New exec code path rebased

Cargo.toml

@@ -179,6 +179,7 @@ console_error_panic_hook = "0.1.7"
 console_log = "0.2.2"
 const_format = "0.2.32"
 core_affinity = "0.5.10"
+cpu-time = "1.0.0"
 criterion = "0.5.1"
 criterion-stats = "0.3.0"
 crossbeam-channel = "0.5.11"
@@ -268,6 +269,7 @@ predicates = "2.1"
 pretty-hex = "0.3.0"
 prio-graph = "0.2.1"
 proc-macro2 = "1.0.78"
+procfs = "0.16.0"
 proptest = "1.4"
 prost = "0.11.9"
 prost-build = "0.11.9"
@@ -288,6 +290,7 @@ reqwest = { version = "0.11.23", default-features = false }
 rolling-file = "0.2.0"
 rpassword = "7.3"
 rustc_version = "0.4"
+rustix = "0.38.21"
 rustls = { version = "0.21.10", default-features = false, features = ["quic"] }
 rustversion = "1.0.14"
 scopeguard = "1.2.0"
@@ -439,6 +442,8 @@ zstd = "0.11.2"
 # for details, see https://github.com/solana-labs/crossbeam/commit/fd279d707025f0e60951e429bf778b4813d1b6bf
 crossbeam-epoch = { git = "https://github.com/solana-labs/crossbeam", rev = "fd279d707025f0e60951e429bf778b4813d1b6bf" }
 
+crossbeam-channel = { git = "https://github.com/ryoqun/crossbeam", rev = "438ec7cdaf6c6a8f593e50344c725fef8a13c7a5" }
+
 # We include the following crates as our dependencies above from crates.io:
 #
 #  * spl-associated-token-account

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/src/drop_bank_service.rs

