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
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
//! Lock-free coordination based on Compare and Swap atomics
use const_init::ConstInit;
use super::{Coord, ReadGrantError, WriteGrantError};
use core::{
cmp::min,
sync::atomic::{AtomicBool, AtomicUsize, Ordering},
};
/// Coordination using CAS atomics
pub struct AtomicCoord {
/// Where the next byte will be written
write: AtomicUsize,
/// Where the next byte will be read from
read: AtomicUsize,
/// Used in the inverted case to mark the end of the
/// readable streak. Otherwise will == sizeof::<self.buf>().
/// Writer is responsible for placing this at the correct
/// place when entering an inverted condition, and Reader
/// is responsible for moving it back to sizeof::<self.buf>()
/// when exiting the inverted condition
last: AtomicUsize,
/// Used by the Writer to remember what bytes are currently
/// allowed to be written to, but are not yet ready to be
/// read from
reserve: AtomicUsize,
/// Is there an active read grant?
read_in_progress: AtomicBool,
/// Is there an active write grant?
write_in_progress: AtomicBool,
}
impl AtomicCoord {
/// Create a new coordination structure that uses atomic CAS operations
pub const fn new() -> Self {
Self {
write: AtomicUsize::new(0),
read: AtomicUsize::new(0),
last: AtomicUsize::new(0),
reserve: AtomicUsize::new(0),
read_in_progress: AtomicBool::new(false),
write_in_progress: AtomicBool::new(false),
}
}
}
impl Default for AtomicCoord {
fn default() -> Self {
Self::new()
}
}
impl ConstInit for AtomicCoord {
#[allow(clippy::declare_interior_mutable_const)]
const INIT: Self = Self::new();
}
unsafe impl Coord for AtomicCoord {
fn reset(&self) {
// Re-initialize the buffer (not totally needed, but nice to do)
self.write.store(0, Ordering::Release);
self.read.store(0, Ordering::Release);
self.reserve.store(0, Ordering::Release);
self.last.store(0, Ordering::Release);
}
fn grant_max_remaining(
&self,
capacity: usize,
mut sz: usize,
) -> Result<(usize, usize), WriteGrantError> {
if self.write_in_progress.swap(true, Ordering::AcqRel) {
return Err(WriteGrantError::GrantInProgress);
}
// Writer component. Must never write to `read`,
// be careful writing to `load`
let write = self.write.load(Ordering::Acquire);
let read = self.read.load(Ordering::Acquire);
let max = capacity;
let already_inverted = write < read;
let start = if already_inverted {
// In inverted case, read is always > write
let remain = read - write - 1;
if remain != 0 {
sz = min(remain, sz);
write
} else {
// Inverted, no room is available
self.write_in_progress.store(false, Ordering::Release);
return Err(WriteGrantError::InsufficientSize);
}
} else {
#[allow(clippy::collapsible_if)]
if write != max {
// Some (or all) room remaining in un-inverted case
sz = min(max - write, sz);
write
} else {
// Not inverted, but need to go inverted
// NOTE: We check read > 1, NOT read >= 1, because
// write must never == read in an inverted condition, since
// we will then not be able to tell if we are inverted or not
if read > 1 {
sz = min(read - 1, sz);
0
} else {
// Not invertible, no space
self.write_in_progress.store(false, Ordering::Release);
return Err(WriteGrantError::InsufficientSize);
}
}
};
// Safe write, only viewed by this task
self.reserve.store(start + sz, Ordering::Release);
Ok((start, sz))
}
fn grant_exact(&self, capacity: usize, sz: usize) -> Result<usize, WriteGrantError> {
if self.write_in_progress.swap(true, Ordering::AcqRel) {
return Err(WriteGrantError::GrantInProgress);
}
// Writer component. Must never write to `read`,
// be careful writing to `load`
let write = self.write.load(Ordering::Acquire);
let read = self.read.load(Ordering::Acquire);
let max = capacity;
let already_inverted = write < read;
let start = if already_inverted {
if (write + sz) < read {
// Inverted, room is still available
write
} else {
// Inverted, no room is available
self.write_in_progress.store(false, Ordering::Release);
return Err(WriteGrantError::InsufficientSize);
}
} else {
#[allow(clippy::collapsible_if)]
if write + sz <= max {
// Non inverted condition
write
} else {
// Not inverted, but need to go inverted
// NOTE: We check sz < read, NOT <=, because
// write must never == read in an inverted condition, since
// we will then not be able to tell if we are inverted or not
if sz < read {
// Invertible situation
0
} else {
// Not invertible, no space
self.write_in_progress.store(false, Ordering::Release);
return Err(WriteGrantError::InsufficientSize);
}
}
};
// Safe write, only viewed by this task
self.reserve.store(start + sz, Ordering::Release);
Ok(start)
}
fn read(&self) -> Result<(usize, usize), ReadGrantError> {
if self.read_in_progress.swap(true, Ordering::AcqRel) {
return Err(ReadGrantError::GrantInProgress);
}
let write = self.write.load(Ordering::Acquire);
let last = self.last.load(Ordering::Acquire);
let mut read = self.read.load(Ordering::Acquire);
// Resolve the inverted case or end of read
if (read == last) && (write < read) {
read = 0;
// This has some room for error, the other thread reads this
// Impact to Grant:
// Grant checks if read < write to see if inverted. If not inverted, but
// no space left, Grant will initiate an inversion, but will not trigger it
// Impact to Commit:
// Commit does not check read, but if Grant has started an inversion,
// grant could move Last to the prior write position
// MOVING READ BACKWARDS!
self.read.store(0, Ordering::Release);
}
let sz = if write < read {
// Inverted, only believe last
last
} else {
// Not inverted, only believe write
write
} - read;
if sz == 0 {
self.read_in_progress.store(false, Ordering::Release);
return Err(ReadGrantError::Empty);
}
Ok((read, sz))
}
fn commit_inner(&self, capacity: usize, grant_len: usize, used: usize) {
// If there is no grant in progress, return early. This
// generally means we are dropping the grant within a
// wrapper structure
if !self.write_in_progress.load(Ordering::Acquire) {
return;
}
// Writer component. Must never write to READ,
// be careful writing to LAST
// Saturate the grant commit
let len = grant_len;
let used = min(len, used);
let write = self.write.load(Ordering::Acquire);
self.reserve.fetch_sub(len - used, Ordering::AcqRel);
let max = capacity;
let last = self.last.load(Ordering::Acquire);
let new_write = self.reserve.load(Ordering::Acquire);
if (new_write < write) && (write != max) {
// We have already wrapped, but we are skipping some bytes at the end of the ring.
// Mark `last` where the write pointer used to be to hold the line here
self.last.store(write, Ordering::Release);
} else if new_write > last {
// We're about to pass the last pointer, which was previously the artificial
// end of the ring. Now that we've passed it, we can "unlock" the section
// that was previously skipped.
//
// Since new_write is strictly larger than last, it is safe to move this as
// the other thread will still be halted by the (about to be updated) write
// value
self.last.store(max, Ordering::Release);
}
// else: If new_write == last, either:
// * last == max, so no need to write, OR
// * If we write in the end chunk again, we'll update last to max next time
// * If we write to the start chunk in a wrap, we'll update last when we
// move write backwards
// Write must be updated AFTER last, otherwise read could think it was
// time to invert early!
self.write.store(new_write, Ordering::Release);
// Allow subsequent grants
self.write_in_progress.store(false, Ordering::Release);
}
fn release_inner(&self, used: usize) {
// If there is no grant in progress, return early. This
// generally means we are dropping the grant within a
// wrapper structure
if !self.read_in_progress.load(Ordering::Acquire) {
return;
}
// // This should always be checked by the public interfaces
// debug_assert!(used <= self.buf.len());
// This should be fine, purely incrementing
let _ = self.read.fetch_add(used, Ordering::Release);
self.read_in_progress.store(false, Ordering::Release);
}
}