arty 0.4.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

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

//! Escaped timer callbacks cannot release storage referenced by live task wakers.

#![cfg(feature = "rt")]
#![cfg(test)]
#![cfg(not(miri))] // The regression isolates intentional process termination in a native child.

testing_aids::init_tracing!();

use std::cell::RefCell;
use std::future::poll_fn;
use std::pin::Pin;
use std::process::Command;
use std::sync::{Arc, mpsc};
use std::task::{Context, Poll, Wake, Waker};
use std::time::Duration;

use arty::runtime::{Runtime, WorkersPolicy};
use testing_aids::TEST_TIMEOUT;
use thread_aware::Unaware;

const CHILD: &str = "ARTY_WORKER_PANIC_BOUNDARY_CHILD";
const MESSAGE: &str = "timer callback crossed the task boundary";

struct PanicWake;

impl Wake for PanicWake {
    fn wake(self: Arc<Self>) {
        panic!("{MESSAGE}");
    }
}

struct WorkerExited(mpsc::Sender<()>);

impl Drop for WorkerExited {
    fn drop(&mut self) {
        let _ = self.0.send(());
    }
}

thread_local! {
    static EXIT: RefCell<Option<WorkerExited>> = const { RefCell::new(None) };
}

#[test]
fn timer_panic_cannot_unwind_past_live_executor_storage() {
    if std::env::var_os(CHILD).is_some() {
        let runtime = Runtime::builder().workers(WorkersPolicy::exactly(1)).build().unwrap();
        let (sent, received) = mpsc::channel();
        let (exited, worker_exited) = mpsc::channel();
        let _task = runtime
            .scheduler()
            .spawn_anywhere(Unaware((sent, exited)), |cx, Unaware((sent, exited))| async move {
                EXIT.with_borrow_mut(|exit| *exit = Some(WorkerExited(exited)));
                let mut sent = Some(sent);
                poll_fn(move |context| {
                    if let Some(sent) = sent.take() {
                        sent.send(context.waker().clone()).unwrap();
                        let mut timer = cx.clock().delay(Duration::from_millis(1));
                        let waker = Waker::from(Arc::new(PanicWake));
                        assert!(Pin::new(&mut timer).poll(&mut Context::from_waker(&waker)).is_pending());
                        // A poisoned timer lock would make its destructor mask the lifecycle failure.
                        std::mem::forget(timer);
                    }
                    Poll::<()>::Pending
                })
                .await;
            });
        let retained = received.recv_timeout(TEST_TIMEOUT).unwrap();
        if worker_exited.recv_timeout(TEST_TIMEOUT).is_err() {
            // A hang is not the required fail-closed termination. Leak objects that
            // may still reference the worker and let the child exit successfully so
            // the parent rejects this outcome.
            std::mem::forget(retained);
            std::mem::forget(runtime);
            return;
        }
        // The boundary must terminate before TLS teardown. Do not dereference a stale waker.
        std::mem::forget(retained);
        std::mem::forget(runtime);
        return;
    }

    let output = Command::new(std::env::current_exe().unwrap())
        .arg("--exact")
        .arg("timer_panic_cannot_unwind_past_live_executor_storage")
        .arg("--nocapture")
        .env(CHILD, "1")
        .output()
        .unwrap();
    assert!(!output.status.success());
    assert!(String::from_utf8_lossy(&output.stderr).contains(MESSAGE));
}