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
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

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

//! Worker clocks advance timers and remain usable after relocation.

#![cfg(all(feature = "rt", feature = "macros"))]

use std::pin::Pin;
use std::sync::{Arc, mpsc};
use std::task::{Context as PollContext, Poll, Wake, Waker};
use std::thread;
use std::time::Duration;

use arty::runtime::{Runtime, WorkersPolicy};
use arty::task::Context;
use tick::{ClockControl, FutureExt};

const TIMER_COUNT: usize = 8;
const CONTROLLED_TIMER_STEP: Duration = Duration::from_secs(1);

#[arty::test]
async fn many_timers_with_relocation_ensure_advanced(context: Context) {
    let scheduler = context.scheduler();
    // Make sure that each relocated task keeps access to the runtime clock.
    let handles: Vec<_> = (0..TIMER_COUNT)
        .map(|_| {
            scheduler.spawn_anywhere(context.clone(), move |context| async move {
                println!("delay(pending) - clock: {:?}, thread: {:?}", context.clock(), context.thread());

                let watch = context.clock().stopwatch();
                context.clock().delay(Duration::from_millis(50)).await;

                println!("delay(done) - clock: {:?}, thread: {:?}", context.clock(), context.thread());

                assert!(watch.elapsed().as_millis() >= 50);
            })
        })
        .collect();

    for handle in handles {
        handle.timeout(context.clock(), Duration::from_secs(10)).await.unwrap().unwrap();
    }
}

#[arty::test]
async fn many_timers_ensure_advanced(context: Context) {
    let handles: Vec<_> = (0..TIMER_COUNT)
        .map(|_| {
            context
                .scheduler()
                .spawn_anywhere(context.clone(), |context| context.clock().delay(Duration::from_millis(1)))
        })
        .collect();

    for handle in handles {
        handle.await.unwrap();
    }
}

#[arty::test]
async fn timer_with_relocated_context(context: Context) {
    context
        .scheduler()
        .spawn_anywhere(context.clone(), |context| context.clock().delay(Duration::from_millis(10)))
        .await
        .unwrap();
}

#[test]
fn controlled_timer_callback_submission_wakes_bound_worker() {
    struct SpawnOnWake {
        scheduler: arty::task::Scheduler,
        task_ran: mpsc::Sender<(thread::ThreadId, thread::ThreadId)>,
    }

    impl Wake for SpawnOnWake {
        fn wake(self: Arc<Self>) {
            let callback_thread = thread::current().id();
            let task_ran = self.task_ran.clone();
            drop(self.scheduler.spawn(move |_| async move {
                task_ran.send((callback_thread, thread::current().id())).unwrap();
            }));
        }
    }

    struct TimerWakeHarness {
        callback_delay: Pin<Box<tick::Delay>>,
        callback_waker: Waker,
        registration: Option<mpsc::Sender<thread::ThreadId>>,
    }

    impl Future for TimerWakeHarness {
        type Output = ();

        fn poll(self: Pin<&mut Self>, _cx: &mut PollContext<'_>) -> Poll<Self::Output> {
            let this = self.get_mut();
            if let Some(registration) = this.registration.take() {
                assert_eq!(
                    this.callback_delay
                        .as_mut()
                        .poll(&mut PollContext::from_waker(&this.callback_waker)),
                    Poll::Pending
                );
                registration.send(thread::current().id()).unwrap();
            }
            Poll::Pending
        }
    }

    let control = ClockControl::new();
    let runtime = Runtime::builder()
        .workers(WorkersPolicy::exactly(1))
        .clock(control.clone())
        .build()
        .unwrap();
    let scheduler = runtime.scheduler().block_on(async |cx| cx.scheduler().clone()).unwrap();
    let (timer_registered, registration) = mpsc::channel();
    let (task_ran, completion) = mpsc::channel();
    let _timer_task = scheduler.spawn(move |cx| TimerWakeHarness {
        callback_delay: Box::pin(cx.clock().delay(CONTROLLED_TIMER_STEP)),
        callback_waker: Waker::from(Arc::new(SpawnOnWake {
            scheduler: cx.scheduler().clone(),
            task_ran,
        })),
        registration: Some(timer_registered),
    });
    let worker_thread = registration.recv_timeout(testing_aids::TEST_TIMEOUT).unwrap();
    let control_thread = thread::current().id();

    control.advance(CONTROLLED_TIMER_STEP);

    let observed_threads = completion.recv_timeout(testing_aids::TEST_TIMEOUT);
    runtime.stop().unwrap();

    assert_eq!(
        observed_threads.expect("the controlled timer callback submission must wake its bound worker"),
        (control_thread, worker_thread)
    );
}