Simulate leader blocks wip

Cargo.toml

@@ -294,6 +294,7 @@ curve25519-dalek = { version = "4.1.3", features = ["digest", "rand_core"] }
 dashmap = "5.5.3"
 derivation-path = { version = "0.2.0", default-features = false }
 derive-where = "1.2.7"
+derive_more = { version = "1.0.0", features = ["full"] }
 dialoguer = "0.10.4"
 digest = "0.10.7"
 dir-diff = "0.3.3"
@@ -331,9 +332,6 @@ index_list = "0.2.14"
 indexmap = "2.7.0"
 indicatif = "0.17.9"
 itertools = "0.12.1"
-jemallocator = { package = "tikv-jemallocator", version = "0.4.1", features = [
-    "unprefixed_malloc_on_supported_platforms",
-] }
 js-sys = "0.3.74"
 json5 = "0.4.1"
 jsonrpc-core = "18.0.0"
@@ -357,9 +355,11 @@ lz4 = "1.28.0"
 memmap2 = "0.5.10"
 memoffset = "0.9"
 merlin = "3"
+mimalloc = "0.1.43"
 min-max-heap = "1.3.0"
 mockall = "0.11.4"
 modular-bitfield = "0.11.2"
+more-asserts = "0.3.1"
 nix = "0.29.0"
 num-bigint = "0.4.6"
 num-derive = "0.4"

banking-bench/Cargo.toml

@@ -9,13 +9,15 @@ license = { workspace = true }
 edition = { workspace = true }
 
 [dependencies]
+assert_matches = { workspace = true }
 clap = { version = "3.1.8", features = ["derive", "cargo"] }
 crossbeam-channel = { workspace = true }
 log = { workspace = true }
+mimalloc = { workspace = true }
 rand = { workspace = true }
 rayon = { workspace = true }
 solana-client = { workspace = true }
-solana-core = { workspace = true }
+solana-core = { workspace = true, features = ["dev-context-only-utils"] }
 solana-gossip = { workspace = true }
 solana-ledger = { workspace = true }
 solana-logger = { workspace = true }
@@ -26,6 +28,7 @@ solana-runtime = { workspace = true, features = ["dev-context-only-utils"] }
 solana-sdk = { workspace = true }
 solana-streamer = { workspace = true }
 solana-tpu-client = { workspace = true }
+solana-unified-scheduler-pool = { workspace = true }
 solana-version = { workspace = true }
 
 [features]

banking-bench/src/main.rs

@@ -1,14 +1,17 @@
 #![allow(clippy::arithmetic_side_effects)]
 use {
+    assert_matches::assert_matches,
     clap::{crate_description, crate_name, Arg, ArgEnum, Command},
     crossbeam_channel::{unbounded, Receiver},
     log::*,
     rand::{thread_rng, Rng},
     rayon::prelude::*,
     solana_client::connection_cache::ConnectionCache,
     solana_core::{
-        banking_stage::BankingStage,
-        banking_trace::{BankingPacketBatch, BankingTracer, BANKING_TRACE_DIR_DEFAULT_BYTE_LIMIT},
+        banking_stage::{update_bank_forks_and_poh_recorder_for_new_tpu_bank, BankingStage},
+        banking_trace::{
+            BankingPacketBatch, BankingTracer, Channels, BANKING_TRACE_DIR_DEFAULT_BYTE_LIMIT,
+        },
         validator::BlockProductionMethod,
     },
     solana_gossip::cluster_info::{ClusterInfo, Node},
@@ -29,20 +32,25 @@ use {
         hash::Hash,
         message::Message,
         pubkey::{self, Pubkey},
+        scheduling::SchedulingMode,
         signature::{Keypair, Signature, Signer},
         system_instruction, system_transaction,
         timing::timestamp,
         transaction::Transaction,
     },
     solana_streamer::socket::SocketAddrSpace,
     solana_tpu_client::tpu_client::DEFAULT_TPU_CONNECTION_POOL_SIZE,
+    solana_unified_scheduler_pool::{DefaultSchedulerPool, SupportedSchedulingMode},
     std::{
         sync::{atomic::Ordering, Arc, RwLock},
         thread::sleep,
         time::{Duration, Instant},
     },
 };
 
