New exec code path rebased for agave

Cargo.toml

@@ -181,9 +181,10 @@ 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.12"
+crossbeam-channel = "0.5.11"
 csv = "1.3.0"
 ctrlc = "3.4.4"
 curve25519-dalek = "3.2.1"
@@ -269,6 +270,7 @@ predicates = "2.1"
 pretty-hex = "0.3.0"
 prio-graph = "0.2.1"
 proc-macro2 = "1.0.79"
+procfs = "0.16.0"
 proptest = "1.4"
 prost = "0.11.9"
 prost-build = "0.11.9"
@@ -289,6 +291,7 @@ reqwest = { version = "0.11.23", default-features = false }
 rolling-file = "0.2.0"
 rpassword = "7.3"
 rustc_version = "0.4"
+rustix = "0.38.32"
 rustls = { version = "0.21.11", default-features = false, features = ["quic"] }
 rustversion = "1.0.14"
 scopeguard = "1.2.0"
@@ -443,6 +446,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/validator.rs

@@ -149,8 +149,8 @@ const WAIT_FOR_WEN_RESTART_SUPERMAJORITY_THRESHOLD_PERCENT: u64 =
 #[derive(Clone, EnumString, EnumVariantNames, Default, IntoStaticStr, Display)]
 #[strum(serialize_all = "kebab-case")]
 pub enum BlockVerificationMethod {
-    #[default]
     BlockstoreProcessor,
+    #[default]
     UnifiedScheduler,
 }
 
@@ -1462,8 +1462,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();
     }
@@ -1499,24 +1501,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
         {
@@ -1525,96 +1532,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);
+        }
     }
 }
 

core/tests/unified_scheduler.rs

@@ -18,7 +18,8 @@ use {
     solana_program_runtime::timings::ExecuteTimings,
     solana_runtime::{
         accounts_background_service::AbsRequestSender, bank::Bank, bank_forks::BankForks,
-        genesis_utils::GenesisConfigInfo, prioritization_fee_cache::PrioritizationFeeCache,
+        genesis_utils::GenesisConfigInfo, installed_scheduler_pool::DefaultScheduleExecutionArg,
+        prioritization_fee_cache::PrioritizationFeeCache,
     },
     solana_sdk::{
         hash::Hash,
@@ -27,11 +28,14 @@ use {
         transaction::{Result, SanitizedTransaction},
     },
     solana_unified_scheduler_pool::{
-        DefaultTaskHandler, HandlerContext, PooledScheduler, SchedulerPool, TaskHandler,
+        DefaultTaskHandler, HandlerContext, PooledScheduler, SchedulerPool, SpawnableScheduler,
+        TaskHandler,
     },
     std::{
         collections::HashMap,
         sync::{Arc, Mutex},
+        thread::sleep,
+        time::Duration,
     },
 };
 
@@ -41,10 +45,11 @@ fn test_scheduler_waited_by_drop_bank_service() {
 
     static LOCK_TO_STALL: Mutex<()> = Mutex::new(());
 
-    #[derive(Debug)]
+    #[derive(Clone, Debug)]
     struct StallingHandler;
-    impl TaskHandler for StallingHandler {
+    impl TaskHandler<DefaultScheduleExecutionArg> for StallingHandler {
         fn handle(
+            &self,
             result: &mut Result<()>,
             timings: &mut ExecuteTimings,
             bank: &Arc<Bank>,
@@ -55,10 +60,24 @@ fn test_scheduler_waited_by_drop_bank_service() {
             info!("Stalling at StallingHandler::handle()...");
             *LOCK_TO_STALL.lock().unwrap();
             // Wait a bit for the replay stage to prune banks
-            std::thread::sleep(std::time::Duration::from_secs(3));
+            sleep(Duration::from_secs(3));
             info!("Now entering into DefaultTaskHandler::handle()...");
 
-            DefaultTaskHandler::handle(result, timings, bank, transaction, index, handler_context);
+            <DefaultTaskHandler as TaskHandler<DefaultScheduleExecutionArg>>::handle(
+                &DefaultTaskHandler,
+                result,
+                timings,
+                bank,
+                transaction,
+                index,
+                handler_context,
+            );
+        }
+
+        fn create<T: SpawnableScheduler<Self, DefaultScheduleExecutionArg>>(
+            _pool: &SchedulerPool<T, Self, DefaultScheduleExecutionArg>,
+        ) -> Self {
+            Self
         }
     }
 
@@ -72,7 +91,7 @@ fn test_scheduler_waited_by_drop_bank_service() {
     let genesis_bank = Bank::new_for_tests(&genesis_config);
     let bank_forks = BankForks::new_rw_arc(genesis_bank);
     let ignored_prioritization_fee_cache = Arc::new(PrioritizationFeeCache::new(0u64));
-    let pool_raw = SchedulerPool::<PooledScheduler<StallingHandler>, _>::new(
+    let pool_raw = SchedulerPool::<PooledScheduler<StallingHandler, _>, _, _>::new(
         None,
         None,
         None,
@@ -107,7 +126,9 @@ fn test_scheduler_waited_by_drop_bank_service() {
     // Delay transaction execution to ensure transaction execution happens after termintion has
     // been started
     let lock_to_stall = LOCK_TO_STALL.lock().unwrap();
-    pruned_bank.schedule_transaction_executions([(&tx, &0)].into_iter());
+    pruned_bank
+        .schedule_transaction_executions([(&tx, &0)].into_iter())
+        .unwrap();
     drop(pruned_bank);
     assert_eq!(pool_raw.pooled_scheduler_count(), 0);
     drop(lock_to_stall);

ledger-tool/src/main.rs

@@ -864,7 +864,6 @@ fn main() {
                 .takes_value(true)
                 .possible_values(BlockVerificationMethod::cli_names())
                 .global(true)
-                .hidden(hidden_unless_forced())
                 .help(BlockVerificationMethod::cli_message()),
         )
         .arg(
@@ -874,7 +873,6 @@ fn main() {
                 .takes_value(true)
                 .validator(|s| is_within_range(s, 1..))
                 .global(true)
-                .hidden(hidden_unless_forced())
                 .help(DefaultSchedulerPool::cli_message()),
         )
         .arg(
@@ -1823,6 +1821,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);