use channels and threads instead of locks

Author: akhi3030

core/src/block_creation_loop.rs

@@ -9,8 +9,7 @@ use {
         banking_trace::BankingTracer,
         replay_stage::{Finalizer, ReplayStage},
     },
-    crossbeam_channel::Receiver,
-    parking_lot::RwLock as PLRwLock,
+    crossbeam_channel::{Receiver, Sender},
     solana_clock::Slot,
     solana_entry::block_component::BlockFooterV1,
     solana_gossip::cluster_info::ClusterInfo,
@@ -34,7 +33,9 @@ use {
     },
     solana_version::version,
     solana_votor::{
-        common::block_timeout, consensus_rewards::ConsensusRewards, event::LeaderWindowInfo,
+        common::block_timeout,
+        consensus_rewards::{BuildRewardCertsRequest, BuildRewardCertsResponse},
+        event::LeaderWindowInfo,
     },
     solana_votor_messages::rewards_certificate::{NotarRewardCertificate, SkipRewardCertificate},
     stats::{BlockCreationLoopMetrics, SlotMetrics},
@@ -82,7 +83,6 @@ pub struct BlockCreationLoopConfig {
     pub poh_recorder: Arc<RwLock<PohRecorder>>,
     pub leader_schedule_cache: Arc<LeaderScheduleCache>,
     pub rpc_subscriptions: Option<Arc<RpcSubscriptions>>,
-    pub consensus_rewards: Arc<PLRwLock<ConsensusRewards>>,
 
     // Notifiers
     pub banking_tracer: Arc<BankingTracer>,
@@ -96,6 +96,9 @@ pub struct BlockCreationLoopConfig {
     // Channel to receive RecordReceiver from PohService
     pub record_receiver_receiver: Receiver<RecordReceiver>,
     pub optimistic_parent_receiver: Receiver<LeaderWindowInfo>,
+
+    pub build_reward_certs_sender: Sender<BuildRewardCertsRequest>,
+    pub reward_certs_receiver: Receiver<BuildRewardCertsResponse>,
 }
 
 struct LeaderContext {
@@ -113,7 +116,8 @@ struct LeaderContext {
     slot_status_notifier: Option<SlotStatusNotifier>,
     banking_tracer: Arc<BankingTracer>,
     replay_highest_frozen: Arc<ReplayHighestFrozen>,
-    consensus_rewards: Arc<PLRwLock<ConsensusRewards>>,
+    build_reward_certs_sender: Sender<BuildRewardCertsRequest>,
+    reward_certs_receiver: Receiver<BuildRewardCertsResponse>,
 
     // Metrics
     metrics: BlockCreationLoopMetrics,
@@ -165,7 +169,8 @@ fn start_loop(config: BlockCreationLoopConfig) {
         replay_highest_frozen,
         highest_parent_ready,
         optimistic_parent_receiver,
-        consensus_rewards,
+        build_reward_certs_sender,
+        reward_certs_receiver,
     } = config;
 
     // Similar to Votor, if this loop dies kill the validator
@@ -203,7 +208,8 @@ fn start_loop(config: BlockCreationLoopConfig) {
         replay_highest_frozen,
         metrics: BlockCreationLoopMetrics::default(),
         slot_metrics: SlotMetrics::default(),
-        consensus_rewards,
+        build_reward_certs_sender,
+        reward_certs_receiver,
     };
 
     // Setup poh
@@ -324,7 +330,8 @@ fn produce_window(
         if let Err(e) = record_and_complete_block(
             ctx.poh_recorder.as_ref(),
             &mut ctx.record_receiver,
-            ctx.consensus_rewards.clone(),
+            &ctx.build_reward_certs_sender,
+            &ctx.reward_certs_receiver,
             skip_timer,
             timeout,
         ) {
@@ -424,17 +431,16 @@ fn produce_block_footer(
 fn record_and_complete_block(
     poh_recorder: &RwLock<PohRecorder>,
     record_receiver: &mut RecordReceiver,
-    consensus_rewards: Arc<PLRwLock<ConsensusRewards>>,
+    build_reward_certs_sender: &Sender<BuildRewardCertsRequest>,
+    cert_receiver: &Receiver<BuildRewardCertsResponse>,
     block_timer: Instant,
     block_timeout: Duration,
 ) -> Result<(), PohRecorderError> {
-    // Taking a read lock on consensus_rewards can contend with the write lock in bls_sigverifier.
-    // We are ready to produce the block footer now, while we gather other bits of data, we can block on the consensus_rewards lock in a separate thread to minimise contention.
-    let handle = std::thread::spawn(move || {
-        // XXX: how to look up the slot.
-        let slot = u64::MAX;
-        consensus_rewards.read().build_rewards_certs(slot)
-    });
+    // XXX: how to look up the slot.
+    let slot = u64::MAX;
+    build_reward_certs_sender
+        .send(BuildRewardCertsRequest { slot })
+        .unwrap();
     loop {
         let remaining_slot_time = block_timeout.saturating_sub(block_timer.elapsed());
         if remaining_slot_time.is_zero() {
@@ -485,11 +491,11 @@ fn record_and_complete_block(
 
     // Produce the footer with the current timestamp
     let working_bank = w_poh_recorder.working_bank().unwrap();
-    let (skip_reward_certificate, notar_reward_certificate) = handle.join().unwrap();
+    let resp = cert_receiver.recv().unwrap();
     let footer = produce_block_footer(
         working_bank.bank.clone_without_scheduler(),
-        skip_reward_certificate,
-        notar_reward_certificate,
+        resp.skip,
+        resp.notar,
     );
 
     BlockComponentProcessor::update_bank_with_footer(

core/src/bls_sigverify/bls_sigverifier.rs

@@ -9,7 +9,6 @@ use {
     },
     bitvec::prelude::{BitVec, Lsb0},
     crossbeam_channel::{Sender, TrySendError},
-    parking_lot::RwLock as PLRwLock,
     rayon::iter::{
         IndexedParallelIterator, IntoParallelIterator, IntoParallelRefIterator, ParallelIterator,
     },
@@ -26,7 +25,7 @@ use {
     solana_streamer::packet::PacketBatch,
     solana_votor::{
         consensus_metrics::{ConsensusMetricsEvent, ConsensusMetricsEventSender},
-        consensus_rewards::ConsensusRewards,
+        consensus_rewards::{AddVoteEntry, AddVoteMessage},
     },
     solana_votor_messages::{
         consensus_message::{Certificate, CertificateType, ConsensusMessage, VoteMessage},
@@ -88,6 +87,7 @@ enum CertVerifyError {
 
 pub struct BLSSigVerifier {
     verified_votes_sender: VerifiedVoteSender,
+    reward_votes_sender: Sender<AddVoteMessage>,
     message_sender: Sender<ConsensusMessage>,
     sharable_banks: SharableBanks,
     stats: BLSSigVerifierStats,
@@ -96,7 +96,6 @@ pub struct BLSSigVerifier {
     consensus_metrics_sender: ConsensusMetricsEventSender,
     last_checked_root_slot: Slot,
     alpenglow_last_voted: Arc<AlpenglowLastVoted>,
-    consensus_rewards: Arc<PLRwLock<ConsensusRewards>>,
 }
 
 impl BLSSigVerifier {
@@ -180,14 +179,8 @@ impl BLSSigVerifier {
 
                     // consensus pool does not need votes for slots other than root slot however the rewards container may still need them.
                     if vote_message.vote.slot() <= root_bank.slot() {
-                        // the only module that takes a write lock on consensus_rewards is this one and it does not take the write lock while it is verifying votes so this should never block.
-                        if self
-                            .consensus_rewards
-                            .read()
-                            .wants_vote(root_bank.slot(), &vote_message)
-                        {
-                            // XXX: actually verify and send the votes.  The verification and sending should happen off the critical path.
-                        }
+                        // XXX: actually verify and send the votes.  The verification and sending should happen off the critical path.
+
                         self.stats.received_old.fetch_add(1, Ordering::Relaxed);
                         packet.meta_mut().set_discard(true);
                         continue;
@@ -249,19 +242,20 @@ impl BLSSigVerifier {
             warn!("could not send consensus metrics, receive side of channel is closed");
         }
 
-        {
-            // This should be the only place that is taking a write lock on consensus_rewards.
-            // This lock should not contend with any other operations in this module.
-            // It can contend with the read lock in the block creation loop though as such we should hold it for as little time as possible.
-            let mut guard = self.consensus_rewards.write();
-            let root_slot = root_bank.slot();
-            for v in rewards_votes {
-                if let Some(rank_map) = get_key_to_rank_map(&root_bank, v.vote.slot()) {
-                    let max_validators = rank_map.len();
-                    guard.add_vote(root_slot, max_validators, v);
-                }
-            }
-        }
+        let votes = rewards_votes
+            .into_iter()
+            .filter_map(|v| {
+                get_key_to_rank_map(&root_bank, v.vote.slot()).map(|rank_map| AddVoteEntry {
+                    max_validators: rank_map.len(),
+                    vote: v,
+                })
+            })
+            .collect();
+        let msg = AddVoteMessage {
+            root_slot: root_bank.slot(),
+            votes,
+        };
+        self.reward_votes_sender.send(msg).unwrap();
 
         self.stats.maybe_report_stats();
 
@@ -273,22 +267,22 @@ impl BLSSigVerifier {
     pub fn new(
         sharable_banks: SharableBanks,
         verified_votes_sender: VerifiedVoteSender,
+        reward_votes_sender: Sender<AddVoteMessage>,
         message_sender: Sender<ConsensusMessage>,
         consensus_metrics_sender: ConsensusMetricsEventSender,
         alpenglow_last_voted: Arc<AlpenglowLastVoted>,
-        consensus_rewards: Arc<PLRwLock<ConsensusRewards>>,
     ) -> Self {
         Self {
             sharable_banks,
             verified_votes_sender,
+            reward_votes_sender,
             message_sender,
             stats: BLSSigVerifierStats::new(),
             verified_certs: RwLock::new(HashSet::new()),
             vote_payload_cache: RwLock::new(HashMap::new()),
             consensus_metrics_sender,
             last_checked_root_slot: 0,
             alpenglow_last_voted,
-            consensus_rewards,
         }
     }
 
@@ -299,20 +293,8 @@ impl BLSSigVerifier {
         votes_to_verify: Vec<VoteToVerify>,
     ) -> Result<Vec<VoteMessage>, BLSSigVerifyServiceError<ConsensusMessage>> {
         let verified_votes = self.verify_votes(votes_to_verify);
-
-        let rewards_votes = {
-            // the only module that takes a write lock on consensus_rewards is this one and it does not take the write lock while it is verifying votes so this should never block.
-            let guard = self.consensus_rewards.read();
-            let root_slot = self.sharable_banks.root().slot();
-            verified_votes
-                .iter()
-                .filter_map(|vote| {
-                    guard
-                        .wants_vote(root_slot, &vote.vote_message)
-                        .then_some(vote.vote_message)
-                })
-                .collect()
-        };
+        // XXX: call consensus_rewards::wants_vote() to do some additional filtering here.
+        let reward_votes = verified_votes.iter().map(|v| v.vote_message).collect();
 
         self.stats
             .total_valid_packets
@@ -366,7 +348,7 @@ impl BLSSigVerifier {
             }
         }
 
-        Ok(rewards_votes)
+        Ok(reward_votes)
     }
 
     fn verify_votes(&self, votes_to_verify: Vec<VoteToVerify>) -> Vec<VoteToVerify> {

core/src/replay_stage.rs

@@ -86,6 +86,7 @@ use {
     solana_vote::vote_transaction::VoteTransaction,
     solana_votor::{
         consensus_metrics::{ConsensusMetricsEventReceiver, ConsensusMetricsEventSender},
+        consensus_rewards::{AddVoteMessage, BuildRewardCertsRequest, BuildRewardCertsResponse},
         event::{
             CompletedBlock, LeaderWindowInfo, VotorEvent, VotorEventReceiver, VotorEventSender,
         },
@@ -299,6 +300,9 @@ pub struct ReplayStageConfig {
     pub consensus_metrics_sender: ConsensusMetricsEventSender,
     pub consensus_metrics_receiver: ConsensusMetricsEventReceiver,
     pub migration_status: Arc<MigrationStatus>,
+    pub reward_certs_sender: Sender<BuildRewardCertsResponse>,
+    pub reward_votes_receiver: Receiver<AddVoteMessage>,
+    pub build_reward_certs_receiver: Receiver<BuildRewardCertsRequest>,
 }
 
 pub struct ReplaySenders {
@@ -620,6 +624,9 @@ impl ReplayStage {
             consensus_metrics_sender,
             consensus_metrics_receiver,
             migration_status,
+            reward_certs_sender,
+            reward_votes_receiver,
+            build_reward_certs_receiver,
         } = config;
 
         let ReplaySenders {
@@ -694,6 +701,9 @@ impl ReplayStage {
             consensus_metrics_sender,
             consensus_metrics_receiver,
             migration_status: migration_status.clone(),
+            reward_certs_sender,
+            reward_votes_receiver,
+            build_reward_certs_receiver,
         };
         let votor = Votor::new(votor_config);
 

core/src/tvu.rs

@@ -27,7 +27,6 @@ use {
     },
     bytes::Bytes,
     crossbeam_channel::{bounded, unbounded, Receiver, Sender},
-    parking_lot::RwLock as PLRwLock,
     solana_client::connection_cache::ConnectionCache,
     solana_clock::Slot,
     solana_geyser_plugin_manager::block_metadata_notifier_interface::BlockMetadataNotifierArc,
@@ -62,7 +61,7 @@ use {
     },
     solana_turbine::{retransmit_stage::RetransmitStage, xdp::XdpSender},
     solana_votor::{
-        consensus_rewards::ConsensusRewards,
+        consensus_rewards::{BuildRewardCertsRequest, BuildRewardCertsResponse},
         event::{LeaderWindowInfo, VotorEventReceiver, VotorEventSender},
         vote_history::VoteHistory,
         vote_history_storage::VoteHistoryStorage,
@@ -218,7 +217,8 @@ impl Tvu {
         key_notifiers: Arc<RwLock<KeyUpdaters>>,
         alpenglow_last_voted: Arc<AlpenglowLastVoted>,
         migration_status: Arc<MigrationStatus>,
-        consensus_rewards: Arc<PLRwLock<ConsensusRewards>>,
+        reward_certs_sender: Sender<BuildRewardCertsResponse>,
+        build_reward_certs_receiver: Receiver<BuildRewardCertsRequest>,
     ) -> Result<Self, String> {
         let (consensus_message_sender, consensus_message_receiver) =
             bounded(MAX_ALPENGLOW_PACKET_NUM);
@@ -271,15 +271,16 @@ impl Tvu {
             alpenglow_quic_server_config,
         )
         .unwrap();
+        let (reward_votes_sender, reward_votes_receiver) = unbounded();
         let alpenglow_sigverify_service = {
             let sharable_banks = bank_forks.read().unwrap().sharable_banks();
             let verifier = BLSSigVerifier::new(
                 sharable_banks,
                 verified_vote_sender.clone(),
+                reward_votes_sender,
                 consensus_message_sender.clone(),
                 consensus_metrics_sender.clone(),
                 alpenglow_last_voted.clone(),
-                consensus_rewards,
             );
             BLSSigverifyService::new(bls_packet_receiver, verifier)
         };
@@ -453,6 +454,9 @@ impl Tvu {
             consensus_metrics_sender: consensus_metrics_sender.clone(),
             consensus_metrics_receiver,
             migration_status,
+            reward_certs_sender,
+            reward_votes_receiver,
+            build_reward_certs_receiver,
         };
 
         let voting_service = VotingService::new(