New exec code path rebased

ledger/src/blockstore_processor.rs

@@ -1945,9 +1945,7 @@ pub mod tests {
             genesis_utils::{
                 self, create_genesis_config_with_vote_accounts, ValidatorVoteKeypairs,
             },
-            installed_scheduler_pool::{
-                MockInstalledScheduler, MockInstalledSchedulerPool, WaitReason,
-            },
+            installed_scheduler_pool::{MockInstalledScheduler, WaitReason},
         },
         solana_sdk::{
             account::{AccountSharedData, WritableAccount},
@@ -4550,18 +4548,10 @@ pub mod tests {
             .in_sequence(&mut seq)
             .returning(|_| None);
         mocked_scheduler
-            .expect_pool()
+            .expect_return_to_pool()
             .times(1)
             .in_sequence(&mut seq)
-            .returning(move || {
-                let mut mocked_pool = MockInstalledSchedulerPool::new();
-                mocked_pool
-                    .expect_return_to_pool()
-                    .times(1)
-                    .in_sequence(&mut seq)
-                    .returning(|_| ());
-                Arc::new(mocked_pool)
-            });
+            .returning(|| ());
         let bank = BankWithScheduler::new(bank, Some(Box::new(mocked_scheduler)));
 
         let batch = bank.prepare_sanitized_batch(&txs);

runtime/src/bank_forks.rs

@@ -239,7 +239,7 @@ impl BankForks {
         let bank = Arc::new(bank);
         let bank = if let Some(scheduler_pool) = &self.scheduler_pool {
             let context = SchedulingContext::new(SchedulingMode::BlockVerification, bank.clone());
-            let scheduler = scheduler_pool.take_from_pool(context);
+            let scheduler = scheduler_pool.take_scheduler(context);
             BankWithScheduler::new(bank, Some(scheduler))
         } else {
             BankWithScheduler::new_without_scheduler(bank)

runtime/src/installed_scheduler_pool.rs

@@ -43,10 +43,8 @@ use {
 #[cfg(feature = "dev-context-only-utils")]
 use {mockall::automock, qualifier_attr::qualifiers};
 
-#[cfg_attr(feature = "dev-context-only-utils", automock)]
 pub trait InstalledSchedulerPool<SEA: ScheduleExecutionArg>: Send + Sync + Debug {
-    fn take_from_pool(&self, context: SchedulingContext) -> Box<dyn InstalledScheduler<SEA>>;
-    fn return_to_pool(&self, scheduler: Box<dyn InstalledScheduler<SEA>>);
+    fn take_scheduler(&self, context: SchedulingContext) -> Box<dyn InstalledScheduler<SEA>>;
 }
 
 #[cfg_attr(doc, aquamarine::aquamarine)]
@@ -111,7 +109,7 @@ pub trait InstalledSchedulerPool<SEA: ScheduleExecutionArg>: Send + Sync + Debug
 )]
 pub trait InstalledScheduler<SEA: ScheduleExecutionArg>: Send + Sync + Debug + 'static {
     fn id(&self) -> SchedulerId;
-    fn pool(&self) -> InstalledSchedulerPoolArc<SEA>;
+    fn return_to_pool(self: Box<Self>);
 
     // Calling this is illegal as soon as wait_for_termination is called.
     fn schedule_execution<'a>(&'a self, transaction_with_index: SEA::TransactionWithIndex<'a>);
@@ -400,7 +398,7 @@ impl BankWithSchedulerInner {
                 .and_then(|scheduler| scheduler.wait_for_termination(&reason));
             if !reason.is_paused() {
                 let scheduler = scheduler.take().expect("scheduler after waiting");
-                scheduler.pool().return_to_pool(scheduler);
+                scheduler.return_to_pool();
             }
             result_with_timings
         } else {
@@ -466,17 +464,6 @@ mod tests {
         solana_sdk::system_transaction,
     };
 
-    fn setup_mocked_scheduler_pool(
-        seq: &mut Sequence,
-    ) -> InstalledSchedulerPoolArc<DefaultScheduleExecutionArg> {
-        let mut mock = MockInstalledSchedulerPool::new();
-        mock.expect_return_to_pool()
-            .times(1)
-            .in_sequence(seq)
-            .returning(|_| ());
-        Arc::new(mock)
-    }
-
     fn setup_mocked_scheduler_with_extra(
         wait_reasons: impl Iterator<Item = WaitReason>,
         f: Option<impl Fn(&mut MockInstalledScheduler<DefaultScheduleExecutionArg>)>,
@@ -503,10 +490,10 @@ mod tests {
                 });
         }
 
-        mock.expect_pool()
+        mock.expect_return_to_pool()
             .times(1)
             .in_sequence(&mut seq)
-            .returning(move || setup_mocked_scheduler_pool(&mut seq));
+            .returning(|| ());
         if let Some(f) = f {
             f(&mut mock);
         }

scheduler-pool/benches/scheduler.rs

@@ -15,8 +15,8 @@ use {
         bank::Bank,
         genesis_utils::{create_genesis_config, GenesisConfigInfo},
         installed_scheduler_pool::{
-            InstalledScheduler, InstalledSchedulerPoolArc, ResultWithTimings, ScheduleExecutionArg,
-            SchedulerId, SchedulingContext, WaitReason, WithTransactionAndIndex,
+            InstalledScheduler, ResultWithTimings, ScheduleExecutionArg, SchedulerId,
+            SchedulingContext, WaitReason, WithTransactionAndIndex,
         },
         prioritization_fee_cache::PrioritizationFeeCache,
     },
@@ -209,7 +209,7 @@ mod nonblocking {
         H: ScheduledTransactionHandler<ScheduleExecutionArgForBench> + Clone,
     > {
         id: SchedulerId,
-        pool: Arc<SchedulerPool<Self, H, ScheduleExecutionArgForBench>>,
+        pub(crate) pool: Arc<SchedulerPool<Self, H, ScheduleExecutionArgForBench>>,
         transaction_sender: crossbeam_channel::Sender<ChainedChannel>,
         result_receiver: crossbeam_channel::Receiver<(Result<()>, ExecuteTimings, usize)>,
         lane_count: usize,
@@ -330,8 +330,8 @@ mod nonblocking {
             self.id
         }
 
-        fn pool(&self) -> InstalledSchedulerPoolArc<ScheduleExecutionArgForBench> {
-            self.pool.clone()
+        fn return_to_pool(self: Box<Self>) {
+            self.pool.clone().return_scheduler(self)
         }
 
         fn schedule_execution(&self, transaction_with_index: TransactionWithIndexForBench) {
@@ -673,8 +673,8 @@ mod thread_utilization {
             self.inner_scheduler.id()
         }
 
-        fn pool(&self) -> InstalledSchedulerPoolArc<ScheduleExecutionArgForBench> {
-            self.inner_scheduler.pool()
+        fn return_to_pool(self: Box<Self>) {
+            self.inner_scheduler.pool.clone().return_scheduler(self)
         }
 
         fn context(&self) -> Option<&SchedulingContext> {

scheduler-pool/src/lib.rs

@@ -47,7 +47,7 @@ pub struct SchedulerPool<
     transaction_status_sender: Option<TransactionStatusSender>,
     replay_vote_sender: Option<ReplayVoteSender>,
     prioritization_fee_cache: Arc<PrioritizationFeeCache>,
-    // weak_self could be elided by changing InstalledScheduler::take_from_pool()'s receiver to
+    // weak_self could be elided by changing InstalledScheduler::take_scheduler()'s receiver to
     // Arc<Self> from &Self, because SchedulerPool is used as in the form of Arc<SchedulerPool>
     // almost always. But, this would cause wasted and noisy Arc::clone()'s at every call sites.
     //
@@ -110,6 +110,15 @@ impl<
             .upgrade()
             .expect("self-referencing Arc-ed pool")
     }
+
+    pub fn return_scheduler(&self, scheduler: Box<dyn InstalledScheduler<SEA>>) {
+        assert!(scheduler.context().is_none());
+
+        self.schedulers
+            .lock()
+            .expect("not poisoned")
+            .push(scheduler);
+    }
 }
 
 impl<
@@ -118,7 +127,7 @@ impl<
         SEA: ScheduleExecutionArg,
     > InstalledSchedulerPool<SEA> for SchedulerPool<T, TH, SEA>
 {
-    fn take_from_pool(&self, context: SchedulingContext) -> Box<dyn InstalledScheduler<SEA>> {
+    fn take_scheduler(&self, context: SchedulingContext) -> Box<dyn InstalledScheduler<SEA>> {
         // pop is intentional for filo, expecting relatively warmed-up scheduler due to having been
         // returned recently
         if let Some(mut scheduler) = self.schedulers.lock().expect("not poisoned").pop() {
@@ -128,15 +137,6 @@ impl<
             T::spawn_boxed(self.self_arc(), context, TH::create(self))
         }
     }
-
-    fn return_to_pool(&self, scheduler: Box<dyn InstalledScheduler<SEA>>) {
-        assert!(scheduler.context().is_none());
-
-        self.schedulers
-            .lock()
-            .expect("not poisoned")
-            .push(scheduler);
-    }
 }
 
 pub trait ScheduledTransactionHandler<SEA: ScheduleExecutionArg>:
@@ -253,8 +253,8 @@ impl<TH: ScheduledTransactionHandler<SEA>, SEA: ScheduleExecutionArg> InstalledS
         self.id
     }
 
-    fn pool(&self) -> InstalledSchedulerPoolArc<SEA> {
-        self.pool.clone()
+    fn return_to_pool(self: Box<Self>) {
+        self.pool.clone().return_scheduler(self)
     }
 
     fn schedule_execution(&self, transaction_with_index: SEA::TransactionWithIndex<'_>) {
@@ -361,7 +361,7 @@ mod tests {
             DefaultSchedulerPool::new_dyn(None, None, None, ignored_prioritization_fee_cache);
         let bank = Arc::new(Bank::default_for_tests());
         let context = SchedulingContext::new(SchedulingMode::BlockVerification, bank);
-        let scheduler = pool.take_from_pool(context);
+        let scheduler = pool.take_scheduler(context);
 
         let debug = format!("{scheduler:#?}");
         assert!(!debug.is_empty());
@@ -372,31 +372,30 @@ mod tests {
         solana_logger::setup();
 
         let ignored_prioritization_fee_cache = Arc::new(PrioritizationFeeCache::new(0u64));
-        let pool =
-            DefaultSchedulerPool::new_dyn(None, None, None, ignored_prioritization_fee_cache);
+        let pool = DefaultSchedulerPool::new(None, None, None, ignored_prioritization_fee_cache);
         let bank = Arc::new(Bank::default_for_tests());
         let context = &SchedulingContext::new(SchedulingMode::BlockVerification, bank);
 
-        let mut scheduler1 = pool.take_from_pool(context.clone());
+        let mut scheduler1 = pool.take_scheduler(context.clone());
         let scheduler_id1 = scheduler1.id();
-        let mut scheduler2 = pool.take_from_pool(context.clone());
+        let mut scheduler2 = pool.take_scheduler(context.clone());
         let scheduler_id2 = scheduler2.id();
         assert_ne!(scheduler_id1, scheduler_id2);
 
         assert_matches!(
             scheduler1.wait_for_termination(&WaitReason::TerminatedToFreeze),
             None
         );
-        pool.return_to_pool(scheduler1);
+        pool.return_scheduler(scheduler1);
         assert_matches!(
             scheduler2.wait_for_termination(&WaitReason::TerminatedToFreeze),
             None
         );
-        pool.return_to_pool(scheduler2);
+        pool.return_scheduler(scheduler2);
 
-        let scheduler3 = pool.take_from_pool(context.clone());
+        let scheduler3 = pool.take_scheduler(context.clone());
         assert_eq!(scheduler_id2, scheduler3.id());
-        let scheduler4 = pool.take_from_pool(context.clone());
+        let scheduler4 = pool.take_scheduler(context.clone());
         assert_eq!(scheduler_id1, scheduler4.id());
     }
 
@@ -410,7 +409,7 @@ mod tests {
         let bank = Arc::new(Bank::default_for_tests());
         let context = &SchedulingContext::new(SchedulingMode::BlockVerification, bank);
 
-        let mut scheduler = pool.take_from_pool(context.clone());
+        let mut scheduler = pool.take_scheduler(context.clone());
 
         assert!(scheduler.context().is_some());
         assert_matches!(
@@ -430,8 +429,7 @@ mod tests {
         solana_logger::setup();
 
         let ignored_prioritization_fee_cache = Arc::new(PrioritizationFeeCache::new(0u64));
-        let pool =
-            DefaultSchedulerPool::new_dyn(None, None, None, ignored_prioritization_fee_cache);
+        let pool = DefaultSchedulerPool::new(None, None, None, ignored_prioritization_fee_cache);
         let old_bank = &Arc::new(Bank::default_for_tests());
         let new_bank = &Arc::new(Bank::default_for_tests());
         assert!(!Arc::ptr_eq(old_bank, new_bank));
@@ -441,15 +439,15 @@ mod tests {
         let new_context =
             &SchedulingContext::new(SchedulingMode::BlockVerification, new_bank.clone());
 
-        let mut scheduler = pool.take_from_pool(old_context.clone());
+        let mut scheduler = pool.take_scheduler(old_context.clone());
         let scheduler_id = scheduler.id();
         assert_matches!(
             scheduler.wait_for_termination(&WaitReason::TerminatedToFreeze),
             None
         );
-        pool.return_to_pool(scheduler);
+        pool.return_scheduler(scheduler);
 
-        let scheduler = pool.take_from_pool(new_context.clone());
+        let scheduler = pool.take_scheduler(new_context.clone());
         assert_eq!(scheduler_id, scheduler.id());
         assert!(Arc::ptr_eq(scheduler.context().unwrap().bank(), new_bank));
     }
@@ -522,7 +520,7 @@ mod tests {
         let context = SchedulingContext::new(SchedulingMode::BlockVerification, bank.clone());
 
         assert_eq!(bank.transaction_count(), 0);
-        let scheduler = pool.take_from_pool(context);
+        let scheduler = pool.take_scheduler(context);
         scheduler.schedule_execution(&(tx0, 0));
         let bank = BankWithScheduler::new(bank, Some(scheduler));
         assert_matches!(bank.wait_for_completed_scheduler(), Some((Ok(()), _)));
@@ -552,7 +550,7 @@ mod tests {
         let context = SchedulingContext::new(SchedulingMode::BlockVerification, bank.clone());
 
         assert_eq!(bank.transaction_count(), 0);
-        let scheduler = pool.take_from_pool(context);
+        let scheduler = pool.take_scheduler(context);
         scheduler.schedule_execution(&(tx0, 0));
         assert_eq!(bank.transaction_count(), 0);
 
@@ -593,8 +591,8 @@ mod tests {
             self.0.id()
         }
 
-        fn pool(&self) -> InstalledSchedulerPoolArc<DefaultScheduleExecutionArg> {
-            self.0.pool()
+        fn return_to_pool(self: Box<Self>) {
+            self.0.pool.clone().return_scheduler(self)
         }
 
         fn context(&self) -> Option<&SchedulingContext> {
@@ -722,7 +720,7 @@ mod tests {
             DefaultTransactionHandler,
             DefaultScheduleExecutionArg,
         >::new_dyn(None, None, None, ignored_prioritization_fee_cache);
-        let scheduler = pool.take_from_pool(context);
+        let scheduler = pool.take_scheduler(context);
 
         let bank = BankWithScheduler::new(bank, Some(scheduler));
         assert_eq!(bank.transaction_count(), 0);