eventador 0.0.18

lock-free pub/sub event-bus with sync and async APIs
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

use crate::sequence::sequence_group::SequenceGroup;
use crate::sequence::Sequence;
use crate::WaitStrategy;
use std::sync::Arc;

pub struct Sequencer {
    cursor: Sequence,
    gating_sequence_cache: Arc<Sequence>,
    gating_sequences: SequenceGroup,
    ring_capacity: u64,
    wait_strategy: WaitStrategy,
}

impl Sequencer {
    pub fn new(ring_capacity: u64, wait_strategy: WaitStrategy) -> Self {
        Self {
            cursor: Sequence::with_value(0),
            gating_sequence_cache: Arc::new(Sequence::with_value(0)),
            gating_sequences: SequenceGroup::new(),
            ring_capacity,
            wait_strategy,
        }
    }

    pub(crate) fn register_gating_sequence(&self, sequence: Arc<Sequence>) {
        self.gating_sequences.add(sequence);
    }

    pub fn get(&self) -> u64 {
        self.cursor.get()
    }

    pub fn next(&self) -> u64 {
        self.next_from(1)
            .expect("sequencer could not get next sequence number from sequence 1")
    }

    #[cfg(feature = "async")]
    pub async fn async_next(&self) -> u64 {
        self.async_next_from(1)
            .await
            .expect("sequencer could not get next sequence number from sequence 1")
    }

    pub fn next_from(&self, n: u64) -> anyhow::Result<u64> {
        if n < 1 || n > self.ring_capacity {
            return Err(anyhow::Error::msg("n must be > 0 and < buffer_size"));
        }

        loop {
            std::hint::spin_loop();

            let current: u64 = self.cursor.get();
            let icurrent: i64 = current as i64;
            let next: i64 = (current + n) as i64;

            let wrap_point: i64 = next - self.ring_capacity as i64;
            let cached_gating_sequence: i64 = self.gating_sequence_cache.get() as i64;

            if wrap_point >= cached_gating_sequence || cached_gating_sequence > icurrent {
                let gating_sequence = self.gating_sequences.minimum_sequence(current);

                match self.wait_strategy {
                    WaitStrategy::AllSubscribers => {
                        if wrap_point > gating_sequence as i64 {
                            std::thread::sleep(std::time::Duration::from_micros(100));
                            continue;
                        }
                    }

                    WaitStrategy::NoWait => {
                        if self.cursor.compare_exchange(current, next as u64) {
                            return Ok(next as u64);
                        }
                    }

                    WaitStrategy::WaitForDuration(wait) => {
                        if self.cursor.compare_exchange(current, next as u64) {
                            std::thread::sleep(wait);

                            return Ok(next as u64);
                        }
                    }
                }

                self.gating_sequence_cache.set(gating_sequence);
            } else if self.cursor.compare_exchange(current, next as u64) {
                return Ok(next as u64);
            }
        }
    }

    #[cfg(feature = "async")]
    pub async fn async_next_from(&self, n: u64) -> anyhow::Result<u64> {
        if n < 1 || n > self.ring_capacity {
            return Err(anyhow::Error::msg("n must be > 0 and < buffer_size"));
        }

        loop {
            let current: u64 = self.cursor.get();
            let icurrent: i64 = current as i64;
            let next: i64 = (current + n) as i64;

            let wrap_point: i64 = next - self.ring_capacity as i64;
            let cached_gating_sequence: i64 = self.gating_sequence_cache.get() as i64;

            if wrap_point >= cached_gating_sequence || cached_gating_sequence > icurrent {
                let gating_sequence = self.gating_sequences.minimum_sequence(current);

                match self.wait_strategy {
                    WaitStrategy::AllSubscribers => {
                        if wrap_point > gating_sequence as i64 {
                            async_std::task::sleep(std::time::Duration::from_micros(100)).await;
                            continue;
                        }
                    }

                    WaitStrategy::NoWait => {
                        if self.cursor.compare_exchange(current, next as u64) {
                            return Ok(next as u64);
                        }
                    }

                    WaitStrategy::WaitForDuration(wait) => {
                        if self.cursor.compare_exchange(current, next as u64) {
                            async_std::task::sleep(wait).await;

                            return Ok(next as u64);
                        }
                    }
                }

                self.gating_sequence_cache.set(gating_sequence);
            } else if self.cursor.compare_exchange(current, next as u64) {
                return Ok(next as u64);
            }
        }
    }
}