event_handler: unify the completed_block_sender and event_sender (anza-xyz#301)

Author: AshwinSekar

core/src/banking_simulation.rs

@@ -45,7 +45,7 @@ use {
     },
     solana_streamer::socket::SocketAddrSpace,
     solana_turbine::broadcast_stage::{BroadcastStage, BroadcastStageType},
-    solana_votor::event::CompletedBlockReceiver,
+    solana_votor::event::VotorEventReceiver,
     std::{
         collections::BTreeMap,
         fmt::Display,
@@ -420,7 +420,7 @@ struct SimulatorLoop {
     leader_schedule_cache: Arc<LeaderScheduleCache>,
     retransmit_slots_sender: Sender<Slot>,
     retracer: Arc<BankingTracer>,
-    _completed_block_receiver: CompletedBlockReceiver,
+    _completed_block_receiver: VotorEventReceiver,
 }
 
 impl SimulatorLoop {

core/src/replay_stage.rs

@@ -83,7 +83,7 @@ use {
     solana_timings::ExecuteTimings,
     solana_vote::vote_transaction::VoteTransaction,
     solana_votor::{
-        event::{CompletedBlock, CompletedBlockReceiver, CompletedBlockSender},
+        event::{CompletedBlock, VotorEvent, VotorEventReceiver, VotorEventSender},
         root_utils,
         vote_history::VoteHistory,
         vote_history_storage::VoteHistoryStorage,
@@ -301,7 +301,7 @@ pub struct ReplaySenders {
     pub dumped_slots_sender: Sender<Vec<(u64, Hash)>>,
     pub alpenglow_vote_sender: AlpenglowVoteSender,
     pub certificate_sender: Sender<(CertificateId, CertificateMessage)>,
-    pub completed_block_sender: CompletedBlockSender,
+    pub votor_event_sender: VotorEventSender,
 }
 
 pub struct ReplayReceivers {
@@ -312,7 +312,7 @@ pub struct ReplayReceivers {
     pub gossip_verified_vote_hash_receiver: Receiver<(Pubkey, u64, Hash)>,
     pub popular_pruned_forks_receiver: Receiver<Vec<u64>>,
     pub bls_verified_message_receiver: BLSVerifiedMessageReceiver,
-    pub completed_block_receiver: CompletedBlockReceiver,
+    pub votor_event_receiver: VotorEventReceiver,
 }
 
 /// Timing information for the ReplayStage main processing loop
@@ -608,7 +608,7 @@ impl ReplayStage {
             dumped_slots_sender,
             alpenglow_vote_sender,
             certificate_sender,
-            completed_block_sender,
+            votor_event_sender,
         } = senders;
 
         let ReplayReceivers {
@@ -619,7 +619,7 @@ impl ReplayStage {
             gossip_verified_vote_hash_receiver,
             popular_pruned_forks_receiver,
             bls_verified_message_receiver,
-            completed_block_receiver,
+            votor_event_receiver,
         } = receivers;
 
         trace!("replay stage");
@@ -653,7 +653,8 @@ impl ReplayStage {
             bank_notification_sender: bank_notification_sender.clone(),
             leader_window_notifier,
             certificate_sender,
-            completed_block_receiver: completed_block_receiver.clone(),
+            event_sender: votor_event_sender.clone(),
+            event_receiver: votor_event_receiver.clone(),
             bls_receiver: bls_verified_message_receiver,
         };
         let votor = Votor::new(votor_config);
@@ -854,7 +855,7 @@ impl ReplayStage {
                     first_alpenglow_slot,
                     (!is_alpenglow_migration_complete).then_some(&mut tbft_structs),
                     &mut is_alpenglow_migration_complete,
-                    &completed_block_sender,
+                    &votor_event_sender,
                 );
                 let did_complete_bank = !new_frozen_slots.is_empty();
                 if is_alpenglow_migration_complete {
@@ -879,7 +880,7 @@ impl ReplayStage {
                 let start_leader_time = if !is_alpenglow_migration_complete {
                     // TODO(ashwin): This will be moved to the event coordinator once we figure out
                     // migration
-                    for _ in completed_block_receiver.try_iter() {}
+                    for _ in votor_event_receiver.try_iter() {}
 
                     // Process cluster-agreed versions of duplicate slots for which we potentially
                     // have the wrong version. Our version was dead or pruned.
@@ -3382,7 +3383,7 @@ impl ReplayStage {
         poh_recorder: &RwLock<PohRecorder>,
         is_alpenglow_migration_complete: &mut bool,
         mut tbft_structs: Option<&mut TowerBFTStructures>,
-        completed_block_sender: &CompletedBlockSender,
+        votor_event_sender: &VotorEventSender,
     ) -> Vec<Slot> {
         // TODO: See if processing of blockstore replay results and bank completion can be made thread safe.
         let mut tx_count = 0;
@@ -3611,18 +3612,18 @@ impl ReplayStage {
                 }
 
                 // For leader banks:
-                // 1) Replay finishes before shredding, broadcast_stage will take care off
+                // 1) Replay finishes before shredding, broadcast_stage will take care of
                 //      notifying votor
                 // 2) Shredding finishes before replay, we notify here
                 //
                 // For non leader banks (2) is always true, so notify here
                 if *is_alpenglow_migration_complete && bank.block_id().is_some() {
                     // Leader blocks will not have a block id, broadcast stage will
                     // take care of notifying the voting loop
-                    let _ = completed_block_sender.send(CompletedBlock {
+                    let _ = votor_event_sender.send(VotorEvent::Block(CompletedBlock {
                         slot: bank.slot(),
                         bank: bank.clone_without_scheduler(),
-                    });
+                    }));
                 }
 
                 if let Some(sender) = bank_notification_sender {
@@ -3721,7 +3722,7 @@ impl ReplayStage {
         first_alpenglow_slot: Option<Slot>,
         tbft_structs: Option<&mut TowerBFTStructures>,
         is_alpenglow_migration_complete: &mut bool,
-        completed_block_sender: &CompletedBlockSender,
+        votor_event_sender: &VotorEventSender,
     ) -> Vec<Slot> /* completed slots */ {
         let active_bank_slots = bank_forks.read().unwrap().active_bank_slots();
         let num_active_banks = active_bank_slots.len();
@@ -3802,7 +3803,7 @@ impl ReplayStage {
             poh_recorder,
             is_alpenglow_migration_complete,
             tbft_structs,
-            completed_block_sender,
+            votor_event_sender,
         )
     }
 

core/src/tpu.rs

@@ -54,7 +54,7 @@ use {
         streamer::StakedNodes,
     },
     solana_turbine::broadcast_stage::{BroadcastStage, BroadcastStageType},
-    solana_votor::event::CompletedBlockSender,
+    solana_votor::event::VotorEventSender,
     std::{
         collections::HashMap,
         net::{SocketAddr, UdpSocket},
@@ -125,7 +125,7 @@ impl Tpu {
         bls_verified_message_sender: BLSVerifiedMessageSender,
         connection_cache: &Arc<ConnectionCache>,
         turbine_quic_endpoint_sender: AsyncSender<(SocketAddr, Bytes)>,
-        completed_block_sender: CompletedBlockSender,
+        votor_event_sender: VotorEventSender,
         keypair: &Keypair,
         log_messages_bytes_limit: Option<usize>,
         staked_nodes: &Arc<RwLock<StakedNodes>>,
@@ -342,7 +342,7 @@ impl Tpu {
             bank_forks,
             shred_version,
             turbine_quic_endpoint_sender,
-            completed_block_sender,
+            votor_event_sender,
         );
 
         (

core/src/tvu.rs

@@ -50,7 +50,7 @@ use {
     solana_sdk::{clock::Slot, pubkey::Pubkey, signature::Keypair},
     solana_turbine::retransmit_stage::RetransmitStage,
     solana_votor::{
-        event::{CompletedBlockReceiver, CompletedBlockSender},
+        event::{VotorEventReceiver, VotorEventSender},
         vote_history::VoteHistory,
         vote_history_storage::VoteHistoryStorage,
         votor::LeaderWindowNotifier,
@@ -177,8 +177,8 @@ impl Tvu {
         replay_highest_frozen: Arc<ReplayHighestFrozen>,
         leader_window_notifier: Arc<LeaderWindowNotifier>,
         voting_service_additional_listeners: Option<&Vec<SocketAddr>>,
-        completed_block_sender: CompletedBlockSender,
-        completed_block_receiver: CompletedBlockReceiver,
+        votor_event_sender: VotorEventSender,
+        votor_event_receiver: VotorEventReceiver,
     ) -> Result<Self, String> {
         let in_wen_restart = wen_restart_repair_slots.is_some();
 
@@ -323,7 +323,7 @@ impl Tvu {
             dumped_slots_sender,
             alpenglow_vote_sender,
             certificate_sender,
-            completed_block_sender,
+            votor_event_sender,
         };
 
         let replay_receivers = ReplayReceivers {
@@ -334,7 +334,7 @@ impl Tvu {
             gossip_verified_vote_hash_receiver,
             popular_pruned_forks_receiver,
             bls_verified_message_receiver,
-            completed_block_receiver,
+            votor_event_receiver,
         };
 
         let replay_stage_config = ReplayStageConfig {
@@ -568,7 +568,7 @@ pub mod tests {
                 DEFAULT_TPU_CONNECTION_POOL_SIZE,
             )
         };
-        let (completed_block_sender, completed_block_receiver) = unbounded();
+        let (votor_event_sender, votor_event_receiver) = unbounded();
 
         let tvu = Tvu::new(
             &vote_keypair.pubkey(),
@@ -638,8 +638,8 @@ pub mod tests {
             Arc::new(ReplayHighestFrozen::default()),
             Arc::new(LeaderWindowNotifier::default()),
             None,
-            completed_block_sender,
-            completed_block_receiver,
+            votor_event_sender,
+            votor_event_receiver,
         )
         .expect("assume success");
         if enable_wen_restart {

core/src/validator.rs

@@ -1557,7 +1557,7 @@ impl Validator {
             Arc::new(crate::cluster_slots_service::cluster_slots::ClusterSlots::default());
         // This channel backing up indicates a serious problem in the voting loop
         // Capping at 1000 for now, TODO: add metrics for channel len
-        let (completed_block_sender, completed_block_receiver) = bounded(1000);
+        let (votor_event_sender, votor_event_receiver) = bounded(1000);
 
         // If RPC is supported and ConnectionCache is used, pass ConnectionCache for being warmup inside Tvu.
         let connection_cache_for_warmup = (json_rpc_service.is_some()
@@ -1632,8 +1632,8 @@ impl Validator {
             replay_highest_frozen,
             leader_window_notifier,
             config.voting_service_additional_listeners.as_ref(),
-            completed_block_sender.clone(),
-            completed_block_receiver,
+            votor_event_sender.clone(),
+            votor_event_receiver,
         )
         .map_err(ValidatorError::Other)?;
 
@@ -1694,7 +1694,7 @@ impl Validator {
             bls_verified_message_sender,
             &connection_cache,
             turbine_quic_endpoint_sender,
-            completed_block_sender,
+            votor_event_sender,
             &identity_keypair,
             config.runtime_config.log_messages_bytes_limit,
             &staked_nodes,

turbine/src/broadcast_stage.rs

@@ -34,7 +34,7 @@ use {
         sendmmsg::{batch_send, SendPktsError},
         socket::SocketAddrSpace,
     },
-    solana_votor::event::CompletedBlockSender,
+    solana_votor::event::VotorEventSender,
     static_assertions::const_assert_eq,
     std::{
         collections::{HashMap, HashSet},
@@ -122,7 +122,7 @@ impl BroadcastStageType {
         bank_forks: Arc<RwLock<BankForks>>,
         shred_version: u16,
         quic_endpoint_sender: AsyncSender<(SocketAddr, Bytes)>,
-        completed_block_sender: CompletedBlockSender,
+        votor_event_sender: VotorEventSender,
     ) -> BroadcastStage {
         match self {
             BroadcastStageType::Standard => BroadcastStage::new(
@@ -134,7 +134,7 @@ impl BroadcastStageType {
                 blockstore,
                 bank_forks,
                 quic_endpoint_sender,
-                completed_block_sender,
+                votor_event_sender.clone(),
                 StandardBroadcastRun::new(shred_version),
             ),
 
@@ -147,7 +147,7 @@ impl BroadcastStageType {
                 blockstore,
                 bank_forks,
                 quic_endpoint_sender,
-                completed_block_sender,
+                votor_event_sender.clone(),
                 FailEntryVerificationBroadcastRun::new(shred_version),
             ),
 
@@ -160,7 +160,7 @@ impl BroadcastStageType {
                 blockstore,
                 bank_forks,
                 quic_endpoint_sender,
-                completed_block_sender,
+                votor_event_sender.clone(),
                 BroadcastFakeShredsRun::new(0, shred_version),
             ),
 
@@ -173,7 +173,7 @@ impl BroadcastStageType {
                 blockstore,
                 bank_forks,
                 quic_endpoint_sender,
-                completed_block_sender,
+                votor_event_sender.clone(),
                 BroadcastDuplicatesRun::new(shred_version, config.clone()),
             ),
         }
@@ -188,7 +188,7 @@ trait BroadcastRun {
         receiver: &Receiver<WorkingBankEntry>,
         socket_sender: &Sender<(Arc<Vec<Shred>>, Option<BroadcastShredBatchInfo>)>,
         blockstore_sender: &Sender<(Arc<Vec<Shred>>, Option<BroadcastShredBatchInfo>)>,
-        completed_block_sender: &CompletedBlockSender,
+        votor_event_sender: &VotorEventSender,
     ) -> Result<()>;
     fn transmit(
         &mut self,
@@ -231,7 +231,7 @@ impl BroadcastStage {
         receiver: &Receiver<WorkingBankEntry>,
         socket_sender: &Sender<(Arc<Vec<Shred>>, Option<BroadcastShredBatchInfo>)>,
         blockstore_sender: &Sender<(Arc<Vec<Shred>>, Option<BroadcastShredBatchInfo>)>,
-        completed_block_sender: &CompletedBlockSender,
+        votor_event_sender: &VotorEventSender,
         mut broadcast_stage_run: impl BroadcastRun,
     ) -> BroadcastStageReturnType {
         loop {
@@ -241,7 +241,7 @@ impl BroadcastStage {
                 receiver,
                 socket_sender,
                 blockstore_sender,
-                completed_block_sender,
+                votor_event_sender,
             );
             let res = Self::handle_error(res, "run");
             if let Some(res) = res {
@@ -294,7 +294,7 @@ impl BroadcastStage {
         blockstore: Arc<Blockstore>,
         bank_forks: Arc<RwLock<BankForks>>,
         quic_endpoint_sender: AsyncSender<(SocketAddr, Bytes)>,
-        completed_block_sender: CompletedBlockSender,
+        votor_event_sender: VotorEventSender,
         mut broadcast_stage_run: impl BroadcastRun + Send + 'static + Clone,
     ) -> Self {
         let (socket_sender, socket_receiver) = unbounded();
@@ -315,7 +315,7 @@ impl BroadcastStage {
                         &receiver,
                         &socket_sender_,
                         &blockstore_sender,
-                        &completed_block_sender,
+                        &votor_event_sender,
                         bs_run,
                     )
                 })
@@ -704,7 +704,7 @@ pub mod test {
         let bank_forks = BankForks::new_rw_arc(bank);
         let bank = bank_forks.read().unwrap().root_bank();
 
-        let (completed_block_sender, _) = unbounded();
+        let (votor_event_sender, _) = unbounded();
 
         // Start up the broadcast stage
         let broadcast_service = BroadcastStage::new(
@@ -716,7 +716,7 @@ pub mod test {
             blockstore.clone(),
             bank_forks,
             quic_endpoint_sender,
-            completed_block_sender,
+            votor_event_sender,
             StandardBroadcastRun::new(0),
         );
 

turbine/src/broadcast_stage/broadcast_duplicates_run.rs

@@ -10,7 +10,7 @@ use {
         signature::{Keypair, Signature, Signer},
         system_transaction,
     },
-    solana_votor::event::CompletedBlockSender,
+    solana_votor::event::VotorEventSender,
     std::collections::HashSet,
 };
 
@@ -83,7 +83,7 @@ impl BroadcastRun for BroadcastDuplicatesRun {
         receiver: &Receiver<WorkingBankEntry>,
         socket_sender: &Sender<(Arc<Vec<Shred>>, Option<BroadcastShredBatchInfo>)>,
         blockstore_sender: &Sender<(Arc<Vec<Shred>>, Option<BroadcastShredBatchInfo>)>,
-        _completed_block_sender: &CompletedBlockSender,
+        _votor_event_sender: &VotorEventSender,
     ) -> Result<()> {
         // 1) Pull entries from banking stage
         let mut receive_results = broadcast_utils::recv_slot_entries(receiver)?;