Poh using mpsc ringbuf

core/src/banking_stage.rs

@@ -821,7 +821,7 @@ mod tests {
     use {
         super::*,
         crate::banking_trace::{BankingPacketBatch, BankingTracer},
-        crossbeam_channel::{unbounded, Receiver},
+        crossbeam_channel::unbounded,
         itertools::Itertools,
         solana_entry::entry::{self, Entry, EntrySlice},
         solana_gossip::cluster_info::Node,
@@ -835,6 +835,7 @@ mod tests {
         },
         solana_perf::packet::{to_packet_batches, PacketBatch},
         solana_poh::{
+            mpsc_ringbuffer::ArrayQueue,
             poh_recorder::{
                 create_test_recorder, PohRecorderError, Record, RecordTransactionsSummary,
             },
@@ -1339,7 +1340,7 @@ mod tests {
     }
 
     pub(crate) fn simulate_poh(
-        record_receiver: Receiver<Record>,
+        record_receiver: Arc<ArrayQueue<Record>>,
         poh_recorder: &Arc<RwLock<PohRecorder>>,
     ) -> JoinHandle<()> {
         let poh_recorder = poh_recorder.clone();
@@ -1349,7 +1350,7 @@ mod tests {
             .spawn(move || loop {
                 PohService::read_record_receiver_and_process(
                     &poh_recorder,
-                    &record_receiver,
+                    record_receiver.clone(),
                     Duration::from_millis(10),
                 );
                 if is_exited.load(Ordering::Relaxed) {

core/src/banking_stage/consumer.rs

@@ -862,7 +862,10 @@ mod tests {
             leader_schedule_cache::LeaderScheduleCache,
         },
         solana_perf::packet::Packet,
-        solana_poh::poh_recorder::{PohRecorder, Record, WorkingBankEntry},
+        solana_poh::{
+            mpsc_ringbuffer::ArrayQueue,
+            poh_recorder::{PohRecorder, Record, WorkingBankEntry},
+        },
         solana_rpc::transaction_status_service::TransactionStatusService,
         solana_runtime::{bank_forks::BankForks, prioritization_fee_cache::PrioritizationFeeCache},
         solana_sdk::{
@@ -1254,25 +1257,25 @@ mod tests {
             let poh_recorder = Arc::new(RwLock::new(poh_recorder));
 
             fn poh_tick_before_returning_record_response(
-                record_receiver: Receiver<Record>,
+                record_receiver: Arc<ArrayQueue<Record>>,
                 poh_recorder: Arc<RwLock<PohRecorder>>,
             ) -> JoinHandle<()> {
                 let is_exited = poh_recorder.read().unwrap().is_exited.clone();
                 let tick_producer = Builder::new()
                     .name("solana-simulate_poh".to_string())
                     .spawn(move || loop {
                         let timeout = Duration::from_millis(10);
-                        let record = record_receiver.recv_timeout(timeout);
-                        if let Ok(record) = record {
+                        let record = record_receiver.pop_with_timeout(timeout);
+                        if let Some(record) = record {
                             let record_response = poh_recorder.write().unwrap().record(
                                 record.slot,
                                 record.mixin,
                                 record.transactions,
                             );
-                            poh_recorder.write().unwrap().tick();
-                            if record.sender.send(record_response).is_err() {
+                            if record_response.is_err() {
                                 panic!("Error returning mixin hash");
                             }
+                            poh_recorder.write().unwrap().tick();
                         }
                         if is_exited.load(Ordering::Relaxed) {
                             break;

core/src/banking_stage/transaction_scheduler/scheduler_controller.rs

@@ -732,7 +732,10 @@ mod tests {
             get_tmp_ledger_path_auto_delete, leader_schedule_cache::LeaderScheduleCache,
         },
         solana_perf::packet::{to_packet_batches, PacketBatch, NUM_PACKETS},
-        solana_poh::poh_recorder::{PohRecorder, Record, WorkingBankEntry},
+        solana_poh::{
+            mpsc_ringbuffer::ArrayQueue,
+            poh_recorder::{PohRecorder, Record, WorkingBankEntry},
+        },
         solana_runtime::bank::Bank,
         solana_sdk::{
             compute_budget::ComputeBudgetInstruction, fee_calculator::FeeRateGovernor, hash::Hash,
@@ -754,7 +757,7 @@ mod tests {
         mint_keypair: Keypair,
         _ledger_path: TempDir,
         _entry_receiver: Receiver<WorkingBankEntry>,
-        _record_receiver: Receiver<Record>,
+        _record_receiver: Arc<ArrayQueue<Record>>,
         poh_recorder: Arc<RwLock<PohRecorder>>,
         banking_packet_sender: Sender<Arc<(Vec<PacketBatch>, Option<SigverifyTracerPacketStats>)>>,
 
@@ -799,7 +802,7 @@ mod tests {
             mint_keypair,
             _ledger_path: ledger_path,
             _entry_receiver: entry_receiver,
-            _record_receiver: record_receiver,
+            _record_receiver: record_receiver.clone(),
             poh_recorder,
             banking_packet_sender,
             consume_work_receivers,

poh/src/lib.rs

@@ -3,6 +3,8 @@ pub mod leader_bank_notifier;
 pub mod poh_recorder;
 pub mod poh_service;
 
+pub mod mpsc_ringbuffer;
+
 #[macro_use]
 extern crate solana_metrics;