lib/http/src/pool.zig
daab053ee43316e1809a84551d573ddd1e5bf3d2
1 const std = @import("std");
2 const alloc_phase = @import("alloc_phase");
3 const thread = @import("sys").thread;
4 const time = @import("sys").time;
5
6 const Allocator = std.mem.Allocator;
7
8 const DrainDeadline = struct {
9 clock: time.AwakeClock,
10 end: time.AwakeInstant,
11
12 fn init(clock: time.AwakeClock, timeout: time.Duration) time.ClockError!DrainDeadline {
13 return .{
14 .clock = clock,
15 .end = (try clock.now()).deadlineAfter(timeout),
16 };
17 }
18
19 fn reached(self: DrainDeadline) time.ClockError!bool {
20 return (try self.clock.now()).reached(self.end);
21 }
22 };
23
24 pub fn ThreadPool(comptime Task: type) type {
25 const Owner = struct {
26 const Self = @This();
27
28 const TaskNode = struct {
29 link: std.DoublyLinkedList.Node = .{},
30 task: Task = undefined,
31
32 fn fromLink(link: *std.DoublyLinkedList.Node) *TaskNode {
33 return @fieldParentPtr("link", link);
34 }
35 };
36
37 const State = struct {
38 workers: []thread.JoinHandle,
39 tasks: []TaskNode,
40 free: std.DoublyLinkedList,
41 ready: std.DoublyLinkedList,
42 queue_mutex: std.atomic.Mutex,
43 shutdown: std.atomic.Value(bool),
44 in_flight_tasks: std.atomic.Value(usize),
45 wake_sem: thread.Semaphore,
46 started_workers: usize,
47 };
48
49 pub const claim: alloc_phase.capacity.Declaration = .{
50 .source = .{
51 .id = "http.thread_pool",
52 .kind = .phase_static,
53 .limit_source = .caller,
54 .storage = .{
55 .covered = &.{
56 .{
57 .id = "state_and_exact_worker_handle_and_task_node_slices",
58 .lifetime = .steady,
59 .detail = "State and exact worker-handle and task-node slices",
60 },
61 },
62 .excluded = &.{
63 "caller-owned storage reachable from task values",
64 "allocations performed by task run and completion methods",
65 "OS thread stacks, handles, semaphore resources, clocks, and sleeps",
66 },
67 },
68 .capacity = .{
69 .inputs = &.{
70 alloc_phase.capacity.bindInput(Limits, "tasks", "tasks"),
71 alloc_phase.capacity.bindInput(Limits, "workers", "workers"),
72 },
73 .type_selectors = &.{},
74 .nodes = &.{
75 .{ .input = 0 },
76 .{ .input = 1 },
77 .{ .add = .{ .left = 0, .right = 1 } },
78 },
79 .assertions = &.{.{
80 .scope = .closure_total,
81 .measure = .retained,
82 .relation = .upper_bound,
83 .expression = 2,
84 }},
85 },
86 .overload = .{
87 .kind = .reject_before_mutation,
88 .detail = "PoolFull is returned before a free slot is removed or queue state changes",
89 },
90 .risks = .{
91 .transitive = .{
92 .status = .open,
93 .detail = "the concrete task run and completion methods remain separate owner-local storage obligations",
94 },
95 .foreign = .{
96 .status = .excluded,
97 .detail = "sys thread, semaphore, yield, time, sleep, and worker-exit TLS cleanup cross the owner-local storage claim",
98 },
99 },
100 .obligations = &.{
101 .{ .key = "http_capacity", .role = .capacity_model },
102 .{ .key = "http_sealed_maximum_overload", .role = .overload },
103 .{ .key = "http_sealed_maximum_transitive_risk", .role = .transitive_risk },
104 .{ .key = "http_oom_retry", .role = .custom },
105 .{ .key = "http_sys_tls", .role = .foreign_risk },
106 },
107 },
108 .bindings = .{
109 .owner = @This(),
110 .seal = .{
111 .family = alloc_phase.capacity.selector(@This().activate),
112 .premise = .{
113 .class = .checked_semantic_fact,
114 .authority = .checker,
115 },
116 },
117 .teardown = .{
118 .family = alloc_phase.capacity.selector(@This().deinit),
119 .premise = .{
120 .class = .checked_semantic_fact,
121 .authority = .checker,
122 },
123 },
124 },
125 };
126 phase: alloc_phase.capacity.Phase,
127 capacity: Capacity,
128 state: *State,
129
130 pub const Limits = struct {
131 workers: usize,
132 tasks: usize,
133 };
134
135 pub const Capacity = struct {
136 workers: usize,
137 tasks: usize,
138 state_bytes: usize,
139 worker_bytes: usize,
140 task_bytes: usize,
141 working_bytes: usize,
142
143 pub fn derive(limits: Limits) error{ InvalidWorkerCount, CapacityOverflow }!Capacity {
144 if (limits.workers == 0) return error.InvalidWorkerCount;
145 const state_bytes = @sizeOf(State);
146 const worker_bytes = try alloc_phase.capacity.mul(
147 usize,
148 limits.workers,
149 @sizeOf(thread.JoinHandle),
150 );
151 const task_bytes = try alloc_phase.capacity.mul(
152 usize,
153 limits.tasks,
154 @sizeOf(TaskNode),
155 );
156 const partial_bytes = try alloc_phase.capacity.add(
157 usize,
158 state_bytes,
159 worker_bytes,
160 );
161 const working_bytes = try alloc_phase.capacity.add(
162 usize,
163 partial_bytes,
164 task_bytes,
165 );
166 return .{
167 .workers = limits.workers,
168 .tasks = limits.tasks,
169 .state_bytes = state_bytes,
170 .worker_bytes = worker_bytes,
171 .task_bytes = task_bytes,
172 .working_bytes = working_bytes,
173 };
174 }
175 };
176
177 pub const Exhaustion = error{PoolFull};
178 pub const SubmitError = Exhaustion || error{ NotStarted, ShuttingDown };
179
180 pub fn init(allocator: Allocator, limits: Limits) !Self {
181 const capacity = try Capacity.derive(limits);
182 const state = try allocator.create(State);
183 errdefer allocator.destroy(state);
184 const workers = try allocator.alloc(thread.JoinHandle, capacity.workers);
185 errdefer allocator.free(workers);
186 const tasks = try allocator.alloc(TaskNode, capacity.tasks);
187 errdefer allocator.free(tasks);
188
189 state.* = .{
190 .workers = workers,
191 .tasks = tasks,
192 .free = .{},
193 .ready = .{},
194 .queue_mutex = .unlocked,
195 .shutdown = std.atomic.Value(bool).init(false),
196 .in_flight_tasks = std.atomic.Value(usize).init(0),
197 .wake_sem = .{},
198 .started_workers = 0,
199 };
200 for (state.tasks) |*task| {
201 task.* = .{};
202 state.free.append(&task.link);
203 }
204
205 errdefer {
206 state.shutdown.store(true, .release);
207 for (0..state.started_workers) |_| state.wake_sem.post();
208 for (state.workers[0..state.started_workers]) |worker| worker.join();
209 }
210
211 for (state.workers) |*worker| {
212 worker.* = try thread.spawn(workerLoop, .{state});
213 state.started_workers += 1;
214 }
215 return .{
216 .phase = .initialization,
217 .capacity = capacity,
218 .state = state,
219 };
220 }
221
222 pub fn activate(self: *Self) error{AlreadyStarted}!void {
223 if (self.phase != .initialization) return error.AlreadyStarted;
224 self.phase = .steady;
225 }
226
227 pub fn deinit(self: *Self, allocator: Allocator) void {
228 std.debug.assert(self.phase != .teardown);
229 const state = self.state;
230 state.shutdown.store(true, .release);
231
232 for (0..state.started_workers) |_| state.wake_sem.post();
233 for (state.workers[0..state.started_workers]) |worker| worker.join();
234
235 self.phase = .teardown;
236 allocator.free(state.tasks);
237 allocator.free(state.workers);
238 allocator.destroy(state);
239 self.state = undefined;
240 }
241
242 pub fn submit(self: *Self, submitted: Task) SubmitError!void {
243 if (self.phase != .steady) return error.NotStarted;
244 const state = self.state;
245
246 lock(&state.queue_mutex);
247 if (state.shutdown.load(.acquire)) {
248 unlock(&state.queue_mutex);
249 return error.ShuttingDown;
250 }
251 const task_node = TaskNode.fromLink(state.free.popFirst() orelse {
252 unlock(&state.queue_mutex);
253 return error.PoolFull;
254 });
255 task_node.task = submitted;
256 state.ready.append(&task_node.link);
257 _ = state.in_flight_tasks.fetchAdd(1, .acq_rel);
258 unlock(&state.queue_mutex);
259 state.wake_sem.post();
260 }
261
262 pub fn drain(
263 self: *Self,
264 clock: time.AwakeClock,
265 timeout: time.Duration,
266 ) time.ClockError!bool {
267 std.debug.assert(self.phase == .steady);
268 const deadline = try DrainDeadline.init(clock, timeout);
269 while (self.state.in_flight_tasks.load(.acquire) != 0) {
270 if (try deadline.reached()) return false;
271 sleepNanoseconds(1 * std.time.ns_per_ms);
272 }
273 return true;
274 }
275
276 fn workerLoop(state: *State) void {
277 while (true) {
278 state.wake_sem.wait();
279 if (state.shutdown.load(.acquire)) return;
280
281 lock(&state.queue_mutex);
282 const task_node = TaskNode.fromLink(state.ready.popFirst() orelse {
283 unlock(&state.queue_mutex);
284 continue;
285 });
286 const task = task_node.task;
287 unlock(&state.queue_mutex);
288
289 task.run();
290
291 lock(&state.queue_mutex);
292 state.free.append(&task_node.link);
293 unlock(&state.queue_mutex);
294 task.complete();
295 _ = state.in_flight_tasks.fetchSub(1, .acq_rel);
296 }
297 }
298 };
299 comptime alloc_phase.capacity.requireAllocatorRejectingOwnerShape(Owner);
300 return Owner;
301 }
302
303 const TestTask = union(enum) {
304 mark_executed: *std.atomic.Value(bool),
305 increment_counter: *std.atomic.Value(usize),
306 wait_for_release: *GateContext,
307 run_until_shutdown: *ShutdownContext,
308 complete_then_follow: *CompletionContext,
309
310 pub fn run(self: TestTask) void {
311 switch (self) {
312 .mark_executed => |flag| markExecuted(flag),
313 .increment_counter => |counter| incrementCounter(counter),
314 .wait_for_release => |context| waitForRelease(context),
315 .run_until_shutdown => |context| runUntilShutdown(context),
316 .complete_then_follow => |context| countCompletion(context),
317 }
318 }
319
320 pub fn complete(self: TestTask) void {
321 switch (self) {
322 .complete_then_follow => |context| waitForCompletionRelease(context),
323 else => {},
324 }
325 }
326 };
327
328 const TestThreadPool = ThreadPool(TestTask);
329 const TestState = TestThreadPool.State;
330 const TestTaskNode = TestThreadPool.TaskNode;
331
332 fn lock(mutex: *std.atomic.Mutex) void {
333 while (!mutex.tryLock()) thread.yield();
334 }
335
336 fn unlock(mutex: *std.atomic.Mutex) void {
337 std.debug.assert(@atomicLoad(std.atomic.Mutex, mutex, .unordered) == .locked);
338 @atomicStore(std.atomic.Mutex, mutex, .unlocked, .release);
339 }
340
341 fn markExecuted(flag: *std.atomic.Value(bool)) void {
342 flag.store(true, .release);
343 }
344
345 fn incrementCounter(counter: *std.atomic.Value(usize)) void {
346 _ = counter.fetchAdd(1, .acq_rel);
347 }
348
349 const GateContext = struct {
350 release: *std.atomic.Value(bool),
351 started: *std.atomic.Value(usize),
352 completed: *std.atomic.Value(usize),
353 };
354
355 fn waitForRelease(context: *GateContext) void {
356 _ = context.started.fetchAdd(1, .acq_rel);
357 var rounds: usize = 0;
358 while (!context.release.load(.acquire) and rounds < 5000) : (rounds += 1) {
359 sleepNanoseconds(1 * std.time.ns_per_ms);
360 }
361 _ = context.completed.fetchAdd(1, .acq_rel);
362 }
363
364 const ShutdownContext = struct {
365 shutdown: *std.atomic.Value(bool),
366 started: *std.atomic.Value(usize),
367 completed: *std.atomic.Value(usize),
368 };
369
370 fn runUntilShutdown(context: *ShutdownContext) void {
371 _ = context.started.fetchAdd(1, .acq_rel);
372 var rounds: usize = 0;
373 while (!context.shutdown.load(.acquire) and rounds < 5000) : (rounds += 1) {
374 sleepNanoseconds(1 * std.time.ns_per_ms);
375 }
376 _ = context.completed.fetchAdd(1, .acq_rel);
377 }
378
379 const ProducerContext = struct {
380 pool: *TestThreadPool,
381 counter: *std.atomic.Value(usize),
382 submissions: usize,
383 failed: *std.atomic.Value(bool),
384 };
385
386 const CompletionContext = struct {
387 counter: *std.atomic.Value(usize),
388 started: *std.atomic.Value(bool),
389 release: *std.atomic.Value(bool),
390 };
391
392 fn countCompletion(context: *CompletionContext) void {
393 incrementCounter(context.counter);
394 }
395
396 fn waitForCompletionRelease(context: *CompletionContext) void {
397 context.started.store(true, .release);
398 var rounds: usize = 0;
399 while (!context.release.load(.acquire) and rounds < 5000) : (rounds += 1) {
400 sleepNanoseconds(1 * std.time.ns_per_ms);
401 }
402 }
403
404 fn movePool(pool: TestThreadPool) TestThreadPool {
405 return pool;
406 }
407
408 const StorageSnapshot = struct {
409 capacity: TestThreadPool.Capacity,
410 state: *TestState,
411 workers_pointer: [*]thread.JoinHandle,
412 workers_length: usize,
413 tasks_pointer: [*]TestTaskNode,
414 tasks_length: usize,
415 started_workers: usize,
416 };
417
418 fn storageSnapshot(pool: *const TestThreadPool) StorageSnapshot {
419 return .{
420 .capacity = pool.capacity,
421 .state = pool.state,
422 .workers_pointer = pool.state.workers.ptr,
423 .workers_length = pool.state.workers.len,
424 .tasks_pointer = pool.state.tasks.ptr,
425 .tasks_length = pool.state.tasks.len,
426 .started_workers = pool.state.started_workers,
427 };
428 }
429
430 const QueueSnapshot = struct {
431 free_count: usize,
432 free_first: ?*std.DoublyLinkedList.Node,
433 free_last: ?*std.DoublyLinkedList.Node,
434 ready_count: usize,
435 ready_first: ?*std.DoublyLinkedList.Node,
436 ready_last: ?*std.DoublyLinkedList.Node,
437 in_flight_tasks: usize,
438 shutdown: bool,
439 };
440
441 fn queueSnapshot(pool: *TestThreadPool) QueueSnapshot {
442 const state = pool.state;
443 lock(&state.queue_mutex);
444 defer unlock(&state.queue_mutex);
445 return .{
446 .free_count = state.free.len(),
447 .free_first = state.free.first,
448 .free_last = state.free.last,
449 .ready_count = state.ready.len(),
450 .ready_first = state.ready.first,
451 .ready_last = state.ready.last,
452 .in_flight_tasks = state.in_flight_tasks.load(.acquire),
453 .shutdown = state.shutdown.load(.acquire),
454 };
455 }
456
457 fn waitForCount(counter: *const std.atomic.Value(usize), expected: usize) bool {
458 var rounds: usize = 0;
459 while (counter.load(.acquire) < expected and rounds < 5000) : (rounds += 1) {
460 sleepNanoseconds(1 * std.time.ns_per_ms);
461 }
462 return counter.load(.acquire) >= expected;
463 }
464
465 fn produceTasks(context: *ProducerContext) void {
466 for (0..context.submissions) |_| {
467 context.pool.submit(.{ .increment_counter = context.counter }) catch {
468 context.failed.store(true, .release);
469 return;
470 };
471 }
472 }
473
474 fn modelThreadPoolCapacity(limits: TestThreadPool.Limits) ?TestThreadPool.Capacity {
475 if (limits.workers == 0) return null;
476 const maximum_bytes = std.math.maxInt(usize);
477 const state_bytes = @sizeOf(TestState);
478 const worker_width = @sizeOf(thread.JoinHandle);
479 const task_width = @sizeOf(TestTaskNode);
480 if (worker_width != 0 and limits.workers > maximum_bytes / worker_width) return null;
481 if (task_width != 0 and limits.tasks > maximum_bytes / task_width) return null;
482 const worker_bytes = limits.workers * worker_width;
483 const task_bytes = limits.tasks * task_width;
484 if (worker_bytes > maximum_bytes - state_bytes) return null;
485 const partial_bytes = state_bytes + worker_bytes;
486 if (task_bytes > maximum_bytes - partial_bytes) return null;
487 return .{
488 .workers = limits.workers,
489 .tasks = limits.tasks,
490 .state_bytes = state_bytes,
491 .worker_bytes = worker_bytes,
492 .task_bytes = task_bytes,
493 .working_bytes = partial_bytes + task_bytes,
494 };
495 }
496
497 test "ThreadPool init and deinit" {
498 var pool = try TestThreadPool.init(std.testing.allocator, .{ .workers = 4, .tasks = 8 });
499 try std.testing.expectEqual(alloc_phase.capacity.Phase.initialization, pool.phase);
500 pool.deinit(std.testing.allocator);
501 try std.testing.expectEqual(alloc_phase.capacity.Phase.teardown, pool.phase);
502 }
503
504 test "ThreadPool requires exactly one activation before submission" {
505 var pool = try TestThreadPool.init(std.testing.allocator, .{ .workers = 1, .tasks = 1 });
506 defer pool.deinit(std.testing.allocator);
507
508 var executed = std.atomic.Value(bool).init(false);
509 try std.testing.expectError(error.NotStarted, pool.submit(.{ .mark_executed = &executed }));
510 try std.testing.expectEqual(alloc_phase.capacity.Phase.initialization, pool.phase);
511
512 try pool.activate();
513 try std.testing.expectEqual(alloc_phase.capacity.Phase.steady, pool.phase);
514 try std.testing.expectError(error.AlreadyStarted, pool.activate());
515
516 try pool.submit(.{ .mark_executed = &executed });
517 try std.testing.expect(try pool.drain(
518 .system(),
519 .fromNanoseconds(1 * std.time.ns_per_s),
520 ));
521 try std.testing.expect(executed.load(.acquire));
522 }
523
524 test "ThreadPool capacity matches an independent typed-byte model" {
525 comptime {
526 @stardustClaim(
527 @import("alloc_phase").capacity.witness(TestThreadPool, "http_capacity"),
528 null,
529 null,
530 null,
531 null,
532 null,
533 null,
534 );
535 }
536
537 const ordinary_limits = TestThreadPool.Limits{ .workers = 3, .tasks = 5 };
538 const ordinary_model = modelThreadPoolCapacity(ordinary_limits);
539 try std.testing.expect(ordinary_model != null);
540 try std.testing.expectEqual(ordinary_model.?, try TestThreadPool.Capacity.derive(ordinary_limits));
541
542 const zero_worker_limits = TestThreadPool.Limits{ .workers = 0, .tasks = 1 };
543 try std.testing.expect(modelThreadPoolCapacity(zero_worker_limits) == null);
544 try std.testing.expectError(
545 error.InvalidWorkerCount,
546 TestThreadPool.Capacity.derive(zero_worker_limits),
547 );
548 try std.testing.expectError(
549 error.InvalidWorkerCount,
550 TestThreadPool.init(std.testing.allocator, zero_worker_limits),
551 );
552
553 const maximum_bytes = std.math.maxInt(usize);
554 const worker_width = @sizeOf(thread.JoinHandle);
555 const task_width = @sizeOf(TestTaskNode);
556 try std.testing.expect(worker_width > 0);
557 try std.testing.expect(task_width > 0);
558
559 const maximum_workers = (maximum_bytes - @sizeOf(TestState)) / worker_width;
560 const worker_boundary = TestThreadPool.Limits{ .workers = maximum_workers, .tasks = 0 };
561 const worker_boundary_model = modelThreadPoolCapacity(worker_boundary);
562 try std.testing.expect(worker_boundary_model != null);
563 try std.testing.expectEqual(
564 worker_boundary_model.?,
565 try TestThreadPool.Capacity.derive(worker_boundary),
566 );
567 const worker_overflow = TestThreadPool.Limits{ .workers = maximum_workers + 1, .tasks = 0 };
568 try std.testing.expect(modelThreadPoolCapacity(worker_overflow) == null);
569 try std.testing.expectError(
570 error.CapacityOverflow,
571 TestThreadPool.Capacity.derive(worker_overflow),
572 );
573 try std.testing.expectError(
574 error.CapacityOverflow,
575 TestThreadPool.init(std.testing.allocator, worker_overflow),
576 );
577
578 const fixed_bytes = @sizeOf(TestState) + worker_width;
579 const maximum_tasks = (maximum_bytes - fixed_bytes) / task_width;
580 const task_boundary = TestThreadPool.Limits{ .workers = 1, .tasks = maximum_tasks };
581 const task_boundary_model = modelThreadPoolCapacity(task_boundary);
582 try std.testing.expect(task_boundary_model != null);
583 try std.testing.expectEqual(
584 task_boundary_model.?,
585 try TestThreadPool.Capacity.derive(task_boundary),
586 );
587 const task_overflow = TestThreadPool.Limits{ .workers = 1, .tasks = maximum_tasks + 1 };
588 try std.testing.expect(modelThreadPoolCapacity(task_overflow) == null);
589 try std.testing.expectError(
590 error.CapacityOverflow,
591 TestThreadPool.Capacity.derive(task_overflow),
592 );
593 try std.testing.expectError(
594 error.CapacityOverflow,
595 TestThreadPool.init(std.testing.allocator, task_overflow),
596 );
597 }
598
599 fn checkThreadPoolInitAllocationFailures(allocator: Allocator) !void {
600 var pool = try TestThreadPool.init(allocator, .{ .workers = 1, .tasks = 2 });
601 try std.testing.expectEqual(alloc_phase.capacity.Phase.initialization, pool.phase);
602 pool.deinit(allocator);
603 try std.testing.expectEqual(alloc_phase.capacity.Phase.teardown, pool.phase);
604 }
605
606 test "ThreadPool init cleans every allocation failure and remains retryable" {
607 comptime {
608 @stardustClaim(
609 @import("alloc_phase").capacity.witness(TestThreadPool, "http_oom_retry"),
610 null,
611 null,
612 null,
613 null,
614 null,
615 null,
616 );
617 }
618
619 try std.testing.checkAllAllocationFailures(
620 std.testing.allocator,
621 checkThreadPoolInitAllocationFailures,
622 .{},
623 );
624
625 var pool = try TestThreadPool.init(std.testing.allocator, .{ .workers = 1, .tasks = 2 });
626 defer pool.deinit(std.testing.allocator);
627 try pool.activate();
628 try std.testing.expectEqual(alloc_phase.capacity.Phase.steady, pool.phase);
629 }
630
631 test "ThreadPool handle remains valid after moving" {
632 const initialized = try TestThreadPool.init(std.testing.allocator, .{ .workers = 1, .tasks = 1 });
633 var pool = movePool(initialized);
634 defer pool.deinit(std.testing.allocator);
635 try pool.activate();
636
637 var executed = std.atomic.Value(bool).init(false);
638 try pool.submit(.{ .mark_executed = &executed });
639 try std.testing.expect(try pool.drain(
640 .system(),
641 .fromNanoseconds(1 * std.time.ns_per_s),
642 ));
643 try std.testing.expect(executed.load(.acquire));
644 }
645
646 test "ThreadPool submits a single caller-owned context" {
647 var pool = try TestThreadPool.init(std.testing.allocator, .{ .workers = 2, .tasks = 1 });
648 defer pool.deinit(std.testing.allocator);
649 try pool.activate();
650
651 var executed = std.atomic.Value(bool).init(false);
652 try pool.submit(.{ .mark_executed = &executed });
653 try std.testing.expect(try pool.drain(
654 .system(),
655 .fromNanoseconds(1 * std.time.ns_per_s),
656 ));
657 try std.testing.expect(executed.load(.acquire));
658 }
659
660 test "ThreadPool reuses task slots" {
661 var pool = try TestThreadPool.init(std.testing.allocator, .{ .workers = 4, .tasks = 100 });
662 defer pool.deinit(std.testing.allocator);
663 try pool.activate();
664
665 var counter = std.atomic.Value(usize).init(0);
666 for (0..100) |_| try pool.submit(.{ .increment_counter = &counter });
667 try std.testing.expect(try pool.drain(
668 .system(),
669 .fromNanoseconds(5 * std.time.ns_per_s),
670 ));
671 for (0..100) |_| try pool.submit(.{ .increment_counter = &counter });
672 try std.testing.expect(try pool.drain(
673 .system(),
674 .fromNanoseconds(5 * std.time.ns_per_s),
675 ));
676 try std.testing.expectEqual(@as(usize, 200), counter.load(.acquire));
677 }
678
679 test "ThreadPool is sealed before its first maximum-capacity workload" {
680 comptime {
681 @stardustClaim(
682 @import("alloc_phase").capacity.witness(TestThreadPool, "http_sealed_maximum_overload"),
683 null,
684 null,
685 null,
686 null,
687 null,
688 null,
689 );
690 }
691 comptime {
692 @stardustClaim(
693 @import("alloc_phase").capacity.witness(TestThreadPool, "http_sealed_maximum_transitive_risk"),
694 null,
695 null,
696 null,
697 null,
698 null,
699 null,
700 );
701 }
702
703 var phase_allocator = try alloc_phase.SealedPhaseAllocator.init(std.testing.allocator);
704 var maybe_pool: ?TestThreadPool = null;
705 var release = std.atomic.Value(bool).init(false);
706 errdefer {
707 release.store(true, .release);
708 if (maybe_pool) |*pool| {
709 if (pool.phase == .steady) {
710 _ = pool.drain(
711 .system(),
712 .fromNanoseconds(5 * std.time.ns_per_s),
713 ) catch false;
714 }
715 }
716 if (phase_allocator.phase() == .initialization) phase_allocator.abortInitialization();
717 if (phase_allocator.phase() == .steady) phase_allocator.beginTeardown();
718 if (maybe_pool) |*pool| {
719 if (pool.phase != .teardown) pool.deinit(phase_allocator.teardownAllocator());
720 }
721 phase_allocator.deinit();
722 }
723
724 maybe_pool = try TestThreadPool.init(
725 phase_allocator.initializationAllocator(),
726 .{ .workers = 1, .tasks = 2 },
727 );
728 const pool = &maybe_pool.?;
729 try std.testing.expectEqual(alloc_phase.capacity.Phase.initialization, phase_allocator.phase());
730 try std.testing.expectEqual(alloc_phase.capacity.Phase.initialization, pool.phase);
731
732 const storage = storageSnapshot(pool);
733 try std.testing.expectEqual(
734 try TestThreadPool.Capacity.derive(.{ .workers = 1, .tasks = 2 }),
735 storage.capacity,
736 );
737 try std.testing.expectEqual(@as(usize, 1), storage.workers_length);
738 try std.testing.expectEqual(@as(usize, 2), storage.tasks_length);
739 try std.testing.expectEqual(@as(usize, 1), storage.started_workers);
740 const idle_queue = queueSnapshot(pool);
741 try std.testing.expectEqual(@as(usize, 2), idle_queue.free_count);
742 try std.testing.expect(idle_queue.free_first != null);
743 try std.testing.expect(idle_queue.free_last != null);
744 try std.testing.expectEqual(@as(usize, 0), idle_queue.ready_count);
745 try std.testing.expect(idle_queue.ready_first == null);
746 try std.testing.expect(idle_queue.ready_last == null);
747 try std.testing.expectEqual(@as(usize, 0), idle_queue.in_flight_tasks);
748 try std.testing.expect(!idle_queue.shutdown);
749
750 phase_allocator.seal();
751 try std.testing.expectEqual(alloc_phase.capacity.Phase.steady, phase_allocator.phase());
752 try std.testing.expectEqual(alloc_phase.capacity.Phase.initialization, pool.phase);
753 try pool.activate();
754 try std.testing.expectEqual(alloc_phase.capacity.Phase.steady, pool.phase);
755
756 var started = std.atomic.Value(usize).init(0);
757 var completed = std.atomic.Value(usize).init(0);
758 var context = GateContext{
759 .release = &release,
760 .started = &started,
761 .completed = &completed,
762 };
763
764 for (0..4) |_| {
765 release.store(false, .release);
766 started.store(0, .release);
767 completed.store(0, .release);
768
769 try pool.submit(.{ .wait_for_release = &context });
770 try pool.submit(.{ .wait_for_release = &context });
771 try std.testing.expect(waitForCount(&started, 1));
772
773 const full_queue = queueSnapshot(pool);
774 try std.testing.expectEqual(@as(usize, 0), full_queue.free_count);
775 try std.testing.expect(full_queue.free_first == null);
776 try std.testing.expect(full_queue.free_last == null);
777 try std.testing.expectEqual(@as(usize, 1), full_queue.ready_count);
778 try std.testing.expect(full_queue.ready_first != null);
779 try std.testing.expect(full_queue.ready_last != null);
780 try std.testing.expectEqual(@as(usize, 2), full_queue.in_flight_tasks);
781 try std.testing.expect(!full_queue.shutdown);
782 try std.testing.expectError(error.PoolFull, pool.submit(.{ .wait_for_release = &context }));
783 try std.testing.expectEqual(full_queue, queueSnapshot(pool));
784 try std.testing.expectEqual(storage, storageSnapshot(pool));
785
786 release.store(true, .release);
787 try std.testing.expect(try pool.drain(
788 .system(),
789 .fromNanoseconds(5 * std.time.ns_per_s),
790 ));
791 try std.testing.expectEqual(@as(usize, 2), started.load(.acquire));
792 try std.testing.expectEqual(@as(usize, 2), completed.load(.acquire));
793 try std.testing.expectEqual(idle_queue, queueSnapshot(pool));
794 try std.testing.expectEqual(storage, storageSnapshot(pool));
795 }
796 try std.testing.expectEqual(alloc_phase.PhaseViolations{}, phase_allocator.violations());
797
798 phase_allocator.beginTeardown();
799 try std.testing.expectEqual(alloc_phase.capacity.Phase.teardown, phase_allocator.phase());
800 pool.deinit(phase_allocator.teardownAllocator());
801 try std.testing.expectEqual(alloc_phase.capacity.Phase.teardown, pool.phase);
802 try std.testing.expectEqual(alloc_phase.PhaseViolations{}, phase_allocator.violations());
803 phase_allocator.deinit();
804 }
805
806 test "ThreadPool completion runs after its task slot is reusable" {
807 var pool = try TestThreadPool.init(std.testing.allocator, .{ .workers = 1, .tasks = 1 });
808 defer pool.deinit(std.testing.allocator);
809 try pool.activate();
810
811 var counter = std.atomic.Value(usize).init(0);
812 var completion_started = std.atomic.Value(bool).init(false);
813 var release_completion = std.atomic.Value(bool).init(false);
814 defer release_completion.store(true, .release);
815 var context = CompletionContext{
816 .counter = &counter,
817 .started = &completion_started,
818 .release = &release_completion,
819 };
820 try pool.submit(.{ .complete_then_follow = &context });
821 var rounds: usize = 0;
822 while (!completion_started.load(.acquire) and rounds < 5000) : (rounds += 1) {
823 sleepNanoseconds(1 * std.time.ns_per_ms);
824 }
825 try std.testing.expect(completion_started.load(.acquire));
826 try pool.submit(.{ .increment_counter = &counter });
827 release_completion.store(true, .release);
828 try std.testing.expect(try pool.drain(
829 .system(),
830 .fromNanoseconds(5 * std.time.ns_per_s),
831 ));
832 try std.testing.expectEqual(@as(usize, 2), counter.load(.acquire));
833 }
834
835 test "ThreadPool accepts concurrent producers" {
836 comptime {
837 @stardustClaim(
838 @import("alloc_phase").capacity.witness(TestThreadPool, "http_sys_tls"),
839 null,
840 null,
841 null,
842 null,
843 null,
844 null,
845 );
846 }
847
848 const producer_count = 4;
849 const submissions_per_producer = 64;
850 var pool = try TestThreadPool.init(std.testing.allocator, .{
851 .workers = 4,
852 .tasks = producer_count * submissions_per_producer,
853 });
854 defer pool.deinit(std.testing.allocator);
855 try pool.activate();
856
857 var counter = std.atomic.Value(usize).init(0);
858 var failed = std.atomic.Value(bool).init(false);
859 var contexts: [producer_count]ProducerContext = undefined;
860 var producers: [producer_count]thread.JoinHandle = undefined;
861 for (&contexts, &producers) |*context, *producer| {
862 context.* = .{
863 .pool = &pool,
864 .counter = &counter,
865 .submissions = submissions_per_producer,
866 .failed = &failed,
867 };
868 producer.* = try thread.spawn(produceTasks, .{context});
869 }
870 for (producers) |producer| producer.join();
871
872 try std.testing.expect(!failed.load(.acquire));
873 try std.testing.expect(try pool.drain(
874 .system(),
875 .fromNanoseconds(5 * std.time.ns_per_s),
876 ));
877 try std.testing.expectEqual(@as(usize, producer_count * submissions_per_producer), counter.load(.acquire));
878 }
879
880 test "ThreadPool drain deadline pauses across suspend gaps and ignores wall jumps" {
881 var clock = time.FakeClock.zero();
882 const deadline = try DrainDeadline.init(
883 clock.awakeClock(),
884 .fromMilliseconds(5),
885 );
886
887 clock.setWall(.fromNanoseconds(std.math.maxInt(u64)));
888 try std.testing.expect(!(try deadline.reached()));
889
890 clock.suspendGap(.fromMilliseconds(10));
891 try std.testing.expect(!(try deadline.reached()));
892
893 clock.advance(.fromMilliseconds(5));
894 try std.testing.expect(try deadline.reached());
895 }
896
897 test "ThreadPool drain times out" {
898 var pool = try TestThreadPool.init(std.testing.allocator, .{ .workers = 1, .tasks = 1 });
899 defer pool.deinit(std.testing.allocator);
900 try pool.activate();
901
902 var release = std.atomic.Value(bool).init(false);
903 var started = std.atomic.Value(usize).init(0);
904 var completed = std.atomic.Value(usize).init(0);
905 var context = GateContext{
906 .release = &release,
907 .started = &started,
908 .completed = &completed,
909 };
910 try pool.submit(.{ .wait_for_release = &context });
911 try std.testing.expect(!(try pool.drain(
912 .system(),
913 .fromNanoseconds(1 * std.time.ns_per_ms),
914 )));
915
916 release.store(true, .release);
917 try std.testing.expect(try pool.drain(
918 .system(),
919 .fromNanoseconds(10 * std.time.ns_per_s),
920 ));
921 }
922
923 test "ThreadPool shutdown cancels queued caller-owned contexts" {
924 var pool = try TestThreadPool.init(std.testing.allocator, .{ .workers = 2, .tasks = 10 });
925 try pool.activate();
926
927 var started = std.atomic.Value(usize).init(0);
928 var completed = std.atomic.Value(usize).init(0);
929 var context = ShutdownContext{
930 .shutdown = &pool.state.shutdown,
931 .started = &started,
932 .completed = &completed,
933 };
934 for (0..10) |_| try pool.submit(.{ .run_until_shutdown = &context });
935
936 var rounds: usize = 0;
937 while (started.load(.acquire) < 2 and rounds < 5000) : (rounds += 1) {
938 sleepNanoseconds(1 * std.time.ns_per_ms);
939 }
940 try std.testing.expectEqual(@as(usize, 2), started.load(.acquire));
941
942 pool.deinit(std.testing.allocator);
943 try std.testing.expectEqual(@as(usize, 2), completed.load(.acquire));
944 }
945
946 fn sleepNanoseconds(ns: u64) void {
947 time.sleepNanoseconds(ns);
948 }