arty 0.5.0

Single-threaded, thread-aware application runtime.
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

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

//! Blocking shutdown must join running work and discard queued captures safely.

#![cfg(feature = "rt")]
#![cfg(test)]

mod panic_support;

mod support;

use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::{Arc, mpsc};

use arty::runtime::RuntimeOperations;
use panic_support::runtime;
use support::JoinHandleExt as _;
use testing_aids::TEST_TIMEOUT;

testing_aids::init_tracing!();

struct CaptureDropPanic {
    drops: Arc<AtomicUsize>,
    dropped: mpsc::Sender<()>,
}

impl Drop for CaptureDropPanic {
    fn drop(&mut self) {
        self.drops.fetch_add(1, Ordering::SeqCst);
        self.dropped.send(()).unwrap();
        assert!(!std::thread::panicking());
        panic!("discarded blocking capture destructor");
    }
}

#[test]
fn queued_blocking_capture_drop_panic_still_completes_its_join() {
    let runtime = runtime();
    let operations = RuntimeOperations::from(&runtime);
    let (started, ready) = mpsc::channel();
    let (release, gate) = mpsc::channel();
    let running = runtime.scheduler().spawn_blocking(move || {
        started.send(()).unwrap();
        gate.recv_timeout(TEST_TIMEOUT).unwrap();
        42
    });
    ready.recv_timeout(TEST_TIMEOUT).unwrap();
    let (dropped, received) = mpsc::channel();
    let drops = Arc::new(AtomicUsize::new(0));
    let invoked = Arc::new(AtomicUsize::new(0));
    let capture = CaptureDropPanic {
        drops: Arc::clone(&drops),
        dropped,
    };
    let captured_invoked = Arc::clone(&invoked);
    let queued = runtime.scheduler().spawn_blocking(move || {
        captured_invoked.fetch_add(1, Ordering::SeqCst);
        drop(capture);
    });
    operations.request_stop();
    release.send(()).unwrap();
    assert_eq!(running.join().unwrap(), 42);
    runtime.stop().unwrap();
    received.recv_timeout(TEST_TIMEOUT).unwrap();
    assert_eq!(drops.load(Ordering::SeqCst), 1);
    assert_eq!(invoked.load(Ordering::SeqCst), 0);
    assert!(queued.join().unwrap_err().is_shutdown());
}

#[test]
fn running_blocking_capture_drop_panic_is_a_join_error() {
    let runtime = runtime();
    let (dropped, received) = mpsc::channel();
    let drops = Arc::new(AtomicUsize::new(0));
    let capture = CaptureDropPanic {
        drops: Arc::clone(&drops),
        dropped,
    };
    let task = runtime.scheduler().spawn_blocking(move || drop(capture));
    assert!(task.join().unwrap_err().is_panic());
    received.recv_timeout(TEST_TIMEOUT).unwrap();
    assert_eq!(drops.load(Ordering::SeqCst), 1);
    assert_eq!(runtime.scheduler().spawn_blocking(|| 7).join().unwrap(), 7);
    runtime.stop().unwrap();
}

#[test]
fn stop_waits_for_running_blocking_panic_and_retains_its_error() {
    let runtime = runtime();
    let operations = RuntimeOperations::from(&runtime);
    let (started, ready) = mpsc::channel();
    let (release, gate) = mpsc::channel();
    let (finished, observed) = mpsc::channel();
    let task: arty::task::JoinHandle<()> = runtime.scheduler().spawn_blocking(move || {
        started.send(()).unwrap();
        gate.recv_timeout(TEST_TIMEOUT).unwrap();
        finished.send(()).unwrap();
        panic!("running blocking callback at shutdown");
    });
    ready.recv_timeout(TEST_TIMEOUT).unwrap();
    operations.request_stop();
    release.send(()).unwrap();
    runtime.stop().unwrap();
    observed.recv_timeout(TEST_TIMEOUT).unwrap();
    assert!(task.join().unwrap_err().is_panic());
}