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
144
145
146
147
148
149
150
151
152
153
154
155
156
// Double-buffered event queue with per-reader cursors. Replaces the
// drained-Vec-as-event pattern, which is lossy within a frame (one reader's
// drain hides the events from every other reader, and multiple sends collapse
// to last-write-wins). Here an event stays readable for two `update` cycles, so
// any reader that runs after the writer sees it, and every reader sees every
// event exactly once.
//
// Each event carries a monotonically increasing sequence id. A reader's cursor
// stores the next id it has not yet seen; reading yields every buffered event
// at or past the cursor and advances it.
use alloc::vec::Vec;
/// A double-buffered event queue: events stay readable for two frames.
pub struct Events<E> {
// Two frame buffers. `newest` indexes the one new events go into; the other
// holds the previous frame's events, still readable.
buffers: [Vec<E>; 2],
newest: usize,
// Id assigned to the next event sent.
next_id: usize,
// Sequence id of the first event in each buffer.
starts: [usize; 2],
}
#[derive(Clone, Copy, Debug, Default)]
/// A reader's position in an [`Events`] queue.
pub struct EventCursor {
// Next sequence id this reader has not yet consumed.
next: usize,
}
impl<E> Default for Events<E> {
fn default() -> Events<E> {
Events {
buffers: [Vec::new(), Vec::new()],
newest: 0,
next_id: 0,
starts: [0, 0],
}
}
}
impl<E> Events<E> {
/// An empty queue.
pub fn new() -> Events<E> {
Events::default()
}
/// Queue an event. It becomes visible to readers immediately and stays
/// readable until the second `update` after this one.
pub fn send(&mut self, event: E) {
self.buffers[self.newest].push(event);
self.next_id += 1;
}
/// Advance one frame: retire the older buffer and start a fresh newest one.
/// Events older than two cycles are dropped.
pub fn update(&mut self) {
let oldest = self.newest ^ 1;
self.buffers[oldest].clear();
self.starts[oldest] = self.next_id;
self.newest = oldest;
}
/// Read every buffered event the cursor has not yet seen, in send order, and
/// advance the cursor past them.
///
/// Lazy: a drain costs no allocation, which matters because every event
/// reader does this every frame. The cursor advances here rather than as the
/// iterator is consumed, so a caller that reads only part of the run still
/// ends up past all of it -- the same thing a returned collection did, and
/// the only behaviour that makes "every reader sees every event exactly
/// once" hold for a partial read.
pub fn read(&self, cursor: &mut EventCursor) -> impl Iterator<Item = &E> {
// Visit buffers oldest-first so events come back in send order.
let (older, newer) = if self.starts[0] > self.starts[1] {
(1, 0)
} else {
(0, 1)
};
let from = cursor.next;
cursor.next = self.next_id;
self.unseen(older, from).chain(self.unseen(newer, from))
}
// The events in one buffer at or past sequence id `from`.
fn unseen(&self, buffer: usize, from: usize) -> impl Iterator<Item = &E> {
let start = self.starts[buffer];
// Ids within a buffer are contiguous from `start`, so the cut is a
// position rather than a per-event test.
let skip = from.saturating_sub(start);
self.buffers[buffer].iter().skip(skip)
}
/// Total events currently buffered across both frames.
pub fn len(&self) -> usize {
self.buffers[0].len() + self.buffers[1].len()
}
/// Whether both frame buffers are empty.
pub fn is_empty(&self) -> bool {
self.buffers[0].is_empty() && self.buffers[1].is_empty()
}
}
#[cfg(test)]
mod tests {
use super::*;
use alloc::vec;
#[test]
fn reader_sees_each_event_once() {
let mut events: Events<u32> = Events::new();
events.send(1);
events.send(2);
let mut cursor = EventCursor::default();
let first: Vec<u32> = events.read(&mut cursor).copied().collect();
assert_eq!(first, vec![1, 2]);
// A second read with the same cursor sees nothing new.
assert_eq!(events.read(&mut cursor).count(), 0);
// A newly sent event is picked up.
events.send(3);
let next: Vec<u32> = events.read(&mut cursor).copied().collect();
assert_eq!(next, vec![3]);
}
#[test]
fn multiple_readers_each_see_all_events() {
let mut events: Events<u32> = Events::new();
events.send(10);
events.send(20);
let mut a = EventCursor::default();
let mut b = EventCursor::default();
let read_a: Vec<u32> = events.read(&mut a).copied().collect();
let read_b: Vec<u32> = events.read(&mut b).copied().collect();
assert_eq!(read_a, vec![10, 20]);
assert_eq!(read_b, vec![10, 20]);
}
#[test]
fn events_survive_one_update_then_drop() {
let mut events: Events<u32> = Events::new();
events.send(1);
events.update();
// Still readable one cycle later, in send order with a later event.
events.send(2);
let mut cursor = EventCursor::default();
let seen: Vec<u32> = events.read(&mut cursor).copied().collect();
assert_eq!(seen, vec![1, 2]);
// Two updates retire the first event entirely.
events.update();
events.update();
assert!(events.is_empty());
}
}