-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathdeque.zig
More file actions
302 lines (258 loc) · 11.4 KB
/
Copy pathdeque.zig
File metadata and controls
302 lines (258 loc) · 11.4 KB
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
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
const std = @import("std");
const Allocator = std.mem.Allocator;
const Atomic = std.atomic.Value;
const Backoff = @import("backoff").Backoff;
/// Deque - A Bounded, High-Performance Work-Stealing Deque
///
/// A bounded work-stealing deque based on the Chase-Lev algorithm.
/// NOTE: This is a fixed-capacity variant without dynamic resizing.
///
/// Architecture:
/// - ONE owner thread: push() and pop() (LIFO stack for cache locality)
/// - MANY thief threads: steal() (FIFO queue from thieves' perspective)
///
/// Performance characteristics:
/// - Owner push: O(1), ~2-5ns (no CAS, just release store)
/// - Owner pop: O(1), ~3-10ns (CAS only for last item race)
/// - Thief steal: O(1), ~20-50ns (acquire loads + CAS)
///
/// Memory model:
/// - Cache-line aligned head/tail to prevent false sharing
/// - Fixed-size ring buffer (returns error.Full when capacity reached)
/// - Uses signed i64 indices for correct size calculations across wraparound
///
/// Usage:
/// ```zig
/// const result = try Deque(Task).init(allocator, 1024);
/// defer result.worker.deinit();
///
/// // Owner thread:
/// try result.worker.push(task);
/// if (result.worker.pop()) |task| { ... }
///
/// // Thief threads (pass result.stealer to other threads):
/// if (stealer.steal()) |task| { ... }
/// ```
pub fn Deque(comptime T: type) type {
// Compile-time validation: enforce pointer types for large T
comptime {
const type_info = @typeInfo(T);
const is_pointer = switch (type_info) {
.pointer => true,
else => false,
};
const size = @sizeOf(T);
if (size > std.atomic.cache_line and !is_pointer) {
@compileError(std.fmt.comptimePrint(
"Deque: Type '{s}' ({d} bytes) exceeds cache line size ({d} bytes). " ++
"Use *{s} instead for better performance and to avoid false sharing.",
.{ @typeName(T), size, std.atomic.cache_line, @typeName(T) }
));
}
}
return struct {
const Self = @This();
/// Internal deque state (not for direct use)
const DequeStore = struct {
/// Index of the oldest item (steal target)
/// Cache-line aligned to isolate thief contention
/// Uses i64 for proper signed comparison semantics
head: Atomic(i64) align(std.atomic.cache_line),
/// Padding to prevent false sharing between head and tail
/// This is critical for performance - keeps owner and thief operations
/// on separate cache lines
_padding: [std.atomic.cache_line - @sizeOf(Atomic(i64))]u8,
/// Index of the next available slot for push
/// Cache-line aligned, only modified by owner thread
/// Uses i64 for proper signed comparison semantics
tail: Atomic(i64) align(std.atomic.cache_line),
/// Fixed-size ring buffer for storing items
buffer: []T,
/// Capacity - 1, for fast modulo via bitwise AND
mask: usize,
/// Allocator for cleanup
allocator: Allocator,
};
/// Owner thread handle - grants access to push() and pop()
pub const Worker = struct {
deque: *DequeStore,
/// Push item to bottom (owner thread only)
///
/// This is the fast path - no CAS operations, just loads and a release store.
/// Returns error.Full when the bounded capacity is reached.
///
/// Memory ordering:
/// - tail load: .monotonic (only we write to it)
/// - head load: .acquire (synchronize with thieves' steals)
/// - tail store: .release (publish the write to buffer)
pub inline fn push(self: *Worker, item: T) !void {
// 1. Read our private tail (monotonic is sufficient)
const tail = self.deque.tail.load(.monotonic);
// 2. Read head with acquire to see latest steals
const head = self.deque.head.load(.acquire);
// 3. Check if buffer is full
if (tail - head >= @as(i64, @intCast(self.deque.buffer.len))) {
return error.Full;
}
// 4. Write item to buffer (non-atomic, we own this slot)
const index: usize = @intCast(tail);
self.deque.buffer[index & self.deque.mask] = item;
// 5. Publish the write with release ordering
// This ensures the item write is visible before tail update
self.deque.tail.store(tail + 1, .release);
}
/// Pop item from bottom (owner thread only)
///
/// Returns null if deque is empty.
/// Uses CAS only when racing with thieves for the last item.
///
/// Memory ordering:
/// - tail load: .monotonic (only we write to it)
/// - tail store: .release (publish the decrement)
/// - head load: .acquire (synchronize with thieves)
/// - CAS: .seq_cst (resolve multi-party race for last item)
pub inline fn pop(self: *Worker) ?T {
// 1. Read our private tail
var tail = self.deque.tail.load(.monotonic);
// 2. Speculatively decrement tail
tail -= 1;
// 3. Store decremented tail with seq_cst (matches paper)
self.deque.tail.store(tail, .seq_cst);
// 4. Read head with seq_cst to synchronize
const head = self.deque.head.load(.seq_cst);
// 5. Check state
if (tail < head) {
// Empty - restore tail and return null
self.deque.tail.store(tail + 1, .monotonic);
return null;
}
// Read the item
const index: usize = @intCast(tail);
const item = self.deque.buffer[index & self.deque.mask];
if (tail > head) {
// More than one item - we're done
return item;
}
// Exactly one item left - race with thieves
// Restore tail BEFORE the CAS (critical for correctness!)
self.deque.tail.store(tail + 1, .seq_cst);
// Try to increment head to claim the item
if (self.deque.head.cmpxchgStrong(head, head + 1, .seq_cst, .monotonic)) |_| {
// CAS failed - thief won
return null;
}
// CAS succeeded - we won
return item;
}
/// Cleanup the deque (consumes the worker handle)
pub fn deinit(self: *Worker) void {
const allocator = self.deque.allocator;
allocator.free(self.deque.buffer);
allocator.destroy(self.deque);
}
/// Get approximate size (racy, for debugging)
pub fn size(self: *const Worker) usize {
const head = self.deque.head.load(.monotonic);
const tail = self.deque.tail.load(.monotonic);
const diff = tail - head;
if (diff <= 0) return 0;
return @intCast(diff);
}
/// Check if empty (racy, for debugging)
pub fn isEmpty(self: *const Worker) bool {
return self.size() == 0;
}
/// Get capacity
pub fn capacity(self: *const Worker) usize {
return self.deque.buffer.len;
}
};
/// Thief thread handle - grants access to steal()
pub const Stealer = struct {
deque: *DequeStore,
/// Steal item from top (thief threads)
///
/// Thread-safe, can be called from any thread.
/// Returns null if deque is empty or if lost race with another thief.
///
/// Uses exponential backoff when CAS fails to reduce contention between thieves.
///
/// Memory ordering:
/// - head/tail load: .acquire (synchronize with owner's pushes)
/// - CAS: .acq_rel (success) / .acquire (failure)
pub fn steal(self: *Stealer) ?T {
var backoff = Backoff.init(.{});
while (true) {
// 1. Read head first with acquire ordering
const head = self.deque.head.load(.acquire);
// 2. Read tail with acquire ordering
const tail = self.deque.tail.load(.acquire);
// 3. Check if empty
if (head >= tail) {
return null;
}
// 4. Speculatively read the item
const index: usize = @intCast(head);
const item = self.deque.buffer[index & self.deque.mask];
// 5. Attempt to claim the item with CAS (seq_cst for correctness)
if (self.deque.head.cmpxchgWeak(head, head + 1, .seq_cst, .monotonic)) |_| {
// CAS failed, another thief won - backoff before retry
// This reduces cache coherency traffic under contention
backoff.snooze();
continue;
} else {
// CAS succeeded, we won the race
return item;
}
}
}
/// Get approximate size (racy, for debugging)
pub fn size(self: *const Stealer) usize {
const head = self.deque.head.load(.monotonic);
const tail = self.deque.tail.load(.monotonic);
const diff = tail - head;
if (diff <= 0) return 0;
return @intCast(diff);
}
/// Check if empty (racy, for debugging)
pub fn isEmpty(self: *const Stealer) bool {
return self.size() == 0;
}
};
/// Initialization result containing both handles
pub const InitResult = struct {
worker: Worker,
stealer: Stealer,
};
/// Initialize a new bounded deque
///
/// Capacity must be a power of two (required for fast & mask modulo).
/// Returns error.CapacityNotPowerOfTwo if capacity is not a power of two.
/// Returns error.OutOfMemory if allocation fails.
pub fn init(allocator: Allocator, capacity: usize) !InitResult {
// Validate power of two
if (!std.math.isPowerOfTwo(capacity)) {
return error.CapacityNotPowerOfTwo;
}
// Allocate deque struct
const deque = try allocator.create(DequeStore);
errdefer allocator.destroy(deque);
// Allocate buffer
const buffer = try allocator.alloc(T, capacity);
errdefer allocator.free(buffer);
// Initialize deque
deque.* = DequeStore{
.head = Atomic(i64).init(0),
._padding = undefined,
.tail = Atomic(i64).init(0),
.buffer = buffer,
.mask = capacity - 1,
.allocator = allocator,
};
return InitResult{
.worker = Worker{ .deque = deque },
.stealer = Stealer{ .deque = deque },
};
}
};
}