Refactor sync out of block_store.rs

.github/workflows/base_benchmarks.yaml

@@ -33,7 +33,6 @@ jobs:
           --threshold-max-sample-size 64 \
           --threshold-upper-boundary 0.99 \
           --thresholds-reset \
-          --err \
           --adapter rust_criterion \
           --github-actions '${{ secrets.GITHUB_TOKEN }}' \
           cargo bench

.github/workflows/pr_benchmarks.yaml

@@ -34,7 +34,6 @@ jobs:
           --start-point-clone-thresholds \
           --start-point-reset \
           --testbed self-hosted \
-          --err \
           --adapter rust_criterion \
           --github-actions '${{ secrets.GITHUB_TOKEN }}' \
           cargo bench

z2/src/converter.rs

@@ -14,18 +14,14 @@ use bitvec::{bitarr, order::Msb0};
 use eth_trie::{EthTrie, MemoryDB, Trie};
 use indicatif::{ProgressBar, ProgressFinish, ProgressIterator, ProgressStyle};
 use itertools::Itertools;
-use libp2p::PeerId;
 use sha2::{Digest, Sha256};
-use tokio::sync::mpsc;
 use tracing::{debug, trace, warn};
 use zilliqa::{
-    block_store::BlockStore,
     cfg::{scilla_ext_libs_path_default, Amount, Config, NodeConfig},
     crypto::{Hash, SecretKey},
     db::Db,
     exec::store_external_libraries,
     message::{Block, QuorumCertificate, Vote, MAX_COMMITTEE_SIZE},
-    node::{MessageSender, RequestId},
     schnorr,
     scilla::{storage_key, CheckOutput, ParamValue, Transition},
     state::{Account, Code, ContractInit, State},
@@ -346,27 +342,15 @@ pub async fn convert_persistence(
         "{msg} {wide_bar} [{per_sec}] {human_pos}/~{human_len} ({elapsed}/~{duration})",
     )?;
 
-    let (outbound_message_sender, _a) = mpsc::unbounded_channel();
-    let (local_message_sender, _b) = mpsc::unbounded_channel();
-    let message_sender = MessageSender {
-        our_shard: 0,
-        our_peer_id: PeerId::random(),
-        outbound_channel: outbound_message_sender,
-        local_channel: local_message_sender,
-        request_id: RequestId::default(),
-    };
+    // let (outbound_message_sender, _a) = mpsc::unbounded_channel();
+    // let (local_message_sender, _b) = mpsc::unbounded_channel();
 
     let zq2_db = Arc::new(zq2_db);
     let node_config = &zq2_config.nodes[0];
-    let block_store = Arc::new(BlockStore::new(
-        node_config,
-        zq2_db.clone(),
-        message_sender.clone(),
-    )?);
     let mut state = State::new_with_genesis(
         zq2_db.clone().state_trie()?,
         node_config.clone(),
-        block_store,
+        zq2_db.clone(),
     )?;
 
     let mut scilla_docker = run_scilla_docker()?;

z2/src/docgen.rs

@@ -14,7 +14,7 @@ use regex::Regex;
 use serde::{Deserialize, Serialize};
 use tera::Tera;
 use tokio::fs;
-use zilliqa::{cfg::NodeConfig, crypto::SecretKey};
+use zilliqa::{cfg::NodeConfig, crypto::SecretKey, sync::SyncPeers};
 
 const SUPPORTED_APIS_PATH_NAME: &str = "index";
 
@@ -352,10 +352,20 @@ pub fn get_implemented_jsonrpc_methods() -> Result<HashMap<ApiMethod, PageStatus
     let (s2, _) = tokio::sync::mpsc::unbounded_channel();
     let (s3, _) = tokio::sync::mpsc::unbounded_channel();
     let (s4, _) = tokio::sync::mpsc::unbounded_channel();
-    let peers = Arc::new(AtomicUsize::new(0));
+    let peers_count = Arc::new(AtomicUsize::new(0));
+
+    let peer_id = secret_key.to_libp2p_keypair().public().to_peer_id();
+    let peers = Arc::new(SyncPeers::new(peer_id));
 
     let my_node = Arc::new(Mutex::new(zilliqa::node::Node::new(
-        config, secret_key, s1, s2, s3, s4, peers,
+        config,
+        secret_key,
+        s1,
+        s2,
+        s3,
+        s4,
+        peers_count,
+        peers,
     )?));
     let module = zilliqa::api::rpc_module(my_node.clone(), &[]);
     for m in module.method_names() {

zilliqa/benches/it.rs

@@ -28,6 +28,7 @@ use zilliqa::{
     message::{Block, ExternalMessage, Proposal, QuorumCertificate, Vote, MAX_COMMITTEE_SIZE},
     node::{MessageSender, RequestId},
     schnorr,
+    sync::SyncPeers,
     test_util::compile_contract,
     time::{self, SystemTime},
     transaction::{
@@ -46,12 +47,13 @@ fn process_empty(c: &mut Criterion) {
         .measurement_time(Duration::from_secs(10));
 
     let secret_key = SecretKey::new().unwrap();
+    let peer_id = secret_key.to_libp2p_keypair().public().to_peer_id();
     let (outbound_message_sender, _a) = mpsc::unbounded_channel();
     let (local_message_sender, _b) = mpsc::unbounded_channel();
     let (reset_timeout_sender, _c) = mpsc::unbounded_channel();
     let message_sender = MessageSender {
         our_shard: 0,
-        our_peer_id: PeerId::random(),
+        our_peer_id: peer_id,
         outbound_channel: outbound_message_sender,
         local_channel: local_message_sender,
         request_id: RequestId::default(),
@@ -88,6 +90,7 @@ fn process_empty(c: &mut Criterion) {
         message_sender,
         reset_timeout_sender,
         Arc::new(db),
+        Arc::new(SyncPeers::new(peer_id)),
     )
     .unwrap();
 
@@ -163,6 +166,7 @@ fn consensus(
     index: usize,
 ) -> Consensus {
     let secret_key = genesis_deposits[index].0;
+    let peer_id = secret_key.to_libp2p_keypair().public().to_peer_id();
     let (outbound_message_sender, a) = mpsc::unbounded_channel();
     let (local_message_sender, b) = mpsc::unbounded_channel();
     let (reset_timeout_sender, c) = mpsc::unbounded_channel();
@@ -208,6 +212,7 @@ fn consensus(
         message_sender,
         reset_timeout_sender,
         Arc::new(db),
+        Arc::new(SyncPeers::new(peer_id)),
     )
     .unwrap()
 }

zilliqa/src/api/zilliqa.rs

@@ -508,7 +508,7 @@ fn get_blockchain_info(_: Params, node: &Arc<Mutex<Node>>) -> Result<BlockchainI
     let num_peers = node.get_peer_num();
     let num_tx_blocks = node.get_chain_tip();
     let num_ds_blocks = (num_tx_blocks / TX_BLOCKS_PER_DS_BLOCK) + 1;
-    let num_transactions = node.consensus.block_store.get_num_transactions()?;
+    let num_transactions = node.consensus.get_num_transactions()?;
     let ds_block_rate = tx_block_rate / TX_BLOCKS_PER_DS_BLOCK as f64;
 
     // num_txns_ds_epoch
@@ -518,7 +518,6 @@ fn get_blockchain_info(_: Params, node: &Arc<Mutex<Node>>) -> Result<BlockchainI
     for i in current_epoch_first..node.get_chain_tip() {
         let block = node
             .consensus
-            .block_store
             .get_canonical_block_by_number(i)?
             .ok_or_else(|| anyhow!("Block not found"))?;
         num_txns_ds_epoch += block.transactions.len();
@@ -527,7 +526,6 @@ fn get_blockchain_info(_: Params, node: &Arc<Mutex<Node>>) -> Result<BlockchainI
     // num_txns_tx_epoch
     let latest_block = node
         .consensus
-        .block_store
         .get_canonical_block_by_number(node.get_chain_tip())?;
     let num_txns_tx_epoch = match latest_block {
         Some(block) => block.transactions.len(),
@@ -1247,11 +1245,7 @@ fn get_recent_transactions(
     let mut txns = Vec::new();
     let mut blocks_searched = 0;
     while block_number > 0 && txns.len() < 100 && blocks_searched < 100 {
-        let block = match node
-            .consensus
-            .block_store
-            .get_canonical_block_by_number(block_number)?
-        {
+        let block = match node.consensus.get_canonical_block_by_number(block_number)? {
             Some(block) => block,
             None => continue,
         };
@@ -1274,7 +1268,7 @@ fn get_recent_transactions(
 // GetNumTransactions
 fn get_num_transactions(_params: Params, node: &Arc<Mutex<Node>>) -> Result<String> {
     let node = node.lock().unwrap();
-    let num_transactions = node.consensus.block_store.get_num_transactions()?;
+    let num_transactions = node.consensus.get_num_transactions()?;
     Ok(num_transactions.to_string())
 }
 
@@ -1283,7 +1277,6 @@ fn get_num_txns_tx_epoch(_params: Params, node: &Arc<Mutex<Node>>) -> Result<Str
     let node = node.lock().unwrap();
     let latest_block = node
         .consensus
-        .block_store
         .get_canonical_block_by_number(node.get_chain_tip())?;
     let num_transactions = match latest_block {
         Some(block) => block.transactions.len(),
@@ -1302,7 +1295,6 @@ fn get_num_txns_ds_epoch(_params: Params, node: &Arc<Mutex<Node>>) -> Result<Str
     for i in current_epoch_first..node.get_chain_tip() {
         let block = node
             .consensus
-            .block_store
             .get_canonical_block_by_number(i)?
             .ok_or_else(|| anyhow!("Block not found"))?;
         num_txns_epoch += block.transactions.len();