starnix_sync/
interruptible_event.rs

1// Copyright 2023 The Fuchsia Authors. All rights reserved.
2// Use of this source code is governed by a BSD-style license that can be
3// found in the LICENSE file.
4
5use std::sync::atomic::Ordering;
6use std::sync::Arc;
7
8/// A blocking object that can either be notified normally or interrupted
9///
10/// To block using an `InterruptibleEvent`, first call `begin_wait`. At this point, the event is
11/// in the "waiting" state, and future calls to `notify` or `interrupt` will terminate the wait.
12///
13/// After `begin_wait` returns, call `block_until` to block the current thread until one of the
14/// following conditions occur:
15///
16///  1. The given deadline expires.
17///  2. At least one of the `notify` or `interrupt` functions were called after `begin_wait`.
18///
19/// It's safe to call `notify` or `interrupt` at any time. However, calls to `begin_wait` and
20/// `block_until` must alternate, starting with `begin_wait`.
21///
22/// `InterruptibleEvent` uses two-phase waiting so that clients can register for notification,
23/// perform some related work, and then start blocking. This approach ensures that clients do not
24/// miss notifications that arrive after they perform the related work but before they actually
25/// start blocking.
26#[derive(Debug)]
27pub struct InterruptibleEvent {
28    futex: zx::Futex,
29}
30
31/// The initial state.
32///
33///  * Transitions to `WAITING` after `begin_wait`.
34const READY: i32 = 0;
35
36/// The event is waiting for a notification or an interruption.
37///
38///  * Transitions to `NOTIFIED` after `notify`.
39///  * Transitions to `INTERRUPTED` after `interrupt`.
40///  * Transitions to `READY` if the deadline for `block_until` expires.
41const WAITING: i32 = 1;
42
43/// The event has been notified and will wake up.
44///
45///  * Transitions to `READY` after `block_until` processes the notification.
46const NOTIFIED: i32 = 2;
47
48/// The event has been interrupted and will wake up.
49///
50///  * Transitions to `READY` after `block_until` processes the interruption.
51const INTERRUPTED: i32 = 3;
52
53/// A guard object to enforce that clients call `begin_wait` before `block_until`.
54#[must_use = "call block_until to advance the event state machine"]
55pub struct EventWaitGuard<'a> {
56    event: &'a Arc<InterruptibleEvent>,
57}
58
59impl<'a> EventWaitGuard<'a> {
60    /// The underlying event associated with this guard.
61    pub fn event(&self) -> &'a Arc<InterruptibleEvent> {
62        self.event
63    }
64
65    /// Block the thread until either `deadline` expires, the event is notified, or the event is
66    /// interrupted.
67    pub fn block_until(
68        self,
69        new_owner: Option<&zx::Thread>,
70        deadline: zx::MonotonicInstant,
71    ) -> Result<(), WakeReason> {
72        self.event.block_until(new_owner, deadline)
73    }
74}
75
76/// A description of why a `block_until` returned without the event being notified.
77#[derive(Debug, PartialEq, Eq)]
78pub enum WakeReason {
79    /// `block_until` returned because another thread interrupted the wait using `interrupt`.
80    Interrupted,
81
82    /// `block_until` returned because the given deadline expired.
83    DeadlineExpired,
84}
85
86impl InterruptibleEvent {
87    pub fn new() -> Arc<Self> {
88        Arc::new(InterruptibleEvent { futex: zx::Futex::new(0) })
89    }
90
91    /// Called to initiate a wait.
92    ///
93    /// Calls to `notify` or `interrupt` after this function returns will cause the event to wake
94    /// up. Calls to those functions prior to calling `begin_wait` will be ignored.
95    ///
96    /// Once called, this function cannot be called again until `block_until` returns. Otherwise,
97    /// this function will panic.
98    pub fn begin_wait<'a>(self: &'a Arc<Self>) -> EventWaitGuard<'a> {
99        self.futex
100            .compare_exchange(READY, WAITING, Ordering::Relaxed, Ordering::Relaxed)
101            .expect("Tried to begin waiting on an event when not ready.");
102        EventWaitGuard { event: self }
103    }
104
105    fn block_until(
106        &self,
107        new_owner: Option<&zx::Thread>,
108        deadline: zx::MonotonicInstant,
109    ) -> Result<(), WakeReason> {
110        // We need to loop around the call to zx_futex_wake because we can receive spurious
111        // wakeups.
112        loop {
113            match self.futex.wait(WAITING, new_owner, deadline) {
114                // The deadline expired while we were sleeping.
115                Err(zx::Status::TIMED_OUT) => {
116                    self.futex.store(READY, Ordering::Relaxed);
117                    return Err(WakeReason::DeadlineExpired);
118                }
119                // The value changed before we went to sleep.
120                Err(zx::Status::BAD_STATE) => (),
121                Err(e) => panic!("Unexpected error from zx_futex_wait: {e}"),
122                Ok(()) => (),
123            }
124
125            let state = self.futex.load(Ordering::Acquire);
126
127            match state {
128                // If we're still in the `WAITING` state, then the wake ended spuriously and we
129                // need to go back to sleep.
130                WAITING => continue,
131                NOTIFIED => {
132                    // We use a store here rather than a compare_exchange because other threads are
133                    // only allowed to write to this value in the `WAITING` state and we are in the
134                    // `NOTIFIED` state.
135                    self.futex.store(READY, Ordering::Relaxed);
136                    return Ok(());
137                }
138                INTERRUPTED => {
139                    // We use a store here rather than a compare_exchange because other threads are
140                    // only allowed to write to this value in the `WAITING` state and we are in the
141                    // `INTERRUPTED` state.
142                    self.futex.store(READY, Ordering::Relaxed);
143                    return Err(WakeReason::Interrupted);
144                }
145                _ => {
146                    panic!("Unexpected event state: {state}");
147                }
148            }
149        }
150    }
151
152    /// Wake up the event normally.
153    ///
154    /// If this function is called before `begin_wait`, this notification is ignored. Calling this
155    /// function repeatedly has no effect. If both `notify` and `interrupt` are called, the state
156    /// observed by `block_until` is a race.
157    pub fn notify(&self) {
158        self.wake(NOTIFIED);
159    }
160
161    /// Wake up the event because of an interruption.
162    ///
163    /// If this function is called before `begin_wait`, this notification is ignored. Calling this
164    /// function repeatedly has no effect. If both `notify` and `interrupt` are called, the state
165    /// observed by `block_until` is a race.
166    pub fn interrupt(&self) {
167        self.wake(INTERRUPTED);
168    }
169
170    fn wake(&self, state: i32) {
171        // See <https://marabos.nl/atomics/hardware.html#failing-compare-exchange> for why we issue
172        // this load before the `compare_exchange` below.
173        let observed = self.futex.load(Ordering::Relaxed);
174        if observed == WAITING
175            && self
176                .futex
177                .compare_exchange(WAITING, state, Ordering::Release, Ordering::Relaxed)
178                .is_ok()
179        {
180            self.futex.wake_all();
181        }
182    }
183}
184
185#[cfg(test)]
186mod test {
187    use super::*;
188    use zx::AsHandleRef;
189
190    #[test]
191    fn test_wait_block_and_notify() {
192        let event = InterruptibleEvent::new();
193
194        let guard = event.begin_wait();
195
196        let other_event = Arc::clone(&event);
197        let thread = std::thread::spawn(move || {
198            other_event.notify();
199        });
200
201        guard.block_until(None, zx::MonotonicInstant::INFINITE).expect("failed to be notified");
202        thread.join().expect("failed to join thread");
203    }
204
205    #[test]
206    fn test_wait_block_and_interrupt() {
207        let event = InterruptibleEvent::new();
208
209        let guard = event.begin_wait();
210
211        let other_event = Arc::clone(&event);
212        let thread = std::thread::spawn(move || {
213            other_event.interrupt();
214        });
215
216        let result = guard.block_until(None, zx::MonotonicInstant::INFINITE);
217        assert_eq!(result, Err(WakeReason::Interrupted));
218        thread.join().expect("failed to join thread");
219    }
220
221    #[test]
222    fn test_wait_block_and_timeout() {
223        let event = InterruptibleEvent::new();
224
225        let guard = event.begin_wait();
226        let result = guard
227            .block_until(None, zx::MonotonicInstant::after(zx::MonotonicDuration::from_millis(20)));
228        assert_eq!(result, Err(WakeReason::DeadlineExpired));
229    }
230
231    #[test]
232    fn futex_ownership_is_transferred() {
233        use zx::HandleBased;
234
235        let event = Arc::new(InterruptibleEvent::new());
236
237        let root_thread_handle =
238            fuchsia_runtime::thread_self().duplicate_handle(zx::Rights::SAME_RIGHTS).unwrap();
239        let root_thread_koid = fuchsia_runtime::thread_self().get_koid().unwrap();
240
241        let event_for_blocked_thread = event.clone();
242
243        let blocked_thread = std::thread::spawn(move || {
244            let event = event_for_blocked_thread;
245            let guard = event.begin_wait();
246            guard.block_until(Some(&root_thread_handle), zx::MonotonicInstant::INFINITE).unwrap();
247        });
248
249        // Wait for the correct owner to appear. If for some reason futex PI breaks, it's likely
250        // that this test will time out rather than panicking outright. It would be nice to have
251        // a clear assertion here, but there's no existing API we can use to atomically wait for a
252        // waiter on the futex *and* see what owner they set. If we find ourselves writing a lot of
253        // tests like this we might consider setting up a fake vdso.
254        while event.futex.get_owner() != Some(root_thread_koid) {
255            std::thread::sleep(std::time::Duration::from_millis(100));
256        }
257
258        event.notify();
259        blocked_thread.join().unwrap();
260    }
261
262    #[test]
263    fn stale_pi_owner_is_noop() {
264        use zx::HandleBased;
265
266        let mut new_owner = None;
267        std::thread::scope(|s| {
268            s.spawn(|| {
269                new_owner = Some(
270                    fuchsia_runtime::thread_self()
271                        .duplicate_handle(zx::Rights::SAME_RIGHTS)
272                        .unwrap(),
273                );
274            });
275        });
276        let new_owner = new_owner.unwrap();
277
278        let event = InterruptibleEvent::new();
279        let guard = event.begin_wait();
280        let result = guard.block_until(
281            Some(&new_owner),
282            zx::MonotonicInstant::after(zx::MonotonicDuration::from_millis(20)),
283        );
284        assert_eq!(result, Err(WakeReason::DeadlineExpired));
285    }
286}