Finish unified scheduler plumbing with min impl (solana-labs#34300)

* Finalize unified scheduler plumbing with min impl * Fix comment * Rename leftover type name... * Make logging text less ambiguous * Make PhantomData simplyer without already used S * Make TaskHandler stateless again * Introduce HandlerContext to simplify TaskHandler * Add comment for coexistence of Pool::{new,new_dyn} * Fix grammar * Remove confusing const for upcoming changes * Demote InstalledScheduler::context() into dcou * Delay drop of context up to return_to_pool()-ing * Revert "Demote InstalledScheduler::context() into dcou" This reverts commit 049a126. * Revert "Delay drop of context up to return_to_pool()-ing" This reverts commit 60b1bd2. * Make context handling really type-safe * Update comment * Fix grammar... * Refine type aliases for boxed traits * Swap the tuple order for readability & semantics * Simplify PooledScheduler::result_with_timings type * Restore .in_sequence() * Use where for aesthetics * Simplify if... * Fix typo... * Polish ::schedule_execution() a bit * Fix rebase conflicts.. * Make test more readable * Fix test failures after rebase...
anza-xyz · Dec 19, 2023 · d2b5afc · d2b5afc
1 parent 4a8d27d
commit d2b5afc
Show file tree

Hide file tree

Showing 17 changed files with 1,031 additions and 85 deletions.
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/Cargo.toml b/Cargo.toml
@@ -108,6 +108,8 @@ members = [
     "transaction-status",
     "turbine",
     "udp-client",
+    "unified-scheduler-logic",
+    "unified-scheduler-pool",
     "upload-perf",
     "validator",
     "version",
@@ -357,6 +359,8 @@ solana-pubsub-client = { path = "pubsub-client", version = "=1.18.0" }
 solana-quic-client = { path = "quic-client", version = "=1.18.0" }
 solana-rayon-threadlimit = { path = "rayon-threadlimit", version = "=1.18.0" }
 solana-remote-wallet = { path = "remote-wallet", version = "=1.18.0", default-features = false }
+solana-unified-scheduler-logic = { path = "unified-scheduler-logic", version = "=1.18.0" }
+solana-unified-scheduler-pool = { path = "unified-scheduler-pool", version = "=1.18.0" }
 solana-rpc = { path = "rpc", version = "=1.18.0" }
 solana-rpc-client = { path = "rpc-client", version = "=1.18.0", default-features = false }
 solana-rpc-client-api = { path = "rpc-client-api", version = "=1.18.0" }

diff --git a/ci/run-sanity.sh b/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
diff --git a/core/Cargo.toml b/core/Cargo.toml
@@ -69,6 +69,7 @@ solana-streamer = { workspace = true }
 solana-tpu-client = { workspace = true }
 solana-transaction-status = { workspace = true }
 solana-turbine = { workspace = true }
+solana-unified-scheduler-pool = { workspace = true }
 solana-version = { workspace = true }
 solana-vote = { workspace = true }
 solana-vote-program = { workspace = true }

diff --git a/core/src/validator.rs b/core/src/validator.rs
@@ -118,6 +118,7 @@ use {
     solana_send_transaction_service::send_transaction_service,
     solana_streamer::{socket::SocketAddrSpace, streamer::StakedNodes},
     solana_turbine::{self, broadcast_stage::BroadcastStageType},
+    solana_unified_scheduler_pool::DefaultSchedulerPool,
     solana_vote_program::vote_state,
     solana_wen_restart::wen_restart::wait_for_wen_restart,
     std::{
@@ -144,6 +145,7 @@ const WAIT_FOR_SUPERMAJORITY_THRESHOLD_PERCENT: u64 = 80;
 pub enum BlockVerificationMethod {
     #[default]
     BlockstoreProcessor,
+    UnifiedScheduler,
 }
 
 impl BlockVerificationMethod {
@@ -813,6 +815,24 @@ impl Validator {
         // (by both replay stage and banking stage)
         let prioritization_fee_cache = Arc::new(PrioritizationFeeCache::default());
 
+        match &config.block_verification_method {
+            BlockVerificationMethod::BlockstoreProcessor => {
+                info!("no scheduler pool is installed for block verification...");
+            }
+            BlockVerificationMethod::UnifiedScheduler => {
+                let scheduler_pool = DefaultSchedulerPool::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 entry_notification_sender = entry_notifier_service
             .as_ref()

diff --git a/ledger-tool/Cargo.toml b/ledger-tool/Cargo.toml
@@ -45,6 +45,7 @@ solana-stake-program = { workspace = true }
 solana-storage-bigtable = { workspace = true }
 solana-streamer = { workspace = true }
 solana-transaction-status = { workspace = true }
+solana-unified-scheduler-pool = { workspace = true }
 solana-version = { workspace = true }
 solana-vote-program = { workspace = true }
 solana_rbpf = { workspace = true, features = ["debugger"] }

diff --git a/ledger-tool/src/ledger_utils.rs b/ledger-tool/src/ledger_utils.rs
@@ -30,6 +30,7 @@ use {
             PrunedBanksRequestHandler, SnapshotRequestHandler,
         },
         bank_forks::BankForks,
+        prioritization_fee_cache::PrioritizationFeeCache,
         snapshot_config::SnapshotConfig,
         snapshot_hash::StartingSnapshotHashes,
         snapshot_utils::{
@@ -42,6 +43,7 @@ use {
         timing::timestamp,
     },
     solana_streamer::socket::SocketAddrSpace,
+    solana_unified_scheduler_pool::DefaultSchedulerPool,
     std::{
         path::{Path, PathBuf},
         process::exit,
@@ -305,6 +307,25 @@ pub fn load_and_process_ledger(
         "Using: block-verification-method: {}",
         block_verification_method,
     );
+    match block_verification_method {
+        BlockVerificationMethod::BlockstoreProcessor => {
+            info!("no scheduler pool is installed for block verification...");
+        }
+        BlockVerificationMethod::UnifiedScheduler => {
+            let no_transaction_status_sender = None;
+            let no_replay_vote_sender = None;
+            let ignored_prioritization_fee_cache = Arc::new(PrioritizationFeeCache::new(0u64));
+            bank_forks
+                .write()
+                .unwrap()
+                .install_scheduler_pool(DefaultSchedulerPool::new_dyn(
+                    process_options.runtime_config.log_messages_bytes_limit,
+                    no_transaction_status_sender,
+                    no_replay_vote_sender,
+                    ignored_prioritization_fee_cache,
+                ));
+        }
+    }
 
     let node_id = Arc::new(Keypair::new());
     let cluster_info = Arc::new(ClusterInfo::new(

diff --git a/ledger/src/blockstore_processor.rs b/ledger/src/blockstore_processor.rs
@@ -74,7 +74,7 @@ use {
     thiserror::Error,
 };
 
-struct TransactionBatchWithIndexes<'a, 'b> {
+pub struct TransactionBatchWithIndexes<'a, 'b> {
     pub batch: TransactionBatch<'a, 'b>,
     pub transaction_indexes: Vec<usize>,
 }
@@ -134,7 +134,7 @@ fn get_first_error(
     first_err
 }
 
-fn execute_batch(
+pub fn execute_batch(
     batch: &TransactionBatchWithIndexes,
     bank: &Arc<Bank>,
     transaction_status_sender: Option<&TransactionStatusSender>,
@@ -1832,7 +1832,7 @@ pub struct TransactionStatusBatch {
     pub transaction_indexes: Vec<usize>,
 }
 
-#[derive(Clone)]
+#[derive(Clone, Debug)]
 pub struct TransactionStatusSender {
     pub sender: Sender<TransactionStatusMessage>,
 }
@@ -1947,7 +1947,9 @@ pub mod tests {
             genesis_utils::{
                 self, create_genesis_config_with_vote_accounts, ValidatorVoteKeypairs,
             },
-            installed_scheduler_pool::{MockInstalledScheduler, SchedulingContext, WaitReason},
+            installed_scheduler_pool::{
+                MockInstalledScheduler, MockUninstalledScheduler, SchedulingContext,
+            },
         },
         solana_sdk::{
             account::{AccountSharedData, WritableAccount},
@@ -4545,27 +4547,34 @@ pub mod tests {
         let txs = create_test_transactions(&mint_keypair, &genesis_config.hash());
 
         let mut mocked_scheduler = MockInstalledScheduler::new();
-        let mut seq = mockall::Sequence::new();
+        let seq = Arc::new(Mutex::new(mockall::Sequence::new()));
+        let seq_cloned = seq.clone();
         mocked_scheduler
             .expect_context()
             .times(1)
-            .in_sequence(&mut seq)
+            .in_sequence(&mut seq.lock().unwrap())
             .return_const(context);
         mocked_scheduler
             .expect_schedule_execution()
             .times(txs.len())
             .returning(|_| ());
         mocked_scheduler
             .expect_wait_for_termination()
-            .with(mockall::predicate::eq(WaitReason::DroppedFromBankForks))
-            .times(1)
-            .in_sequence(&mut seq)
-            .returning(|_| None);
-        mocked_scheduler
-            .expect_return_to_pool()
+            .with(mockall::predicate::eq(true))
             .times(1)
-            .in_sequence(&mut seq)
-            .returning(|| ());
+            .in_sequence(&mut seq.lock().unwrap())
+            .returning(move |_| {
+                let mut mocked_uninstalled_scheduler = MockUninstalledScheduler::new();
+                mocked_uninstalled_scheduler
+                    .expect_return_to_pool()
+                    .times(1)
+                    .in_sequence(&mut seq_cloned.lock().unwrap())
+                    .returning(|| ());
+                (
+                    (Ok(()), ExecuteTimings::default()),
+                    Box::new(mocked_uninstalled_scheduler),
+                )
+            });
         let bank = BankWithScheduler::new(bank, Some(Box::new(mocked_scheduler)));
 
         let batch = bank.prepare_sanitized_batch(&txs);

diff --git a/local-cluster/tests/local_cluster.rs b/local-cluster/tests/local_cluster.rs
@@ -4,6 +4,7 @@ use {
     crossbeam_channel::{unbounded, Receiver},
     gag::BufferRedirect,
     log::*,
+    rand::seq::IteratorRandom,
     serial_test::serial,
     solana_accounts_db::{
         accounts_db::create_accounts_run_and_snapshot_dirs, hardened_unpack::open_genesis_config,
@@ -15,7 +16,7 @@ use {
         },
         optimistic_confirmation_verifier::OptimisticConfirmationVerifier,
         replay_stage::DUPLICATE_THRESHOLD,
-        validator::ValidatorConfig,
+        validator::{BlockVerificationMethod, ValidatorConfig},
     },
     solana_download_utils::download_snapshot_archive,
     solana_entry::entry::create_ticks,
@@ -5456,6 +5457,44 @@ fn test_duplicate_shreds_switch_failure() {
     );
 }
 
+#[test]
+#[serial]
+fn test_randomly_mixed_block_verification_methods_between_bootstrap_and_not() {
+    // tailored logging just to see two block verification methods are working correctly
+    solana_logger::setup_with_default(
+        "solana_metrics::metrics=warn,\
+         solana_core=warn,\
+         solana_runtime::installed_scheduler_pool=trace,\
+         solana_ledger::blockstore_processor=debug,\
+         info",
+    );
+
+    let num_nodes = 2;
+    let mut config = ClusterConfig::new_with_equal_stakes(
+        num_nodes,
+        DEFAULT_CLUSTER_LAMPORTS,
+        DEFAULT_NODE_STAKE,
+    );
+
+    // Randomly switch to use unified scheduler
+    config
+        .validator_configs
+        .iter_mut()
+        .choose(&mut rand::thread_rng())
+        .unwrap()
+        .block_verification_method = BlockVerificationMethod::UnifiedScheduler;
+
+    let local = LocalCluster::new(&mut config, SocketAddrSpace::Unspecified);
+    cluster_tests::spend_and_verify_all_nodes(
+        &local.entry_point_info,
+        &local.funding_keypair,
+        num_nodes,
+        HashSet::new(),
+        SocketAddrSpace::Unspecified,
+        &local.connection_cache,
+    );
+}
+
 /// Forks previous marked invalid should be marked as such in fork choice on restart
 #[test]
 #[serial]

diff --git a/programs/sbf/Cargo.lock b/programs/sbf/Cargo.lock
diff --git a/runtime/src/bank.rs b/runtime/src/bank.rs
@@ -4290,7 +4290,7 @@ impl Bank {
     }
 
     /// Prepare a transaction batch from a single transaction without locking accounts
-    pub(crate) fn prepare_unlocked_batch_from_single_tx<'a>(
+    pub fn prepare_unlocked_batch_from_single_tx<'a>(
         &'a self,
         transaction: &'a SanitizedTransaction,
     ) -> TransactionBatch<'_, '_> {