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         2028 :     pub fn new() -> Timing {
      38         2028 :         Timing {
      39         2028 :             current_time: AtomicU64::new(0),
      40         2028 :             queue: Mutex::new(BinaryHeap::new()),
      41         2028 :             nonce: AtomicU32::new(0),
      42         2028 :             fake_context: Arc::new(ThreadContext::new()),
      43         2028 :         }
      44         2028 :     }
      45              : 
      46              :     /// Return the current world's time.
      47      8229458 :     pub fn now(&self) -> u64 {
      48      8229458 :         self.current_time.load(std::sync::atomic::Ordering::SeqCst)
      49      8229458 :     }
      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       854004 :     pub(crate) fn step(&self) -> Option<Arc<ThreadContext>> {
      54       854004 :         let mut queue = self.queue.lock();
      55       854004 : 
      56       854004 :         if queue.is_empty() {
      57              :             // no future events
      58         2048 :             return None;
      59       851956 :         }
      60       851956 : 
      61       851956 :         if !self.is_event_ready(queue.deref_mut()) {
      62       628456 :             let next_time = queue.peek().unwrap().time;
      63       628456 :             self.current_time
      64       628456 :                 .store(next_time, std::sync::atomic::Ordering::SeqCst);
      65       628456 :             trace!("rewind time to {}", next_time);
      66       628456 :             assert!(self.is_event_ready(queue.deref_mut()));
      67       223500 :         }
      68              : 
      69       851956 :         Some(queue.pop().unwrap().wake_context)
      70       854004 :     }
      71              : 
      72              :     /// Append an event to the queue, to wakeup the thread in `ms` milliseconds.
      73       803177 :     pub(crate) fn schedule_wakeup(&self, ms: u64, wake_context: Arc<ThreadContext>) {
      74       803177 :         self.nonce.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
      75       803177 :         let nonce = self.nonce.load(std::sync::atomic::Ordering::SeqCst);
      76       803177 :         self.queue.lock().push(Pending {
      77       803177 :             time: self.now() + ms,
      78       803177 :             nonce,
      79       803177 :             wake_context,
      80       803177 :         })
      81       803177 :     }
      82              : 
      83              :     /// Append a fake event to the queue, to prevent clocks from skipping this time.
      84        98870 :     pub fn schedule_fake(&self, ms: u64) {
      85        98870 :         self.queue.lock().push(Pending {
      86        98870 :             time: self.now() + ms,
      87        98870 :             nonce: 0,
      88        98870 :             wake_context: self.fake_context.clone(),
      89        98870 :         });
      90        98870 :     }
      91              : 
      92              :     /// Return true if there is a ready event.
      93      1480412 :     fn is_event_ready(&self, queue: &mut BinaryHeap<Pending>) -> bool {
      94      1480412 :         queue.peek().map_or(false, |x| x.time <= self.now())
      95      1480412 :     }
      96              : 
      97              :     /// Clear all pending events.
      98         2003 :     pub(crate) fn clear(&self) {
      99         2003 :         self.queue.lock().clear();
     100         2003 :     }
     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      6960266 :     fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
     113      6960266 :         Some(self.cmp(other))
     114      6960266 :     }
     115              : }
     116              : 
     117              : impl Ord for Pending {
     118      6960266 :     fn cmp(&self, other: &Self) -> Ordering {
     119      6960266 :         (other.time, other.nonce).cmp(&(self.time, self.nonce))
     120      6960266 :     }
     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 {}