arty 0.6.0

Thread-aware async runtime that keeps each task on one worker thread.
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
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154

// Copyright (c) Microsoft Corporation.
// Licensed under the MIT License.

//! Wake notification shared by an asynchronous worker and its callers.

use std::sync::{Arc as StdArc, OnceLock};
use std::task::{Wake, Waker};
use std::time::Duration;

use performables::arc::Arc;
use performables::sync::notify::Notify;

/// A coalescing notification shared by task wakers and the command dispatcher.
#[derive(Debug, Default)]
pub(crate) struct Signal {
    ready: Notify,
    driver_waker: OnceLock<Waker>,
    #[cfg(test)]
    wait_observer: Option<std::sync::mpsc::Sender<()>>,
}

impl Signal {
    #[cfg(test)]
    pub(crate) fn with_wait_observer(wait_observer: std::sync::mpsc::Sender<()>) -> Self {
        Self {
            wait_observer: Some(wait_observer),
            ..Self::default()
        }
    }

    pub(crate) fn wait_for_notification(&self) {
        if !self.ready.try_wait() {
            #[cfg(test)]
            if let Some(wait_observer) = &self.wait_observer {
                _ = wait_observer.send(());
            }
            self.ready.wait();
        }
    }

    pub(crate) fn take_notification(&self) -> bool {
        self.ready.try_wait()
    }

    pub(crate) fn wait(&self, timeout: Duration) {
        _ = self.ready.wait_timeout(timeout);
    }

    pub(crate) fn waker(signal: &Arc<Self>) -> Waker {
        // Wake requires a standard Arc; both handles share the same allocation.
        Waker::from(Arc::into_std_arc(Arc::clone(signal)))
    }

    #[cfg_attr(test, mutants::skip)] // Removing driver interruption strands primary waits across unrelated runtime tests.
    pub(crate) fn set_driver_waker(&self, waker: Waker) {
        assert!(
            self.driver_waker.set(waker).is_ok(),
            "a worker installs its I/O driver interrupt waker once"
        );
    }

    #[cfg(test)]
    #[cfg_attr(coverage_nightly, coverage(off))]
    #[cfg_attr(test, mutants::skip)]
    pub(crate) fn is_notified(&self) -> bool {
        self.ready.is_notified()
    }

    #[cfg_attr(test, mutants::skip)] // Removing all worker and driver wake effects strands runtime tests; each effect is tested directly.
    fn notify(&self) {
        self.ready.notify_one();
        if let Some(driver_waker) = self.driver_waker.get() {
            driver_waker.wake_by_ref();
        }
    }
}

impl Wake for Signal {
    fn wake(self: StdArc<Self>) {
        self.notify();
    }

    fn wake_by_ref(self: &StdArc<Self>) {
        self.notify();
    }
}

#[cfg(test)]
#[cfg_attr(coverage_nightly, coverage(off))] // Test scaffolding is not runtime behavior.
mod tests {
    use std::sync::mpsc;
    use std::thread;

    use testing_aids::{TEST_TIMEOUT, is_mutation_testing};

    use super::*;

    #[test]
    fn notification_before_wait_is_retained_and_consumed() {
        let signal = Arc::new(Signal::default());
        let waker = Signal::waker(&signal);
        waker.wake_by_ref();
        assert!(signal.is_notified());
        signal.wait(Duration::ZERO);
        assert!(!signal.is_notified());
        waker.wake();
        assert!(signal.is_notified());
        signal.wait(Duration::ZERO);
        assert!(!signal.is_notified());
    }

    #[test]
    fn remote_wake_releases_an_indefinite_waiter() {
        let (wait_entered, wait_entry) = mpsc::channel();
        let signal = Arc::new(Signal::with_wait_observer(wait_entered));
        let (done_tx, done_rx) = mpsc::channel();
        let waiter = Arc::clone(&signal);
        let worker = thread::spawn(move || {
            waiter.wait_for_notification();
            done_tx.send(()).unwrap();
        });
        wait_entry.recv_timeout(TEST_TIMEOUT).unwrap();
        assert_eq!(
            done_rx.recv_timeout(Duration::from_millis(20)),
            Err(mpsc::RecvTimeoutError::Timeout)
        );
        Signal::waker(&signal).wake_by_ref();
        done_rx.recv_timeout(Duration::from_secs(1)).unwrap();
        worker.join().unwrap();
    }

    #[test]
    fn notification_before_indefinite_wait_is_retained_and_consumed() {
        let signal = Arc::new(Signal::default());
        Signal::waker(&signal).wake_by_ref();

        signal.wait_for_notification();

        assert!(!signal.is_notified());
    }

    #[test]
    fn an_unnotified_wait_can_time_out() {
        if is_mutation_testing() {
            return;
        }
        let signal = Signal::default();
        let timeout = Duration::from_millis(20);
        let start = std::time::Instant::now();
        signal.wait(timeout);
        assert!(start.elapsed() >= timeout);
        assert!(!signal.is_notified());
    }
}