Scheduler Frequency Fixes

core/src/banking_stage/transaction_scheduler/prio_graph_scheduler.rs

@@ -43,7 +43,7 @@ type SchedulerPrioGraph = PrioGraph<
 
 pub(crate) struct PrioGraphSchedulerConfig {
     pub max_scheduled_cus: u64,
-    pub max_transactions_per_scheduling_pass: usize,
+    pub max_scanned_transactions_per_scheduling_pass: usize,
     pub look_ahead_window_size: usize,
     pub target_transactions_per_batch: usize,
 }
@@ -52,8 +52,8 @@ impl Default for PrioGraphSchedulerConfig {
     fn default() -> Self {
         Self {
             max_scheduled_cus: MAX_BLOCK_UNITS,
-            max_transactions_per_scheduling_pass: 100_000,
-            look_ahead_window_size: 2048,
+            max_scanned_transactions_per_scheduling_pass: 1000,
+            look_ahead_window_size: 256,
             target_transactions_per_batch: TARGET_NUM_TRANSACTIONS_PER_BATCH,
         }
     }
@@ -192,16 +192,18 @@ impl<Tx: TransactionWithMeta> PrioGraphScheduler<Tx> {
         let mut unblock_this_batch = Vec::with_capacity(
             self.consume_work_senders.len() * self.config.target_transactions_per_batch,
         );
+        let mut num_scanned: usize = 0;
         let mut num_scheduled: usize = 0;
         let mut num_sent: usize = 0;
         let mut num_unschedulable: usize = 0;
-        while num_scheduled < self.config.max_transactions_per_scheduling_pass {
+        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() {
                 break;
             }
 
             while let Some(id) = self.prio_graph.pop() {
+                num_scanned += 1;
                 unblock_this_batch.push(id);
 
                 // Should always be in the container, during initial testing phase panic.
@@ -267,12 +269,12 @@ impl<Tx: TransactionWithMeta> PrioGraphScheduler<Tx> {
                                 break;
                             }
                         }
-
-                        if num_scheduled >= self.config.max_transactions_per_scheduling_pass {
-                            break;
-                        }
                     }
                 }
+
+                if num_scanned >= self.config.max_scanned_transactions_per_scheduling_pass {
+                    break;
+                }
             }
 
             // Send all non-empty batches

core/src/banking_stage/transaction_scheduler/receive_and_buffer.rs

@@ -80,11 +80,10 @@ impl ReceiveAndBuffer for SanitizedTransactionReceiveAndBuffer {
         count_metrics: &mut SchedulerCountMetrics,
         decision: &BufferedPacketsDecision,
     ) -> Result<usize, ()> {
-        let remaining_queue_capacity = container.remaining_capacity();
-
+        const MAX_RECEIVE_PACKETS: usize = 5_000;
         const MAX_PACKET_RECEIVE_TIME: Duration = Duration::from_millis(10);
         let (recv_timeout, should_buffer) = match decision {
-            BufferedPacketsDecision::Consume(_) => (
+            BufferedPacketsDecision::Consume(_) | BufferedPacketsDecision::Hold => (
                 if container.is_empty() {
                     MAX_PACKET_RECEIVE_TIME
                 } else {
@@ -93,14 +92,12 @@ impl ReceiveAndBuffer for SanitizedTransactionReceiveAndBuffer {
                 true,
             ),
             BufferedPacketsDecision::Forward => (MAX_PACKET_RECEIVE_TIME, self.forwarding_enabled),
-            BufferedPacketsDecision::ForwardAndHold | BufferedPacketsDecision::Hold => {
-                (MAX_PACKET_RECEIVE_TIME, true)
-            }
+            BufferedPacketsDecision::ForwardAndHold => (MAX_PACKET_RECEIVE_TIME, true),
         };
 
         let (received_packet_results, receive_time_us) = measure_us!(self
             .packet_receiver
-            .receive_packets(recv_timeout, remaining_queue_capacity, |packet| {
+            .receive_packets(recv_timeout, MAX_RECEIVE_PACKETS, |packet| {
                 packet.check_excessive_precompiles()?;
                 Ok(packet)
             }));

core/src/banking_stage/transaction_scheduler/scheduler_controller.rs

@@ -105,8 +105,8 @@ impl<C: LikeClusterInfo, R: ReceiveAndBuffer> SchedulerController<C, R> {
             self.timing_metrics
                 .maybe_report_and_reset_slot(new_leader_slot);
 
-            self.process_transactions(&decision)?;
             self.receive_completed()?;
+            self.process_transactions(&decision)?;
             if self.receive_and_buffer_packets(&decision).is_err() {
                 break;
             }