Merge pull request #3115 from ljedrz/cache_latest_leader

[Optimize] Cache the latest leader

node/bft/ledger-service/Cargo.toml

@@ -18,7 +18,7 @@ edition = "2021"
 
 [features]
 default = [ ]
-ledger = [ "parking_lot", "rand", "tokio", "tracing" ]
+ledger = [ "lru", "parking_lot", "rand", "tokio", "tracing" ]
 ledger-write = [ ]
 mock = [ "parking_lot", "tracing" ]
 prover = [ ]
@@ -34,6 +34,7 @@ features = [ "serde", "rayon" ]
 
 [dependencies.lru]
 version = "0.12"
+optional = true
 
 [dependencies.parking_lot]
 version = "0.12"

node/bft/ledger-service/src/ledger.rs

@@ -22,12 +22,12 @@ use snarkvm::{
         store::ConsensusStorage,
         Ledger,
     },
-    prelude::{bail, Field, Network, Result},
+    prelude::{bail, Address, Field, Network, Result},
 };
 
 use indexmap::IndexMap;
 use lru::LruCache;
-use parking_lot::Mutex;
+use parking_lot::{Mutex, RwLock};
 use std::{
     fmt,
     ops::Range,
@@ -41,10 +41,12 @@ use std::{
 const COMMITTEE_CACHE_SIZE: usize = 16;
 
 /// A core ledger service.
+#[allow(clippy::type_complexity)]
 pub struct CoreLedgerService<N: Network, C: ConsensusStorage<N>> {
     ledger: Ledger<N, C>,
     coinbase_verifying_key: Arc<CoinbaseVerifyingKey<N>>,
     committee_cache: Arc<Mutex<LruCache<u64, Committee<N>>>>,
+    latest_leader: Arc<RwLock<Option<(u64, Address<N>)>>>,
     shutdown: Arc<AtomicBool>,
 }
 
@@ -53,7 +55,7 @@ impl<N: Network, C: ConsensusStorage<N>> CoreLedgerService<N, C> {
     pub fn new(ledger: Ledger<N, C>, shutdown: Arc<AtomicBool>) -> Self {
         let coinbase_verifying_key = Arc::new(ledger.coinbase_puzzle().coinbase_verifying_key().clone());
         let committee_cache = Arc::new(Mutex::new(LruCache::new(COMMITTEE_CACHE_SIZE.try_into().unwrap())));
-        Self { ledger, coinbase_verifying_key, committee_cache, shutdown }
+        Self { ledger, coinbase_verifying_key, committee_cache, latest_leader: Default::default(), shutdown }
     }
 }
 
@@ -81,6 +83,16 @@ impl<N: Network, C: ConsensusStorage<N>> LedgerService<N> for CoreLedgerService<
         self.ledger.latest_block()
     }
 
+    /// Returns the latest cached leader and its associated round.
+    fn latest_leader(&self) -> Option<(u64, Address<N>)> {
+        *self.latest_leader.read()
+    }
+
+    /// Updates the latest cached leader and its associated round.
+    fn update_latest_leader(&self, round: u64, leader: Address<N>) {
+        *self.latest_leader.write() = Some((round, leader));
+    }
+
     /// Returns `true` if the given block height exists in the ledger.
     fn contains_block_height(&self, height: u32) -> bool {
         self.ledger.contains_block_height(height).unwrap_or(false)

node/bft/ledger-service/src/mock.rs

@@ -20,7 +20,7 @@ use snarkvm::{
         committee::Committee,
         narwhal::{BatchCertificate, Data, Subdag, Transmission, TransmissionID},
     },
-    prelude::{bail, ensure, Field, Network, Result},
+    prelude::{bail, ensure, Address, Field, Network, Result},
 };
 
 use indexmap::IndexMap;
@@ -68,6 +68,14 @@ impl<N: Network> LedgerService<N> for MockLedgerService<N> {
         unreachable!("MockLedgerService does not support latest_block")
     }
 
+    /// Returns the latest cached leader and its associated round.
+    fn latest_leader(&self) -> Option<(u64, Address<N>)> {
+        None
+    }
+
+    /// Updates the latest cached leader and its associated round.
+    fn update_latest_leader(&self, _round: u64, _leader: Address<N>) {}
+
     /// Returns `true` if the given block height exists in the canonical ledger.
     fn contains_block_height(&self, height: u32) -> bool {
         self.height_to_hash.lock().contains_key(&height)

node/bft/ledger-service/src/prover.rs

@@ -20,7 +20,7 @@ use snarkvm::{
         committee::Committee,
         narwhal::{BatchCertificate, Data, Subdag, Transmission, TransmissionID},
     },
-    prelude::{bail, Field, Network, Result},
+    prelude::{bail, Address, Field, Network, Result},
 };
 
 use indexmap::IndexMap;
@@ -56,6 +56,16 @@ impl<N: Network> LedgerService<N> for ProverLedgerService<N> {
         unreachable!("Latest block does not exist in prover")
     }
 
+    /// Returns the latest cached leader and its associated round.
+    fn latest_leader(&self) -> Option<(u64, Address<N>)> {
+        unreachable!("Latest leader does not exist in prover");
+    }
+
+    /// Updates the latest cached leader and its associated round.
+    fn update_latest_leader(&self, _round: u64, _leader: Address<N>) {
+        unreachable!("Latest leader does not exist in prover");
+    }
+
     /// Returns `true` if the given block height exists in the ledger.
     fn contains_block_height(&self, _height: u32) -> bool {
         false

node/bft/ledger-service/src/traits.rs

@@ -19,7 +19,7 @@ use snarkvm::{
         committee::Committee,
         narwhal::{BatchCertificate, Data, Subdag, Transmission, TransmissionID},
     },
-    prelude::{Field, Network, Result},
+    prelude::{Address, Field, Network, Result},
 };
 
 use indexmap::IndexMap;
@@ -36,6 +36,12 @@ pub trait LedgerService<N: Network>: Debug + Send + Sync {
     /// Returns the latest block in the ledger.
     fn latest_block(&self) -> Block<N>;
 
+    /// Returns the latest cached leader and its associated round.
+    fn latest_leader(&self) -> Option<(u64, Address<N>)>;
+
+    /// Updates the latest cached leader and its associated round.
+    fn update_latest_leader(&self, round: u64, leader: Address<N>);
+
     /// Returns `true` if the given block height exists in the ledger.
     fn contains_block_height(&self, height: u32) -> bool;
 

node/bft/ledger-service/src/translucent.rs

@@ -24,7 +24,7 @@ use snarkvm::{
         store::ConsensusStorage,
         Ledger,
     },
-    prelude::{narwhal::BatchCertificate, Field, Network, Result},
+    prelude::{narwhal::BatchCertificate, Address, Field, Network, Result},
 };
 use std::{
     fmt,
@@ -67,6 +67,16 @@ impl<N: Network, C: ConsensusStorage<N>> LedgerService<N> for TranslucentLedgerS
         self.inner.latest_block()
     }
 
+    /// Returns the latest cached leader and its associated round.
+    fn latest_leader(&self) -> Option<(u64, Address<N>)> {
+        self.inner.latest_leader()
+    }
+
+    /// Updates the latest cached leader and its associated round.
+    fn update_latest_leader(&self, round: u64, leader: Address<N>) {
+        self.inner.update_latest_leader(round, leader);
+    }
+
     /// Returns `true` if the given block height exists in the ledger.
     fn contains_block_height(&self, height: u32) -> bool {
         self.inner.contains_block_height(height)

node/bft/src/bft.rs

@@ -298,11 +298,22 @@ impl<N: Network> BFT<N> {
             }
         };
         // Determine the leader of the current round.
-        let leader = match committee_lookback.get_leader(current_round) {
-            Ok(leader) => leader,
-            Err(e) => {
-                error!("BFT failed to compute the leader for the even round {current_round} - {e}");
-                return false;
+        let leader = match self.ledger().latest_leader() {
+            Some((cached_round, cached_leader)) if cached_round == current_round => cached_leader,
+            _ => {
+                // Compute the leader for the current round.
+                let computed_leader = match committee_lookback.get_leader(current_round) {
+                    Ok(leader) => leader,
+                    Err(e) => {
+                        error!("BFT failed to compute the leader for the even round {current_round} - {e}");
+                        return false;
+                    }
+                };
+
+                // Cache the computed leader.
+                self.ledger().update_latest_leader(current_round, computed_leader);
+
+                computed_leader
             }
         };
         // Find and set the leader certificate, if the leader was present in the current even round.
@@ -453,10 +464,23 @@ impl<N: Network> BFT<N> {
         let Ok(committee_lookback) = self.ledger().get_committee_lookback_for_round(commit_round) else {
             bail!("BFT failed to retrieve the committee with lag for commit round {commit_round}");
         };
-        // Compute the leader for the commit round.
-        let Ok(leader) = committee_lookback.get_leader(commit_round) else {
-            bail!("BFT failed to compute the leader for commit round {commit_round}");
+
+        // Either retrieve the cached leader or compute it.
+        let leader = match self.ledger().latest_leader() {
+            Some((cached_round, cached_leader)) if cached_round == commit_round => cached_leader,
+            _ => {
+                // Compute the leader for the commit round.
+                let Ok(computed_leader) = committee_lookback.get_leader(commit_round) else {
+                    bail!("BFT failed to compute the leader for commit round {commit_round}");
+                };
+
+                // Cache the computed leader.
+                self.ledger().update_latest_leader(commit_round, computed_leader);
+
+                computed_leader
+            }
         };
+
         // Retrieve the leader certificate for the commit round.
         let Some(leader_certificate) = self.dag.read().get_certificate_for_round_with_author(commit_round, leader)
         else {
@@ -514,11 +538,22 @@ impl<N: Network> BFT<N> {
                         bail!("BFT failed to retrieve a previous committee lookback for the even round {round} - {e}");
                     }
                 };
-                // Compute the leader address for the leader round.
-                let leader = match previous_committee_lookback.get_leader(round) {
-                    Ok(leader) => leader,
-                    Err(e) => {
-                        bail!("BFT failed to compute the leader for the even round {round} - {e}");
+                // Either retrieve the cached leader or compute it.
+                let leader = match self.ledger().latest_leader() {
+                    Some((cached_round, cached_leader)) if cached_round == round => cached_leader,
+                    _ => {
+                        // Compute the leader for the commit round.
+                        let computed_leader = match previous_committee_lookback.get_leader(round) {
+                            Ok(leader) => leader,
+                            Err(e) => {
+                                bail!("BFT failed to compute the leader for the even round {round} - {e}");
+                            }
+                        };
+
+                        // Cache the computed leader.
+                        self.ledger().update_latest_leader(round, computed_leader);
+
+                        computed_leader
                     }
                 };
                 // Retrieve the previous leader certificate.

node/bft/src/worker.rs

@@ -477,6 +477,7 @@ mod tests {
             committee::Committee,
             narwhal::{BatchCertificate, Subdag, Transmission, TransmissionID},
         },
+        prelude::Address,
     };
 
     use bytes::Bytes;
@@ -505,6 +506,8 @@ mod tests {
             fn latest_round(&self) -> u64;
             fn latest_block_height(&self) -> u32;
             fn latest_block(&self) -> Block<N>;
+            fn latest_leader(&self) -> Option<(u64, Address<N>)>;
+            fn update_latest_leader(&self, round: u64, leader: Address<N>);
             fn contains_block_height(&self, height: u32) -> bool;
             fn get_block_height(&self, hash: &N::BlockHash) -> Result<u32>;
             fn get_block_hash(&self, height: u32) -> Result<N::BlockHash>;