Write the dag relationship in batch

commons/stream-task/src/event.rs

@@ -278,6 +278,7 @@ impl TaskEventHandle for TaskEventCounterHandle {
     }
 
     fn on_error(&self) {
+        println!("jacktest: on_error()");
         if let Some(counter) = self.current_counter.lock().as_ref() {
             counter.error_counter.fetch_add(1, Ordering::Release);
         }

flexidag/src/blockdag.rs

@@ -223,10 +223,12 @@ impl BlockDAG {
         );
         let reachability_store = self.storage.reachability_store.clone();
 
-        let mut merge_set = ghostdata
-            .unordered_mergeset_without_selected_parent()
-            .filter(|hash| self.storage.reachability_store.read().has(*hash).unwrap())
-            .collect::<Vec<_>>()
+        let mut merge_set = self
+            .ghost_dag_manager()
+            .unordered_mergeset_without_selected_parent(
+                ghostdata.selected_parent,
+                &header.parents(),
+            )
             .into_iter();
         let add_block_result = {
             let mut reachability_writer = reachability_store.write();
@@ -482,8 +484,8 @@ impl BlockDAG {
         let dag_state = self.get_dag_state(previous_pruning_point)?;
         let next_ghostdata = self.ghostdata(&dag_state.tips)?;
         info!(
-            "start to calculate the mergeset and tips for tips: {:?}, and last pruning point: {:?} and next ghostdata: {:?}",
-            dag_state.tips, previous_pruning_point, next_ghostdata,
+            "start to calculate the mergeset and tips for tips: {:?}, and last pruning point: {:?} and next ghostdata's selected parents: {:?} and blues set are {:?}",
+            dag_state.tips, previous_pruning_point, next_ghostdata.selected_parent, next_ghostdata.mergeset_blues,
         );
         let next_pruning_point = self.pruning_point_manager().next_pruning_point(
             previous_pruning_point,

flexidag/src/consensusdb/access.rs

@@ -108,6 +108,13 @@ where
         Ok(())
     }
 
+    pub fn flush_cache(&self, data: &[(S::Key, S::Value)]) -> Result<(), StoreError> {
+        for (key, value) in data {
+            self.cache.insert(key.clone(), value.clone());
+        }
+        Ok(())
+    }
+
     /// Write directly from an iterator and do not cache any data. NOTE: this action also clears the cache
     pub fn write_many_without_cache(
         &self,

flexidag/src/consensusdb/consensus_relations.rs

@@ -1,14 +1,15 @@
 use super::schema::{KeyCodec, ValueCodec};
 use super::{
     db::DBStorage,
-    prelude::{BatchDbWriter, CachedDbAccess, DirectDbWriter, StoreError},
+    prelude::{CachedDbAccess, StoreError},
 };
 use crate::define_schema;
-use rocksdb::WriteBatch;
 use starcoin_crypto::HashValue as Hash;
+use starcoin_storage::batch::{WriteBatch, WriteBatchData, WriteBatchWithColumn};
+use starcoin_storage::storage::{InnerStore, WriteOp};
 use starcoin_types::blockhash::{BlockHashes, BlockLevel};
+use std::collections::HashMap;
 use std::sync::Arc;
-
 /// Reader API for `RelationsStore`.
 pub trait RelationsStoreReader {
     fn get_parents(&self, hash: Hash) -> Result<BlockHashes, StoreError>;
@@ -90,41 +91,6 @@ impl DbRelationsStore {
     pub fn clone_with_new_cache(&self, cache_size: usize) -> Self {
         Self::new(Arc::clone(&self.db), self.level, cache_size)
     }
-
-    pub fn insert_batch(
-        &mut self,
-        batch: &mut WriteBatch,
-        hash: Hash,
-        parents: BlockHashes,
-    ) -> Result<(), StoreError> {
-        if self.has(hash)? {
-            return Err(StoreError::KeyAlreadyExists(hash.to_string()));
-        }
-
-        // Insert a new entry for `hash`
-        self.parents_access
-            .write(BatchDbWriter::new(batch, &self.db), hash, parents.clone())?;
-
-        // The new hash has no children yet
-        self.children_access.write(
-            BatchDbWriter::new(batch, &self.db),
-            hash,
-            BlockHashes::new(Vec::new()),
-        )?;
-
-        // Update `children` for each parent
-        for parent in parents.iter().cloned() {
-            let mut children = (*self.get_children(parent)?).clone();
-            children.push(hash);
-            self.children_access.write(
-                BatchDbWriter::new(batch, &self.db),
-                parent,
-                BlockHashes::new(children),
-            )?;
-        }
-
-        Ok(())
-    }
 }
 
 impl RelationsStoreReader for DbRelationsStore {
@@ -147,35 +113,69 @@ impl RelationsStoreReader for DbRelationsStore {
 }
 
 impl RelationsStore for DbRelationsStore {
-    /// See `insert_batch` as well
-    /// TODO: use one function with DbWriter for both this function and insert_batch
     fn insert(&self, hash: Hash, parents: BlockHashes) -> Result<(), StoreError> {
         if self.has(hash)? {
             return Err(StoreError::KeyAlreadyExists(hash.to_string()));
         }
 
-        // Insert a new entry for `hash`
-        self.parents_access
-            .write(DirectDbWriter::new(&self.db), hash, parents.clone())?;
+        let mut parent_to_children = HashMap::new();
+        parent_to_children.insert(hash, vec![]);
 
-        // The new hash has no children yet
-        self.children_access.write(
-            DirectDbWriter::new(&self.db),
-            hash,
-            BlockHashes::new(Vec::new()),
-        )?;
-
-        // Update `children` for each parent
         for parent in parents.iter().cloned() {
-            let mut children = (*self.get_children(parent)?).clone();
+            let mut children = match self.get_children(parent) {
+                Ok(children) => (*children).clone(),
+                Err(e) => match e {
+                    StoreError::KeyNotFound(_) => vec![],
+                    _ => return std::result::Result::Err(e),
+                },
+            };
             children.push(hash);
-            self.children_access.write(
-                DirectDbWriter::new(&self.db),
-                parent,
-                BlockHashes::new(children),
-            )?;
+            parent_to_children.insert(parent, children);
         }
 
+        let batch = WriteBatchWithColumn {
+            data: vec![
+                WriteBatchData {
+                    column: PARENTS_CF.to_string(),
+                    row_data: WriteBatch::new_with_rows(vec![(
+                        hash.to_vec(),
+                        WriteOp::Value(
+                            <Arc<Vec<Hash>> as ValueCodec<RelationParent>>::encode_value(&parents)?,
+                        ),
+                    )]),
+                },
+                WriteBatchData {
+                    column: CHILDREN_CF.to_string(),
+                    row_data: WriteBatch::new_with_rows(
+                        parent_to_children
+                            .iter()
+                            .map(|(key, value)| {
+                                std::result::Result::Ok((
+                                    key.to_vec(),
+                                    WriteOp::Value(<Arc<Vec<Hash>> as ValueCodec<
+                                        RelationChildren,
+                                    >>::encode_value(
+                                        &Arc::new(value.clone())
+                                    )?),
+                                ))
+                            })
+                            .collect::<std::result::Result<Vec<_>, StoreError>>()?,
+                    ),
+                },
+            ],
+        };
+        self.db
+            .write_batch_with_column(batch)
+            .map_err(|e| StoreError::DBIoError(e.to_string()))?;
+
+        self.parents_access.flush_cache(&[(hash, parents)])?;
+        self.children_access.flush_cache(
+            &parent_to_children
+                .into_iter()
+                .map(|(key, value)| (key, BlockHashes::new(value)))
+                .collect::<Vec<_>>(),
+        )?;
+
         Ok(())
     }
 }

flexidag/src/ghostdag/protocol.rs

@@ -210,22 +210,29 @@ impl<
             }
         }
 
+        let remote_blue_set = blue_blocks
+            .iter()
+            .map(|header| header.id())
+            .collect::<HashSet<_>>();
         if new_block_data
             .mergeset_blues
             .iter()
             .skip(1)
             .cloned()
             .collect::<HashSet<_>>()
-            != blue_blocks
+            != remote_blue_set
+        {
+            warn!("The data of blue set is not equal when executing the block: {:?}, for {:?}, checking data: {:?}", header.id(), blue_blocks.iter().map(|header| header.id()).collect::<Vec<_>>(), new_block_data.mergeset_blues);
+            let ghostdata = self.ghostdag(&header.parents_hash())?;
+            if ghostdata
+                .mergeset_blues
                 .iter()
-                .map(|header| header.id())
+                .skip(1)
+                .cloned()
                 .collect::<HashSet<_>>()
-        {
-            if header.number() < 10000000 {
-                // no bail before 10000000
-                warn!("The data of blue set is not equal when executing the block: {:?}, for {:?}, checking data: {:?}", header.id(), blue_blocks.iter().map(|header| header.id()).collect::<Vec<_>>(), new_block_data.mergeset_blues);
-            } else {
-                bail!("The data of blue set is not equal when executing the block: {:?}, for {:?}, checking data: {:?}", header.id(), blue_blocks.iter().map(|header| header.id()).collect::<Vec<_>>(), new_block_data.mergeset_blues);
+                != remote_blue_set
+            {
+                bail!("The ghost data of blue set is not equal when executing the block: {:?}, for {:?}, checking data: {:?}", header.id(), blue_blocks.iter().map(|header| header.id()).collect::<Vec<_>>(), ghostdata.mergeset_blues);
             }
         }
 
@@ -377,12 +384,20 @@ impl<
             *candidate_blue_anticone_size = (*candidate_blue_anticone_size).checked_add(1).unwrap();
             if *candidate_blue_anticone_size > self.k {
                 // k-cluster violation: The candidate's blue anticone exceeded k
+                info!(
+                    "Checking blue candidate: {} failed, blue anticone exceeded k",
+                    blue_candidate
+                );
                 return Ok(ColoringState::Red);
             }
 
             if *candidate_blues_anticone_sizes.get(&block).unwrap() == self.k {
                 // k-cluster violation: A block in candidate's blue anticone already
                 // has k blue blocks in its own anticone
+                info!(
+                    "Checking blue candidate: {} failed, block {} has k blue blocks in its anticone",
+                    blue_candidate, block
+                );
                 return Ok(ColoringState::Red);
             }
 
@@ -431,6 +446,10 @@ impl<
         // The maximum length of new_block_data.mergeset_blues can be K+1 because
         // it contains the selected parent.
         if new_block_data.mergeset_blues.len() as KType == self.k.checked_add(1).unwrap() {
+            info!(
+                "Checking blue candidate: {} failed, mergeset blues size is K+1",
+                blue_candidate
+            );
             return Ok(ColoringOutput::Red);
         }
 

flexidag/src/prune/pruning_point_manager.rs

@@ -75,6 +75,11 @@ impl<T: ReachabilityStoreReader + Clone> PruningPointManagerT<T> {
             min_required_blue_score_for_next_pruning_point
         );
 
+        debug!("previous_pruning_point: {:?}, previous_ghostdata: {:?}, next_ghostdata: {:?}, pruning_depth: {:?}, pruning_finality: {:?}", 
+            previous_pruning_point, previous_ghostdata, next_ghostdata,
+            pruning_depth, pruning_finality,
+        );
+
         let mut latest_pruning_ghost_data = previous_ghostdata.to_compact();
         if min_required_blue_score_for_next_pruning_point + pruning_depth
             <= next_ghostdata.blue_score