+#[global_allocator]
+static GLOBAL: mimalloc::MiMalloc = mimalloc::MiMalloc;
+
 // transfer transaction cost = 1 * SIGNATURE_COST +
 //                             2 * WRITE_LOCK_UNITS +
 //                             1 * system_program
@@ -347,7 +355,7 @@ fn main() {
     let (replay_vote_sender, _replay_vote_receiver) = unbounded();
     let bank0 = Bank::new_for_benches(&genesis_config);
     let bank_forks = BankForks::new_rw_arc(bank0);
-    let mut bank = bank_forks.read().unwrap().working_bank();
+    let mut bank = bank_forks.read().unwrap().working_bank_with_scheduler();
 
     // set cost tracker limits to MAX so it will not filter out TXs
     bank.write_cost_tracker()
@@ -440,9 +448,36 @@ fn main() {
             BANKING_TRACE_DIR_DEFAULT_BYTE_LIMIT,
         )))
         .unwrap();
-    let (non_vote_sender, non_vote_receiver) = banking_tracer.create_channel_non_vote();
-    let (tpu_vote_sender, tpu_vote_receiver) = banking_tracer.create_channel_tpu_vote();
-    let (gossip_vote_sender, gossip_vote_receiver) = banking_tracer.create_channel_gossip_vote();
+    let prioritization_fee_cache = Arc::new(PrioritizationFeeCache::new(0u64));
+    let scheduler_pool = if matches!(
+        block_production_method,
+        BlockProductionMethod::UnifiedScheduler
+    ) {
+        let pool = DefaultSchedulerPool::new(
+            SupportedSchedulingMode::Either(SchedulingMode::BlockProduction),
+            None,
+            None,
+            None,
+            Some(replay_vote_sender.clone()),
+            prioritization_fee_cache.clone(),
+            poh_recorder.read().unwrap().new_recorder(),
+        );
+        bank_forks
+            .write()
+            .unwrap()
+            .install_scheduler_pool(pool.clone());
+        Some(pool)
+    } else {
+        None
+    };
+    let Channels {
+        non_vote_sender,
+        non_vote_receiver,
+        tpu_vote_sender,
+        tpu_vote_receiver,
+        gossip_vote_sender,
+        gossip_vote_receiver,
+    } = banking_tracer.create_channels(scheduler_pool.as_ref());
     let cluster_info = {
         let keypair = Arc::new(Keypair::new());
         let node = Node::new_localhost_with_pubkey(&keypair.pubkey());
@@ -461,7 +496,7 @@ fn main() {
         ),
     };
     let banking_stage = BankingStage::new_num_threads(
-        block_production_method,
+        block_production_method.clone(),
         &cluster_info,
         &poh_recorder,
         non_vote_receiver,
@@ -473,10 +508,23 @@ fn main() {
         None,
         Arc::new(connection_cache),
         bank_forks.clone(),
-        &Arc::new(PrioritizationFeeCache::new(0u64)),
+        &prioritization_fee_cache,
         false,
+        scheduler_pool,
     );
 
+    // This bench processes transactions, starting from the very first bank, so special-casing is
+    // needed for unified scheduler.
+    if matches!(
+        block_production_method,
+        BlockProductionMethod::UnifiedScheduler
+    ) {
+        bank = bank_forks
+            .write()
+            .unwrap()
+            .reinstall_block_production_scheduler_into_working_genesis_bank();
+    }
+
     // This is so that the signal_receiver does not go out of scope after the closure.
     // If it is dropped before poh_service, then poh_service will error when
     // calling send() on the channel.
@@ -537,33 +585,31 @@ fn main() {
             tx_total_us += now.elapsed().as_micros() as u64;
 
             let mut poh_time = Measure::start("poh_time");
-            poh_recorder
+            let cleared_bank = poh_recorder
                 .write()
                 .unwrap()
                 .reset(bank.clone(), Some((bank.slot(), bank.slot() + 1)));
+            assert_matches!(cleared_bank, None);
             poh_time.stop();
 
             let mut new_bank_time = Measure::start("new_bank");
+            if let Some((result, _timings)) = bank.wait_for_completed_scheduler() {
+                assert_matches!(result, Ok(_));
+            }
             let new_slot = bank.slot() + 1;
-            let new_bank = Bank::new_from_parent(bank, &collector, new_slot);
+            let new_bank = Bank::new_from_parent(bank.clone(), &collector, new_slot);
             new_bank_time.stop();
 
             let mut insert_time = Measure::start("insert_time");
-            bank_forks.write().unwrap().insert(new_bank);
-            bank = bank_forks.read().unwrap().working_bank();
+            update_bank_forks_and_poh_recorder_for_new_tpu_bank(
+                &bank_forks,
+                &poh_recorder,
+                new_bank,
+                false,
+            );
+            bank = bank_forks.read().unwrap().working_bank_with_scheduler();
             insert_time.stop();
 
-            // set cost tracker limits to MAX so it will not filter out TXs
-            bank.write_cost_tracker()
-                .unwrap()
-                .set_limits(u64::MAX, u64::MAX, u64::MAX);
-
-            assert!(poh_recorder.read().unwrap().bank().is_none());
-            poh_recorder
-                .write()
-                .unwrap()
-                .set_bank_for_test(bank.clone());
-            assert!(poh_recorder.read().unwrap().bank().is_some());
             debug!(
                 "new_bank_time: {}us insert_time: {}us poh_time: {}us",
                 new_bank_time.as_us(),

bench-tps/Cargo.toml

@@ -14,6 +14,7 @@ clap = { workspace = true }
 crossbeam-channel = { workspace = true }
 csv = { workspace = true }
 log = { workspace = true }
+mimalloc = { workspace = true }
 rand = { workspace = true }
 rayon = { workspace = true }
 serde = { workspace = true }

bench-tps/src/main.rs

@@ -31,6 +31,9 @@ use {
     },
 };
 
+#[global_allocator]
+static GLOBAL: mimalloc::MiMalloc = mimalloc::MiMalloc;
+
 /// Number of signatures for all transactions in ~1 week at ~100K TPS
 pub const NUM_SIGNATURES_FOR_TXS: u64 = 100_000 * 60 * 60 * 24 * 7;
 

ci/test-checks.sh

@@ -83,7 +83,7 @@ _ scripts/check-dev-context-only-utils.sh tree
 
 _ scripts/cargo-for-all-lock-files.sh -- "+${rust_nightly}" fmt --all -- --check
 
-_ ci/do-audit.sh
+# _ ci/do-audit.sh
 
 if [[ -n $CI ]] && [[ $CHANNEL = "stable" ]]; then
   _ ci/check-install-all.sh

core/Cargo.toml

@@ -24,6 +24,7 @@ bytes = { workspace = true }
 chrono = { workspace = true, features = ["default", "serde"] }
 crossbeam-channel = { workspace = true }
 dashmap = { workspace = true, features = ["rayon", "raw-api"] }
+derive_more = { workspace = true }
 etcd-client = { workspace = true, features = ["tls"] }
 futures = { workspace = true }
 histogram = { workspace = true }
@@ -87,6 +88,7 @@ solana-tls-utils = { workspace = true }
 solana-tpu-client = { workspace = true }
 solana-transaction-status = { workspace = true }
 solana-turbine = { workspace = true }
+solana-unified-scheduler-logic = { workspace = true }
 solana-unified-scheduler-pool = { workspace = true }
 solana-version = { workspace = true }
 solana-vote = { workspace = true }

core/benches/banking_stage.rs

@@ -2,7 +2,7 @@
 #![feature(test)]
 
 use {
-    solana_core::validator::BlockProductionMethod,
+    solana_core::{banking_trace::Channels, validator::BlockProductionMethod},
     solana_vote_program::{vote_state::TowerSync, vote_transaction::new_tower_sync_transaction},
 };
 
@@ -211,9 +211,14 @@ fn bench_banking(bencher: &mut Bencher, tx_type: TransactionType) {
     genesis_config.ticks_per_slot = 10_000;
 
     let banking_tracer = BankingTracer::new_disabled();
-    let (non_vote_sender, non_vote_receiver) = banking_tracer.create_channel_non_vote();
-    let (tpu_vote_sender, tpu_vote_receiver) = banking_tracer.create_channel_tpu_vote();
-    let (gossip_vote_sender, gossip_vote_receiver) = banking_tracer.create_channel_gossip_vote();
+    let Channels {
+        non_vote_sender,
+        non_vote_receiver,
+        tpu_vote_sender,
+        tpu_vote_receiver,
+        gossip_vote_sender,
+        gossip_vote_receiver,
+    } = banking_tracer.create_channels(None);
 
     let mut bank = Bank::new_for_benches(&genesis_config);
     // Allow arbitrary transaction processing time for the purposes of this bench
@@ -304,6 +309,7 @@ fn bench_banking(bencher: &mut Bencher, tx_type: TransactionType) {
         bank_forks,
         &Arc::new(PrioritizationFeeCache::new(0u64)),
         false,
+        None,
     );
 
     let chunk_len = verified.len() / CHUNKS;

core/benches/banking_trace.rs

@@ -7,7 +7,7 @@ use {
         for_test::{
             drop_and_clean_temp_dir_unless_suppressed, sample_packet_batch, terminate_tracer,
         },
-        receiving_loop_with_minimized_sender_overhead, BankingPacketBatch, BankingTracer,
+        receiving_loop_with_minimized_sender_overhead, BankingPacketBatch, BankingTracer, Channels,
         TraceError, TracerThreadResult, BANKING_TRACE_DIR_DEFAULT_BYTE_LIMIT,
     },
     std::{
@@ -35,7 +35,11 @@ fn black_box_packet_batch(packet_batch: BankingPacketBatch) -> TracerThreadResul
 fn bench_banking_tracer_main_thread_overhead_noop_baseline(bencher: &mut Bencher) {
     let exit = Arc::<AtomicBool>::default();
     let tracer = BankingTracer::new_disabled();
-    let (non_vote_sender, non_vote_receiver) = tracer.create_channel_non_vote();
+    let Channels {
+        non_vote_sender,
+        non_vote_receiver,
+        ..
+    } = tracer.create_channels(None);
 
     let exit_for_dummy_thread = exit.clone();
     let dummy_main_thread = thread::spawn(move || {
@@ -64,7 +68,11 @@ fn bench_banking_tracer_main_thread_overhead_under_peak_write(bencher: &mut Benc
         BANKING_TRACE_DIR_DEFAULT_BYTE_LIMIT,
     )))
     .unwrap();
-    let (non_vote_sender, non_vote_receiver) = tracer.create_channel_non_vote();
+    let Channels {
+        non_vote_sender,
+        non_vote_receiver,
+        ..
+    } = tracer.create_channels(None);
 
     let exit_for_dummy_thread = exit.clone();
     let dummy_main_thread = thread::spawn(move || {
@@ -101,7 +109,11 @@ fn bench_banking_tracer_main_thread_overhead_under_sustained_write(bencher: &mut
         1024 * 1024, // cause more frequent trace file rotation
     )))
     .unwrap();
-    let (non_vote_sender, non_vote_receiver) = tracer.create_channel_non_vote();
+    let Channels {
+        non_vote_sender,
+        non_vote_receiver,
+        ..
+    } = tracer.create_channels(None);
 
     let exit_for_dummy_thread = exit.clone();
     let dummy_main_thread = thread::spawn(move || {
@@ -142,7 +154,11 @@ fn bench_banking_tracer_background_thread_throughput(bencher: &mut Bencher) {
 
         let (tracer, tracer_thread) =
             BankingTracer::new(Some((&path, exit.clone(), 50 * 1024 * 1024))).unwrap();
-        let (non_vote_sender, non_vote_receiver) = tracer.create_channel_non_vote();
+        let Channels {
+            non_vote_sender,
+            non_vote_receiver,
+            ..
+        } = tracer.create_channels(None);
 
         let dummy_main_thread = thread::spawn(move || {
             receiving_loop_with_minimized_sender_overhead::<_, TraceError, 0>(

core/benches/sigverify_stage.rs

@@ -185,8 +185,9 @@ fn bench_sigverify_stage(bencher: &mut Bencher, use_same_tx: bool) {
             if let Ok(message) = verified_r.recv_timeout(Duration::from_millis(10)) {
                 let (verifieds, tracer_packet_stats) = (&message.0, message.1.as_ref().unwrap());
                 received += verifieds.iter().map(|batch| batch.len()).sum::<usize>();
-                total_tracer_packets_received_in_sigverify_stage +=
-                    tracer_packet_stats.total_tracer_packets_received_in_sigverify_stage;
+                total_tracer_packets_received_in_sigverify_stage += tracer_packet_stats
+                    .total_tracer_packets_received_in_sigverify_stage
+                    .0;
                 test::black_box(message);
                 if total_tracer_packets_received_in_sigverify_stage >= sent_len {
                     break;