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

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

use std::sync::{Arc, OnceLock};

use arty_executor::{ShutdownCause, ShutdownError};

/// Shared runtime shutdown state.
#[derive(Clone, Default, derive_more::Debug)]
pub(crate) struct Shutdown {
    #[debug(ignore)]
    inner: Arc<OnceLock<ShutdownCause>>,
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub(crate) enum ShutdownState {
    Running,
    Started,
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub(crate) enum ShutdownStart {
    Started,
    AlreadyStarted,
}

impl Shutdown {
    /// Starts shutdown and reports whether this call started it.
    ///
    /// The first published cause is authoritative. Later requests do not
    /// replace it, including a default cause without an error.
    pub(crate) fn begin(&self, cause: ShutdownCause) -> ShutdownStart {
        match self.inner.set(cause) {
            Ok(()) => ShutdownStart::Started,
            Err(_) => ShutdownStart::AlreadyStarted,
        }
    }

    #[must_use]
    pub(crate) fn state(&self) -> ShutdownState {
        if self.inner.get().is_some() {
            ShutdownState::Started
        } else {
            ShutdownState::Running
        }
    }

    #[must_use]
    pub(crate) fn cause(&self) -> Option<ShutdownCause> {
        self.inner.get().cloned()
    }

    #[must_use]
    pub(crate) fn error(&self) -> Option<ShutdownError> {
        self.cause().and_then(ShutdownCause::into_error)
    }
}

#[cfg(test)]
#[cfg_attr(coverage_nightly, coverage(off))]
mod tests {
    use std::error::Error;
    use std::fmt::{self, Debug, Display};
    use std::sync::Barrier;
    use std::thread;

    use static_assertions::assert_impl_all;

    use super::*;

    #[derive(Debug)]
    struct TestError(&'static str);

    impl Display for TestError {
        fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
            f.write_str(self.0)
        }
    }

    impl Error for TestError {}

    fn cause(message: &'static str) -> ShutdownCause {
        ShutdownError::new(TestError(message)).into()
    }

    #[test]
    fn default_cause_remains_authoritative() {
        assert_impl_all!(Shutdown: Clone, Debug, Default, Send, Sync);

        let shutdown = Shutdown::default();

        assert_eq!(shutdown.begin(ShutdownCause::default()), ShutdownStart::Started);
        assert_eq!(shutdown.begin(cause("later")), ShutdownStart::AlreadyStarted);

        assert_eq!(shutdown.state(), ShutdownState::Started);
        assert!(shutdown.error().is_none());
    }

    #[test]
    fn error_cause_remains_authoritative() {
        let shutdown = Shutdown::default();

        assert_eq!(shutdown.begin(cause("first")), ShutdownStart::Started);
        assert_eq!(shutdown.begin(cause("later")), ShutdownStart::AlreadyStarted);

        assert_eq!(shutdown.error().unwrap().inner().to_string(), "first");
    }

    #[test]
    fn concurrent_requests_publish_one_stable_cause() {
        let shutdown = Shutdown::default();
        let barrier = Arc::new(Barrier::new(3));
        let workers: [_; 2] = ["first", "second"].map(|message| {
            let shutdown = shutdown.clone();
            let barrier = Arc::clone(&barrier);
            thread::spawn(move || {
                barrier.wait();
                shutdown.begin(cause(message))
            })
        });

        barrier.wait();
        let started = workers.map(|worker| worker.join().unwrap());

        assert_eq!(started.into_iter().filter(|started| *started == ShutdownStart::Started).count(), 1);
        let observed = shutdown.error().unwrap().inner().to_string();
        assert!(matches!(observed.as_str(), "first" | "second"));
        assert_eq!(shutdown.error().unwrap().inner().to_string(), observed);
    }
}