consensus_core/
leader_timeout.rs

1// Copyright (c) Mysten Labs, Inc.
2// SPDX-License-Identifier: Apache-2.0
3use std::{sync::Arc, time::Duration};
4
5use consensus_types::block::Round;
6use tokio::{
7    sync::{
8        oneshot::{Receiver, Sender},
9        watch,
10    },
11    task::JoinHandle,
12    time::{Instant, sleep_until},
13};
14use tracing::{debug, warn};
15
16use crate::{context::Context, core::CoreSignalsReceivers, core_thread::CoreThreadDispatcher};
17
18pub(crate) struct LeaderTimeoutTaskHandle {
19    handle: JoinHandle<()>,
20    stop: Sender<()>,
21}
22
23impl LeaderTimeoutTaskHandle {
24    pub async fn stop(self) {
25        self.stop.send(()).ok();
26        self.handle.await.ok();
27    }
28}
29
30pub(crate) struct LeaderTimeoutTask<D: CoreThreadDispatcher> {
31    dispatcher: Arc<D>,
32    new_round_receiver: watch::Receiver<Round>,
33    leader_timeout: Duration,
34    min_round_delay: Duration,
35    stop: Receiver<()>,
36}
37
38impl<D: CoreThreadDispatcher> LeaderTimeoutTask<D> {
39    pub fn start(
40        dispatcher: Arc<D>,
41        signals_receivers: &CoreSignalsReceivers,
42        context: Arc<Context>,
43    ) -> LeaderTimeoutTaskHandle {
44        let (stop_sender, stop) = tokio::sync::oneshot::channel();
45        let mut me = Self {
46            dispatcher,
47            stop,
48            new_round_receiver: signals_receivers.new_round_receiver(),
49            leader_timeout: context.parameters.leader_timeout,
50            min_round_delay: context.parameters.min_round_delay,
51        };
52        let handle = tokio::spawn(async move { me.run().await });
53
54        LeaderTimeoutTaskHandle {
55            handle,
56            stop: stop_sender,
57        }
58    }
59
60    async fn run(&mut self) {
61        let new_round = &mut self.new_round_receiver;
62        let mut leader_round: Round = *new_round.borrow_and_update();
63        let mut min_leader_round_timed_out = false;
64        let mut max_leader_round_timed_out = false;
65        let timer_start = Instant::now();
66        let min_leader_timeout = sleep_until(timer_start + self.min_round_delay);
67        let max_leader_timeout = sleep_until(timer_start + self.leader_timeout);
68
69        tokio::pin!(min_leader_timeout);
70        tokio::pin!(max_leader_timeout);
71
72        loop {
73            tokio::select! {
74                // when the min leader timer expires then we attempt to trigger the creation of a new block.
75                // If we already timed out before then the branch gets disabled so we don't attempt
76                // all the time to produce already produced blocks for that round.
77                () = &mut min_leader_timeout, if !min_leader_round_timed_out => {
78                    if let Err(err) = self.dispatcher.new_block(leader_round, false).await {
79                        warn!("Error received while calling dispatcher, probably dispatcher is shutting down, will now exit: {err:?}");
80                        return;
81                    }
82                    min_leader_round_timed_out = true;
83                },
84                // When the max leader timer expires then we attempt to trigger the creation of a new block. This
85                // call is made with `force = true` to bypass any checks that allow to propose immediately if block
86                // not already produced.
87                // Keep in mind that first the min timeout should get triggered and then the max timeout, only
88                // if the round has not advanced in the meantime. Otherwise, the max timeout will not get
89                // triggered at all.
90                () = &mut max_leader_timeout, if !max_leader_round_timed_out => {
91                    debug!("Max leader timeout, will attempt new block for leader round {leader_round}");
92                    if let Err(err) = self.dispatcher.new_block(leader_round, true).await {
93                        warn!("Error received while calling dispatcher, probably dispatcher is shutting down, will now exit: {err:?}");
94                        return;
95                    }
96                    max_leader_round_timed_out = true;
97                }
98
99                // a new round has been produced. Reset the leader timeout.
100                Ok(_) = new_round.changed() => {
101                    leader_round = *new_round.borrow_and_update();
102                    debug!("New round has been received {leader_round}, resetting timer");
103
104                    min_leader_round_timed_out = false;
105                    max_leader_round_timed_out = false;
106
107                    let now = Instant::now();
108                    min_leader_timeout
109                    .as_mut()
110                    .reset(now + self.min_round_delay);
111                    max_leader_timeout
112                    .as_mut()
113                    .reset(now + self.leader_timeout);
114                },
115                _ = &mut self.stop => {
116                    debug!("Stop signal has been received, now shutting down");
117                    return;
118                }
119            }
120        }
121    }
122}
123
124#[cfg(test)]
125mod tests {
126    use std::{collections::BTreeSet, sync::Arc, time::Duration};
127
128    use async_trait::async_trait;
129    use consensus_config::Parameters;
130    use consensus_types::block::{BlockRef, Round};
131    use parking_lot::Mutex;
132    use tokio::time::{Instant, sleep};
133
134    use crate::{
135        block::VerifiedBlock,
136        commit::CertifiedCommits,
137        context::Context,
138        core::CoreSignals,
139        core_thread::{CoreError, CoreThreadDispatcher},
140        leader_timeout::LeaderTimeoutTask,
141    };
142
143    #[derive(Clone, Default)]
144    struct MockCoreThreadDispatcher {
145        new_block_calls: Arc<Mutex<Vec<(Round, bool, Instant)>>>,
146    }
147
148    impl MockCoreThreadDispatcher {
149        async fn get_new_block_calls(&self) -> Vec<(Round, bool, Instant)> {
150            let mut binding = self.new_block_calls.lock();
151            let all_calls = binding.drain(0..);
152            all_calls.into_iter().collect()
153        }
154    }
155
156    #[async_trait]
157    impl CoreThreadDispatcher for MockCoreThreadDispatcher {
158        async fn add_blocks(
159            &self,
160            _blocks: Vec<VerifiedBlock>,
161        ) -> Result<BTreeSet<BlockRef>, CoreError> {
162            todo!()
163        }
164
165        async fn check_block_refs(
166            &self,
167            _block_refs: Vec<BlockRef>,
168        ) -> Result<BTreeSet<BlockRef>, CoreError> {
169            todo!()
170        }
171
172        async fn add_certified_commits(
173            &self,
174            _commits: CertifiedCommits,
175        ) -> Result<BTreeSet<BlockRef>, CoreError> {
176            todo!()
177        }
178
179        async fn new_block(&self, round: Round, force: bool) -> Result<(), CoreError> {
180            self.new_block_calls
181                .lock()
182                .push((round, force, Instant::now()));
183            Ok(())
184        }
185
186        async fn get_missing_blocks(&self) -> Result<BTreeSet<BlockRef>, CoreError> {
187            todo!()
188        }
189
190        fn set_propagation_delay(&self, _propagation_delay: Round) -> Result<(), CoreError> {
191            todo!()
192        }
193
194        fn set_last_known_proposed_round(&self, _round: Round) -> Result<(), CoreError> {
195            todo!()
196        }
197    }
198
199    #[tokio::test(flavor = "current_thread", start_paused = true)]
200    async fn basic_leader_timeout() {
201        let (context, _signers) = Context::new_for_test(4);
202        let dispatcher = Arc::new(MockCoreThreadDispatcher::default());
203        let leader_timeout = Duration::from_millis(500);
204        let min_round_delay = Duration::from_millis(50);
205        let parameters = Parameters {
206            leader_timeout,
207            min_round_delay,
208            ..Default::default()
209        };
210        let context = Arc::new(context.with_parameters(parameters));
211        let start = Instant::now();
212
213        let (mut signals, signal_receivers) = CoreSignals::new(context.clone());
214
215        // spawn the task
216        let _handle = LeaderTimeoutTask::start(dispatcher.clone(), &signal_receivers, context);
217
218        // send a signal that a new round has been produced.
219        signals.new_round(10);
220
221        // wait enough until the min round delay has passed and a new_block call is triggered
222        sleep(2 * min_round_delay).await;
223        let all_calls = dispatcher.get_new_block_calls().await;
224        assert_eq!(all_calls.len(), 1);
225
226        let (round, force, timestamp) = all_calls[0];
227        assert_eq!(round, 10);
228        assert!(!force);
229        assert!(
230            min_round_delay <= timestamp - start,
231            "Leader timeout min setting {:?} should be less than actual time difference {:?}",
232            min_round_delay,
233            timestamp - start
234        );
235
236        // wait enough until a new_block has been received
237        sleep(2 * leader_timeout).await;
238        let all_calls = dispatcher.get_new_block_calls().await;
239        assert_eq!(all_calls.len(), 1);
240
241        let (round, force, timestamp) = all_calls[0];
242        assert_eq!(round, 10);
243        assert!(force);
244        assert!(
245            leader_timeout <= timestamp - start,
246            "Leader timeout setting {:?} should be less than actual time difference {:?}",
247            leader_timeout,
248            timestamp - start
249        );
250
251        // now wait another 2 * leader_timeout, no other call should be received
252        sleep(2 * leader_timeout).await;
253        let all_calls = dispatcher.get_new_block_calls().await;
254
255        assert_eq!(all_calls.len(), 0);
256    }
257
258    #[tokio::test(flavor = "current_thread", start_paused = true)]
259    async fn multiple_leader_timeouts() {
260        let (context, _signers) = Context::new_for_test(4);
261        let dispatcher = Arc::new(MockCoreThreadDispatcher::default());
262        let leader_timeout = Duration::from_millis(500);
263        let min_round_delay = Duration::from_millis(50);
264        let parameters = Parameters {
265            leader_timeout,
266            min_round_delay,
267            ..Default::default()
268        };
269        let context = Arc::new(context.with_parameters(parameters));
270        let now = Instant::now();
271
272        let (mut signals, signal_receivers) = CoreSignals::new(context.clone());
273
274        // spawn the task
275        let _handle = LeaderTimeoutTask::start(dispatcher.clone(), &signal_receivers, context);
276
277        // now send some signals with some small delay between them, but not enough so every round
278        // manages to timeout and call the force new block method.
279        signals.new_round(13);
280        sleep(min_round_delay / 2).await;
281        signals.new_round(14);
282        sleep(min_round_delay / 2).await;
283        signals.new_round(15);
284        sleep(2 * leader_timeout).await;
285
286        // only the last one should be received
287        let all_calls = dispatcher.get_new_block_calls().await;
288        let (round, force, timestamp) = all_calls[0];
289        assert_eq!(round, 15);
290        assert!(!force);
291        assert!(min_round_delay < timestamp - now);
292
293        let (round, force, timestamp) = all_calls[1];
294        assert_eq!(round, 15);
295        assert!(force);
296        assert!(leader_timeout < timestamp - now);
297    }
298}
consensus_core/leader_timeout.rs

consensus_core/
leader_timeout.rs
