Line data    Source code

       1              : use std::{
       2              :     cmp::Ordering,
       3              :     collections::BinaryHeap,
       4              :     ops::DerefMut,
       5              :     sync::{
       6              :         atomic::{AtomicU32, AtomicU64},
       7              :         Arc,
       8              :     },
       9              : };
      10              : 
      11              : use parking_lot::Mutex;
      12              : use tracing::trace;
      13              : 
      14              : use crate::executor::ThreadContext;
      15              : 
      16              : /// Holds current time and all pending wakeup events.
      17              : pub struct Timing {
      18              :     /// Current world's time.
      19              :     current_time: AtomicU64,
      20              :     /// Pending timers.
      21              :     queue: Mutex<BinaryHeap<Pending>>,
      22              :     /// Global nonce. Makes picking events from binary heap queue deterministic
      23              :     /// by appending a number to events with the same timestamp.
      24              :     nonce: AtomicU32,
      25              :     /// Used to schedule fake events.
      26              :     fake_context: Arc<ThreadContext>,
      27              : }
      28              : 
      29              : impl Default for Timing {
      30            0 :     fn default() -> Self {
      31            0 :         Self::new()
      32            0 :     }
      33              : }
      34              : 
      35              : impl Timing {
      36              :     /// Create a new empty clock with time set to 0.
      37         4056 :     pub fn new() -> Timing {
      38         4056 :         Timing {
      39         4056 :             current_time: AtomicU64::new(0),
      40         4056 :             queue: Mutex::new(BinaryHeap::new()),
      41         4056 :             nonce: AtomicU32::new(0),
      42         4056 :             fake_context: Arc::new(ThreadContext::new()),
      43         4056 :         }
      44         4056 :     }
      45              : 
      46              :     /// Return the current world's time.
      47     16448683 :     pub fn now(&self) -> u64 {
      48     16448683 :         self.current_time.load(std::sync::atomic::Ordering::SeqCst)
      49     16448683 :     }
      50              : 
      51              :     /// Tick-tock the global clock. Return the event ready to be processed
      52              :     /// or move the clock forward and then return the event.
      53      1701382 :     pub(crate) fn step(&self) -> Option<Arc<ThreadContext>> {
      54      1701382 :         let mut queue = self.queue.lock();
      55      1701382 : 
      56      1701382 :         if queue.is_empty() {
      57              :             // no future events
      58         4096 :             return None;
      59      1697286 :         }
      60      1697286 : 
      61      1697286 :         if !self.is_event_ready(queue.deref_mut()) {
      62      1260458 :             let next_time = queue.peek().unwrap().time;
      63      1260458 :             self.current_time
      64      1260458 :                 .store(next_time, std::sync::atomic::Ordering::SeqCst);
      65      1260458 :             trace!("rewind time to {}", next_time);
      66      1260458 :             assert!(self.is_event_ready(queue.deref_mut()));
      67       436828 :         }
      68              : 
      69      1697286 :         Some(queue.pop().unwrap().wake_context)
      70      1701382 :     }
      71              : 
      72              :     /// Append an event to the queue, to wakeup the thread in `ms` milliseconds.
      73      1596909 :     pub(crate) fn schedule_wakeup(&self, ms: u64, wake_context: Arc<ThreadContext>) {
      74      1596909 :         self.nonce.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
      75      1596909 :         let nonce = self.nonce.load(std::sync::atomic::Ordering::SeqCst);
      76      1596909 :         self.queue.lock().push(Pending {
      77      1596909 :             time: self.now() + ms,
      78      1596909 :             nonce,
      79      1596909 :             wake_context,
      80      1596909 :         })
      81      1596909 :     }
      82              : 
      83              :     /// Append a fake event to the queue, to prevent clocks from skipping this time.
      84       201905 :     pub fn schedule_fake(&self, ms: u64) {
      85       201905 :         self.queue.lock().push(Pending {
      86       201905 :             time: self.now() + ms,
      87       201905 :             nonce: 0,
      88       201905 :             wake_context: self.fake_context.clone(),
      89       201905 :         });
      90       201905 :     }
      91              : 
      92              :     /// Return true if there is a ready event.
      93      2957744 :     fn is_event_ready(&self, queue: &mut BinaryHeap<Pending>) -> bool {
      94      2957744 :         queue.peek().map_or(false, |x| x.time <= self.now())
      95      2957744 :     }
      96              : 
      97              :     /// Clear all pending events.
      98         4006 :     pub(crate) fn clear(&self) {
      99         4006 :         self.queue.lock().clear();
     100         4006 :     }
     101              : }
     102              : 
     103              : struct Pending {
     104              :     time: u64,
     105              :     nonce: u32,
     106              :     wake_context: Arc<ThreadContext>,
     107              : }
     108              : 
     109              : // BinaryHeap is a max-heap, and we want a min-heap. Reverse the ordering here
     110              : // to get that.
     111              : impl PartialOrd for Pending {
     112     13788457 :     fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
     113     13788457 :         Some(self.cmp(other))
     114     13788457 :     }
     115              : }
     116              : 
     117              : impl Ord for Pending {
     118     13788457 :     fn cmp(&self, other: &Self) -> Ordering {
     119     13788457 :         (other.time, other.nonce).cmp(&(self.time, self.nonce))
     120     13788457 :     }
     121              : }
     122              : 
     123              : impl PartialEq for Pending {
     124            0 :     fn eq(&self, other: &Self) -> bool {
     125            0 :         (other.time, other.nonce) == (self.time, self.nonce)
     126            0 :     }
     127              : }
     128              : 
     129              : impl Eq for Pending {}