Synchronize optimization into parallel processing, and replace the c…

…alculation of ghost data with validation methods during synchronization to improve efficiency (#4192)

* add parallel code

* use try recv to avoid blocking

* parallel execution

* create chain when the parents are ready in parallel execution

* set executing state in sender

* check the parents when dispatching

* add 10000 buffer for parallel execution

* save the block in local store to buffer the block

* add log for saving time

* check local store for the absent block

* add test case

* add some fix

* add some test

* add false testing case

* add more testing code

* add check data

* add verify blue block in verifier

* add some code

* add verification

* fix some bugs

* add yeilding after execution for processing the main chain in other service

* finish debug, it works

* fmt and clippy

* rebase master
set the number 4500000 for vega updating

* remove some commented code

* fix: return continue if the dag block is ready

* add loop if the connection of the workers are not closed

* 3300000 will be version 1 in vega

* use selected parent to get he parent

* remove comments

* remove bracket

* fix fmt

chain/api/src/chain.rs

@@ -3,6 +3,7 @@
 
 use anyhow::Result;
 use starcoin_crypto::HashValue;
+use starcoin_dag::types::ghostdata::GhostdagData;
 use starcoin_state_api::ChainStateReader;
 use starcoin_statedb::ChainStateDB;
 use starcoin_time_service::TimeService;
@@ -22,7 +23,10 @@ pub use starcoin_types::block::ExecutedBlock;
 use starcoin_vm_types::access_path::AccessPath;
 use starcoin_vm_types::contract_event::ContractEvent;
 
-pub struct VerifiedBlock(pub Block);
+pub struct VerifiedBlock {
+    pub block: Block,
+    pub ghostdata: Option<GhostdagData>,
+}
 pub type MintedUncleNumber = u64;
 
 pub trait ChainReader {
@@ -105,6 +109,11 @@ pub trait ChainReader {
     fn current_tips_hash(&self) -> Result<Vec<HashValue>>;
     fn has_dag_block(&self, header_id: HashValue) -> Result<bool>;
     fn check_chain_type(&self) -> Result<ChainType>;
+    fn verify_and_ghostdata(
+        &self,
+        uncles: &[BlockHeader],
+        header: &BlockHeader,
+    ) -> Result<GhostdagData>;
 }
 
 pub trait ChainWriter {
@@ -115,6 +124,9 @@ pub trait ChainWriter {
     /// Verify, Execute and Connect block to current chain.
     fn apply(&mut self, block: Block) -> Result<ExecutedBlock>;
 
+    /// Verify, Execute and Connect block to current chain.
+    fn apply_for_sync(&mut self, block: Block) -> Result<ExecutedBlock>;
+
     fn chain_state(&mut self) -> &ChainStateDB;
 }
 

chain/src/chain.rs

@@ -1,7 +1,7 @@
 // Copyright (c) The Starcoin Core Contributors
 // SPDX-License-Identifier: Apache-2.0
 
-use crate::verifier::{BlockVerifier, DagVerifier, FullVerifier};
+use crate::verifier::{BlockVerifier, DagVerifier, DagVerifierWithGhostData, FullVerifier};
 use anyhow::{bail, ensure, format_err, Ok, Result};
 use sp_utils::stop_watch::{watch, CHAIN_WATCH_NAME};
 use starcoin_accumulator::inmemory::InMemoryAccumulator;
@@ -458,8 +458,8 @@ impl BlockChain {
     }
 
     fn execute_dag_block(&mut self, verified_block: VerifiedBlock) -> Result<ExecutedBlock> {
-        info!("execute dag block:{:?}", verified_block.0);
-        let block = verified_block.0;
+        info!("execute dag block:{:?}", verified_block.block);
+        let block = verified_block.block;
         let selected_parent = block.parent_hash();
         let block_info_past = self
             .storage
@@ -645,9 +645,16 @@ impl BlockChain {
             .storage
             .get_block_header_by_hash(self.genesis_hash)?
             .ok_or_else(|| format_err!("failed to get genesis because it is none"))?;
-        let result = self
-            .dag
-            .commit(header.to_owned(), genesis_header.parent_hash());
+        let result = match verified_block.ghostdata {
+            Some(trusted_ghostdata) => self.dag.commit_trusted_block(
+                header.to_owned(),
+                genesis_header.parent_hash(),
+                Arc::new(trusted_ghostdata),
+            ),
+            None => self
+                .dag
+                .commit(header.to_owned(), genesis_header.parent_hash()),
+        };
         match result {
             anyhow::Result::Ok(_) => info!("finish to commit dag block: {:?}", block_id),
             Err(e) => {
@@ -1210,8 +1217,7 @@ impl ChainReader for BlockChain {
     }
 
     fn verify(&self, block: Block) -> Result<VerifiedBlock> {
-        DagVerifier::verify_header(self, block.header())?;
-        Ok(VerifiedBlock(block))
+        DagVerifier::verify_block(self, block)
     }
 
     fn execute(&mut self, verified_block: VerifiedBlock) -> Result<ExecutedBlock> {
@@ -1225,7 +1231,7 @@ impl ChainReader for BlockChain {
             self.block_accumulator.fork(None),
             &self.epoch,
             Some(self.status.status.clone()),
-            verified_block.0,
+            verified_block.block,
             self.vm_metrics.clone(),
         )
     }
@@ -1348,6 +1354,14 @@ impl ChainReader for BlockChain {
     fn check_chain_type(&self) -> Result<ChainType> {
         Ok(ChainType::Dag)
     }
+
+    fn verify_and_ghostdata(
+        &self,
+        uncles: &[BlockHeader],
+        header: &BlockHeader,
+    ) -> Result<starcoin_dag::types::ghostdata::GhostdagData> {
+        self.dag().verify_and_ghostdata(uncles, header)
+    }
 }
 
 impl BlockChain {
@@ -1467,10 +1481,9 @@ impl BlockChain {
         }
         // Caculate the ghostdata of the virutal node created by all tips.
         // And the ghostdata.selected of the tips will be the latest head.
-        let block_hash = {
-            let ghost_of_tips = dag.ghostdata(tips.as_slice())?;
-            ghost_of_tips.selected_parent
-        };
+        let block_hash = dag
+            .ghost_dag_manager()
+            .find_selected_parent(tips.iter().copied())?;
         let (block, block_info) = {
             let block = self
                 .storage
@@ -1532,6 +1545,10 @@ impl ChainWriter for BlockChain {
     fn chain_state(&mut self) -> &ChainStateDB {
         &self.statedb
     }
+
+    fn apply_for_sync(&mut self, block: Block) -> Result<ExecutedBlock> {
+        self.apply_with_verifier::<DagVerifierWithGhostData>(block)
+    }
 }
 
 pub(crate) fn info_2_accumulator(

chain/src/verifier/mod.rs

@@ -7,6 +7,7 @@ use starcoin_chain_api::{
     verify_block, ChainReader, ConnectBlockError, VerifiedBlock, VerifyBlockField,
 };
 use starcoin_consensus::{Consensus, ConsensusVerifyError};
+use starcoin_dag::types::ghostdata::GhostdagData;
 use starcoin_logger::prelude::debug;
 use starcoin_open_block::AddressFilter;
 use starcoin_types::block::{Block, BlockHeader, ALLOWED_FUTURE_BLOCKTIME};
@@ -76,13 +77,16 @@ pub trait BlockVerifier {
         StaticVerifier::verify_body_hash(&new_block)?;
         watch(CHAIN_WATCH_NAME, "n13");
         //verify uncles
-        Self::verify_uncles(
+        let ghostdata = Self::verify_uncles(
             current_chain,
             new_block.uncles().unwrap_or_default(),
             new_block_header,
         )?;
         watch(CHAIN_WATCH_NAME, "n14");
-        Ok(VerifiedBlock(new_block))
+        Ok(VerifiedBlock {
+            block: new_block,
+            ghostdata,
+        })
     }
 
     fn verify_blacklisted_txns(new_block: &Block) -> Result<()> {
@@ -101,7 +105,7 @@ pub trait BlockVerifier {
         current_chain: &R,
         uncles: &[BlockHeader],
         header: &BlockHeader,
-    ) -> Result<()>
+    ) -> Result<Option<GhostdagData>>
     where
         R: ChainReader,
     {
@@ -118,7 +122,7 @@ pub trait BlockVerifier {
         }
 
         if uncles.is_empty() {
-            return Ok(());
+            return Ok(None);
         }
         verify_block!(
             VerifyBlockField::Uncle,
@@ -163,7 +167,7 @@ pub trait BlockVerifier {
             Self::verify_header(&uncle_branch, uncle)?;
             uncle_ids.insert(uncle_id);
         }
-        Ok(())
+        Ok(None)
     }
 
     fn can_be_uncle<R>(current_chain: &R, block_header: &BlockHeader) -> Result<bool>
@@ -318,25 +322,27 @@ impl BlockVerifier for NoneVerifier {
     where
         R: ChainReader,
     {
-        Ok(VerifiedBlock(new_block))
+        Ok(VerifiedBlock {
+            block: new_block,
+            ghostdata: None,
+        })
     }
 
     fn verify_uncles<R>(
         _current_chain: &R,
         _uncles: &[BlockHeader],
         _header: &BlockHeader,
-    ) -> Result<()>
+    ) -> Result<Option<GhostdagData>>
     where
         R: ChainReader,
     {
-        Ok(())
+        Ok(None)
     }
 }
 
-//TODO: Implement it.
-pub struct DagVerifier;
-impl BlockVerifier for DagVerifier {
-    fn verify_header<R>(current_chain: &R, new_block_header: &BlockHeader) -> Result<()>
+struct BasicDagVerifier;
+impl BasicDagVerifier {
+    pub fn verify_header<R>(current_chain: &R, new_block_header: &BlockHeader) -> Result<()>
     where
         R: ChainReader,
     {
@@ -380,50 +386,60 @@ impl BlockVerifier for DagVerifier {
         ConsensusVerifier::verify_header(current_chain, new_block_header)
     }
 
+    fn verify_blue_blocks<R>(
+        current_chain: &R,
+        uncles: &[BlockHeader],
+        header: &BlockHeader,
+    ) -> Result<GhostdagData>
+    where
+        R: ChainReader,
+    {
+        current_chain.verify_and_ghostdata(uncles, header)
+    }
+}
+//TODO: Implement it.
+pub struct DagVerifier;
+impl BlockVerifier for DagVerifier {
+    fn verify_header<R>(current_chain: &R, new_block_header: &BlockHeader) -> Result<()>
+    where
+        R: ChainReader,
+    {
+        BasicDagVerifier::verify_header(current_chain, new_block_header)
+    }
+
     fn verify_uncles<R>(
         _current_chain: &R,
         _uncles: &[BlockHeader],
         _header: &BlockHeader,
-    ) -> Result<()>
+    ) -> Result<Option<GhostdagData>>
     where
         R: ChainReader,
     {
-        // let mut uncle_ids = HashSet::new();
-        // for uncle in uncles {
-        //     let uncle_id = uncle.id();
-        //     verify_block!(
-        //         VerifyBlockField::Uncle,
-        //         !uncle_ids.contains(&uncle.id()),
-        //         "repeat uncle {:?} in current block {:?}",
-        //         uncle_id,
-        //         header.id()
-        //     );
-
-        //     if !header.is_dag() {
-        //         verify_block!(
-        //             VerifyBlockField::Uncle,
-        //             uncle.number() < header.number() ,
-        //         "uncle block number bigger than or equal to current block ,uncle block number is {} , current block number is {}", uncle.number(), header.number()
-        //         );
-        //     }
-
-        //     verify_block!(
-        //         VerifyBlockField::Uncle,
-        //         current_chain.get_block_info(Some(uncle_id))?.is_some(),
-        //         "Invalid block: uncle {} does not exist",
-        //         uncle_id
-        //     );
-
-        //     debug!(
-        //         "verify_uncle header number {} hash {:?} uncle number {} hash {:?}",
-        //         header.number(),
-        //         header.id(),
-        //         uncle.number(),
-        //         uncle.id()
-        //     );
-        //     uncle_ids.insert(uncle_id);
-        // }
+        Ok(None)
+    }
+}
 
-        Ok(())
+pub struct DagVerifierWithGhostData;
+impl BlockVerifier for DagVerifierWithGhostData {
+    fn verify_header<R>(current_chain: &R, new_block_header: &BlockHeader) -> Result<()>
+    where
+        R: ChainReader,
+    {
+        BasicDagVerifier::verify_header(current_chain, new_block_header)
+    }
+
+    fn verify_uncles<R>(
+        current_chain: &R,
+        uncles: &[BlockHeader],
+        header: &BlockHeader,
+    ) -> Result<Option<GhostdagData>>
+    where
+        R: ChainReader,
+    {
+        Ok(Some(BasicDagVerifier::verify_blue_blocks(
+            current_chain,
+            uncles,
+            header,
+        )?))
     }
 }