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

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

//! A user poll panic racing shutdown cannot lose its join or destroy a task twice.

use std::cell::Cell;
use std::marker::PhantomPinned;
use std::pin::Pin;
use std::rc::Rc;
use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
use std::sync::{Arc, Barrier, mpsc};
use std::task::{Context, Poll, Waker};

use testing_aids::TEST_TIMEOUT;
use thread_aware::Unaware;

use crate::runtime_support::runtime;
use crate::support::{self, JoinHandleExt as _};

struct RacingPanic {
    first: Cell<Option<mpsc::Sender<Waker>>>,
    ready: Arc<AtomicBool>,
    drops: Arc<AtomicUsize>,
    address: Cell<Option<usize>>,
    owner: std::thread::ThreadId,
    _local: Rc<()>,
    _pinned: PhantomPinned,
}

impl Future for RacingPanic {
    type Output = ();

    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<()> {
        let address = std::ptr::from_ref(self.as_ref().get_ref()).addr();
        if let Some(previous) = self.address.replace(Some(address)) {
            assert_eq!(previous, address);
        }
        if let Some(first) = self.first.take() {
            first.send(cx.waker().clone()).unwrap();
        }
        assert!(!self.ready.load(Ordering::SeqCst), "task poll racing stop");
        Poll::Pending
    }
}

impl Drop for RacingPanic {
    fn drop(&mut self) {
        assert_eq!(Some(std::ptr::from_ref(self).addr()), self.address.get());
        assert_eq!(std::thread::current().id(), self.owner);
        self.drops.fetch_add(1, Ordering::SeqCst);
    }
}

#[test]
fn poll_panic_racing_repeated_stop_always_completes_one_join_and_drop() {
    let repetitions = if cfg!(miri) { 2 } else { 24 };
    for _ in 0..repetitions {
        let runtime = runtime();
        let control = support::runtime_control(&runtime);
        let (first, received) = mpsc::channel();
        let ready = Arc::new(AtomicBool::new(false));
        let drops = Arc::new(AtomicUsize::new(0));
        let task = runtime.scheduler().spawn_anywhere(
            Unaware((first, Arc::clone(&ready), Arc::clone(&drops))),
            |_, Unaware((first, ready, drops))| RacingPanic {
                first: Cell::new(Some(first)),
                ready,
                drops,
                address: Cell::new(None),
                owner: std::thread::current().id(),
                _local: Rc::new(()),
                _pinned: PhantomPinned,
            },
        );
        let retained = received.recv_timeout(TEST_TIMEOUT).unwrap();
        let barrier = Arc::new(Barrier::new(3));
        let stop_barrier = Arc::clone(&barrier);
        let stopper = std::thread::spawn(move || {
            stop_barrier.wait();
            for _ in 0..8 {
                control.request_stop();
            }
        });
        let panic_barrier = Arc::clone(&barrier);
        let panicker = std::thread::spawn(move || {
            panic_barrier.wait();
            ready.store(true, Ordering::SeqCst);
            retained.wake();
        });
        barrier.wait();
        stopper.join().unwrap();
        panicker.join().unwrap();
        runtime.stop().unwrap();
        let error = task.join().unwrap_err();
        assert!(error.is_panic() || error.is_shutdown());
        assert_ne!(error.is_panic(), error.is_shutdown());
        assert_eq!(drops.load(Ordering::SeqCst), 1);
    }
}