feat: group P3 & banking into 50ms batches

Author: OliverNChalk

banking-bench/src/main.rs

@@ -485,6 +485,7 @@ fn main() {
         false,
         HashSet::default(),
         BundleAccountLocker::default(),
+        Duration::from_millis(50),
     );
 
     // This is so that the signal_receiver does not go out of scope after the closure.

core/src/banking_simulation.rs

@@ -819,6 +819,7 @@ impl BankingSimulator {
             false,
             collections::HashSet::default(),
             BundleAccountLocker::default(),
+            Duration::from_millis(50),
         );
 
         let (&_slot, &raw_base_event_time) = freeze_time_by_slot

core/src/banking_stage.rs

@@ -82,6 +82,8 @@ pub(crate) mod transaction_scheduler;
 // Fixed thread size seems to be fastest on GCP setup
 pub const NUM_THREADS: u32 = 6;
 
+pub const DEFAULT_BATCH_INTERVAL: Duration = Duration::from_millis(50);
+
 const TOTAL_BUFFERED_PACKETS: usize = 100_000;
 
 const NUM_VOTE_PROCESSING_THREADS: u32 = 2;
@@ -365,6 +367,7 @@ impl BankingStage {
         enable_forwarding: bool,
         blacklisted_accounts: HashSet<Pubkey>,
         bundle_account_locker: BundleAccountLocker,
+        batch_interval: Duration,
     ) -> Self {
         Self::new_num_threads(
             block_production_method,
@@ -383,6 +386,7 @@ impl BankingStage {
             enable_forwarding,
             blacklisted_accounts,
             bundle_account_locker,
+            batch_interval,
         )
     }
 
@@ -404,6 +408,7 @@ impl BankingStage {
         enable_forwarding: bool,
         blacklisted_accounts: HashSet<Pubkey>,
         bundle_account_locker: BundleAccountLocker,
+        batch_interval: Duration,
     ) -> Self {
         match block_production_method {
             BlockProductionMethod::ThreadLocalMultiIterator => {
@@ -440,6 +445,7 @@ impl BankingStage {
                 enable_forwarding,
                 blacklisted_accounts,
                 bundle_account_locker,
+                batch_interval,
             ),
         }
     }
@@ -551,6 +557,7 @@ impl BankingStage {
         enable_forwarding: bool,
         blacklisted_accounts: HashSet<Pubkey>,
         bundle_account_locker: BundleAccountLocker,
+        batch_interval: Duration,
     ) -> Self {
         assert!(num_threads >= MIN_TOTAL_THREADS);
         // Single thread to generate entries from many banks.
@@ -659,6 +666,7 @@ impl BankingStage {
                 scheduler,
                 worker_metrics,
                 forwarder,
+                batch_interval,
             );
             Builder::new()
                 .name("solBnkTxSched".to_string())
@@ -939,6 +947,7 @@ mod tests {
                 false,
                 HashSet::default(),
                 BundleAccountLocker::default(),
+                Duration::from_millis(50),
             );
             drop(non_vote_sender);
             drop(tpu_vote_sender);
@@ -997,6 +1006,7 @@ mod tests {
                 false,
                 HashSet::default(),
                 BundleAccountLocker::default(),
+                Duration::from_millis(50),
             );
             trace!("sending bank");
             drop(non_vote_sender);
@@ -1084,6 +1094,7 @@ mod tests {
                 false,
                 HashSet::default(),
                 BundleAccountLocker::default(),
+                Duration::from_millis(50),
             );
 
             // fund another account so we can send 2 good transactions in a single batch.
@@ -1464,6 +1475,7 @@ mod tests {
                 false,
                 HashSet::default(),
                 BundleAccountLocker::default(),
+                Duration::from_millis(50),
             );
 
             let keypairs = (0..100).map(|_| Keypair::new()).collect_vec();

core/src/banking_stage/read_write_account_set.rs

@@ -140,6 +140,7 @@ mod tests {
             VersionedTransaction::try_new(message, &[write_keypair]).unwrap(),
             MessageHash::Compute,
             Some(false),
+            false,
             bank,
             bank.get_reserved_account_keys(),
         )

core/src/banking_stage/transaction_scheduler/scheduler_controller.rs

@@ -71,6 +71,9 @@ pub(crate) struct SchedulerController<T: LikeClusterInfo> {
     worker_metrics: Vec<Arc<ConsumeWorkerMetrics>>,
     /// State for forwarding packets to the leader, if enabled.
     forwarder: Option<Forwarder<T>>,
+    batch: Vec<ImmutableDeserializedPacket>,
+    batch_start: Instant,
+    batch_interval: Duration,
 }
 
 impl<T: LikeClusterInfo> SchedulerController<T> {
@@ -81,6 +84,7 @@ impl<T: LikeClusterInfo> SchedulerController<T> {
         scheduler: GreedyScheduler,
         worker_metrics: Vec<Arc<ConsumeWorkerMetrics>>,
         forwarder: Option<Forwarder<T>>,
+        batch_interval: Duration,
     ) -> Self {
         Self {
             decision_maker,
@@ -94,6 +98,10 @@ impl<T: LikeClusterInfo> SchedulerController<T> {
             timing_metrics: SchedulerTimingMetrics::default(),
             worker_metrics,
             forwarder,
+
+            batch: Vec::default(),
+            batch_start: Instant::now(),
+            batch_interval,
         }
     }
 
@@ -123,6 +131,7 @@ impl<T: LikeClusterInfo> SchedulerController<T> {
                 .maybe_report_and_reset_slot(new_leader_slot);
 
             self.receive_completed()?;
+            self.maybe_queue_batch();
             self.process_transactions(&decision)?;
             if !self.receive_and_buffer_packets(&decision) {
                 break;
@@ -146,6 +155,23 @@ impl<T: LikeClusterInfo> SchedulerController<T> {
         Ok(())
     }
 
+    fn maybe_queue_batch(&mut self) {
+        // Early return if the current batch is empty.
+        if self.batch.is_empty() {
+            return;
+        }
+
+        if self.batch_start.elapsed() > self.batch_interval {
+            Self::buffer_packets(
+                &self.bank_forks,
+                &mut self.container,
+                &mut self.transaction_id_generator,
+                &mut self.count_metrics,
+                self.batch.drain(..),
+            );
+        }
+    }
+
     /// Process packets based on decision.
     fn process_transactions(
         &mut self,
@@ -474,9 +500,11 @@ impl<T: LikeClusterInfo> SchedulerController<T> {
                 });
 
                 if should_buffer {
-                    let (_, buffer_time_us) = measure_us!(
-                        self.buffer_packets(receive_packet_results.deserialized_packets)
-                    );
+                    // Packets are not immediately schedulable but are instead
+                    // grouped into 50ms batches (measured from the recv time of
+                    // the first packet).
+                    let (_, buffer_time_us) =
+                        measure_us!(self.extend_batch(receive_packet_results.deserialized_packets));
                     self.timing_metrics.update(|timing_metrics| {
                         saturating_add_assign!(timing_metrics.buffer_time_us, buffer_time_us);
                     });
@@ -496,12 +524,27 @@ impl<T: LikeClusterInfo> SchedulerController<T> {
         true
     }
 
-    fn buffer_packets(&mut self, packets: Vec<ImmutableDeserializedPacket>) {
+    fn extend_batch(&mut self, packets: Vec<ImmutableDeserializedPacket>) {
+        // If this is the first packet in the batch, set the
+        // start timestamp for the batch.
+        if self.batch.is_empty() {
+            self.batch_start = Instant::now();
+        }
+        self.batch.extend(packets);
+    }
+
+    fn buffer_packets(
+        bank_forks: &Arc<RwLock<BankForks>>,
+        container: &mut TransactionStateContainer,
+        transaction_id_generator: &mut TransactionIdGenerator,
+        count_metrics: &mut SchedulerCountMetrics,
+        packets: impl Iterator<Item = ImmutableDeserializedPacket>,
+    ) {
         // Convert to Arcs
         let packets: Vec<_> = packets.into_iter().map(Arc::new).collect();
         // Sanitize packets, generate IDs, and insert into the container.
         let (root_bank, working_bank) = {
-            let bank_forks = self.bank_forks.read().unwrap();
+            let bank_forks = bank_forks.read().unwrap();
             let root_bank = bank_forks.root_bank();
             let working_bank = bank_forks.working_bank();
             (root_bank, working_bank)
@@ -586,7 +629,7 @@ impl<T: LikeClusterInfo> SchedulerController<T> {
                     .filter(|(_, check_result)| check_result.is_ok())
             {
                 saturating_add_assign!(post_transaction_check_count, 1);
-                let transaction_id = self.transaction_id_generator.next();
+                let transaction_id = transaction_id_generator.next();
 
                 let (priority, cost) = Self::calculate_priority_and_cost(
                     &transaction,
@@ -598,7 +641,7 @@ impl<T: LikeClusterInfo> SchedulerController<T> {
                     max_age,
                 };
 
-                if self.container.insert_new_transaction(
+                if container.insert_new_transaction(
                     transaction_id,
                     transaction_ttl,
                     packet,
@@ -617,7 +660,7 @@ impl<T: LikeClusterInfo> SchedulerController<T> {
             let num_dropped_on_transaction_checks =
                 post_lock_validation_count.saturating_sub(post_transaction_check_count);
 
-            self.count_metrics.update(|count_metrics| {
+            count_metrics.update(|count_metrics| {
                 saturating_add_assign!(
                     count_metrics.num_dropped_on_capacity,
                     num_dropped_on_capacity
@@ -809,6 +852,7 @@ mod tests {
             GreedyScheduler::new(consume_work_senders, finished_consume_work_receiver),
             vec![], // no actual workers with metrics to report, this can be empty
             None,
+            Duration::from_millis(50),
         );
 
         (test_frame, scheduler_controller)

core/src/bundle_stage/bundle_consumer.rs

@@ -224,7 +224,7 @@ impl BundleConsumer {
                             &locked_bundle,
                             bank_start,
                             bundle_stage_leader_metrics,
-                            Some(10u64.pow(6) / 10), // 0.1 lamports per CU
+                            Some(10u64.pow(6) / 1000), // 0.001 lamports per CU (filling a block at this rate pays ~1c).
                         ));
                         bundle_stage_leader_metrics
                             .leader_slot_metrics_tracker()
@@ -277,7 +277,7 @@ impl BundleConsumer {
         if bank_start.working_bank.slot() != *last_tip_updated_slot
             && Self::bundle_touches_tip_pdas(
                 locked_bundle.sanitized_bundle(),
-                &tip_manager.get_tip_accounts(),
+                tip_manager.get_tip_accounts(),
             )
         {
             let start = Instant::now();
@@ -482,6 +482,7 @@ impl BundleConsumer {
         ))
     }
 
+    #[allow(clippy::too_many_arguments)]
     fn update_qos_and_execute_record_commit_bundle(
         committer: &Committer,
         recorder: &TransactionRecorder,
@@ -1139,6 +1140,7 @@ mod tests {
             leader_schedule_cache,
             TipManagerConfig {
                 funnel,
+                rewards_split: None,
                 tip_payment_program_id: Pubkey::from_str(
                     "T1pyyaTNZsKv2WcRAB8oVnk93mLJw2XzjtVYqCsaHqt",
                 )