ckb_shared/
shared.rs

1//! Provide Shared
2#![allow(missing_docs)]
3use crate::block_status::BlockStatus;
4use crate::{HeaderMap, Snapshot, SnapshotMgr};
5use arc_swap::{ArcSwap, Guard};
6use ckb_async_runtime::Handle;
7use ckb_chain_spec::consensus::Consensus;
8use ckb_constant::store::TX_INDEX_UPPER_BOUND;
9use ckb_constant::sync::MAX_TIP_AGE;
10use ckb_db::{Direction, IteratorMode};
11use ckb_db_schema::{COLUMN_BLOCK_BODY, COLUMN_NUMBER_HASH};
12use ckb_error::{AnyError, Error};
13use ckb_logger::debug;
14use ckb_notify::NotifyController;
15use ckb_proposal_table::ProposalView;
16use ckb_stop_handler::{new_crossbeam_exit_rx, register_thread};
17use ckb_store::{ChainDB, ChainStore};
18use ckb_systemtime::unix_time_as_millis;
19use ckb_tx_pool::{BlockTemplate, TokioRwLock, TxPoolController};
20use ckb_types::{
21    core::{BlockNumber, EpochExt, EpochNumber, HeaderView, Version},
22    packed::{self, Byte32},
23    prelude::*,
24    H256, U256,
25};
26use ckb_util::{shrink_to_fit, Mutex, MutexGuard};
27use ckb_verification::cache::TxVerificationCache;
28use dashmap::DashMap;
29use std::cmp;
30use std::collections::BTreeMap;
31use std::sync::atomic::{AtomicBool, Ordering};
32use std::sync::Arc;
33use std::thread;
34use std::time::Duration;
35
36const FREEZER_INTERVAL: Duration = Duration::from_secs(60);
37const THRESHOLD_EPOCH: EpochNumber = 2;
38const MAX_FREEZE_LIMIT: BlockNumber = 30_000;
39
40pub const SHRINK_THRESHOLD: usize = 300;
41
42/// An owned permission to close on a freezer thread
43pub struct FreezerClose {
44    stopped: Arc<AtomicBool>,
45}
46
47impl Drop for FreezerClose {
48    fn drop(&mut self) {
49        self.stopped.store(true, Ordering::SeqCst);
50    }
51}
52
53/// TODO(doc): @quake
54#[derive(Clone)]
55pub struct Shared {
56    pub(crate) store: ChainDB,
57    pub(crate) tx_pool_controller: TxPoolController,
58    pub(crate) notify_controller: NotifyController,
59    pub(crate) txs_verify_cache: Arc<TokioRwLock<TxVerificationCache>>,
60    pub(crate) consensus: Arc<Consensus>,
61    pub(crate) snapshot_mgr: Arc<SnapshotMgr>,
62    pub(crate) async_handle: Handle,
63    pub(crate) ibd_finished: Arc<AtomicBool>,
64
65    pub(crate) assume_valid_targets: Arc<Mutex<Option<Vec<H256>>>>,
66    pub(crate) assume_valid_target_specified: Arc<Option<H256>>,
67
68    pub header_map: Arc<HeaderMap>,
69    pub(crate) block_status_map: Arc<DashMap<Byte32, BlockStatus>>,
70    pub(crate) unverified_tip: Arc<ArcSwap<crate::HeaderIndex>>,
71}
72
73impl Shared {
74    /// Construct new Shared
75    #[allow(clippy::too_many_arguments)]
76    pub fn new(
77        store: ChainDB,
78        tx_pool_controller: TxPoolController,
79        notify_controller: NotifyController,
80        txs_verify_cache: Arc<TokioRwLock<TxVerificationCache>>,
81        consensus: Arc<Consensus>,
82        snapshot_mgr: Arc<SnapshotMgr>,
83        async_handle: Handle,
84        ibd_finished: Arc<AtomicBool>,
85
86        assume_valid_targets: Arc<Mutex<Option<Vec<H256>>>>,
87        assume_valid_target_specified: Arc<Option<H256>>,
88        header_map: Arc<HeaderMap>,
89        block_status_map: Arc<DashMap<Byte32, BlockStatus>>,
90    ) -> Shared {
91        let header = store
92            .get_tip_header()
93            .unwrap_or(consensus.genesis_block().header());
94        let unverified_tip = Arc::new(ArcSwap::new(Arc::new(crate::HeaderIndex::new(
95            header.number(),
96            header.hash(),
97            header.difficulty(),
98        ))));
99
100        Shared {
101            store,
102            tx_pool_controller,
103            notify_controller,
104            txs_verify_cache,
105            consensus,
106            snapshot_mgr,
107            async_handle,
108            ibd_finished,
109            assume_valid_targets,
110            assume_valid_target_specified,
111            header_map,
112            block_status_map,
113            unverified_tip,
114        }
115    }
116    /// Spawn freeze background thread that periodically checks and moves ancient data from the kv database into the freezer.
117    pub fn spawn_freeze(&self) -> Option<FreezerClose> {
118        if let Some(freezer) = self.store.freezer() {
119            ckb_logger::info!("Freezer enabled");
120            let signal_receiver = new_crossbeam_exit_rx();
121            let shared = self.clone();
122            let freeze_jh = thread::Builder::new()
123                .spawn(move || loop {
124                    match signal_receiver.recv_timeout(FREEZER_INTERVAL) {
125                        Err(_) => {
126                            if let Err(e) = shared.freeze() {
127                                ckb_logger::error!("Freezer error {}", e);
128                                break;
129                            }
130                        }
131                        Ok(_) => {
132                            ckb_logger::info!("Freezer closing");
133                            break;
134                        }
135                    }
136                })
137                .expect("Start FreezerService failed");
138
139            register_thread("freeze", freeze_jh);
140
141            return Some(FreezerClose {
142                stopped: Arc::clone(&freezer.stopped),
143            });
144        }
145        None
146    }
147
148    fn freeze(&self) -> Result<(), Error> {
149        let freezer = self.store.freezer().expect("freezer inited");
150        let snapshot = self.snapshot();
151        let current_epoch = snapshot.epoch_ext().number();
152
153        if self.is_initial_block_download() {
154            ckb_logger::trace!("is_initial_block_download freeze skip");
155            return Ok(());
156        }
157
158        if current_epoch <= THRESHOLD_EPOCH {
159            ckb_logger::trace!("Freezer idles");
160            return Ok(());
161        }
162
163        let limit_block_hash = snapshot
164            .get_epoch_index(current_epoch + 1 - THRESHOLD_EPOCH)
165            .and_then(|index| snapshot.get_epoch_ext(&index))
166            .expect("get_epoch_ext")
167            .last_block_hash_in_previous_epoch();
168
169        let frozen_number = freezer.number();
170
171        let threshold = cmp::min(
172            snapshot
173                .get_block_number(&limit_block_hash)
174                .expect("get_block_number"),
175            frozen_number + MAX_FREEZE_LIMIT,
176        );
177
178        ckb_logger::trace!(
179            "Freezer current_epoch {} number {} threshold {}",
180            current_epoch,
181            frozen_number,
182            threshold
183        );
184
185        let store = self.store();
186        let get_unfrozen_block = |number: BlockNumber| {
187            store
188                .get_block_hash(number)
189                .and_then(|hash| store.get_unfrozen_block(&hash))
190        };
191
192        let ret = freezer.freeze(threshold, get_unfrozen_block)?;
193
194        let stopped = freezer.stopped.load(Ordering::SeqCst);
195
196        // Wipe out frozen data
197        self.wipe_out_frozen_data(&snapshot, ret, stopped)?;
198
199        ckb_logger::trace!("Freezer completed");
200
201        Ok(())
202    }
203
204    fn wipe_out_frozen_data(
205        &self,
206        snapshot: &Snapshot,
207        frozen: BTreeMap<packed::Byte32, (BlockNumber, u32)>,
208        stopped: bool,
209    ) -> Result<(), Error> {
210        let mut side = BTreeMap::new();
211        let mut batch = self.store.new_write_batch();
212
213        ckb_logger::trace!("freezer wipe_out_frozen_data {} ", frozen.len());
214
215        if !frozen.is_empty() {
216            // remain header
217            for (hash, (number, txs)) in &frozen {
218                batch.delete_block_body(*number, hash, *txs).map_err(|e| {
219                    ckb_logger::error!("Freezer delete_block_body failed {}", e);
220                    e
221                })?;
222
223                let pack_number: packed::Uint64 = number.pack();
224                let prefix = pack_number.as_slice();
225                for (key, value) in snapshot
226                    .get_iter(
227                        COLUMN_NUMBER_HASH,
228                        IteratorMode::From(prefix, Direction::Forward),
229                    )
230                    .take_while(|(key, _)| key.starts_with(prefix))
231                {
232                    let reader = packed::NumberHashReader::from_slice_should_be_ok(key.as_ref());
233                    let block_hash = reader.block_hash().to_entity();
234                    if &block_hash != hash {
235                        let txs =
236                            packed::Uint32Reader::from_slice_should_be_ok(value.as_ref()).unpack();
237                        side.insert(block_hash, (reader.number().to_entity(), txs));
238                    }
239                }
240            }
241            self.store.write_sync(&batch).map_err(|e| {
242                ckb_logger::error!("Freezer write_batch delete failed {}", e);
243                e
244            })?;
245            batch.clear()?;
246
247            if !stopped {
248                let start = frozen.keys().min().expect("frozen empty checked");
249                let end = frozen.keys().max().expect("frozen empty checked");
250                self.compact_block_body(start, end);
251            }
252        }
253
254        if !side.is_empty() {
255            // Wipe out side chain
256            for (hash, (number, txs)) in &side {
257                batch
258                    .delete_block(number.unpack(), hash, *txs)
259                    .map_err(|e| {
260                        ckb_logger::error!("Freezer delete_block_body failed {}", e);
261                        e
262                    })?;
263            }
264
265            self.store.write(&batch).map_err(|e| {
266                ckb_logger::error!("Freezer write_batch delete failed {}", e);
267                e
268            })?;
269
270            if !stopped {
271                let start = side.keys().min().expect("side empty checked");
272                let end = side.keys().max().expect("side empty checked");
273                self.compact_block_body(start, end);
274            }
275        }
276        Ok(())
277    }
278
279    fn compact_block_body(&self, start: &packed::Byte32, end: &packed::Byte32) {
280        let start_t = packed::TransactionKey::new_builder()
281            .block_hash(start.clone())
282            .index(0u32.pack())
283            .build();
284
285        let end_t = packed::TransactionKey::new_builder()
286            .block_hash(end.clone())
287            .index(TX_INDEX_UPPER_BOUND.pack())
288            .build();
289
290        if let Err(e) = self.store.compact_range(
291            COLUMN_BLOCK_BODY,
292            Some(start_t.as_slice()),
293            Some(end_t.as_slice()),
294        ) {
295            ckb_logger::error!("Freezer compact_range {}-{} error {}", start, end, e);
296        }
297    }
298
299    /// TODO(doc): @quake
300    pub fn tx_pool_controller(&self) -> &TxPoolController {
301        &self.tx_pool_controller
302    }
303
304    /// TODO(doc): @quake
305    pub fn txs_verify_cache(&self) -> Arc<TokioRwLock<TxVerificationCache>> {
306        Arc::clone(&self.txs_verify_cache)
307    }
308
309    /// TODO(doc): @quake
310    pub fn notify_controller(&self) -> &NotifyController {
311        &self.notify_controller
312    }
313
314    /// TODO(doc): @quake
315    pub fn snapshot(&self) -> Guard<Arc<Snapshot>> {
316        self.snapshot_mgr.load()
317    }
318
319    /// Return arc cloned snapshot
320    pub fn cloned_snapshot(&self) -> Arc<Snapshot> {
321        Arc::clone(&self.snapshot())
322    }
323
324    /// TODO(doc): @quake
325    pub fn store_snapshot(&self, snapshot: Arc<Snapshot>) {
326        self.snapshot_mgr.store(snapshot)
327    }
328
329    /// TODO(doc): @quake
330    pub fn refresh_snapshot(&self) {
331        let new = self.snapshot().refresh(self.store.get_snapshot());
332        self.store_snapshot(Arc::new(new));
333    }
334
335    /// TODO(doc): @quake
336    pub fn new_snapshot(
337        &self,
338        tip_header: HeaderView,
339        total_difficulty: U256,
340        epoch_ext: EpochExt,
341        proposals: ProposalView,
342    ) -> Arc<Snapshot> {
343        Arc::new(Snapshot::new(
344            tip_header,
345            total_difficulty,
346            epoch_ext,
347            self.store.get_snapshot(),
348            proposals,
349            Arc::clone(&self.consensus),
350        ))
351    }
352
353    /// TODO(doc): @quake
354    pub fn consensus(&self) -> &Consensus {
355        &self.consensus
356    }
357
358    /// Return arc cloned consensus re
359    pub fn cloned_consensus(&self) -> Arc<Consensus> {
360        Arc::clone(&self.consensus)
361    }
362
363    /// Return async runtime handle
364    pub fn async_handle(&self) -> &Handle {
365        &self.async_handle
366    }
367
368    /// TODO(doc): @quake
369    pub fn genesis_hash(&self) -> Byte32 {
370        self.consensus.genesis_hash()
371    }
372
373    /// TODO(doc): @quake
374    pub fn store(&self) -> &ChainDB {
375        &self.store
376    }
377
378    /// Return whether chain is in initial block download
379    pub fn is_initial_block_download(&self) -> bool {
380        // Once this function has returned false, it must remain false.
381        if self.ibd_finished.load(Ordering::Acquire) {
382            false
383        } else if unix_time_as_millis().saturating_sub(self.snapshot().tip_header().timestamp())
384            > MAX_TIP_AGE
385        {
386            true
387        } else {
388            self.ibd_finished.store(true, Ordering::Release);
389            false
390        }
391    }
392
393    /// Generate and return block_template
394    pub fn get_block_template(
395        &self,
396        bytes_limit: Option<u64>,
397        proposals_limit: Option<u64>,
398        max_version: Option<Version>,
399    ) -> Result<Result<BlockTemplate, AnyError>, AnyError> {
400        self.tx_pool_controller().get_block_template(
401            bytes_limit,
402            proposals_limit,
403            max_version.map(Into::into),
404        )
405    }
406
407    pub fn set_unverified_tip(&self, header: crate::HeaderIndex) {
408        self.unverified_tip.store(Arc::new(header));
409    }
410    pub fn get_unverified_tip(&self) -> crate::HeaderIndex {
411        self.unverified_tip.load().as_ref().clone()
412    }
413
414    pub fn header_map(&self) -> &HeaderMap {
415        &self.header_map
416    }
417    pub fn remove_header_view(&self, hash: &Byte32) {
418        self.header_map.remove(hash);
419    }
420
421    pub fn block_status_map(&self) -> &DashMap<Byte32, BlockStatus> {
422        &self.block_status_map
423    }
424
425    pub fn get_block_status(&self, block_hash: &Byte32) -> BlockStatus {
426        match self.block_status_map().get(block_hash) {
427            Some(status_ref) => *status_ref.value(),
428            None => {
429                if self.header_map().contains_key(block_hash) {
430                    BlockStatus::HEADER_VALID
431                } else {
432                    let verified = self
433                        .snapshot()
434                        .get_block_ext(block_hash)
435                        .map(|block_ext| block_ext.verified);
436                    match verified {
437                        None => BlockStatus::UNKNOWN,
438                        Some(None) => BlockStatus::BLOCK_STORED,
439                        Some(Some(true)) => BlockStatus::BLOCK_VALID,
440                        Some(Some(false)) => BlockStatus::BLOCK_INVALID,
441                    }
442                }
443            }
444        }
445    }
446
447    pub fn contains_block_status<T: ChainStore>(
448        &self,
449        block_hash: &Byte32,
450        status: BlockStatus,
451    ) -> bool {
452        self.get_block_status(block_hash).contains(status)
453    }
454
455    pub fn insert_block_status(&self, block_hash: Byte32, status: BlockStatus) {
456        self.block_status_map.insert(block_hash, status);
457    }
458
459    pub fn remove_block_status(&self, block_hash: &Byte32) {
460        let log_now = std::time::Instant::now();
461        self.block_status_map.remove(block_hash);
462        debug!("remove_block_status cost {:?}", log_now.elapsed());
463        shrink_to_fit!(self.block_status_map, SHRINK_THRESHOLD);
464        debug!(
465            "remove_block_status shrink_to_fit cost {:?}",
466            log_now.elapsed()
467        );
468    }
469
470    pub fn assume_valid_targets(&self) -> MutexGuard<Option<Vec<H256>>> {
471        self.assume_valid_targets.lock()
472    }
473
474    pub fn assume_valid_target_specified(&self) -> Arc<Option<H256>> {
475        Arc::clone(&self.assume_valid_target_specified)
476    }
477}
ckb_shared/shared.rs

ckb_shared/
shared.rs
