cfxcore/pos/mempool/core_mempool/
mempool.rs

1// Copyright (c) The Diem Core Contributors
2// SPDX-License-Identifier: Apache-2.0
3
4// Copyright 2021 Conflux Foundation. All rights reserved.
5// Conflux is free software and distributed under GNU General Public License.
6// See http://www.gnu.org/licenses/
7
8//! Mempool is used to track transactions which have been submitted but not yet
9//! agreed upon.
10use crate::pos::mempool::{
11    core_mempool::{
12        index::TxnPointer,
13        transaction::{MempoolTransaction, TimelineState},
14        transaction_store::TransactionStore,
15    },
16    logging::{LogEntry, LogSchema, TxnsLog},
17};
18use diem_config::config::NodeConfig;
19use diem_crypto::{hash::CryptoHash, HashValue};
20use diem_logger::prelude::*;
21use diem_types::{
22    account_address::AccountAddress,
23    mempool_status::MempoolStatus,
24    term_state::PosState,
25    transaction::{
26        authenticator::TransactionAuthenticator, SignedTransaction,
27        TransactionPayload,
28    },
29    validator_verifier::ValidatorVerifier,
30};
31use executor::vm::verify_dispute;
32use std::{collections::HashSet, time::Duration};
33
34pub struct Mempool {
35    // Stores the metadata of all transactions in mempool (of all states).
36    pub transactions: TransactionStore,
37
38    pub system_transaction_timeout: Duration,
39}
40
41impl Mempool {
42    pub fn new(config: &NodeConfig) -> Self {
43        Mempool {
44            transactions: TransactionStore::new(&config.mempool),
45            system_transaction_timeout: Duration::from_secs(
46                config.mempool.system_transaction_timeout_secs,
47            ),
48        }
49    }
50
51    /// This function will be called once the transaction has been stored.
52    pub(crate) fn remove_transaction(&mut self, hash: HashValue) {
53        self.transactions.commit_transaction(hash);
54    }
55
56    /// Used to add a transaction to the Mempool.
57    /// Performs basic validation: checks account's sequence number.
58    pub(crate) fn add_txn(
59        &mut self, txn: SignedTransaction, timeline_state: TimelineState,
60    ) -> MempoolStatus {
61        diem_trace!(LogSchema::new(LogEntry::AddTxn)
62            .txns(TxnsLog::new_txn(txn.sender(), txn.hash())),);
63
64        let expiration_time = std::time::SystemTime::now()
65            .duration_since(std::time::UNIX_EPOCH)
66            .expect("System time is before UNIX_EPOCH")
67            + self.system_transaction_timeout;
68
69        let txn_info =
70            MempoolTransaction::new(txn, expiration_time, timeline_state);
71
72        self.transactions.insert(txn_info)
73    }
74
75    /// Fetches next block of transactions for consensus.
76    /// `batch_size` - size of requested block.
77    /// `seen_txns` - transactions that were sent to Consensus but were not
78    /// committed yet,  mempool should filter out such transactions.
79    #[allow(clippy::explicit_counter_loop)]
80    pub(crate) fn get_block(
81        &mut self, _batch_size: u64, mut seen: HashSet<TxnPointer>,
82        pos_state: &PosState, validators: ValidatorVerifier,
83    ) -> Vec<SignedTransaction> {
84        let mut block = vec![];
85        let mut block_log = TxnsLog::new();
86        let seen_size = seen.len();
87        let mut txn_walked = 0usize;
88        for txn in self.transactions.iter() {
89            txn_walked += 1;
90            if seen.contains(&TxnPointer::from(txn)) {
91                continue;
92            }
93            let validate_result = match txn.txn.payload() {
94                TransactionPayload::Election(election_payload) => {
95                    pos_state.validate_election(election_payload)
96                }
97                TransactionPayload::PivotDecision(_) => {
98                    seen.insert((txn.get_sender(), txn.get_hash()));
99                    continue;
100                }
101                TransactionPayload::Dispute(dispute_payload) => {
102                    // Every candidate sees the same unchanged state, so this
103                    // thins duplicates out without ruling them out.
104                    verify_dispute(dispute_payload, pos_state.current_view())
105                        .ok_or(anyhow::anyhow!("invalid dispute"))
106                        .and_then(|offense_epoch| {
107                            pos_state.validate_dispute(
108                                dispute_payload,
109                                offense_epoch,
110                            )
111                        })
112                }
113                _ => {
114                    continue;
115                }
116            };
117            if validate_result.is_ok() {
118                block.push(txn.txn.clone());
119                block_log.add(txn.get_sender(), txn.get_hash());
120                seen.insert((txn.get_sender(), txn.get_hash()));
121            }
122        }
123        let mut max_pivot_height = 0;
124        let mut chosen_pivot_tx = None;
125        // iterate all pivot decision transaction
126        for pivot_decision_set in self.transactions.iter_pivot_decision() {
127            let mut pivot_decision_opt = None;
128            diem_debug!("get_block: 0 {:?}", pivot_decision_set.len());
129            for (account, hash) in pivot_decision_set.iter() {
130                if validators.get_public_key(account).is_some() {
131                    if pivot_decision_opt.is_none() {
132                        if let Some(txn) = self.transactions.get(hash) {
133                            pivot_decision_opt = Some(txn);
134                        }
135                    }
136                }
137            }
138            diem_debug!("get_block: 1 {:?}", pivot_decision_opt);
139            if validators
140                .check_voting_power(
141                    pivot_decision_set.iter().map(|(addr, _)| addr),
142                )
143                .is_ok()
144            {
145                let pivot_decision = pivot_decision_opt.unwrap();
146                let pivot_height = match pivot_decision.payload() {
147                    TransactionPayload::PivotDecision(decision) => {
148                        decision.height
149                    }
150                    _ => unreachable!(),
151                };
152                if pivot_height > max_pivot_height
153                    && pivot_height > pos_state.pivot_decision().height
154                {
155                    max_pivot_height = pivot_height;
156                    chosen_pivot_tx = Some(pivot_decision);
157                }
158            }
159            diem_debug!("get_block: 2 {:?}", chosen_pivot_tx);
160        }
161        if let Some(tx) = chosen_pivot_tx {
162            let pivot_decision_hash = match tx.payload() {
163                TransactionPayload::PivotDecision(decision) => decision.hash(),
164                _ => unreachable!(),
165            };
166            // aggregate signatures
167            let txn_hashes =
168                self.transactions.get_pivot_decisions(&pivot_decision_hash);
169            let senders: Vec<AccountAddress> =
170                validators.get_ordered_account_addresses_iter().collect();
171            let mut signatures = vec![];
172            for hash in &txn_hashes {
173                if let Some(txn) = self.transactions.get(hash) {
174                    match txn.authenticator() {
175                        TransactionAuthenticator::BLS { signature, .. } => {
176                            if let Ok(index) =
177                                senders.binary_search(&txn.sender())
178                            {
179                                signatures.push((signature, index));
180                            }
181                        }
182                        _ => unreachable!(),
183                    }
184                }
185            }
186            let new_tx =
187                SignedTransaction::new_multisig(tx.raw_txn(), signatures);
188            block_log.add(new_tx.sender(), new_tx.hash());
189            block.push(new_tx);
190        }
191
192        diem_debug!(
193            LogSchema::new(LogEntry::GetBlock).txns(block_log),
194            seen_consensus = seen_size,
195            walked = txn_walked,
196            seen_after = seen.len(),
197            result_size = block.len(),
198            block_size = block.len()
199        );
200        block
201    }
202
203    /// Periodic core mempool garbage collection. Removes all expired
204    /// transactions by system TTL.
205    pub(crate) fn gc(&mut self) { self.transactions.gc_by_system_ttl(); }
206
207    /// Read `count` transactions from timeline since `timeline_id`.
208    /// Returns block of transactions and new last_timeline_id.
209    pub(crate) fn read_timeline(
210        &mut self, timeline_id: u64, count: usize,
211    ) -> (Vec<SignedTransaction>, u64) {
212        self.transactions.read_timeline(timeline_id, count)
213    }
214
215    /// Read transactions from timeline from `start_id` (exclusive) to `end_id`
216    /// (inclusive).
217    pub(crate) fn timeline_range(
218        &mut self, start_id: u64, end_id: u64,
219    ) -> Vec<SignedTransaction> {
220        self.transactions.timeline_range(start_id, end_id)
221    }
222}
cfxcore/pos/mempool/core_mempool/mempool.rs

cfxcore/pos/mempool/core_mempool/
mempool.rs
