authorgravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2025-10-30 10:26:19-07:00
committergravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2025-11-21 07:09:44-08:00
logc91dec3b6fee7f549eb91e67ea8a0d81cbdbfa4d
tree5551010074a25abe1686a366025e5f282f563139
parentfba9fdf54feff811108d12a4f14506446dd82faf

std.Io.Threaded: store thread capacity differently

2 files changed, 41 insertions(+), 9 deletions(-)

lib/std/Io/Threaded.zig+39-7
......@@ -25,7 +25,8 @@ run_queue: std.SinglyLinkedList = .{},
2525join_requested: bool = false,
2626threads: std.ArrayList(std.Thread),
2727stack_size: usize,
28cpu_count: std.Thread.CpuCountError!usize,
28thread_capacity: std.atomic.Value(ThreadCapacity),
29thread_capacity_error: ?std.Thread.CpuCountError,
2930concurrent_count: usize,
3031
3132wsa: if (is_windows) Wsa else struct {} = .{},
......@@ -34,6 +35,21 @@ have_signal_handler: bool,
3435old_sig_io: if (have_sig_io) posix.Sigaction else void,
3536old_sig_pipe: if (have_sig_pipe) posix.Sigaction else void,
3637
38pub const ThreadCapacity = enum(usize) {
39 unknown = 0,
40 _,
41
42 pub fn init(n: usize) ThreadCapacity {
43 assert(n != 0);
44 return @enumFromInt(n);
45 }
46
47 pub fn get(tc: ThreadCapacity) ?usize {
48 if (tc == .unknown) return null;
49 return @intFromEnum(tc);
50 }
51};
52
3753threadlocal var current_closure: ?*Closure = null;
3854
3955const max_iovecs_len = 8;
......@@ -104,18 +120,21 @@ pub fn init(
104120 /// here.
105121 gpa: Allocator,
106122) Threaded {
123 const cpu_count = std.Thread.getCpuCount();
124
107125 var t: Threaded = .{
108126 .allocator = gpa,
109127 .threads = .empty,
110128 .stack_size = std.Thread.SpawnConfig.default_stack_size,
111 .cpu_count = std.Thread.getCpuCount(),
129 .thread_capacity = .init(if (cpu_count) |n| .init(n) else |_| .unknown),
130 .thread_capacity_error = if (cpu_count) |_| null else |e| e,
112131 .concurrent_count = 0,
113132 .old_sig_io = undefined,
114133 .old_sig_pipe = undefined,
115134 .have_signal_handler = false,
116135 };
117136
118 if (t.cpu_count) |n| {
137 if (cpu_count) |n| {
119138 t.threads.ensureTotalCapacityPrecise(gpa, n - 1) catch {};
120139 } else |_| {}
121140
......@@ -145,7 +164,8 @@ pub const init_single_threaded: Threaded = .{
145164 .allocator = .failing,
146165 .threads = .empty,
147166 .stack_size = std.Thread.SpawnConfig.default_stack_size,
148 .cpu_count = 1,
167 .thread_capacity = .init(.init(1)),
168 .thread_capacity_error = null,
149169 .concurrent_count = 0,
150170 .old_sig_io = undefined,
151171 .old_sig_pipe = undefined,
......@@ -166,6 +186,18 @@ pub fn deinit(t: *Threaded) void {
166186 t.* = undefined;
167187}
168188
189pub fn setThreadCapacity(t: *Threaded, n: usize) void {
190 t.thread_capacity.store(.init(n), .monotonic);
191}
192
193pub fn getThreadCapacity(t: *Threaded) ?usize {
194 return t.thread_capacity.load(.monotonic).get();
195}
196
197pub fn getCurrentThreadId() usize {
198 @panic("TODO");
199}
200
169201fn join(t: *Threaded) void {
170202 if (builtin.single_threaded) return;
171203 {
......@@ -497,7 +529,7 @@ fn async(
497529 }
498530
499531 const t: *Threaded = @ptrCast(@alignCast(userdata));
500 const cpu_count = t.cpu_count catch {
532 const cpu_count = t.getThreadCapacity() orelse {
501533 return concurrent(userdata, result.len, result_alignment, context, context_alignment, start) catch {
502534 start(context.ptr, result.ptr);
503535 return null;
......@@ -556,7 +588,7 @@ fn concurrent(
556588 if (builtin.single_threaded) return error.ConcurrencyUnavailable;
557589
558590 const t: *Threaded = @ptrCast(@alignCast(userdata));
559 const cpu_count = t.cpu_count catch 1;
591 const cpu_count = t.getThreadCapacity() orelse 1;
560592
561593 const gpa = t.allocator;
562594 const ac = AsyncClosure.init(gpa, .concurrent, result_len, result_alignment, context, context_alignment, start) catch {
......@@ -685,7 +717,7 @@ fn groupAsync(
685717 if (builtin.single_threaded) return start(group, context.ptr);
686718
687719 const t: *Threaded = @ptrCast(@alignCast(userdata));
688 const cpu_count = t.cpu_count catch 1;
720 const cpu_count = t.getThreadCapacity() orelse 1;
689721
690722 const gpa = t.allocator;
691723 const gc = GroupClosure.init(gpa, t, group, context, context_alignment, start) catch {
lib/std/Io/Threaded/test.zig+2-2
......@@ -10,7 +10,7 @@ test "concurrent vs main prevents deadlock via oversubscription" {
1010 defer threaded.deinit();
1111 const io = threaded.io();
1212
13 threaded.cpu_count = 1;
13 threaded.setThreadCapacity(1);
1414
1515 var queue: Io.Queue(u8) = .init(&.{});
1616
......@@ -38,7 +38,7 @@ test "concurrent vs concurrent prevents deadlock via oversubscription" {
3838 defer threaded.deinit();
3939 const io = threaded.io();
4040
41 threaded.cpu_count = 1;
41 threaded.setThreadCapacity(1);
4242
4343 var queue: Io.Queue(u8) = .init(&.{});
4444