Merge pull request #4065 from ProvableHQ/refactor/no-bft-channels

[Refactor] Remove BFT channels

Author: vicsn

Cargo.lock

@@ -99,21 +99,30 @@ fn check_locktick_imports<P: AsRef<Path>>(path: P) {
 
             // Modify the lock balance based on the type of the relevant import.
             if [ImportOfInterest::ParkingLot, ImportOfInterest::Tokio].contains(&ioi) {
-                if line.contains("Mutex") {
-                    lock_balance += 1;
+                lock_balance += line.matches("Mutex").count() as i8;
+                lock_balance += line.matches("RwLock").count() as i8;
+
+                // A correction in case of the `use tokio::Mutex as TMutex` convention.
+                if line.contains("TMutex") {
+                    lock_balance -= 1;
                 }
-                if line.contains("RwLock") {
-                    lock_balance += 1;
+
+                // Account for lock guards, which do not have a locktick counterpart.
+                if line.contains("MutexGuard") {
+                    lock_balance -= 1;
                 }
-            } else if ioi == ImportOfInterest::Locktick {
-                // Use `matches` instead of just `contains` here, as more than a single
-                // lock type entry is possible in a locktick import.
-                for _hit in line.matches("Mutex") {
+                if line.contains("RwLockReadGuard") {
                     lock_balance -= 1;
                 }
-                for _hit in line.matches("RwLock") {
+                if line.contains("RwLockWriteGuard") {
                     lock_balance -= 1;
                 }
+            } else if ioi == ImportOfInterest::Locktick {
+                // Use `matches` instead of just `contains` here, as more than a single
+                // lock type entry is possible in a locktick import.
+                lock_balance -= line.matches("Mutex").count() as i8;
+                lock_balance -= line.matches("RwLock").count() as i8;
+
                 // A correction in case of the `use tokio::Mutex as TMutex` convention.
                 if line.contains("TMutex") {
                     lock_balance += 1;

build.rs

@@ -200,7 +200,7 @@ pub async fn start_primary(
     let trusted_peers_only = false;
     // Initialize the primary instance.
     let block_sync = Arc::new(BlockSync::new(ledger.clone()));
-    let mut primary = Primary::<CurrentNetwork>::new(
+    let primary = Primary::<CurrentNetwork>::new(
         account,
         storage,
         ledger,
@@ -212,7 +212,7 @@ pub async fn start_primary(
         None,
     )?;
     // Run the primary instance.
-    primary.run(None, None, sender.clone(), receiver).await?;
+    primary.run(None, None, None, sender.clone(), receiver).await?;
     // Handle OS signals.
     handle_signals(&primary);
     // Return the primary instance.

node/bft/examples/simple_node.rs

@@ -15,18 +15,9 @@
 
 use crate::{
     MAX_LEADER_CERTIFICATE_DELAY_IN_SECS,
-    Primary,
-    helpers::{
-        BFTReceiver,
-        ConsensusSender,
-        DAG,
-        PrimaryReceiver,
-        PrimarySender,
-        Storage,
-        fmt_id,
-        init_bft_channels,
-        now,
-    },
+    helpers::{ConsensusSender, DAG, PrimaryReceiver, PrimarySender, Storage, fmt_id, now},
+    primary::{Primary, PrimaryCallback},
+    sync::SyncCallback,
 };
 
 use snarkos_account::Account;
@@ -50,15 +41,11 @@ use anyhow::Context;
 use colored::Colorize;
 use indexmap::{IndexMap, IndexSet};
 #[cfg(feature = "locktick")]
-use locktick::{
-    parking_lot::{Mutex, RwLock},
-    tokio::Mutex as TMutex,
-};
+use locktick::{parking_lot::RwLock, tokio::Mutex as TMutex};
 #[cfg(not(feature = "locktick"))]
-use parking_lot::{Mutex, RwLock};
+use parking_lot::RwLock;
 use std::{
     collections::{BTreeMap, HashSet},
-    future::Future,
     net::SocketAddr,
     sync::{
         Arc,
@@ -67,10 +54,7 @@ use std::{
 };
 #[cfg(not(feature = "locktick"))]
 use tokio::sync::Mutex as TMutex;
-use tokio::{
-    sync::{OnceCell, oneshot},
-    task::JoinHandle,
-};
+use tokio::sync::{OnceCell, oneshot};
 
 #[derive(Clone)]
 pub struct BFT<N: Network> {
@@ -84,8 +68,6 @@ pub struct BFT<N: Network> {
     leader_certificate_timer: Arc<AtomicI64>,
     /// The consensus sender.
     consensus_sender: Arc<OnceCell<ConsensusSender<N>>>,
-    /// Handles for all spawned tasks.
-    handles: Arc<Mutex<Vec<JoinHandle<()>>>>,
     /// The BFT lock.
     lock: Arc<TMutex<()>>,
 }
@@ -120,7 +102,6 @@ impl<N: Network> BFT<N> {
             leader_certificate: Default::default(),
             leader_certificate_timer: Default::default(),
             consensus_sender: Default::default(),
-            handles: Default::default(),
             lock: Default::default(),
         })
     }
@@ -137,14 +118,16 @@ impl<N: Network> BFT<N> {
         primary_receiver: PrimaryReceiver<N>,
     ) -> Result<()> {
         info!("Starting the BFT instance...");
-        // Initialize the BFT channels.
-        let (bft_sender, bft_receiver) = init_bft_channels::<N>();
-        // First, start the BFT handlers.
-        self.start_handlers(bft_receiver);
+        // Set up callbacks.
+        let primary_callback = Some(Arc::new(self.clone()) as Arc<dyn PrimaryCallback<N>>);
+
+        let sync_callback = Some(Arc::new(self.clone()) as Arc<dyn SyncCallback<N>>);
+
         // Next, run the primary instance.
-        self.primary.run(ping, Some(bft_sender), primary_sender, primary_receiver).await?;
+        self.primary.run(ping, primary_callback, sync_callback, primary_sender, primary_receiver).await?;
+
         // Lastly, set the consensus sender.
-        // Note: This ensures during initial syncing, that the BFT does not advance the ledger.
+        // Note: This ensures that the BFT does not advance the ledger during initial syncing.
         if let Some(consensus_sender) = consensus_sender {
             self.consensus_sender.set(consensus_sender).expect("Consensus sender already set");
         }
@@ -226,8 +209,9 @@ impl<N: Network> BFT<N> {
     }
 }
 
-impl<N: Network> BFT<N> {
-    /// Stores the certificate in the DAG, and attempts to commit one or more anchors.
+#[async_trait::async_trait]
+impl<N: Network> PrimaryCallback<N> for BFT<N> {
+    /// Notification that a new round has started.
     fn update_to_next_round(&self, current_round: u64) -> bool {
         // Ensure the current round is at least the storage round (this is a sanity check).
         let storage_round = self.storage().current_round();
@@ -299,6 +283,39 @@ impl<N: Network> BFT<N> {
         is_ready
     }
 
+    /// Notification about a new certificated generated by `Primary` or received by the `Primary` from a peer.
+    async fn add_new_certificate(&self, certificate: BatchCertificate<N>) -> Result<()> {
+        // Update the DAG with the certificate.
+        self.update_dag::<true, false>(certificate).await
+    }
+}
+
+#[async_trait::async_trait]
+impl<N: Network> SyncCallback<N> for BFT<N> {
+    /// Syncs the BFT DAG with the given batch certificates. These batch certificates **must**
+    /// already exist in the ledger.
+    ///
+    /// This method commits all the certificates into the DAG.
+    /// Note that there is no need to insert the certificates into the DAG, because these certificates
+    /// already exist in the ledger and therefore do not need to be re-ordered into future committed subdags.
+    async fn sync_dag_at_bootup(&self, certificates: Vec<BatchCertificate<N>>) {
+        // Acquire the BFT write lock.
+        let mut dag = self.dag.write();
+
+        // Commit all the certificates.
+        for certificate in certificates {
+            dag.commit(&certificate, self.storage().max_gc_rounds());
+        }
+    }
+
+    /// Notification about a new certificate detected by the `Sync` instance after fetching a new block.
+    async fn add_new_certificate(&self, certificate: BatchCertificate<N>) -> Result<()> {
+        // Update the DAG with the certificate.
+        self.update_dag::<true, true>(certificate).await
+    }
+}
+
+impl<N: Network> BFT<N> {
     /// Updates the leader certificate to the current even round,
     /// returning `true` if the BFT is ready to update to the next round.
     ///
@@ -901,86 +918,13 @@ impl<N: Network> BFT<N> {
 }
 
 impl<N: Network> BFT<N> {
-    /// Starts the BFT handlers.
-    fn start_handlers(&self, bft_receiver: BFTReceiver<N>) {
-        let BFTReceiver {
-            mut rx_primary_round,
-            mut rx_primary_certificate,
-            mut rx_sync_bft_dag_at_bootup,
-            mut rx_sync_bft,
-        } = bft_receiver;
-
-        // Process the current round from the primary.
-        let self_ = self.clone();
-        self.spawn(async move {
-            while let Some((current_round, callback)) = rx_primary_round.recv().await {
-                callback.send(self_.update_to_next_round(current_round)).ok();
-            }
-        });
-
-        // Process the certificate from the primary.
-        let self_ = self.clone();
-        self.spawn(async move {
-            while let Some((certificate, callback)) = rx_primary_certificate.recv().await {
-                // Update the DAG with the certificate.
-                let result = self_.update_dag::<true, false>(certificate).await;
-                // Send the callback **after** updating the DAG.
-                // Note: We must await the DAG update before proceeding.
-                callback.send(result).ok();
-            }
-        });
-
-        // Process the request to sync the BFT DAG at bootup.
-        let self_ = self.clone();
-        self.spawn(async move {
-            while let Some(certificates) = rx_sync_bft_dag_at_bootup.recv().await {
-                self_.sync_bft_dag_at_bootup(certificates).await;
-            }
-        });
-
-        // Handler for new certificates that were fetched by the sync module.
-        let self_ = self.clone();
-        self.spawn(async move {
-            while let Some((certificate, callback)) = rx_sync_bft.recv().await {
-                // Update the DAG with the certificate.
-                let result = self_.update_dag::<true, true>(certificate).await;
-                // Send the callback **after** updating the DAG.
-                // Note: We must await the DAG update before proceeding.
-                callback.send(result).ok();
-            }
-        });
-    }
-
-    /// Syncs the BFT DAG with the given batch certificates. These batch certificates **must**
-    /// already exist in the ledger.
-    ///
-    /// This method commits all the certificates into the DAG.
-    /// Note that there is no need to insert the certificates into the DAG, because these certificates
-    /// already exist in the ledger and therefore do not need to be re-ordered into future committed subdags.
-    async fn sync_bft_dag_at_bootup(&self, certificates: Vec<BatchCertificate<N>>) {
-        // Acquire the BFT write lock.
-        let mut dag = self.dag.write();
-
-        // Commit all the certificates.
-        for certificate in certificates {
-            dag.commit(&certificate, self.storage().max_gc_rounds());
-        }
-    }
-
-    /// Spawns a task with the given future; it should only be used for long-running tasks.
-    fn spawn<T: Future<Output = ()> + Send + 'static>(&self, future: T) {
-        self.handles.lock().push(tokio::spawn(future));
-    }
-
     /// Shuts down the BFT.
     pub async fn shut_down(&self) {
         info!("Shutting down the BFT...");
         // Acquire the lock.
         let _lock = self.lock.lock().await;
         // Shut down the primary.
         self.primary.shut_down().await;
-        // Abort the tasks.
-        self.handles.lock().iter().for_each(|handle| handle.abort());
     }
 }
 
@@ -990,6 +934,7 @@ mod tests {
         BFT,
         MAX_LEADER_CERTIFICATE_DELAY_IN_SECS,
         helpers::{Storage, dag::test_helpers::mock_dag_with_modified_last_committed_round},
+        sync::SyncCallback,
     };
 
     use snarkos_account::Account;
@@ -1587,7 +1532,7 @@ mod tests {
         let bootup_bft = initialize_bft(account.clone(), storage_2, ledger)?;
 
         // Sync the BFT DAG at bootup.
-        bootup_bft.sync_bft_dag_at_bootup(certificates.clone()).await;
+        bootup_bft.sync_dag_at_bootup(certificates.clone()).await;
 
         // Check that the BFT starts from the same last committed round.
         assert_eq!(bft.dag.read().last_committed_round(), bootup_bft.dag.read().last_committed_round());
@@ -1766,7 +1711,7 @@ mod tests {
         let bootup_bft = initialize_bft(account.clone(), bootup_storage.clone(), ledger.clone())?;
 
         // Sync the BFT DAG at bootup.
-        bootup_bft.sync_bft_dag_at_bootup(pre_shutdown_certificates.clone()).await;
+        bootup_bft.sync_dag_at_bootup(pre_shutdown_certificates.clone()).await;
 
         // Insert the post shutdown certificates to the storage and BFT with bootup.
         for certificate in post_shutdown_certificates.iter() {
@@ -1946,7 +1891,7 @@ mod tests {
         // Insert a mock DAG in the BFT without bootup.
         *bootup_bft.dag.write() = crate::helpers::dag::test_helpers::mock_dag_with_modified_last_committed_round(0);
         // Sync the BFT DAG at bootup.
-        bootup_bft.sync_bft_dag_at_bootup(pre_shutdown_certificates.clone()).await;
+        bootup_bft.sync_dag_at_bootup(pre_shutdown_certificates.clone()).await;
 
         // Insert the post shutdown certificates into the storage.
         let mut post_shutdown_certificates: Vec<snarkvm::ledger::narwhal::BatchCertificate<CurrentNetwork>> =

node/bft/src/bft.rs

@@ -1746,7 +1746,6 @@ mod prop_tests {
         MAX_WORKERS,
         MEMORY_POOL_PORT,
         Worker,
-        gateway::prop_tests::GatewayAddress::{Dev, Prod},
         helpers::{Storage, init_primary_channels, init_worker_channels},
     };
 
@@ -1848,7 +1847,7 @@ mod prop_tests {
                     Just(account_selector.select(validators)),
                     0u8..,
                 )
-                    .prop_map(|(a, b, c, d)| (a, b, c.private_key, Dev(d)))
+                    .prop_map(|(a, b, c, d)| (a, b, c.private_key, GatewayAddress::Dev(d)))
             })
             .boxed()
     }
@@ -1863,7 +1862,7 @@ mod prop_tests {
                     Just(account_selector.select(validators)),
                     any::<Option<SocketAddr>>(),
                 )
-                    .prop_map(|(a, b, c, d)| (a, b, c.private_key, Prod(d)))
+                    .prop_map(|(a, b, c, d)| (a, b, c.private_key, GatewayAddress::Prod(d)))
             })
             .boxed()
     }