authorgravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2025-12-21 19:22:42-08:00
committergravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2025-12-23 22:15:12-08:00
loga29d79313a0c62843694ea723723447661239110
tree4d5cfd244696cd5933b8b815b5500398f0117f50
parent60481b82236fff1198975578c27c02ce6a7fa3fa

std.Io.Threaded: accept argv and environ on init

This is needed unfortunately for OpenBSD and Haiku for process executable path. I made it so that you can omit the options usually, but you get a compile error if you omit the options on those targets.

5 files changed, 103 insertions(+), 46 deletions(-)

lib/compiler/build_runner.zig+1-1
...@@ -39,7 +39,7 @@ pub fn main() !void {...@@ -39,7 +39,7 @@ pub fn main() !void {
3939
40 const args = try process.argsAlloc(arena);40 const args = try process.argsAlloc(arena);
4141
42 var threaded: std.Io.Threaded = .init(gpa);42 var threaded: std.Io.Threaded = .init(gpa, .{});
43 defer threaded.deinit();43 defer threaded.deinit();
44 const io = threaded.io();44 const io = threaded.io();
4545
lib/compiler/test_runner.zig+2-2
...@@ -131,7 +131,7 @@ fn mainServer() !void {...@@ -131,7 +131,7 @@ fn mainServer() !void {
131131
132 .run_test => {132 .run_test => {
133 testing.allocator_instance = .{};133 testing.allocator_instance = .{};
134 testing.io_instance = .init(testing.allocator);134 testing.io_instance = .init(testing.allocator, .{});
135 log_err_count = 0;135 log_err_count = 0;
136 const index = try server.receiveBody_u32();136 const index = try server.receiveBody_u32();
137 const test_fn = builtin.test_functions[index];137 const test_fn = builtin.test_functions[index];
...@@ -233,7 +233,7 @@ fn mainTerminal() void {...@@ -233,7 +233,7 @@ fn mainTerminal() void {
233 var leaks: usize = 0;233 var leaks: usize = 0;
234 for (test_fn_list, 0..) |test_fn, i| {234 for (test_fn_list, 0..) |test_fn, i| {
235 testing.allocator_instance = .{};235 testing.allocator_instance = .{};
236 testing.io_instance = .init(testing.allocator);236 testing.io_instance = .init(testing.allocator, .{});
237 defer {237 defer {
238 testing.io_instance.deinit();238 testing.io_instance.deinit();
239 if (testing.allocator_instance.deinit() == .leak) leaks += 1;239 if (testing.allocator_instance.deinit() == .leak) leaks += 1;
lib/std/Io/Threaded.zig+92-37
...@@ -29,23 +29,7 @@ join_requested: bool = false,...@@ -29,23 +29,7 @@ join_requested: bool = false,
29stack_size: usize,29stack_size: usize,
30/// All threads are spawned detached; this is how we wait until they all exit.30/// All threads are spawned detached; this is how we wait until they all exit.
31wait_group: std.Thread.WaitGroup = .{},31wait_group: std.Thread.WaitGroup = .{},
32/// Maximum thread pool size (excluding main thread) when dispatching async
33/// tasks. Until this limit, calls to `Io.async` when all threads are busy will
34/// cause a new thread to be spawned and permanently added to the pool. After
35/// this limit, calls to `Io.async` when all threads are busy run the task
36/// immediately.
37///
38/// Defaults to a number equal to logical CPU cores.
39///
40/// Protected by `mutex` once the I/O instance is already in use. See
41/// `setAsyncLimit`.
42async_limit: Io.Limit,32async_limit: Io.Limit,
43/// Maximum thread pool size (excluding main thread) for dispatching concurrent
44/// tasks. Until this limit, calls to `Io.concurrent` will increase the thread
45/// pool size.
46///
47/// concurrent tasks. After this number, calls to `Io.concurrent` return
48/// `error.ConcurrencyUnavailable`.
49concurrent_limit: Io.Limit = .unlimited,33concurrent_limit: Io.Limit = .unlimited,
50/// Error from calling `std.Thread.getCpuCount` in `init`.34/// Error from calling `std.Thread.getCpuCount` in `init`.
51cpu_count_error: ?std.Thread.CpuCountError,35cpu_count_error: ?std.Thread.CpuCountError,
...@@ -55,17 +39,7 @@ cpu_count_error: ?std.Thread.CpuCountError,...@@ -55,17 +39,7 @@ cpu_count_error: ?std.Thread.CpuCountError,
55busy_count: usize = 0,39busy_count: usize = 0,
56main_thread: Thread,40main_thread: Thread,
57pid: Pid = .unknown,41pid: Pid = .unknown,
58/// When a cancel request is made, blocking syscalls can be unblocked by42robust_cancel: RobustCancel,
59/// issuing a signal. However, if the signal arrives after the check and before
60/// the syscall instruction, it is missed.
61///
62/// This option solves the race condition by retrying the signal delivery
63/// until it is acknowledged, with an exponential backoff.
64///
65/// Unfortunately, trying again until the cancellation request is acknowledged
66/// has been observed to be relatively slow, and usually strong cancellation
67/// guarantees are not needed, so this defaults to off.
68robust_cancel: RobustCancel = .disabled,
6943
70wsa: if (is_windows) Wsa else struct {} = .{},44wsa: if (is_windows) Wsa else struct {} = .{},
7145
...@@ -86,6 +60,32 @@ stderr_writer: File.Writer = .{...@@ -86,6 +60,32 @@ stderr_writer: File.Writer = .{
86stderr_mode: Io.Terminal.Mode = .no_color,60stderr_mode: Io.Terminal.Mode = .no_color,
87stderr_writer_initialized: bool = false,61stderr_writer_initialized: bool = false,
8862
63environ: Environ,
64args: Args,
65
66pub const Environ = switch (native_os) {
67 .openbsd, .haiku => struct {
68 PATH: ?[]const u8,
69
70 pub const empty: @This() = .{
71 .PATH = null,
72 };
73 },
74 else => struct {
75 pub const empty: @This() = .{};
76 },
77};
78
79pub const Args = switch (native_os) {
80 .openbsd, .haiku => struct {
81 list: []const []const u8,
82 pub const empty: @This() = .{ .list = &.{} };
83 },
84 else => struct {
85 pub const empty: @This() = .{};
86 },
87};
88
89pub const RobustCancel = if (std.Thread.use_pthreads or native_os == .linux) enum {89pub const RobustCancel = if (std.Thread.use_pthreads or native_os == .linux) enum {
90 enabled,90 enabled,
91 disabled,91 disabled,
...@@ -557,6 +557,54 @@ const Closure = struct {...@@ -557,6 +557,54 @@ const Closure = struct {
557 }557 }
558};558};
559559
560pub const InitOptions = struct {
561 /// Affects how many bytes are memory-mapped for threads.
562 stack_size: usize = std.Thread.SpawnConfig.default_stack_size,
563 /// Maximum thread pool size (excluding main thread) when dispatching async
564 /// tasks. Until this limit, calls to `Io.async` when all threads are busy will
565 /// cause a new thread to be spawned and permanently added to the pool. After
566 /// this limit, calls to `Io.async` when all threads are busy run the task
567 /// immediately.
568 ///
569 /// Defaults to a number equal to logical CPU cores.
570 ///
571 /// Protected by `Threaded.mutex` once the I/O instance is already in use. See
572 /// `setAsyncLimit`.
573 async_limit: ?Io.Limit = null,
574 /// Maximum thread pool size (excluding main thread) for dispatching concurrent
575 /// tasks. Until this limit, calls to `Io.concurrent` will increase the thread
576 /// pool size.
577 ///
578 /// concurrent tasks. After this number, calls to `Io.concurrent` return
579 /// `error.ConcurrencyUnavailable`.
580 concurrent_limit: Io.Limit = .unlimited,
581 /// When a cancel request is made, blocking syscalls can be unblocked by
582 /// issuing a signal. However, if the signal arrives after the check and before
583 /// the syscall instruction, it is missed.
584 ///
585 /// This option solves the race condition by retrying the signal delivery
586 /// until it is acknowledged, with an exponential backoff.
587 ///
588 /// Unfortunately, trying again until the cancellation request is acknowledged
589 /// has been observed to be relatively slow, and usually strong cancellation
590 /// guarantees are not needed, so this defaults to off.
591 robust_cancel: RobustCancel = .disabled,
592 /// Affects the following operations:
593 /// * `processExecutablePath` on OpenBSD and Haiku.
594 ///
595 /// The default value causes this to be a compile error on systems that need to
596 /// initialize this field. `Environ.empty` can be used to omit this field on
597 /// all targets.
598 environ: Environ = .{},
599 /// Affects the following operations:
600 /// * `processExecutablePath` on OpenBSD and Haiku.
601 ///
602 /// The default value causes this to be a compile error on systems that need to
603 /// initialize this field. `Args.empty` can be used to omit this field on all
604 /// targets.
605 args: Args = .{},
606};
607
560/// Related:608/// Related:
561/// * `init_single_threaded`609/// * `init_single_threaded`
562pub fn init(610pub fn init(
...@@ -568,6 +616,7 @@ pub fn init(...@@ -568,6 +616,7 @@ pub fn init(
568 /// If these functions are avoided, then `Allocator.failing` may be passed616 /// If these functions are avoided, then `Allocator.failing` may be passed
569 /// here.617 /// here.
570 gpa: Allocator,618 gpa: Allocator,
619 options: InitOptions,
571) Threaded {620) Threaded {
572 if (builtin.single_threaded) return .init_single_threaded;621 if (builtin.single_threaded) return .init_single_threaded;
573622
...@@ -575,8 +624,9 @@ pub fn init(...@@ -575,8 +624,9 @@ pub fn init(
575624
576 var t: Threaded = .{625 var t: Threaded = .{
577 .allocator = gpa,626 .allocator = gpa,
578 .stack_size = std.Thread.SpawnConfig.default_stack_size,627 .stack_size = options.stack_size,
579 .async_limit = if (cpu_count) |n| .limited(n - 1) else |_| .nothing,628 .async_limit = options.async_limit orelse if (cpu_count) |n| .limited(n - 1) else |_| .nothing,
629 .concurrent_limit = options.concurrent_limit,
580 .cpu_count_error = if (cpu_count) |_| null else |e| e,630 .cpu_count_error = if (cpu_count) |_| null else |e| e,
581 .old_sig_io = undefined,631 .old_sig_io = undefined,
582 .old_sig_pipe = undefined,632 .old_sig_pipe = undefined,
...@@ -586,6 +636,9 @@ pub fn init(...@@ -586,6 +636,9 @@ pub fn init(
586 .current_closure = null,636 .current_closure = null,
587 .cancel_protection = undefined,637 .cancel_protection = undefined,
588 },638 },
639 .environ = options.environ,
640 .args = options.args,
641 .robust_cancel = options.robust_cancel,
589 };642 };
590643
591 if (posix.Sigaction != void) {644 if (posix.Sigaction != void) {
...@@ -624,6 +677,9 @@ pub const init_single_threaded: Threaded = .{...@@ -624,6 +677,9 @@ pub const init_single_threaded: Threaded = .{
624 .current_closure = null,677 .current_closure = null,
625 .cancel_protection = undefined,678 .cancel_protection = undefined,
626 },679 },
680 .robust_cancel = .disabled,
681 .environ = .empty,
682 .args = .empty,
627};683};
628684
629var global_single_threaded_instance: Threaded = .init_single_threaded;685var global_single_threaded_instance: Threaded = .init_single_threaded;
...@@ -7186,18 +7242,17 @@ fn processExecutablePath(userdata: ?*anyopaque, out_buffer: []u8) std.process.Ex...@@ -7186,18 +7242,17 @@ fn processExecutablePath(userdata: ?*anyopaque, out_buffer: []u8) std.process.Ex
7186 }7242 }
7187 },7243 },
7188 .openbsd, .haiku => {7244 .openbsd, .haiku => {
7189 // OpenBSD doesn't support getting the path of a running process, so try to guess it7245 // The best we can do on these operating systems is check based on CLI args.
7190 if (std.os.argv.len == 0)7246 const argv = t.args.list;
7191 return error.FileNotFound;7247 if (argv.len == 0) return error.OperationUnsupported;
71927248 const argv0 = argv[0];
7193 const argv0 = std.mem.span(std.os.argv[0]);
7194 if (std.mem.findScalar(u8, argv0, '/') != null) {7249 if (std.mem.findScalar(u8, argv0, '/') != null) {
7195 // argv[0] is a path (relative or absolute): use realpath(3) directly7250 // argv[0] is a path (relative or absolute): use realpath(3) directly
7196 const current_thread = Thread.getCurrent(t);7251 const current_thread = Thread.getCurrent(t);
7197 var resolved_buf: [std.c.PATH_MAX]u8 = undefined;7252 var resolved_buf: [std.c.PATH_MAX]u8 = undefined;
7198 try current_thread.beginSyscall();7253 try current_thread.beginSyscall();
7199 while (true) {7254 while (true) {
7200 if (std.c.realpath(std.os.argv[0], &resolved_buf)) |p| {7255 if (std.c.realpath(argv[0], &resolved_buf)) |p| {
7201 assert(p == &resolved_buf);7256 assert(p == &resolved_buf);
7202 break current_thread.endSyscall();7257 break current_thread.endSyscall();
7203 } else switch (@as(std.c.E, @enumFromInt(std.c._errno().*))) {7258 } else switch (@as(std.c.E, @enumFromInt(std.c._errno().*))) {
...@@ -7229,12 +7284,12 @@ fn processExecutablePath(userdata: ?*anyopaque, out_buffer: []u8) std.process.Ex...@@ -7229,12 +7284,12 @@ fn processExecutablePath(userdata: ?*anyopaque, out_buffer: []u8) std.process.Ex
7229 } else if (argv0.len != 0) {7284 } else if (argv0.len != 0) {
7230 // argv[0] is not empty (and not a path): search PATH7285 // argv[0] is not empty (and not a path): search PATH
7231 const current_thread = Thread.getCurrent(t);7286 const current_thread = Thread.getCurrent(t);
7232 const PATH = posix.getenvZ("PATH") orelse return error.FileNotFound;7287 const PATH = t.environ.PATH orelse return error.FileNotFound;
7233 var it = std.mem.tokenizeScalar(u8, PATH, ':');7288 var it = std.mem.tokenizeScalar(u8, PATH, ':');
7234 it: while (it.next()) |dir| {7289 it: while (it.next()) |dir| {
7235 var resolved_path_buf: [std.c.PATH_MAX]u8 = undefined;7290 var resolved_path_buf: [std.c.PATH_MAX]u8 = undefined;
7236 const resolved_path = std.fmt.bufPrintSentinel(&resolved_path_buf, "{s}/{s}", .{7291 const resolved_path = std.fmt.bufPrintSentinel(&resolved_path_buf, "{s}/{s}", .{
7237 dir, std.os.argv[0],7292 dir, argv[0],
7238 }, 0) catch continue;7293 }, 0) catch continue;
72397294
7240 var resolved_buf: [std.c.PATH_MAX]u8 = undefined;7295 var resolved_buf: [std.c.PATH_MAX]u8 = undefined;
lib/std/Io/Threaded/test.zig+7-5
...@@ -1,3 +1,5 @@...@@ -1,3 +1,5 @@
1//! Tests belong here if they access internal state of std.Io.Threaded or
2//! otherwise assume details of that particular implementation.
1const builtin = @import("builtin");3const builtin = @import("builtin");
24
3const std = @import("std");5const std = @import("std");
...@@ -11,7 +13,7 @@ test "concurrent vs main prevents deadlock via oversubscription" {...@@ -11,7 +13,7 @@ test "concurrent vs main prevents deadlock via oversubscription" {
11 return error.SkipZigTest;13 return error.SkipZigTest;
12 }14 }
1315
14 var threaded: Io.Threaded = .init(std.testing.allocator);16 var threaded: Io.Threaded = .init(std.testing.allocator, .{});
15 defer threaded.deinit();17 defer threaded.deinit();
16 const io = threaded.io();18 const io = threaded.io();
1719
...@@ -44,7 +46,7 @@ test "concurrent vs concurrent prevents deadlock via oversubscription" {...@@ -44,7 +46,7 @@ test "concurrent vs concurrent prevents deadlock via oversubscription" {
44 return error.SkipZigTest;46 return error.SkipZigTest;
45 }47 }
4648
47 var threaded: Io.Threaded = .init(std.testing.allocator);49 var threaded: Io.Threaded = .init(std.testing.allocator, .{});
48 defer threaded.deinit();50 defer threaded.deinit();
49 const io = threaded.io();51 const io = threaded.io();
5052
...@@ -78,7 +80,7 @@ test "async/concurrent context and result alignment" {...@@ -78,7 +80,7 @@ test "async/concurrent context and result alignment" {
78 var buffer: [2048]u8 align(@alignOf(ByteArray512)) = undefined;80 var buffer: [2048]u8 align(@alignOf(ByteArray512)) = undefined;
79 var fba: std.heap.FixedBufferAllocator = .init(&buffer);81 var fba: std.heap.FixedBufferAllocator = .init(&buffer);
8082
81 var threaded: std.Io.Threaded = .init(fba.allocator());83 var threaded: std.Io.Threaded = .init(fba.allocator(), .{});
82 defer threaded.deinit();84 defer threaded.deinit();
83 const io = threaded.io();85 const io = threaded.io();
8486
...@@ -111,7 +113,7 @@ test "Group.async context alignment" {...@@ -111,7 +113,7 @@ test "Group.async context alignment" {
111 var buffer: [2048]u8 align(@alignOf(ByteArray512)) = undefined;113 var buffer: [2048]u8 align(@alignOf(ByteArray512)) = undefined;
112 var fba: std.heap.FixedBufferAllocator = .init(&buffer);114 var fba: std.heap.FixedBufferAllocator = .init(&buffer);
113115
114 var threaded: std.Io.Threaded = .init(fba.allocator());116 var threaded: std.Io.Threaded = .init(fba.allocator(), .{});
115 defer threaded.deinit();117 defer threaded.deinit();
116 const io = threaded.io();118 const io = threaded.io();
117119
...@@ -131,7 +133,7 @@ fn returnArray() [32]u8 {...@@ -131,7 +133,7 @@ fn returnArray() [32]u8 {
131}133}
132134
133test "async with array return type" {135test "async with array return type" {
134 var threaded: std.Io.Threaded = .init(std.testing.allocator);136 var threaded: std.Io.Threaded = .init(std.testing.allocator, .{});
135 defer threaded.deinit();137 defer threaded.deinit();
136 const io = threaded.io();138 const io = threaded.io();
137139
src/main.zig+1-1
...@@ -241,7 +241,7 @@ fn mainArgs(gpa: Allocator, arena: Allocator, args: []const []const u8) !void {...@@ -241,7 +241,7 @@ fn mainArgs(gpa: Allocator, arena: Allocator, args: []const []const u8) !void {
241 }241 }
242 }242 }
243243
244 var threaded: Io.Threaded = .init(gpa);244 var threaded: Io.Threaded = .init(gpa, .{});
245 defer threaded.deinit();245 defer threaded.deinit();
246 threaded_impl_ptr = &threaded;246 threaded_impl_ptr = &threaded;
247 threaded.stack_size = thread_stack_size;247 threaded.stack_size = thread_stack_size;