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,6 +234,7 @@ 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"
 num-bigint = "0.4.3"
@@ -340,6 +344,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" }

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 }

core/src/replay_stage.rs

@@ -40,7 +40,8 @@ use {
         block_error::BlockError,
         blockstore::Blockstore,
         blockstore_processor::{
-            self, BlockstoreProcessorError, ConfirmationProgress, TransactionStatusSender,
+            self, BlockstoreProcessorError, ConfirmationProgress, ExecuteBatchesInternalMetrics,
+            TransactionStatusSender,
         },
         leader_schedule_cache::LeaderScheduleCache,
         leader_schedule_utils::first_of_consecutive_leader_slots,
@@ -2609,7 +2610,6 @@ impl ReplayStage {
                 match replay_result {
                     Ok(replay_tx_count) => tx_count += replay_tx_count,
                     Err(err) => {
-                        // Error means the slot needs to be marked as dead
                         Self::mark_dead_slot(
                             blockstore,
                             bank,
@@ -2625,8 +2625,7 @@ impl ReplayStage {
                             ancestor_hashes_replay_update_sender,
                             purge_repair_slot_counter,
                         );
-                        // If the bank was corrupted, don't try to run the below logic to check if the
-                        // bank is completed
+                        // don't try to run the below logic to check if the bank is completed
                         continue;
                     }
                 }
@@ -2640,6 +2639,40 @@ impl ReplayStage {
                     .expect("Bank fork progress entry missing for completed bank");
 
                 let replay_stats = bank_progress.replay_stats.clone();
+
+                if let Some((result, complete_execute_timings)) =
+                    bank.wait_for_completed_scheduler()
+                {
+                    let metrics = ExecuteBatchesInternalMetrics::new_with_timings_from_all_threads(
+                        complete_execute_timings,
+                    );
+                    replay_stats
+                        .write()
+                        .unwrap()
+                        .batch_execute
+                        .accumulate(metrics);
+
+                    if let Err(err) = result {
+                        Self::mark_dead_slot(
+                            blockstore,
+                            bank,
+                            bank_forks.read().unwrap().root(),
+                            &BlockstoreProcessorError::InvalidTransaction(err),
+                            rpc_subscriptions,
+                            duplicate_slots_tracker,
+                            gossip_duplicate_confirmed_slots,
+                            epoch_slots_frozen_slots,
+                            progress,
+                            heaviest_subtree_fork_choice,
+                            duplicate_slots_to_repair,
+                            ancestor_hashes_replay_update_sender,
+                            purge_repair_slot_counter,
+                        );
+                        // don't try to run the remaining normal processing for the completed bank
+                        continue;
+                    }
+                }
+
                 let r_replay_stats = replay_stats.read().unwrap();
                 let replay_progress = bank_progress.replay_progress.clone();
                 let r_replay_progress = replay_progress.read().unwrap();

core/src/validator.rs

@@ -92,6 +92,7 @@ use {
         snapshot_hash::StartingSnapshotHashes,
         snapshot_utils::{self, clean_orphaned_account_snapshot_dirs, move_and_async_delete_path},
     },
+    solana_scheduler_pool::SchedulerPool,
     solana_sdk::{
         clock::Slot,
         epoch_schedule::MAX_LEADER_SCHEDULE_EPOCH_OFFSET,
@@ -129,6 +130,7 @@ const WAIT_FOR_SUPERMAJORITY_THRESHOLD_PERCENT: u64 = 80;
 pub enum BlockVerificationMethod {
     #[default]
     BlockstoreProcessor,
+    UnifiedScheduler,
 }
 
 impl BlockVerificationMethod {
@@ -747,6 +749,30 @@ impl Validator {
             config.block_verification_method, config.block_production_method
         );
 
+        let (replay_vote_sender, replay_vote_receiver) = unbounded();
+
+        // block min prioritization fee cache should be readable by RPC, and writable by validator
+        // (by both replay stage and banking stage)
+        let prioritization_fee_cache = Arc::new(PrioritizationFeeCache::default());
+
+        match &config.block_verification_method {
+            BlockVerificationMethod::BlockstoreProcessor => {
+                info!("not installing scheduler pool...");
+            }
+            BlockVerificationMethod::UnifiedScheduler => {
+                let scheduler_pool = SchedulerPool::new_dyn(
+                    config.runtime_config.log_messages_bytes_limit,
+                    transaction_status_sender.clone(),
+                    Some(replay_vote_sender.clone()),
+                    prioritization_fee_cache.clone(),
+                );
+                bank_forks
+                    .write()
+                    .unwrap()
+                    .install_scheduler_pool(scheduler_pool);
+            }
+        }
+
         let leader_schedule_cache = Arc::new(leader_schedule_cache);
         let mut process_blockstore = ProcessBlockStore::new(
             &id,
@@ -866,10 +892,6 @@ impl Validator {
             false => Arc::new(ConnectionCache::with_udp(tpu_connection_pool_size)),
         };
 
-        // block min prioritization fee cache should be readable by RPC, and writable by validator
-        // (by both replay stage and banking stage)
-        let prioritization_fee_cache = Arc::new(PrioritizationFeeCache::default());
-
         let rpc_override_health_check = Arc::new(AtomicBool::new(false));
         let (
             json_rpc_service,
@@ -1068,7 +1090,6 @@ impl Validator {
             info!("Disabled banking tracer");
         }
 
-        let (replay_vote_sender, replay_vote_receiver) = unbounded();
         let tvu = Tvu::new(
             vote_account,
             authorized_voter_keypairs,

ledger-tool/Cargo.toml

@@ -35,6 +35,7 @@ solana-logger = { workspace = true }
 solana-measure = { workspace = true }
 solana-rpc = { workspace = true }
 solana-runtime = { workspace = true }
+solana-scheduler-pool = { workspace = true }
 solana-sdk = { workspace = true }
 solana-stake-program = { workspace = true }
 solana-storage-bigtable = { workspace = true }