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

consensus_core/
leader_timeout.rs
