votor: unify own vote sender with bls receiver, unify skip receiver (anza-xyz#308)

Author: AshwinSekar

core/src/replay_stage.rs

@@ -69,7 +69,10 @@ use {
         commitment::BlockCommitmentCache,
         installed_scheduler_pool::BankWithScheduler,
         prioritization_fee_cache::PrioritizationFeeCache,
-        vote_sender_types::{AlpenglowVoteSender, BLSVerifiedMessageReceiver, ReplayVoteSender},
+        vote_sender_types::{
+            AlpenglowVoteSender, BLSVerifiedMessageReceiver, BLSVerifiedMessageSender,
+            ReplayVoteSender,
+        },
     },
     solana_sdk::{
         clock::{BankId, Slot, NUM_CONSECUTIVE_LEADER_SLOTS},
@@ -302,6 +305,7 @@ pub struct ReplaySenders {
     pub alpenglow_vote_sender: AlpenglowVoteSender,
     pub certificate_sender: Sender<(CertificateId, CertificateMessage)>,
     pub votor_event_sender: VotorEventSender,
+    pub own_vote_sender: BLSVerifiedMessageSender,
 }
 
 pub struct ReplayReceivers {
@@ -609,6 +613,7 @@ impl ReplayStage {
             alpenglow_vote_sender,
             certificate_sender,
             votor_event_sender,
+            own_vote_sender,
         } = senders;
 
         let ReplayReceivers {
@@ -655,6 +660,7 @@ impl ReplayStage {
             certificate_sender,
             event_sender: votor_event_sender.clone(),
             event_receiver: votor_event_receiver.clone(),
+            own_vote_sender,
             bls_receiver: bls_verified_message_receiver,
         };
         let votor = Votor::new(votor_config);

core/src/tvu.rs

@@ -45,7 +45,10 @@ use {
         bank_forks::BankForks,
         commitment::BlockCommitmentCache,
         prioritization_fee_cache::PrioritizationFeeCache,
-        vote_sender_types::{AlpenglowVoteSender, BLSVerifiedMessageReceiver, ReplayVoteSender},
+        vote_sender_types::{
+            AlpenglowVoteSender, BLSVerifiedMessageReceiver, BLSVerifiedMessageSender,
+            ReplayVoteSender,
+        },
     },
     solana_sdk::{clock::Slot, pubkey::Pubkey, signature::Keypair},
     solana_turbine::retransmit_stage::RetransmitStage,
@@ -153,6 +156,7 @@ impl Tvu {
         bank_notification_sender: Option<BankNotificationSenderConfig>,
         duplicate_confirmed_slots_receiver: DuplicateConfirmedSlotsReceiver,
         alpenglow_vote_sender: AlpenglowVoteSender,
+        own_vote_sender: BLSVerifiedMessageSender,
         bls_verified_message_receiver: BLSVerifiedMessageReceiver,
         tvu_config: TvuConfig,
         max_slots: &Arc<MaxSlots>,
@@ -324,6 +328,7 @@ impl Tvu {
             alpenglow_vote_sender,
             certificate_sender,
             votor_event_sender,
+            own_vote_sender,
         };
 
         let replay_receivers = ReplayReceivers {
@@ -546,7 +551,7 @@ pub mod tests {
         let (replay_vote_sender, _replay_vote_receiver) = unbounded();
         let (alpenglow_vote_sender, _alpenglow_vote_receiver) = unbounded();
         let (_, gossip_confirmed_slots_receiver) = unbounded();
-        let (_, alpenglow_vote_receiver) = unbounded();
+        let (bls_verified_message_sender, bls_verified_message_receiver) = unbounded();
         let max_complete_transaction_status_slot = Arc::new(AtomicU64::default());
         let max_complete_rewards_slot = Arc::new(AtomicU64::default());
         let ignored_prioritization_fee_cache = Arc::new(PrioritizationFeeCache::new(0u64));
@@ -614,7 +619,8 @@ pub mod tests {
             None,
             gossip_confirmed_slots_receiver,
             alpenglow_vote_sender,
-            alpenglow_vote_receiver,
+            bls_verified_message_sender,
+            bls_verified_message_receiver,
             TvuConfig::default(),
             &Arc::new(MaxSlots::default()),
             None,

core/src/validator.rs

@@ -1596,6 +1596,7 @@ impl Validator {
             bank_notification_sender.clone(),
             duplicate_confirmed_slots_receiver,
             alpenglow_vote_sender.clone(),
+            bls_verified_message_sender.clone(),
             bls_verified_message_receiver,
             TvuConfig {
                 max_ledger_shreds: config.max_ledger_shreds,

votor/src/certificate_pool_service.rs

@@ -51,7 +51,6 @@ pub(crate) struct CertificatePoolContext {
     // just like regular votes. However do we need to convert
     // Vote -> BLSMessage -> Vote?
     // consider adding a separate pathway in cert_pool.add_transaction for ingesting own votes
-    pub(crate) own_vote_receiver: BLSVerifiedMessageReceiver,
     pub(crate) bls_receiver: BLSVerifiedMessageReceiver,
 
     pub(crate) bls_sender: Sender<BLSOp>,
@@ -248,9 +247,6 @@ impl CertificatePoolService {
             }
 
             let bls_messages: Vec<BLSMessage> = select! {
-                recv(ctx.own_vote_receiver) -> msg => {
-                    std::iter::once(msg?).chain(ctx.own_vote_receiver.try_iter()).collect()
-                },
                 recv(ctx.bls_receiver) -> msg => {
                     std::iter::once(msg?).chain(ctx.bls_receiver.try_iter()).collect()
                 },

votor/src/event_handler.rs

@@ -41,11 +41,7 @@ pub(crate) struct EventHandlerContext {
     pub(crate) exit: Arc<AtomicBool>,
     pub(crate) start: Arc<(Mutex<bool>, Condvar)>,
 
-    // VotorEvent receivers
     pub(crate) event_receiver: VotorEventReceiver,
-    pub(crate) skip_timeout_receiver: VotorEventReceiver,
-
-    // Skip timer
     pub(crate) skip_timer: Arc<RwLock<SkipTimerManager>>,
 
     // Contexts
@@ -97,7 +93,6 @@ impl EventHandler {
             exit,
             start,
             event_receiver,
-            skip_timeout_receiver,
             skip_timer,
             shared_context: ctx,
             voting_context: mut vctx,
@@ -133,9 +128,6 @@ impl EventHandler {
                 recv(event_receiver) -> msg => {
                     msg?
                 },
-                recv(skip_timeout_receiver) -> msg => {
-                    msg?
-                }
                 default(Duration::from_secs(1))  => continue
             };
 

votor/src/votor.rs

@@ -54,7 +54,7 @@ use {
         CertificateId,
     },
     alpenglow_vote::bls_message::CertificateMessage,
-    crossbeam_channel::{bounded, Sender},
+    crossbeam_channel::Sender,
     solana_gossip::cluster_info::ClusterInfo,
     solana_ledger::{blockstore::Blockstore, leader_schedule_cache::LeaderScheduleCache},
     solana_pubkey::Pubkey,
@@ -63,9 +63,11 @@ use {
         rpc_subscriptions::RpcSubscriptions,
     },
     solana_runtime::{
-        accounts_background_service::AbsRequestSender, bank_forks::BankForks,
-        installed_scheduler_pool::BankWithScheduler, root_bank_cache::RootBankCache,
-        vote_sender_types::BLSVerifiedMessageReceiver,
+        accounts_background_service::AbsRequestSender,
+        bank_forks::BankForks,
+        installed_scheduler_pool::BankWithScheduler,
+        root_bank_cache::RootBankCache,
+        vote_sender_types::{BLSVerifiedMessageReceiver, BLSVerifiedMessageSender},
     },
     solana_sdk::{clock::Slot, signature::Keypair, signer::Signer},
     std::{
@@ -113,6 +115,7 @@ pub struct VotorConfig {
     pub leader_window_notifier: Arc<LeaderWindowNotifier>,
     pub certificate_sender: Sender<(CertificateId, CertificateMessage)>,
     pub event_sender: VotorEventSender,
+    pub own_vote_sender: BLSVerifiedMessageSender,
 
     // Receivers
     pub event_receiver: VotorEventReceiver,
@@ -164,6 +167,7 @@ impl Votor {
             certificate_sender,
             event_sender,
             event_receiver,
+            own_vote_sender,
             bls_receiver,
         } = config;
 
@@ -173,10 +177,6 @@ impl Votor {
         let my_pubkey = identity_keypair.pubkey();
         let has_new_vote_been_rooted = !wait_for_vote_to_start_leader;
 
-        // These should not backup, TODO: add metrics for length
-        let (skip_timeout_sender, skip_timeout_receiver) = bounded(1000);
-        let (own_vote_sender, own_vote_receiver) = bounded(1000);
-
         let shared_context = SharedContext {
             blockstore: blockstore.clone(),
             bank_forks: bank_forks.clone(),
@@ -209,13 +209,12 @@ impl Votor {
         };
 
         let (skip_timer_service, skip_timer) =
-            SkipTimerService::new(exit.clone(), 100, skip_timeout_sender);
+            SkipTimerService::new(exit.clone(), 100, event_sender.clone());
 
         let event_handler_context = EventHandlerContext {
             exit: exit.clone(),
             start: start.clone(),
             event_receiver,
-            skip_timeout_receiver,
             skip_timer,
             shared_context,
             voting_context,
@@ -230,7 +229,6 @@ impl Votor {
             blockstore,
             root_bank_cache: RootBankCache::new(bank_forks.clone()),
             leader_schedule_cache,
-            own_vote_receiver,
             bls_receiver,
             bls_sender,
             event_sender,