[go: up one dir, main page]

tokio 1.46.1

An event-driven, non-blocking I/O platform for writing asynchronous I/O backed applications.
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

#![warn(rust_2018_idioms)]
#![cfg(feature = "full")]

use std::sync::Arc;
use tokio::runtime::Runtime;
use tokio::sync::{mpsc, Barrier};

#[test]
#[cfg_attr(panic = "abort", ignore)]
fn basic_enter() {
    let rt1 = rt();
    let rt2 = rt();

    let enter1 = rt1.enter();
    let enter2 = rt2.enter();

    drop(enter2);
    drop(enter1);
}

#[test]
#[should_panic]
#[cfg_attr(panic = "abort", ignore)]
fn interleave_enter_different_rt() {
    let rt1 = rt();
    let rt2 = rt();

    let enter1 = rt1.enter();
    let enter2 = rt2.enter();

    drop(enter1);
    drop(enter2);
}

#[test]
#[should_panic]
#[cfg_attr(panic = "abort", ignore)]
fn interleave_enter_same_rt() {
    let rt1 = rt();

    let _enter1 = rt1.enter();
    let enter2 = rt1.enter();
    let enter3 = rt1.enter();

    drop(enter2);
    drop(enter3);
}

#[test]
#[cfg(not(target_os = "wasi"))]
#[cfg_attr(panic = "abort", ignore)]
fn interleave_then_enter() {
    let _ = std::panic::catch_unwind(|| {
        let rt1 = rt();
        let rt2 = rt();

        let enter1 = rt1.enter();
        let enter2 = rt2.enter();

        drop(enter1);
        drop(enter2);
    });

    // Can still enter
    let rt3 = rt();
    let _enter = rt3.enter();
}

// If the cycle causes a leak, then miri will catch it.
#[test]
fn drop_tasks_with_reference_cycle() {
    rt().block_on(async {
        let (tx, mut rx) = mpsc::channel(1);

        let barrier = Arc::new(Barrier::new(3));
        let barrier_a = barrier.clone();
        let barrier_b = barrier.clone();

        let a = tokio::spawn(async move {
            let b = rx.recv().await.unwrap();

            // Poll the JoinHandle once. This registers the waker.
            // The other task cannot have finished at this point due to the barrier below.
            futures::future::select(b, std::future::ready(())).await;

            barrier_a.wait().await;
        });

        let b = tokio::spawn(async move {
            // Poll the JoinHandle once. This registers the waker.
            // The other task cannot have finished at this point due to the barrier below.
            futures::future::select(a, std::future::ready(())).await;

            barrier_b.wait().await;
        });

        tx.send(b).await.unwrap();

        barrier.wait().await;
    });
}

#[cfg(tokio_unstable)]
mod unstable {
    use super::*;

    #[test]
    fn runtime_id_is_same() {
        let rt = rt();

        let handle1 = rt.handle();
        let handle2 = rt.handle();

        assert_eq!(handle1.id(), handle2.id());
    }

    #[test]
    fn runtime_ids_different() {
        let rt1 = rt();
        let rt2 = rt();

        assert_ne!(rt1.handle().id(), rt2.handle().id());
    }
}

fn rt() -> Runtime {
    tokio::runtime::Builder::new_current_thread()
        .build()
        .unwrap()
}