Merge pull request AleoNet#2054 from AleoHQ/fix/parallel-checks

Add parallelism to ledger and introduce `timer` for  Ledger and Beacon initialization

node/Cargo.toml

@@ -16,6 +16,14 @@ categories = [ "cryptography", "operating-systems" ]
 license = "GPL-3.0"
 edition = "2021"
 
+[features]
+default = [ ]
+timer = [ "aleo-std/timer", "snarkos-node-ledger/timer" ]
+
+[dependencies.aleo-std]
+version = "0.1.15"
+default-features = false
+
 [dependencies.anyhow]
 version = "1"
 
@@ -68,3 +76,9 @@ features = ["rt", "signal"]
 
 [dependencies.tracing]
 version = "0.1"
+
+[dev-dependencies.tokio-test]
+version = "0.4"
+
+[dev-dependencies.rand_chacha]
+version = "0.3.0"

node/ledger/Cargo.toml

@@ -16,6 +16,15 @@ categories = [ "cryptography", "operating-systems" ]
 license = "GPL-3.0"
 edition = "2021"
 
+[features]
+default = [ "parallel" ]
+parallel = [ "rayon" ]
+timer = [ "aleo-std/timer" ]
+
+[dependencies.aleo-std]
+version = "0.1.15"
+default-features = false
+
 [dependencies.anyhow]
 version = "1"
 
@@ -32,6 +41,10 @@ version = "0.12"
 [dependencies.rand]
 version = "0.8"
 
+[dependencies.rayon]
+version = "1"
+optional = true
+
 [dependencies.snarkos-node-store]
 path = "../store"
 

node/ledger/src/get.rs

@@ -53,6 +53,11 @@ impl<N: Network, C: ConsensusStorage<N>> Ledger<N, C> {
         }
     }
 
+    /// Returns the blocks in the given block range.
+    pub fn get_blocks(&self, heights: Range<u32>) -> Result<Vec<Block<N>>> {
+        cfg_into_iter!(heights).map(|height| self.get_block(height)).collect()
+    }
+
     /// Returns the block for the given block hash.
     pub fn get_block_by_hash(&self, block_hash: &N::BlockHash) -> Result<Block<N>> {
         // Retrieve the block.

node/ledger/src/lib.rs

@@ -43,9 +43,12 @@ use snarkvm::{
     },
 };
 
+use aleo_std::prelude::{finish, lap, timer};
 use anyhow::Result;
+use core::ops::Range;
 use indexmap::IndexMap;
 use parking_lot::RwLock;
+use rand::{prelude::IteratorRandom, rngs::OsRng};
 use std::{borrow::Cow, sync::Arc};
 
 #[cfg(feature = "parallel")]
