bach 0.1.2

Discrete-event simulation environment for async workflows
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
131
132
133
134
135
136
137
138
139
140
141
142
143

pub trait Entry: Sized {
    type Queue: Queue<Self>;

    fn delay(&self) -> u64;
    fn start_tick(&self) -> u64;
    fn set_start_tick(&mut self, tick: u64);

    /// Returns `true` if this entry has been cancelled and should be dropped
    /// during cascading rather than re-inserted into the wheel.
    fn is_cancelled(&self) -> bool {
        false
    }
}

pub trait Queue<Entry> {
    fn new() -> Self;
    fn is_empty(&self) -> bool;
    fn push(&mut self, entry: Entry);
    fn pop(&mut self) -> Option<Entry>;
    fn take(&mut self) -> Self;
}

pub mod atomic {
    use super::*;
    use alloc::sync::Arc;
    use atomic_waker::AtomicWaker;
    use core::{
        sync::atomic::{AtomicBool, AtomicU64, Ordering},
        task::Waker,
    };
    use intrusive_collections::{intrusive_adapter, LinkedList, LinkedListLink};

    intrusive_adapter!(pub Adapter = ArcEntry: Entry { link => LinkedListLink });

    pub type ArcEntry = Arc<Entry>;

    #[derive(Debug)]
    pub struct Entry {
        waker: AtomicWaker,
        expired: AtomicBool,
        registered: AtomicBool,
        delay: u64,
        start_tick: AtomicU64,
        link: LinkedListLink,
    }

    unsafe impl Send for Entry {}
    unsafe impl Sync for Entry {}

    pub fn wake(entry: ArcEntry) {
        entry.wake();
    }

    impl Entry {
        pub fn new(delay: u64) -> Arc<Self> {
            Arc::new(Self {
                waker: AtomicWaker::new(),
                expired: AtomicBool::new(false),
                registered: AtomicBool::new(false),
                delay,
                start_tick: AtomicU64::new(0),
                link: LinkedListLink::new(),
            })
        }

        pub fn wake(&self) {
            self.expired.store(true, Ordering::SeqCst);
            self.registered.store(false, Ordering::SeqCst);

            if let Some(waker) = self.waker.take() {
                waker.wake();
            }
        }

        pub fn should_register(&self) -> bool {
            !self.registered.swap(true, Ordering::SeqCst)
        }

        pub fn cancel(&self) {
            self.waker.take();
        }

        pub fn take_expired(&self) -> bool {
            self.expired.swap(false, Ordering::SeqCst)
        }

        pub fn register(&self, waker: &Waker) {
            self.waker.register(waker)
        }
    }

    impl super::Entry for Arc<Entry> {
        type Queue = LinkedList<Adapter>;

        fn delay(&self) -> u64 {
            self.delay
        }

        fn start_tick(&self) -> u64 {
            self.start_tick.load(Ordering::SeqCst)
        }

        fn set_start_tick(&mut self, tick: u64) {
            self.start_tick.store(tick, Ordering::SeqCst);
        }

        fn is_cancelled(&self) -> bool {
            // The application holds one Arc clone (via `Timer`); the wheel
            // holds another.  When the application drops its `Timer` the
            // strong count falls to 1, meaning only the wheel still holds the
            // entry and it should be discarded rather than cascaded further.
            Arc::strong_count(self) == 1
        }
    }

    impl Drop for Entry {
        fn drop(&mut self) {
            self.cancel();
        }
    }

    impl Queue<ArcEntry> for LinkedList<Adapter> {
        fn new() -> Self {
            LinkedList::new(Adapter::new())
        }

        fn is_empty(&self) -> bool {
            LinkedList::is_empty(self)
        }

        fn push(&mut self, entry: ArcEntry) {
            self.push_back(entry);
        }

        fn pop(&mut self) -> Option<ArcEntry> {
            self.pop_front()
        }

        fn take(&mut self) -> Self {
            LinkedList::take(self)
        }
    }
}