refactor: rip out the epoch stakes service. (#350)

Author: akhi3030

core/src/sigverifier/bls_sigverifier.rs

@@ -11,17 +11,27 @@ use {
     crossbeam_channel::{Sender, TrySendError},
     solana_clock::Slot,
     solana_pubkey::Pubkey,
-    solana_runtime::epoch_stakes_service::EpochStakesService,
+    solana_runtime::{
+        bank::Bank, epoch_stakes::BLSPubkeyToRankMap, root_bank_cache::RootBankCache,
+    },
     solana_streamer::packet::PacketBatch,
     solana_votor_messages::bls_message::BLSMessage,
     stats::{BLSSigVerifierStats, StatsUpdater},
     std::{collections::HashMap, sync::Arc},
 };
 
+fn get_key_to_rank_map(bank: &Bank, slot: Slot) -> Option<&Arc<BLSPubkeyToRankMap>> {
+    let stakes = bank.epoch_stakes_map();
+    let epoch = bank.epoch_schedule().get_epoch(slot);
+    stakes
+        .get(&epoch)
+        .map(|stake| stake.bls_pubkey_to_rank_map())
+}
+
 pub struct BLSSigVerifier {
     verified_votes_sender: VerifiedVoteSender,
     message_sender: Sender<BLSMessage>,
-    epoch_stakes_service: Arc<EpochStakesService>,
+    root_bank_cache: RootBankCache,
     stats: BLSSigVerifierStats,
 }
 
@@ -60,8 +70,8 @@ impl SigVerifier for BLSSigVerifier {
                 }
             };
 
