authorgravatar for kappaloris@gmail.comLoris Cro <kappaloris@gmail.com> 2025-11-04 21:11:40+01:00
committergravatar for kappaloris@gmail.comLoris Cro <kappaloris@gmail.com> 2025-11-07 12:43:25+01:00
log0156d68f0c9626074ca14d6c2cf677135e47e585
tree8527f8972db535a6c578bd1cec656c8c1bf3d1a0
parent19af9fa488ba7ddf2e7fd8f2f06ab594f7e23f91

Io.Threaded PoC reimplementation

This is a reimplementation of Io.Threaded that fixes the issues highlighted in the recent Zulip discussion. It's poorly tested but it does successfully run to completion the litmust test example that I offered in the discussion. This implementation has the following key design decisions: - `t.cpu_count` is used as the threadpool size. - `t.concurrency_limit` is used as the maximum number of "burst, one-shot" threads that can be spawned by `io.concurrent` past `t.cpu_count`. - `t.available_thread_count` is the number of threads in the pool that is not currently busy with work (the bookkeeping happens in the worker function). - `t.one_shot_thread_count` is the number of active threads that were spawned by `io.concurrent` past `t.cpu_count`. In this implementation: - `io.async` first tries to decrement `t.available_thread_count`. If there are no threads available, it tries to spawn a new one if possible, otherwise it runs the task immediately. - `io.concurrent` first tries to use a thread in the pool same as `io.async`, but on failure (no available threads and pool size limit reached) it tries to spawn a new one-shot thread. One shot threads run a different main function that just executes one task, decrements the number of active one shot threads, and then exits. A relevant future improvement is to have one-shot threads stay on for a few seconds (and potentially pick up a new task) to amortize spawning costs.

1 files changed, 114 insertions(+), 79 deletions(-)

