atap 0.1.0

Threadsafe futureless async runtime for macOS
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

//! # Shutdown

use atap::{
    Runtime, RuntimeError,
    sleep::{Sleep, SleepMode},
};
use std::{
    thread,
    time::{Duration, Instant},
};

/// A shutdown drains the backlog, refuses new work, and leaves
/// `block` working, then `init` starts the runtime again
#[test]
fn shutdown_drains_the_backlog_then_init_starts_it_again() {
    let _ = Runtime::init();

    let tasks = 2_000;

    // Half on workers and half on sleep threads
    let handles: Vec<_> = (0..tasks)
        .map(|task| {
            let mode = match task % 2 == 0 {
                true => SleepMode::Precise,
                false => SleepMode::Relaxed,
            };

            Runtime::task(Sleep::sleep(Duration::from_micros(200)).mode(mode)).spawn()
        })
        .collect();

    let started = Instant::now();
    Runtime::shutdown();
    let drained = started.elapsed();

    println!("shutdown drained {tasks} tasks in {drained:?}");

    // Everything spawned before it ran to the end
    for (task, handle) in handles.into_iter().enumerate() {
        handle
            .join()
            .unwrap_or_else(|error| panic!("task {task} was dropped by the shutdown: {error}"));
    }

    // Closed to new work
    let late =
        Runtime::task(Sleep::sleep(Duration::from_millis(10)).mode(SleepMode::Relaxed)).spawn();
    let kept = late.clone();

    assert_eq!(
        late.join(),
        Err(RuntimeError::TaskFailed),
        "a spawn after a shutdown settles instead of waiting for something that has gone",
    );

    // Blocking calls still work
    let slept = Runtime::block(Sleep::sleep(Duration::from_millis(20)).mode(SleepMode::Relaxed));

    println!("a blocking call after the shutdown still slept {slept:?}");

    assert!(
        slept >= Duration::from_millis(20),
        "the blocking call came back early, so the reactor went down with the pool",
    );

    let status = Runtime::status();
    println!("{status}");

    assert!(status.shut_down(), "the status says what happened");
    assert!(!status.manager_alive(), "the manager's queue was closed");
    assert!(!status.healthy());

    // Safe twice
    Runtime::shutdown();

    for cycle in 0..2 {
        // Starts again, and only once
        assert_eq!(
            Runtime::init(),
            Ok(()),
            "cycle {cycle}: the runtime didn't start again"
        );

        assert_eq!(
            Runtime::init(),
            Err(RuntimeError::AlreadyInit),
            "cycle {cycle}: a running runtime was started twice",
        );

        let status = Runtime::status();
        println!("cycle {cycle}: {status}");

        assert!(
            status.healthy(),
            "cycle {cycle}: the runtime came back degraded"
        );

        // A handle from before the shutdown still reads what its
        // task ended with
        assert_eq!(
            kept.try_join(),
            Err(RuntimeError::TaskFailed),
            "cycle {cycle}: a restart changed what an old handle reads",
        );

        let quick = Runtime::task(Sleep::sleep(Duration::from_micros(200))).spawn();
        let blocking =
            Runtime::task(Sleep::sleep(Duration::from_millis(5)).mode(SleepMode::Relaxed)).spawn();

        // Both need the new manager's timers
        let delayed =
            Runtime::task(Sleep::sleep(Duration::from_millis(1)).mode(SleepMode::Relaxed))
                .after(Duration::from_millis(50))
                .spawn();

        let counted =
            Runtime::task(Sleep::sleep(Duration::from_millis(1)).mode(SleepMode::Relaxed))
                .repeat()
                .every(Duration::from_millis(10))
                .count(3)
                .spawn();

        quick.join().expect("a task spawned after the restart runs");
        blocking
            .join()
            .expect("a blocking task spawned after the restart runs");
        delayed
            .join()
            .expect("a delayed task spawned after the restart runs");

        let deadline = Instant::now() + Duration::from_secs(10);

        while !counted.is_finished() && Instant::now() < deadline {
            thread::sleep(Duration::from_millis(1));
        }

        assert!(
            counted.is_finished(),
            "cycle {cycle}: a timed repeat never got through its count"
        );
        assert!(
            !counted.is_failed(),
            "cycle {cycle}: a timed repeat failed after the restart"
        );

        Runtime::shutdown();

        assert!(
            Runtime::status().shut_down(),
            "cycle {cycle}: the second shutdown didn't take"
        );
    }
}