desync 0.9.0

A hassle-free data type for asynchronous programming
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

use desync::scheduler::*;

use super::timeout::*;

use std::thread;
use std::time::*;
use std::sync::*;
use std::sync::mpsc::*;

#[test]
#[cfg(not(miri))]   // slow!
fn suspend_queue_with_background_thread() {
    for _x in 0..1000 {
        timeout(|| {
            use futures::executor;

            let queue           = queue();
            let scheduler       = scheduler();
            let (tx, rx)        = channel();

            let pos             = Arc::new(Mutex::new(0));

            // Send the current position when the async methods run
            let pos2            = pos.clone();
            let tx2             = tx.clone();
            desync(&queue, move || { tx2.send(*pos2.lock().unwrap()).unwrap(); });

            // Suspend after the first send
            let suspended   = scheduler.suspend(&queue);
            let resumer     = executor::block_on(suspended).unwrap();

            // Send again
            let pos2 = pos.clone();
            let tx2 = tx.clone();
            desync(&queue, move || { tx2.send(*pos2.lock().unwrap()).unwrap(); });

            // Wait for the first queue to send
            assert!(rx.recv().unwrap() == 0);

            thread::yield_now();

            // Update the position and resume
            *pos.lock().unwrap() = 1;
            resumer.resume();

            // The resumption will send us a value when it occurs
            assert!(rx.recv().unwrap() == 1);
        }, 500);
    }
}

#[test]
fn suspend_queue_with_local_drain() {
    timeout(|| {
        use futures::executor;

        // Want a scheduler with 0 threads to force a 'drain on current thread' situation
        let scheduler = Arc::new(Scheduler::new());
        scheduler.set_max_threads(0);
        scheduler.despawn_threads_if_overloaded();

        let queue = scheduler.create_job_queue();
        
        // Job so there's something to drain
        scheduler.desync(&queue, ||{});

        // Start suspending the queue
        let suspended   = scheduler.suspend(&queue);
        let resumer     = executor::block_on(suspended).unwrap();

        // Resume after a delay
        let resume_scheduler    = scheduler.clone();
        thread::spawn(move || {
            // Wait 100ms with the queue suspended
            thread::sleep(Duration::from_millis(100));

            // Resume it
            resumer.resume();

            // The scheduler will need to be able to finish the task on a thread
            resume_scheduler.set_max_threads(1);
        });

        // Should be able to retrieve a value once the queue resumes
        assert!(scheduler.sync(&queue, || 42) == 42);
    }, 500);
}

#[test]
fn safe_to_drop_suspended_queue() {
    timeout(|| {
        use futures::executor;

        let queue       = queue();
        let scheduler   = scheduler();

        let pos         = Arc::new(Mutex::new(0));

        // Increment the position, suspend the queue, increment it again
        let pos2        = pos.clone();
        desync(&queue, move || { let mut pos2 = pos2.lock().unwrap(); *pos2 += 1 });
        let suspended   = scheduler.suspend(&queue);
        let pos2        = pos.clone();
        desync(&queue, move || { let mut pos2 = pos2.lock().unwrap(); *pos2 += 1 });

        // Wait for long enough for these events to take place and check the queue
        while *pos.lock().unwrap() == 0 {
            thread::sleep(Duration::from_millis(100));
        }
        assert!(*pos.lock().unwrap() == 1);

        // Make sure the queue is actually suspended
        executor::block_on(suspended).unwrap();
    }, 500);
}