lib/std/Io/Threaded.zig+114-79
...@@ -24,8 +24,10 @@ run_queue: std.SinglyLinkedList = .{},...@@ -24,8 +24,10 @@ run_queue: std.SinglyLinkedList = .{},
24join_requested: bool = false,24join_requested: bool = false,
25threads: std.ArrayListUnmanaged(std.Thread),25threads: std.ArrayListUnmanaged(std.Thread),
26stack_size: usize,26stack_size: usize,
27cpu_count: std.Thread.CpuCountError!usize,27cpu_count: usize, // 0 means no limit
28concurrent_count: usize,28concurrency_limit: usize, // 0 means no limit
29available_thread_count: usize = 0,
30one_shot_thread_count: usize = 0,
2931
30wsa: if (is_windows) Wsa else struct {} = .{},32wsa: if (is_windows) Wsa else struct {} = .{},
3133
...@@ -70,8 +72,6 @@ const Closure = struct {...@@ -70,8 +72,6 @@ const Closure = struct {
70 start: Start,72 start: Start,
71 node: std.SinglyLinkedList.Node = .{},73 node: std.SinglyLinkedList.Node = .{},
72 cancel_tid: CancelId,74 cancel_tid: CancelId,
73 /// Whether this task bumps minimum number of threads in the pool.
74 is_concurrent: bool,
7575
76 const Start = *const fn (*Closure) void;76 const Start = *const fn (*Closure) void;
7777
...@@ -103,20 +103,20 @@ pub fn init(...@@ -103,20 +103,20 @@ pub fn init(
103 /// here.103 /// here.
104 gpa: Allocator,104 gpa: Allocator,
105) Threaded {105) Threaded {
106 assert(!builtin.single_threaded); // use 'init_single_threaded' instead
107
106 var t: Threaded = .{108 var t: Threaded = .{
107 .allocator = gpa,109 .allocator = gpa,
108 .threads = .empty,110 .threads = .empty,
109 .stack_size = std.Thread.SpawnConfig.default_stack_size,111 .stack_size = std.Thread.SpawnConfig.default_stack_size,
110 .cpu_count = std.Thread.getCpuCount(),112 .cpu_count = std.Thread.getCpuCount() catch 0,
111 .concurrent_count = 0,113 .concurrency_limit = 0,
112 .old_sig_io = undefined,114 .old_sig_io = undefined,
113 .old_sig_pipe = undefined,115 .old_sig_pipe = undefined,
114 .have_signal_handler = false,116 .have_signal_handler = false,
115 };117 };
116118
117 if (t.cpu_count) |n| {119 t.threads.ensureTotalCapacity(gpa, t.cpu_count) catch {};
118 t.threads.ensureTotalCapacityPrecise(gpa, n - 1) catch {};
119 } else |_| {}
120120
121 if (posix.Sigaction != void) {121 if (posix.Sigaction != void) {
122 // This causes sending `posix.SIG.IO` to thread to interrupt blocking122 // This causes sending `posix.SIG.IO` to thread to interrupt blocking
...@@ -145,7 +145,7 @@ pub const init_single_threaded: Threaded = .{...@@ -145,7 +145,7 @@ pub const init_single_threaded: Threaded = .{
145 .threads = .empty,145 .threads = .empty,
146 .stack_size = std.Thread.SpawnConfig.default_stack_size,146 .stack_size = std.Thread.SpawnConfig.default_stack_size,
147 .cpu_count = 1,147 .cpu_count = 1,
148 .concurrent_count = 0,148 .concurrency_limit = 0,
149 .old_sig_io = undefined,149 .old_sig_io = undefined,
150 .old_sig_pipe = undefined,150 .old_sig_pipe = undefined,
151 .have_signal_handler = false,151 .have_signal_handler = false,
...@@ -184,18 +184,22 @@ fn worker(t: *Threaded) void {...@@ -184,18 +184,22 @@ fn worker(t: *Threaded) void {
184 while (t.run_queue.popFirst()) |closure_node| {184 while (t.run_queue.popFirst()) |closure_node| {
185 t.mutex.unlock();185 t.mutex.unlock();
186 const closure: *Closure = @fieldParentPtr("node", closure_node);186 const closure: *Closure = @fieldParentPtr("node", closure_node);
187 const is_concurrent = closure.is_concurrent;
188 closure.start(closure);187 closure.start(closure);
189 t.mutex.lock();188 t.mutex.lock();
190 if (is_concurrent) {189 t.available_thread_count += 1;
191 t.concurrent_count -= 1;
192 }
193 }190 }
194 if (t.join_requested) break;191 if (t.join_requested) break;
195 t.cond.wait(&t.mutex);192 t.cond.wait(&t.mutex);
196 }193 }
197}194}
198195
196fn oneShotWorker(t: *Threaded, closure: *Closure) void {
197 closure.start(closure);
198 t.mutex.lock();
199 defer t.mutex.unlock();
200 t.one_shot_thread_count -= 1;
201}
202
199pub fn io(t: *Threaded) Io {203pub fn io(t: *Threaded) Io {
200 return .{204 return .{
201 .userdata = t,205 .userdata = t,
...@@ -448,22 +452,22 @@ fn async(...@@ -448,22 +452,22 @@ fn async(
448 context_alignment: std.mem.Alignment,452 context_alignment: std.mem.Alignment,
449 start: *const fn (context: *const anyopaque, result: *anyopaque) void,453 start: *const fn (context: *const anyopaque, result: *anyopaque) void,
450) ?*Io.AnyFuture {454) ?*Io.AnyFuture {
451 if (builtin.single_threaded) {455 const t: *Threaded = @ptrCast(@alignCast(userdata));
456 if (t.cpu_count == 1) {
452 start(context.ptr, result.ptr);457 start(context.ptr, result.ptr);
453 return null;458 return null;
454 }459 }
455 const t: *Threaded = @ptrCast(@alignCast(userdata));460
456 const cpu_count = t.cpu_count catch {
457 return concurrent(userdata, result.len, result_alignment, context, context_alignment, start) catch {
458 start(context.ptr, result.ptr);
459 return null;
460 };
461 };
462 const gpa = t.allocator;461 const gpa = t.allocator;
463 const context_offset = context_alignment.forward(@sizeOf(AsyncClosure));462 const context_offset = context_alignment.forward(@sizeOf(AsyncClosure));
464 const result_offset = result_alignment.forward(context_offset + context.len);463 const result_offset = result_alignment.forward(context_offset + context.len);
465 const n = result_offset + result.len;464 const n = result_offset + result.len;
466 const ac: *AsyncClosure = @ptrCast(@alignCast(gpa.alignedAlloc(u8, .of(AsyncClosure), n) catch {465
466 const ac: *AsyncClosure = @ptrCast(@alignCast(gpa.alignedAlloc(
467 u8,
468 .of(AsyncClosure),
469 n,
470 ) catch {
467 start(context.ptr, result.ptr);471 start(context.ptr, result.ptr);
468 return null;472 return null;
469 }));473 }));
...@@ -472,7 +476,6 @@ fn async(...@@ -472,7 +476,6 @@ fn async(
472 .closure = .{476 .closure = .{
473 .cancel_tid = .none,477 .cancel_tid = .none,
474 .start = AsyncClosure.start,478 .start = AsyncClosure.start,
475 .is_concurrent = false,
476 },479 },
477 .func = start,480 .func = start,
478 .context_alignment = context_alignment,481 .context_alignment = context_alignment,
...@@ -485,34 +488,38 @@ fn async(...@@ -485,34 +488,38 @@ fn async(
485488
486 t.mutex.lock();489 t.mutex.lock();
487490
488 const thread_capacity = cpu_count - 1 + t.concurrent_count;491 if (t.available_thread_count == 0) {
489492 if (t.cpu_count != 0 and t.threads.items.len >= t.cpu_count) {
490 t.threads.ensureTotalCapacityPrecise(gpa, thread_capacity) catch {493 t.mutex.unlock();
491 t.mutex.unlock();494 ac.free(gpa, result.len);
492 ac.free(gpa, result.len);495 start(context.ptr, result.ptr);
493 start(context.ptr, result.ptr);496 return null;
494 return null;497 }
495 };
496498
497 t.run_queue.prepend(&ac.closure.node);499 t.threads.ensureUnusedCapacity(gpa, 1) catch {
500 t.mutex.unlock();
501 ac.free(gpa, result.len);
502 start(context.ptr, result.ptr);
503 return null;
504 };
498505
499 if (t.threads.items.len < thread_capacity) {506 const thread = std.Thread.spawn(
500 const thread = std.Thread.spawn(.{ .stack_size = t.stack_size }, worker, .{t}) catch {507 .{ .stack_size = t.stack_size },
501 if (t.threads.items.len == 0) {508 worker,
502 assert(t.run_queue.popFirst() == &ac.closure.node);509 .{t},
503 t.mutex.unlock();510 ) catch {
504 ac.free(gpa, result.len);
505 start(context.ptr, result.ptr);
506 return null;
507 }
508 // Rely on other workers to do it.
509 t.mutex.unlock();511 t.mutex.unlock();
510 t.cond.signal();512 ac.free(gpa, result.len);
511 return @ptrCast(ac);513 start(context.ptr, result.ptr);
514 return null;
512 };515 };
516
513 t.threads.appendAssumeCapacity(thread);517 t.threads.appendAssumeCapacity(thread);
518 } else {
519 t.available_thread_count -= 1;
514 }520 }
515521
522 t.run_queue.prepend(&ac.closure.node);
516 t.mutex.unlock();523 t.mutex.unlock();
517 t.cond.signal();524 t.cond.signal();
518 return @ptrCast(ac);525 return @ptrCast(ac);
...@@ -529,20 +536,21 @@ fn concurrent(...@@ -529,20 +536,21 @@ fn concurrent(
529 if (builtin.single_threaded) return error.ConcurrencyUnavailable;536 if (builtin.single_threaded) return error.ConcurrencyUnavailable;
530537
531 const t: *Threaded = @ptrCast(@alignCast(userdata));538 const t: *Threaded = @ptrCast(@alignCast(userdata));
532 const cpu_count = t.cpu_count catch 1;
533 const gpa = t.allocator;539 const gpa = t.allocator;
534 const context_offset = context_alignment.forward(@sizeOf(AsyncClosure));540 const context_offset = context_alignment.forward(@sizeOf(AsyncClosure));
535 const result_offset = result_alignment.forward(context_offset + context.len);541 const result_offset = result_alignment.forward(context_offset + context.len);
536 const n = result_offset + result_len;542 const n = result_offset + result_len;
537 const ac_bytes = gpa.alignedAlloc(u8, .of(AsyncClosure), n) catch543
544 const ac_bytes = gpa.alignedAlloc(u8, .of(AsyncClosure), n) catch {
538 return error.ConcurrencyUnavailable;545 return error.ConcurrencyUnavailable;
539 const ac: *AsyncClosure = @ptrCast(@alignCast(ac_bytes));546 };
547 errdefer gpa.free(ac_bytes);
540548
549 const ac: *AsyncClosure = @ptrCast(@alignCast(ac_bytes));
541 ac.* = .{550 ac.* = .{
542 .closure = .{551 .closure = .{
543 .cancel_tid = .none,552 .cancel_tid = .none,
544 .start = AsyncClosure.start,553 .start = AsyncClosure.start,
545 .is_concurrent = true,
546 },554 },
547 .func = start,555 .func = start,
548 .context_alignment = context_alignment,556 .context_alignment = context_alignment,
...@@ -550,33 +558,52 @@ fn concurrent(...@@ -550,33 +558,52 @@ fn concurrent(
550 .reset_event = .unset,558 .reset_event = .unset,
551 .select_condition = null,559 .select_condition = null,
552 };560 };
561
553 @memcpy(ac.contextPointer()[0..context.len], context);562 @memcpy(ac.contextPointer()[0..context.len], context);
554563
555 t.mutex.lock();564 t.mutex.lock();
565 defer t.mutex.unlock();
556566
557 t.concurrent_count += 1;567 // If there's an avilable thread, use it.
558 const thread_capacity = cpu_count - 1 + t.concurrent_count;568 if (t.available_thread_count > 0) {
569 t.available_thread_count -= 1;
570 t.run_queue.prepend(&ac.closure.node);
571 t.cond.signal();
572 return @ptrCast(ac);
573 }
559574
560 t.threads.ensureTotalCapacity(gpa, thread_capacity) catch {575 // If we can spawn a normal worker, spawn it and use it.
561 t.mutex.unlock();576 if (t.cpu_count == 0 or t.threads.items.len < t.cpu_count) {
562 ac.free(gpa, result_len);577 t.threads.ensureUnusedCapacity(gpa, 1) catch return error.ConcurrencyUnavailable;
563 return error.ConcurrencyUnavailable;
564 };
565578
566 t.run_queue.prepend(&ac.closure.node);579 const thread = std.Thread.spawn(
580 .{ .stack_size = t.stack_size },
581 worker,
582 .{t},
583 ) catch return error.ConcurrencyUnavailable;
567584
568 if (t.threads.items.len < thread_capacity) {
569 const thread = std.Thread.spawn(.{ .stack_size = t.stack_size }, worker, .{t}) catch {
570 assert(t.run_queue.popFirst() == &ac.closure.node);
571 t.mutex.unlock();
572 ac.free(gpa, result_len);
573 return error.ConcurrencyUnavailable;
574 };
575 t.threads.appendAssumeCapacity(thread);585 t.threads.appendAssumeCapacity(thread);
586 t.run_queue.prepend(&ac.closure.node);
587 t.cond.signal();
588 return @ptrCast(ac);
576 }589 }
577590
578 t.mutex.unlock();591 // If we have a concurrencty limit and we havent' hit it yet,
579 t.cond.signal();592 // spawn a new one-shot thread.
593 if (t.concurrency_limit != 0 and t.one_shot_thread_count >= t.concurrency_limit) {
594 return error.ConcurrencyUnavailable;
595 }
596
597 t.one_shot_thread_count += 1;
598 errdefer t.one_shot_thread_count -= 1;
599
600 const thread = std.Thread.spawn(
601 .{ .stack_size = t.stack_size },
602 oneShotWorker,
603 .{ t, &ac.closure },
604 ) catch return error.ConcurrencyUnavailable;
605 thread.detach();
606
580 return @ptrCast(ac);607 return @ptrCast(ac);
581}608}
582609
...@@ -647,7 +674,6 @@ fn groupAsync(...@@ -647,7 +674,6 @@ fn groupAsync(
647) void {674) void {
648 if (builtin.single_threaded) return start(group, context.ptr);675 if (builtin.single_threaded) return start(group, context.ptr);
649 const t: *Threaded = @ptrCast(@alignCast(userdata));676 const t: *Threaded = @ptrCast(@alignCast(userdata));
650 const cpu_count = t.cpu_count catch 1;
651 const gpa = t.allocator;677 const gpa = t.allocator;
652 const n = GroupClosure.contextEnd(context_alignment, context.len);678 const n = GroupClosure.contextEnd(context_alignment, context.len);
653 const gc: *GroupClosure = @ptrCast(@alignCast(gpa.alignedAlloc(u8, .of(GroupClosure), n) catch {679 const gc: *GroupClosure = @ptrCast(@alignCast(gpa.alignedAlloc(u8, .of(GroupClosure), n) catch {
...@@ -657,7 +683,6 @@ fn groupAsync(...@@ -657,7 +683,6 @@ fn groupAsync(
657 .closure = .{683 .closure = .{
658 .cancel_tid = .none,684 .cancel_tid = .none,
659 .start = GroupClosure.start,685 .start = GroupClosure.start,
660 .is_concurrent = false,
661 },686 },
662 .t = t,687 .t = t,
663 .group = group,688 .group = group,
...@@ -674,26 +699,36 @@ fn groupAsync(...@@ -674,26 +699,36 @@ fn groupAsync(
674 gc.node = .{ .next = @ptrCast(@alignCast(group.token)) };699 gc.node = .{ .next = @ptrCast(@alignCast(group.token)) };
675 group.token = &gc.node;700 group.token = &gc.node;
676701
677 const thread_capacity = cpu_count - 1 + t.concurrent_count;702 if (t.available_thread_count == 0) {
678703 if (t.cpu_count != 0 and t.threads.items.len >= t.cpu_count) {
679 t.threads.ensureTotalCapacityPrecise(gpa, thread_capacity) catch {704 t.mutex.unlock();
680 t.mutex.unlock();705 gc.free(gpa);
681 gc.free(gpa);706 return start(group, context.ptr);
682 return start(group, context.ptr);707 }
683 };
684708
685 t.run_queue.prepend(&gc.closure.node);709 t.threads.ensureUnusedCapacity(gpa, 1) catch {
710 t.mutex.unlock();
711 gc.free(gpa);
712 return start(group, context.ptr);
713 };
686714
687 if (t.threads.items.len < thread_capacity) {715 const thread = std.Thread.spawn(
688 const thread = std.Thread.spawn(.{ .stack_size = t.stack_size }, worker, .{t}) catch {716 .{ .stack_size = t.stack_size },
689 assert(t.run_queue.popFirst() == &gc.closure.node);717 worker,
718 .{t},
719 ) catch {
690 t.mutex.unlock();720 t.mutex.unlock();
691 gc.free(gpa);721 gc.free(gpa);
692 return start(group, context.ptr);722 return start(group, context.ptr);
693 };723 };
724
694 t.threads.appendAssumeCapacity(thread);725 t.threads.appendAssumeCapacity(thread);
726 } else {
727 t.available_thread_count -= 1;
695 }728 }
696729
730 t.run_queue.prepend(&gc.closure.node);
731
697 // This needs to be done before unlocking the mutex to avoid a race with732 // This needs to be done before unlocking the mutex to avoid a race with
698 // the associated task finishing.733 // the associated task finishing.
699 const group_state: *std.atomic.Value(usize) = @ptrCast(&group.state);734 const group_state: *std.atomic.Value(usize) = @ptrCast(&group.state);