switchyard 0.3.1

Real-time compute focused async executor
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

use std::{cell::Cell, time::Duration};
use switchyard::{threads::single_thread, Switchyard};

#[test]
fn single_task() {
    let yard = Switchyard::new(single_thread(None, None), || ()).unwrap();
    let handle = yard.spawn(0, async { 12 });
    let result = futures_executor::block_on(handle);

    assert_eq!(result, 12);
}

#[test]
fn single_local_task() {
    let yard = Switchyard::new(single_thread(None, None), || Cell::new(30)).unwrap();
    let handle = yard.spawn_local(0, |value| async move { value.take() + 12 });
    let result = futures_executor::block_on(handle);

    assert_eq!(result, 42);
}

#[test]
fn single_waiting_task() {
    let yard = Switchyard::new(single_thread(None, None), || ()).unwrap();

    // The design of this test is to only send the value to wait for after we confirm that the
    // future is running, which gives it a very high chance of the await returning Pending
    // allowing us to test our wakers.
    let (sender1, receiver1) = flume::unbounded();
    let (sender2, receiver2) = flume::unbounded();

    let handle = yard.spawn(0, async move {
        // We're running the future
        sender1.send(()).unwrap();
        // Wait for data
        receiver2.recv_async().await.unwrap()
    });

    // Wait until the future is running
    receiver1.recv().unwrap();
    // Send data
    sender2.send(42).unwrap();

    let result = futures_executor::block_on(handle);

    assert_eq!(result, 42);
}

#[test]
fn single_local_waiting_task() {
    let yard = Switchyard::new(single_thread(None, None), || ()).unwrap();

    // The design of this test is to only send the value to wait for after we confirm that the
    // future is running, which gives it a very high chance of the await returning Pending
    // allowing us to test our wakers.
    let (sender1, receiver1) = flume::unbounded();
    let (sender2, receiver2) = flume::unbounded();

    let handle = yard.spawn_local(0, move |_| async move {
        // We're running the future
        sender1.send(()).unwrap();
        // Wait for data
        receiver2.recv_async().await.unwrap()
    });

    // Wait until the future is running
    receiver1.recv().unwrap();

    assert_eq!(yard.jobs(), 1);

    // Send data
    sender2.send(42).unwrap();

    let result = futures_executor::block_on(handle);

    assert_eq!(result, 42);
}

#[test]
fn multi_tasks() {
    let yard = Switchyard::new(single_thread(None, None), || ()).unwrap();

    let task_one = yard.spawn(0, async move { 2 * 2 });
    let task_two = yard.spawn(0, async move { 4 * 4 });

    assert_eq!(futures_executor::block_on(task_one), 2 * 2);
    assert_eq!(futures_executor::block_on(task_two), 4 * 4);
}

#[test]
fn wait_for_idle() {
    let yard = Switchyard::new(single_thread(None, None), || ()).unwrap();

    let (sender, receiver) = flume::unbounded();
    let task = yard.spawn(0, async move {
        // Regular sleep to hard block the thread
        std::thread::sleep(Duration::from_millis(100));
        sender.send(2 * 2).unwrap()
    });

    futures_executor::block_on(yard.wait_for_idle());

    assert_eq!(receiver.recv().unwrap(), 4);
    assert_eq!(futures_executor::block_on(task), ());
    assert_eq!(yard.jobs(), 0);
}

#[test]
fn wait_for_idle_stalled() {
    let yard = Switchyard::new(single_thread(None, None), || ()).unwrap();

    let (sender, receiver) = flume::unbounded();
    let task = yard.spawn(0, async move {
        receiver.recv_async().await.unwrap();
        2 * 2
    });

    assert_eq!(yard.jobs(), 1);

    // nothing has been sent, wait for idle should clear immediately
    futures_executor::block_on(yard.wait_for_idle());

    assert_eq!(yard.jobs(), 1);

    sender.send(()).unwrap();

    assert_eq!(futures_executor::block_on(task), 4);
}