nuclei/sync/
event.rs

1//! A one-shot event: set once, and every wait completes from then on.
2
3#![allow(unsafe_code)]
4
5use core::fmt;
6use core::future::Future;
7use core::pin::Pin;
8use core::task::{Context, Poll};
9
10use super::internal::{Ordering, UnsafeCell};
11use super::list::List;
12use super::queue::{Arrivals, FIRST_BIT, Policy, QUEUED, Step, WaitQueue};
13use super::waiter::{Cancel, Wait, WakeList};
14
15const SET: usize = 1 << FIRST_BIT;
16
17/// An event that is set once and stays set.
18///
19/// [`wait`](Self::wait) completes as soon as the event is set, whether
20/// that happened before or after the wait began; [`set`](Self::set) wakes
21/// every waiter. Checking is one atomic load.
22///
23/// ```
24/// use nuclei::sync::Event;
25///
26/// let ready = Event::new();
27/// assert!(!ready.is_set());
28/// ready.set();
29/// assert!(ready.is_set());
30/// ```
31pub struct Event {
32    queue: WaitQueue,
33    waiting: UnsafeCell<List>,
34}
35
36// SAFETY: the waiting list is touched only by the drain role's holder.
37unsafe impl Send for Event {}
38// SAFETY: as above.
39unsafe impl Sync for Event {}
40
41impl Event {
42    loom_const_fn! {
43        /// An event that is not set.
44        pub fn new() -> Self {
45            Self::with_state(0)
46        }
47    }
48
49    loom_const_fn! {
50        /// An event that is already set when `set` is true.
51        pub(super) fn new_set(set: bool) -> Self {
52            Self::with_state(if set { SET } else { 0 })
53        }
54    }
55
56    loom_const_fn! {
57        fn with_state(state: usize) -> Self {
58            Self { queue: WaitQueue::new(state), waiting: UnsafeCell::new(List::new()) }
59        }
60    }
61
62    /// Sets the event and wakes every waiter. Setting it again does
63    /// nothing.
64    pub fn set(&self) {
65        self.transition(|state| {
66            if state & SET != 0 {
67                Step::Keep
68            } else if state & QUEUED != 0 {
69                Step::Drain(state | SET)
70            } else {
71                Step::Set(state | SET)
72            }
73        });
74    }
75
76    /// Whether the event is set.
77    pub fn is_set(&self) -> bool {
78        self.queue.state.load(Ordering::Acquire) & SET != 0
79    }
80
81    /// A future that completes once the event is set.
82    pub fn wait(&self) -> EventWait<'_> {
83        EventWait {
84            event: self,
85            wait: Wait::new(),
86        }
87    }
88}
89
90impl Policy for Event {
91    fn queue(&self) -> &WaitQueue {
92        &self.queue
93    }
94
95    fn pass(&self, arrivals: Arrivals, woken: &mut WakeList<'_>, _wake_front: bool) -> usize {
96        let pool = &self.queue;
97        let set = self.is_set();
98        // SAFETY: only the drain role's holder runs a pass.
99        self.waiting.with_mut(|waiting| unsafe {
100            let waiting = &mut *waiting;
101            waiting.absorb(arrivals, pool);
102            waiting.tidy(&self.queue);
103            while set && let Some(node) = waiting.front(pool) {
104                waiting.pop_front();
105                woken.grant(node);
106            }
107            if waiting.front(pool).is_none() {
108                QUEUED
109            } else {
110                0
111            }
112        })
113    }
114}
115
116impl Drop for Event {
117    fn drop(&mut self) {
118        let pool = &self.queue;
119        // SAFETY: `&mut self` rules out a concurrent pass.
120        self.waiting
121            .with_mut(|waiting| unsafe { (*waiting).clear(pool) });
122    }
123}
124
125impl Default for Event {
126    fn default() -> Self {
127        Self::new()
128    }
129}
130
131impl fmt::Debug for Event {
132    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
133        f.debug_struct("Event")
134            .field("set", &self.is_set())
135            .finish()
136    }
137}
138
139/// The future [`Event::wait`] returns.
140#[must_use = "futures do nothing unless polled"]
141pub struct EventWait<'a> {
142    event: &'a Event,
143    wait: Wait,
144}
145
146impl Future for EventWait<'_> {
147    type Output = ();
148
149    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<()> {
150        let this = self.get_mut();
151        let event = this.event;
152        if this.wait.is_queued() {
153            return this.wait.poll(&event.queue, Some(cx.waker())).map(drop);
154        }
155        if event.is_set() {
156            return Poll::Ready(());
157        }
158        event.enqueue(&mut this.wait, 0, cx.waker());
159        this.wait.poll(&event.queue, None).map(drop)
160    }
161}
162
163impl Drop for EventWait<'_> {
164    fn drop(&mut self) {
165        if let Cancel::Withdrawn(_) = self.wait.cancel(&self.event.queue) {
166            self.event.withdrawn(false);
167        }
168    }
169}
170
171impl fmt::Debug for EventWait<'_> {
172    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
173        f.debug_struct("EventWait").finish_non_exhaustive()
174    }
175}
176
177#[cfg(all(test, not(loom)))]
178mod tests {
179    use super::*;
180    use crate::sync::test_support::{Polled, block_on};
181    use std::sync::Arc;
182
183    #[test]
184    fn setting_wakes_every_waiter_and_later_waits_pass() {
185        let event = Event::new();
186        let mut waiters: Vec<_> = (0..3).map(|_| Polled::new(event.wait())).collect();
187        for waiter in &mut waiters {
188            waiter.pending();
189        }
190        drop(waiters.pop());
191        event.set();
192        event.set();
193        for waiter in &mut waiters {
194            assert_eq!(waiter.wakes(), 1);
195            waiter.ready();
196        }
197        Polled::new(event.wait()).ready();
198    }
199
200    #[test]
201    fn a_set_from_another_thread_is_seen() {
202        let event = Arc::new(Event::new());
203        let setter = Arc::clone(&event);
204        let thread = std::thread::spawn(move || setter.set());
205        block_on(event.wait());
206        thread.join().expect("setter");
207        assert!(event.is_set());
208    }
209}
nuclei/sync/event.rs

nuclei/sync/
event.rs
