Scheduler Frequency Fixes (#4545)

* Change PrioGraphSchedulerConfig::default for 1k maxs, 256 look ahead

* Break loop on scanned transaction count

* make Hold decision behave same as Consume during receive

* receive maximum of 5_000 packets - loose max

* receive_completed before process_transactions

(cherry picked from commit 6e93845)

# Conflicts:
#	core/src/banking_stage/transaction_scheduler/prio_graph_scheduler.rs
#	core/src/banking_stage/transaction_scheduler/receive_and_buffer.rs
#	core/src/banking_stage/transaction_scheduler/scheduler_controller.rs

core/src/banking_stage/transaction_scheduler/prio_graph_scheduler.rs

@@ -39,7 +39,29 @@ type SchedulerPrioGraph = PrioGraph<
     fn(&TransactionPriorityId, &GraphNode<TransactionPriorityId>) -> TransactionPriorityId,
 >;
 
+<<<<<<< HEAD
 pub(crate) struct PrioGraphScheduler {
+=======
+pub(crate) struct PrioGraphSchedulerConfig {
+    pub max_scheduled_cus: u64,
+    pub max_scanned_transactions_per_scheduling_pass: usize,
+    pub look_ahead_window_size: usize,
+    pub target_transactions_per_batch: usize,
+}
+
+impl Default for PrioGraphSchedulerConfig {
+    fn default() -> Self {
+        Self {
+            max_scheduled_cus: MAX_BLOCK_UNITS,
+            max_scanned_transactions_per_scheduling_pass: 1000,
+            look_ahead_window_size: 256,
+            target_transactions_per_batch: TARGET_NUM_TRANSACTIONS_PER_BATCH,
+        }
+    }
+}
+
+pub(crate) struct PrioGraphScheduler<Tx> {
+>>>>>>> 6e93845fb (Scheduler Frequency Fixes (#4545))
     in_flight_tracker: InFlightTracker,
     account_locks: ThreadAwareAccountLocks,
     consume_work_senders: Vec<Sender<ConsumeWork>>,
@@ -168,19 +190,31 @@ impl PrioGraphScheduler {
         // Check transactions against filter, remove from container if it fails.
         chunked_pops(container, &mut self.prio_graph, &mut window_budget);
 
+<<<<<<< HEAD
         let mut unblock_this_batch =
             Vec::with_capacity(self.consume_work_senders.len() * TARGET_NUM_TRANSACTIONS_PER_BATCH);
         const MAX_TRANSACTIONS_PER_SCHEDULING_PASS: usize = 100_000;
         let mut num_scheduled: usize = 0;
         let mut num_sent: usize = 0;
         let mut num_unschedulable: usize = 0;
         while num_scheduled < MAX_TRANSACTIONS_PER_SCHEDULING_PASS {
+=======
+        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_scanned < self.config.max_scanned_transactions_per_scheduling_pass {
+>>>>>>> 6e93845fb (Scheduler Frequency Fixes (#4545))
             // 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.
@@ -245,12 +279,19 @@ impl PrioGraphScheduler {
                                 break;
                             }
                         }
+<<<<<<< HEAD
 
                         if num_scheduled >= MAX_TRANSACTIONS_PER_SCHEDULING_PASS {
                             break;
                         }
+=======
+>>>>>>> 6e93845fb (Scheduler Frequency Fixes (#4545))
                     }
                 }
+
+                if num_scanned >= self.config.max_scanned_transactions_per_scheduling_pass {
+                    break;
+                }
             }
 
             // Send all non-empty batches