@@ -80,21 +83,31 @@ pub struct Ledger<N: Network, C: ConsensusStorage<N>> {
 impl<N: Network, C: ConsensusStorage<N>> Ledger<N, C> {
     /// Loads the ledger from storage.
     pub fn load(genesis: Option<Block<N>>, dev: Option<u16>) -> Result<Self> {
+        let timer = timer!("Ledger::load");
+
         // Retrieve the genesis hash.
         let genesis_hash = match genesis {
             Some(ref genesis) => genesis.hash(),
             None => Block::<N>::from_bytes_le(N::genesis_bytes())?.hash(),
         };
+        lap!(timer, "Load genesis hash");
 
         // Initialize the consensus store.
         let store = match ConsensusStore::<N, C>::open(dev) {
             Ok(store) => store,
             _ => bail!("Failed to load ledger (run 'snarkos clean' and try again)"),
         };
+        lap!(timer, "Load consensus store");
+
         // Initialize a new VM.
         let vm = VM::from(store)?;
+        lap!(timer, "Initialize a new VM");
+
         // Initialize the ledger.
         let ledger = Self::from(vm, genesis)?;
+        lap!(timer, "Initialize ledger");
+
+        finish!(timer);
 
         // Ensure the ledger contains the correct genesis block.
         match ledger.contains_block_hash(&genesis_hash)? {
@@ -105,11 +118,14 @@ impl<N: Network, C: ConsensusStorage<N>> Ledger<N, C> {
 
     /// Initializes the ledger from storage, with an optional genesis block.
     pub fn from(vm: VM<N, C>, genesis: Option<Block<N>>) -> Result<Self> {
+        let timer = timer!("Ledger::from");
+
         // Load the genesis block.
         let genesis = match genesis {
             Some(genesis) => genesis,
             None => Block::<N>::from_bytes_le(N::genesis_bytes())?,
         };
+        lap!(timer, "Load genesis block");
 
         // Initialize the ledger.
         let mut ledger = Self {
@@ -136,12 +152,19 @@ impl<N: Network, C: ConsensusStorage<N>> Ledger<N, C> {
         ledger.current_block = Arc::new(RwLock::new(block));
         // Set the current epoch challenge.
         ledger.current_epoch_challenge = Arc::new(RwLock::new(Some(ledger.get_epoch_challenge(latest_height)?)));
-
-        // // Safety check the existence of every block.
-        // cfg_into_iter!((0..=latest_height)).try_for_each(|height| {
-        //     ledger.get_block(height)?;
-        //     Ok::<_, Error>(())
-        // })?;
+        lap!(timer, "Initialize ledger state");
+
+        // Safety check the existence of `NUM_BLOCKS` random blocks.
+        const NUM_BLOCKS: usize = 1000;
+        let block_heights: Vec<u32> = (0..=latest_height)
+            .choose_multiple(&mut OsRng::default(), core::cmp::min(NUM_BLOCKS, latest_height as usize));
+        cfg_into_iter!(block_heights).try_for_each(|height| {
+            ledger.get_block(height)?;
+            Ok::<_, Error>(())
+        })?;
+        lap!(timer, "Check existence of {NUM_BLOCKS} random blocks");
+
+        finish!(timer);
 
         Ok(ledger)
     }

node/src/beacon/mod.rs

@@ -46,6 +46,7 @@ use snarkvm::prelude::{
     Zero,
 };
 
+use aleo_std::prelude::{finish, lap, timer};
 use anyhow::{bail, Result};
 use core::{str::FromStr, time::Duration};
 use parking_lot::RwLock;
@@ -90,29 +91,46 @@ impl<N: Network> Beacon<N> {
         genesis: Option<Block<N>>,
         dev: Option<u16>,
     ) -> Result<Self> {
+        let timer = timer!("Beacon::new");
+
         // Initialize the node account.
         let account = Account::from(private_key)?;
+        lap!(timer, "Initialize the account");
+
         // Initialize the ledger.
         let ledger = Ledger::load(genesis, dev)?;
+        lap!(timer, "Initialize the ledger");
+
         // Initialize the consensus.
         let consensus = Consensus::new(ledger.clone())?;
+        lap!(timer, "Initialize consensus");
+
         // Initialize the node router.
         let (router, router_receiver) = Router::new::<Self>(node_ip, account.address(), trusted_peers).await?;
+        lap!(timer, "Initialize the router");
+
         // Initialize the REST server.
         let rest = match rest_ip {
-            Some(rest_ip) => Some(Arc::new(Rest::start(
-                rest_ip,
-                account.address(),
-                Some(consensus.clone()),
-                ledger.clone(),
-                router.clone(),
-            )?)),
+            Some(rest_ip) => {
+                let server = Arc::new(Rest::start(
+                    rest_ip,
+                    account.address(),
+                    Some(consensus.clone()),
+                    ledger.clone(),
+                    router.clone(),
+                )?);
+                lap!(timer, "Initialize REST server");
+                Some(server)
+            }
             None => None,
         };
+
         // Initialize the block generation time.
         let block_generation_time = Arc::new(AtomicU64::new(2));
         // Retrieve the unspent records.
         let unspent_records = ledger.find_unspent_records(account.view_key())?;
+        lap!(timer, "Retrieve the unspent records");
+
         // Initialize the node.
         let node = Self {
             account,
@@ -126,11 +144,13 @@ impl<N: Network> Beacon<N> {
         };
         // Initialize the router handler.
         router.initialize_handler(node.clone(), router_receiver).await;
-
         // Initialize the block production.
         node.initialize_block_production().await;
         // Initialize the signal handler.
         node.handle_signals();
+        lap!(timer, "Initialize the handlers");
+
+        finish!(timer);
         // Return the node.
         Ok(node)
     }
@@ -440,3 +460,47 @@ impl<N: Network> Beacon<N> {
         Ok(())
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use snarkvm::prelude::{ConsensusMemory, ConsensusStore, Testnet3, VM};
+
+    use rand::SeedableRng;
+    use rand_chacha::ChaChaRng;
+
+    type CurrentNetwork = Testnet3;
+
+    /// Use `RUST_MIN_STACK=67108864 cargo test --release profiler --features timer` to run this test.
+    #[ignore]
+    #[tokio::test]
+    async fn test_profiler() -> Result<()> {
+        // Specify the node attributes.
+        let node = SocketAddr::from_str("0.0.0.0:4133").unwrap();
+        let rest = SocketAddr::from_str("0.0.0.0:3033").unwrap();
+
+        let dev = Some(0);
+
+        // Initialize an (insecure) fixed RNG.
+        let mut rng = ChaChaRng::seed_from_u64(1234567890u64);
+        // Initialize the beacon private key.
+        let beacon_private_key = PrivateKey::<CurrentNetwork>::new(&mut rng)?;
+        // Initialize a new VM.
+        let vm = VM::from(ConsensusStore::<CurrentNetwork, ConsensusMemory<CurrentNetwork>>::open(None)?)?;
+        // Initialize the genesis block.
+        let genesis = Block::genesis(&vm, &beacon_private_key, &mut rng)?;
+
+        println!("Initializing beacon node...");
+
+        let beacon =
+            Beacon::<CurrentNetwork>::new(node, Some(rest), beacon_private_key, &[], Some(genesis), dev).await.unwrap();
+
+        println!(
+            "Loaded beacon node with {} blocks and {} records",
+            beacon.ledger.latest_height(),
+            beacon.unspent_records.read().len()
+        );
+
+        bail!("\n\nRemember to #[ignore] this test!\n\n")
+    }
+}

node/src/lib.rs

@@ -15,6 +15,7 @@
 // along with the snarkOS library. If not, see <https://www.gnu.org/licenses/>.
 
 #![forbid(unsafe_code)]
+#![recursion_limit = "256"]
 
 #[macro_use]
 extern crate async_trait;