Scheduler: slot-gated Retry

core/src/banking_stage/consume_worker.rs

@@ -1,6 +1,9 @@
 use {
     super::{
-        consumer::{Consumer, ExecuteAndCommitTransactionsOutput, ProcessTransactionBatchOutput},
+        consumer::{
+            Consumer, ExecuteAndCommitTransactionsOutput, ProcessTransactionBatchOutput,
+            RetryableIndexKind,
+        },
         leader_slot_timing_metrics::LeaderExecuteAndCommitTimings,
         scheduler_messages::{ConsumeWork, FinishedConsumeWork},
     },
@@ -122,6 +125,7 @@ impl<Tx: TransactionWithMeta> ConsumeWorker<Tx> {
         self.metrics.has_data.store(true, Ordering::Relaxed);
 
         self.consumed_sender.send(FinishedConsumeWork {
+            slot: Some(bank.slot()),
             work,
             retryable_indexes: output
                 .execute_and_commit_transactions_output
@@ -152,7 +156,9 @@ impl<Tx: TransactionWithMeta> ConsumeWorker<Tx> {
 
     /// Send transactions back to scheduler as retryable.
     fn retry(&self, work: ConsumeWork<Tx>) -> Result<(), ConsumeWorkerError<Tx>> {
-        let retryable_indexes: Vec<_> = (0..work.transactions.len()).collect();
+        let retryable_indexes: Vec<_> = (0..work.transactions.len())
+            .map(RetryableIndexKind::InvalidBank)
+            .collect();
         let num_retryable = retryable_indexes.len();
         self.metrics
             .count_metrics
@@ -164,6 +170,7 @@ impl<Tx: TransactionWithMeta> ConsumeWorker<Tx> {
             .fetch_add(num_retryable, Ordering::Relaxed);
         self.metrics.has_data.store(true, Ordering::Relaxed);
         self.consumed_sender.send(FinishedConsumeWork {
+            slot: None,
             work,
             retryable_indexes,
         })?;
@@ -907,7 +914,10 @@ mod tests {
         assert_eq!(consumed.work.batch_id, bid);
         assert_eq!(consumed.work.ids, vec![id]);
         assert_eq!(consumed.work.max_ages, vec![max_age]);
-        assert_eq!(consumed.retryable_indexes, vec![0]);
+        assert_eq!(
+            consumed.retryable_indexes,
+            vec![RetryableIndexKind::InvalidBank(0)]
+        );
 
         drop(test_frame);
         let _ = worker_thread.join().unwrap();
@@ -956,7 +966,7 @@ mod tests {
         assert_eq!(consumed.work.batch_id, bid);
         assert_eq!(consumed.work.ids, vec![id]);
         assert_eq!(consumed.work.max_ages, vec![max_age]);
-        assert_eq!(consumed.retryable_indexes, Vec::<usize>::new());
+        assert_eq!(consumed.retryable_indexes, Vec::new());
 
         drop(test_frame);
         let _ = worker_thread.join().unwrap();
@@ -1008,7 +1018,10 @@ mod tests {
         assert_eq!(consumed.work.batch_id, bid);
         assert_eq!(consumed.work.ids, vec![id1, id2]);
         assert_eq!(consumed.work.max_ages, vec![max_age, max_age]);
-        assert_eq!(consumed.retryable_indexes, vec![1]); // id2 is retryable since lock conflict
+        assert_eq!(
+            consumed.retryable_indexes,
+            vec![RetryableIndexKind::AccountInUse(1)]
+        ); // id2 is retryable since lock conflict
 
         drop(test_frame);
         let _ = worker_thread.join().unwrap();
@@ -1077,13 +1090,13 @@ mod tests {
         assert_eq!(consumed.work.batch_id, bid1);
         assert_eq!(consumed.work.ids, vec![id1]);
         assert_eq!(consumed.work.max_ages, vec![max_age]);
-        assert_eq!(consumed.retryable_indexes, Vec::<usize>::new());
+        assert_eq!(consumed.retryable_indexes, Vec::new());
 
         let consumed = consumed_receiver.recv().unwrap();
         assert_eq!(consumed.work.batch_id, bid2);
         assert_eq!(consumed.work.ids, vec![id2]);
         assert_eq!(consumed.work.max_ages, vec![max_age]);
-        assert_eq!(consumed.retryable_indexes, Vec::<usize>::new());
+        assert_eq!(consumed.retryable_indexes, Vec::new());
 
         drop(test_frame);
         let _ = worker_thread.join().unwrap();
@@ -1213,7 +1226,7 @@ mod tests {
             .unwrap();
 
         let consumed = consumed_receiver.recv().unwrap();
-        assert_eq!(consumed.retryable_indexes, Vec::<usize>::new());
+        assert_eq!(consumed.retryable_indexes, Vec::new());
         // all but one succeed. 6 for initial funding
         assert_eq!(bank.transaction_count(), 6 + 5);
 

core/src/banking_stage/consumer.rs

@@ -34,6 +34,16 @@ use {
 /// Consumer will create chunks of transactions from buffer with up to this size.
 pub const TARGET_NUM_TRANSACTIONS_PER_BATCH: usize = 64;
 
+#[cfg_attr(test, derive(Debug, PartialEq, Eq, PartialOrd, Ord))]
+pub enum RetryableIndexKind {
+    /// Retryable index due to account lock failures (Jito)
+    AccountInUse(usize),
+    /// Retryable index due to block-limits.
+    BlockLimits(usize),
+    /// Retryable index due to recording failure or missing bank i.e. block ended during execution.
+    InvalidBank(usize),
+}
+
 pub struct ProcessTransactionBatchOutput {
     // The number of transactions filtered out by the cost model
     pub(crate) cost_model_throttled_transactions_count: u64,
@@ -48,7 +58,7 @@ pub struct ExecuteAndCommitTransactionsOutput {
     pub(crate) transaction_counts: LeaderProcessedTransactionCounts,
     // Transactions that either were not executed, or were executed and failed to be committed due
     // to the block ending.
-    pub(crate) retryable_transaction_indexes: Vec<usize>,
+    pub(crate) retryable_transaction_indexes: Vec<RetryableIndexKind>,
     // A result that indicates whether transactions were successfully
     // committed into the Poh stream.
     pub commit_transactions_result: Result<Vec<CommitTransactionDetails>, PohRecorderError>,
@@ -274,23 +284,23 @@ impl Consumer {
                 // following are retryable errors
                 Err(TransactionError::AccountInUse) => {
                     error_counters.account_in_use += 1;
-                    Some(index)
+                    Some(RetryableIndexKind::AccountInUse(index))
                 }
                 Err(TransactionError::WouldExceedMaxBlockCostLimit) => {
                     error_counters.would_exceed_max_block_cost_limit += 1;
-                    Some(index)
+                    Some(RetryableIndexKind::BlockLimits(index))
                 }
                 Err(TransactionError::WouldExceedMaxVoteCostLimit) => {
                     error_counters.would_exceed_max_vote_cost_limit += 1;
-                    Some(index)
+                    Some(RetryableIndexKind::BlockLimits(index))
                 }
                 Err(TransactionError::WouldExceedMaxAccountCostLimit) => {
                     error_counters.would_exceed_max_account_cost_limit += 1;
-                    Some(index)
+                    Some(RetryableIndexKind::BlockLimits(index))
                 }
                 Err(TransactionError::WouldExceedAccountDataBlockLimit) => {
                     error_counters.would_exceed_account_data_block_limit += 1;
-                    Some(index)
+                    Some(RetryableIndexKind::BlockLimits(index))
                 }
                 // following are non-retryable errors
                 Err(TransactionError::TooManyAccountLocks) => {
@@ -367,7 +377,11 @@ impl Consumer {
 
         if let Err(recorder_err) = record_transactions_result {
             retryable_transaction_indexes.extend(processing_results.iter().enumerate().filter_map(
-                |(index, processing_result)| processing_result.was_processed().then_some(index),
+                |(index, processing_result)| {
+                    processing_result
+                        .was_processed()
+                        .then_some(RetryableIndexKind::InvalidBank(index))
+                },
             ));
 
             return ExecuteAndCommitTransactionsOutput {
@@ -756,7 +770,10 @@ mod tests {
                 processed_with_successful_result_count: 1,
             }
         );
-        assert_eq!(retryable_transaction_indexes, vec![0]);
+        assert_eq!(
+            retryable_transaction_indexes,
+            vec![RetryableIndexKind::InvalidBank(0)]
+        );
         assert_matches!(
             commit_transactions_result,
             Err(PohRecorderError::MaxHeightReached)
@@ -1131,7 +1148,10 @@ mod tests {
             commit_transactions_result.get(1),
             Some(CommitTransactionDetails::NotCommitted)
         );
-        assert_eq!(retryable_transaction_indexes, vec![1]);
+        assert_eq!(
+            retryable_transaction_indexes,
+            vec![RetryableIndexKind::AccountInUse(1)]
+        );
 
         let expected_block_cost = {
             let (actual_programs_execution_cost, actual_loaded_accounts_data_size_cost) =
@@ -1246,7 +1266,10 @@ mod tests {
                 processed_with_successful_result_count: 1,
             }
         );
-        assert_eq!(retryable_transaction_indexes, vec![1]);
+        assert_eq!(
+            retryable_transaction_indexes,
+            vec![RetryableIndexKind::AccountInUse(1)]
+        );
         assert!(commit_transactions_result.is_ok());
     }
 
@@ -1302,7 +1325,9 @@ mod tests {
 
         assert_eq!(
             execute_and_commit_transactions_output.retryable_transaction_indexes,
-            (1..transactions_len - 1).collect::<Vec<usize>>()
+            (1..transactions_len - 1)
+                .map(RetryableIndexKind::InvalidBank)
+                .collect::<Vec<_>>()
         );
     }
 
@@ -1360,7 +1385,9 @@ mod tests {
         // Everything except first of the transactions failed and are retryable
         assert_eq!(
             execute_and_commit_transactions_output.retryable_transaction_indexes,
-            (1..transactions_len).collect::<Vec<usize>>()
+            (1..transactions_len)
+                .map(RetryableIndexKind::AccountInUse)
+                .collect::<Vec<_>>()
         );
     }
 
@@ -1447,7 +1474,9 @@ mod tests {
         execute_and_commit_transactions_output
             .retryable_transaction_indexes
             .sort_unstable();
-        let expected: Vec<usize> = (0..transactions.len()).collect();
+        let expected = (0..transactions.len())
+            .map(RetryableIndexKind::InvalidBank)
+            .collect::<Vec<_>>();
         assert_eq!(
             execute_and_commit_transactions_output.retryable_transaction_indexes,
             expected

core/src/banking_stage/scheduler_messages.rs

@@ -1,4 +1,5 @@
 use {
+    super::consumer::RetryableIndexKind,
     solana_sdk::clock::{Epoch, Slot},
     std::fmt::Display,
 };
@@ -46,6 +47,8 @@ pub struct ConsumeWork<Tx> {
 /// Message: [Worker -> Scheduler]
 /// Processed transactions.
 pub struct FinishedConsumeWork<Tx> {
+    /// Slot that work was attempted on. `None` if sent back without attempt.
+    pub slot: Option<Slot>,
     pub work: ConsumeWork<Tx>,
-    pub retryable_indexes: Vec<usize>,
+    pub retryable_indexes: Vec<RetryableIndexKind>,
 }

core/src/banking_stage/transaction_scheduler/greedy_scheduler.rs

@@ -98,6 +98,7 @@ impl<Tx: TransactionWithMeta> Scheduler<Tx> for GreedyScheduler<Tx> {
         let mut num_sent: usize = 0;
         let mut num_unschedulable_conflicts: usize = 0;
         let mut num_unschedulable_threads: usize = 0;
+        let mut num_skipped_retry: usize = 0;
 
         let mut batches = Batches::new(num_threads, self.config.target_transactions_per_batch);
         while num_scanned < self.config.max_scanned_transactions_per_scheduling_pass
@@ -152,6 +153,10 @@ impl<Tx: TransactionWithMeta> Scheduler<Tx> for GreedyScheduler<Tx> {
                     num_unschedulable_threads += 1;
                     self.unschedulables.push(id);
                 }
+                Err(TransactionSchedulingError::Skipped) => {
+                    num_skipped_retry += 1;
+                    self.unschedulables.push(id);
+                }
                 Ok(TransactionSchedulingInfo {
                     thread_id,
                     transaction,
@@ -209,6 +214,7 @@ impl<Tx: TransactionWithMeta> Scheduler<Tx> for GreedyScheduler<Tx> {
             num_scheduled,
             num_unschedulable_conflicts,
             num_unschedulable_threads,
+            num_skipped_retry,
             num_filtered_out: 0,
             filter_time_us: 0,
         })
@@ -228,6 +234,7 @@ fn try_schedule_transaction<Tx: TransactionWithMeta>(
 ) -> Result<TransactionSchedulingInfo<Tx>, TransactionSchedulingError> {
     match pre_lock_filter(transaction_state) {
         PreLockFilterAction::AttemptToSchedule => {}
+        PreLockFilterAction::SkipAndRetain => return Err(TransactionSchedulingError::Skipped),
     }
 
     // Schedule the transaction if it can be.

core/src/banking_stage/transaction_scheduler/prio_graph_scheduler.rs

@@ -197,6 +197,7 @@ impl<Tx: TransactionWithMeta> Scheduler<Tx> for PrioGraphScheduler<Tx> {
         let mut num_sent: usize = 0;
         let mut num_unschedulable_conflicts: usize = 0;
         let mut num_unschedulable_threads: usize = 0;
+        let mut num_skipped_retry: usize = 0;
         while num_scanned < self.config.max_scanned_transactions_per_scheduling_pass {
             // If nothing is in the main-queue of the `PrioGraph` then there's nothing left to schedule.
             if self.prio_graph.is_empty() {
@@ -239,6 +240,10 @@ impl<Tx: TransactionWithMeta> Scheduler<Tx> for PrioGraphScheduler<Tx> {
                         num_unschedulable_threads += 1;
                         unschedulable_ids.push(id);
                     }
+                    Err(TransactionSchedulingError::Skipped) => {
+                        num_skipped_retry += 1;
+                        unschedulable_ids.push(id);
+                    }
                     Ok(TransactionSchedulingInfo {
                         thread_id,
                         transaction,
@@ -331,6 +336,7 @@ impl<Tx: TransactionWithMeta> Scheduler<Tx> for PrioGraphScheduler<Tx> {
             num_scheduled,
             num_unschedulable_conflicts,
             num_unschedulable_threads,
+            num_skipped_retry,
             num_filtered_out,
             filter_time_us: total_filter_time_us,
         })
@@ -370,6 +376,7 @@ fn try_schedule_transaction<Tx: TransactionWithMeta>(
 ) -> Result<TransactionSchedulingInfo<Tx>, TransactionSchedulingError> {
     match pre_lock_filter(transaction_state) {
         PreLockFilterAction::AttemptToSchedule => {}
+        PreLockFilterAction::SkipAndRetain => return Err(TransactionSchedulingError::Skipped),
     }
 
     // Check if this transaction conflicts with any blocked transactions
@@ -694,6 +701,7 @@ mod tests {
         // Complete batch on thread 0. Remaining txs can be scheduled onto thread 1
         finished_work_sender
             .send(FinishedConsumeWork {
+                slot: None,
                 work: thread_0_work.into_iter().next().unwrap(),
                 retryable_indexes: vec![],
             })

core/src/banking_stage/transaction_scheduler/scheduler.rs

@@ -50,6 +50,8 @@ pub(crate) trait Scheduler<Tx: TransactionWithMeta> {
 pub(crate) enum PreLockFilterAction {
     /// Attempt to schedule the transaction.
     AttemptToSchedule,
+    /// Skip the transaction but do not drop it.
+    SkipAndRetain,
 }
 
 /// Metrics from scheduling transactions.
@@ -67,6 +69,8 @@ pub(crate) struct SchedulingSummary {
     pub num_unschedulable_conflicts: usize,
     /// Number of transactions that were skipped due to thread capacity.
     pub num_unschedulable_threads: usize,
+    /// Number of transactions that were skipped due too recently attempted.
+    pub num_skipped_retry: usize,
     /// Number of transactions that were dropped due to filter.
     pub num_filtered_out: usize,
     /// Time spent filtering transactions