async-reify 0.1.1

Async computation inspector: extract continuation structure as step graphs
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

use async_reify::{PollResult, Trace};
use async_reify_macros::trace_async;
use std::sync::{Arc, Mutex};

async fn fetch() -> i32 {
    42
}

async fn slow() -> i32 {
    tokio::task::yield_now().await;
    7
}

#[trace_async(trace = trace)]
async fn workflow_simple(trace: Arc<Mutex<Trace>>) -> i32 {
    let a = fetch().await;
    let b = fetch().await;
    a + b
}

#[tokio::test]
async fn rewrites_simple_awaits() {
    let trace = Trace::shared();
    let value = workflow_simple(trace.clone()).await;
    assert_eq!(value, 84);

    let snapshot = Arc::try_unwrap(trace).unwrap().into_inner().unwrap();
    // Each `fetch().await` should record at least one Ready event.
    let ready_count = snapshot
        .events
        .iter()
        .filter(|e| matches!(e.result, PollResult::Ready))
        .count();
    assert!(
        ready_count >= 2,
        "expected >=2 Ready events, got {ready_count}: {:?}",
        snapshot.events
    );

    // Labels should include source location info.
    assert!(snapshot.events.iter().all(|e| {
        e.label
            .as_ref()
            .map(|l| l.contains('@') && l.contains(".rs:"))
            .unwrap_or(false)
    }));
}

#[trace_async(trace = trace)]
async fn workflow_pending(trace: Arc<Mutex<Trace>>) -> i32 {
    slow().await + slow().await
}

#[tokio::test]
async fn rewrites_pending_awaits() {
    let trace = Trace::shared();
    let value = workflow_pending(trace.clone()).await;
    assert_eq!(value, 14);

    let snapshot = Arc::try_unwrap(trace).unwrap().into_inner().unwrap();
    let pending_count = snapshot
        .events
        .iter()
        .filter(|e| matches!(e.result, PollResult::Pending))
        .count();
    let ready_count = snapshot
        .events
        .iter()
        .filter(|e| matches!(e.result, PollResult::Ready))
        .count();
    // yield_now produces at least one Pending then Ready per call.
    assert!(pending_count >= 2, "expected pending events");
    assert!(ready_count >= 2, "expected ready events");
}

#[trace_async(trace = trace)]
async fn workflow_chained(trace: Arc<Mutex<Trace>>) -> i32 {
    fetch().await + fetch().await
}

#[tokio::test]
async fn rewrites_chained_method_awaits() {
    let trace = Trace::shared();
    let value = workflow_chained(trace.clone()).await;
    assert_eq!(value, 84);

    let snapshot = Arc::try_unwrap(trace).unwrap().into_inner().unwrap();
    assert!(snapshot.events.len() >= 2);
}

// A nested async block inside the function. Awaits inside a nested closure
// or item should be left alone, but awaits in nested async blocks within
// the same scope are still rewritten (they share the outer trace).
#[trace_async(trace = trace)]
async fn workflow_nested_block(trace: Arc<Mutex<Trace>>) -> i32 {
    let inner = async { fetch().await }.await;
    inner
}

#[tokio::test]
async fn rewrites_awaits_in_nested_async_block() {
    let trace = Trace::shared();
    let value = workflow_nested_block(trace.clone()).await;
    assert_eq!(value, 42);

    let snapshot = Arc::try_unwrap(trace).unwrap().into_inner().unwrap();
    assert!(!snapshot.events.is_empty());
}