Something went wrong. Try again.
fork of gitlab.com/devnw/zig/thread-pool for zig 0.17
Something went wrong. Try again.
81 kB · 2014 lines
Zig
at main
12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103110411051106110711081109111011111112111311141115111611171118111911201121112211231124112511261127112811291130113111321133113411351136113711381139114011411142114311441145114611471148114911501151115211531154115511561157115811591160116111621163116411651166116711681169117011711172117311741175117611771178117911801181118211831184118511861187118811891190119111921193119411951196119711981199120012011202120312041205120612071208120912101211121212131214121512161217121812191220122112221223122412251226122712281229123012311232123312341235123612371238123912401241124212431244124512461247124812491250125112521253125412551256125712581259126012611262126312641265126612671268126912701271127212731274127512761277127812791280128112821283128412851286128712881289129012911292129312941295129612971298129913001301130213031304130513061307130813091310131113121313131413151316131713181319132013211322132313241325132613271328132913301331133213331334133513361337133813391340134113421343134413451346134713481349135013511352135313541355135613571358135913601361136213631364136513661367136813691370137113721373137413751376137713781379138013811382138313841385138613871388138913901391139213931394139513961397139813991400140114021403140414051406140714081409141014111412141314141415141614171418141914201421142214231424142514261427142814291430143114321433143414351436143714381439144014411442144314441445144614471448144914501451145214531454145514561457145814591460146114621463146414651466146714681469147014711472147314741475147614771478147914801481148214831484148514861487148814891490149114921493149414951496149714981499150015011502150315041505150615071508150915101511151215131514151515161517151815191520152115221523152415251526152715281529153015311532153315341535153615371538153915401541154215431544154515461547154815491550155115521553155415551556155715581559156015611562156315641565156615671568156915701571157215731574157515761577157815791580158115821583158415851586158715881589159015911592159315941595159615971598159916001601160216031604160516061607160816091610161116121613161416151616161716181619162016211622162316241625162616271628162916301631163216331634163516361637163816391640164116421643164416451646164716481649165016511652165316541655165616571658165916601661166216631664166516661667166816691670167116721673167416751676167716781679168016811682168316841685168616871688168916901691169216931694169516961697169816991700170117021703170417051706170717081709171017111712171317141715171617171718171917201721172217231724172517261727172817291730173117321733173417351736173717381739174017411742174317441745174617471748174917501751175217531754175517561757175817591760176117621763176417651766176717681769177017711772177317741775177617771778177917801781178217831784178517861787178817891790179117921793179417951796179717981799180018011802180318041805180618071808180918101811181218131814181518161817181818191820182118221823182418251826182718281829183018311832183318341835183618371838183918401841184218431844184518461847184818491850185118521853185418551856185718581859186018611862186318641865186618671868186918701871187218731874187518761877187818791880188118821883188418851886188718881889189018911892189318941895189618971898189919001901190219031904190519061907190819091910191119121913191419151916191719181919192019211922192319241925192619271928192919301931193219331934193519361937193819391940194119421943194419451946194719481949195019511952195319541955195619571958195919601961196219631964196519661967196819691970197119721973197419751976197719781979198019811982198319841985198619871988198919901991199219931994199519961997199819992000200120022003200420052006200720082009201020112012201320142015const std = @import("std");const thread_pool = @import("thread_pool");const common = @import("common.zig");
test "work-stealing scheduler executes each accepted task exactly once" { var pool = try thread_pool.ThreadPool.init(std.testing.allocator, common.defaultConfig(.work_stealing)); defer pool.deinit();
var wg = pool.waitGroup(); var counter = common.Counter{}; for (0..128) |_| { _ = try pool.submit(common.counterTask, @ptrCast(&counter), .{ .wait_group = &wg }); } wg.wait();
try std.testing.expectEqual(@as(u64, 128), counter.value.load(.acquire)); const stats = pool.statsSnapshot(); try std.testing.expectEqual(@as(u64, 128), stats.completed);}
test "global scheduler waitable handle completes" { var pool = try thread_pool.ThreadPool.init(std.testing.allocator, common.defaultConfig(.global_queue)); defer pool.deinit();
var counter = common.Counter{}; const handle = try pool.submit(common.counterTask, @ptrCast(&counter), .{ .waitable = true }); try pool.wait(handle); try std.testing.expectEqual(@as(u64, 1), counter.value.load(.acquire));}
test "wait rejects out-of-bounds handle index" { var pool = try thread_pool.ThreadPool.init(std.testing.allocator, .{ .min_threads = 1, .max_threads = 1, .queue_capacity = 64, }); defer pool.deinit();
const bad_handle = thread_pool.types.TaskHandle{ .index = 9999, .generation = 1, .waitable = true, }; try std.testing.expectError(error.InvalidHandle, pool.wait(bad_handle));}
test "shutdown rejects new submissions" { var pool = try thread_pool.ThreadPool.init(std.testing.allocator, common.defaultConfig(.global_queue)); defer pool.deinit();
try pool.shutdown(.cancel_pending, null); var counter = common.Counter{}; try std.testing.expectError(error.ShutdownInProgress, pool.submit(common.counterTask, @ptrCast(&counter), .{}));}
test "worker cache purge resets current scratch usage" { var pool = try thread_pool.ThreadPool.init(std.testing.allocator, common.defaultConfig(.work_stealing)); defer pool.deinit();
var wg = pool.waitGroup(); var counter = common.Counter{}; var scratch_ctx = common.ScratchTaskContext{ .counter = &counter, .bytes = 1024 }; for (0..16) |_| { _ = try pool.submit(common.scratchTask, @ptrCast(&scratch_ctx), .{ .wait_group = &wg }); } wg.wait(); try pool.purgeWorkerCaches();
const stats = pool.statsSnapshot(); for (stats.worker_stats) |worker| { try std.testing.expectEqual(@as(u64, 0), worker.scratch_current_bytes); }}
test "worker cache purge rejects queued work even when no task is active" { var pool = try thread_pool.ThreadPool.init(std.testing.allocator, common.defaultConfig(.work_stealing)); defer pool.deinit();
pool.core.counters.queued_tasks.store(1, .release); defer pool.core.counters.queued_tasks.store(0, .release);
try std.testing.expectError(error.PoolBusy, pool.purgeWorkerCaches());}
test "waitable handle generation increments across reuses" { var pool = try thread_pool.ThreadPool.init(std.testing.allocator, common.defaultConfig(.global_queue)); defer pool.deinit();
var counter = common.Counter{}; const handle1 = try pool.submit(common.counterTask, @ptrCast(&counter), .{ .waitable = true }); try pool.wait(handle1);
// Same descriptor index may be reused; generation must differ. const handle2 = try pool.submit(common.counterTask, @ptrCast(&counter), .{ .waitable = true }); try pool.wait(handle2);
if (handle1.index == handle2.index) { try std.testing.expect(handle2.generation > handle1.generation); }}
// SEC-018 / F-008: concurrent double-wait on the same TaskHandle.//// Pre-fix, two threads that raced on the same waitable handle could both// pass the "state == .completed" check and both call releaseDescriptor,// producing a double-push to the freelist and (depending on ABA timing) a// double-free the allocator never sees because the freelist is a bounded// pool. Post-fix, `releaseDescriptor` is CAS-guarded on// `descriptor.state` so exactly one caller performs the release; the other// finds `.free` with a matching generation and returns success.test "SEC-018 concurrent double-wait on same handle is safe" { var pool = try thread_pool.ThreadPool.init(std.testing.allocator, common.defaultConfig(.global_queue)); defer pool.deinit();
const Barrier = struct { pool_ptr: *thread_pool.ThreadPool, handle: thread_pool.types.TaskHandle, start_gate: std.atomic.Value(u32) = std.atomic.Value(u32).init(0), results: [2]anyerror!void = .{ {}, {} },
fn waiter(self: *@This(), slot: usize) void { // Spin until the main thread releases both waiters together so // they race the completed broadcast under the completion_mutex. while (self.start_gate.load(.acquire) == 0) { std.atomic.spinLoopHint(); } self.results[slot] = self.pool_ptr.wait(self.handle); } };
// Run multiple rounds to shake out the race; each round submits a // fresh waitable handle and fires two waiter threads at it. var round: usize = 0; while (round < 64) : (round += 1) { var counter = common.Counter{}; const handle = try pool.submit(common.counterTask, @ptrCast(&counter), .{ .waitable = true });
var barrier = Barrier{ .pool_ptr = &pool, .handle = handle };
const t0 = try std.Thread.spawn(.{}, Barrier.waiter, .{ &barrier, @as(usize, 0) }); const t1 = try std.Thread.spawn(.{}, Barrier.waiter, .{ &barrier, @as(usize, 1) });
// Release both waiters simultaneously. barrier.start_gate.store(1, .release);
t0.join(); t1.join();
// Both waiters must see success. If the CAS guard regressed, one // would return error.InvalidHandle or (worse) the pool would // double-push to the freelist — the DebugAllocator check at // pool.deinit() plus the task_desc_live invariant would catch the // latter on the stats snapshot below. try barrier.results[0]; try barrier.results[1]; try std.testing.expectEqual(@as(u64, 1), counter.value.load(.acquire)); }
// After all rounds, task_desc_live must be zero — no leaked descriptors. const stats = pool.statsSnapshot(); try std.testing.expectEqual(@as(u64, 0), stats.task_desc_live);}
// SEC-018: back-to-back sequential wait() on the same handle must not// panic and must not double-free the descriptor slot. Post-fix, the// second wait() observes `.free` with a matching generation (if the slot// hasn't been reused) and returns success, or observes a generation// mismatch and returns InvalidHandle. Either outcome is acceptable; what// must NOT happen is a panic or a double freelist push.test "SEC-018 sequential double-wait on same handle does not panic" { var pool = try thread_pool.ThreadPool.init(std.testing.allocator, common.defaultConfig(.global_queue)); defer pool.deinit();
var counter = common.Counter{}; const handle = try pool.submit(common.counterTask, @ptrCast(&counter), .{ .waitable = true }); try pool.wait(handle);
// Second wait() is a no-op or InvalidHandle — both are acceptable // post-SEC-018. The key invariant is "no panic, no double-free". pool.wait(handle) catch |err| { try std.testing.expectEqual(@as(anyerror, error.InvalidHandle), err); };
try std.testing.expectEqual(@as(u64, 1), counter.value.load(.acquire)); const stats = pool.statsSnapshot(); try std.testing.expectEqual(@as(u64, 0), stats.task_desc_live);}
test "recursive submission through local queues does not deadlock" { var pool = try thread_pool.ThreadPool.init(std.testing.allocator, common.defaultConfig(.work_stealing)); defer pool.deinit();
var wg = pool.waitGroup(); var counter = common.Counter{}; var recursive = common.RecursiveContext{ .pool = &pool, .counter = &counter, .wait_group = &wg, .child_count = 8, };
_ = try pool.submit(common.recursiveTask, @ptrCast(&recursive), .{ .wait_group = &wg }); wg.wait();
try std.testing.expectEqual(@as(u64, 9), counter.value.load(.acquire));}
test "submitBatch from worker context completes all child tasks" { const allocator = std.testing.allocator; var pool = try thread_pool.ThreadPool.init(allocator, .{ .min_threads = 4, .max_threads = 4, .queue_capacity = 256, .local_queue_capacity = 64, .scheduler_variant = .work_stealing, }); defer pool.deinit();
const CTX = struct { pool: *thread_pool.ThreadPool, wg: *thread_pool.WaitGroup, done: *std.atomic.Value(u64),
fn parent(raw: *anyopaque, _: *thread_pool.TaskContext) void { const self: *@This() = @ptrCast(@alignCast(raw)); var tasks: [4]thread_pool.BatchTask = undefined; for (&tasks) |*t| { t.* = .{ .task_fn = child, .ctx = @ptrCast(self) }; } _ = self.pool.submitBatch(&tasks, .{ .wait_group = self.wg, .backpressure = .block }) catch return; } fn child(raw: *anyopaque, _: *thread_pool.TaskContext) void { const self: *@This() = @ptrCast(@alignCast(raw)); _ = self.done.fetchAdd(1, .monotonic); } };
var done = std.atomic.Value(u64).init(0); var wg = pool.waitGroup(); var ctx = CTX{ .pool = &pool, .wg = &wg, .done = &done }; _ = try pool.submit(CTX.parent, @ptrCast(&ctx), .{ .wait_group = &wg, .backpressure = .block }); wg.wait(); try std.testing.expectEqual(@as(u64, 4), done.load(.acquire));}
test "submitBatch rejects waitable options (Q-NEW-02 descriptor leak guard)" { // submitBatch returns only a count; waitable handles would be discarded // and their descriptors leaked until pool exhaustion (#107). var pool = try thread_pool.ThreadPool.init(std.testing.allocator, .{ .min_threads = 1, .max_threads = 1, .queue_capacity = 8, }); defer pool.deinit();
var counter = common.Counter{}; var batch: [2]thread_pool.BatchTask = undefined; for (&batch) |*t| { t.* = .{ .task_fn = common.counterTask, .ctx = @ptrCast(&counter) }; }
try std.testing.expectError( error.WaitableUnsupported, pool.submitBatch(&batch, .{ .waitable = true }), );
// The rejected batch must not have enqueued anything: nothing runs. try std.testing.expectEqual(@as(u64, 0), counter.value.load(.acquire));}
test "pre-cancelled token prevents task body execution" { var pool = try thread_pool.ThreadPool.init(std.testing.allocator, .{ .min_threads = 1, .max_threads = 1, .queue_capacity = 64, }); defer pool.deinit();
var token = thread_pool.CancellationToken{}; // Cancel BEFORE submitting -- the task body must never run. token.cancel(); try std.testing.expect(token.isCancelled());
var wg = pool.waitGroup(); var counter = common.Counter{}; _ = try pool.submit(common.counterTask, @ptrCast(&counter), .{ .wait_group = &wg, .cancellation_token = &token, }); wg.wait();
// Task body was NOT executed; counter stays 0. try std.testing.expectEqual(@as(u64, 0), counter.value.load(.acquire));
// Stats must reflect the cancellation. const stats = pool.statsSnapshot(); try std.testing.expect(stats.cancelled > 0);}
test "trySubmit returns error when queue is full" { // Tiny queue + 1 worker + sleep tasks = guaranteed full queue. var pool = try thread_pool.ThreadPool.init(std.testing.allocator, .{ .min_threads = 1, .max_threads = 1, .queue_capacity = 4, .scheduler_variant = .global_queue, }); defer pool.deinit();
var wg = pool.waitGroup(); var counter = common.Counter{}; var sleeper = common.SleepTaskContext{ .counter = &counter, .sleep_ns = 500_000_000 };
// Fill every queue slot with long-running sleep tasks. for (0..4) |_| { _ = try pool.submit(common.sleepTask, @ptrCast(&sleeper), .{ .wait_group = &wg }); }
// trySubmit must fail immediately with a queue/pool exhaustion error. // Either error.QueueFull or error.TaskPoolExhausted is acceptable -- // TaskPoolExhausted fires when all descriptors are consumed before // the ring buffer itself is full. const result = pool.trySubmit(common.counterTask, @ptrCast(&counter), .{}); if (result) |_| { return error.TestUnexpectedResult; } else |err| { try std.testing.expect(err == error.QueueFull or err == error.TaskPoolExhausted); }
// Clean up: let the sleep tasks finish. wg.wait();}
test "block_with_timeout returns Timeout when queue stays full" { var pool = try thread_pool.ThreadPool.init(std.testing.allocator, .{ .min_threads = 1, .max_threads = 1, .queue_capacity = 4, .scheduler_variant = .global_queue, }); defer pool.deinit();
var wg = pool.waitGroup(); var counter = common.Counter{}; // Sleep long enough that the queue stays full during the 100 ms // block_with_timeout below (5× margin: 500 ms task vs 100 ms timeout). var sleeper = common.SleepTaskContext{ .counter = &counter, .sleep_ns = 500_000_000 };
// Fill the queue with long-running tasks. for (0..4) |_| { _ = try pool.submit(common.sleepTask, @ptrCast(&sleeper), .{ .wait_group = &wg }); }
// submit() with block_with_timeout should time out because the queue // stays full for longer than the 100ms timeout. const result = pool.submit(common.counterTask, @ptrCast(&counter), .{ .backpressure = .block_with_timeout, .timeout_ns = 100_000_000, }); try std.testing.expectError(error.Timeout, result);
// Clean up: let the pool drain. wg.wait();}
test "shutdown with timeout returns Timeout for long-running tasks" { var pool = try thread_pool.ThreadPool.init(std.testing.allocator, .{ .min_threads = 1, .max_threads = 1, .queue_capacity = 64, }); // NOTE: no defer pool.deinit() here -- we call shutdown manually, then // must still deinit. The deinit calls shutdown(.drain, null) internally // which is safe to call after a timed-out shutdown. defer pool.deinit();
var counter = common.Counter{}; // 300 ms task vs 50 ms shutdown timeout = 6× margin; task will not finish // before the shutdown timeout fires. var sleeper = common.SleepTaskContext{ .counter = &counter, .sleep_ns = 300_000_000 }; _ = try pool.submit(common.sleepTask, @ptrCast(&sleeper), .{});
// Give the worker a moment to pick up the task so it becomes active. thread_pool.clock.sleep(10_000_000);
// Shutdown with a 50ms timeout -- the 300ms task won't finish in time. const result = pool.core.shutdown(.drain, 50_000_000); try std.testing.expectError(error.Timeout, result);}
test "WaitGroup done underflow guard present in debug build modes" { // Verify the WaitGroup correctly tracks pending count down to zero. // The underflow guard (@panic) fires if done() is called when pending == 0 // in Debug and ReleaseSafe (std.debug.runtime_safety == true). // In ReleaseFast the invariant is restored silently (F-004 fix). var wg = thread_pool.types.WaitGroup{}; wg.add(1); try std.testing.expectEqual(@as(u32, 1), wg.pendingCount()); wg.done(); try std.testing.expectEqual(@as(u32, 0), wg.pendingCount()); // At this point, calling wg.done() in Debug/ReleaseSafe would trigger: // @panic("WaitGroup: done() called more times than add()") // In ReleaseFast it returns early after restoring pending to 0.}
test "F-004: WaitGroup.done underflow does not crash ReleaseFast" { // This test only runs under ReleaseFast (runtime_safety == false) // where the fix makes double-done resilient. In Debug/ReleaseSafe // the panic is preserved as a caller-bug catcher. if (std.debug.runtime_safety) return error.SkipZigTest;
var wg = thread_pool.types.WaitGroup{}; wg.add(1); wg.done(); wg.done(); // pre-fix: would crash; post-fix: must return cleanly // pending must be saturated at 0, not wrapped negative try std.testing.expectEqual(@as(u32, 0), wg.pendingCount());}
test "local_queue_capacity zero rejected for all scheduler variants" { var cfg = thread_pool.types.Config{ .local_queue_capacity = 0, .scheduler_variant = .global_queue }; cfg = cfg.normalized(); try std.testing.expectError(error.InvalidLocalQueueCapacity, cfg.validate());}
test "concurrent push-pop-steal does not lose or duplicate items" { const WorkDeque = @import("thread_pool").WorkDeque; const Deque = WorkDeque(u32); var deque = try Deque.init(std.testing.allocator, 256); defer deque.deinit();
const total_items: u32 = 10_000; var owner_collected = std.atomic.Value(u32).init(0); var stolen_collected = std.atomic.Value(u32).init(0);
const Stealer = struct { fn run(d: *Deque, owner_done: *std.atomic.Value(u32), stolen: *std.atomic.Value(u32)) void { var count: u32 = 0; while (true) { if (d.stealTop()) |_| { count += 1; } else { if (d.len() == 0 and owner_done.load(.acquire) > 0) { if (d.stealTop()) |_| { count += 1; } else break; } std.Thread.yield() catch {}; } } _ = stolen.fetchAdd(count, .release); } };
var threads: [3]std.Thread = undefined; for (&threads) |*t| { t.* = try std.Thread.spawn(.{}, Stealer.run, .{ &deque, &owner_collected, &stolen_collected }); }
var owner_count: u32 = 0; var pushed: u32 = 0; while (pushed < total_items) { if (deque.tryPushBottom(pushed)) { pushed += 1; } if (pushed % 4 == 0) { if (deque.popBottom()) |_| { owner_count += 1; } } } while (deque.popBottom()) |_| { owner_count += 1; }
_ = owner_collected.fetchAdd(owner_count, .release);
for (&threads) |t| t.join();
const total_recovered = owner_collected.load(.acquire) + stolen_collected.load(.acquire); try std.testing.expectEqual(total_items, total_recovered);}
test "Stats.wakeups aggregates from per-worker counters" { var pool = try thread_pool.ThreadPool.init(std.testing.allocator, .{ .max_threads = 2, .queue_capacity = 64, }); defer pool.deinit();
var wg = pool.waitGroup(); var counter = common.Counter{}; for (0..8) |_| { _ = try pool.submit(common.counterTask, @ptrCast(&counter), .{ .wait_group = &wg }); } wg.wait();
// Pool-level wakeups must equal the sum of per-worker wakeups. This is // the correctness invariant Q-022 fixed: the pool-level counter was // previously never incremented while workers did increment their own. // The actual value may be zero under ReleaseFast when workers never // needed to park (8 trivial tasks finish before the idle threshold), // so we only assert the aggregation invariant, not a minimum count. const stats = pool.statsSnapshot(); var expected: u64 = 0; for (stats.worker_stats) |w| { expected += w.wakeups; } try std.testing.expectEqual(expected, stats.wakeups);}
test "Stats.steals/parks aggregate per-worker under work-stealing" { var pool = try thread_pool.ThreadPool.init(std.testing.allocator, .{ .max_threads = 4, .queue_capacity = 1024, .local_queue_capacity = 64, .scheduler_variant = .work_stealing, }); defer pool.deinit();
var wg = pool.waitGroup(); var counter = common.Counter{}; for (0..256) |_| { _ = try pool.submit(common.counterTask, @ptrCast(&counter), .{ .wait_group = &wg }); } wg.wait();
const stats = pool.statsSnapshot();
var expected_steals: u64 = 0; var expected_steal_failures: u64 = 0; var expected_parks: u64 = 0; for (stats.worker_stats) |w| { expected_steals += w.steals; expected_steal_failures += w.steal_failures; expected_parks += w.parks; } try std.testing.expectEqual(expected_steals, stats.steals); try std.testing.expectEqual(expected_steal_failures, stats.steal_failures); try std.testing.expectEqual(expected_parks, stats.parks);}
test "latency summary exposes p999 and p9999 tail fields" { // OBSERVABILITY-PIVOT / R-01: the LatencyHistogram exposes p99.9 and // p99.99 derived from the same 64-bucket log2 histogram used for p50/ // p95/p99. The quantisation is coarse at the tails, so we only assert // monotonicity (p99 <= p999 <= p9999 <= max) plus non-zeroness under // realistic load. Skipped when timestamps are compiled out, since all // histogram fields are intentionally zero in that build. if (thread_pool.build_options.timestamp_mode == .disabled) return error.SkipZigTest; var pool = try thread_pool.ThreadPool.init(std.testing.allocator, .{ .min_threads = 2, .max_threads = 2, .queue_capacity = 4096, .default_backpressure = .block, }); defer pool.deinit();
var wg = pool.waitGroup(); var counter = common.Counter{}; // 2000 tasks is enough to populate tail buckets under realistic scheduling // jitter without making this test slow. The histogram has 64 log2 buckets // so even sparse data hits multiple bins. `.block` backpressure keeps the // test deterministic if the scheduler stalls submissions. for (0..2000) |_| { _ = try pool.submit(common.counterTask, @ptrCast(&counter), .{ .wait_group = &wg }); } wg.wait();
const stats = pool.statsSnapshot(); try std.testing.expectEqual(@as(u64, 2000), counter.value.load(.acquire)); try std.testing.expectEqual(@as(u64, 2000), stats.task_wait_ns.count); try std.testing.expectEqual(@as(u64, 2000), stats.task_exec_ns.count);
// p999 / p9999 must be monotone non-decreasing with p99 and bounded by max. // These are the R-01 acceptance assertions. try std.testing.expect(stats.task_wait_ns.p99_ns <= stats.task_wait_ns.p999_ns); try std.testing.expect(stats.task_wait_ns.p999_ns <= stats.task_wait_ns.p9999_ns); try std.testing.expect(stats.task_wait_ns.p9999_ns <= stats.task_wait_ns.max_ns);
try std.testing.expect(stats.task_exec_ns.p99_ns <= stats.task_exec_ns.p999_ns); try std.testing.expect(stats.task_exec_ns.p999_ns <= stats.task_exec_ns.p9999_ns); try std.testing.expect(stats.task_exec_ns.p9999_ns <= stats.task_exec_ns.max_ns);
// At 2000 samples in realistic load, p9999 cannot be zero — that would // indicate the summary path is silently skipping the scaled quantile. try std.testing.expect(stats.task_exec_ns.p9999_ns > 0);}
test "per-lane histograms populated when allow_blocking_lane=true" { // OBSERVABILITY-PIVOT / R-02: with the blocking lane enabled, the // snapshot carries separate default/blocking wait+exec summaries in // addition to the pool-wide `task_{wait,exec}_ns`. We exercise both // lanes and assert that both per-lane histograms receive samples. // // SONNY-2427: the split is keyed on `Worker.serves_lane`, not on the // task's lane, and the blocking worker always helps the default lane // (Slice L3). Nothing used to decide which workers drained the 64 // default-lane tasks; whenever the blocking worker reached them first // it ran all 80 tasks and `default_wait_ns.count` was legitimately 0. // Pin the placement instead: hold the only blocking worker inside a // blocking-lane gate task while the default-lane tasks run, so only // default workers can run them. if (thread_pool.build_options.timestamp_mode == .disabled) return error.SkipZigTest; try common.skipIfSchedulerVariantIncompatible(.lane_extension); var pool = try thread_pool.ThreadPool.init(std.testing.allocator, .{ .min_threads = 4, .max_threads = 4, .queue_capacity = 1024, .local_queue_capacity = 32, .scheduler_variant = .lane_extension, .allow_blocking_lane = true, .blocking_workers = 1, .blocking_queue_capacity = 512, // Bounds deinit if a task is ever lost (the waits below then fail // with WaitGroupDeadline instead of hanging). .deinit_timeout_ns = 5 * std.time.ns_per_s, }); defer pool.deinit();
var counter = common.Counter{}; var gate = common.GateTaskContext{ .counter = &counter }; // Runs before `pool.deinit()` (defers are LIFO), so a failed // assertion can never leave the blocking worker stuck in the gate. defer gate.release.store(true, .release);
var blocking_wg = pool.waitGroup(); _ = try pool.submit(common.gateTask, @ptrCast(&gate), .{ .wait_group = &blocking_wg, .lane = .blocking }); const gate_deadline = thread_pool.clock.nanoTimestamp() + 10 * std.time.ns_per_s; while (!gate.started.load(.acquire)) { if (thread_pool.clock.nanoTimestamp() > gate_deadline) return error.GateTaskNeverStarted; thread_pool.clock.sleep(50 * std.time.ns_per_us); }
// 64 default-lane tasks and 16 blocking-lane tasks (the gate plus 15); // each lane's histograms must count exactly its own tasks, and the // pool-wide summary count must equal their sum. var default_wg = pool.waitGroup(); for (0..64) |_| { _ = try pool.submit(common.counterTask, @ptrCast(&counter), .{ .wait_group = &default_wg }); } // Deadline-bounded, not `wait()`: lane_extension workers use local // ring queues, and a lost task (SONNY-2448) must fail, not hang. try common.waitPendingZero(&default_wg, 10 * std.time.ns_per_s);
gate.release.store(true, .release); for (0..15) |_| { _ = try pool.submit( common.counterTask, @ptrCast(&counter), .{ .wait_group = &blocking_wg, .lane = .blocking }, ); } try common.waitPendingZero(&blocking_wg, 10 * std.time.ns_per_s);
const stats = pool.statsSnapshot(); try std.testing.expectEqual(@as(u64, 80), counter.value.load(.acquire)); // With placement pinned, the per-lane counts are exact. try std.testing.expectEqual(@as(u64, 64), stats.default_wait_ns.count); try std.testing.expectEqual(@as(u64, 64), stats.default_exec_ns.count); try std.testing.expectEqual(@as(u64, 16), stats.blocking_wait_ns.count); try std.testing.expectEqual(@as(u64, 16), stats.blocking_exec_ns.count); // Pool-wide count is the sum of lane counts. try std.testing.expectEqual( stats.task_wait_ns.count, stats.default_wait_ns.count + stats.blocking_wait_ns.count, ); try std.testing.expectEqual( stats.task_exec_ns.count, stats.default_exec_ns.count + stats.blocking_exec_ns.count, ); try std.testing.expectEqual(@as(u64, 80), stats.task_wait_ns.count);}
test "memory_budget_used_bytes mirrors live_bytes when budget is unlimited" { // OBSERVABILITY-PIVOT / R-10: default `max_pool_memory_bytes = 0` // (unlimited) must preserve pre-R-10 behaviour — scratch // allocations succeed unconditionally and `memory_budget_used_bytes` // reports the same number as `live_bytes`. var pool = try thread_pool.ThreadPool.init(std.testing.allocator, .{ .min_threads = 2, .max_threads = 2, .queue_capacity = 64, .worker_scratch_cap_bytes = 4096, // max_pool_memory_bytes defaults to 0 (unlimited) }); defer pool.deinit();
var wg = pool.waitGroup(); var counter = common.Counter{}; var ctx = common.ScratchTaskContext{ .counter = &counter, .bytes = 256 }; for (0..8) |_| { _ = try pool.submit(common.scratchTask, @ptrCast(&ctx), .{ .wait_group = &wg }); } wg.wait();
const stats = pool.statsSnapshot(); try std.testing.expectEqual(stats.live_bytes, stats.memory_budget_used_bytes); try std.testing.expect(stats.live_bytes > 0);}
test "max_pool_memory_bytes rejects scratch growth past budget" { // OBSERVABILITY-PIVOT / R-10: setting a tiny budget causes the // scratch arena's `ensureBuffer` allocation to return // `error.PoolMemoryBudgetExceeded`; the error bubbles out of // `allocScratch` and the task gracefully continues (the scratch // task uses `catch {}` so this is observable via the counter // incrementing 0 for scratch allocs but the task itself still // executing). Most importantly, `memory_budget_used_bytes` stays // <= budget + slack. var pool = try thread_pool.ThreadPool.init(std.testing.allocator, .{ .min_threads = 2, .max_threads = 2, .queue_capacity = 64, .worker_scratch_cap_bytes = 64 * 1024, // Budget is 1 byte — effectively "no scratch growth allowed". // Init-time allocations (workers, queues, descriptors) are // exempt from the budget per the R-10 design note. .max_pool_memory_bytes = 1, }); defer pool.deinit();
var wg = pool.waitGroup(); var counter = common.Counter{}; var ctx = common.ScratchTaskContext{ .counter = &counter, .bytes = 256 }; for (0..4) |_| { _ = try pool.submit(common.scratchTask, @ptrCast(&ctx), .{ .wait_group = &wg }); } wg.wait();
// The task always runs (scratch allocation uses `catch {}`), so // the counter advances even when the scratch alloc was rejected. try std.testing.expectEqual(@as(u64, 4), counter.value.load(.acquire));
const stats = pool.statsSnapshot(); // No scratch arena was successfully allocated because the budget // (1 byte) is smaller than `worker_scratch_cap_bytes` (64 KiB). // The per-worker scratch_current_bytes therefore stays at 0. var scratch_total: u64 = 0; for (stats.worker_stats) |ws| scratch_total += ws.scratch_current_bytes; try std.testing.expectEqual(@as(u64, 0), scratch_total);}
test "wake-latency histogram records a signal-driven wake of a parked worker" { // OBSERVABILITY-PIVOT / R-03: when timestamps are enabled, the pool // records the delta between parking on `park_cond.timedWait` and that // call returning before `idle_park_after_ns`. Timeout returns are // excluded. // // SONNY-2427: this test used to run bursts separated by 200 µs sleeps // with a 10 ms park timeout, and needed a signal-driven wake to land // within 10 ms of a park. Wake coalescing left only one submit that // could supply it, so under CPU contention it failed (1/1,200 runs). // Now: both workers are observed parked before the submit, every // submit runs the wake scan, and the park timeout is far longer than // the watchdog, so the task can only run through a recorded wake. if (thread_pool.build_options.timestamp_mode == .disabled) return error.SkipZigTest; const watchdog_ns: i96 = 10 * std.time.ns_per_s; const park_timeout_ns: u64 = 60 * std.time.ns_per_s;
var pool = try thread_pool.ThreadPool.init(std.testing.allocator, .{ .min_threads = 2, .max_threads = 2, .queue_capacity = 256, .spin_before_park_iters = 0, // A timeout return is not recorded, and cannot happen within the // watchdog: only a wake can bring a worker back. .idle_park_after_ns = park_timeout_ns, // Every submit scans for a parked worker (the default coalesces // scans to every 32nd submit). .wake_coalesce_batch = 1, // Bounds deinit on the failure path: joinAllWorkers broadcasts // the parked workers awake after this deadline. .deinit_timeout_ns = 5 * std.time.ns_per_s, }); defer pool.deinit();
// Wait until both workers are inside the park handshake. A worker // sets `is_parked` before its final queue check, and nothing is // queued yet, so it goes on to wait; the submitter's scan sees // `is_parked` and signals under the park mutex. const setup_ns = thread_pool.clock.nanoTimestamp(); while (true) { var parked: usize = 0; for (pool.core.workers) |*w| { if (w.is_parked.load(.seq_cst) == 1) parked += 1; } if (parked == pool.core.workers.len) break; if (thread_pool.clock.nanoTimestamp() - setup_ns > watchdog_ns) return error.WorkersNeverParked; thread_pool.clock.sleep(50 * std.time.ns_per_us); } // Baseline rather than an exact-zero assertion: `timedWait` can return // early on a spurious OS wake (not a timeout, not our submit's signal) // before either worker ever sees a task, and the recorder (src/pool.zig, // `parkWorker`) can't tell that apart from a real wake — it only checks // that the return happened before `idle_park_after_ns` elapsed. Record // count/max now and assert the delta after the submit instead. const baseline = pool.statsSnapshot().wake_latency_ns;
var wg = pool.waitGroup(); var counter = common.Counter{}; _ = try pool.submit(common.counterTask, @ptrCast(&counter), .{ .wait_group = &wg }); try common.waitPendingZero(&wg, watchdog_ns); try std.testing.expectEqual(@as(u64, 1), counter.value.load(.acquire));
// The worker that ran the task recorded its wake before running it, // and the wait group's completion orders that record before this read. // No-lost-wakeup property: at least one additional wake was recorded // between the baseline above and here, i.e. the submit's signal (or a // spurious wake racing it) was not dropped. const stats = pool.statsSnapshot(); try std.testing.expect(stats.wake_latency_ns.count > baseline.count); // Recorded wakes are strictly shorter than the park timeout. Note: this // test no longer independently proves that a timeout return is never // recorded as a wake — that guarantee now rests solely on the // `delta_ns < idle_park_after_ns` gate in `parkWorker`, not on this // test's baseline being exactly zero. That check was deliberately given // up to make the test robust to a spurious pre-submit wake. try std.testing.expect(stats.wake_latency_ns.max_ns < park_timeout_ns); try std.testing.expect(stats.wake_latency_ns.mean_ns > 0);}
test "per-lane histograms zero when allow_blocking_lane=false" { // OBSERVABILITY-PIVOT / R-02: single-lane configs must leave the // per-lane fields at the zero `LatencySummary` so operators can // distinguish "blocking lane disabled" from "blocking lane saw zero // tasks". The pool-wide `task_{wait,exec}_ns` still carries the // overall data. if (thread_pool.build_options.timestamp_mode == .disabled) return error.SkipZigTest; var pool = try thread_pool.ThreadPool.init(std.testing.allocator, common.defaultConfig(.global_queue)); defer pool.deinit();
var wg = pool.waitGroup(); var counter = common.Counter{}; for (0..128) |_| { _ = try pool.submit(common.counterTask, @ptrCast(&counter), .{ .wait_group = &wg }); } wg.wait();
const stats = pool.statsSnapshot(); try std.testing.expect(stats.task_wait_ns.count > 0); try std.testing.expectEqual(@as(u64, 0), stats.default_wait_ns.count); try std.testing.expectEqual(@as(u64, 0), stats.default_exec_ns.count); try std.testing.expectEqual(@as(u64, 0), stats.blocking_wait_ns.count); try std.testing.expectEqual(@as(u64, 0), stats.blocking_exec_ns.count);}
test "timestamp_mode disabled zeroes histograms" { // COMPTIME-ELISION: when the library is compiled with // -Dtimestamp-mode=full the runtime config is rejected by design. // Skip the disabled-mode behavioral check in that build. if (thread_pool.build_options.timestamp_mode == .full) return error.SkipZigTest; var pool = try thread_pool.ThreadPool.init(std.testing.allocator, .{ .max_threads = 2, .queue_capacity = 64, .timestamp_mode = .disabled, }); defer pool.deinit();
var wg = pool.waitGroup(); var counter = common.Counter{}; for (0..32) |_| { _ = try pool.submit(common.counterTask, @ptrCast(&counter), .{ .wait_group = &wg }); } wg.wait();
const stats = pool.statsSnapshot(); try std.testing.expectEqual(@as(u64, 0), stats.task_wait_ns.count); try std.testing.expectEqual(@as(u64, 0), stats.task_exec_ns.count); try std.testing.expectEqual(@as(u64, 32), stats.completed);}
test "timestamp_mode disabled still purges scratch on idle" { // COMPTIME-ELISION: skip under -Dtimestamp-mode=full (same reasoning // as the histogram-zeroing test above). if (thread_pool.build_options.timestamp_mode == .full) return error.SkipZigTest; var pool = try thread_pool.ThreadPool.init(std.testing.allocator, .{ .max_threads = 2, .queue_capacity = 64, .worker_scratch_cap_bytes = 1024, .idle_park_after_ns = 100_000, // 0.1ms — immediate park .idle_purge_after_ns = 200_000, // 0.2ms — quick purge trigger .timestamp_mode = .disabled, }); defer pool.deinit();
var counter = common.Counter{}; var ctx = common.ScratchTaskContext{ .counter = &counter, .bytes = 256 };
var wg = pool.waitGroup(); _ = try pool.submit(common.scratchTask, @ptrCast(&ctx), .{ .wait_group = &wg }); wg.wait();
// Sleep long enough for: // 1. Worker to enter idle path (spin, then park for idle_park_after_ns) // 2. Timeout to fire (idle_park_after_ns) // 3. Purge check to succeed (idle_purge_after_ns since task completion) // Use a generous sleep to account for scheduling delays thread_pool.clock.sleep(100_000_000); // 100ms — very ample time for purge to occur
const stats = pool.statsSnapshot(); // After purge, scratch memory should be released. live_bytes may still include // other allocations (queues, etc.), but scratch_current_bytes should be near 0. if (stats.worker_stats.len > 0) { var total_scratch: u64 = 0; for (stats.worker_stats) |ws| { total_scratch += ws.scratch_current_bytes; } try std.testing.expect(total_scratch == 0); }}
test "fuzz BoundedRing push/pop consistency" { try std.testing.fuzz({}, struct { fn testOne(_: void, smith: *std.testing.Smith) !void { var buf: [256]u8 = undefined; const len = smith.sliceWithHash(&buf, 0); const input = buf[0..len];
const Ring = thread_pool.BoundedRing(u32); var ring = Ring.init(std.testing.allocator, 16) catch return; defer ring.deinit();
var pushed: u32 = 0; var popped: u32 = 0;
for (input) |byte| { if (byte % 2 == 0) { if (ring.tryPush(pushed)) pushed += 1; } else { if (ring.pop()) |_| popped += 1; } }
while (ring.pop()) |_| popped += 1;
try std.testing.expectEqual(pushed, popped); } }.testOne, .{});}
test "fuzz Config validation never panics" { try std.testing.fuzz({}, struct { fn testOne(_: void, smith: *std.testing.Smith) !void { const max_threads = smith.valueWithHash(u32, 1); const queue_cap = smith.valueWithHash(u32, 2); const local_cap = smith.valueWithHash(u32, 3); const scheduler_byte = smith.valueWithHash(u8, 4) % 3;
var cfg = thread_pool.Config{ .max_threads = max_threads, .queue_capacity = queue_cap, .local_queue_capacity = local_cap, .scheduler_variant = @enumFromInt(scheduler_byte), }; cfg = cfg.normalized(); cfg.validate() catch {}; } }.testOne, .{});}
test "phase g2: a parked worker wakes for work submitted after the spin window" { // SONNY-2427: this test used to assert that the task ran < 1 ms after // submit. That bound belongs to the OS scheduler, not the pool: with // both workers parked, `scanAndWake` signals one within microseconds // of the submit, but under CPU contention the kernel can take several // ms to run the woken worker, and pinned to one CPU the old // busy-waiting main thread starved it until preempted. The end-of-spin // local-work probe the name refers to was dropped by the G2 re-land // (see `Core.idle`); what the pool guarantees is that the // spin-then-park handshake never loses the submit's wakeup. Assert // that without a latency bound: the park timeout is far longer than // the dispatch watchdog, so the task runs in time only if the wake // reached a parked worker. const allocator = std.testing.allocator; const watchdog_ns: i96 = 10 * std.time.ns_per_s;
var pool = try thread_pool.ThreadPool.init(allocator, .{ .min_threads = 2, .max_threads = 2, .scheduler_variant = .work_stealing, .queue_capacity = 256, // Workers spin before parking, as in the original G2 scenario. .spin_before_park_iters = 10_000, // A lost wakeup would strand the task until this timeout. .idle_park_after_ns = 60 * std.time.ns_per_s, // Bounds deinit on that failure path: shutdown(.drain) waits for // the stranded task until this deadline, then joinAllWorkers // broadcasts the parked workers awake. .deinit_timeout_ns = 5 * std.time.ns_per_s, }); defer pool.deinit();
// Replaces the old 5 ms sleep: wait until both workers are past their // spin window and on the park path. const setup_ns = thread_pool.clock.nanoTimestamp(); while (pool.statsSnapshot().parks < 2) { if (thread_pool.clock.nanoTimestamp() - setup_ns > watchdog_ns) return error.WorkersNeverParked; thread_pool.clock.sleep(50 * std.time.ns_per_us); }
const Ctx = struct { ran: std.atomic.Value(bool) = std.atomic.Value(bool).init(false), fn run(ctx_ptr: *anyopaque, _: *thread_pool.TaskContext) void { const self: *@This() = @ptrCast(@alignCast(ctx_ptr)); self.ran.store(true, .release); } }; var ctx = Ctx{};
// The pool's first submit always runs the wake scan (it is the head // of a `wake_coalesce_batch` window), so it must signal a parked // worker. _ = try pool.submit(Ctx.run, &ctx, .{});
// Poll with a sleep, not a spin, so this thread never competes for // CPU with the worker it is waiting on. const submit_ns = thread_pool.clock.nanoTimestamp(); while (!ctx.ran.load(.acquire)) { if (thread_pool.clock.nanoTimestamp() - submit_ns > watchdog_ns) break; thread_pool.clock.sleep(50 * std.time.ns_per_us); } try std.testing.expect(ctx.ran.load(.acquire));}
test "per-worker park: submit wakes a parked worker" { var pool = try thread_pool.ThreadPool.init(std.testing.allocator, .{ .max_threads = 4, .queue_capacity = 128, .idle_park_after_ns = 10_000_000, // 10ms — long enough for workers to park }); defer pool.deinit();
// Let workers park before we submit. thread_pool.clock.sleep(50_000_000); // 50ms
var wg = pool.waitGroup(); var counter = common.Counter{}; _ = try pool.submit(common.counterTask, @ptrCast(&counter), .{ .wait_group = &wg }); wg.wait();
try std.testing.expectEqual(@as(u64, 1), counter.value.load(.acquire)); // At least one worker must have woken up. const stats = pool.statsSnapshot(); try std.testing.expect(stats.wakeups > 0);}
test "Stats.started and .completed aggregate from per-worker counters" { var pool = try thread_pool.ThreadPool.init(std.testing.allocator, .{ .max_threads = 2, .queue_capacity = 64, }); defer pool.deinit();
var wg = pool.waitGroup(); var counter = common.Counter{}; for (0..64) |_| { _ = try pool.submit(common.counterTask, @ptrCast(&counter), .{ .wait_group = &wg }); } wg.wait();
const stats = pool.statsSnapshot(); try std.testing.expectEqual(@as(u64, 64), stats.started); try std.testing.expectEqual(@as(u64, 64), stats.completed);}
test "runtime_stats omits parked_workers counter after simplification" { // Compile-time assertion: RuntimeStats (types.Stats) must NOT have // a `parked_workers` field after this refactor. comptime std.debug.assert(!@hasField(thread_pool.types.Stats, "parked_workers"));}
test "backpressure block policy returns when capacity frees" { var pool = try thread_pool.ThreadPool.init(std.testing.allocator, .{ .max_threads = 1, .queue_capacity = 4, .local_queue_capacity = 2, }); defer pool.deinit();
var wg = pool.waitGroup(); var counter = common.Counter{}; // Fill the descriptor pool with tasks. for (0..4) |_| { _ = try pool.submit(common.counterTask, @ptrCast(&counter), .{ .wait_group = &wg }); } // This one must block (no descriptors available) then succeed once the // worker drains a task and releases a descriptor, which signals the // backpressure condvar. _ = try pool.submit(common.counterTask, @ptrCast(&counter), .{ .wait_group = &wg, .backpressure = .block, }); wg.wait(); try std.testing.expectEqual(@as(u64, 5), counter.value.load(.acquire));}
test "LIFO slot: submit-from-worker uses lifo_slot before local ring" { const allocator = std.testing.allocator; var pool = try thread_pool.ThreadPool.init(allocator, .{ .max_threads = 1, .min_threads = 1, .queue_capacity = 256, .local_queue_capacity = 16, .scheduler_variant = .work_stealing, }); defer pool.deinit();
// Pass both pool pointer and wait-group to outerTask so inner tasks // can register with the same wait-group and be awaited. const OuterCtx = struct { pool: *thread_pool.ThreadPool, wg: *thread_pool.WaitGroup, };
const Nested = struct { var outer_counter: std.atomic.Value(u64) = std.atomic.Value(u64).init(0); var inner_counter: std.atomic.Value(u64) = std.atomic.Value(u64).init(0);
fn innerTask(raw_ctx: *anyopaque, runtime: *thread_pool.TaskContext) void { _ = raw_ctx; _ = runtime; _ = inner_counter.fetchAdd(1, .release); } fn outerTask(raw_ctx: *anyopaque, runtime: *thread_pool.TaskContext) void { _ = runtime; const octx: *OuterCtx = @ptrCast(@alignCast(raw_ctx)); _ = outer_counter.fetchAdd(1, .release); _ = octx.pool.submit(innerTask, @ptrCast(octx.pool), .{ .wait_group = octx.wg }) catch {}; } };
Nested.outer_counter.store(0, .release); Nested.inner_counter.store(0, .release);
var wg = pool.waitGroup(); var octx = OuterCtx{ .pool = &pool, .wg = &wg }; for (0..100) |_| { _ = try pool.submit(Nested.outerTask, @ptrCast(&octx), .{ .wait_group = &wg }); } wg.wait(); // waits for all 100 outer + 100 inner tasks
try std.testing.expectEqual(@as(u64, 100), Nested.outer_counter.load(.acquire)); try std.testing.expectEqual(@as(u64, 100), Nested.inner_counter.load(.acquire));}
test "backpressure deadline: returns error.Timeout when queue stays full" { // Queue=2, single worker whose two slots stay occupied by blocking tasks. // A 3rd submit with a tight deadline must return Timeout within ~100 ms, // not hang. Exercises the spin-backoff deadline-clipping path in // waitForCapacity. var pool = try thread_pool.ThreadPool.init(std.testing.allocator, .{ .min_threads = 1, .max_threads = 1, .queue_capacity = 2, .local_queue_capacity = 2, }); defer pool.deinit();
var block_sem = thread_pool.sync.Semaphore{};
const BlockCtx = struct { sem: *thread_pool.sync.Semaphore, fn run(raw: *anyopaque, rt: *thread_pool.TaskContext) void { _ = rt; const ctx: *@This() = @ptrCast(@alignCast(raw)); ctx.sem.wait(); } }; var bctx = BlockCtx{ .sem = &block_sem };
// Fill the queue with blocking tasks. _ = try pool.submit(BlockCtx.run, @ptrCast(&bctx), .{}); _ = try pool.submit(BlockCtx.run, @ptrCast(&bctx), .{});
// Third submit must time out within ~100 ms. const start = thread_pool.clock.nanoTimestamp(); const result = pool.submit(BlockCtx.run, @ptrCast(&bctx), .{ .backpressure = .block_with_timeout, .timeout_ns = 10 * std.time.ns_per_ms, }); const elapsed = thread_pool.clock.nanoTimestamp() - start; try std.testing.expectError(error.Timeout, result); try std.testing.expect(elapsed >= 10 * std.time.ns_per_ms); try std.testing.expect(elapsed < 100 * std.time.ns_per_ms); // hang guard
// Release the blocked tasks so shutdown drains cleanly. block_sem.post(); block_sem.post();}
test "Config validate rejects max_threads > 64" { const cfg = thread_pool.Config{ .max_threads = 65, .min_threads = 1 }; try std.testing.expectError(error.TooManyWorkers, cfg.validate());}
test "Config validate accepts max_threads = 64" { const cfg = thread_pool.Config{ .max_threads = 64, .min_threads = 1 }; try cfg.validate();}
test "Config validate: explicit max_threads > 64 still rejected after normalized()" { // normalized() only clamps auto-detect (max_threads == 0) to 64. // Explicit over-limit values must still be rejected by validate(). const cfg = (thread_pool.Config{ .max_threads = 65, .min_threads = 1 }).normalized(); try std.testing.expectError(error.TooManyWorkers, cfg.validate());}
test "non_empty_workers bitmap invariant under racing push/pop" { // 8 workers pushing/popping 80k tasks; after quiesce, counter must // equal total, queued/active must be 0, and a follow-up submit must // still execute. Smoke test that the scheduler protocol doesn't break // correctness under concurrent push, pop-to-empty, and steal drain. var pool = try thread_pool.ThreadPool.init(std.testing.allocator, .{ .min_threads = 8, .max_threads = 8, .queue_capacity = 4096, .local_queue_capacity = 256, .scheduler_variant = .work_stealing, }); defer pool.deinit();
var counter = common.Counter{}; var wg = pool.waitGroup();
const total: u64 = 80_000; var i: u64 = 0; while (i < total) : (i += 1) { // Block on backpressure so the test works on CI runners with // fewer cores than workers (queue fills before workers drain). _ = try pool.submit(common.counterTask, @ptrCast(&counter), .{ .wait_group = &wg, .backpressure = .block }); } wg.wait(); try std.testing.expectEqual(total, counter.value.load(.acquire));
// Give workers a brief window to drain late stragglers so counters settle. thread_pool.clock.sleep(5 * std.time.ns_per_ms); const stats = pool.statsSnapshot(); try std.testing.expectEqual(@as(u64, 0), stats.queue_depth_current); try std.testing.expectEqual(@as(u64, 0), stats.active_tasks);
// Follow-up submit confirms end-to-end wakeup path still functions // post-quiesce. var final_counter = common.Counter{}; var final_wg = pool.waitGroup(); _ = try pool.submit(common.counterTask, @ptrCast(&final_counter), .{ .wait_group = &final_wg, .backpressure = .block }); final_wg.wait(); try std.testing.expectEqual(@as(u64, 1), final_counter.value.load(.acquire));}
test "Config validate rejects hot_spinners >= max_threads" { const cfg = thread_pool.Config{ .max_threads = 4, .min_threads = 1, .hot_spinners = 4 }; try std.testing.expectError(error.InvalidHotSpinners, cfg.validate());}
test "Config validate accepts hot_spinners < max_threads" { const cfg = thread_pool.Config{ .max_threads = 4, .min_threads = 1, .hot_spinners = 1 }; try cfg.validate();}
test "Config validate rejects hot_spinners > 0 with wake_every_n_tasks == 0" { const cfg = thread_pool.Config{ .max_threads = 4, .min_threads = 1, .hot_spinners = 1, .wake_every_n_tasks = 0, }; try std.testing.expectError(error.InvalidWakeThreshold, cfg.validate());}
test "Config validate accepts hot_spinners = 0 with wake_every_n_tasks = 0" { const cfg = thread_pool.Config{ .max_threads = 4, .min_threads = 1, .hot_spinners = 0, .wake_every_n_tasks = 0, }; try cfg.validate();}
test "active_spinners: counter stays non-negative under concurrent workers" { // Integration-level sanity: submit a steady stream of tasks to a // multi-worker pool, assert total task count matches after quiesce. // If the active_spinners fetchAdd/fetchSub pairing is broken (e.g., // missing a decrement on some exit path), the counter drifts and // the submitter either skips all wakes (tasks never picked up) or // always wakes (no perf benefit but still correct). This test // catches the first failure mode — a stuck pool. var pool = try thread_pool.ThreadPool.init(std.testing.allocator, .{ .min_threads = 4, .max_threads = 4, .queue_capacity = 4096, .local_queue_capacity = 256, .scheduler_variant = .work_stealing, // Short spin to force rapid spin→park→wake cycles, exercising // every exit path from idle(). .spin_before_park_iters = 8, .idle_park_after_ns = 50_000, }); defer pool.deinit();
var counter = common.Counter{}; var wg = pool.waitGroup();
const total: u64 = 40_000; var i: u64 = 0; while (i < total) : (i += 1) { _ = try pool.submit(common.counterTask, @ptrCast(&counter), .{ .wait_group = &wg, .backpressure = .block }); } wg.wait(); try std.testing.expectEqual(total, counter.value.load(.acquire));
// Brief quiesce so workers drain any stragglers and return to idle. thread_pool.clock.sleep(5 * std.time.ns_per_ms);
// Submit one more task; must execute. If active_spinners is stuck >0 // with all workers parked, the submitter skips the wake and this // task starves (test timeout via wg.wait semantics on single task). var final_counter = common.Counter{}; var final_wg = pool.waitGroup(); _ = try pool.submit(common.counterTask, @ptrCast(&final_counter), .{ .wait_group = &final_wg, .backpressure = .block }); final_wg.wait(); try std.testing.expectEqual(@as(u64, 1), final_counter.value.load(.acquire));}
test "hot_spinners = 1 executes all submitted tasks" { var pool = try thread_pool.ThreadPool.init(std.testing.allocator, .{ .min_threads = 4, .max_threads = 4, .queue_capacity = 4096, .local_queue_capacity = 256, .scheduler_variant = .work_stealing, .hot_spinners = 1, .wake_every_n_tasks = 8, }); defer pool.deinit();
var counter = common.Counter{}; var wg = pool.waitGroup();
const total: u64 = 10_000; var i: u64 = 0; while (i < total) : (i += 1) { _ = try pool.submit(common.counterTask, @ptrCast(&counter), .{ .wait_group = &wg, .backpressure = .block }); } wg.wait(); try std.testing.expectEqual(total, counter.value.load(.acquire));}
test "hot_spinners = 1 executes follow-up task after quiesce" { // Regression guard: after the initial burst drains, the hot-spinner // must still execute the next submitted task without requiring any // external wake. var pool = try thread_pool.ThreadPool.init(std.testing.allocator, .{ .min_threads = 2, .max_threads = 2, .queue_capacity = 1024, .local_queue_capacity = 128, .scheduler_variant = .work_stealing, .hot_spinners = 1, .wake_every_n_tasks = 4, }); defer pool.deinit();
var counter = common.Counter{}; var wg = pool.waitGroup(); var i: u64 = 0; while (i < 500) : (i += 1) { _ = try pool.submit(common.counterTask, @ptrCast(&counter), .{ .wait_group = &wg }); } wg.wait(); try std.testing.expectEqual(@as(u64, 500), counter.value.load(.acquire));
// Brief idle period so the non-hot worker parks. thread_pool.clock.sleep(10 * std.time.ns_per_ms);
// One more task — must execute via the hot-spinner. var solo = common.Counter{}; var solo_wg = pool.waitGroup(); _ = try pool.submit(common.counterTask, @ptrCast(&solo), .{ .wait_group = &solo_wg }); solo_wg.wait(); try std.testing.expectEqual(@as(u64, 1), solo.value.load(.acquire));}
test "phase h regression: state CAS gate rejects duplicate descriptor dispatch" { // Regression guard for the Phase H state-CAS fix in runTask. The wider // reproducer lives in tests/stress/double_dequeue_race.zig; this is a // smaller/faster unit test intended to exercise the same race window // (spin=0 + tight idle park + 16-worker submit flood) on every unit // test run so a future refactor that drops the state CAS surfaces here // rather than only in the stress suite. // // The state CAS at the top of runTask transitions .queued → .running; // any duplicate descriptor_index emission from some queue path arrives // after the first claimant already released the descriptor (task_fn is // null, state is .free or recycled), fails the CAS, and returns // silently. Pre-fix, the same race would SIGSEGV on a NULL task_fn call. const noop = struct { fn run(_: *anyopaque, _: *thread_pool.TaskContext) void {} }.run;
var pool = try thread_pool.ThreadPool.init(std.testing.allocator, .{ .min_threads = 16, .max_threads = 16, .queue_capacity = 4096, .local_queue_capacity = 256, .scheduler_variant = .work_stealing, .spin_before_park_iters = 0, .idle_park_after_ns = 10_000, // 10us — high park churn }); defer pool.deinit();
// Let workers reach steady state before the flood. thread_pool.clock.sleep(5 * std.time.ns_per_ms);
const per_producer: u64 = 200_000; var wg = pool.waitGroup();
const Submitter = struct { fn run(p: *thread_pool.ThreadPool, g: *thread_pool.WaitGroup, n: u64) void { var dummy: u8 = 0; var i: u64 = 0; while (i < n) : (i += 1) { _ = p.submit(noop, @ptrCast(&dummy), .{ .backpressure = .block, .wait_group = g, }) catch return; } } };
const t1 = try std.Thread.spawn(.{}, Submitter.run, .{ &pool, &wg, per_producer }); const t2 = try std.Thread.spawn(.{}, Submitter.run, .{ &pool, &wg, per_producer }); t1.join(); t2.join();
// Deadline-based polling so a lost wakeup surfaces as a test failure // with a clear pendingCount rather than a process-level hang. const deadline_ns: i128 = 60 * std.time.ns_per_s; const start = thread_pool.clock.nanoTimestamp(); while (thread_pool.clock.nanoTimestamp() - start < deadline_ns) { if (wg.pendingCount() == 0) break; thread_pool.clock.sleep(1 * std.time.ns_per_ms); }
try std.testing.expectEqual(@as(u32, 0), wg.pendingCount());}
test "g3: Stats.submitted reports exact spawn count" { var pool = try thread_pool.ThreadPool.init(std.testing.allocator, common.defaultConfig(.work_stealing)); defer pool.deinit();
const N: u64 = 1000; var wg = pool.waitGroup(); var counter = common.Counter{}; var i: u64 = 0; while (i < N) : (i += 1) { _ = try pool.submit(common.counterTask, @ptrCast(&counter), .{ .wait_group = &wg, .backpressure = .block }); } wg.wait();
try std.testing.expectEqual(N, counter.value.load(.acquire)); const stats = pool.statsSnapshot(); try std.testing.expectEqual(N, stats.submitted);}
test "sync.Mutex lock/unlock is thread-safe" { var m: thread_pool.sync.Mutex = .{}; var counter: u64 = 0; const N = 4; const IterPerThread = 1000; const Worker = struct { fn run(mutex: *thread_pool.sync.Mutex, c: *u64) void { var i: usize = 0; while (i < IterPerThread) : (i += 1) { mutex.lock(); c.* += 1; mutex.unlock(); } } }; var threads: [N]std.Thread = undefined; for (&threads) |*t| t.* = try std.Thread.spawn(.{}, Worker.run, .{ &m, &counter }); for (threads) |t| t.join(); try std.testing.expectEqual(@as(u64, N * IterPerThread), counter);}
test "sync.Mutex tryLock succeeds when unlocked, fails when locked" { var m: thread_pool.sync.Mutex = .{}; try std.testing.expect(m.tryLock()); try std.testing.expect(!m.tryLock()); m.unlock(); try std.testing.expect(m.tryLock()); m.unlock();}
test "sync.Condition signal wakes waiter" { var m: thread_pool.sync.Mutex = .{}; var c: thread_pool.sync.Condition = .{}; var ready: bool = false;
const Ctx = struct { m: *thread_pool.sync.Mutex, c: *thread_pool.sync.Condition, ready: *bool }; const ctx = Ctx{ .m = &m, .c = &c, .ready = &ready };
const Waiter = struct { fn run(x: Ctx) void { x.m.lock(); defer x.m.unlock(); while (!x.ready.*) x.c.wait(x.m); } }; var t = try std.Thread.spawn(.{}, Waiter.run, .{ctx});
// Yield a handful of times to let the waiter park; no external sleep. var i: usize = 0; while (i < 1000) : (i += 1) std.Thread.yield() catch {};
m.lock(); ready = true; c.signal(); m.unlock();
t.join();}
test "sync.Condition: a signal racing a timedWait timeout is never orphaned" { // SONNY-2427: signal()/broadcast() reserve a wake by bumping // `state.signals` while `waiters > signals`. A timed-out waiter used to // leave with a blind `fetchSub` of its waiter count, so a signal landing // between its last check for signals and that fetchSub was orphaned // (signals > waiters). While an orphan sits in `state`, signal() and // broadcast() skip the next waiter (they only fire while waiters > // signals): it sleeps its whole timeout, or forever in `wait`. // // This thread is the only waiter and signals are only sent under `m`, // so each time timedWait returns (holding `m`) no waiter or reserved // signal may remain. The race is rare on one CPU (it needs a preemption // between the waiter's last signal check and its exit), so this test // only detects it with two or more CPUs: against the unfixed exit path // these 20k waits found orphans in 99/100 unpinned runs and 90/100 runs // on two pinned CPUs (Debug + ReleaseSafe), none on one CPU // (SONNY-2427). The fixed exit path cannot orphan. var m: thread_pool.sync.Mutex = .{}; var c: thread_pool.sync.Condition = .{}; var stop = std.atomic.Value(bool).init(false); var waiting = std.atomic.Value(bool).init(false);
const Signaler = struct { fn run( mm: *thread_pool.sync.Mutex, cc: *thread_pool.sync.Condition, stop_flag: *std.atomic.Value(bool), waiting_flag: *std.atomic.Value(bool), ) void { var prng = std.Random.DefaultPrng.init(0x2427); const random = prng.random(); while (!stop_flag.load(.acquire)) { if (!waiting_flag.load(.acquire)) { std.atomic.spinLoopHint(); continue; } // Jitter the signal across the waiter's timeout so some // land in its exit path. var spins = random.uintLessThan(u32, 4096); while (spins > 0) : (spins -= 1) std.atomic.spinLoopHint(); mm.lock(); cc.signal(); mm.unlock(); } } }; const signaler = try std.Thread.spawn(.{}, Signaler.run, .{ &m, &c, &stop, &waiting }); defer { stop.store(true, .release); signaler.join(); }
var orphaned: usize = 0; var round: usize = 0; while (round < 20_000) : (round += 1) { m.lock(); waiting.store(true, .release); c.timedWait(&m, 5 * std.time.ns_per_us) catch {}; waiting.store(false, .release); const state = c.state.load(.monotonic); if (state.waiters != 0 or state.signals != 0) { orphaned += 1; // Reset under `m` (no signal() in flight) so each orphan is // counted once. c.state.store(.{ .waiters = 0, .signals = 0 }, .monotonic); } m.unlock(); } try std.testing.expectEqual(@as(usize, 0), orphaned);}
test "sync.Semaphore permits balance" { var s: thread_pool.sync.Semaphore = .{}; s.post(); s.post(); s.wait(); s.wait(); // Non-hang = success.}
test "clock.nanoTimestamp increases over time" { const a = thread_pool.clock.nanoTimestamp(); thread_pool.clock.sleep(1 * std.time.ns_per_ms); const b = thread_pool.clock.nanoTimestamp(); try std.testing.expect(b > a);}
test "clock.sleep sleeps at least the requested duration" { const want_ns: u64 = 5 * std.time.ns_per_ms; const start = thread_pool.clock.nanoTimestamp(); thread_pool.clock.sleep(want_ns); const elapsed = thread_pool.clock.nanoTimestamp() - start; // Allow 0.5ms slop for scheduler jitter. try std.testing.expect(elapsed >= @as(i96, @intCast(want_ns - 500_000)));}
test "DescCache stores and returns indices LIFO" { var cache = thread_pool.DescCache.init(); try std.testing.expectEqual(@as(usize, 0), cache.len()); try std.testing.expect(cache.tryPop() == null);
_ = cache.push(7); _ = cache.push(9); try std.testing.expectEqual(@as(?u32, 9), cache.tryPop()); try std.testing.expectEqual(@as(?u32, 7), cache.tryPop()); try std.testing.expect(cache.tryPop() == null);}
test "DescCache full push returns false" { var cache = thread_pool.DescCache.init(); var i: u32 = 0; while (i < thread_pool.DescCache.capacity) : (i += 1) try std.testing.expect(cache.push(i)); try std.testing.expect(!cache.push(9999));}
test "DescCache drainAll yields all held indices" { var cache = thread_pool.DescCache.init(); _ = cache.push(1); _ = cache.push(2); _ = cache.push(3);
var drained: [4]u32 = undefined; const n = cache.drainAll(&drained); try std.testing.expectEqual(@as(usize, 3), n); try std.testing.expectEqualSlices(u32, &[_]u32{ 3, 2, 1 }, drained[0..n]); try std.testing.expectEqual(@as(usize, 0), cache.len());}
test "acquireDescriptorBatch returns distinct indices in one CAS" { const allocator = std.testing.allocator; var pool = try thread_pool.ThreadPool.init(allocator, .{ .min_threads = 1, .max_threads = 1, .queue_capacity = 64, .local_queue_capacity = 32, .scheduler_variant = .work_stealing, }); defer pool.deinit();
var out: [16]u32 = undefined; const n = pool.acquireDescriptorBatchForTesting(&out); try std.testing.expect(n > 0 and n <= out.len);
var seen = std.AutoHashMap(u32, void).init(allocator); defer seen.deinit(); for (out[0..n]) |idx| { const gop = try seen.getOrPut(idx); try std.testing.expect(!gop.found_existing); }
for (out[0..n]) |idx| pool.releaseDescriptorForTesting(idx);}
test "registered submitter handles N > DescCache.capacity submits correctly" { const allocator = std.testing.allocator; var pool = try thread_pool.ThreadPool.init(allocator, .{ .min_threads = 2, .max_threads = 2, .queue_capacity = 256, .local_queue_capacity = 32, .scheduler_variant = .work_stealing, .max_submitter_queues = 1, .submitter_queue_capacity = 128, }); defer pool.deinit();
const slot = try pool.registerSubmitter(); defer pool.releaseSubmitter(slot);
var done = std.atomic.Value(u64).init(0); const noop = struct { fn run(raw: *anyopaque, _: *thread_pool.TaskContext) void { const d: *std.atomic.Value(u64) = @ptrCast(@alignCast(raw)); _ = d.fetchAdd(1, .monotonic); } }.run;
var wg = pool.waitGroup(); const N: usize = thread_pool.DescCache.capacity * 3 + 5; var i: usize = 0; while (i < N) : (i += 1) { _ = try pool.submit(noop, @ptrCast(&done), .{ .wait_group = &wg, .backpressure = .block }); } wg.wait(); try std.testing.expectEqual(@as(u64, N), done.load(.acquire));}
test "releaseSubmitter drains desc_cache back to global freelist" { const allocator = std.testing.allocator; var pool = try thread_pool.ThreadPool.init(allocator, .{ .min_threads = 2, .max_threads = 2, .queue_capacity = 64, .local_queue_capacity = 32, .scheduler_variant = .work_stealing, .max_submitter_queues = 1, .submitter_queue_capacity = 64, }); defer pool.deinit();
const stats_before = pool.statsSnapshot(); const live_before = stats_before.task_desc_live;
const slot = try pool.registerSubmitter();
var done = std.atomic.Value(u64).init(0); const noop = struct { fn run(raw: *anyopaque, _: *thread_pool.TaskContext) void { const d: *std.atomic.Value(u64) = @ptrCast(@alignCast(raw)); _ = d.fetchAdd(1, .monotonic); } }.run;
var wg = pool.waitGroup(); // 16 submits will refill the cache (take 32 from freelist on first miss, // consume 1, cache 31 — then 15 more tryPop hits). After wg.wait() returns, // 16 descriptors are queued but the other 16 cached indices are still live. for (0..16) |_| { _ = try pool.submit(noop, @ptrCast(&done), .{ .wait_group = &wg, .backpressure = .block }); } wg.wait();
// Release the submitter — must return the remaining ~16 cached // descriptors to the global freelist, restoring task_desc_live to // the pre-registration value. pool.releaseSubmitter(slot);
// SONNY-2427: join the workers before reading task_desc_live. // finalizeTask calls `wait_group.done()` before `releaseDescriptor()`, // so wg.wait() can return while a worker still holds the last task's // descriptor. shutdown(.drain) does not touch submitter caches, so the // count below still depends on releaseSubmitter's drain. try pool.shutdown(.drain, 10 * std.time.ns_per_s);
const stats_after = pool.statsSnapshot(); try std.testing.expectEqual(live_before, stats_after.task_desc_live);}
test "Worker struct is within the 4096-byte ceiling post-desc_cache" { const allocator = std.testing.allocator; var pool = try thread_pool.ThreadPool.init(allocator, .{ .min_threads = 4, .max_threads = 4, .queue_capacity = 64, .local_queue_capacity = 32, .scheduler_variant = .work_stealing, }); defer pool.deinit();
try pool.assertWorkerSizeBoundedForTesting(4096);}
test "worker-spawned task consumes from per-worker DescCache" { const allocator = std.testing.allocator; var pool = try thread_pool.ThreadPool.init(allocator, .{ .min_threads = 2, .max_threads = 2, .queue_capacity = 256, .local_queue_capacity = 64, .scheduler_variant = .work_stealing, }); defer pool.deinit();
var done = std.atomic.Value(u64).init(0); const Ctx = struct { pool: *thread_pool.ThreadPool, wg: *thread_pool.WaitGroup, done: *std.atomic.Value(u64), fn parent(raw: *anyopaque, _: *thread_pool.TaskContext) void { const self: *@This() = @ptrCast(@alignCast(raw)); var i: usize = 0; while (i < 64) : (i += 1) { _ = self.pool.submit(child, @ptrCast(self), .{ .wait_group = self.wg, .backpressure = .block, }) catch return; } } fn child(raw: *anyopaque, _: *thread_pool.TaskContext) void { const self: *@This() = @ptrCast(@alignCast(raw)); _ = self.done.fetchAdd(1, .monotonic); } };
var wg = pool.waitGroup(); var ctx = Ctx{ .pool = &pool, .wg = &wg, .done = &done }; _ = try pool.submit(Ctx.parent, @ptrCast(&ctx), .{ .wait_group = &wg, .backpressure = .block, }); wg.wait(); try std.testing.expectEqual(@as(u64, 64), done.load(.monotonic));}
test "pool.deinit drains per-worker desc_caches" { // Verifies that descriptor indices cached in Worker.desc_cache are // returned to the global freelist at pool teardown, so task_desc_live // reaches zero. The explicit drainWorkerDescCachesForTesting hook // mirrors the drain that deinitOwned runs, letting us assert the // invariant before deinit frees the pool (statsSnapshot needs it). // // SONNY-2427: both snapshots follow `shutdown(.drain)`, which joins // the workers — the same order deinit uses (join, then drain). // `wg.wait()` alone does not order them: finalizeTask calls // `wait_group.done()` before `releaseDescriptor()`, so a worker // preempted between the two still counts a live descriptor when // wg.wait() returns. const allocator = std.testing.allocator; var pool = try thread_pool.ThreadPool.init(allocator, .{ .min_threads = 2, .max_threads = 2, .queue_capacity = 256, .local_queue_capacity = 64, .scheduler_variant = .work_stealing, }); defer pool.deinit();
// Every child is submitted from the parent's worker thread, so only // that worker's DescCache refills: 100 children take 4 batches of 32 // and leave 28 indices cached but unused. (The old 96 is a multiple of // 32, which left the cache empty and the drain with nothing to do.) const children = 100; var done = std.atomic.Value(u64).init(0); const Ctx = struct { pool: *thread_pool.ThreadPool, wg: *thread_pool.WaitGroup, done: *std.atomic.Value(u64), fn parent(raw: *anyopaque, _: *thread_pool.TaskContext) void { const self: *@This() = @ptrCast(@alignCast(raw)); var i: usize = 0; while (i < children) : (i += 1) { _ = self.pool.submit(child, @ptrCast(self), .{ .wait_group = self.wg, .backpressure = .block, }) catch return; } } fn child(raw: *anyopaque, _: *thread_pool.TaskContext) void { const self: *@This() = @ptrCast(@alignCast(raw)); _ = self.done.fetchAdd(1, .monotonic); } };
var wg = pool.waitGroup(); var ctx = Ctx{ .pool = &pool, .wg = &wg, .done = &done }; _ = try pool.submit(Ctx.parent, @ptrCast(&ctx), .{ .wait_group = &wg, .backpressure = .block, }); wg.wait(); try pool.shutdown(.drain, 10 * std.time.ns_per_s); try std.testing.expectEqual(@as(u64, children), done.load(.monotonic));
// Workers are joined, so every descriptor still live is a cached, // never-used index: the drain below has work to do. try std.testing.expect(pool.statsSnapshot().task_desc_live > 0);
// Drain worker caches before asserting; this exercises the same code // path that deinitOwned calls internally at shutdown. pool.drainWorkerDescCachesForTesting();
const stats = pool.statsSnapshot(); try std.testing.expectEqual(@as(u64, 0), stats.task_desc_live);}
test "clock: nanoTimestamp is monotonic non-decreasing" { // GAP-0036 / AP-01: the clock source must be monotonic (.awake) so that // interval subtractions in pool.zig never underflow when the wall clock // steps backward (NTP correction, VM resume). var prev = thread_pool.clock.nanoTimestamp(); var i: usize = 0; while (i < 1000) : (i += 1) { const now = thread_pool.clock.nanoTimestamp(); try std.testing.expect(now >= prev); prev = now; }}
test "version constant is non-empty and parses as semver" { // AP2-01: thread_pool.version is now derived from build.zig.zon via // build_options so a release bump requires only one edit. Verify the // injected value is non-empty and well-formed; the exact string need // not be hard-coded here because it flows from the manifest automatically. try std.testing.expect(thread_pool.version.len > 0); _ = try std.SemanticVersion.parse(thread_pool.version);}
test "SubmitError is an explicit, exhaustively switchable set" { // QR-004 (#43): the public submit contract is a closed error set. The // else-less switch fails to compile if a member is ever added without // updating the declared set (or vice versa), pinning the API contract. var e: thread_pool.SubmitError = error.QueueFull; _ = &e; switch (e) { error.ShutdownInProgress => {}, error.QueueFull => {}, error.TaskPoolExhausted => {}, error.BlockingLaneDisabled => {}, error.Timeout => {}, }}
test "SubmitBatchError is SubmitError plus WaitableUnsupported" { // QR-004 (#43): batch submit adds exactly one member (Q-NEW-02, #107). var e: thread_pool.SubmitBatchError = error.WaitableUnsupported; _ = &e; switch (e) { error.WaitableUnsupported => {}, error.ShutdownInProgress => {}, error.QueueFull => {}, error.TaskPoolExhausted => {}, error.BlockingLaneDisabled => {}, error.Timeout => {}, }}
test "wait/shutdown/registerSubmitter error sets are closed" { // QR-004 (#43): single-member sets — the else-less switches prove no // hidden members exist. var we: thread_pool.WaitError = error.InvalidHandle; _ = &we; switch (we) { error.InvalidHandle => {}, } var se: thread_pool.ShutdownError = error.Timeout; _ = &se; switch (se) { error.Timeout => {}, } var re: thread_pool.RegisterSubmitterError = error.NoSubmitterSlots; _ = &re; switch (re) { error.NoSubmitterSlots => {}, }}
test "ConfigError is an explicit, exhaustively switchable set" { // QR-003 (#41): Config.validate returns a closed, documented set. var e: thread_pool.ConfigError = error.InvalidMaxThreads; _ = &e; switch (e) { error.InvalidMaxThreads => {}, error.TooManyWorkers => {}, error.InvalidMinThreads => {}, error.InvalidThreadBounds => {}, error.InvalidQueueCapacity => {}, error.InvalidLocalQueueCapacity => {}, error.LaneExtensionRequiresBlockingWorkers => {}, error.InvalidInjectBatchSize => {}, error.InvalidHotSpinners => {}, error.InvalidWakeThreshold => {}, error.InvalidBlockingWorkers => {}, error.BlockingLaneWithoutFlag => {}, error.AllowBlockingLaneRequiresWorkers => {}, error.InvalidBlockingQueueCapacity => {}, error.BlockingQueueCapacityWithoutLane => {}, error.BlockingBackpressureWithoutLane => {}, error.CpuHelpsBlockingWithoutLane => {}, error.WatchdogWithoutOverflowCap => {}, error.OverflowCapExceedsBitmap => {}, error.SubmitterCapExceedsBitmap => {}, }}
test "deinit is not blocked by a long watchdog interval (Q-NEW-08, #113)" { // Q-NEW-08 (#113): watchdogMain used a non-interruptible clock.sleep, so // joinAllWorkers (and therefore deinit) blocked up to one full // watchdog_interval_ns. With a 5s interval, deinit must still return // promptly because the stop signal wakes the watchdog's timed wait. var pool = try thread_pool.ThreadPool.init(std.testing.allocator, .{ .min_threads = 2, .max_threads = 2, .queue_capacity = 64, .local_queue_capacity = 16, .scheduler_variant = .work_stealing, .watchdog_interval_ns = 5 * std.time.ns_per_s, .blocking_promotion_timeout_ns = 10 * std.time.ns_per_ms, .max_overflow_workers = 1, .worker_scratch_cap_bytes = 512, }); // Give the watchdog thread time to enter its interval wait. Without this, // deinit can set watchdog_stop before the watchdog's first loop iteration // and the test passes vacuously even with a non-interruptible sleep. thread_pool.clock.sleep(100 * std.time.ns_per_ms); const t0 = thread_pool.clock.nanoTimestamp(); pool.deinit(); const elapsed = thread_pool.clock.nanoTimestamp() - t0; try std.testing.expect(elapsed < 2 * std.time.ns_per_s);}