lib/simd/src/thread/pool.zig
daab053ee43316e1809a84551d573ddd1e5bf3d2
1 const std = @import("std");
2 const sys = @import("sys");
3 const simd = @import("../root.zig");
4 const spin = @import("spin.zig");
5 const wait = @import("wait.zig");
6 const autotune = simd.autotune;
7 const topology = simd.topology;
8
9 pub const max_clusters: usize = 33;
10 pub const all_clusters: usize = max_clusters - 1;
11 pub const max_threads: usize = 127;
12 pub const max_workers: usize = max_threads + 1;
13 pub const max_callers: usize = 60;
14 pub const caller_name_capacity: usize = 64;
15 pub const max_victims: usize = 4;
16 pub const max_configs: usize = 4;
17
18 pub const PoolWaitMode = enum(u8) {
19 block = 1,
20 spin,
21 };
22
23 pub const WaitType = enum(u8) {
24 block,
25 spin_shared,
26 spin_separate,
27 };
28
29 pub fn waitName(wait_type: WaitType) []const u8 {
30 return switch (wait_type) {
31 .block => "Block",
32 .spin_shared => "Single",
33 .spin_separate => "Separate",
34 };
35 }
36
37 pub const Exit = enum(u32) {
38 none,
39 loop,
40 thread,
41 };
42
43 pub const Config = struct {
44 spin_type: spin.SpinType = .pause,
45 wait_type: WaitType = .spin_separate,
46 reserved: [2]u8 = @splat(0),
47
48 pub fn formatName(
49 self: Config,
50 storage: []u8,
51 ) error{NoSpaceLeft}![]const u8 {
52 return std.fmt.bufPrint(
53 storage,
54 "{s:<14} {s:<9}",
55 .{ spin.name(self.spin_type), waitName(self.wait_type) },
56 );
57 }
58
59 pub fn candidates(
60 mode: PoolWaitMode,
61 storage: *[max_configs]Config,
62 ) []const Config {
63 return switch (mode) {
64 .block => blk: {
65 storage[0] = .{ .wait_type = .block };
66 break :blk storage[0..1];
67 },
68 .spin => blk: {
69 const detected = spin.detectSpin(0);
70 const spin_types: [2]spin.SpinType = .{ detected, .pause };
71 const spin_count: usize = if (detected == .pause) 1 else 2;
72 var count: usize = 0;
73 for (spin_types[0..spin_count]) |spin_type| {
74 storage[count] = .{
75 .spin_type = spin_type,
76 .wait_type = .spin_shared,
77 };
78 count += 1;
79 storage[count] = .{
80 .spin_type = spin_type,
81 .wait_type = .spin_separate,
82 };
83 count += 1;
84 }
85 break :blk storage[0..count];
86 },
87 };
88 }
89
90 fn encode(self: Config) u16 {
91 return @as(u16, @backingInt(self.spin_type)) |
92 @as(u16, @backingInt(self.wait_type)) << 8;
93 }
94
95 fn decode(bits: u16) Config {
96 return .{
97 .spin_type = @fromBackingInt(@intCast(@as(u8, @truncate(bits)))),
98 .wait_type = @fromBackingInt(@intCast(@as(u8, @truncate(bits >> 8)))),
99 };
100 }
101 };
102
103 pub const PoolWorkerMapping = struct {
104 cluster_index: u8 = 0,
105 max_cluster_workers: usize = 0,
106
107 pub fn init(
108 cluster_index: usize,
109 max_cluster_workers: usize,
110 ) error{ InvalidCluster, EmptyCluster }!PoolWorkerMapping {
111 if (cluster_index > all_clusters) return error.InvalidCluster;
112 if (max_cluster_workers == 0) return error.EmptyCluster;
113 return .{
114 .cluster_index = @intCast(cluster_index),
115 .max_cluster_workers = max_cluster_workers,
116 };
117 }
118
119 pub fn clusterIndex(self: PoolWorkerMapping) usize {
120 return self.cluster_index;
121 }
122
123 pub fn maxClusterWorkers(self: PoolWorkerMapping) usize {
124 return self.max_cluster_workers;
125 }
126
127 pub fn globalIndex(
128 self: PoolWorkerMapping,
129 worker_index: usize,
130 ) usize {
131 if (self.max_cluster_workers == 0) return worker_index;
132 if (self.cluster_index == all_clusters) {
133 std.debug.assert(worker_index < all_clusters);
134 return worker_index * self.max_cluster_workers;
135 }
136 std.debug.assert(worker_index < self.max_cluster_workers);
137 return self.cluster_index * self.max_cluster_workers + worker_index;
138 }
139 };
140
141 pub const ShuffledIota = struct {
142 coprime: u32 = 1,
143
144 pub fn init(coprime: u32) ShuffledIota {
145 return .{ .coprime = coprime };
146 }
147
148 pub fn next(
149 self: ShuffledIota,
150 current: u32,
151 size: u32,
152 ) u32 {
153 std.debug.assert(size != 0);
154 std.debug.assert(current < size);
155 return @intCast(
156 (@as(u64, current) + self.coprime) % @as(u64, size),
157 );
158 }
159
160 pub fn coprimeNonzero(a_value: u32, b_value: u32) bool {
161 std.debug.assert(a_value != 0);
162 std.debug.assert(b_value != 0);
163 var a = a_value;
164 var b = b_value;
165 const trailing_a = @ctz(a);
166 const trailing_b = @ctz(b);
167 if (@min(trailing_a, trailing_b) != 0) return false;
168 a >>= @intCast(trailing_a);
169 b >>= @intCast(trailing_b);
170 while (true) {
171 const previous_a = a;
172 a = @max(previous_a, b);
173 b = @min(previous_a, b);
174 if (b == 1) return true;
175 a -= b;
176 if (a == 0) return false;
177 a >>= @intCast(@ctz(a));
178 }
179 }
180
181 pub fn findAnotherCoprime(size: u32, start: u32) u32 {
182 std.debug.assert(size != 0);
183 if (size <= 2) return 1;
184 const increment: u32 = if (size & 1 == 0) 2 else 1;
185 var candidate = start | 1;
186 var attempts: u64 = 0;
187 const max_attempts = @as(u64, size) * 16;
188 while (attempts < max_attempts) : (attempts += 1) {
189 if (coprimeNonzero(candidate, size)) return candidate;
190 candidate +%= increment;
191 if (candidate == 0) candidate = 1;
192 }
193 unreachable;
194 }
195 };
196
197 pub const Caller = struct {
198 index: u8 = 0,
199
200 pub fn id(self: Caller) usize {
201 return self.index;
202 }
203 };
204
205 pub const CallerStats = struct {
206 runs: u64 = 0,
207 tasks: u64 = 0,
208 workers: u64 = 0,
209 elapsed_ns: u64 = 0,
210 };
211
212 pub const PoolStats = struct {
213 runs: u64 = 0,
214 serial_runs: u64 = 0,
215 threaded_runs: u64 = 0,
216 tasks: u64 = 0,
217 stolen_tasks: u64 = 0,
218 elapsed_ns: u64 = 0,
219 };
220
221 const CallerRegistry = struct {
222 guard: std.atomic.Mutex = .unlocked,
223 count: usize = 1,
224 lengths: [max_callers]u8 = @splat(0),
225 names: [max_callers][caller_name_capacity]u8 = @splat(@splat(0)),
226
227 fn add(
228 self: *CallerRegistry,
229 caller_name: []const u8,
230 ) error{ EmptyCallerName, CallerNameTooLong, TooManyCallers }!Caller {
231 if (caller_name.len == 0) return error.EmptyCallerName;
232 if (caller_name.len > caller_name_capacity) {
233 return error.CallerNameTooLong;
234 }
235 while (!self.guard.tryLock()) std.atomic.spinLoopHint();
236 defer self.guard.unlock();
237 for (self.names[1..self.count], 1..) |stored, index| {
238 const length = self.lengths[index];
239 if (std.mem.eql(u8, stored[0..length], caller_name)) {
240 return .{ .index = @intCast(index) };
241 }
242 }
243 if (self.count == max_callers) return error.TooManyCallers;
244 const index = self.count;
245 @memcpy(self.names[index][0..caller_name.len], caller_name);
246 self.lengths[index] = @intCast(caller_name.len);
247 self.count += 1;
248 return .{ .index = @intCast(index) };
249 }
250
251 fn name(self: *CallerRegistry, caller: Caller) ?[]const u8 {
252 while (!self.guard.tryLock()) std.atomic.spinLoopHint();
253 defer self.guard.unlock();
254 if (caller.index == 0 or caller.index >= self.count) return null;
255 const length = self.lengths[caller.index];
256 return self.names[caller.index][0..length];
257 }
258 };
259
260 var caller_registry = CallerRegistry{};
261
262 pub fn addCaller(
263 name: []const u8,
264 ) error{ EmptyCallerName, CallerNameTooLong, TooManyCallers }!Caller {
265 return caller_registry.add(name);
266 }
267
268 pub fn callerName(caller: Caller) ?[]const u8 {
269 return caller_registry.name(caller);
270 }
271
272 pub fn workerRange(
273 begin: u64,
274 end: u64,
275 worker_count: usize,
276 worker_index: usize,
277 ) error{ InvalidRange, EmptyWorkers, WorkerOutOfBounds, TooManyTasks }!struct {
278 begin: u64,
279 end: u64,
280 } {
281 if (begin > end) return error.InvalidRange;
282 if (worker_count == 0) return error.EmptyWorkers;
283 if (worker_index >= worker_count) return error.WorkerOutOfBounds;
284 const task_count_u64 = end - begin;
285 if (task_count_u64 > std.math.maxInt(usize)) return error.TooManyTasks;
286 const task_count: usize = @intCast(task_count_u64);
287 const minimum = task_count / worker_count;
288 const remainder = task_count % worker_count;
289 const local_begin = worker_index * minimum + @min(worker_index, remainder);
290 const local_count = minimum + @intFromBool(worker_index < remainder);
291 return .{
292 .begin = begin + local_begin,
293 .end = begin + local_begin + local_count,
294 };
295 }
296
297 const Worker = struct {
298 begin: std.atomic.Value(usize) = std.atomic.Value(usize).init(0),
299 end: usize = 0,
300 last_tasks: usize = 0,
301 last_stolen: usize = 0,
302 wait_epoch: std.atomic.Value(u32) = std.atomic.Value(u32).init(0),
303 barrier_epoch: std.atomic.Value(u32) = std.atomic.Value(u32).init(0),
304 victims: [max_victims]u8 = @splat(0),
305 victim_count: u8 = 0,
306 padding: [if (@sizeOf(usize) == 8) 19 else 35]u8 = @splat(0),
307 };
308
309 comptime {
310 if (@sizeOf(Worker) != 64) @compileError("thread worker must occupy one cache line");
311 }
312
313 const Callback = *const fn (*anyopaque, u64, usize) void;
314 const AutoTuner = autotune.AutoTune(Config, max_configs, 30);
315
316 pub const ThreadPool = struct {
317 pub const State = enum(u8) {
318 initialization,
319 steady,
320 teardown,
321 };
322
323 pub const Capacity = struct {
324 requested_threads: usize,
325 threads: u8,
326 workers: u8,
327
328 pub fn derive(requested_threads: usize) Capacity {
329 const supported = topology.haveThreadingSupport() and
330 sys.thread.threadsSupported();
331 const thread_count: usize = if (supported)
332 @min(requested_threads, max_threads)
333 else
334 0;
335 const worker_count: usize = thread_count + 1;
336 return .{
337 .requested_threads = requested_threads,
338 .threads = @intCast(thread_count),
339 .workers = @intCast(worker_count),
340 };
341 }
342
343 pub fn wasClamped(self: Capacity) bool {
344 return self.requested_threads != self.threads;
345 }
346 };
347
348 pub const StartError = sys.thread.SpawnError || error{
349 AlreadyDeinitialized,
350 PoolMoved,
351 };
352
353 pub const RunError = StartError || error{
354 Busy,
355 InvalidRange,
356 TooManyTasks,
357 };
358
359 state: State = .initialization,
360 capacity: Capacity,
361 mapping: PoolWorkerMapping,
362 address: usize = 0,
363 wait_mode: std.atomic.Value(u8) =
364 std.atomic.Value(u8).init(@backingInt(PoolWaitMode.block)),
365 config_bits: std.atomic.Value(u16) =
366 std.atomic.Value(u16).init((Config{ .wait_type = .block }).encode()),
367 epoch: std.atomic.Value(u32) = std.atomic.Value(u32).init(0),
368 work_available: std.atomic.Value(bool) =
369 std.atomic.Value(bool).init(false),
370 exiting: std.atomic.Value(bool) = std.atomic.Value(bool).init(false),
371 busy: std.atomic.Value(bool) = std.atomic.Value(bool).init(false),
372 synchronization: sys.thread.Mutex = .{},
373 done: sys.thread.Condition = .{},
374 started_threads: usize = 0,
375 handles: [max_threads]sys.thread.JoinHandle = undefined,
376 workers: [max_workers]Worker align(64) = @splat(.{}),
377 task_context: *anyopaque = undefined,
378 task_callback: Callback = undefined,
379 current_begin: u64 = 0,
380 current_tasks: usize = 0,
381 stats: PoolStats = .{},
382 caller_stats: [max_callers]CallerStats = @splat(.{}),
383 block_tuner: AutoTuner = .{},
384 spin_tuner: AutoTuner = .{},
385
386 pub fn init(
387 requested_threads: usize,
388 mapping: PoolWorkerMapping,
389 ) ThreadPool {
390 var self = ThreadPool{
391 .capacity = Capacity.derive(requested_threads),
392 .mapping = mapping,
393 };
394 var candidates: [max_configs]Config = undefined;
395 self.block_tuner.setCandidates(
396 Config.candidates(.block, &candidates),
397 ) catch unreachable;
398 self.spin_tuner.setCandidates(
399 Config.candidates(.spin, &candidates),
400 ) catch unreachable;
401 self.config_bits.store(
402 self.block_tuner.nextConfig().encode(),
403 .monotonic,
404 );
405 self.configureWorkers();
406 return self;
407 }
408
409 pub fn maxThreads() usize {
410 if (!topology.haveThreadingSupport() or
411 !sys.thread.threadsSupported())
412 {
413 return 0;
414 }
415 var affinity = topology.LogicalProcessorSet{};
416 if (topology.getThreadAffinity(&affinity)) {
417 const count = affinity.count();
418 if (count != 0) return @min(count - 1, max_threads);
419 }
420 return @min(
421 topology.totalLogicalProcessors() -| 1,
422 max_threads,
423 );
424 }
425
426 pub fn numThreadsFromCores(
427 allocator: std.mem.Allocator,
428 ) std.mem.Allocator.Error!usize {
429 if (!topology.haveThreadingSupport() or
430 !sys.thread.threadsSupported())
431 {
432 return 0;
433 }
434 var detected = try topology.init(allocator);
435 defer detected.deinit(allocator);
436 if (detected.packages.len == 0) return maxThreads();
437 return @min(detected.packages[0].cores.len -| 1, max_threads);
438 }
439
440 pub fn start(self: *ThreadPool) StartError!void {
441 if (self.state == .teardown) return error.AlreadyDeinitialized;
442 if (self.state == .steady) {
443 try self.requireStableAddress();
444 return;
445 }
446 self.address = @intFromPtr(self);
447 if (self.capacity.threads == 0) {
448 self.state = .steady;
449 return;
450 }
451 var spawned: usize = 0;
452 while (spawned < self.capacity.threads) : (spawned += 1) {
453 self.handles[spawned] = sys.thread.spawn(workerMain, .{
454 WorkerContext{
455 .pool = self,
456 .worker_index = spawned + 1,
457 },
458 }) catch |err| {
459 self.stopSpawned(spawned);
460 return err;
461 };
462 nameWorker(self.handles[spawned], spawned);
463 }
464 self.synchronization.lock();
465 while (self.started_threads != self.capacity.threads) {
466 self.done.wait(&self.synchronization);
467 }
468 self.synchronization.unlock();
469 self.state = .steady;
470 }
471
472 pub fn deinit(self: *ThreadPool) void {
473 if (self.state == .teardown) return;
474 if (self.state == .initialization) {
475 self.state = .teardown;
476 self.address = 0;
477 return;
478 }
479 self.requireStableAddress() catch unreachable;
480 std.debug.assert(!self.busy.load(.acquire));
481 if (self.capacity.threads != 0) {
482 self.signalExit();
483 for (self.handles[0..self.capacity.threads]) |handle| {
484 handle.join();
485 }
486 }
487 self.state = .teardown;
488 self.address = 0;
489 }
490
491 pub fn numWorkers(self: *const ThreadPool) usize {
492 return self.capacity.workers;
493 }
494
495 pub fn wasClamped(self: *const ThreadPool) bool {
496 return self.capacity.wasClamped();
497 }
498
499 pub fn setWaitMode(
500 self: *ThreadPool,
501 mode: PoolWaitMode,
502 ) (error{Busy} || StartError)!void {
503 if (self.state == .teardown) return error.AlreadyDeinitialized;
504 if (self.busy.cmpxchgStrong(
505 false,
506 true,
507 .acquire,
508 .monotonic,
509 ) != null) return error.Busy;
510 defer self.busy.store(false, .release);
511 self.wait_mode.store(@backingInt(mode), .release);
512 const mode_tuner = self.tuner(mode);
513 const selected = mode_tuner.best() orelse mode_tuner.nextConfig();
514 self.config_bits.store(selected.encode(), .release);
515 if (self.state == .steady and self.capacity.threads != 0) {
516 try self.requireStableAddress();
517 const epoch = self.wakeWorkers(false, true);
518 self.waitForWorkers(epoch);
519 }
520 }
521
522 pub fn waitMode(self: *const ThreadPool) PoolWaitMode {
523 return @fromBackingInt(@intCast(self.wait_mode.load(.acquire)));
524 }
525
526 pub fn config(self: *const ThreadPool) Config {
527 return Config.decode(self.config_bits.load(.acquire));
528 }
529
530 pub fn autoTuneComplete(self: *const ThreadPool) bool {
531 return self.tunerConst(self.waitMode()).best() != null;
532 }
533
534 pub fn autoTuneCosts(self: *ThreadPool) []autotune.CostDistribution {
535 return self.tuner(self.waitMode()).costs();
536 }
537
538 pub fn statsSnapshot(self: *ThreadPool) PoolStats {
539 self.synchronization.lock();
540 defer self.synchronization.unlock();
541 return self.stats;
542 }
543
544 pub fn callerStats(
545 self: *ThreadPool,
546 caller: Caller,
547 ) ?CallerStats {
548 if (caller.index >= max_callers) return null;
549 self.synchronization.lock();
550 defer self.synchronization.unlock();
551 return self.caller_stats[caller.index];
552 }
553
554 pub fn resetStats(self: *ThreadPool) error{Busy}!void {
555 if (self.busy.cmpxchgStrong(
556 false,
557 true,
558 .acquire,
559 .monotonic,
560 ) != null) return error.Busy;
561 defer self.busy.store(false, .release);
562 self.synchronization.lock();
563 defer self.synchronization.unlock();
564 self.stats = .{};
565 self.caller_stats = @splat(.{});
566 }
567
568 pub fn globalWorkerIndex(
569 self: *const ThreadPool,
570 local_worker_index: usize,
571 ) error{WorkerOutOfBounds}!usize {
572 if (local_worker_index >= self.numWorkers()) {
573 return error.WorkerOutOfBounds;
574 }
575 return self.mapping.globalIndex(local_worker_index);
576 }
577
578 pub fn run(
579 self: *ThreadPool,
580 begin: u64,
581 end: u64,
582 context: anytype,
583 comptime body: fn (@TypeOf(context), u64, usize) void,
584 ) RunError!void {
585 return self.runWithCaller(begin, end, .{}, context, body);
586 }
587
588 pub fn runWithCaller(
589 self: *ThreadPool,
590 begin: u64,
591 end: u64,
592 caller: Caller,
593 context: anytype,
594 comptime body: fn (@TypeOf(context), u64, usize) void,
595 ) RunError!void {
596 if (begin > end) return error.InvalidRange;
597 const task_count_u64 = end - begin;
598 if (task_count_u64 > std.math.maxInt(usize)) {
599 return error.TooManyTasks;
600 }
601 const task_count: usize = @intCast(task_count_u64);
602 const Context = @TypeOf(context);
603 const Adapter = CallbackAdapter(Context, body);
604 var context_storage = context;
605 const started_ns = nowNanoseconds();
606 if (task_count <= 1 or self.numWorkers() == 1) {
607 var task = begin;
608 while (task < end) : (task += 1) body(context, task, 0);
609 self.recordRun(
610 caller,
611 task_count,
612 false,
613 0,
614 elapsedNanoseconds(started_ns),
615 );
616 return;
617 }
618 if (self.busy.cmpxchgStrong(
619 false,
620 true,
621 .acquire,
622 .monotonic,
623 ) != null) return error.Busy;
624 defer self.busy.store(false, .release);
625 try self.start();
626 try self.requireStableAddress();
627 self.task_context = @ptrCast(&context_storage);
628 self.task_callback = Adapter.call;
629 self.current_begin = begin;
630 self.current_tasks = task_count;
631 self.divideRange(begin, end);
632 for (self.workers[0..self.numWorkers()]) |*worker| {
633 worker.last_tasks = 0;
634 worker.last_stolen = 0;
635 }
636 const epoch = self.wakeWorkers(
637 true,
638 self.config().wait_type == .block,
639 );
640 self.runWorker(0);
641 self.waitForWorkers(epoch);
642 var stolen_tasks: usize = 0;
643 for (self.workers[0..self.numWorkers()]) |worker| {
644 stolen_tasks += worker.last_stolen;
645 }
646 const elapsed_ns = elapsedNanoseconds(started_ns);
647 self.recordRun(
648 caller,
649 task_count,
650 true,
651 stolen_tasks,
652 elapsed_ns,
653 );
654 self.notifyAutotune(elapsed_ns);
655 }
656
657 fn configureWorkers(self: *ThreadPool) void {
658 const worker_count: u32 = self.capacity.workers;
659 for (self.workers[0..worker_count], 0..) |*worker, worker_index| {
660 worker.* = .{};
661 worker.victim_count = @intCast(@min(max_victims, worker_count));
662 const coprime = ShuffledIota.findAnotherCoprime(
663 worker_count,
664 @intCast((worker_index + 1) * 257 + worker_index * 13),
665 );
666 const shuffled = ShuffledIota.init(coprime);
667 worker.victims[0] = @intCast(worker_index);
668 for (1..worker.victim_count) |victim_index| {
669 worker.victims[victim_index] = @intCast(shuffled.next(
670 worker.victims[victim_index - 1],
671 worker_count,
672 ));
673 }
674 }
675 }
676
677 fn divideRange(self: *ThreadPool, begin: u64, end: u64) void {
678 for (self.workers[0..self.numWorkers()], 0..) |*worker, index| {
679 const range = workerRange(
680 0,
681 end - begin,
682 self.numWorkers(),
683 index,
684 ) catch unreachable;
685 worker.begin.store(@intCast(range.begin), .monotonic);
686 worker.end = @intCast(range.end);
687 }
688 }
689
690 fn runWorker(self: *ThreadPool, worker_index: usize) void {
691 var task_count: usize = 0;
692 var stolen_count: usize = 0;
693 if (self.current_tasks <= self.numWorkers()) {
694 const offset = self.workers[0].begin.load(.monotonic) +
695 worker_index;
696 const end = self.workers[self.numWorkers() - 1].end;
697 if (offset < end) {
698 self.task_callback(
699 self.task_context,
700 self.current_begin + offset,
701 worker_index,
702 );
703 task_count = 1;
704 }
705 } else {
706 const worker = &self.workers[worker_index];
707 for (worker.victims[0..worker.victim_count]) |victim_index| {
708 const victim = &self.workers[victim_index];
709 while (true) {
710 const offset = victim.begin.fetchAdd(1, .monotonic);
711 if (offset >= victim.end) break;
712 self.task_callback(
713 self.task_context,
714 self.current_begin + offset,
715 worker_index,
716 );
717 task_count += 1;
718 stolen_count += @intFromBool(victim_index != worker_index);
719 }
720 }
721 }
722 self.workers[worker_index].last_tasks = task_count;
723 self.workers[worker_index].last_stolen = stolen_count;
724 }
725
726 fn wakeWorkers(
727 self: *ThreadPool,
728 work_available: bool,
729 wake_blocked: bool,
730 ) u32 {
731 self.work_available.store(work_available, .monotonic);
732 const epoch = self.epoch.load(.monotonic) +% 1;
733 self.epoch.store(epoch, .release);
734 for (self.workers[1..self.numWorkers()]) |*worker| {
735 worker.wait_epoch.store(epoch, .release);
736 }
737 if (wake_blocked and self.capacity.threads != 0) {
738 wait.wakeAll(&self.workers[1].wait_epoch);
739 }
740 return epoch;
741 }
742
743 fn waitForWorkers(self: *ThreadPool, epoch: u32) void {
744 spin.callWithSpin(
745 self.config().spin_type,
746 BarrierContext{ .pool = self, .epoch = epoch },
747 waitAtBarrier,
748 );
749 }
750
751 fn waitForWork(
752 self: *ThreadPool,
753 worker_index: usize,
754 observed_epoch: u32,
755 ) ?u32 {
756 while (true) {
757 if (self.exiting.load(.acquire)) return null;
758 const config_value = self.config();
759 switch (config_value.wait_type) {
760 .block => {
761 const next_epoch = wait.blockUntilDifferent(
762 observed_epoch,
763 &self.workers[1].wait_epoch,
764 );
765 if (self.exiting.load(.acquire)) return null;
766 return next_epoch;
767 },
768 .spin_shared, .spin_separate => {
769 const watched = if (config_value.wait_type ==
770 .spin_separate)
771 &self.workers[worker_index].wait_epoch
772 else
773 &self.epoch;
774 var next_epoch: u32 = observed_epoch;
775 spin.callWithSpin(
776 config_value.spin_type,
777 SpinWaitContext{
778 .previous = observed_epoch,
779 .watched = watched,
780 .next = &next_epoch,
781 },
782 waitUntilDifferent,
783 );
784 if (self.exiting.load(.acquire)) return null;
785 return next_epoch;
786 },
787 }
788 }
789 }
790
791 fn workerStarted(self: *ThreadPool) void {
792 self.synchronization.lock();
793 self.started_threads += 1;
794 self.done.broadcast();
795 self.synchronization.unlock();
796 }
797
798 fn workerReached(self: *ThreadPool, worker_index: usize, epoch: u32) void {
799 self.workers[worker_index].barrier_epoch.store(epoch, .release);
800 }
801
802 fn stopSpawned(self: *ThreadPool, spawned: usize) void {
803 self.signalExit();
804 for (self.handles[0..spawned]) |handle| handle.join();
805 self.exiting.store(false, .monotonic);
806 self.epoch.store(0, .monotonic);
807 self.started_threads = 0;
808 self.address = 0;
809 self.work_available.store(false, .monotonic);
810 self.configureWorkers();
811 }
812
813 fn signalExit(self: *ThreadPool) void {
814 self.exiting.store(true, .release);
815 const epoch = self.epoch.load(.monotonic) +% 1;
816 self.epoch.store(epoch, .release);
817 for (self.workers[1..self.numWorkers()]) |*worker| {
818 worker.wait_epoch.store(epoch, .release);
819 }
820 if (self.capacity.threads != 0) {
821 wait.wakeAll(&self.workers[1].wait_epoch);
822 }
823 }
824
825 fn requireStableAddress(self: *const ThreadPool) error{PoolMoved}!void {
826 if (self.address != @intFromPtr(self)) return error.PoolMoved;
827 }
828
829 fn recordRun(
830 self: *ThreadPool,
831 caller: Caller,
832 tasks: usize,
833 threaded: bool,
834 stolen_tasks: usize,
835 elapsed_ns: u64,
836 ) void {
837 self.synchronization.lock();
838 defer self.synchronization.unlock();
839 self.stats.runs +|= 1;
840 self.stats.serial_runs +|= @intFromBool(!threaded);
841 self.stats.threaded_runs +|= @intFromBool(threaded);
842 self.stats.tasks +|= tasks;
843 self.stats.stolen_tasks +|= stolen_tasks;
844 self.stats.elapsed_ns +|= elapsed_ns;
845 const caller_index = if (caller.index < max_callers)
846 caller.index
847 else
848 0;
849 const stats = &self.caller_stats[caller_index];
850 stats.runs +|= 1;
851 stats.tasks +|= tasks;
852 stats.workers +|= if (threaded) self.numWorkers() else 1;
853 stats.elapsed_ns +|= elapsed_ns;
854 }
855
856 fn notifyAutotune(self: *ThreadPool, elapsed_ns: u64) void {
857 const mode = self.waitMode();
858 const selected = blk: {
859 const tuner_value = self.tuner(mode);
860 if (tuner_value.best()) |best| break :blk best.*;
861 tuner_value.notifyCost(@max(elapsed_ns, 1));
862 break :blk if (tuner_value.best()) |best|
863 best.*
864 else
865 tuner_value.nextConfig().*;
866 };
867 self.config_bits.store(selected.encode(), .release);
868 }
869
870 fn tuner(self: *ThreadPool, mode: PoolWaitMode) *AutoTuner {
871 return switch (mode) {
872 .block => &self.block_tuner,
873 .spin => &self.spin_tuner,
874 };
875 }
876
877 fn tunerConst(self: *const ThreadPool, mode: PoolWaitMode) *const AutoTuner {
878 return switch (mode) {
879 .block => &self.block_tuner,
880 .spin => &self.spin_tuner,
881 };
882 }
883 };
884
885 const WorkerContext = struct {
886 pool: *ThreadPool,
887 worker_index: usize,
888 };
889
890 const SpinWaitContext = struct {
891 previous: u32,
892 watched: *const std.atomic.Value(u32),
893 next: *u32,
894 };
895
896 fn waitUntilDifferent(context: SpinWaitContext, policy: anytype) void {
897 context.next.* = policy.untilDifferent(
898 context.previous,
899 context.watched,
900 ).value;
901 }
902
903 const BarrierContext = struct {
904 pool: *ThreadPool,
905 epoch: u32,
906 };
907
908 fn waitAtBarrier(context: BarrierContext, policy: anytype) void {
909 for (context.pool.workers[1..context.pool.numWorkers()]) |*worker| {
910 _ = policy.untilEqual(context.epoch, &worker.barrier_epoch);
911 }
912 }
913
914 fn workerMain(context: WorkerContext) void {
915 const pool = context.pool;
916 pool.workerStarted();
917 var observed_epoch: u32 = 0;
918 while (pool.waitForWork(context.worker_index, observed_epoch)) |epoch| {
919 if (pool.work_available.load(.acquire)) {
920 pool.runWorker(context.worker_index);
921 }
922 pool.workerReached(context.worker_index, epoch);
923 observed_epoch = epoch;
924 }
925 }
926
927 fn nameWorker(handle: sys.thread.JoinHandle, index: usize) void {
928 if (comptime std.Thread.max_name_len < 9) return;
929 var storage: [std.Thread.max_name_len]u8 = undefined;
930 const name_value = std.fmt.bufPrint(
931 &storage,
932 "worker{d:0>3}",
933 .{index},
934 ) catch return;
935 handle.setName(name_value) catch {};
936 }
937
938 fn CallbackAdapter(
939 comptime Context: type,
940 comptime body: fn (Context, u64, usize) void,
941 ) type {
942 return struct {
943 fn call(
944 opaque_context: *anyopaque,
945 task: u64,
946 worker: usize,
947 ) void {
948 const context: *Context = @ptrCast(@alignCast(opaque_context));
949 body(context.*, task, worker);
950 }
951 };
952 }
953
954 fn nowNanoseconds() u64 {
955 return @intCast(@max(sys.time.nanoTimestamp(), 0));
956 }
957
958 fn elapsedNanoseconds(started: u64) u64 {
959 return nowNanoseconds() -| started;
960 }
961
962 test "Highway shuffled iota detects coprimes and exact permutations" {
963 for (1..40) |size_usize| {
964 const size: u32 = @intCast(size_usize);
965 const coprime = ShuffledIota.findAnotherCoprime(size, 1);
966 try std.testing.expect(ShuffledIota.coprimeNonzero(coprime, size));
967 const shuffled = ShuffledIota.init(coprime);
968 for (0..size) |start_usize| {
969 var visited: [40]u8 = @splat(0);
970 var current: u32 = @intCast(start_usize);
971 for (0..size) |_| {
972 visited[current] += 1;
973 current = shuffled.next(current, size);
974 }
975 for (visited[0..size]) |count| {
976 try std.testing.expectEqual(@as(u8, 1), count);
977 }
978 }
979 }
980 for (1..500) |value| {
981 try std.testing.expect(ShuffledIota.coprimeNonzero(1, @intCast(value)));
982 try std.testing.expect(ShuffledIota.coprimeNonzero(@intCast(value), 1));
983 }
984 }
985
986 test "Highway binary coprime agrees for powers products and primes" {
987 for (1..20) |i| {
988 const a = @as(u32, 1) << @intCast(i);
989 for (1..20) |j| {
990 const b = @as(u32, 1) << @intCast(j);
991 try std.testing.expect(!ShuffledIota.coprimeNonzero(a, b));
992 }
993 }
994 for (1..30) |i| {
995 const power = @as(u32, 1) << @intCast(i);
996 try std.testing.expect(ShuffledIota.coprimeNonzero(power, power + 1));
997 try std.testing.expect(ShuffledIota.coprimeNonzero(power, power - 1));
998 try std.testing.expect(ShuffledIota.coprimeNonzero(power + 1, power));
999 try std.testing.expect(ShuffledIota.coprimeNonzero(power - 1, power));
1000 }
1001 var random_state: u32 = 0x4f1b_c3d9;
1002 for (0..5_000) |_| {
1003 random_state = random_state *% 1_664_525 +% 1_013_904_223;
1004 const x = (random_state & 0xfff7) + 2;
1005 random_state = random_state *% 1_664_525 +% 1_013_904_223;
1006 const y = (random_state & 0xfff7) + 2;
1007 const product = x * y;
1008 try std.testing.expect(!ShuffledIota.coprimeNonzero(product, x));
1009 try std.testing.expect(!ShuffledIota.coprimeNonzero(product, y));
1010 try std.testing.expect(!ShuffledIota.coprimeNonzero(x, product));
1011 try std.testing.expect(!ShuffledIota.coprimeNonzero(y, product));
1012 }
1013 const primes = [_]u32{
1014 2, 3, 5, 7, 11, 13, 17, 19, 23, 29, 31, 37,
1015 41, 43, 47, 53, 59, 61, 67, 71, 73, 79, 83, 89,
1016 97, 101, 103, 107, 109, 113, 127, 131, 137, 139, 149, 151,
1017 157, 163, 167, 173, 179, 181, 191, 193, 197, 199, 211, 223,
1018 227, 229, 233, 239, 241, 251, 257, 263, 269, 271,
1019 };
1020 for (primes, 0..) |a, i| {
1021 for (primes[i + 1 ..]) |b| {
1022 try std.testing.expect(ShuffledIota.coprimeNonzero(a, b));
1023 try std.testing.expect(ShuffledIota.coprimeNonzero(b, a));
1024 }
1025 }
1026 }
1027
1028 test "Highway independent shuffles retain coverage with bounded contention" {
1029 var shuffles: [40]ShuffledIota = @splat(.{});
1030 var current: [40]u32 = @splat(0);
1031 var visited_all: [40]u8 = @splat(0);
1032 for (1..40) |size_usize| {
1033 const size: u32 = @intCast(size_usize);
1034 @memset(visited_all[0..size], 0);
1035 for (0..size) |index_usize| {
1036 const index: u32 = @intCast(index_usize);
1037 shuffles[index] = .{ .coprime = ShuffledIota.findAnotherCoprime(
1038 size,
1039 (index + 1) * 257 + index * 13,
1040 ) };
1041 current[index] = index;
1042 }
1043 var bad_steps: usize = 0;
1044 for (0..size) |_| {
1045 var visited: [40]u8 = @splat(0);
1046 for (current[0..size]) |value| {
1047 visited[value] += 1;
1048 visited_all[value] = 1;
1049 }
1050 var contended: usize = 0;
1051 var maximum: u8 = 0;
1052 for (visited[0..size]) |count| {
1053 contended += @intFromBool(count > 1);
1054 maximum = @max(maximum, count);
1055 }
1056 const expected: usize = @intFromFloat(
1057 std.math.sqrt(@as(f32, @floatFromInt(size))) * 2.0,
1058 );
1059 bad_steps += @intFromBool(contended > expected and maximum > 3);
1060 for (current[0..size], 0..) |*value, index| {
1061 value.* = shuffles[index].next(value.*, size);
1062 }
1063 }
1064 for (visited_all[0..size]) |visited| {
1065 try std.testing.expectEqual(@as(u8, 1), visited);
1066 }
1067 try std.testing.expect(bad_steps < 4);
1068 }
1069 }
1070
1071 test "Highway pool configurations preserve layout and candidate order" {
1072 try std.testing.expectEqual(@as(usize, 4), @sizeOf(Config));
1073 var storage: [max_configs]Config = undefined;
1074 const block = Config.candidates(.block, &storage);
1075 try std.testing.expectEqual(@as(usize, 1), block.len);
1076 try std.testing.expectEqual(WaitType.block, block[0].wait_type);
1077 try std.testing.expectEqual(spin.SpinType.pause, block[0].spin_type);
1078 const block_config = block[0];
1079 const detected = spin.detectSpin(0);
1080 const candidates = Config.candidates(.spin, &storage);
1081 const expected_count: usize = if (detected == .pause) 2 else 4;
1082 try std.testing.expectEqual(expected_count, candidates.len);
1083 for (candidates, 0..) |candidate, index| {
1084 try std.testing.expect(candidate.wait_type != .block);
1085 try std.testing.expectEqual(
1086 if (index & 1 == 0)
1087 WaitType.spin_shared
1088 else
1089 WaitType.spin_separate,
1090 candidate.wait_type,
1091 );
1092 try std.testing.expectEqual(
1093 if (index < 2) detected else spin.SpinType.pause,
1094 candidate.spin_type,
1095 );
1096 }
1097 var name_storage: [64]u8 = undefined;
1098 const formatted = try block_config.formatName(&name_storage);
1099 try std.testing.expect(std.mem.startsWith(u8, formatted, "Pause"));
1100 try std.testing.expect(std.mem.endsWith(u8, formatted, "Block "));
1101 }
1102
1103 test "Highway worker range division is balanced and contiguous" {
1104 for (1..40) |worker_count| {
1105 for (0..80) |task_count| {
1106 var cursor: u64 = 11;
1107 var minimum: usize = std.math.maxInt(usize);
1108 var maximum: usize = 0;
1109 for (0..worker_count) |worker_index| {
1110 const range = try workerRange(
1111 11,
1112 11 + task_count,
1113 worker_count,
1114 worker_index,
1115 );
1116 try std.testing.expectEqual(cursor, range.begin);
1117 const count: usize = @intCast(range.end - range.begin);
1118 minimum = @min(minimum, count);
1119 maximum = @max(maximum, count);
1120 cursor = range.end;
1121 }
1122 try std.testing.expectEqual(11 + task_count, cursor);
1123 try std.testing.expect(maximum - minimum <= 1);
1124 }
1125 }
1126 }
1127
1128 test "Highway pool worker mapping preserves local and cluster indices" {
1129 const local = PoolWorkerMapping{};
1130 try std.testing.expectEqual(@as(usize, 7), local.globalIndex(7));
1131 const cluster = try PoolWorkerMapping.init(3, 8);
1132 try std.testing.expectEqual(@as(usize, 27), cluster.globalIndex(3));
1133 const across = try PoolWorkerMapping.init(all_clusters, 8);
1134 try std.testing.expectEqual(@as(usize, 24), across.globalIndex(3));
1135 try std.testing.expectError(
1136 error.InvalidCluster,
1137 PoolWorkerMapping.init(all_clusters + 1, 1),
1138 );
1139 try std.testing.expectError(
1140 error.EmptyCluster,
1141 PoolWorkerMapping.init(0, 0),
1142 );
1143 var pool = ThreadPool.init(3, cluster);
1144 defer pool.deinit();
1145 for (0..pool.numWorkers()) |worker| {
1146 try std.testing.expectEqual(
1147 cluster.globalIndex(worker),
1148 try pool.globalWorkerIndex(worker),
1149 );
1150 }
1151 try std.testing.expectError(
1152 error.WorkerOutOfBounds,
1153 pool.globalWorkerIndex(pool.numWorkers()),
1154 );
1155 }
1156
1157 const HitContext = struct {
1158 begin: u64,
1159 hits: []std.atomic.Value(u32),
1160 worker_bits: *std.atomic.Value(usize),
1161 };
1162
1163 fn recordHit(context: *HitContext, task: u64, worker: usize) void {
1164 _ = context.hits[@intCast(task - context.begin)].fetchAdd(1, .monotonic);
1165 _ = context.worker_bits.fetchOr(
1166 @as(usize, 1) << @intCast(worker),
1167 .monotonic,
1168 );
1169 }
1170
1171 test "Highway thread pool runs every task once in block and spin modes" {
1172 if (!topology.haveThreadingSupport() or
1173 !sys.thread.threadsSupported())
1174 {
1175 return error.SkipZigTest;
1176 }
1177 var pool = ThreadPool.init(@min(ThreadPool.maxThreads(), 6), .{});
1178 defer pool.deinit();
1179 var hits: [97]std.atomic.Value(u32) =
1180 @splat(std.atomic.Value(u32).init(0));
1181 var worker_bits = std.atomic.Value(usize).init(0);
1182 var context = HitContext{
1183 .begin = 23,
1184 .hits = &hits,
1185 .worker_bits = &worker_bits,
1186 };
1187 for ([_]PoolWaitMode{ .spin, .block }) |mode| {
1188 try pool.setWaitMode(mode);
1189 for (&hits) |*hit| hit.store(0, .monotonic);
1190 worker_bits.store(0, .monotonic);
1191 try pool.run(23, 120, &context, recordHit);
1192 for (&hits) |*hit| {
1193 try std.testing.expectEqual(
1194 @as(u32, 1),
1195 hit.load(.monotonic),
1196 );
1197 }
1198 try std.testing.expect(worker_bits.load(.monotonic) != 0);
1199 }
1200 }
1201
1202 test "Highway thread pool preserves ranges near the u64 limit" {
1203 if (!topology.haveThreadingSupport() or
1204 !sys.thread.threadsSupported() or ThreadPool.maxThreads() == 0)
1205 {
1206 return error.SkipZigTest;
1207 }
1208 const begin = std.math.maxInt(u64) - 17;
1209 var pool = ThreadPool.init(@min(ThreadPool.maxThreads(), 2), .{});
1210 defer pool.deinit();
1211 var hits: [17]std.atomic.Value(u32) =
1212 @splat(std.atomic.Value(u32).init(0));
1213 var worker_bits = std.atomic.Value(usize).init(0);
1214 var context = HitContext{
1215 .begin = begin,
1216 .hits = &hits,
1217 .worker_bits = &worker_bits,
1218 };
1219 try pool.run(begin, std.math.maxInt(u64), &context, recordHit);
1220 for (&hits) |*hit| {
1221 try std.testing.expectEqual(@as(u32, 1), hit.load(.monotonic));
1222 }
1223 }
1224
1225 const AssignmentContext = struct {
1226 expected_tasks: usize,
1227 expected_workers: usize,
1228 calls: *std.atomic.Value(usize),
1229 worker_bits: *std.atomic.Value(usize),
1230 };
1231
1232 fn recordAssignment(
1233 context: *AssignmentContext,
1234 task: u64,
1235 worker: usize,
1236 ) void {
1237 std.debug.assert(task < context.expected_tasks);
1238 std.debug.assert(worker < context.expected_workers);
1239 _ = context.calls.fetchAdd(1, .monotonic);
1240 _ = context.worker_bits.fetchOr(
1241 @as(usize, 1) << @intCast(worker),
1242 .monotonic,
1243 );
1244 }
1245
1246 test "Highway small assignments retain valid worker identities" {
1247 for ([_]usize{ 0, 1, 3, 5, 8 }) |requested| {
1248 var pool = ThreadPool.init(@min(ThreadPool.maxThreads(), requested), .{});
1249 defer pool.deinit();
1250 for (1..3) |multiplier| {
1251 const tasks = pool.numWorkers() * multiplier;
1252 var calls = std.atomic.Value(usize).init(0);
1253 var worker_bits = std.atomic.Value(usize).init(0);
1254 var context = AssignmentContext{
1255 .expected_tasks = tasks,
1256 .expected_workers = pool.numWorkers(),
1257 .calls = &calls,
1258 .worker_bits = &worker_bits,
1259 };
1260 try pool.run(0, tasks, &context, recordAssignment);
1261 try std.testing.expectEqual(tasks, calls.load(.monotonic));
1262 try std.testing.expect(
1263 @popCount(worker_bits.load(.monotonic)) <= pool.numWorkers(),
1264 );
1265 }
1266 }
1267 }
1268
1269 const SumContext = struct {
1270 counters: []std.atomic.Value(usize),
1271 };
1272
1273 fn addTask(context: *SumContext, task: u64, worker: usize) void {
1274 _ = context.counters[worker].fetchAdd(@intCast(task), .monotonic);
1275 }
1276
1277 test "Highway pool wait modes preserve the task sum and caller statistics" {
1278 if (!topology.haveThreadingSupport() or
1279 !sys.thread.threadsSupported())
1280 {
1281 return error.SkipZigTest;
1282 }
1283 const caller = try addCaller("thread-pool-test");
1284 try std.testing.expectEqual(
1285 caller.id(),
1286 (try addCaller("thread-pool-test")).id(),
1287 );
1288 try std.testing.expectEqualStrings(
1289 "thread-pool-test",
1290 callerName(caller).?,
1291 );
1292 var pool = ThreadPool.init(@min(ThreadPool.maxThreads(), 9), .{});
1293 defer pool.deinit();
1294 var counters: [max_workers]std.atomic.Value(usize) =
1295 @splat(std.atomic.Value(usize).init(0));
1296 var context = SumContext{ .counters = &counters };
1297 const task_count = pool.numWorkers() * 19;
1298 for ([_]PoolWaitMode{ .spin, .block }) |mode| {
1299 try pool.setWaitMode(mode);
1300 for (counters[0..pool.numWorkers()]) |*counter| {
1301 counter.store(0, .monotonic);
1302 }
1303 try pool.runWithCaller(
1304 0,
1305 task_count,
1306 caller,
1307 &context,
1308 addTask,
1309 );
1310 var actual: usize = 0;
1311 for (counters[0..pool.numWorkers()]) |*counter| {
1312 actual += counter.load(.monotonic);
1313 }
1314 try std.testing.expectEqual(
1315 task_count * (task_count - 1) / 2,
1316 actual,
1317 );
1318 }
1319 const caller_stats = pool.callerStats(caller).?;
1320 try std.testing.expectEqual(@as(u64, 2), caller_stats.runs);
1321 try std.testing.expectEqual(@as(u64, task_count * 2), caller_stats.tasks);
1322 try std.testing.expectEqual(@as(u64, 2), pool.statsSnapshot().threaded_runs);
1323 try pool.resetStats();
1324 try std.testing.expectEqual(PoolStats{}, pool.statsSnapshot());
1325 try std.testing.expectEqual(CallerStats{}, pool.callerStats(caller).?);
1326 }
1327
1328 const NestedContext = struct {
1329 begin: u64,
1330 end: u64,
1331 hits: []std.atomic.Value(u32),
1332 inner_calls: *std.atomic.Value(usize),
1333 };
1334
1335 fn innerTask(context: *NestedContext, task: u64, worker: usize) void {
1336 std.debug.assert(worker == 0);
1337 std.debug.assert(context.begin <= task);
1338 std.debug.assert(task < context.end);
1339 _ = context.inner_calls.fetchAdd(1, .monotonic);
1340 }
1341
1342 fn outerTask(context: *NestedContext, task: u64, _: usize) void {
1343 std.debug.assert(context.begin <= task);
1344 std.debug.assert(task < context.end);
1345 _ = context.hits[@intCast(task - context.begin)].fetchAdd(1, .monotonic);
1346 var inner = ThreadPool.init(0, .{});
1347 defer inner.deinit();
1348 inner.run(context.begin, context.end, context, innerTask) catch unreachable;
1349 }
1350
1351 test "Highway pools reuse shifted ranges and allow nested serial runs" {
1352 var hits: [20]std.atomic.Value(u32) =
1353 @splat(std.atomic.Value(u32).init(0));
1354 var inner_calls = std.atomic.Value(usize).init(0);
1355 for ([_]usize{ 0, 3, 6 }) |requested| {
1356 var pool = ThreadPool.init(@min(ThreadPool.maxThreads(), requested), .{});
1357 defer pool.deinit();
1358 for ([_]PoolWaitMode{ .spin, .block }) |mode| {
1359 try pool.setWaitMode(mode);
1360 for (0..20) |task_count| {
1361 for (0..8) |begin| {
1362 for (&hits) |*hit| hit.store(0, .monotonic);
1363 inner_calls.store(0, .monotonic);
1364 var context = NestedContext{
1365 .begin = begin,
1366 .end = begin + task_count,
1367 .hits = &hits,
1368 .inner_calls = &inner_calls,
1369 };
1370 try pool.run(
1371 context.begin,
1372 context.end,
1373 &context,
1374 outerTask,
1375 );
1376 for (hits[0..task_count]) |*hit| {
1377 try std.testing.expectEqual(
1378 @as(u32, 1),
1379 hit.load(.monotonic),
1380 );
1381 }
1382 try std.testing.expectEqual(
1383 task_count * task_count,
1384 inner_calls.load(.monotonic),
1385 );
1386 }
1387 }
1388 }
1389 }
1390 }
1391
1392 fn emptyTask(_: void, _: u64, _: usize) void {}
1393
1394 test "Highway live pools switch wait modes repeatedly" {
1395 if (!topology.haveThreadingSupport() or
1396 !sys.thread.threadsSupported() or ThreadPool.maxThreads() == 0)
1397 {
1398 return error.SkipZigTest;
1399 }
1400 var pool = ThreadPool.init(@min(ThreadPool.maxThreads(), 9), .{});
1401 defer pool.deinit();
1402 try pool.run(0, 2, {}, emptyTask);
1403 for (0..100) |iteration| {
1404 const mode: PoolWaitMode = if ((iteration * 17 + 5) & 1 == 0)
1405 .spin
1406 else
1407 .block;
1408 try pool.setWaitMode(mode);
1409 try std.testing.expectEqual(mode, pool.waitMode());
1410 try std.testing.expectEqual(
1411 mode == .block,
1412 pool.config().wait_type == .block,
1413 );
1414 }
1415 try pool.setWaitMode(.block);
1416 }
1417
1418 test "Highway pool autotuning visits candidates and converges" {
1419 if (!topology.haveThreadingSupport() or
1420 !sys.thread.threadsSupported() or ThreadPool.maxThreads() == 0)
1421 {
1422 return error.SkipZigTest;
1423 }
1424 var pool = ThreadPool.init(@min(ThreadPool.maxThreads(), 2), .{});
1425 defer pool.deinit();
1426 try pool.setWaitMode(.spin);
1427 var runs: usize = 0;
1428 while (!pool.autoTuneComplete() and runs < max_configs * 64) : (runs += 1) {
1429 try pool.run(0, 2, {}, emptyTask);
1430 }
1431 try std.testing.expect(pool.autoTuneComplete());
1432 try std.testing.expect(runs >= 30);
1433 const costs = pool.autoTuneCosts();
1434 try std.testing.expect(costs.len == 2 or costs.len == 4);
1435 for (costs) |*cost| {
1436 try std.testing.expect(
1437 cost.bufferedCount() != 0 or cost.onlineCount() != 0,
1438 );
1439 }
1440 }
1441
1442 test "Highway pool clamps capacity and rejects invalid ranges" {
1443 var pool = ThreadPool.init(max_threads + 100, .{});
1444 defer pool.deinit();
1445 try std.testing.expect(pool.wasClamped());
1446 try std.testing.expectEqual(max_workers, pool.numWorkers());
1447 try std.testing.expectError(
1448 error.InvalidRange,
1449 pool.run(2, 1, {}, struct {
1450 fn call(_: void, _: u64, _: usize) void {}
1451 }.call),
1452 );
1453 try pool.resetStats();
1454 try std.testing.expectEqual(PoolStats{}, pool.statsSnapshot());
1455 }