add time to log

add log

add read_all_with_callback

split state dump into multiple task

add more log to indicate progress

Author: Pana

Cargo.lock

@@ -80,6 +80,7 @@ cfx-tasks = { workspace = true }
 cfx-config = { workspace = true }
 cfxcore-types = { workspace = true }
 fallible-iterator = { workspace = true }
+chrono = { workspace = true }
 
 [dev-dependencies]
 criterion = { workspace = true }

crates/client/Cargo.toml

@@ -3,21 +3,18 @@ use cfx_config::Configuration;
 use cfx_rpc_eth_types::{AccountState, StateDump, EOA_STORAGE_ROOT_H256};
 use cfx_rpc_primitives::Bytes;
 use cfx_statedb::{StateDbExt, StateDbGeneric};
-use cfx_storage::{
-    state_manager::StateManagerTrait, utils::to_key_prefix_iter_upper_bound,
-    KeyValueDbIterableTrait,
-};
+use cfx_storage::state_manager::StateManagerTrait;
 use cfx_types::{Address, Space, H256};
 use cfxcore::NodeType;
-use fallible_iterator::FallibleIterator;
+use chrono::Utc;
 use keccak_hash::{keccak, KECCAK_EMPTY};
 use parking_lot::{Condvar, Mutex};
 use primitives::{
     Account, SkipInputCheck, StorageKey, StorageKeyWithSpace, StorageValue,
 };
 use rlp::Rlp;
 use std::{
-    collections::{BTreeMap, HashMap, HashSet},
+    collections::{BTreeMap, HashMap},
     ops::Deref,
     sync::Arc,
     thread,
@@ -32,6 +29,10 @@ pub struct StateDumpConfig {
     pub no_storage: bool,
 }
 
+// This method will read all data (k, v) from the Conflux state tree (including
+// core space and espace accounts, code, storage, deposit, vote_list) into
+// memory at once, then parse and assemble them and assemble all account states
+// into a StateDump struct and return it
 pub fn dump_whole_state(
     conf: &mut Configuration, exit_cond_var: Arc<(Mutex<bool>, Condvar)>,
     config: &StateDumpConfig,
@@ -52,14 +53,19 @@ pub fn dump_whole_state(
     Ok(state_dump)
 }
 
+// This method will iterate through the entire state tree, storing each found
+// account in a temporary map After iterating through all accounts, it will
+// retrieve the code and storage for each account, then call the callback method
+// Pass the AccountState as a parameter to the callback method, which will
+// handle the AccountState
 pub fn iterate_dump_whole_state<F: Fn(AccountState)>(
     conf: &mut Configuration, exit_cond_var: Arc<(Mutex<bool>, Condvar)>,
     config: &StateDumpConfig, callback: F,
 ) -> Result<H256, String> {
     let (mut state_db, state_root) =
         prepare_state_db(conf, exit_cond_var, config)?;
 
-    export_space_accounts_with_iterator(
+    export_space_accounts_with_callback(
         &mut state_db,
         Space::Ethereum,
         config,
@@ -74,7 +80,7 @@ fn prepare_state_db(
     conf: &mut Configuration, exit_cond_var: Arc<(Mutex<bool>, Condvar)>,
     config: &StateDumpConfig,
 ) -> Result<(StateDbGeneric, H256), String> {
-    println!("Preparing state...");
+    println("Preparing state...");
     let (
         data_man,
         _,
@@ -140,6 +146,7 @@ fn prepare_state_db(
 fn export_space_accounts(
     state: &mut StateDbGeneric, space: Space, config: &StateDumpConfig,
 ) -> Result<BTreeMap<Address, AccountState>, Box<dyn std::error::Error>> {
+    println("Start to iterate state...");
     let empty_key = StorageKey::EmptyKey.with_space(space);
     let kv_pairs = state.read_all(empty_key, None)?;
 
@@ -156,7 +163,7 @@ fn export_space_accounts(
         match storage_key_with_space.key {
             StorageKey::AccountKey(address_bytes) => {
                 let address = Address::from_slice(address_bytes);
-                println!("Find account: {:?}", address);
+                println(&format!("Find account: {:?}", address));
                 let account =
                     Account::new_from_rlp(address, &Rlp::new(&value))?;
                 accounts_map.insert(address, account);
@@ -208,16 +215,16 @@ fn export_space_accounts(
             codes_map.get(&address).cloned()
         } else {
             if let Some(code) = codes_map.get(&address) {
-                println!("no-contract account have code: {:?}", code);
+                println(&format!("no-contract account have code: {:?}", code));
             }
             None
         };
 
         let storage = if is_contract {
             storage_map.get(&address).cloned()
         } else {
-            if let Some(storage) = storage_map.get(&address) {
-                println!("no-contract account have storage: {:?}", storage);
+            if let Some(_storage) = storage_map.get(&address) {
+                println(&format!("no-contract account have storage"));
             }
             None
         };
@@ -243,86 +250,61 @@ fn export_space_accounts(
     Ok(accounts)
 }
 
-fn export_space_accounts_with_iterator<F: Fn(AccountState)>(
+pub fn export_space_accounts_with_callback<F: Fn(AccountState)>(
     state: &mut StateDbGeneric, space: Space, config: &StateDumpConfig,
     callback: F,
 ) -> Result<(), Box<dyn std::error::Error>> {
-    let empty_key = StorageKey::EmptyKey.with_space(space);
-    let (kvs, maybe_kv_iterator) = state.read_all_iterator(empty_key)?;
-
-    let mut deleted_keys = HashSet::new();
+    println("Start to iterate state...");
     let mut found_accounts = 0;
+    let mut core_space_key_count: u64 = 0;
+    let mut total_key_count: u64 = 0;
 
-    // Iterate key value pairs from delta trie and intermediate trie
-    for (k, v) in kvs {
-        let storage_key = StorageKeyWithSpace::from_delta_mpt_key(&k);
-        let key = storage_key.to_key_bytes();
-        deleted_keys.insert(key.clone());
-
-        let storage_key_with_space =
-            StorageKeyWithSpace::from_key_bytes::<SkipInputCheck>(&key);
-        if storage_key_with_space.space != space {
-            continue;
-        }
-
-        if let StorageKey::AccountKey(address_bytes) =
-            storage_key_with_space.key
-        {
-            let address = Address::from_slice(address_bytes);
-            println!("Find account: {:?}", address);
-            let account = Account::new_from_rlp(address, &Rlp::new(&v))?;
-
-            let account_state = get_account_state(state, &account, config)?;
-            callback(account_state);
-            found_accounts += 1;
-
-            if config.limit > 0 && found_accounts >= config.limit as usize {
-                break;
-            }
-        } else {
-            continue;
-        }
-    }
-
-    let lower_bound_incl = empty_key.to_key_bytes();
-    let upper_bound_excl = to_key_prefix_iter_upper_bound(&lower_bound_incl);
+    for i in 0..=255 {
+        let prefix = [i];
+        let start_key = StorageKey::AddressPrefixKey(&prefix).with_space(space);
 
-    if let Some(mut kv_iterator) = maybe_kv_iterator {
-        let mut kvs = kv_iterator
-            .iter_range(
-                lower_bound_incl.as_slice(),
-                upper_bound_excl.as_ref().map(|v| &**v),
-            )?
-            .take();
-
-        while let Some((key, value)) = kvs.next()? {
-            if deleted_keys.contains(&key) {
-                continue;
-            }
+        let mut account_states = BTreeMap::new();
 
+        let mut inner_callback = |(key, value): (Vec<u8>, Box<[u8]>)| {
+            total_key_count += 1;
             let storage_key_with_space =
                 StorageKeyWithSpace::from_key_bytes::<SkipInputCheck>(&key);
             if storage_key_with_space.space != space {
-                continue;
+                core_space_key_count += 1;
+                return;
+            }
+
+            if total_key_count % 10000 == 0 {
+                println(&format!(
+                    "total_key_count: {}, core_space_key_count: {}",
+                    total_key_count, core_space_key_count
+                ));
             }
 
             if let StorageKey::AccountKey(address_bytes) =
                 storage_key_with_space.key
             {
                 let address = Address::from_slice(address_bytes);
-                println!("Find account: {:?}", address);
-                let account =
-                    Account::new_from_rlp(address, &Rlp::new(&value))?;
+                println(&format!("Find account: {:?}", address));
+                let account = Account::new_from_rlp(address, &Rlp::new(&value))
+                    .expect("Failed to decode account");
 
-                let account_state = get_account_state(state, &account, config)?;
-                callback(account_state);
-                found_accounts += 1;
+                account_states.insert(address, account);
+            }
+        };
 
-                if config.limit > 0 && found_accounts >= config.limit as usize {
-                    break;
-                }
-            } else {
-                continue;
+        state.read_all_with_callback(start_key, &mut inner_callback)?;
+
+        if account_states.len() > 0 {
+            println("Start to read account code and storage data...");
+        }
+
+        for (_address, account) in account_states {
+            let account_state = get_account_state(state, &account, config)?;
+            callback(account_state);
+            found_accounts += 1;
+            if config.limit > 0 && found_accounts >= config.limit as usize {
+                break;
             }
         }
     }
@@ -369,3 +351,7 @@ fn get_account_state(
         address_hash: Some(address_hash),
     })
 }
+
+fn println(message: &str) {
+    println!("[{}] {}", Utc::now().format("%Y-%m-%d %H:%M:%S"), message);
+}

crates/client/src/state_dump.rs

@@ -188,12 +188,12 @@ mod impls {
             self.delete_all::<access_mode::Read>(key_prefix, debug_record)
         }
 
-        pub fn read_all_iterator(
+        pub fn read_all_with_callback(
             &mut self, access_key_prefix: StorageKeyWithSpace,
-        ) -> Result<(Vec<MptKeyValue>, Option<KvdbSqliteSharded<Box<[u8]>>>)>
-        {
+            callback: &mut dyn FnMut(MptKeyValue),
+        ) -> Result<()> {
             self.storage
-                .read_all_iterator(access_key_prefix)
+                .read_all_with_callback(access_key_prefix, callback)
                 .map_err(|err| err.into())
         }
 
@@ -544,7 +544,7 @@ mod impls {
     };
     use cfx_storage::{
         utils::{access_mode, to_key_prefix_iter_upper_bound},
-        KvdbSqliteSharded, MptKeyValue, StorageStateTrait,
+        MptKeyValue, StorageStateTrait,
     };
     use cfx_types::{
         address_util::AddressUtil, Address, AddressWithSpace, Space,

crates/dbs/statedb/src/lib.rs

@@ -622,6 +622,57 @@ impl CowNodeRef {
         Ok(())
     }
 
+    pub fn iterate_internal_with_callback(
+        &self, owned_node_set: &OwnedNodeSet, trie: &DeltaMpt,
+        guarded_trie_node: GuardedMaybeOwnedTrieNodeAsCowCallParam,
+        key_prefix: CompressedPathRaw, db: &mut DeltaDbOwnedReadTraitObj,
+        callback: &mut dyn FnMut(MptKeyValue),
+    ) -> Result<()> {
+        if guarded_trie_node.as_ref().as_ref().has_value() {
+            assert!(CompressedPathRaw::has_second_nibble(
+                key_prefix.path_mask()
+            ));
+            callback((
+                key_prefix.path_slice().to_vec(),
+                guarded_trie_node.as_ref().as_ref().value_clone().unwrap(),
+            ));
+        }
+
+        let children_table =
+            guarded_trie_node.as_ref().as_ref().children_table.clone();
+        // Free the lock for trie_node.
+        // FIXME: try to share the lock.
+        drop(guarded_trie_node);
+
+        let node_memory_manager = trie.get_node_memory_manager();
+        let allocator = node_memory_manager.get_allocator();
+        for (i, node_ref) in children_table.iter() {
+            let mut cow_child_node =
+                Self::new((*node_ref).into(), owned_node_set, self.mpt_id);
+            let child_node = cow_child_node.get_trie_node(
+                node_memory_manager,
+                &allocator,
+                db,
+            )?;
+            let key_prefix = CompressedPathRaw::join_connected_paths(
+                &key_prefix,
+                i,
+                &child_node.compressed_path_ref(),
+            );
+            let child_node = GuardedValue::take(child_node);
+            cow_child_node.iterate_internal_with_callback(
+                owned_node_set,
+                trie,
+                child_node,
+                key_prefix,
+                db,
+                callback,
+            )?;
+        }
+
+        Ok(())
+    }
+
     /// Recursively commit dirty nodes.
     pub fn commit_dirty_recursively<
         Transaction: BorrowMut<DeltaDbTransactionTraitObj>,