@@ -1,19 +1,16 @@
 use {
     crossbeam_channel::Receiver,
     solana_measure::measure::Measure,
-    solana_runtime::bank::Bank,
-    std::{
-        sync::Arc,
-        thread::{self, Builder, JoinHandle},
-    },
+    solana_runtime::installed_scheduler_pool::BankWithScheduler,
+    std::thread::{self, Builder, JoinHandle},
 };
 
 pub struct DropBankService {
     thread_hdl: JoinHandle<()>,
 }
 
 impl DropBankService {
-    pub fn new(bank_receiver: Receiver<Vec<Arc<Bank>>>) -> Self {
+    pub fn new(bank_receiver: Receiver<Vec<BankWithScheduler>>) -> Self {
         let thread_hdl = Builder::new()
             .name("solDropBankSrvc".to_string())
             .spawn(move || {

core/src/replay_stage.rs

@@ -536,7 +536,7 @@ impl ReplayStage {
         cluster_slots_update_sender: ClusterSlotsUpdateSender,
         cost_update_sender: Sender<CostUpdate>,
         voting_sender: Sender<VoteOp>,
-        drop_bank_sender: Sender<Vec<Arc<Bank>>>,
+        drop_bank_sender: Sender<Vec<BankWithScheduler>>,
         block_metadata_notifier: Option<BlockMetadataNotifierArc>,
         log_messages_bytes_limit: Option<usize>,
         prioritization_fee_cache: Arc<PrioritizationFeeCache>,
@@ -1618,7 +1618,7 @@ impl ReplayStage {
 
         // Grab the Slot and BankId's of the banks we need to purge, then clear the banks
         // from BankForks
-        let (slots_to_purge, removed_banks): (Vec<(Slot, BankId)>, Vec<Arc<Bank>>) = {
+        let (slots_to_purge, removed_banks): (Vec<(Slot, BankId)>, Vec<BankWithScheduler>) = {
             let mut w_bank_forks = bank_forks.write().unwrap();
             slot_descendants
                 .iter()
@@ -2275,7 +2275,7 @@ impl ReplayStage {
         replay_timing: &mut ReplayTiming,
         voting_sender: &Sender<VoteOp>,
         epoch_slots_frozen_slots: &mut EpochSlotsFrozenSlots,
-        drop_bank_sender: &Sender<Vec<Arc<Bank>>>,
+        drop_bank_sender: &Sender<Vec<BankWithScheduler>>,
         wait_to_vote_slot: Option<Slot>,
     ) {
         if bank.is_empty() {
@@ -4093,7 +4093,7 @@ impl ReplayStage {
         has_new_vote_been_rooted: &mut bool,
         voted_signatures: &mut Vec<Signature>,
         epoch_slots_frozen_slots: &mut EpochSlotsFrozenSlots,
-        drop_bank_sender: &Sender<Vec<Arc<Bank>>>,
+        drop_bank_sender: &Sender<Vec<BankWithScheduler>>,
     ) {
         bank_forks.read().unwrap().prune_program_cache(new_root);
         let removed_banks = bank_forks.write().unwrap().set_root(

core/src/validator.rs

@@ -142,8 +142,8 @@ const WAIT_FOR_SUPERMAJORITY_THRESHOLD_PERCENT: u64 = 80;
 #[derive(Clone, EnumString, EnumVariantNames, Default, IntoStaticStr, Display)]
 #[strum(serialize_all = "kebab-case")]
 pub enum BlockVerificationMethod {
-    #[default]
     BlockstoreProcessor,
+    #[default]
     UnifiedScheduler,
 }
 
@@ -1434,8 +1434,10 @@ impl Validator {
 
     // Used for notifying many nodes in parallel to exit
     pub fn exit(&mut self) {
+        info!("exit1");
         self.validator_exit.write().unwrap().exit();
 
+        info!("exit2");
         // drop all signals in blockstore
         self.blockstore.drop_signal();
     }
@@ -1471,24 +1473,29 @@ impl Validator {
     }
 
     pub fn join(self) {
-        drop(self.bank_forks);
+        info!("join1");
         drop(self.cluster_info);
 
+        info!("join2");
         self.poh_service.join().expect("poh_service");
         drop(self.poh_recorder);
 
+        info!("join3");
         if let Some(json_rpc_service) = self.json_rpc_service {
             json_rpc_service.join().expect("rpc_service");
         }
 
+        info!("join4");
         if let Some(pubsub_service) = self.pubsub_service {
             pubsub_service.join().expect("pubsub_service");
         }
 
+        info!("join5");
         self.rpc_completed_slots_service
             .join()
             .expect("rpc_completed_slots_service");
 
+        info!("join6");
         if let Some(optimistically_confirmed_bank_tracker) =
             self.optimistically_confirmed_bank_tracker
         {
@@ -1497,96 +1504,126 @@ impl Validator {
                 .expect("optimistically_confirmed_bank_tracker");
         }
 
+        info!("join7");
         if let Some(transaction_status_service) = self.transaction_status_service {
             transaction_status_service
                 .join()
                 .expect("transaction_status_service");
         }
 
+        info!("join8");
         if let Some(rewards_recorder_service) = self.rewards_recorder_service {
             rewards_recorder_service
                 .join()
                 .expect("rewards_recorder_service");
         }
 
+        info!("join9");
         if let Some(cache_block_meta_service) = self.cache_block_meta_service {
             cache_block_meta_service
                 .join()
                 .expect("cache_block_meta_service");
         }
 
+        info!("join10");
         if let Some(system_monitor_service) = self.system_monitor_service {
             system_monitor_service
                 .join()
                 .expect("system_monitor_service");
         }
 
+        info!("join11");
         if let Some(sample_performance_service) = self.sample_performance_service {
             sample_performance_service
                 .join()
                 .expect("sample_performance_service");
         }
 
+        info!("join12");
         if let Some(entry_notifier_service) = self.entry_notifier_service {
             entry_notifier_service
                 .join()
                 .expect("entry_notifier_service");
         }
 
+        info!("join13");
         if let Some(s) = self.snapshot_packager_service {
             s.join().expect("snapshot_packager_service");
         }
 
+        info!("join14");
         self.gossip_service.join().expect("gossip_service");
         if let Some(repair_quic_endpoint) = &self.repair_quic_endpoint {
             repair::quic_endpoint::close_quic_endpoint(repair_quic_endpoint);
         }
+        info!("join15");
         self.serve_repair_service
             .join()
             .expect("serve_repair_service");
+        info!("join15");
         if let Some(repair_quic_endpoint_join_handle) = self.repair_quic_endpoint_join_handle {
             self.repair_quic_endpoint_runtime
                 .map(|runtime| runtime.block_on(repair_quic_endpoint_join_handle))
                 .transpose()
                 .unwrap();
         };
+        info!("join16");
         self.stats_reporter_service
             .join()
             .expect("stats_reporter_service");
+        info!("join17");
         self.blockstore_metric_report_service
             .join()
             .expect("ledger_metric_report_service");
+        info!("join18");
         self.accounts_background_service
             .join()
             .expect("accounts_background_service");
+        info!("join19");
         self.accounts_hash_verifier
             .join()
             .expect("accounts_hash_verifier");
+        info!("join20");
         if let Some(turbine_quic_endpoint) = &self.turbine_quic_endpoint {
             solana_turbine::quic_endpoint::close_quic_endpoint(turbine_quic_endpoint);
         }
+        info!("join21");
         self.tpu.join().expect("tpu");
+        info!("join22");
         self.tvu.join().expect("tvu");
+        info!("join23");
         if let Some(turbine_quic_endpoint_join_handle) = self.turbine_quic_endpoint_join_handle {
             self.turbine_quic_endpoint_runtime
                 .map(|runtime| runtime.block_on(turbine_quic_endpoint_join_handle))
                 .transpose()
                 .unwrap();
         }
+        info!("join24");
         self.completed_data_sets_service
             .join()
             .expect("completed_data_sets_service");
+        info!("join25");
         if let Some(ip_echo_server) = self.ip_echo_server {
             ip_echo_server.shutdown_background();
         }
 
+        info!("join26");
         if let Some(geyser_plugin_service) = self.geyser_plugin_service {
             geyser_plugin_service.join().expect("geyser_plugin_service");
         }
 
+        info!("join27");
         self.poh_timing_report_service
             .join()
             .expect("poh_timing_report_service");
+        info!("join28");
+        self.bank_forks.write().unwrap().prepare_to_drop();
+        let sc = Arc::strong_count(&self.bank_forks);
+        if let Some(bank_forks) = Arc::into_inner(self.bank_forks) {
+            drop::<BankForks>(bank_forks.into_inner().unwrap());
+        } else {
+            warn!("seems bankforks are leaking...{}:", sc);
+        }
     }
 }
 

ledger-tool/src/ledger_utils.rs

@@ -304,6 +304,8 @@ pub fn load_and_process_ledger(
             }
         }
         BlockVerificationMethod::UnifiedScheduler => {
+            let unified_scheduler_handler_threads =
+                value_t!(arg_matches, "unified_scheduler_handler_threads", usize).ok();
             let no_transaction_status_sender = None;
             let no_replay_vote_sender = None;
             let ignored_prioritization_fee_cache = Arc::new(PrioritizationFeeCache::new(0u64));

ledger-tool/src/main.rs

@@ -1654,6 +1654,8 @@ fn main() {
                     }
                     exit_signal.store(true, Ordering::Relaxed);
                     system_monitor_service.join().unwrap();
+                    bank_forks.write().unwrap().prepare_to_drop();
+                    drop::<BankForks>(Arc::into_inner(bank_forks).unwrap().into_inner().unwrap());
                 }
                 ("graph", Some(arg_matches)) => {
                     let output_file = value_t_or_exit!(arg_matches, "graph_filename", String);

ledger/src/blockstore_processor.rs

@@ -338,8 +338,7 @@ fn process_batches(
         // scheduling always succeeds here without being blocked on actual transaction executions.
         // The transaction execution errors will be collected via the blocking fn called
         // BankWithScheduler::wait_for_completed_scheduler(), if any.
-        schedule_batches_for_execution(bank, batches);
-        Ok(())
+        schedule_batches_for_execution(bank, batches)
     } else {
         debug!(
             "process_batches()/rebatch_and_execute_batches({} batches)",
@@ -360,7 +359,7 @@ fn process_batches(
 fn schedule_batches_for_execution(
     bank: &BankWithScheduler,
     batches: &[TransactionBatchWithIndexes],
-) {
+) -> Result<()> {
     for TransactionBatchWithIndexes {
         batch,
         transaction_indexes,
@@ -371,8 +370,9 @@ fn schedule_batches_for_execution(
                 .sanitized_transactions()
                 .iter()
                 .zip(transaction_indexes.iter()),
-        );
+        )?;
     }
+    Ok(())
 }
 
 fn rebatch_transactions<'a>(
@@ -440,9 +440,7 @@ fn rebatch_and_execute_batches(
     {
         let mut cost_tracker = bank.write_cost_tracker().unwrap();
         for tx_cost in &tx_costs {
-            cost_tracker
-                .try_add(tx_cost)
-                .map_err(TransactionError::from)?;
+            cost_tracker.try_add(tx_cost)?;
         }
     }
 
@@ -1960,6 +1958,7 @@ pub mod tests {
             instruction::{Instruction, InstructionError},
             native_token::LAMPORTS_PER_SOL,
             pubkey::Pubkey,
+            scheduling::SchedulingMode,
             signature::{Keypair, Signer},
             system_instruction::SystemError,
             system_transaction,
@@ -4544,7 +4543,7 @@ pub mod tests {
             ..
         } = create_genesis_config_with_leader(500, &dummy_leader_pubkey, 100);
         let bank = Arc::new(Bank::new_for_tests(&genesis_config));
-        let context = SchedulingContext::new(bank.clone());
+        let context = SchedulingContext::new(SchedulingMode::BlockVerification, bank.clone());
 
         let txs = create_test_transactions(&mint_keypair, &genesis_config.hash());
 
@@ -4559,7 +4558,7 @@ pub mod tests {
         mocked_scheduler
             .expect_schedule_execution()
             .times(txs.len())
-            .returning(|_| ());
+            .returning(|_| Ok(()));
         mocked_scheduler
             .expect_wait_for_termination()
             .with(mockall::predicate::eq(true))

local-cluster/tests/local_cluster.rs

@@ -4,7 +4,7 @@ use {
     crossbeam_channel::{unbounded, Receiver},
     gag::BufferRedirect,
     log::*,
-    rand::seq::IteratorRandom,
+    rand::seq::SliceRandom,
     serial_test::serial,
     solana_accounts_db::{
         hardened_unpack::open_genesis_config, utils::create_accounts_run_and_snapshot_dirs,
@@ -5499,12 +5499,14 @@ fn test_randomly_mixed_block_verification_methods_between_bootstrap_and_not() {
     );
 
     // Randomly switch to use unified scheduler
-    config
-        .validator_configs
-        .iter_mut()
-        .choose(&mut rand::thread_rng())
-        .unwrap()
-        .block_verification_method = BlockVerificationMethod::UnifiedScheduler;
+    let mut methods = [
+        BlockVerificationMethod::UnifiedScheduler,
+        BlockVerificationMethod::BlockstoreProcessor,
+    ];
+    methods.shuffle(&mut rand::thread_rng());
+    for (validator_config, method) in config.validator_configs.iter_mut().zip(methods) {
+        validator_config.block_verification_method = method;
+    }
 
     let local = LocalCluster::new(&mut config, SocketAddrSpace::Unspecified);
     cluster_tests::spend_and_verify_all_nodes(