uexec 0.1.4

Simple work-stealing global and local 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

use crate::{future_holder::FutureHolder, threads::Threads};

use std::{
    collections::{HashMap, HashSet, VecDeque},
    sync::Arc,
};

use parking_lot::Mutex;

/* The common state containing all the futures */
#[derive(Clone, Default)]
pub(crate) struct State(Arc<Mutex<Futures>>);

#[derive(Default)]
struct Futures {
    /* Pending futures */
    pending: HashMap<u64, FutureHolder>,
    /* List of tasks that need polling */
    pollable: VecDeque<FutureHolder>,
    /* List of tasks that are already pollable but not yet registered */
    pollable_next: HashSet<u64>,
}

impl State {
    pub(crate) fn register_future(&self, future: FutureHolder, threads: Threads) {
        let mut futures = self.0.lock();
        if futures.pollable_next.remove(&future.id()) {
            futures.pollable.push_back(future);
            threads.unpark_random();
        } else {
            futures.pending.insert(future.id(), future);
        }
    }

    pub(crate) fn register_pollable(&self, id: u64) {
        let mut futures = self.0.lock();
        if let Some(future) = futures.pending.remove(&id) {
            // Future was pending, mark it as pollable
            futures.pollable.push_back(future);
        } else {
            // Future hasn't been registered yet
            futures.pollable_next.insert(id);
        }
    }

    pub(crate) fn has_pollable_tasks(&self) -> bool {
        !self.0.lock().pollable.is_empty()
    }

    pub(crate) fn cancel(&self, id: u64) {
        let mut futures = self.0.lock();
        // If the future was pending, do nothing, otherwise...
        if futures.pending.remove(&id).is_none() {
            // ...check if it was pollable
            if let Some((index, _)) = futures
                .pollable
                .iter()
                .enumerate()
                .find(|(_, future)| future.id() == id)
            {
                if let Some(future) = futures.pollable.remove(index) {
                    // If it was pollable, give it one last chance
                    future.last_run();
                }
            }
        }
    }

    pub(crate) fn next(&self) -> Option<FutureHolder> {
        self.0.lock().pollable.pop_front()
    }
}