agave-votor 4.2.0

//! Controls the queueing and firing of skip timer events for use
//! in the event loop.

mod stats;
mod timers;

use {
    crate::{common::DELTA_TIMEOUT, event::VotorEvent},
    agave_votor_messages::migration::MigrationStatus,
    crossbeam_channel::Sender,
    parking_lot::RwLock as PlRwLock,
    solana_clock::Slot,
    std::{
        sync::{
            Arc,
            atomic::{AtomicBool, Ordering},
        },
        thread::{self, JoinHandle},
        time::{Duration, Instant},
    },
    timers::Timers,
};

/// A manager of timer states.  Uses a background thread to trigger next ready
/// timers and send events.
pub(crate) struct TimerManager {
    timers: Arc<PlRwLock<Timers>>,
    handle: JoinHandle<()>,
}

impl TimerManager {
    pub(crate) fn new(
        event_sender: Sender<VotorEvent>,
        exit: Arc<AtomicBool>,
        migration_status: Arc<MigrationStatus>,
    ) -> Self {
        let timers = Arc::new(PlRwLock::new(Timers::new(DELTA_TIMEOUT, event_sender)));
        let handle = {
            let timers = Arc::clone(&timers);
            thread::spawn(move || {
                let _ = migration_status.wait_for_migration_or_exit(exit.as_ref());
                while !exit.load(Ordering::Relaxed) {
                    let duration = match timers.write().progress(Instant::now()) {
                        None => {
                            // No active timers, sleep for an arbitrary amount.
                            // This should be smaller than the minimum amount
                            // of time any newly added timers would take to expire.
                            Duration::from_millis(100)
                        }
                        Some(next_fire) => next_fire.duration_since(Instant::now()),
                    };
                    thread::sleep(duration);
                }
            })
        };

        Self { timers, handle }
    }

    pub(crate) fn set_timeouts(
        &self,
        slot: Slot,
        standstill_slot: Option<Slot>,
        delta_first_fec_set: Duration,
        delta_block: Duration,
    ) -> bool {
        self.timers.write().set_timeouts(
            slot,
            Instant::now(),
            standstill_slot,
            delta_first_fec_set,
            delta_block,
        )
    }

    pub(crate) fn join(self) {
        self.handle.join().unwrap();
    }

    #[cfg(test)]
    pub(crate) fn is_timeout_set(&self, slot: Slot) -> bool {
        self.timers.read().is_timeout_set(slot)
    }
}

#[cfg(test)]
mod tests {
    use {
        super::*, crate::event::VotorEvent, crossbeam_channel::bounded,
        solana_clock::DEFAULT_MS_PER_SLOT, std::time::Duration,
    };

    #[test]
    fn test_timer_manager() {
        let (event_sender, event_receiver) = bounded(1024);
        let exit = Arc::new(AtomicBool::new(false));
        let timer_manager = TimerManager::new(
            event_sender,
            exit.clone(),
            Arc::new(MigrationStatus::post_migration_status()),
        );
        let delta_block = Duration::from_millis(DEFAULT_MS_PER_SLOT);
        let delta_first_fec_set = delta_block;
        let slot = 52;
        let start = Instant::now();
        assert!(timer_manager.set_timeouts(slot, None, delta_first_fec_set, delta_block));
        assert!(!timer_manager.set_timeouts(slot, None, delta_first_fec_set, delta_block));
        // Should see the first two timeout events at DELTA_TIMEOUT + delta_block.
        let mut timeouts_received = 0;
        while timeouts_received < 2 && Instant::now().duration_since(start) < Duration::from_secs(2)
        {
            let recv = event_receiver.recv_timeout(Duration::from_millis(200));
            if let Ok(event) = recv {
                match event {
                    VotorEvent::Timeout(s) => {
                        assert_eq!(s, slot);
                        assert!(
                            Instant::now().duration_since(start) >= DELTA_TIMEOUT + delta_block
                        );
                        timeouts_received += 1;
                    }
                    VotorEvent::TimeoutCrashedLeader(s) => {
                        assert_eq!(s, slot);
                        assert!(
                            Instant::now().duration_since(start)
                                >= DELTA_TIMEOUT + delta_first_fec_set
                        );
                        timeouts_received += 1;
                    }
                    _ => panic!("Unexpected event: {event:?}"),
                }
            }
        }
        assert!(
            timeouts_received == 2,
            "Did not receive all expected timeouts"
        );
        exit.store(true, Ordering::Relaxed);
        timer_manager.join();
    }
}