lib/gpalloc/src/properties/concurrency.zig
daab053ee43316e1809a84551d573ddd1e5bf3d2
1 const std = @import("std");
2 const hypothesis = @import("hypothesis");
3 const gpalloc = @import("gpalloc");
4 const sys = @import("sys");
5
6 const allocator_properties = @import("allocator.zig");
7 const draw = @import("draw.zig");
8
9 const Allocator = std.mem.Allocator;
10 const GpAllocator = gpalloc.GpAllocator;
11 const max_small_size = gpalloc.max_small_size;
12 const drawUsize = draw.drawUsize;
13
14 const failed_sentinel: usize = 1;
15
16 const Record = struct {
17 bytes: []u8,
18 seed: u64,
19 };
20
21 fn patternByte(seed: u64, index: usize) u8 {
22 return @truncate(seed +% index *% 33 +% (index >> 3));
23 }
24
25 fn fillPattern(bytes: []u8, seed: u64) void {
26 const span = @min(bytes.len, 256);
27 for (bytes[0..span], 0..) |*byte, index| byte.* = patternByte(seed, index);
28 bytes[bytes.len - 1] = patternByte(seed, bytes.len - 1);
29 }
30
31 fn checkPattern(bytes: []const u8, seed: u64) bool {
32 const span = @min(bytes.len, 256);
33 for (bytes[0..span], 0..) |byte, index| {
34 if (byte != patternByte(seed, index)) return false;
35 }
36 return bytes[bytes.len - 1] == patternByte(seed, bytes.len - 1);
37 }
38
39 fn randomLen(random: std.Random, include_large: bool) usize {
40 const variant = random.uintLessThan(u8, 10);
41 if (include_large and variant == 9) {
42 return random.intRangeAtMost(usize, max_small_size + 1, 128 * 1024);
43 }
44 if (variant >= 7) {
45 return random.intRangeAtMost(usize, 1025, max_small_size);
46 }
47 return random.intRangeAtMost(usize, 1, 1024);
48 }
49
50 fn reportFailure(failures: *std.atomic.Value(u32)) void {
51 _ = failures.fetchAdd(1, .monotonic);
52 }
53
54 fn checkRecord(record: Record, failures: *std.atomic.Value(u32)) void {
55 if (!checkPattern(record.bytes, record.seed)) reportFailure(failures);
56 }
57
58 const HandoffPair = struct {
59 heap_allocator: Allocator,
60 len: usize,
61 seed: u64,
62 items: usize,
63 slots: []std.atomic.Value(usize),
64 failures: *std.atomic.Value(u32),
65 };
66
67 pub const HandoffProperty = struct {
68 const max_pairs = 2;
69 const max_items = 96;
70
71 fn produce(pair: *const HandoffPair) void {
72 for (pair.slots[0..pair.items], 0..) |*slot, index| {
73 const bytes = pair.heap_allocator.alloc(u8, pair.len) catch {
74 reportFailure(pair.failures);
75 slot.store(failed_sentinel, .release);
76 continue;
77 };
78 fillPattern(bytes, pair.seed +% index);
79 slot.store(@intFromPtr(bytes.ptr), .release);
80 }
81 }
82
83 fn consume(pair: *const HandoffPair) void {
84 for (pair.slots[0..pair.items], 0..) |*slot, index| {
85 var value = slot.load(.acquire);
86 while (value == 0) {
87 sys.thread.yield();
88 value = slot.load(.acquire);
89 }
90 if (value == failed_sentinel) continue;
91 const ptr: [*]u8 = @ptrFromInt(value);
92 const bytes = ptr[0..pair.len];
93 checkRecord(.{ .bytes = bytes, .seed = pair.seed +% index }, pair.failures);
94 pair.heap_allocator.free(bytes);
95 }
96 }
97
98 fn drawLen(conjecture: *hypothesis.ConjectureData) !usize {
99 const variant = try drawUsize(conjecture, 0, 9, 0);
100 return switch (variant) {
101 0...6 => try drawUsize(conjecture, 1, 1024, 16),
102 7...8 => try drawUsize(conjecture, 1025, max_small_size, 1025),
103 else => try drawUsize(conjecture, max_small_size + 1, 128 * 1024, max_small_size + 1),
104 };
105 }
106
107 pub fn property(
108 conjecture: *hypothesis.ConjectureData,
109 property_allocator: Allocator,
110 ) !void {
111 var heap = GpAllocator.init(property_allocator, .{});
112 defer heap.deinit();
113 const heap_allocator = heap.allocator();
114
115 var failures = std.atomic.Value(u32).init(0);
116 var slot_storage: [max_pairs][max_items]std.atomic.Value(usize) = undefined;
117 var pairs: [max_pairs]HandoffPair = undefined;
118
119 const pair_count = try drawUsize(conjecture, 1, max_pairs, 1);
120 for (0..pair_count) |pair_index| {
121 for (&slot_storage[pair_index]) |*slot| slot.* = .init(0);
122 pairs[pair_index] = .{
123 .heap_allocator = heap_allocator,
124 .len = try drawLen(conjecture),
125 .seed = 0x9e3779b97f4a7c15 +% @as(u64, pair_index) *% 0x100,
126 .items = try drawUsize(conjecture, 4, max_items, 8),
127 .slots = &slot_storage[pair_index],
128 .failures = &failures,
129 };
130 }
131
132 var handles: [max_pairs * 2]sys.thread.JoinHandle = undefined;
133 var spawned: usize = 0;
134 {
135 errdefer for (handles[0..spawned]) |handle| handle.join();
136 for (pairs[0..pair_count]) |*pair| {
137 handles[spawned] = try sys.thread.spawn(produce, .{pair});
138 spawned += 1;
139 handles[spawned] = try sys.thread.spawn(consume, .{pair});
140 spawned += 1;
141 }
142 }
143 for (handles[0..spawned]) |handle| handle.join();
144
145 if (failures.load(.monotonic) != 0) return error.PropertyFailed;
146 }
147 };
148
149 const ChurnWork = struct {
150 heap_allocator: Allocator,
151 prng_seed: u64,
152 rounds: usize,
153 slots: usize,
154 survivors: []?Record,
155 failures: *std.atomic.Value(u32),
156 };
157
158 pub const ChurnProperty = struct {
159 const max_threads = 4;
160 const max_slots = 32;
161
162 fn churn(work: *const ChurnWork) void {
163 var prng = std.Random.DefaultPrng.init(work.prng_seed);
164 const random = prng.random();
165 var active = @as([max_slots]?Record, @splat(null));
166 var round: usize = 0;
167 while (round < work.rounds) : (round += 1) {
168 for (active[0..work.slots]) |*slot| {
169 if (slot.*) |record| {
170 checkRecord(record, work.failures);
171 if (random.boolean()) {
172 work.heap_allocator.free(record.bytes);
173 slot.* = null;
174 }
175 } else {
176 const len = randomLen(random, true);
177 const bytes = work.heap_allocator.alloc(u8, len) catch {
178 reportFailure(work.failures);
179 continue;
180 };
181 const seed = random.int(u64);
182 fillPattern(bytes, seed);
183 slot.* = .{ .bytes = bytes, .seed = seed };
184 }
185 }
186 }
187 for (active[0..work.slots], work.survivors[0..work.slots]) |slot, *survivor| {
188 survivor.* = slot;
189 }
190 }
191
192 pub fn property(
193 conjecture: *hypothesis.ConjectureData,
194 property_allocator: Allocator,
195 ) !void {
196 var heap = GpAllocator.init(property_allocator, .{});
197 defer heap.deinit();
198 const heap_allocator = heap.allocator();
199
200 var failures = std.atomic.Value(u32).init(0);
201 var survivor_storage: [max_threads][max_slots]?Record =
202 @as([max_threads][max_slots]?Record, @splat(@as([max_slots]?Record, @splat(null))));
203 var work_items: [max_threads]ChurnWork = undefined;
204
205 const thread_count = try drawUsize(conjecture, 2, max_threads, 2);
206 const base_seed = try drawUsize(conjecture, 0, 1 << 32, 0);
207 for (0..thread_count) |thread_index| {
208 work_items[thread_index] = .{
209 .heap_allocator = heap_allocator,
210 .prng_seed = @as(u64, base_seed) +% thread_index,
211 .rounds = try drawUsize(conjecture, 1, 6, 2),
212 .slots = try drawUsize(conjecture, 4, max_slots, 8),
213 .survivors = &survivor_storage[thread_index],
214 .failures = &failures,
215 };
216 }
217
218 var handles: [max_threads]sys.thread.JoinHandle = undefined;
219 var spawned: usize = 0;
220 {
221 errdefer for (handles[0..spawned]) |handle| handle.join();
222 for (work_items[0..thread_count]) |*work| {
223 handles[spawned] = try sys.thread.spawn(churn, .{work});
224 spawned += 1;
225 }
226 }
227 for (handles[0..spawned]) |handle| handle.join();
228
229 for (survivor_storage[0..thread_count]) |*row| {
230 for (row) |*slot| {
231 if (slot.*) |record| {
232 checkRecord(record, &failures);
233 heap_allocator.free(record.bytes);
234 slot.* = null;
235 }
236 }
237 }
238
239 if (failures.load(.monotonic) != 0) return error.PropertyFailed;
240 }
241 };
242
243 const LifecycleWork = struct {
244 heap_allocator: Allocator,
245 prng_seed: u64,
246 burst: usize,
247 adopt: []?Record,
248 publish: []?Record,
249 failures: *std.atomic.Value(u32),
250 };
251
252 pub const LifecycleProperty = struct {
253 const max_waves = 3;
254 const max_workers = 3;
255 const max_burst = 48;
256 const publish_capacity = max_burst / 2 + 1;
257
258 fn wave(work: *const LifecycleWork) void {
259 for (work.adopt) |*entry| {
260 if (entry.*) |record| {
261 checkRecord(record, work.failures);
262 work.heap_allocator.free(record.bytes);
263 entry.* = null;
264 }
265 }
266 var prng = std.Random.DefaultPrng.init(work.prng_seed);
267 const random = prng.random();
268 var index: usize = 0;
269 while (index < work.burst) : (index += 1) {
270 const len = randomLen(random, false);
271 const bytes = work.heap_allocator.alloc(u8, len) catch {
272 reportFailure(work.failures);
273 continue;
274 };
275 const seed = random.int(u64);
276 fillPattern(bytes, seed);
277 if (index % 2 == 0) {
278 work.publish[index / 2] = .{ .bytes = bytes, .seed = seed };
279 } else {
280 checkRecord(.{ .bytes = bytes, .seed = seed }, work.failures);
281 work.heap_allocator.free(bytes);
282 }
283 }
284 }
285
286 pub fn property(
287 conjecture: *hypothesis.ConjectureData,
288 property_allocator: Allocator,
289 ) !void {
290 var heap = GpAllocator.init(property_allocator, .{
291 .thread_cache_active_limit = try drawUsize(conjecture, 0, 2, 0),
292 .thread_cache_retention_limit = try drawUsize(conjecture, 0, 2, 1),
293 });
294 defer heap.deinit();
295 const heap_allocator = heap.allocator();
296
297 var failures = std.atomic.Value(u32).init(0);
298 var publish_storage: [max_waves][max_workers][publish_capacity]?Record =
299 @as([max_waves][max_workers][publish_capacity]?Record, @splat(@as([max_workers][publish_capacity]?Record, @splat(@as([publish_capacity]?Record, @splat(null))))));
300 var work_storage: [max_workers]LifecycleWork = undefined;
301
302 const wave_count = try drawUsize(conjecture, 1, max_waves, 1);
303 const base_seed = try drawUsize(conjecture, 0, 1 << 32, 0);
304 var previous_workers: usize = 0;
305 for (0..wave_count) |wave_index| {
306 const worker_count = try drawUsize(conjecture, 1, max_workers, 1);
307 var handles: [max_workers]sys.thread.JoinHandle = undefined;
308 var spawned: usize = 0;
309 errdefer for (handles[0..spawned]) |handle| handle.join();
310 for (0..worker_count) |worker_index| {
311 const adopt: []?Record = if (wave_index > 0 and worker_index < previous_workers)
312 &publish_storage[wave_index - 1][worker_index]
313 else
314 &.{};
315 work_storage[worker_index] = .{
316 .heap_allocator = heap_allocator,
317 .prng_seed = @as(u64, base_seed) +% wave_index *% 31 +% worker_index,
318 .burst = try drawUsize(conjecture, 4, max_burst, 8),
319 .adopt = adopt,
320 .publish = &publish_storage[wave_index][worker_index],
321 .failures = &failures,
322 };
323 handles[spawned] = try sys.thread.spawn(wave, .{&work_storage[worker_index]});
324 spawned += 1;
325 }
326 for (handles[0..spawned]) |handle| handle.join();
327 previous_workers = worker_count;
328 }
329
330 for (&publish_storage) |*wave_rows| {
331 for (wave_rows) |*worker_row| {
332 for (worker_row) |*slot| {
333 if (slot.*) |record| {
334 checkRecord(record, &failures);
335 heap_allocator.free(record.bytes);
336 slot.* = null;
337 }
338 }
339 }
340 }
341
342 if (failures.load(.monotonic) != 0) return error.PropertyFailed;
343 }
344 };
345
346 const StatsTotals = struct {
347 small_allocations: u64 = 0,
348 small_frees: u64 = 0,
349 large_allocations: u64 = 0,
350 large_frees: u64 = 0,
351 };
352
353 const StatsWork = struct {
354 heap_allocator: Allocator,
355 prng_seed: u64,
356 ops: usize,
357 totals: *StatsTotals,
358 failures: *std.atomic.Value(u32),
359 };
360
361 pub const StatsProperty = struct {
362 const max_threads = 4;
363 const max_ops = 48;
364 const live_limit = 8;
365
366 fn recordAlloc(totals: *StatsTotals, len: usize) void {
367 if (len <= max_small_size) {
368 totals.small_allocations += 1;
369 } else {
370 totals.large_allocations += 1;
371 }
372 }
373
374 fn recordFree(totals: *StatsTotals, len: usize) void {
375 if (len <= max_small_size) {
376 totals.small_frees += 1;
377 } else {
378 totals.large_frees += 1;
379 }
380 }
381
382 fn exercise(work: *const StatsWork) void {
383 var prng = std.Random.DefaultPrng.init(work.prng_seed);
384 const random = prng.random();
385 var active = @as([live_limit]?Record, @splat(null));
386 var totals: StatsTotals = .{};
387 var op: usize = 0;
388 while (op < work.ops) : (op += 1) {
389 const slot = &active[random.uintLessThan(usize, live_limit)];
390 if (slot.*) |record| {
391 checkRecord(record, work.failures);
392 recordFree(&totals, record.bytes.len);
393 work.heap_allocator.free(record.bytes);
394 slot.* = null;
395 } else {
396 const len = randomLen(random, true);
397 const bytes = work.heap_allocator.alloc(u8, len) catch {
398 reportFailure(work.failures);
399 continue;
400 };
401 recordAlloc(&totals, len);
402 const seed = random.int(u64);
403 fillPattern(bytes, seed);
404 slot.* = .{ .bytes = bytes, .seed = seed };
405 }
406 }
407 for (&active) |*slot| {
408 if (slot.*) |record| {
409 checkRecord(record, work.failures);
410 recordFree(&totals, record.bytes.len);
411 work.heap_allocator.free(record.bytes);
412 slot.* = null;
413 }
414 }
415 work.totals.* = totals;
416 }
417
418 pub fn property(
419 conjecture: *hypothesis.ConjectureData,
420 property_allocator: Allocator,
421 ) !void {
422 var heap = GpAllocator.init(property_allocator, .{ .collect_stats = true });
423 defer heap.deinit();
424 const heap_allocator = heap.allocator();
425
426 var failures = std.atomic.Value(u32).init(0);
427 var totals_storage = @as([max_threads]StatsTotals, @splat(.{}));
428 var work_items: [max_threads]StatsWork = undefined;
429
430 const thread_count = try drawUsize(conjecture, 2, max_threads, 2);
431 const base_seed = try drawUsize(conjecture, 0, 1 << 32, 0);
432 for (0..thread_count) |thread_index| {
433 work_items[thread_index] = .{
434 .heap_allocator = heap_allocator,
435 .prng_seed = @as(u64, base_seed) +% thread_index *% 977,
436 .ops = try drawUsize(conjecture, 8, max_ops, 8),
437 .totals = &totals_storage[thread_index],
438 .failures = &failures,
439 };
440 }
441
442 var handles: [max_threads]sys.thread.JoinHandle = undefined;
443 var spawned: usize = 0;
444 {
445 errdefer for (handles[0..spawned]) |handle| handle.join();
446 for (work_items[0..thread_count]) |*work| {
447 handles[spawned] = try sys.thread.spawn(exercise, .{work});
448 spawned += 1;
449 }
450 }
451 for (handles[0..spawned]) |handle| handle.join();
452
453 var expected: StatsTotals = .{};
454 for (totals_storage[0..thread_count]) |totals| {
455 expected.small_allocations += totals.small_allocations;
456 expected.small_frees += totals.small_frees;
457 expected.large_allocations += totals.large_allocations;
458 expected.large_frees += totals.large_frees;
459 }
460
461 const actual = heap.stats();
462 try std.testing.expectEqual(expected.small_allocations, actual.small_allocations);
463 try std.testing.expectEqual(expected.small_frees, actual.small_frees);
464 try std.testing.expectEqual(expected.large_allocations, actual.large_allocations);
465 try std.testing.expectEqual(expected.large_frees, actual.large_frees);
466 try std.testing.expectEqual(@as(usize, 0), actual.active_small_bytes);
467 try std.testing.expectEqual(@as(usize, 0), actual.active_large_bytes);
468
469 if (failures.load(.monotonic) != 0) return error.PropertyFailed;
470 }
471 };
472
473 test "property: concurrent handoff preserves remote-freed block contents" {
474 try hypothesis.checkNamed(
475 HandoffProperty,
476 "gpalloc-concurrent-handoff",
477 allocator_properties.settings(),
478 );
479 }
480
481 test "property: concurrent churn survives thread exit reclamation" {
482 try hypothesis.checkNamed(
483 ChurnProperty,
484 "gpalloc-concurrent-churn",
485 allocator_properties.settings(),
486 );
487 }
488
489 test "property: thread lifecycle waves preserve blocks under cache caps" {
490 try hypothesis.checkNamed(
491 LifecycleProperty,
492 "gpalloc-concurrent-lifecycle",
493 allocator_properties.settings(),
494 );
495 }
496
497 test "property: concurrent stats reconcile with per-thread totals" {
498 try hypothesis.checkNamed(
499 StatsProperty,
500 "gpalloc-concurrent-stats",
501 allocator_properties.settings(),
502 );
503 }