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}