gwr_engine/time/
clock.rs

1// Copyright (c) 2023 Graphcore Ltd. All rights reserved.
2
3//! This module represents the time during a simulation.
4//!
5//! Time is made up of a tick count and a phase.
6
7use core::cmp::Ordering;
8use std::cell::{Cell, RefCell};
9use std::future::Future;
10use std::pin::Pin;
11use std::rc::Rc;
12use std::task::{Context, Poll, Waker};
13
14use futures::future::FusedFuture;
15
16use crate::traits::{Resolve, Resolver};
17use crate::types::{SimError, SimResult};
18
19pub mod phase {
20    pub const BEGIN: u32 = 0;
21    pub const END: u32 = u32::MAX;
22}
23
24/// ClockTick structure for representing a number of Clock ticks and a phase.
25#[derive(Clone, Copy, Debug, PartialEq, Eq)]
26pub struct ClockTick {
27    /// Clock ticks.
28    tick: u64,
29
30    /// Clock phase.
31    phase: u32,
32}
33
34impl ClockTick {
35    #[must_use]
36    pub fn new() -> Self {
37        Self {
38            tick: 0,
39            phase: phase::BEGIN,
40        }
41    }
42
43    /// Get the current clock tick.
44    #[must_use]
45    pub fn tick(&self) -> u64 {
46        self.tick
47    }
48
49    /// Get the current clock phase.
50    #[must_use]
51    pub fn phase(&self) -> u32 {
52        self.phase
53    }
54
55    /// Change the default constructor value of `tick`.
56    pub fn set_tick(&mut self, tick: u64) -> ClockTick {
57        self.tick = tick;
58        *self
59    }
60
61    /// Change the default constructor value of `phase`.
62    pub fn set_phase(&mut self, phase: u32) -> ClockTick {
63        self.phase = phase;
64        *self
65    }
66}
67
68impl Default for ClockTick {
69    fn default() -> Self {
70        Self::new()
71    }
72}
73
74/// Define the comparison operation for SimTime.
75impl Ord for ClockTick {
76    fn cmp(&self, other: &Self) -> Ordering {
77        match self.tick.cmp(&other.tick) {
78            Ordering::Greater => Ordering::Greater,
79            Ordering::Less => Ordering::Less,
80            Ordering::Equal => self.phase.cmp(&other.phase),
81        }
82    }
83}
84
85impl PartialOrd for ClockTick {
86    fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
87        Some(self.cmp(other))
88    }
89}
90
91impl std::fmt::Display for ClockTick {
92    fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
93        write!(f, "{}.{:?}", self.tick, self.phase)
94    }
95}
96
97/// State representing a clock.
98#[derive(Clone)]
99pub struct Clock {
100    /// Frequency of the clock in MHz.
101    /// *Note*: Should never be changed as it is registered at this frequency.
102    freq_mhz: f64,
103
104    pub shared_state: Rc<ClockState>,
105}
106
107pub struct TaskWaker {
108    /// Internal identifier for a scheduled clock wait.
109    pub id: u64,
110
111    /// The Waker to use to make a task active again.
112    pub waker: Waker,
113
114    /// When a task is scheduled in the future it may be a background task
115    /// that will simply run forever in which case it will set `can_exit` to
116    /// true.
117    pub can_exit: bool,
118}
119
120/// Shared state between futures using a Clock and the Clock itself.
121pub struct ClockState {
122    now: RefCell<ClockTick>,
123
124    next_waiter_id: Cell<u64>,
125
126    /// Queue of futures waiting for the right time.
127    pub waiting: RefCell<Vec<Vec<TaskWaker>>>,
128
129    /// Queue of times at which those futures are to be woken. This is kept
130    /// sorted by time so that the first entry is the next to be woken.
131    pub waiting_times: RefCell<Vec<ClockTick>>,
132
133    /// Registered [`Resolve`] functions.
134    pub to_resolve: RefCell<Vec<Rc<dyn Resolve + 'static>>>,
135}
136
137impl ClockState {
138    fn schedule(&self, schedule_time: ClockTick, cx: &mut Context<'_>, can_exit: bool) -> u64 {
139        let waiter_id = self.next_waiter_id.get();
140        self.next_waiter_id.set(waiter_id + 1);
141
142        let mut waiting_times = self.waiting_times.borrow_mut();
143        let mut waiting = self.waiting.borrow_mut();
144        if let Some(index) = waiting_times.iter().position(|&x| x == schedule_time) {
145            // Time already exists, add this task
146            waiting[index].push(TaskWaker {
147                id: waiter_id,
148                waker: cx.waker().clone(),
149                can_exit,
150            });
151        } else {
152            // Time not found, insert at the correct location
153            match waiting_times.iter().position(|x| *x < schedule_time) {
154                Some(index) => {
155                    // Insert at an arbitrary index
156                    waiting_times.insert(index, schedule_time);
157                    waiting.insert(
158                        index,
159                        vec![TaskWaker {
160                            id: waiter_id,
161                            waker: cx.waker().clone(),
162                            can_exit,
163                        }],
164                    );
165                }
166                None => {
167                    // Insert at the head
168                    waiting_times.push(schedule_time);
169                    waiting.push(vec![TaskWaker {
170                        id: waiter_id,
171                        waker: cx.waker().clone(),
172                        can_exit,
173                    }]);
174                }
175            }
176        }
177
178        waiter_id
179    }
180
181    fn unschedule(&self, schedule_time: ClockTick, waiter_id: u64) {
182        let mut waiting_times = self.waiting_times.borrow_mut();
183        let mut waiting = self.waiting.borrow_mut();
184
185        if let Some(time_index) = waiting_times.iter().position(|&x| x == schedule_time)
186            && let Some(waiter_index) = waiting[time_index].iter().position(|w| w.id == waiter_id)
187        {
188            waiting[time_index].remove(waiter_index);
189            if waiting[time_index].is_empty() {
190                waiting.remove(time_index);
191                waiting_times.remove(time_index);
192            }
193        }
194    }
195
196    fn advance_time(&self, to_time: ClockTick) {
197        self.resolve();
198
199        assert!(to_time >= *self.now.borrow(), "Time moving backwards");
200        *self.now.borrow_mut() = to_time;
201    }
202
203    fn resolve(&self) {
204        for r in self.to_resolve.borrow_mut().drain(..) {
205            r.resolve();
206        }
207    }
208}
209
210impl Clock {
211    /// Create a new [Clock] at the specified frequency.
212    #[must_use]
213    pub fn new(freq_mhz: f64) -> Self {
214        let shared_state = Rc::new(ClockState {
215            now: RefCell::new(ClockTick {
216                tick: 0,
217                phase: phase::BEGIN,
218            }),
219            next_waiter_id: Cell::new(0),
220            waiting: RefCell::new(Vec::new()),
221            waiting_times: RefCell::new(Vec::new()),
222            to_resolve: RefCell::new(Vec::new()),
223        });
224
225        Self {
226            freq_mhz,
227            shared_state,
228        }
229    }
230
231    /// Advance the time on this clock
232    pub fn advance_time(&self, to_time: ClockTick) {
233        self.shared_state.advance_time(to_time);
234    }
235
236    /// Returns the clocks frequency in MHz.
237    #[must_use]
238    pub fn freq_mhz(&self) -> f64 {
239        self.freq_mhz
240    }
241
242    /// Returns the current [ClockTick].
243    #[must_use]
244    pub fn tick_now(&self) -> ClockTick {
245        *self.shared_state.now.borrow()
246    }
247
248    /// Returns the current time in `ns`.
249    #[must_use]
250    pub fn time_now_ns(&self) -> f64 {
251        let now = *self.shared_state.now.borrow();
252        self.to_ns(&now)
253    }
254
255    /// Returns the time in `ns` of the next event registered with this clock.
256    #[must_use]
257    pub fn time_of_next(&self) -> f64 {
258        match self.shared_state.waiting_times.borrow().last() {
259            Some(clock_time) => self.to_ns(clock_time),
260            None => f64::MAX,
261        }
262    }
263
264    /// Returns the phase of the next event registered with this clock.
265    #[must_use]
266    pub fn phase_of_next(&self) -> u32 {
267        match self.shared_state.waiting_times.borrow().last() {
268            Some(clock_time) => clock_time.phase(),
269            None => u32::MAX,
270        }
271    }
272
273    /// Convert the given [ClockTick] to a time in `ns` for this clock.
274    #[must_use]
275    pub fn to_ns(&self, clock_time: &ClockTick) -> f64 {
276        clock_time.tick as f64 / self.freq_mhz * 1000.0
277    }
278
279    /// Returns a [ClockDelay] future which must be `await`ed to delay the
280    /// specified number of ticks, rejecting a deadline that overflows.
281    #[must_use = "Futures do nothing unless you `.await` or otherwise use them"]
282    pub fn wait_ticks(&self, ticks: u64) -> Result<ClockDelay, SimError> {
283        let mut until = self.tick_now();
284        until.tick = until
285            .tick
286            .checked_add(ticks)
287            .ok_or_else(|| SimError("clock wait deadline overflows".into()))?;
288        until.phase = phase::BEGIN;
289        Ok(ClockDelay {
290            shared_state: self.shared_state.clone(),
291            until,
292            can_exit: false,
293            waiter_id: None,
294            done: false,
295        })
296    }
297
298    /// Returns a [ClockDelay] future which must be `await`ed to delay the
299    /// specified number of ticks. However, if the remainder of the simulation
300    /// completes then this future is allowed to not complete. This allows the
301    /// user to create tasks that can run continuously as long as the rest of
302    /// the simulation continues to run. Returns an error if the deadline
303    /// overflows.
304    #[must_use = "Futures do nothing unless you `.await` or otherwise use them"]
305    pub fn wait_ticks_or_exit(&self, ticks: u64) -> Result<ClockDelay, SimError> {
306        let mut delay = self.wait_ticks(ticks)?;
307        delay.can_exit = true;
308        Ok(delay)
309    }
310
311    #[must_use = "Futures do nothing unless you `.await` or otherwise use them"]
312    pub fn next_tick_and_phase(&self, phase: u32) -> Result<ClockDelay, SimError> {
313        let mut delay = self.wait_ticks(1)?;
314        delay.until.phase = phase;
315        Ok(delay)
316    }
317
318    #[must_use = "Futures do nothing unless you `.await` or otherwise use them"]
319    pub fn wait_phase(&self, phase: u32) -> ClockDelay {
320        let mut until = self.tick_now();
321        assert!(phase >= until.phase, "Time going backwards");
322        until.phase = phase;
323        ClockDelay {
324            shared_state: self.shared_state.clone(),
325            until,
326            can_exit: false,
327            waiter_id: None,
328            done: false,
329        }
330    }
331
332    /// Advance to the next tick after the specified time, rejecting
333    /// unrepresentable durations and tick overflow.
334    pub fn advance_to(&self, time_ns: f64) -> SimResult {
335        let now_ns = self.time_now_ns();
336        assert!(now_ns < time_ns);
337        let diff_ns = time_ns - now_ns;
338        let ticks = (diff_ns * (self.freq_mhz / 1000.0)).ceil();
339        // u64::MAX rounds up to 2^64 as f64, so the upper bound is exclusive.
340        if !ticks.is_finite() || !(0.0..u64::MAX as f64).contains(&ticks) {
341            return Err(SimError(
342                "clock advance duration is outside u64 range".into(),
343            ));
344        }
345
346        let mut until = self.tick_now();
347        until.tick = until
348            .tick
349            .checked_add(ticks as u64)
350            .ok_or_else(|| SimError("clock advance deadline overflows".into()))?;
351        until.phase = phase::BEGIN;
352
353        self.shared_state.advance_time(until);
354        Ok(())
355    }
356}
357
358/// The default clocks is simply to use a 1GHz clock so ticks are 1ns.
359impl Default for Clock {
360    fn default() -> Self {
361        Self::new(1000.0)
362    }
363}
364
365/// The comparison operators for Clocks - use the next pending Waker time.
366impl PartialEq for Clock {
367    fn eq(&self, other: &Self) -> bool {
368        self.time_of_next() == other.time_of_next() && self.phase_of_next() == other.phase_of_next()
369    }
370}
371impl Eq for Clock {}
372
373impl Ord for Clock {
374    fn cmp(&self, other: &Self) -> Ordering {
375        self.time_of_next()
376            .total_cmp(&other.time_of_next())
377            .then_with(|| self.phase_of_next().cmp(&other.phase_of_next()))
378    }
379}
380
381impl PartialOrd for Clock {
382    fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
383        Some(self.cmp(other))
384    }
385}
386
387impl Resolver for Clock {
388    fn add_resolve(&self, resolve: Rc<dyn Resolve + 'static>) {
389        self.shared_state.to_resolve.borrow_mut().push(resolve);
390    }
391}
392
393/// Future returned by the clock to manage advancing time using async functions.
394pub struct ClockDelay {
395    shared_state: Rc<ClockState>,
396    until: ClockTick,
397    can_exit: bool,
398    waiter_id: Option<u64>,
399    done: bool,
400}
401
402impl Future for ClockDelay {
403    type Output = ();
404    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
405        if self.done {
406            return Poll::Ready(());
407        }
408
409        if self.until > *self.shared_state.now.borrow() {
410            if let Some(waiter_id) = self.waiter_id {
411                self.shared_state.unschedule(self.until, waiter_id);
412            }
413            let waiter_id = self.shared_state.schedule(self.until, cx, self.can_exit);
414            self.waiter_id = Some(waiter_id);
415            Poll::Pending
416        } else {
417            self.waiter_id = None;
418            self.done = true;
419            Poll::Ready(())
420        }
421    }
422}
423
424impl FusedFuture for ClockDelay {
425    fn is_terminated(&self) -> bool {
426        self.done
427    }
428}
429
430impl Drop for ClockDelay {
431    fn drop(&mut self) {
432        if let Some(waiter_id) = self.waiter_id.take() {
433            self.shared_state.unschedule(self.until, waiter_id);
434        }
435    }
436}
437
438#[cfg(test)]
439mod tests {
440    use std::mem::drop;
441    use std::pin::Pin;
442    use std::task::Context;
443
444    use futures::future::FusedFuture;
445    use futures::task::noop_waker;
446
447    use super::*;
448
449    #[test]
450    fn wait_ticks_or_exit_rejects_overflow_and_accepts_maximum_deadline() {
451        let clock = Clock::default();
452        clock.advance_time(ClockTick::new().set_tick(u64::MAX - 1));
453        assert!(clock.wait_ticks_or_exit(2).is_err());
454        let delay = clock.wait_ticks_or_exit(1).unwrap();
455        assert_eq!(delay.until.tick(), u64::MAX);
456        assert_eq!(delay.until.phase(), phase::BEGIN);
457        assert!(delay.can_exit);
458        assert!(clock.shared_state.waiting_times.borrow().is_empty());
459    }
460
461    #[test]
462    fn next_tick_and_phase_rejects_overflow_and_preserves_phase() {
463        let clock = Clock::default();
464        clock.advance_time(ClockTick::new().set_tick(u64::MAX - 1));
465        let delay = clock.next_tick_and_phase(3).unwrap();
466        assert_eq!(delay.until.tick(), u64::MAX);
467        assert_eq!(delay.until.phase(), 3);
468        clock.advance_time(ClockTick::new().set_tick(u64::MAX));
469        assert!(clock.next_tick_and_phase(3).is_err());
470        assert!(clock.shared_state.waiting_times.borrow().is_empty());
471    }
472
473    #[test]
474    fn advance_to_rejects_tick_overflow_without_advancing_time() {
475        let clock = Clock::default();
476        // The duration is a representable 2048 ticks, but its deadline is 2^64.
477        let start_tick = u64::MAX - 2047;
478        clock.advance_time(ClockTick::new().set_tick(start_tick));
479        assert!(clock.advance_to(u64::MAX as f64).is_err());
480        assert_eq!(clock.tick_now().tick(), start_tick);
481    }
482
483    #[test]
484    fn advance_to_rejects_duration_at_u64_upper_bound() {
485        let clock = Clock::default();
486        assert!(clock.advance_to(u64::MAX as f64).is_err());
487        assert_eq!(clock.tick_now(), ClockTick::default());
488    }
489
490    #[test]
491    fn advance_to_rejects_infinite_duration() {
492        let clock = Clock::default();
493        assert!(clock.advance_to(f64::INFINITY).is_err());
494        assert_eq!(clock.tick_now(), ClockTick::default());
495    }
496
497    #[test]
498    fn advance_to_accepts_largest_f64_duration_below_u64_upper_bound() {
499        let clock = Clock::default();
500        // The largest f64 below 2^64 is 2^64 - 2048.
501        let ticks = 18_446_744_073_709_549_568_u64;
502        clock.advance_to(ticks as f64).unwrap();
503        assert_eq!(clock.tick_now().tick(), ticks);
504        assert_eq!(clock.tick_now().phase(), phase::BEGIN);
505    }
506
507    #[test]
508    fn wait_ticks_rejects_deadline_overflow() {
509        let clock = Clock::default();
510        clock.advance_time(ClockTick::new().set_tick(u64::MAX - 1));
511        assert!(clock.wait_ticks(2).is_err());
512        assert!(clock.shared_state.waiting_times.borrow().is_empty());
513        assert!(clock.shared_state.waiting.borrow().is_empty());
514    }
515
516    #[test]
517    fn wait_ticks_accepts_maximum_deadline_and_zero_wait() {
518        let clock = Clock::default();
519        clock.advance_time(ClockTick::new().set_tick(u64::MAX - 1));
520        let delay = clock.wait_ticks(1).unwrap();
521        assert_eq!(delay.until.tick(), u64::MAX);
522        clock.advance_time(ClockTick::new().set_tick(u64::MAX));
523        let delay = clock.wait_ticks(0).unwrap();
524        assert_eq!(delay.until.tick(), u64::MAX);
525    }
526
527    #[test]
528    fn convert_to_ns() {
529        let clk_ghz = Clock::new(1000.0);
530        assert_eq!(1.0, clk_ghz.to_ns(&ClockTick::new().set_tick(1)));
531
532        let slow_clk = Clock::new(0.5);
533        assert_eq!(2000.0, slow_clk.to_ns(&ClockTick::new().set_tick(1)));
534    }
535
536    #[test]
537    fn clock_tick_accessors_default_display_and_ordering() {
538        let default_tick = ClockTick::default();
539        assert_eq!(default_tick.tick(), 0);
540        assert_eq!(default_tick.phase(), phase::BEGIN);
541        assert_eq!(default_tick.to_string(), "0.0");
542
543        let earlier = ClockTick::new().set_tick(1);
544        let later = ClockTick::new().set_tick(2);
545        assert!(earlier < later);
546        assert_eq!(earlier.partial_cmp(&later), Some(Ordering::Less));
547    }
548
549    #[test]
550    fn clock_tick_phase_accessors_display_and_ordering() {
551        let tick = ClockTick::new().set_tick(1).set_phase(2);
552        assert_eq!(tick.phase(), 2);
553        assert_eq!(tick.to_string(), "1.2");
554
555        let earlier_phase = ClockTick::new().set_tick(1).set_phase(1);
556        assert!(earlier_phase < tick);
557
558        let later_tick = ClockTick::new().set_tick(2).set_phase(0);
559        assert!(later_tick > tick);
560    }
561
562    #[test]
563    fn clock_default_advance_and_ordering() {
564        let clock = Clock::default();
565        assert_eq!(clock.freq_mhz(), 1000.0);
566
567        clock.advance_to(2.1).unwrap();
568        assert_eq!(clock.tick_now().tick(), 3);
569        assert_eq!(clock.tick_now().phase(), phase::BEGIN);
570        assert_eq!(clock.time_now_ns(), 3.0);
571
572        let earlier = Clock::new(1000.0);
573        let later = Clock::new(1000.0);
574        drop(earlier.wait_ticks(1).unwrap());
575        drop(later.wait_ticks(2).unwrap());
576
577        assert!(earlier == later);
578        assert_eq!(earlier.partial_cmp(&later), Some(Ordering::Equal));
579    }
580
581    #[test]
582    fn unschedule_unknown_waiter_is_a_noop() {
583        let clock = Clock::new(1000.0);
584        let scheduled_time = ClockTick::new().set_tick(1);
585
586        clock
587            .shared_state
588            .waiting_times
589            .borrow_mut()
590            .push(scheduled_time);
591        clock.shared_state.waiting.borrow_mut().push(Vec::new());
592
593        clock.shared_state.unschedule(scheduled_time, 7);
594
595        assert_eq!(clock.shared_state.waiting_times.borrow().len(), 1);
596        assert_eq!(clock.shared_state.waiting.borrow().len(), 1);
597    }
598
599    #[test]
600    fn unschedule_waiter_keeps_time_when_other_waiters_remain() {
601        let clock = Clock::new(1000.0);
602        let scheduled_time = ClockTick::new().set_tick(1);
603        let waker = noop_waker();
604        let mut cx = Context::from_waker(&waker);
605
606        let first_waiter_id = clock.shared_state.schedule(scheduled_time, &mut cx, false);
607        let second_waiter_id = clock.shared_state.schedule(scheduled_time, &mut cx, false);
608
609        clock
610            .shared_state
611            .unschedule(scheduled_time, first_waiter_id);
612
613        let waiting_times = clock.shared_state.waiting_times.borrow();
614        let waiting = clock.shared_state.waiting.borrow();
615
616        assert_eq!(waiting_times.as_slice(), &[scheduled_time]);
617        assert_eq!(waiting.len(), 1);
618        assert_eq!(waiting[0].len(), 1);
619        assert_eq!(waiting[0][0].id, second_waiter_id);
620    }
621
622    #[test]
623    fn clock_delay_is_fused_after_completion() {
624        let clock = Clock::new(1000.0);
625        let waker = noop_waker();
626        let mut cx = Context::from_waker(&waker);
627        let mut delay = clock.wait_ticks(1).unwrap();
628
629        assert!(!delay.is_terminated());
630        assert_eq!(Pin::new(&mut delay).poll(&mut cx), Poll::Pending);
631        assert!(!delay.is_terminated());
632
633        clock.advance_time(ClockTick::new().set_tick(1));
634        assert_eq!(Pin::new(&mut delay).poll(&mut cx), Poll::Ready(()));
635        assert!(delay.is_terminated());
636        assert_eq!(Pin::new(&mut delay).poll(&mut cx), Poll::Ready(()));
637        assert!(delay.is_terminated());
638    }
639
640    #[test]
641    fn phase_wait_helpers_schedule_phase_delays() {
642        let clock = Clock::new(1000.0);
643
644        let next_tick = clock.next_tick_and_phase(3).unwrap();
645        assert_eq!(next_tick.until.tick(), 1);
646        assert_eq!(next_tick.until.phase(), 3);
647
648        let same_tick = clock.wait_phase(1);
649        assert_eq!(same_tick.until.tick(), 0);
650        assert_eq!(same_tick.until.phase(), 1);
651    }
652
653    #[test]
654    fn tick_waits_resume_in_end_phase() {
655        let clock = Clock::new(1000.0);
656
657        let begin = clock.wait_phase(phase::END);
658        assert_eq!(begin.until.tick(), 0);
659        assert_eq!(begin.until.phase(), phase::END);
660
661        clock.advance_time(ClockTick::new().set_phase(phase::END));
662
663        let delay = clock.wait_ticks(1).unwrap();
664        assert_eq!(delay.until.tick(), 1);
665        assert_eq!(delay.until.phase(), phase::BEGIN);
666    }
667}
gwr_engine/time/clock.rs

gwr_engine/time/
clock.rs
