tick 0.3.4 

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148

// Copyright (c) Microsoft Corporation.
// Licensed under the MIT License.

use std::sync::Mutex;
use std::time::Instant;

use thread_aware::affinity::Affinity;
use thread_aware::{PerCore, ThreadAware};

use crate::timers::Timers;

#[derive(Debug, Clone)]
pub(crate) enum ClockState {
    System(SynchronizedTimers),
    #[cfg(any(feature = "test-util", test))]
    ClockControl(crate::ClockControl),
}

impl ThreadAware for ClockState {
    fn relocate(&mut self, source: Option<Affinity>, destination: Affinity) {
        match self {
            Self::System(synchronized_timers) => synchronized_timers.relocate(source, destination),
            #[cfg(any(feature = "test-util", test))]
            Self::ClockControl(clock_control) => clock_control.relocate(source, destination),
        }
    }
}

impl ClockState {
    pub(crate) fn new_system() -> Self {
        Self::System(SynchronizedTimers::new_isolated())
    }

    /// Creates a `System` clock state backed by a single shared timer set. Used by clocks driven
    /// by a single global driver task (e.g. the Tokio-driven clock created by
    /// [`Clock::new_tokio`][crate::Clock::new_tokio]). [`ThreadAware::relocate`] is a no-op for
    /// this variant.
    #[cfg(any(feature = "rt-shared", test))]
    pub(crate) fn new_system_shared() -> Self {
        Self::System(SynchronizedTimers::new_shared())
    }
}

impl ClockState {
    pub(crate) fn timers_len(&self) -> usize {
        match self {
            #[cfg(any(feature = "test-util", test))]
            Self::ClockControl(control) => control.timers_len(),
            Self::System(timers) => timers.with_timers(|t| t.len()),
        }
    }

    pub(crate) fn alive(&self) -> bool {
        match self {
            #[cfg(any(feature = "test-util", test))]
            Self::ClockControl(_) => true,
            Self::System(timers) => timers.with_timers(|t| t.alive()),
        }
    }

    #[cfg_attr(test, mutants::skip)] // causes test timeout
    pub(crate) fn is_unique(&self) -> bool {
        match self {
            Self::System(timers) => timers.is_unique(),
            #[cfg(any(feature = "test-util", test))]
            Self::ClockControl(control) => control.is_unique(),
        }
    }
}

// The mutex here is not accessed on a hot path. Timers are accessed only when:
//
// 1. A new timer is registered.
// 2. A timer is unregistered.
// 3. Timers are evaluated. Timer evaluation is very fast when there are no timers to fire. If
//    there are timers to fire, the time to evaluate them is proportional to the number of timers
//    that are ready to fire, and taking the lock is not the bottleneck.
#[derive(Debug, Clone)]
pub(crate) enum SynchronizedTimers {
    /// A single shared timer set. [`ThreadAware::relocate`] is a no-op, so all clones observe
    /// the same timers regardless of thread affinity. Used by clocks driven by a single global
    /// driver task (e.g. the Tokio-driven clock created by [`Clock::new_tokio`][crate::Clock::new_tokio]).
    #[cfg(any(feature = "rt-shared", test))]
    Shared(std::sync::Arc<Mutex<Timers>>),

    /// Per-core isolated timer storage. [`ThreadAware::relocate`] creates a fresh timer set on
    /// the destination core, enabling thread-per-core runtimes to operate on independent timers
    /// with no cross-thread lock contention.
    Isolated(thread_aware::Arc<Mutex<Timers>, PerCore>),
}

impl ThreadAware for SynchronizedTimers {
    fn relocate(&mut self, source: Option<Affinity>, destination: Affinity) {
        match self {
            #[cfg(any(feature = "rt-shared", test))]
            Self::Shared(_) => {}
            Self::Isolated(timers) => timers.relocate(source, destination),
        }
    }
}

impl SynchronizedTimers {
    pub(crate) fn new_isolated() -> Self {
        Self::Isolated(thread_aware::Arc::new(|| Mutex::new(Timers::default())))
    }

    #[cfg(any(feature = "rt-shared", test))]
    pub(crate) fn new_shared() -> Self {
        Self::Shared(std::sync::Arc::new(Mutex::new(Timers::default())))
    }

    pub(super) fn with_timers<F, R>(&self, f: F) -> R
    where
        F: FnOnce(&mut Timers) -> R,
    {
        let mut timers = match self {
            #[cfg(any(feature = "rt-shared", test))]
            Self::Shared(timers) => timers.lock().expect("timers lock poisoned"),
            Self::Isolated(timers) => timers.lock().expect("timers lock poisoned"),
        };
        f(&mut timers)
    }

    #[cfg_attr(test, mutants::skip)] // Causes test timeout.
    pub(crate) fn try_advance_timers(&self, now: Instant) -> Option<Instant> {
        self.with_timers(|timers| timers.advance_timers(now))
    }

    #[cfg_attr(test, mutants::skip)] // causes test timeout
    pub(crate) fn is_unique(&self) -> bool {
        match self {
            #[cfg(any(feature = "rt-shared", test))]
            Self::Shared(timers) => std::sync::Arc::strong_count(timers) == 1,
            Self::Isolated(timers) => thread_aware::Arc::strong_count(timers) == 1,
        }
    }
}

#[cfg_attr(coverage_nightly, coverage(off))]
#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn clock_state_send_and_sync() {
        static_assertions::assert_impl_all!(ClockState: Send, Sync);
    }
}