reactors 0.1.4

Performance-focused cross-platform asynchronous IO implementation
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

//! Hashed [`timewheel`](https://blog.acolyer.org/2015/11/23/hashed-and-hierarchical-timing-wheels/)
//! implementation for [`IoReactor`](crate::io::IoReactor)

use std::collections::HashMap;
use std::task::Poll;
// Time wheel algorithem impl
#[derive(Debug)]
struct Slot<T> {
    round: u64,
    t: T,
}

/// Timewheel implementation for [`IoReactor`](crate::io::IoReactor)
#[derive(Debug)]
pub struct TimeWheel<T> {
    hashed: HashMap<u64, Vec<Slot<T>>>,
    steps: u64,
    tick: u64,
}

impl<T> TimeWheel<T> {
    // create new hashed time wheel instance
    pub fn new(steps: u64) -> Self {
        TimeWheel {
            steps: steps,
            hashed: HashMap::new(),
            tick: 0,
        }
    }

    pub fn add(&mut self, timeout: u64, value: T) {
        log::trace!(
            "add timeout({}) steps({}) tick({})",
            timeout,
            self.steps,
            self.tick
        );

        let slot = (timeout + self.tick) % self.steps;
        let round = timeout / self.steps;

        log::trace!(
            "add timeout({}) to slot({}) with round({}), current tick is {}",
            timeout,
            slot,
            round,
            self.tick
        );

        let slots = self.hashed.entry(slot).or_insert(Vec::new());

        slots.push(Slot { t: value, round });
    }

    pub fn tick(&mut self) -> Poll<Vec<T>> {
        let step = self.tick % self.steps;

        self.tick += 1;

        if let Some(slots) = self.hashed.remove(&step) {
            let mut current: Vec<T> = vec![];
            let mut reserved: Vec<Slot<T>> = vec![];

            for slot in slots {
                if slot.round == 0 {
                    current.push(slot.t);
                } else {
                    reserved.push(Slot::<T> {
                        t: slot.t,
                        round: slot.round - 1,
                    });
                }
            }

            if !reserved.is_empty() {
                self.hashed.insert(step, reserved);
            }

            if !current.is_empty() {
                return Poll::Ready(current);
            }
        }

        Poll::Pending
    }
}