events 0.6.0

Async manual-reset and auto-reset events for multi-use signaling
Documentation 
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

use std::cell::Cell;
use std::sync::atomic::{AtomicBool, Ordering};
use std::task::{RawWaker, RawWakerVTable, Waker};

/// A custom waker that executes a caller-supplied closure when woken.
///
/// This is used to test that `wake()` is never called while holding borrows
/// on interior-mutable state (e.g. `UnsafeCell<WaiterList>`). If a re-entrant
/// waker accesses the same state, Miri detects the aliased access as UB.
///
/// # Thread safety
///
/// This type is `!Send` and must only be used with single-threaded (`Local`)
/// event types. The backing state uses `Cell` which is not thread-safe.
/// For thread-safe event types, use [`AtomicWakeTracker`] instead.
pub(crate) struct ReentrantWakerData {
    action: Box<dyn Fn()>,
    was_woken: Cell<bool>,
}

impl ReentrantWakerData {
    /// The returned `Box` ensures a stable address for the waker data pointer.
    #[expect(
        clippy::unnecessary_box_returns,
        reason = "the Box ensures a stable address for the raw waker data pointer"
    )]
    pub(crate) fn new(action: impl Fn() + 'static) -> Box<Self> {
        Box::new(Self {
            action: Box::new(action),
            was_woken: Cell::new(false),
        })
    }

    /// Creates a [`Waker`] backed by this data.
    ///
    /// # Safety
    ///
    /// * The caller must ensure this `ReentrantWakerData` outlives all wakers
    ///   (and their clones) created from it.
    /// * The returned [`Waker`] (and any clones) must not be sent to or used
    ///   from any thread other than the one that created it. The backing
    ///   state uses [`Cell`] and a non-thread-safe closure.
    pub(crate) unsafe fn waker(&self) -> Waker {
        let data: *const () = std::ptr::from_ref(self).cast();

        // SAFETY: The caller upholds the safety contract of `waker()`:
        // the data outlives all wakers, and the waker is only used on
        // the creating thread.
        unsafe { Waker::from_raw(RawWaker::new(data, &VTABLE)) }
    }

    pub(crate) fn was_woken(&self) -> bool {
        self.was_woken.get()
    }
}

fn clone_fn(data: *const ()) -> RawWaker {
    RawWaker::new(data, &VTABLE)
}

fn wake_fn(data: *const ()) {
    wake_by_ref_fn(data);
}

fn wake_by_ref_fn(data: *const ()) {
    // SAFETY: data points to a valid ReentrantWakerData that outlives this
    // waker per the contract on `ReentrantWakerData::waker()`.
    let this = unsafe { &*(data as *const ReentrantWakerData) };
    this.was_woken.set(true);
    (this.action)();
}

fn drop_fn(_data: *const ()) {}

static VTABLE: RawWakerVTable = RawWakerVTable::new(clone_fn, wake_fn, wake_by_ref_fn, drop_fn);

/// A thread-safe waker that tracks whether `wake()` was called.
///
/// Unlike [`ReentrantWakerData`], this type is `Send + Sync` and uses atomic
/// operations, making it suitable for testing thread-safe event types.
pub(crate) struct AtomicWakeTracker {
    woken: AtomicBool,
}

impl AtomicWakeTracker {
    /// The returned `Box` ensures a stable address for the raw waker pointer.
    #[expect(
        clippy::unnecessary_box_returns,
        reason = "the Box ensures a stable address for the raw waker data pointer"
    )]
    pub(crate) fn new() -> Box<Self> {
        Box::new(Self {
            woken: AtomicBool::new(false),
        })
    }

    /// Creates a [`Waker`] backed by this tracker.
    ///
    /// # Safety
    ///
    /// The caller must ensure this `AtomicWakeTracker` outlives all wakers
    /// (and their clones) created from it.
    pub(crate) unsafe fn waker(&self) -> Waker {
        let data: *const () = std::ptr::from_ref(self).cast();

        // SAFETY: The caller ensures this AtomicWakeTracker outlives all
        // wakers created from it.
        unsafe { Waker::from_raw(RawWaker::new(data, &ATOMIC_VTABLE)) }
    }

    pub(crate) fn was_woken(&self) -> bool {
        self.woken.load(Ordering::Relaxed)
    }
}

fn atomic_clone_fn(data: *const ()) -> RawWaker {
    RawWaker::new(data, &ATOMIC_VTABLE)
}

fn atomic_wake_fn(data: *const ()) {
    atomic_wake_by_ref_fn(data);
}

fn atomic_wake_by_ref_fn(data: *const ()) {
    // SAFETY: data points to a valid AtomicWakeTracker that outlives this
    // waker per the contract on `AtomicWakeTracker::waker()`.
    let this = unsafe { &*(data as *const AtomicWakeTracker) };
    this.woken.store(true, Ordering::Relaxed);
}

fn atomic_drop_fn(_data: *const ()) {}

static ATOMIC_VTABLE: RawWakerVTable = RawWakerVTable::new(
    atomic_clone_fn,
    atomic_wake_fn,
    atomic_wake_by_ref_fn,
    atomic_drop_fn,
);