cfxcore/pos/consensus/
state_computer.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
8use super::state_replication::StateComputer;
9use crate::pos::mempool::{CommitNotification, CommittedTransaction};
10use anyhow::Result;
11use consensus_types::block::Block;
12use diem_crypto::HashValue;
13use diem_logger::prelude::*;
14use diem_types::{
15    ledger_info::LedgerInfoWithSignatures, transaction::Transaction,
16};
17use executor_types::{
18    BlockExecutor, Error as ExecutionError, StateComputeResult,
19};
20use fail::fail_point;
21use futures::channel::{mpsc, oneshot};
22use parking_lot::Mutex;
23use std::boxed::Box;
24
25/// Basic communication with the Execution module;
26/// implements StateComputer traits.
27pub struct ExecutionProxy {
28    executor: Mutex<Box<dyn BlockExecutor>>,
29    mempool_commit_sender: mpsc::Sender<CommitNotification>,
30    mempool_commit_timeout_ms: u64,
31}
32
33impl ExecutionProxy {
34    pub fn new(
35        executor: Box<dyn BlockExecutor>,
36        mempool_commit_sender: mpsc::Sender<CommitNotification>,
37        mempool_commit_timeout_ms: u64,
38    ) -> Self {
39        Self {
40            executor: Mutex::new(executor),
41            mempool_commit_sender,
42            mempool_commit_timeout_ms,
43        }
44    }
45
46    /// Notify mempool of committed transactions so it can prune them.
47    async fn notify_mempool(&self, committed_txns: Vec<Transaction>) {
48        let user_txns: Vec<CommittedTransaction> = committed_txns
49            .iter()
50            .filter_map(|txn| match txn {
51                Transaction::UserTransaction(signed_txn) => {
52                    Some(CommittedTransaction {
53                        sender: signed_txn.sender(),
54                        hash: signed_txn.hash(),
55                    })
56                }
57                _ => None,
58            })
59            .collect();
60
61        if user_txns.is_empty() {
62            return;
63        }
64
65        let (callback, cb_receiver) = oneshot::channel();
66        let notification = CommitNotification {
67            transactions: user_txns,
68            callback,
69        };
70
71        if let Err(e) =
72            self.mempool_commit_sender.clone().try_send(notification)
73        {
74            diem_error!(
75                error = ?e,
76                "Failed to send commit notification to mempool"
77            );
78            return;
79        }
80
81        match tokio::time::timeout(
82            std::time::Duration::from_millis(self.mempool_commit_timeout_ms),
83            cb_receiver,
84        )
85        .await
86        {
87            Ok(Ok(Ok(response))) => {
88                if !response.success {
89                    diem_error!(
90                        "Mempool commit failed: {:?}",
91                        response.error_message
92                    );
93                }
94            }
95            Ok(Ok(Err(e))) => {
96                diem_error!(
97                    error = ?e,
98                    "Mempool commit returned error"
99                );
100            }
101            Ok(Err(_)) => {
102                diem_error!("Mempool commit callback dropped");
103            }
104            Err(_) => {
105                diem_error!(
106                    "Mempool commit notification timed out after {} ms",
107                    self.mempool_commit_timeout_ms
108                );
109            }
110        }
111    }
112}
113
114#[async_trait::async_trait]
115impl StateComputer for ExecutionProxy {
116    fn compute(
117        &self,
118        // The block to be executed.
119        block: &Block,
120        // The parent block id.
121        parent_block_id: HashValue,
122        catch_up_mode: bool,
123    ) -> Result<StateComputeResult, ExecutionError> {
124        fail_point!("consensus::compute", |_| {
125            Err(ExecutionError::InternalError {
126                error: "Injected error in compute".into(),
127            })
128        });
129        diem_debug!(
130            block_id = block.id(),
131            parent_id = block.parent_id(),
132            "Executing block",
133        );
134
135        self.executor.lock().execute_block(
136            id_and_transactions_from_block(block),
137            parent_block_id,
138            catch_up_mode,
139        )
140    }
141
142    async fn commit(
143        &self, block_ids: Vec<HashValue>,
144        finality_proof: LedgerInfoWithSignatures,
145    ) -> Result<(), ExecutionError> {
146        let committed_txns = self
147            .executor
148            .lock()
149            .commit_blocks(block_ids, finality_proof)?;
150        self.notify_mempool(committed_txns).await;
151        Ok(())
152    }
153}
154
155fn id_and_transactions_from_block(
156    block: &Block,
157) -> (HashValue, Vec<Transaction>) {
158    let id = block.id();
159    let mut transactions = vec![Transaction::BlockMetadata(block.into())];
160    transactions.extend(
161        block
162            .payload()
163            .unwrap_or(&vec![])
164            .iter()
165            .map(|txn| Transaction::UserTransaction(txn.clone())),
166    );
167    (id, transactions)
168}
cfxcore/pos/consensus/state_computer.rs

cfxcore/pos/consensus/
state_computer.rs
