feat: Updates timers to send Crashed leader and timeout events (#344)

Part of #243.

    Rename SkipTimer to TimerManager as now the timers are not just for skip but also for crashed leaders.
    Refactors the core state machine of sending out different types of events out of the main loop of the timers.
    Sends out additional crashed leader events on top of skip events.

Co-authored-by: Quentin Kniep <[email protected]>

Author: 2 people

Cargo.lock

@@ -247,6 +247,7 @@ impl Tvu {
             rpc_subscriptions.clone(),
             slot_status_notifier.clone(),
             tvu_config.xdp_sender,
+            votor_event_sender.clone(),
         );
 
         let (ancestor_duplicate_slots_sender, ancestor_duplicate_slots_receiver) = unbounded();

core/src/tvu.rs

@@ -7,7 +7,7 @@ use {
         xdp::{XdpSender, XdpShredPayload},
     },
     bytes::Bytes,
-    crossbeam_channel::{Receiver, RecvError, TryRecvError},
+    crossbeam_channel::{Receiver, RecvError, Sender, TryRecvError},
     lru::LruCache,
     rand::Rng,
     rayon::{prelude::*, ThreadPool, ThreadPoolBuilder},
@@ -34,6 +34,7 @@ use {
         socket::SocketAddrSpace,
     },
     solana_time_utils::timestamp,
+    solana_votor::event::VotorEvent,
     static_assertions::const_assert_eq,
     std::{
         borrow::Cow,
@@ -242,6 +243,7 @@ fn retransmit(
     rpc_subscriptions: Option<&RpcSubscriptions>,
     slot_status_notifier: Option<&SlotStatusNotifier>,
     shred_buf: &mut Vec<Vec<shred::Payload>>,
+    votor_event_sender: &Sender<VotorEvent>,
 ) -> Result<(), RecvError> {
     // Try to receive shreds from the channel without blocking. If the channel
     // is empty precompute turbine trees speculatively. If no cache updates are
@@ -380,6 +382,7 @@ fn retransmit(
         addr_cache,
         rpc_subscriptions,
         slot_status_notifier,
+        votor_event_sender,
     );
     timer_start.stop();
     stats.total_time += timer_start.as_us();
@@ -593,6 +596,7 @@ impl RetransmitStage {
         rpc_subscriptions: Option<Arc<RpcSubscriptions>>,
         slot_status_notifier: Option<SlotStatusNotifier>,
         xdp_sender: Option<XdpSender>,
+        votor_event_sender: Sender<VotorEvent>,
     ) -> Self {
         let cluster_nodes_cache = ClusterNodesCache::<RetransmitStage>::new(
             CLUSTER_NODES_CACHE_NUM_EPOCH_CAP,
@@ -614,30 +618,29 @@ impl RetransmitStage {
 
         let retransmit_thread_handle = Builder::new()
             .name("solRetransmittr".to_string())
-            .spawn({
-                move || {
-                    let mut shred_buf = Vec::with_capacity(RETRANSMIT_BATCH_SIZE);
-                    while retransmit(
-                        &thread_pool,
-                        &bank_forks,
-                        &leader_schedule_cache,
-                        &cluster_info,
-                        &retransmit_receiver,
-                        &retransmit_sockets,
-                        &quic_endpoint_sender,
-                        xdp_sender.as_ref(),
-                        &mut stats,
-                        &cluster_nodes_cache,
-                        &mut addr_cache,
-                        &mut shred_deduper,
-                        &max_slots,
-                        rpc_subscriptions.as_deref(),
-                        slot_status_notifier.as_ref(),
-                        &mut shred_buf,
-                    )
-                    .is_ok()
-                    {}
-                }
+            .spawn(move || {
+                let mut shred_buf = Vec::with_capacity(RETRANSMIT_BATCH_SIZE);
+                while retransmit(
+                    &thread_pool,
+                    &bank_forks,
+                    &leader_schedule_cache,
+                    &cluster_info,
+                    &retransmit_receiver,
+                    &retransmit_sockets,
+                    &quic_endpoint_sender,
+                    xdp_sender.as_ref(),
+                    &mut stats,
+                    &cluster_nodes_cache,
+                    &mut addr_cache,
+                    &mut shred_deduper,
+                    &max_slots,
+                    rpc_subscriptions.as_deref(),
+                    slot_status_notifier.as_ref(),
+                    &mut shred_buf,
+                    &votor_event_sender,
+                )
+                .is_ok()
+                {}
             })
             .unwrap();
 
@@ -717,6 +720,7 @@ impl RetransmitStats {
         addr_cache: &mut AddrCache,
         rpc_subscriptions: Option<&RpcSubscriptions>,
         slot_status_notifier: Option<&SlotStatusNotifier>,
+        votor_event_sender: &Sender<VotorEvent>,
     ) {
         for (slot, mut slot_stats) in feed {
             addr_cache.record(slot, &mut slot_stats);
@@ -728,6 +732,7 @@ impl RetransmitStats {
                             slot_stats.outset,
                             rpc_subscriptions,
                             slot_status_notifier,
+                            votor_event_sender,
                         );
                     }
                     self.slot_stats.put(slot, slot_stats);
@@ -821,6 +826,7 @@ fn notify_subscribers(
     timestamp: u64, // When the first shred in the slot was received.
     rpc_subscriptions: Option<&RpcSubscriptions>,
     slot_status_notifier: Option<&SlotStatusNotifier>,
+    votor_event_sender: &Sender<VotorEvent>,
 ) {
     if let Some(rpc_subscriptions) = rpc_subscriptions {
         let slot_update = SlotUpdate::FirstShredReceived { slot, timestamp };
@@ -833,6 +839,9 @@ fn notify_subscribers(
             .unwrap()
             .notify_first_shred_received(slot);
     }
+    votor_event_sender
+        .send(VotorEvent::FirstShred(slot))
+        .unwrap();
 }
 
 #[cfg(test)]

turbine/src/retransmit_stage.rs

@@ -33,6 +33,7 @@ dashmap = { workspace = true, features = ["rayon", "raw-api"] }
 etcd-client = { workspace = true, features = ["tls"] }
 itertools = { workspace = true }
 log = { workspace = true }
+parking_lot = { workspace = true }
 qualifier_attr = { workspace = true }
 rayon = { workspace = true }
 serde = { workspace = true }

votor/Cargo.toml

@@ -35,10 +35,17 @@ pub enum VotorEvent {
     /// The block has received a notarization certificate
     BlockNotarized(Block),
 
+    /// Received the first shred for the slot.
+    FirstShred(Slot),
+
     /// The pool has marked the given block as a ready parent for `slot`
     ParentReady { slot: Slot, parent_block: Block },
 
-    /// The skip timer has fired for the given slot
+    //// Timeout to early detect that a honest that has crashed and
+    /// if the leader window should be skipped.
+    TimeoutCrashedLeader(Slot),
+
+    /// Timeout to inspect whether the remaining leader window should be skipped.
     Timeout(Slot),
 
     /// The given block has reached the safe to notar status
@@ -67,16 +74,18 @@ impl VotorEvent {
     pub(crate) fn should_ignore(&self, root: Slot) -> bool {
         match self {
             VotorEvent::Block(completed_block) => completed_block.slot <= root,
-            VotorEvent::BlockNotarized((s, _)) => *s <= root,
-            VotorEvent::ParentReady {
-                slot,
+            VotorEvent::Timeout(s)
+            | VotorEvent::SafeToSkip(s)
+            | VotorEvent::TimeoutCrashedLeader(s)
+            | VotorEvent::FirstShred(s)
+            | VotorEvent::SafeToNotar((s, _))
+            | VotorEvent::Finalized((s, _))
+            | VotorEvent::BlockNotarized((s, _))
+            | VotorEvent::ParentReady {
+                slot: s,
                 parent_block: _,
-            } => *slot <= root,
-            VotorEvent::Timeout(s) => *s <= root,
-            VotorEvent::SafeToNotar((s, _)) => *s <= root,
-            VotorEvent::SafeToSkip(s) => *s <= root,
+            } => s <= &root,
             VotorEvent::ProduceWindow(_) => false,
-            VotorEvent::Finalized((s, _)) => *s <= root,
             VotorEvent::Standstill(_) => false,
             VotorEvent::SetIdentity => false,
         }

votor/src/event.rs

@@ -6,12 +6,13 @@ use {
         commitment::{alpenglow_update_commitment_cache, AlpenglowCommitmentType},
         event::{CompletedBlock, VotorEvent, VotorEventReceiver},
         root_utils::{self, RootContext},
-        skip_timer::SkipTimerManager,
+        timer_manager::TimerManager,
         vote_history::{VoteHistory, VoteHistoryError},
         voting_utils::{self, BLSOp, VoteError, VotingContext},
         votor::{SharedContext, Votor},
     },
     crossbeam_channel::{select, RecvError, SendError},
+    parking_lot::RwLock,
     solana_clock::Slot,
     solana_hash::Hash,
     solana_ledger::leader_schedule_utils::{
@@ -25,7 +26,7 @@ use {
         collections::{BTreeMap, BTreeSet},
         sync::{
             atomic::{AtomicBool, Ordering},
-            Arc, Condvar, Mutex, RwLock,
+            Arc, Condvar, Mutex,
         },
         thread::{self, Builder, JoinHandle},
         time::Duration,
@@ -43,7 +44,7 @@ pub(crate) struct EventHandlerContext {
     pub(crate) start: Arc<(Mutex<bool>, Condvar)>,
 
     pub(crate) event_receiver: VotorEventReceiver,
-    pub(crate) skip_timer: Arc<RwLock<SkipTimerManager>>,
+    pub(crate) timer_manager: Arc<RwLock<TimerManager>>,
 
     // Contexts
     pub(crate) shared_context: SharedContext,
@@ -94,14 +95,15 @@ impl EventHandler {
             exit,
             start,
             event_receiver,
-            skip_timer,
+            timer_manager,
             shared_context: ctx,
             voting_context: mut vctx,
             root_context: rctx,
         } = context;
         let mut my_pubkey = vctx.identity_keypair.pubkey();
         let mut pending_blocks = PendingBlocks::default();
         let mut finalized_blocks = BTreeSet::default();
+        let mut received_shred = BTreeSet::default();
 
         // Wait until migration has completed
         info!("{my_pubkey}: Event loop initialized");
@@ -139,12 +141,13 @@ impl EventHandler {
             let votes = Self::handle_event(
                 &mut my_pubkey,
                 event,
-                &skip_timer,
+                &timer_manager,
                 &ctx,
                 &mut vctx,
                 &rctx,
                 &mut pending_blocks,
                 &mut finalized_blocks,
+                &mut received_shred,
             )?;
 
             // TODO: properly bubble up error handling here and in call graph
@@ -159,12 +162,13 @@ impl EventHandler {
     fn handle_event(
         my_pubkey: &mut Pubkey,
         event: VotorEvent,
-        skip_timer: &RwLock<SkipTimerManager>,
+        timer_manager: &RwLock<TimerManager>,
         ctx: &SharedContext,
         vctx: &mut VotingContext,
         rctx: &RootContext,
         pending_blocks: &mut PendingBlocks,
         finalized_blocks: &mut BTreeSet<Block>,
+        received_shred: &mut BTreeSet<Slot>,
     ) -> Result<Vec<Result<BLSOp, VoteError>>, EventLoopError> {
         let mut votes = vec![];
         match event {
@@ -195,6 +199,7 @@ impl EventHandler {
                     rctx,
                     pending_blocks,
                     finalized_blocks,
+                    received_shred,
                 )?;
             }
 
@@ -205,16 +210,29 @@ impl EventHandler {
                 Self::try_final(my_pubkey, block, vctx, &mut votes);
             }
 
+            VotorEvent::FirstShred(slot) => {
+                info!("{my_pubkey}: First shred {slot}");
+                received_shred.insert(slot);
+            }
+
             // Received a parent ready notification for `slot`
             VotorEvent::ParentReady { slot, parent_block } => {
                 info!("{my_pubkey}: Parent ready {slot} {parent_block:?}");
                 let should_set_timeouts = vctx.vote_history.add_parent_ready(slot, parent_block);
                 Self::check_pending_blocks(my_pubkey, pending_blocks, vctx, &mut votes);
                 if should_set_timeouts {
-                    skip_timer.write().unwrap().set_timeouts(slot);
+                    timer_manager.write().set_timeouts(slot);
                 }
             }
 
+            VotorEvent::TimeoutCrashedLeader(slot) => {
+                info!("{my_pubkey}: TimeoutCrashedLeader {slot}");
+                if vctx.vote_history.voted(slot) || received_shred.contains(&slot) {
+                    return Ok(votes);
+                }
+                Self::try_skip_window(my_pubkey, slot, vctx, &mut votes);
+            }
+
             // Skip timer for the slot has fired
             VotorEvent::Timeout(slot) => {
                 info!("{my_pubkey}: Timeout {slot}");
@@ -289,6 +307,7 @@ impl EventHandler {
                     rctx,
                     pending_blocks,
                     finalized_blocks,
+                    received_shred,
                 )?;
             }
 
@@ -538,6 +557,7 @@ impl EventHandler {
         rctx: &RootContext,
         pending_blocks: &mut PendingBlocks,
         finalized_blocks: &mut BTreeSet<Block>,
+        received_shred: &mut BTreeSet<Slot>,
     ) -> Result<(), SetRootError> {
         let bank_forks_r = ctx.bank_forks.read().unwrap();
         let old_root = bank_forks_r.root();
@@ -565,6 +585,7 @@ impl EventHandler {
             rctx,
             pending_blocks,
             finalized_blocks,
+            received_shred,
         )
     }
 

votor/src/event_handler.rs

@@ -10,7 +10,7 @@ pub mod commitment;
 pub mod event;
 mod event_handler;
 pub mod root_utils;
-mod skip_timer;
+mod timer_manager;
 pub mod vote_history;
 pub mod vote_history_storage;
 pub mod voting_utils;

votor/src/lib.rs

@@ -41,11 +41,13 @@ pub(crate) fn set_root(
     rctx: &RootContext,
     pending_blocks: &mut PendingBlocks,
     finalized_blocks: &mut BTreeSet<Block>,
+    received_shred: &mut BTreeSet<Slot>,
 ) -> Result<(), SetRootError> {
     info!("{my_pubkey}: setting root {new_root}");
     vctx.vote_history.set_root(new_root);
     *pending_blocks = pending_blocks.split_off(&new_root);
     *finalized_blocks = finalized_blocks.split_off(&(new_root, Hash::default()));
+    *received_shred = received_shred.split_off(&new_root);
 
     check_and_handle_new_root(
         new_root,