Add new execution code-path for unified scheduler

Cargo.toml

@@ -76,6 +76,8 @@ members = [
     "rpc-test",
     "runtime",
     "runtime/store-tool",
+    "scheduler",
+    "scheduler-pool",
     "sdk",
     "sdk/cargo-build-bpf",
     "sdk/cargo-build-sbf",
@@ -121,6 +123,7 @@ license = "Apache-2.0"
 edition = "2021"
 
 [workspace.dependencies]
+aquamarine = "0.3.1"
 aes-gcm-siv = "0.10.3"
 ahash = "0.8.3"
 anyhow = "1.0.70"
@@ -231,8 +234,10 @@ memmap2 = "0.5.10"
 memoffset = "0.8"
 merlin = "3"
 min-max-heap = "1.3.0"
+mockall = "0.11.4"
 modular-bitfield = "0.11.2"
 nix = "0.25.1"
+nohash-hasher = "0.2.0"
 num-bigint = "0.4.3"
 num_cpus = "1.13.1"
 num_enum = "0.5.7"
@@ -340,6 +345,8 @@ solana-rpc-client = { path = "rpc-client", version = "=1.16.0", default-features
 solana-rpc-client-api = { path = "rpc-client-api", version = "=1.16.0" }
 solana-rpc-client-nonce-utils = { path = "rpc-client-nonce-utils", version = "=1.16.0" }
 solana-runtime = { path = "runtime", version = "=1.16.0" }
+solana-scheduler = { path = "scheduler", version = "=1.16.0" }
+solana-scheduler-pool = { path = "scheduler-pool", version = "=1.16.0" }
 solana-sdk = { path = "sdk", version = "=1.16.0" }
 solana-sdk-macro = { path = "sdk/macro", version = "=1.16.0" }
 solana-send-transaction-service = { path = "send-transaction-service", version = "=1.16.0" }

banking-bench/Cargo.toml

@@ -21,7 +21,7 @@ solana-ledger = { workspace = true }
 solana-logger = { workspace = true }
 solana-measure = { workspace = true }
 solana-perf = { workspace = true }
-solana-poh = { workspace = true }
+solana-poh = { workspace = true, features = ["test-in-workspace"] }
 solana-runtime = { workspace = true }
 solana-sdk = { workspace = true }
 solana-streamer = { workspace = true }

banking-bench/src/main.rs

@@ -538,7 +538,7 @@ fn main() {
                 );
 
                 assert!(poh_recorder.read().unwrap().bank().is_none());
-                poh_recorder.write().unwrap().set_bank(&bank, false);
+                poh_recorder.write().unwrap().set_bank_for_test(&bank);
                 assert!(poh_recorder.read().unwrap().bank().is_some());
                 debug!(
                     "new_bank_time: {}us insert_time: {}us poh_time: {}us",

ci/run-sanity.sh

@@ -39,4 +39,5 @@ $solana_ledger_tool create-snapshot --ledger config/ledger "$snapshot_slot" conf
 cp config/ledger/genesis.tar.bz2 config/snapshot-ledger
 $solana_ledger_tool copy --ledger config/ledger \
   --target-db config/snapshot-ledger --starting-slot "$snapshot_slot" --ending-slot "$latest_slot"
-$solana_ledger_tool verify --ledger config/snapshot-ledger
+$solana_ledger_tool verify --ledger config/snapshot-ledger --block-verification-method blockstore-processor
+$solana_ledger_tool verify --ledger config/snapshot-ledger --block-verification-method unified-scheduler

ci/test-bench.sh

@@ -56,6 +56,10 @@ _ $cargoNightly bench --manifest-path gossip/Cargo.toml ${V:+--verbose} \
 _ $cargoNightly bench --manifest-path poh/Cargo.toml ${V:+--verbose} \
   -- -Z unstable-options --format=json | tee -a "$BENCH_FILE"
 
+# Run scheduler-pool benches
+_ $cargoNightly bench --manifest-path scheduler-pool/Cargo.toml ${V:+--verbose} \
+  -- -Z unstable-options --format=json | tee -a "$BENCH_FILE"
+
 # Run core benches
 _ $cargoNightly bench --manifest-path core/Cargo.toml ${V:+--verbose} \
   -- -Z unstable-options --format=json | tee -a "$BENCH_FILE"

core/Cargo.toml

@@ -55,6 +55,7 @@ solana-rayon-threadlimit = { workspace = true }
 solana-rpc = { workspace = true }
 solana-rpc-client-api = { workspace = true }
 solana-runtime = { workspace = true }
+solana-scheduler-pool = { workspace = true }
 solana-sdk = { workspace = true }
 solana-send-transaction-service = { workspace = true }
 solana-streamer = { workspace = true }
@@ -76,7 +77,9 @@ raptorq = { workspace = true }
 serde_json = { workspace = true }
 serial_test = { workspace = true }
 solana-logger = { workspace = true }
+solana-poh = { workspace = true, features = ["test-in-workspace"] }
 solana-program-runtime = { workspace = true }
+solana-runtime = { workspace = true, features = ["test-in-workspace"] }
 solana-stake-program = { workspace = true }
 static_assertions = { workspace = true }
 systemstat = { workspace = true }

core/benches/banking_stage.rs

@@ -30,7 +30,8 @@ use {
     solana_perf::{packet::to_packet_batches, test_tx::test_tx},
     solana_poh::poh_recorder::{create_test_recorder, WorkingBankEntry},
     solana_runtime::{
-        bank::Bank, bank_forks::BankForks, prioritization_fee_cache::PrioritizationFeeCache,
+        bank::Bank, bank_forks::BankForks, installed_scheduler_pool::BankWithScheduler,
+        prioritization_fee_cache::PrioritizationFeeCache,
     },
     solana_sdk::{
         genesis_config::GenesisConfig,
@@ -385,6 +386,7 @@ fn simulate_process_entries(
     num_accounts: usize,
 ) {
     let bank = Arc::new(Bank::new_for_benches(genesis_config));
+    let bank = BankWithScheduler::new_without_scheduler(bank);
 
     for i in 0..(num_accounts / 2) {
         bank.transfer(initial_lamports, mint_keypair, &keypairs[i * 2].pubkey())

core/src/banking_stage.rs

@@ -1018,7 +1018,7 @@ mod tests {
 
             let poh_simulator = simulate_poh(record_receiver, &poh_recorder);
 
-            poh_recorder.write().unwrap().set_bank(&bank, false);
+            poh_recorder.write().unwrap().set_bank_for_test(&bank);
             let pubkey = solana_sdk::pubkey::new_rand();
             let keypair2 = Keypair::new();
             let pubkey2 = solana_sdk::pubkey::new_rand();

core/src/banking_stage/consumer.rs

@@ -747,7 +747,7 @@ mod tests {
         let recorder = poh_recorder.new_recorder();
         let poh_recorder = Arc::new(RwLock::new(poh_recorder));
 
-        poh_recorder.write().unwrap().set_bank(&bank, false);
+        poh_recorder.write().unwrap().set_bank_for_test(&bank);
 
         let poh_simulator = simulate_poh(record_receiver, &poh_recorder);
 
@@ -894,7 +894,7 @@ mod tests {
 
             let poh_simulator = simulate_poh(record_receiver, &poh_recorder);
 
-            poh_recorder.write().unwrap().set_bank(&bank, false);
+            poh_recorder.write().unwrap().set_bank_for_test(&bank);
             let (replay_vote_sender, _replay_vote_receiver) = unbounded();
             let committer = Committer::new(
                 None,
@@ -1021,7 +1021,7 @@ mod tests {
 
             let poh_simulator = simulate_poh(record_receiver, &poh_recorder);
 
-            poh_recorder.write().unwrap().set_bank(&bank, false);
+            poh_recorder.write().unwrap().set_bank_for_test(&bank);
             let (replay_vote_sender, _replay_vote_receiver) = unbounded();
             let committer = Committer::new(
                 None,
@@ -1093,7 +1093,7 @@ mod tests {
 
             let poh_simulator = simulate_poh(record_receiver, &poh_recorder);
 
-            poh_recorder.write().unwrap().set_bank(&bank, false);
+            poh_recorder.write().unwrap().set_bank_for_test(&bank);
             let (replay_vote_sender, _replay_vote_receiver) = unbounded();
             let committer = Committer::new(
                 None,
@@ -1212,7 +1212,7 @@ mod tests {
             let recorder = poh_recorder.new_recorder();
             let poh_recorder = Arc::new(RwLock::new(poh_recorder));
 
-            poh_recorder.write().unwrap().set_bank(&bank, false);
+            poh_recorder.write().unwrap().set_bank_for_test(&bank);
 
             let poh_simulator = simulate_poh(record_receiver, &poh_recorder);
 
@@ -1512,7 +1512,7 @@ mod tests {
 
             let poh_simulator = simulate_poh(record_receiver, &poh_recorder);
 
-            poh_recorder.write().unwrap().set_bank(&bank, false);
+            poh_recorder.write().unwrap().set_bank_for_test(&bank);
 
             let shreds = entries_to_test_shreds(
                 &entries,
@@ -1650,7 +1650,7 @@ mod tests {
 
             let poh_simulator = simulate_poh(record_receiver, &poh_recorder);
 
-            poh_recorder.write().unwrap().set_bank(&bank, false);
+            poh_recorder.write().unwrap().set_bank_for_test(&bank);
 
             let shreds = entries_to_test_shreds(
                 &entries,
@@ -1751,7 +1751,7 @@ mod tests {
             assert_eq!(buffered_packet_batches.len(), num_conflicting_transactions);
             // When the working bank in poh_recorder is Some, all packets should be processed.
             // Multi-Iterator will process them 1-by-1 if all txs are conflicting.
-            poh_recorder.write().unwrap().set_bank(&bank, false);
+            poh_recorder.write().unwrap().set_bank_for_test(&bank);
             let bank_start = poh_recorder.read().unwrap().bank_start().unwrap();
             let banking_stage_stats = BankingStageStats::default();
             consumer.consume_buffered_packets(
@@ -1831,7 +1831,7 @@ mod tests {
             assert_eq!(buffered_packet_batches.len(), num_conflicting_transactions);
             // When the working bank in poh_recorder is Some, all packets should be processed.
             // Multi-Iterator will process them 1-by-1 if all txs are conflicting.
-            poh_recorder.write().unwrap().set_bank(&bank, false);
+            poh_recorder.write().unwrap().set_bank_for_test(&bank);
             let bank_start = poh_recorder.read().unwrap().bank_start().unwrap();
             consumer.consume_buffered_packets(
                 &bank_start,
@@ -1886,7 +1886,7 @@ mod tests {
             // When the working bank in poh_recorder is Some, all packets should be processed
             // except except for retryable errors. Manually take the lock of a transaction to
             // simulate another thread processing a transaction with that lock.
-            poh_recorder.write().unwrap().set_bank(&bank, false);
+            poh_recorder.write().unwrap().set_bank_for_test(&bank);
             let bank_start = poh_recorder.read().unwrap().bank_start().unwrap();
 
             let lock_account = transactions[0].message.account_keys[1];