consensus_core/
core_thread.rs

1// Copyright (c) Mysten Labs, Inc.
2// SPDX-License-Identifier: Apache-2.0
3
4use std::{collections::BTreeSet, fmt::Debug, sync::Arc};
5
6use async_trait::async_trait;
7use consensus_types::block::{BlockRef, Round};
8use mysten_metrics::{
9    monitored_mpsc::{Receiver, Sender, WeakSender, channel},
10    monitored_scope, spawn_logged_monitored_task,
11};
12use parking_lot::RwLock;
13use thiserror::Error;
14use tokio::sync::{oneshot, watch};
15use tracing::warn;
16
17use crate::{
18    block::VerifiedBlock,
19    commit::CertifiedCommits,
20    context::Context,
21    core::Core,
22    core_thread::CoreError::Shutdown,
23    dag_state::DagState,
24    error::{ConsensusError, ConsensusResult},
25    task::join_and_propagate_panic,
26};
27
28const CORE_THREAD_COMMANDS_CHANNEL_SIZE: usize = 2000;
29
30enum CoreThreadCommand {
31    /// Add blocks to be processed and accepted
32    AddBlocks(Vec<VerifiedBlock>, oneshot::Sender<BTreeSet<BlockRef>>),
33    /// Checks if block refs exist locally and sync missing ones.
34    CheckBlockRefs(Vec<BlockRef>, oneshot::Sender<BTreeSet<BlockRef>>),
35    /// Adds certified commits and their certification blocks for processing and acceptance.
36    /// Returns missing ancestors of certification voting blocks. Blocks included in certified commits
37    /// cannot have missing ancestors.
38    AddCertifiedCommits(CertifiedCommits, oneshot::Sender<BTreeSet<BlockRef>>),
39    /// Called when the min round has passed or the leader timeout occurred and a block should be produced.
40    /// When the command is called with `force = true`, then the block will be created for `round` skipping
41    /// any checks (ex leader existence of previous round). More information can be found on the `Core` component.
42    NewBlock(Round, oneshot::Sender<()>, bool),
43    /// Request missing blocks that need to be synced.
44    GetMissing(oneshot::Sender<BTreeSet<BlockRef>>),
45}
46
47#[derive(Error, Debug)]
48pub enum CoreError {
49    #[error("Core thread shutdown: {0}")]
50    Shutdown(String),
51}
52
53/// The interface to dispatch commands to CoreThread and Core.
54/// Also this allows the easier mocking during unit tests.
55#[async_trait]
56pub trait CoreThreadDispatcher: Sync + Send + 'static {
57    async fn add_blocks(&self, blocks: Vec<VerifiedBlock>)
58    -> Result<BTreeSet<BlockRef>, CoreError>;
59
60    async fn check_block_refs(
61        &self,
62        block_refs: Vec<BlockRef>,
63    ) -> Result<BTreeSet<BlockRef>, CoreError>;
64
65    async fn add_certified_commits(
66        &self,
67        commits: CertifiedCommits,
68    ) -> Result<BTreeSet<BlockRef>, CoreError>;
69
70    async fn new_block(&self, round: Round, force: bool) -> Result<(), CoreError>;
71
72    async fn get_missing_blocks(&self) -> Result<BTreeSet<BlockRef>, CoreError>;
73
74    /// Sets the estimated delay to propagate a block to a quorum of peers, in
75    /// number of rounds.
76    fn set_propagation_delay(&self, delay: Round) -> Result<(), CoreError>;
77
78    fn set_last_known_proposed_round(&self, round: Round) -> Result<(), CoreError>;
79}
80
81pub(crate) struct CoreThreadHandle {
82    sender: Sender<CoreThreadCommand>,
83    join_handle: tokio::task::JoinHandle<()>,
84}
85
86impl CoreThreadHandle {
87    pub async fn stop(self) {
88        // drop the sender, that will force all the other weak senders to not able to upgrade.
89        drop(self.sender);
90        join_and_propagate_panic(self.join_handle).await;
91    }
92}
93
94struct CoreThread {
95    core: Core,
96    receiver: Receiver<CoreThreadCommand>,
97    rx_propagation_delay: watch::Receiver<Round>,
98    rx_last_known_proposed_round: watch::Receiver<Round>,
99    context: Arc<Context>,
100}
101
102impl CoreThread {
103    pub async fn run(mut self) -> ConsensusResult<()> {
104        let result = self.run_inner().await;
105        self.core.stop().await;
106        result
107    }
108
109    async fn run_inner(&mut self) -> ConsensusResult<()> {
110        tracing::debug!("Started core thread");
111
112        loop {
113            tokio::select! {
114                command = self.receiver.recv() => {
115                    let Some(command) = command else {
116                        break;
117                    };
118                    self.context.metrics.node_metrics.core_lock_dequeued.inc();
119                    match command {
120                        CoreThreadCommand::AddBlocks(blocks, sender) => {
121                            let _scope = monitored_scope("CoreThread::loop::add_blocks");
122                            let missing_block_refs = self.core.add_blocks(blocks)?;
123                            sender.send(missing_block_refs).ok();
124                        }
125                        CoreThreadCommand::CheckBlockRefs(block_refs, sender) => {
126                            let _scope = monitored_scope("CoreThread::loop::check_block_refs");
127                            let missing_block_refs = self.core.check_block_refs(block_refs)?;
128                            sender.send(missing_block_refs).ok();
129                        }
130                        CoreThreadCommand::AddCertifiedCommits(commits, sender) => {
131                            let _scope = monitored_scope("CoreThread::loop::add_certified_commits");
132                            let missing_block_refs = self.core.add_certified_commits(commits)?;
133                            sender.send(missing_block_refs).ok();
134                        }
135                        CoreThreadCommand::NewBlock(round, sender, force) => {
136                            let _scope = monitored_scope("CoreThread::loop::new_block");
137                            self.core.new_block(round, force)?;
138                            sender.send(()).ok();
139                        }
140                        CoreThreadCommand::GetMissing(sender) => {
141                            let _scope = monitored_scope("CoreThread::loop::get_missing");
142                            sender.send(self.core.get_missing_blocks()).ok();
143                        }
144                    }
145                }
146                _ = self.rx_last_known_proposed_round.changed() => {
147                    let _scope = monitored_scope("CoreThread::loop::set_last_known_proposed_round");
148                    let round = *self.rx_last_known_proposed_round.borrow();
149                    self.core.set_last_known_proposed_round(round);
150                    // `round` arg is meant to avoid proposing below already proposed round.
151                    // Passing Round::MAX to select the threshold clock round for proposing.
152                    self.core.new_block(Round::MAX, true)?;
153                }
154                _ = self.rx_propagation_delay.changed() => {
155                    let _scope = monitored_scope("CoreThread::loop::set_propagation_delay");
156                    let should_propose_before = self.core.should_propose();
157                    let propagation_delay = *self.rx_propagation_delay.borrow();
158                    self.core.set_propagation_delay(
159                        propagation_delay
160                    );
161                    if !should_propose_before && self.core.should_propose() {
162                        // If core cannot propose before but can propose now, try to produce a new block to ensure liveness,
163                        // because block proposal could have been skipped.
164                        self.core.new_block(Round::MAX, true)?;
165                    }
166                }
167            }
168        }
169
170        Ok(())
171    }
172}
173
174#[derive(Clone)]
175pub(crate) struct ChannelCoreThreadDispatcher {
176    context: Arc<Context>,
177    sender: WeakSender<CoreThreadCommand>,
178    tx_propagation_delay: Arc<watch::Sender<Round>>,
179    tx_last_known_proposed_round: Arc<watch::Sender<Round>>,
180}
181
182impl ChannelCoreThreadDispatcher {
183    pub(crate) fn start(
184        context: Arc<Context>,
185        _dag_state: &RwLock<DagState>,
186        core: Core,
187    ) -> (Self, CoreThreadHandle) {
188        let (sender, receiver) =
189            channel("consensus_core_commands", CORE_THREAD_COMMANDS_CHANNEL_SIZE);
190        let (tx_propagation_delay, mut rx_propagation_delay) = watch::channel(0);
191        let (tx_last_known_proposed_round, mut rx_last_known_proposed_round) = watch::channel(0);
192        rx_propagation_delay.mark_unchanged();
193        rx_last_known_proposed_round.mark_unchanged();
194        let core_thread = CoreThread {
195            core,
196            receiver,
197            rx_propagation_delay,
198            rx_last_known_proposed_round,
199            context: context.clone(),
200        };
201
202        let join_handle = spawn_logged_monitored_task!(
203            async move {
204                if let Err(err) = core_thread.run().await
205                    && !matches!(err, ConsensusError::Shutdown)
206                {
207                    panic!("Fatal error occurred: {err}");
208                }
209            },
210            "ConsensusCoreThread"
211        );
212
213        // Explicitly using downgraded sender in order to allow sharing the CoreThreadDispatcher but
214        // able to shutdown the CoreThread by dropping the original sender.
215        let dispatcher = ChannelCoreThreadDispatcher {
216            context,
217            sender: sender.downgrade(),
218            tx_propagation_delay: Arc::new(tx_propagation_delay),
219            tx_last_known_proposed_round: Arc::new(tx_last_known_proposed_round),
220        };
221        let handle = CoreThreadHandle {
222            join_handle,
223            sender,
224        };
225        (dispatcher, handle)
226    }
227
228    async fn send(&self, command: CoreThreadCommand) {
229        self.context.metrics.node_metrics.core_lock_enqueued.inc();
230        if let Some(sender) = self.sender.upgrade()
231            && let Err(err) = sender.send(command).await
232        {
233            warn!(
234                "Couldn't send command to core thread, probably is shutting down: {}",
235                err
236            );
237        }
238    }
239}
240
241#[async_trait]
242impl CoreThreadDispatcher for ChannelCoreThreadDispatcher {
243    async fn add_blocks(
244        &self,
245        blocks: Vec<VerifiedBlock>,
246    ) -> Result<BTreeSet<BlockRef>, CoreError> {
247        let (sender, receiver) = oneshot::channel();
248        self.send(CoreThreadCommand::AddBlocks(blocks, sender))
249            .await;
250        let missing_block_refs = receiver.await.map_err(|e| Shutdown(e.to_string()))?;
251
252        Ok(missing_block_refs)
253    }
254
255    async fn check_block_refs(
256        &self,
257        block_refs: Vec<BlockRef>,
258    ) -> Result<BTreeSet<BlockRef>, CoreError> {
259        let (sender, receiver) = oneshot::channel();
260        self.send(CoreThreadCommand::CheckBlockRefs(block_refs, sender))
261            .await;
262        let missing_block_refs = receiver.await.map_err(|e| Shutdown(e.to_string()))?;
263
264        Ok(missing_block_refs)
265    }
266
267    async fn add_certified_commits(
268        &self,
269        commits: CertifiedCommits,
270    ) -> Result<BTreeSet<BlockRef>, CoreError> {
271        let (sender, receiver) = oneshot::channel();
272        self.send(CoreThreadCommand::AddCertifiedCommits(commits, sender))
273            .await;
274        let missing_block_refs = receiver.await.map_err(|e| Shutdown(e.to_string()))?;
275        Ok(missing_block_refs)
276    }
277
278    async fn new_block(&self, round: Round, force: bool) -> Result<(), CoreError> {
279        let (sender, receiver) = oneshot::channel();
280        self.send(CoreThreadCommand::NewBlock(round, sender, force))
281            .await;
282        receiver.await.map_err(|e| Shutdown(e.to_string()))
283    }
284
285    async fn get_missing_blocks(&self) -> Result<BTreeSet<BlockRef>, CoreError> {
286        let (sender, receiver) = oneshot::channel();
287        self.send(CoreThreadCommand::GetMissing(sender)).await;
288        receiver.await.map_err(|e| Shutdown(e.to_string()))
289    }
290
291    fn set_propagation_delay(&self, propagation_delay: Round) -> Result<(), CoreError> {
292        self.tx_propagation_delay
293            .send(propagation_delay)
294            .map_err(|e| Shutdown(e.to_string()))
295    }
296
297    fn set_last_known_proposed_round(&self, round: Round) -> Result<(), CoreError> {
298        self.tx_last_known_proposed_round
299            .send(round)
300            .map_err(|e| Shutdown(e.to_string()))
301    }
302}
303
304// TODO: complete the Mock for thread dispatcher to be used from several tests
305#[cfg(test)]
306#[derive(Default)]
307pub(crate) struct MockCoreThreadDispatcher {
308    add_blocks: parking_lot::Mutex<Vec<VerifiedBlock>>,
309    missing_blocks: parking_lot::Mutex<BTreeSet<BlockRef>>,
310    last_known_proposed_round: parking_lot::Mutex<Vec<Round>>,
311}
312
313#[cfg(test)]
314impl MockCoreThreadDispatcher {
315    #[cfg(test)]
316    pub(crate) async fn get_add_blocks(&self) -> Vec<VerifiedBlock> {
317        let mut add_blocks = self.add_blocks.lock();
318        add_blocks.drain(0..).collect()
319    }
320
321    #[cfg(test)]
322    pub(crate) async fn stub_missing_blocks(&self, block_refs: BTreeSet<BlockRef>) {
323        let mut missing_blocks = self.missing_blocks.lock();
324        missing_blocks.extend(block_refs);
325    }
326
327    #[cfg(test)]
328    pub(crate) async fn get_last_own_proposed_round(&self) -> Vec<Round> {
329        let last_known_proposed_round = self.last_known_proposed_round.lock();
330        last_known_proposed_round.clone()
331    }
332}
333
334#[cfg(test)]
335#[async_trait]
336impl CoreThreadDispatcher for MockCoreThreadDispatcher {
337    async fn add_blocks(
338        &self,
339        blocks: Vec<VerifiedBlock>,
340    ) -> Result<BTreeSet<BlockRef>, CoreError> {
341        let mut add_blocks = self.add_blocks.lock();
342        add_blocks.extend(blocks);
343        Ok(BTreeSet::new())
344    }
345
346    async fn check_block_refs(
347        &self,
348        _block_refs: Vec<BlockRef>,
349    ) -> Result<BTreeSet<BlockRef>, CoreError> {
350        Ok(BTreeSet::new())
351    }
352
353    async fn add_certified_commits(
354        &self,
355        _commits: CertifiedCommits,
356    ) -> Result<BTreeSet<BlockRef>, CoreError> {
357        todo!()
358    }
359
360    async fn new_block(&self, _round: Round, _force: bool) -> Result<(), CoreError> {
361        Ok(())
362    }
363
364    async fn get_missing_blocks(&self) -> Result<BTreeSet<BlockRef>, CoreError> {
365        let mut missing_blocks = self.missing_blocks.lock();
366        let result = missing_blocks.clone();
367        missing_blocks.clear();
368        Ok(result)
369    }
370
371    fn set_propagation_delay(&self, _propagation_delay: Round) -> Result<(), CoreError> {
372        todo!();
373    }
374
375    fn set_last_known_proposed_round(&self, round: Round) -> Result<(), CoreError> {
376        let mut last_known_proposed_round = self.last_known_proposed_round.lock();
377        last_known_proposed_round.push(round);
378        Ok(())
379    }
380}
381
382#[cfg(test)]
383mod test {
384    use std::time::Duration;
385
386    use parking_lot::RwLock;
387    use tokio::time::timeout;
388
389    use super::*;
390    use crate::{
391        CommitConsumerArgs,
392        block::{BlockAPI, TestBlock, genesis_blocks},
393        block_manager::BlockManager,
394        block_verifier::NoopBlockVerifier,
395        commit_observer::CommitObserver,
396        context::Context,
397        core::CoreSignals,
398        dag_state::DagState,
399        leader_schedule::LeaderSchedule,
400        round_tracker::RoundTracker,
401        storage::{Store, WriteBatch, mem_store::MemStore},
402        transaction::{TransactionClient, TransactionConsumer, TransactionConsumerPool},
403        transaction_vote_tracker::TransactionVoteTracker,
404    };
405
406    #[tokio::test]
407    async fn test_core_thread() {
408        telemetry_subscribers::init_for_testing();
409        let (context, mut key_pairs) = Context::new_for_test(4);
410        let context = Arc::new(context);
411        let store = Arc::new(MemStore::new());
412        let dag_state = Arc::new(RwLock::new(DagState::new(context.clone(), store.clone())));
413        let block_manager = BlockManager::new(context.clone(), dag_state.clone());
414        let (_transaction_client, tx_receiver, priority_tx_receiver) =
415            TransactionClient::new(context.clone());
416        let transaction_pool = Arc::new(TransactionConsumerPool::new(TransactionConsumer::new(
417            tx_receiver,
418            priority_tx_receiver,
419            context.clone(),
420        )));
421        let transaction_vote_tracker = TransactionVoteTracker::new(
422            context.clone(),
423            Arc::new(NoopBlockVerifier {}),
424            dag_state.clone(),
425        );
426        let (signals, signal_receivers) = CoreSignals::new(context.clone());
427        let _block_receiver = signal_receivers.block_broadcast_receiver();
428        let (commit_consumer, _commit_receiver) = CommitConsumerArgs::new(0, 0);
429        let commit_observer = CommitObserver::new(
430            context.clone(),
431            commit_consumer,
432            dag_state.clone(),
433            transaction_vote_tracker.clone(),
434        )
435        .await;
436        let leader_schedule = Arc::new(LeaderSchedule::from_store(
437            context.clone(),
438            dag_state.clone(),
439        ));
440        let round_tracker = Arc::new(RwLock::new(RoundTracker::new(context.clone(), vec![])));
441        let core = Core::new_validator(
442            context.clone(),
443            leader_schedule,
444            transaction_pool,
445            transaction_vote_tracker,
446            block_manager,
447            commit_observer,
448            signals,
449            key_pairs.remove(context.own_index.value()).1,
450            dag_state.clone(),
451            false,
452            round_tracker,
453        );
454
455        let (core_dispatcher, handle) =
456            ChannelCoreThreadDispatcher::start(context, &dag_state, core);
457
458        // Now create some clones of the dispatcher
459        let dispatcher_1 = core_dispatcher.clone();
460        let dispatcher_2 = core_dispatcher.clone();
461
462        // Try to send some commands
463        assert!(dispatcher_1.add_blocks(vec![]).await.is_ok());
464        assert!(dispatcher_2.add_blocks(vec![]).await.is_ok());
465
466        // Now shutdown the dispatcher
467        handle.stop().await;
468
469        // Try to send some commands
470        assert!(dispatcher_1.add_blocks(vec![]).await.is_err());
471        assert!(dispatcher_2.add_blocks(vec![]).await.is_err());
472    }
473
474    #[tokio::test]
475    async fn test_last_known_sync_wakes_threshold_clock_round() {
476        telemetry_subscribers::init_for_testing();
477        let (context, mut key_pairs) = Context::new_for_test(4);
478        let context = Arc::new(context);
479        let store = Arc::new(MemStore::new());
480
481        let mut last_round_blocks = genesis_blocks(&context);
482        let mut all_blocks = last_round_blocks.clone();
483        for round in 1..=2 {
484            let mut this_round_blocks = Vec::new();
485            for (index, _authority) in context.committee.authorities() {
486                let block = VerifiedBlock::new_for_test(
487                    TestBlock::new(round, index.value() as u32)
488                        .set_ancestors(last_round_blocks.iter().map(|b| b.reference()).collect())
489                        .build(),
490                );
491                this_round_blocks.push(block);
492            }
493            all_blocks.extend(this_round_blocks.clone());
494            last_round_blocks = this_round_blocks;
495        }
496        store
497            .write(WriteBatch::default().blocks(all_blocks))
498            .expect("Storage error");
499
500        let dag_state = Arc::new(RwLock::new(DagState::new(context.clone(), store)));
501        assert_eq!(
502            dag_state.read().get_last_proposed_block().unwrap().round(),
503            2
504        );
505        assert_eq!(dag_state.read().threshold_clock_round(), 3);
506
507        let block_manager = BlockManager::new(context.clone(), dag_state.clone());
508        let (_transaction_client, tx_receiver, priority_tx_receiver) =
509            TransactionClient::new(context.clone());
510        let transaction_pool = Arc::new(TransactionConsumerPool::new(TransactionConsumer::new(
511            tx_receiver,
512            priority_tx_receiver,
513            context.clone(),
514        )));
515        let transaction_vote_tracker = TransactionVoteTracker::new(
516            context.clone(),
517            Arc::new(NoopBlockVerifier {}),
518            dag_state.clone(),
519        );
520        transaction_vote_tracker.recover_blocks_after_round(dag_state.read().gc_round());
521        let (signals, signal_receivers) = CoreSignals::new(context.clone());
522        let mut block_receiver = signal_receivers.block_broadcast_receiver();
523        let (commit_consumer, _commit_receiver) = CommitConsumerArgs::new(0, 0);
524        let commit_observer = CommitObserver::new(
525            context.clone(),
526            commit_consumer,
527            dag_state.clone(),
528            transaction_vote_tracker.clone(),
529        )
530        .await;
531        let leader_schedule = Arc::new(LeaderSchedule::from_store(
532            context.clone(),
533            dag_state.clone(),
534        ));
535        let round_tracker = Arc::new(RwLock::new(RoundTracker::new(context.clone(), vec![])));
536        let core = Core::new_validator(
537            context.clone(),
538            leader_schedule,
539            transaction_pool,
540            transaction_vote_tracker,
541            block_manager,
542            commit_observer,
543            signals,
544            key_pairs.remove(context.own_index.value()).1,
545            dag_state.clone(),
546            true,
547            round_tracker,
548        );
549
550        let (core_dispatcher, handle) =
551            ChannelCoreThreadDispatcher::start(context, &dag_state, core);
552
553        let recovered_block = timeout(Duration::from_secs(5), block_receiver.recv())
554            .await
555            .expect("timed out waiting for recovered block")
556            .expect("block broadcast closed");
557        assert_eq!(recovered_block.block.round(), 2);
558
559        assert!(
560            timeout(Duration::from_millis(100), block_receiver.recv())
561                .await
562                .is_err(),
563            "round 3 must not be proposed before last-known sync completes"
564        );
565
566        core_dispatcher
567            .set_last_known_proposed_round(1)
568            .expect("core thread should be running");
569
570        let proposed_block = timeout(Duration::from_secs(5), async {
571            loop {
572                let block = block_receiver.recv().await.expect("block broadcast closed");
573                if block.block.round() == 3 {
574                    return block;
575                }
576            }
577        })
578        .await
579        .expect("timed out waiting for threshold-clock proposal");
580        assert_eq!(
581            proposed_block.block.author(),
582            core_dispatcher.context.own_index
583        );
584
585        handle.stop().await;
586    }
587}
consensus_core/core_thread.rs

consensus_core/
core_thread.rs
