starnix_sync/
object_waiter.rs

1// Copyright 2026 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 smallvec::SmallVec;
6use std::sync::Arc;
7use std::sync::atomic::{AtomicU8, Ordering};
8
9/// A blocking object that waits for signals on Zircon objects, and that can be interrupted.
10///
11/// `InterruptibleEvent` and `PortEvent` wake up a thread when another thread notifies or
12/// interrupts it. A thread that must also wake up when a Zircon object is signaled, such as a
13/// channel it reads requests from, would have to observe the object through a port, or an async
14/// executor. Instead, `ObjectWaiter` lets the thread wait for the objects directly with
15/// `zx_object_wait_many`: the waiter adds an event of its own to the objects the thread waits for,
16/// and `interrupt` signals that event.
17///
18/// To block using an `ObjectWaiter`, first call `begin_wait`. At this point, the waiter is in the
19/// "waiting" state, and future calls to `interrupt` will terminate the wait.
20///
21/// After `begin_wait` returns, call `wait_many` on the returned guard to block the current thread
22/// until one of the following conditions occur:
23///
24///  1. One of the given objects asserts one of the signals that it is waited for.
25///  2. The given deadline expires.
26///  3. `interrupt` was called after `begin_wait`.
27///
28/// It's safe to call `interrupt` at any time. However, only one thread can wait on an
29/// `ObjectWaiter` at a time: `begin_wait` cannot be called again until the guard it returns is
30/// consumed by `wait_many`, or dropped.
31///
32/// `ObjectWaiter` uses two-phase waiting so that clients can register the waiter with the code
33/// that interrupts them, for example in the `RunState` of a task, and then start blocking. This
34/// approach ensures that clients do not miss interruptions that arrive after they register the
35/// waiter but before they actually start blocking, while the interruptions of previous waits,
36/// which arrive before `begin_wait`, are ignored.
37///
38/// The event of the waiter is only signaled when a wait is actually interrupted, and only cleared
39/// after such a wait, so that a wait that is not interrupted costs no more syscall than
40/// `zx_object_wait_many` itself.
41///
42/// As with `InterruptibleEvent` and `PortEvent`, a wait that remains blocked for
43/// `STACK_CLEAN_DELAY` wakes up once to decommit the unused pages of the stack of the thread with
44/// `starnix_stack::clean_stack`, and then blocks again until the deadline.
45#[derive(Debug)]
46pub struct ObjectWaiter {
47    /// The event that `interrupt` signals with `INTERRUPTED_SIGNAL`, and that `wait_many` waits for
48    /// along with the objects of the caller.
49    ///
50    /// The signal is only asserted while `state` is `INTERRUPTING` or `INTERRUPTED`, so that an
51    /// interruption that arrives before `wait_many` blocks still ends the wait. The end of an
52    /// interrupted wait deasserts it.
53    event: zx::Event,
54
55    /// The state of the waiter: `IDLE`, `WAITING`, `INTERRUPTING` or `INTERRUPTED`.
56    state: AtomicU8,
57}
58
59/// No wait is in progress, and the event is not signaled.
60///
61///  * Transitions to `WAITING` after `begin_wait`.
62const IDLE: u8 = 0;
63
64/// A wait is in progress, and was not interrupted.
65///
66///  * Transitions to `INTERRUPTING` after `interrupt`.
67///  * Transitions to `IDLE` when the wait ends.
68const WAITING: u8 = 1;
69
70/// `interrupt` is signaling the event.
71///
72///  * Transitions to `INTERRUPTED` once `interrupt` has signaled the event.
73const INTERRUPTING: u8 = 2;
74
75/// The wait was interrupted, and the event is signaled.
76///
77///  * Transitions to `IDLE` when the wait ends, after the event is cleared.
78const INTERRUPTED: u8 = 3;
79
80/// The signal that `interrupt` asserts on the event of the waiter.
81const INTERRUPTED_SIGNAL: zx::Signals = zx::Signals::EVENT_SIGNALED;
82
83/// The number of wait items, including the event of the waiter, that `wait_many` keeps on the
84/// stack. Waits for more objects allocate.
85const WAIT_ITEMS_COMMON_SIZE: usize = 4;
86
87/// A guard object to enforce that clients call `begin_wait` before `wait_many`.
88///
89/// The wait ends when the guard is consumed by `wait_many`, or dropped.
90#[must_use = "call wait_many to block until the objects are signaled or the wait is interrupted"]
91#[derive(Debug)]
92pub struct ObjectWaitGuard<'a> {
93    waiter: &'a Arc<ObjectWaiter>,
94}
95
96/// A description of why `wait_many` returned without any of the given objects being signaled.
97#[derive(Debug, PartialEq, Eq)]
98pub enum ObjectWaitError {
99    /// `wait_many` returned because another thread interrupted the wait using `interrupt`.
100    Interrupted,
101
102    /// `wait_many` returned because the given deadline expired.
103    DeadlineExpired,
104
105    /// `wait_many` returned because `zx_object_wait_many` failed with the given status, for
106    /// example because the handle of an object does not have the `ZX_RIGHT_WAIT` right.
107    Failed(zx::Status),
108}
109
110impl ObjectWaiter {
111    pub fn new() -> Arc<Self> {
112        Arc::new(Self { event: zx::Event::create(), state: AtomicU8::new(IDLE) })
113    }
114
115    /// Called to initiate a wait.
116    ///
117    /// Calls to `interrupt` after this function returns, and before the wait ends, will cause
118    /// `wait_many` to return `ObjectWaitError::Interrupted`. Calls to `interrupt` prior to calling
119    /// `begin_wait`, or after the wait ended, are ignored.
120    ///
121    /// Once called, this function cannot be called again until the returned guard is consumed by
122    /// `wait_many`, or dropped. Otherwise, this function will panic.
123    pub fn begin_wait<'a>(self: &'a Arc<Self>) -> ObjectWaitGuard<'a> {
124        // The event is not signaled in the `IDLE` state, so this needs no syscall. The `Release`
125        // pairs with the `Acquire` of `interrupt`, so that an `interrupt` that observes `WAITING`
126        // signals the event after the end of the previous wait cleared it.
127        self.state
128            .compare_exchange(IDLE, WAITING, Ordering::AcqRel, Ordering::Relaxed)
129            .expect("Tried to begin waiting on an ObjectWaiter that is already waiting.");
130        ObjectWaitGuard { waiter: self }
131    }
132
133    /// Interrupts the wait in progress.
134    ///
135    /// If no wait is in progress, that is, before `begin_wait` or after the wait ended, the
136    /// interruption is ignored. Otherwise, `wait_many` returns `ObjectWaitError::Interrupted`,
137    /// whether it is already blocked or not. Calling this function repeatedly has no effect.
138    pub fn interrupt(&self) {
139        // Only signal the event of a wait in progress. An `interrupt` that observes the state of a
140        // previous wait, or of no wait, does nothing, so it cannot signal the event of a later
141        // wait.
142        if self
143            .state
144            .compare_exchange(WAITING, INTERRUPTING, Ordering::Acquire, Ordering::Relaxed)
145            .is_ok()
146        {
147            // The event is owned by this object, so signaling it cannot fail.
148            let _ = self.event.signal(zx::Signals::NONE, INTERRUPTED_SIGNAL);
149            // The `Release` pairs with the `Acquire` of `end_wait`, so that the end of the wait
150            // clears the event after this signal.
151            self.state.store(INTERRUPTED, Ordering::Release);
152        }
153    }
154
155    /// Ends the current wait, and returns to the `IDLE` state.
156    ///
157    /// Returns whether the wait was interrupted, that is, whether `interrupt` moved the state out
158    /// of `WAITING` before the wait ended.
159    fn end_wait(&self) -> bool {
160        loop {
161            match self.state.compare_exchange(WAITING, IDLE, Ordering::Release, Ordering::Acquire) {
162                // The wait was not interrupted, so the event is not signaled.
163                Ok(_) => return false,
164                // `interrupt` is signaling the event. Wait for it to finish, so that the event can
165                // be cleared before the next wait. This only lasts until `interrupt` returns from
166                // `zx_object_signal`.
167                Err(INTERRUPTING) => std::thread::yield_now(),
168                Err(INTERRUPTED) => {
169                    // The event is owned by this object, so signaling it cannot fail.
170                    let _ = self.event.signal(INTERRUPTED_SIGNAL, zx::Signals::NONE);
171                    self.state.store(IDLE, Ordering::Release);
172                    return true;
173                }
174                Err(state) => unreachable!("Unexpected ObjectWaiter state: {state}"),
175            }
176        }
177    }
178}
179
180impl<'a> ObjectWaitGuard<'a> {
181    /// The waiter associated with this guard.
182    pub fn waiter(&self) -> &'a Arc<ObjectWaiter> {
183        self.waiter
184    }
185
186    /// Blocks the thread until one of `items` is signaled, `deadline` expires, or the wait is
187    /// interrupted.
188    ///
189    /// Each of `items` names an object and the signals to wait for on it, as returned by the
190    /// `wait_item` method of the handle of the object. Once at least one of the objects asserts
191    /// one of the signals it is waited for, returns the signals pending on each object, in the
192    /// order of `items`.
193    ///
194    /// Returns `ObjectWaitError::Interrupted` if `interrupt` was called after `begin_wait` and
195    /// before the wait ended, even if some of the objects are also signaled, so that
196    /// interruptions are never missed.
197    ///
198    /// `items` must contain fewer than `ZX_WAIT_MANY_MAX_ITEMS` items, because the waiter adds its
199    /// own event to them.
200    pub fn wait_many<const N: usize>(
201        self,
202        items: [zx::WaitItem<'_>; N],
203        deadline: zx::MonotonicInstant,
204    ) -> Result<[zx::Signals; N], ObjectWaitError> {
205        self.wait_many_with_clean_delay(items, deadline, crate::STACK_CLEAN_DELAY)
206    }
207
208    fn wait_many_with_clean_delay<const N: usize>(
209        self,
210        items: [zx::WaitItem<'_>; N],
211        deadline: zx::MonotonicInstant,
212        clean_delay: zx::MonotonicDuration,
213    ) -> Result<[zx::Signals; N], ObjectWaitError> {
214        const { assert!(N < zx::sys::ZX_WAIT_MANY_MAX_ITEMS, "Too many items to wait for") };
215
216        // `zx_object_wait_many` takes a single array, so append the event of the waiter to
217        // `items`. Reserve room for all of them first, so that the items are not moved to the heap
218        // after being written inline when there are `WAIT_ITEMS_COMMON_SIZE` of them.
219        let waiter = self.waiter;
220        let mut all_items: SmallVec<[zx::WaitItem<'_>; WAIT_ITEMS_COMMON_SIZE]> =
221            SmallVec::with_capacity(N + 1);
222        all_items.extend(items);
223        all_items.push(waiter.event.wait_item(INTERRUPTED_SIGNAL));
224
225        // Clean the stack once if the wait remains blocked for `clean_delay`.
226        let mut clean_deadline = crate::initial_clean_deadline(deadline, clean_delay);
227        let result = loop {
228            match zx::object_wait_many(&mut all_items, clean_deadline) {
229                Err(zx::Status::TIMED_OUT) if clean_deadline < deadline => {
230                    starnix_stack::clean_stack();
231                    clean_deadline = deadline;
232                }
233                result => break result,
234            }
235        };
236
237        // The state of the waiter, rather than the signals that `zx_object_wait_many` returned,
238        // tells whether the wait was interrupted: `interrupt` may have moved the state out of
239        // `WAITING` before signaling the event, or after the syscall returned.
240        if self.end() {
241            return Err(ObjectWaitError::Interrupted);
242        }
243        match result {
244            // `Ok(true)` means that a handle was closed during the wait, which cannot happen
245            // because `items` borrow their handles.
246            Ok(_) => Ok(std::array::from_fn(|i| all_items[i].pending())),
247            Err(zx::Status::TIMED_OUT) => Err(ObjectWaitError::DeadlineExpired),
248            Err(status) => Err(ObjectWaitError::Failed(status)),
249        }
250    }
251
252    /// Ends the wait instead of `drop`, and returns whether the wait was interrupted.
253    fn end(self) -> bool {
254        let waiter = self.waiter;
255        std::mem::forget(self);
256        waiter.end_wait()
257    }
258}
259
260impl Drop for ObjectWaitGuard<'_> {
261    fn drop(&mut self) {
262        self.waiter.end_wait();
263    }
264}
265
266#[cfg(test)]
267mod test {
268    use super::*;
269    use std::sync::atomic::AtomicBool;
270
271    fn signal(event: &zx::Event) {
272        event.signal(zx::Signals::NONE, zx::Signals::USER_0).expect("failed to signal event");
273    }
274
275    /// Waits with `guard` until `event` asserts `USER_0`, `deadline` expires, or the wait is
276    /// interrupted.
277    fn wait_for(
278        guard: ObjectWaitGuard<'_>,
279        event: &zx::Event,
280        deadline: zx::MonotonicInstant,
281    ) -> Result<[zx::Signals; 1], ObjectWaitError> {
282        guard.wait_many([event.wait_item(zx::Signals::USER_0)], deadline)
283    }
284
285    /// Runs `f` on a new thread once the calling thread is blocked in `zx_object_wait_many`.
286    fn spawn_when_blocked(f: impl FnOnce() + Send + 'static) -> std::thread::JoinHandle<()> {
287        let waiting_thread = fuchsia_runtime::with_thread_self(|thread| {
288            thread.duplicate_handle(zx::Rights::SAME_RIGHTS).expect("duplicate_handle")
289        });
290        std::thread::spawn(move || {
291            while waiting_thread.info().expect("info").state
292                != zx::ThreadState::Blocked(zx::ThreadBlockType::WaitMany)
293            {
294                std::thread::sleep(std::time::Duration::from_millis(1));
295            }
296            f();
297        })
298    }
299
300    /// Returns whether the event of `waiter` is signaled.
301    fn is_event_signaled(waiter: &ObjectWaiter) -> bool {
302        waiter.event.wait_one(INTERRUPTED_SIGNAL, zx::MonotonicInstant::INFINITE_PAST).is_ok()
303    }
304
305    #[test]
306    fn test_wait_block_and_signal() {
307        let waiter = ObjectWaiter::new();
308        let event = zx::Event::create();
309        let other_event = event.duplicate_handle(zx::Rights::SAME_RIGHTS).unwrap();
310        let thread = spawn_when_blocked(move || signal(&other_event));
311
312        let result = wait_for(waiter.begin_wait(), &event, zx::MonotonicInstant::INFINITE);
313        assert_eq!(result, Ok([zx::Signals::USER_0]));
314        thread.join().expect("failed to join thread");
315    }
316
317    #[test]
318    fn test_wait_block_and_interrupt() {
319        let waiter = ObjectWaiter::new();
320        let event = zx::Event::create();
321        let other_waiter = Arc::clone(&waiter);
322        let thread = spawn_when_blocked(move || other_waiter.interrupt());
323
324        let result = wait_for(waiter.begin_wait(), &event, zx::MonotonicInstant::INFINITE);
325        assert_eq!(result, Err(ObjectWaitError::Interrupted));
326        thread.join().expect("failed to join thread");
327        // The end of the interrupted wait cleared the event.
328        assert!(!is_event_signaled(&waiter));
329    }
330
331    #[test]
332    fn test_wait_block_and_timeout() {
333        let waiter = ObjectWaiter::new();
334        let event = zx::Event::create();
335        let deadline = zx::MonotonicInstant::after(zx::MonotonicDuration::from_millis(20));
336
337        let result = wait_for(waiter.begin_wait(), &event, deadline);
338        assert_eq!(result, Err(ObjectWaitError::DeadlineExpired));
339        assert!(zx::MonotonicInstant::get() >= deadline);
340    }
341
342    #[test]
343    fn test_poll_with_past_deadline() {
344        let waiter = ObjectWaiter::new();
345        let event = zx::Event::create();
346        let deadline = zx::MonotonicInstant::INFINITE_PAST;
347
348        let result = wait_for(waiter.begin_wait(), &event, deadline);
349        assert_eq!(result, Err(ObjectWaitError::DeadlineExpired));
350        signal(&event);
351        assert_eq!(wait_for(waiter.begin_wait(), &event, deadline), Ok([zx::Signals::USER_0]));
352    }
353
354    #[test]
355    fn test_wait_returns_pending_signals_in_order() {
356        let waiter = ObjectWaiter::new();
357        let event = zx::Event::create();
358        let (channel, peer) = zx::Channel::create();
359        std::mem::drop(peer);
360
361        let result = waiter.begin_wait().wait_many(
362            [
363                event.wait_item(zx::Signals::USER_0),
364                channel.wait_item(zx::Signals::CHANNEL_PEER_CLOSED),
365            ],
366            zx::MonotonicInstant::INFINITE,
367        );
368        let [event_signals, channel_signals] = result.expect("failed to wait");
369        assert_eq!(event_signals, zx::Signals::NONE);
370        assert!(channel_signals.contains(zx::Signals::CHANNEL_PEER_CLOSED));
371    }
372
373    #[test]
374    fn test_wait_without_items() {
375        let waiter = ObjectWaiter::new();
376
377        let result = waiter.begin_wait().wait_many([], zx::MonotonicInstant::INFINITE_PAST);
378        assert_eq!(result, Err(ObjectWaitError::DeadlineExpired));
379        let guard = waiter.begin_wait();
380        waiter.interrupt();
381        let result = guard.wait_many([], zx::MonotonicInstant::INFINITE);
382        assert_eq!(result, Err(ObjectWaitError::Interrupted));
383    }
384
385    #[test]
386    fn test_wait_fails_without_wait_right() {
387        let waiter = ObjectWaiter::new();
388        let event = zx::Event::create().duplicate_handle(zx::Rights::TRANSFER).unwrap();
389
390        let result = wait_for(waiter.begin_wait(), &event, zx::MonotonicInstant::INFINITE);
391        assert_eq!(result, Err(ObjectWaitError::Failed(zx::Status::ACCESS_DENIED)));
392    }
393
394    #[test]
395    fn test_interrupt_outside_wait_is_ignored() {
396        let waiter = ObjectWaiter::new();
397        let event = zx::Event::create();
398
399        // Neither an interruption before the first wait, nor one after a wait ended, signals the
400        // event or interrupts the next wait.
401        for _ in 0..2 {
402            waiter.interrupt();
403            assert!(!is_event_signaled(&waiter));
404            let result = wait_for(waiter.begin_wait(), &event, zx::MonotonicInstant::INFINITE_PAST);
405            assert_eq!(result, Err(ObjectWaitError::DeadlineExpired));
406        }
407    }
408
409    #[test]
410    fn test_interrupt_before_block_ends_wait() {
411        let waiter = ObjectWaiter::new();
412        let event = zx::Event::create();
413
414        // The interruption arrives after `begin_wait`, but before the thread blocks.
415        let guard = waiter.begin_wait();
416        waiter.interrupt();
417        let result = wait_for(guard, &event, zx::MonotonicInstant::INFINITE);
418        assert_eq!(result, Err(ObjectWaitError::Interrupted));
419
420        // It takes precedence over the signals of the objects.
421        signal(&event);
422        let guard = waiter.begin_wait();
423        waiter.interrupt();
424        let result = wait_for(guard, &event, zx::MonotonicInstant::INFINITE);
425        assert_eq!(result, Err(ObjectWaitError::Interrupted));
426
427        // The next wait ignores the previous interruptions.
428        let result = wait_for(waiter.begin_wait(), &event, zx::MonotonicInstant::INFINITE);
429        assert_eq!(result, Ok([zx::Signals::USER_0]));
430    }
431
432    #[test]
433    fn test_interrupt_signals_event_once() {
434        let waiter = ObjectWaiter::new();
435
436        let guard = waiter.begin_wait();
437        assert!(!is_event_signaled(&waiter));
438        waiter.interrupt();
439        assert!(is_event_signaled(&waiter));
440        // Repeated interruptions keep the wait interrupted.
441        waiter.interrupt();
442        assert!(is_event_signaled(&waiter));
443
444        // Dropping the guard of the interrupted wait ends the wait, and clears the event.
445        std::mem::drop(guard);
446        assert!(!is_event_signaled(&waiter));
447        std::mem::drop(waiter.begin_wait());
448    }
449
450    #[test]
451    fn test_interrupt_in_progress_ends_wait() {
452        let waiter = ObjectWaiter::new();
453        let event = zx::Event::create();
454        signal(&event);
455
456        // Start an interruption that has not signaled the event yet, as if `interrupt` were
457        // preempted right after leaving `WAITING`, so that `zx_object_wait_many` returns because
458        // `event` is signaled, without observing the interruption.
459        let guard = waiter.begin_wait();
460        waiter.state.store(INTERRUPTING, Ordering::Relaxed);
461        let other_waiter = Arc::clone(&waiter);
462        let thread = std::thread::spawn(move || {
463            std::thread::sleep(std::time::Duration::from_millis(10));
464            // Finish the interruption, as `interrupt` does.
465            other_waiter.event.signal(zx::Signals::NONE, INTERRUPTED_SIGNAL).expect("signal");
466            other_waiter.state.store(INTERRUPTED, Ordering::Release);
467        });
468
469        // The wait reports the interruption anyway, and its end clears the event.
470        let result = wait_for(guard, &event, zx::MonotonicInstant::INFINITE);
471        assert_eq!(result, Err(ObjectWaitError::Interrupted));
472        thread.join().expect("failed to join thread");
473        assert!(!is_event_signaled(&waiter));
474    }
475
476    #[test]
477    fn test_interrupt_races_with_end_of_wait() {
478        let waiter = ObjectWaiter::new();
479        let event = zx::Event::create();
480        signal(&event);
481        let done = Arc::new(AtomicBool::new(false));
482        let interrupter = std::thread::spawn({
483            let (waiter, done) = (Arc::clone(&waiter), Arc::clone(&done));
484            move || {
485                while !done.load(Ordering::Relaxed) {
486                    waiter.interrupt();
487                }
488            }
489        });
490
491        // Each wait either returns because `event` is signaled, or is interrupted, and no
492        // interruption outlives the wait that it interrupted.
493        for _ in 0..1000 {
494            let result = wait_for(waiter.begin_wait(), &event, zx::MonotonicInstant::INFINITE);
495            assert!(
496                result == Ok([zx::Signals::USER_0]) || result == Err(ObjectWaitError::Interrupted),
497                "Unexpected result: {result:?}"
498            );
499            assert!(!is_event_signaled(&waiter));
500        }
501        done.store(true, Ordering::Relaxed);
502        interrupter.join().expect("failed to join thread");
503    }
504
505    #[test]
506    #[should_panic(expected = "Tried to begin waiting on an ObjectWaiter that is already waiting.")]
507    fn test_concurrent_begin_wait_panics() {
508        let waiter = ObjectWaiter::new();
509        let _guard = waiter.begin_wait();
510        let _ = waiter.begin_wait();
511    }
512
513    #[test]
514    fn test_wait_clean_stack() {
515        let waiter = ObjectWaiter::new();
516        let event = zx::Event::create();
517        let wait = |guard: ObjectWaitGuard<'_>, deadline| {
518            guard.wait_many_with_clean_delay(
519                [event.wait_item(zx::Signals::USER_0)],
520                deadline,
521                zx::MonotonicDuration::from_millis(10),
522            )
523        };
524
525        // After cleaning the stack, the wait blocks again until the deadline expires...
526        let deadline = zx::MonotonicInstant::after(zx::MonotonicDuration::from_millis(30));
527        assert_eq!(wait(waiter.begin_wait(), deadline), Err(ObjectWaitError::DeadlineExpired));
528        assert!(zx::MonotonicInstant::get() >= deadline);
529
530        // ...the wait is interrupted...
531        let guard = waiter.begin_wait();
532        let other_waiter = Arc::clone(&waiter);
533        let thread = std::thread::spawn(move || {
534            std::thread::sleep(std::time::Duration::from_millis(30));
535            other_waiter.interrupt();
536        });
537        let result = wait(guard, zx::MonotonicInstant::INFINITE);
538        assert_eq!(result, Err(ObjectWaitError::Interrupted));
539        thread.join().expect("failed to join thread");
540
541        // ...or an object is signaled.
542        let other_event = event.duplicate_handle(zx::Rights::SAME_RIGHTS).unwrap();
543        let thread = std::thread::spawn(move || {
544            std::thread::sleep(std::time::Duration::from_millis(30));
545            signal(&other_event);
546        });
547        let result = wait(waiter.begin_wait(), zx::MonotonicInstant::INFINITE);
548        assert_eq!(result, Ok([zx::Signals::USER_0]));
549        thread.join().expect("failed to join thread");
550    }
551}
starnix_sync/object_waiter.rs

starnix_sync/
object_waiter.rs