-            let Some(rank_to_pubkey_map) = self.epoch_stakes_service.get_key_to_rank_map(slot)
-            else {
+            let bank = self.root_bank_cache.root_bank();
+            let Some(rank_to_pubkey_map) = get_key_to_rank_map(&bank, slot) else {
                 stats_updater.received_no_epoch_stakes += 1;
                 continue;
             };
@@ -102,12 +112,12 @@ impl SigVerifier for BLSSigVerifier {
 
 impl BLSSigVerifier {
     pub fn new(
-        epoch_stakes_service: Arc<EpochStakesService>,
+        root_bank_cache: RootBankCache,
         verified_votes_sender: VerifiedVoteSender,
         message_sender: Sender<BLSMessage>,
     ) -> Self {
         Self {
-            epoch_stakes_service,
+            root_bank_cache,
             verified_votes_sender,
             message_sender,
             stats: BLSSigVerifierStats::new(),
@@ -137,12 +147,13 @@ mod tests {
     use {
         super::*,
         bitvec::prelude::*,
-        crossbeam_channel::{unbounded, Receiver},
+        crossbeam_channel::Receiver,
         solana_bls_signatures::Signature,
         solana_hash::Hash,
         solana_perf::packet::{Packet, PinnedPacketBatch},
         solana_runtime::{
             bank::Bank,
+            bank_forks::BankForks,
             genesis_utils::{
                 create_genesis_config_with_alpenglow_vote_accounts_no_program,
                 ValidatorVoteKeypairs,
@@ -175,13 +186,12 @@ mod tests {
             &validator_keypairs,
             stakes_vec,
         );
-        let bank = Arc::new(Bank::new_for_tests(&genesis.genesis_config));
-        let epoch = bank.epoch();
-        let (_tx, rx) = unbounded();
-        let epoch_stakes_service = Arc::new(EpochStakesService::new(bank, epoch, rx));
+        let bank0 = Bank::new_for_tests(&genesis.genesis_config);
+        let bank_forks = BankForks::new_rw_arc(bank0);
+        let root_bank_cache = RootBankCache::new(bank_forks);
         (
             validator_keypairs,
-            BLSSigVerifier::new(epoch_stakes_service, verified_vote_sender, message_sender),
+            BLSSigVerifier::new(root_bank_cache, verified_vote_sender, message_sender),
         )
     }
 

core/src/tpu.rs

@@ -53,7 +53,6 @@ use {
     },
     solana_runtime::{
         bank_forks::BankForks,
-        epoch_stakes_service::EpochStakesService,
         prioritization_fee_cache::PrioritizationFeeCache,
         root_bank_cache::RootBankCache,
         vote_sender_types::{BLSVerifiedMessageSender, ReplayVoteReceiver, ReplayVoteSender},
@@ -325,13 +324,9 @@ impl Tpu {
         };
 
         let alpenglow_sigverify_stage = {
-            let (tx, rx) = unbounded();
-            bank_forks.write().unwrap().register_new_bank_subscriber(tx);
-            let bank = bank_forks.read().unwrap().root_bank();
-            let epoch = bank.epoch();
-            let epoch_stakes_service = Arc::new(EpochStakesService::new(bank, epoch, rx));
+            let root_bank_cache = RootBankCache::new(bank_forks.clone());
             let verifier = BLSSigVerifier::new(
-                epoch_stakes_service,
+                root_bank_cache,
                 verified_vote_sender.clone(),
                 bls_verified_message_sender,
             );

runtime/src/bank_forks.rs

@@ -9,7 +9,6 @@ use {
         },
         snapshot_controller::SnapshotController,
     },
-    crossbeam_channel::{SendError, Sender},
     log::*,
     solana_clock::{BankId, Slot},
     solana_hash::Hash,
@@ -74,8 +73,6 @@ pub struct BankForks {
     highest_slot_at_startup: Slot,
     scheduler_pool: Option<InstalledSchedulerPoolArc>,
     dumped_slot_subscribers: Vec<DumpedSlotSubscription>,
-    /// Tracks subscribers interested in hearing about new `Bank`s.
-    new_bank_subscribers: Vec<Sender<Arc<Bank>>>,
 }
 
 impl Index<u64> for BankForks {
@@ -124,7 +121,6 @@ impl BankForks {
             highest_slot_at_startup: 0,
             scheduler_pool: None,
             dumped_slot_subscribers: vec![],
-            new_bank_subscribers: vec![],
         }));
 
         root_bank.set_fork_graph_in_program_cache(Arc::downgrade(&bank_forks));
@@ -323,24 +319,6 @@ impl BankForks {
         self.dumped_slot_subscribers.push(notifier);
     }
 
-    /// Register a new subscriber interested in hearing about new `Bank`s.
-    pub fn register_new_bank_subscriber(&mut self, tx: Sender<Arc<Bank>>) {
-        self.new_bank_subscribers.push(tx);
-    }
-
-    /// Call to notify subscribers of new `Bank`s.
-    fn notify_new_bank_subscribers(&mut self, root_bank: &Arc<Bank>) {
-        let mut channels_to_drop = vec![];
-        for (ind, tx) in self.new_bank_subscribers.iter().enumerate() {
-            if let Err(SendError(_)) = tx.send(root_bank.clone()) {
-                channels_to_drop.push(ind);
-            }
-        }
-        for ind in channels_to_drop {
-            self.new_bank_subscribers.remove(ind);
-        }
-    }
-
     /// Clears associated banks from BankForks and notifies subscribers that a dump has occured.
     pub fn dump_slots<'a, I>(&mut self, slots: I) -> (Vec<(Slot, BankId)>, Vec<BankWithScheduler>)
     where
@@ -403,7 +381,6 @@ impl BankForks {
                 "Clearing epoch rewards cache for epoch {old_epoch} after setting root to slot {root}"
             );
             root_bank.clear_epoch_rewards_cache();
-            self.notify_new_bank_subscribers(root_bank);
         }
         let root_tx_count = root_bank
             .parents()

runtime/src/lib.rs

@@ -11,7 +11,6 @@ pub mod bank_utils;
 pub mod commitment;
 pub mod dependency_tracker;
 pub mod epoch_stakes;
-pub mod epoch_stakes_service;
 pub mod genesis_utils;
 pub mod inflation_rewards;
 pub mod installed_scheduler_pool;