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

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

//! Shutdown cancels pending work and releases its captured resources.

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

testing_aids::init_tracing!();

use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};

use arty::runtime::{Runtime, RuntimeOperations};
use events_once::{BoxedReceiver, Event};
use testing_aids::execute_or_abandon;
use thread_aware::Unaware;

fn canary() -> (impl Future<Output = ()>, BoxedReceiver<()>, std::sync::Weak<()>) {
    let lifetime = Arc::new(());
    let observer = Arc::downgrade(&lifetime);
    let (started, notification) = Event::boxed();
    let future = async move {
        started.send(());
        std::future::pending::<()>().await;
        drop(lifetime);
    };
    (future, notification, observer)
}

#[test]
fn stop_via_runtime() {
    execute_or_abandon(|| {
        let runtime = Runtime::new().unwrap();

        let (canary, started, observer) = canary();

        runtime.scheduler().spawn_anywhere(Unaware(canary), |_, Unaware(canary)| canary);
        futures::executor::block_on(started).unwrap();

        runtime.stop().unwrap();

        // We expect the canary to have died. Otherwise, the runtime is still running!
        assert!(observer.upgrade().is_none());
    })
    .unwrap();
}

#[test]
fn stop_via_async_task() {
    execute_or_abandon(|| {
        let runtime = Runtime::new().unwrap();

        let (canary, started, observer) = canary();

        runtime.scheduler().spawn_anywhere(Unaware(canary), |_, Unaware(canary)| canary);

        futures::executor::block_on(started).unwrap();

        let (requested, stop_requested) = Event::boxed();
        runtime
            .scheduler()
            .spawn_anywhere(Unaware(requested), |cx, Unaware(requested)| async move {
                RuntimeOperations::from(&cx).request_stop();
                requested.send(());
            });

        futures::executor::block_on(stop_requested).unwrap();
        runtime.stop().unwrap();

        // We expect the canary to have died. Otherwise, the runtime is still running!
        assert!(observer.upgrade().is_none());
    })
    .unwrap();
}

#[test]
fn stop_scoped() {
    execute_or_abandon(move || {
        _ = Runtime::new().unwrap();
    })
    .unwrap();
}

#[test]
fn stop_in_run() {
    let task_was_executed = Arc::new(AtomicBool::new(false));

    execute_or_abandon({
        let task_was_executed = Arc::clone(&task_was_executed);

        move || {
            Runtime::new()
                .unwrap()
                .scheduler()
                .block_on(async move |_| {
                    task_was_executed.store(true, Ordering::Relaxed);
                })
                .unwrap();
        }
    })
    .unwrap();

    assert!(task_was_executed.load(Ordering::Relaxed));
}