cached_pos_ledger_db/
lib.rs

1// Copyright 2020 Conflux Foundation. All rights reserved.
2// Conflux is free software and distributed under GNU General Public License.
3// See http://www.gnu.org/licenses/
4
5use std::sync::Arc;
6
7use anyhow::{format_err, Result};
8
9use diem_crypto::HashValue;
10use diem_infallible::Mutex;
11use diem_logger::prelude::*;
12use diem_types::{
13    account_address::AccountAddress,
14    block_info::PivotBlockDecision,
15    contract_event::ContractEvent,
16    ledger_info::LedgerInfo,
17    term_state::{NodeID, PosState},
18    transaction::Transaction,
19};
20use executor_types::{Error, ExecutedTrees, ProcessedVMOutput};
21pub use speculation_cache::{SpeculationBlock, SpeculationCache};
22use storage_interface::{DbReaderWriter, TreeState};
23
24mod logging;
25mod speculation_cache;
26
27pub struct CachedPosLedgerDB {
28    pub db: DbReaderWriter,
29    pub cache: Mutex<SpeculationCache>,
30}
31
32impl CachedPosLedgerDB {
33    pub fn new(db: DbReaderWriter) -> Self {
34        let startup_info = db
35            .reader
36            .get_startup_info(true)
37            .expect("Shouldn't fail")
38            .expect("DB not bootstrapped.");
39
40        Self {
41            db,
42            cache: Mutex::new(SpeculationCache::new_with_startup_info(
43                startup_info,
44            )),
45        }
46    }
47
48    fn get_executed_trees(
49        &self, block_id: HashValue,
50    ) -> Result<ExecutedTrees, Error> {
51        diem_debug!(
52            "get_executed_trees:{} {}",
53            block_id,
54            self.cache.lock().committed_block_id()
55        );
56        let executed_trees =
57            if block_id == self.cache.lock().committed_block_id() {
58                self.cache.lock().committed_trees().clone()
59            } else {
60                self.get_block(&block_id)?
61                    .lock()
62                    .output()
63                    .executed_trees()
64                    .clone()
65            };
66
67        Ok(executed_trees)
68    }
69
70    pub fn get_pos_state(
71        &self, block_id: &HashValue,
72    ) -> Result<PosState, Error> {
73        if let Ok(executed_tree) = self.get_executed_trees(*block_id) {
74            Ok(executed_tree.pos_state().clone())
75        } else {
76            self.db.reader.get_pos_state(block_id).map_err(|_| {
77                Error::InternalError {
78                    error: "pos state not found".to_string(),
79                }
80            })
81        }
82    }
83
84    pub fn reset_cache(&self) -> Result<(), Error> {
85        let startup_info = self
86            .db
87            .reader
88            .get_startup_info(true)?
89            .ok_or_else(|| format_err!("DB not bootstrapped."))?;
90        *(self.cache.lock()) =
91            SpeculationCache::new_with_startup_info(startup_info);
92        Ok(())
93    }
94
95    pub fn new_on_unbootstrapped_db(
96        db: DbReaderWriter, tree_state: TreeState, initial_seed: Vec<u8>,
97        initial_nodes: Vec<(NodeID, u64)>,
98        initial_committee: Vec<(AccountAddress, u64)>,
99        genesis_pivot_decision: Option<PivotBlockDecision>,
100    ) -> Self {
101        // if initial_nodes.is_empty() {
102        //     let access_paths = ON_CHAIN_CONFIG_REGISTRY
103        //         .iter()
104        //         .map(|config_id| config_id.access_path())
105        //         .collect();
106        //     let configs = db
107        //         .reader
108        //         .as_ref()
109        //         .batch_fetch_resources_by_version(access_paths, 0)
110        //         .unwrap();
111        //     let validators: ValidatorSet = OnChainConfigPayload::new(
112        //         0,
113        //         Arc::new(
114        //             ON_CHAIN_CONFIG_REGISTRY
115        //                 .iter()
116        //                 .cloned()
117        //                 .zip_eq(configs)
118        //                 .collect(),
119        //         ),
120        //     )
121        //     .get()
122        //     .unwrap();
123        //     for node in validators {
124        //         let node_id = NodeID::new(
125        //             node.consensus_public_key().clone(),
126        //             node.vrf_public_key().clone().unwrap(),
127        //         );
128        //         initial_nodes.push((node_id, node.consensus_voting_power()));
129        //     }
130        // }
131        // TODO(lpl): The default value is only for pos-tool.
132        let genesis_pivot_decision =
133            genesis_pivot_decision.unwrap_or(PivotBlockDecision {
134                block_hash: Default::default(),
135                height: 0,
136            });
137        let pos_state = PosState::new(
138            initial_seed,
139            initial_nodes,
140            initial_committee,
141            genesis_pivot_decision,
142        );
143        Self {
144            db,
145            cache: Mutex::new(SpeculationCache::new_for_db_bootstrapping(
146                tree_state, pos_state,
147            )),
148        }
149    }
150
151    pub fn committed_block_id(&self) -> HashValue {
152        return self.cache.lock().committed_block_id();
153    }
154
155    pub fn update_block_tree_root(
156        &self, committed_trees: ExecutedTrees,
157        committed_ledger_info: &LedgerInfo, committed_txns: Vec<Transaction>,
158        reconfig_events: Vec<ContractEvent>,
159    ) {
160        self.cache.lock().update_block_tree_root(
161            committed_trees,
162            committed_ledger_info,
163            committed_txns,
164            reconfig_events,
165        )
166    }
167
168    pub fn update_synced_trees(&self, new_trees: ExecutedTrees) {
169        self.cache.lock().update_synced_trees(new_trees)
170    }
171
172    pub fn add_block(
173        &self, parent_block_id: HashValue,
174        block: (
175            HashValue,         /* block id */
176            Vec<Transaction>,  /* block transactions */
177            ProcessedVMOutput, /* block execution output */
178        ),
179    ) -> Result<(), Error> {
180        self.cache.lock().add_block(parent_block_id, block)
181    }
182
183    pub fn reset(&self) { self.cache.lock().reset() }
184
185    pub fn prune(
186        &self, committed_ledger_info: &LedgerInfo,
187        committed_txns: Vec<Transaction>, reconfig_events: Vec<ContractEvent>,
188    ) -> Result<HashValue, Error> {
189        self.cache.lock().prune(
190            committed_ledger_info,
191            committed_txns,
192            reconfig_events,
193        )
194    }
195
196    pub fn get_block(
197        &self, block_id: &HashValue,
198    ) -> Result<Arc<Mutex<SpeculationBlock>>, Error> {
199        self.cache.lock().get_block(block_id)
200    }
201}
cached_pos_ledger_db/lib.rs

cached_pos_ledger_db/
lib.rs
