tokio 0.2.13

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

use crate::runtime::thread_pool::queue;
use crate::task::{self, Task};
use crate::tests::mock_schedule::{Noop, NOOP_SCHEDULE};

use loom::thread;

use std::cell::Cell;
use std::rc::Rc;

#[test]
fn multi_worker() {
    const THREADS: usize = 2;
    const PER_THREAD: usize = 7;

    fn work(_i: usize, q: queue::Worker<Noop>, rem: Rc<Cell<usize>>) {
        let mut rem_local = PER_THREAD;

        while rem.get() != 0 {
            for _ in 0..3 {
                if rem_local > 0 {
                    q.push(val(0));
                    rem_local -= 1;
                }
            }

            // Try to work
            while let Some(task) = q.pop_local_first() {
                assert!(task.run(&mut || Some(From::from(&NOOP_SCHEDULE))).is_none());
                let r = rem.get();
                assert!(r > 0);
                rem.set(r - 1);
            }

            // Try to steal
            if let Some(task) = q.steal(0) {
                assert!(task.run(&mut || Some(From::from(&NOOP_SCHEDULE))).is_none());
                let r = rem.get();
                assert!(r > 0);
                rem.set(r - 1);
            }

            thread::yield_now();
        }
    }

    loom::model(|| {
        let rem = Rc::new(Cell::new(THREADS * PER_THREAD));

        let mut qs = queue::build(THREADS);
        let q1 = qs.remove(0);

        for i in 1..THREADS {
            let q = qs.remove(0);
            let rem = rem.clone();
            thread::spawn(move || {
                work(i, q, rem);
            });
        }

        work(0, q1, rem);

        // th.join().unwrap();
    });
}

fn val(num: u32) -> Task<Noop> {
    let (task, _) = task::joinable(async move { num });
    task
}