ckb_block_filter/
filter.rs

1use ckb_async_runtime::tokio::{self, task::block_in_place};
2use ckb_logger::{debug, info, warn};
3use ckb_shared::Shared;
4use ckb_stop_handler::{CancellationToken, new_tokio_exit_rx};
5use ckb_store::{ChainDB, ChainStore};
6use ckb_types::{
7    core::HeaderView,
8    packed::{Byte32, CellOutput, OutPoint},
9    utilities::{FilterDataProvider, build_filter_data},
10};
11
12const NAME: &str = "BlockFilter";
13
14/// A block filter creation service
15#[derive(Clone)]
16pub struct BlockFilter {
17    shared: Shared,
18}
19
20struct WrappedChainDB<'a> {
21    inner: &'a ChainDB,
22}
23
24impl<'a> FilterDataProvider for WrappedChainDB<'a> {
25    fn cell(&self, out_point: &OutPoint) -> Option<CellOutput> {
26        self.inner
27            .get_transaction(&out_point.tx_hash())
28            .and_then(|(tx, _)| tx.outputs().get(out_point.index().into()))
29    }
30}
31
32impl<'a> WrappedChainDB<'a> {
33    fn new(inner: &'a ChainDB) -> Self {
34        Self { inner }
35    }
36}
37
38impl BlockFilter {
39    /// Create a new block filter service
40    pub fn new(shared: Shared) -> Self {
41        Self { shared }
42    }
43
44    /// start background single-threaded service to create block filter data
45    pub fn start(self) {
46        let notify_controller = self.shared.notify_controller().clone();
47        let async_handle = self.shared.async_handle().clone();
48        let stop_rx: CancellationToken = new_tokio_exit_rx();
49        let filter_data_builder = self.clone();
50
51        let build_filter_data =
52            async_handle.spawn_blocking(move || filter_data_builder.build_filter_data());
53
54        async_handle.spawn(async move {
55            let mut new_block_watcher = notify_controller.watch_new_block(NAME.to_string()).await;
56            let _build_filter_data_finished = build_filter_data.await;
57
58            loop {
59                tokio::select! {
60                    Ok(_) = new_block_watcher.changed() => {
61                        block_in_place(|| self.build_filter_data());
62                        new_block_watcher.borrow_and_update();
63                    }
64                    _ = stop_rx.cancelled() => {
65                        info!("BlockFilter received exit signal, exit now");
66                        break
67                    },
68                    else => break,
69                }
70            }
71        });
72    }
73
74    /// build block filter data to the latest block
75    fn build_filter_data(&self) {
76        let snapshot = self.shared.snapshot();
77        let tip_header = snapshot.get_tip_header().expect("tip stored");
78        let start_number = match snapshot.get_latest_built_filter_data_block_hash() {
79            Some(block_hash) => {
80                debug!("Hash of the latest created block {:#x}", block_hash);
81                if snapshot.is_main_chain(&block_hash) {
82                    let header = snapshot
83                        .get_block_header(&block_hash)
84                        .expect("header stored");
85                    debug!(
86                        "Latest created block on the main chain, starting from {}",
87                        header.number() + 1
88                    );
89                    header.number() + 1
90                } else {
91                    // find fork chain number
92                    let mut header = snapshot
93                        .get_block_header(&block_hash)
94                        .expect("header stored");
95                    while !snapshot.is_main_chain(&header.parent_hash()) {
96                        header = snapshot
97                            .get_block_header(&header.parent_hash())
98                            .expect("parent header stored");
99                    }
100                    debug!(
101                        "Block with the latest built filter data on the forked chain, starting from {}",
102                        header.number()
103                    );
104                    header.number()
105                }
106            }
107            None => 0,
108        };
109
110        for block_number in start_number..=tip_header.number() {
111            if ckb_stop_handler::has_received_stop_signal() {
112                info!("ckb has received stop signal, BlockFilter exit now");
113                return;
114            }
115
116            let block_hash = snapshot.get_block_hash(block_number).expect("index stored");
117            let header = snapshot
118                .get_block_header(&block_hash)
119                .expect("header stored");
120            self.build_filter_data_for_block(&header);
121        }
122    }
123
124    fn build_filter_data_for_block(&self, header: &HeaderView) {
125        debug!(
126            "Start building filter data for block: {}, hash: {:#x}",
127            header.number(),
128            header.hash()
129        );
130        let db = self.shared.store();
131        if db.get_block_filter_hash(&header.hash()).is_some() {
132            debug!(
133                "Filter data for block {:#x} already exists. Skip building.",
134                header.hash()
135            );
136            return;
137        }
138        let parent_block_filter_hash = if header.is_genesis() {
139            Byte32::zero()
140        } else {
141            db.get_block_filter_hash(&header.parent_hash())
142                .expect("parent block filter data stored")
143        };
144
145        let transactions = db.get_block_body(&header.hash());
146        let transactions_size: usize = transactions.iter().map(|tx| tx.data().total_size()).sum();
147        let provider = WrappedChainDB::new(db);
148        let (filter_data, missing_out_points) = build_filter_data(provider, &transactions);
149        for out_point in missing_out_points {
150            warn!(
151                "Unable to find the input cell for the out_point: {:#x}, \
152                Skip adding it to the filter. This should only happen during testing.",
153                out_point
154            );
155        }
156        let db_transaction = db.begin_transaction();
157        db_transaction
158            .insert_block_filter(
159                &header.hash(),
160                &(filter_data.clone().into()),
161                &parent_block_filter_hash,
162            )
163            .expect("insert_block_filter should be ok");
164        db_transaction.commit().expect("commit should be ok");
165        debug!(
166            "Inserted filter data for block: {}, hash: {:#x}, filter data size: {}, transactions size: {}",
167            header.number(),
168            header.hash(),
169            filter_data.len(),
170            transactions_size
171        );
172    }
173}
ckb_block_filter/filter.rs

ckb_block_filter/
filter.rs
