authorgravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2026-01-08 12:55:38-08:00
committergravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2026-01-30 12:10:01-08:00
logf391adc3af832cf9a5ec7be8f7c23134fc07984a
tree201c013021814f0d2b6feba83ccdfdbed913e71c
parent6d22f7b4d7bd6d29ecbc5e3bd4a2e1c085f293a3

std.Io: simplify operate function

- no timeout - no n_wait - infallible

5 files changed, 34 insertions(+), 54 deletions(-)

lib/std/Io.zig+4-15
...@@ -149,7 +149,7 @@ pub const VTable = struct {...@@ -149,7 +149,7 @@ pub const VTable = struct {
149 futexWaitUncancelable: *const fn (?*anyopaque, ptr: *const u32, expected: u32) void,149 futexWaitUncancelable: *const fn (?*anyopaque, ptr: *const u32, expected: u32) void,
150 futexWake: *const fn (?*anyopaque, ptr: *const u32, max_waiters: u32) void,150 futexWake: *const fn (?*anyopaque, ptr: *const u32, max_waiters: u32) void,
151151
152 operate: *const fn (?*anyopaque, []Operation, n_wait: usize, Timeout) OperateError!void,152 operate: *const fn (?*anyopaque, []Operation) void,
153153
154 dirCreateDir: *const fn (?*anyopaque, Dir, []const u8, Dir.Permissions) Dir.CreateDirError!void,154 dirCreateDir: *const fn (?*anyopaque, Dir, []const u8, Dir.Permissions) Dir.CreateDirError!void,
155 dirCreateDirPath: *const fn (?*anyopaque, Dir, []const u8, Dir.Permissions) Dir.CreateDirPathError!Dir.CreatePathStatus,155 dirCreateDirPath: *const fn (?*anyopaque, Dir, []const u8, Dir.Permissions) Dir.CreateDirPathError!Dir.CreatePathStatus,
...@@ -266,22 +266,11 @@ pub const Operation = union(enum) {...@@ -266,22 +266,11 @@ pub const Operation = union(enum) {
266 };266 };
267};267};
268268
269pub const OperateError = error{ Canceled, Timeout };
270
271/// Performs all `operations` in a non-deterministic order. Returns after all269/// Performs all `operations` in a non-deterministic order. Returns after all
272/// `operations` have been attempted. The degree to which the operations are270/// `operations` have been completed. The degree to which the operations are
273/// performed concurrently is determined by the `Io` implementation.271/// performed concurrently is determined by the `Io` implementation.
274///272pub fn operate(io: Io, operations: []Operation) void {
275/// `n_wait` is an amount of operations between `0` and `operations.len` that273 return io.vtable.operate(io.userdata, operations);
276/// determines how many attempted operations must complete before `operate`
277/// returns. Operation completion is defined by returning a value other than
278/// `error.WouldBlock`. If the operation cannot return `error.WouldBlock`, it
279/// always counts as completing.
280///
281/// In the event `error.Canceled` is returned, any number of `operations` may
282/// still have been completed successfully.
283pub fn operate(io: Io, operations: []Operation, n_wait: usize, timeout: Timeout) OperateError!void {
284 return io.vtable.operate(io.userdata, operations, n_wait, timeout);
285}274}
286275
287pub const Limit = enum(usize) {276pub const Limit = enum(usize) {
lib/std/Io/File.zig+1-1
...@@ -559,7 +559,7 @@ pub fn readStreaming(file: File, io: Io, buffer: []const []u8) Reader.Error!usiz...@@ -559,7 +559,7 @@ pub fn readStreaming(file: File, io: Io, buffer: []const []u8) Reader.Error!usiz
559 .data = buffer,559 .data = buffer,
560 .result = undefined,560 .result = undefined,
561 } };561 } };
562 io.vtable.operate(io.userdata, (&operation)[0..1], 1, .none) catch unreachable;562 io.vtable.operate(io.userdata, (&operation)[0..1]);
563 return operation.file_read_streaming.result;563 return operation.file_read_streaming.result;
564}564}
565565
lib/std/Io/Threaded.zig+28-33
...@@ -2449,20 +2449,15 @@ fn futexWake(userdata: ?*anyopaque, ptr: *const u32, max_waiters: u32) void {...@@ -2449,20 +2449,15 @@ fn futexWake(userdata: ?*anyopaque, ptr: *const u32, max_waiters: u32) void {
2449 Thread.futexWake(ptr, max_waiters);2449 Thread.futexWake(ptr, max_waiters);
2450}2450}
24512451
2452fn operate(userdata: ?*anyopaque, operations: []Io.Operation, n_wait: usize, timeout: Io.Timeout) Io.OperateError!void {2452fn operate(userdata: ?*anyopaque, operations: []Io.Operation) void {
2453 const t: *Threaded = @ptrCast(@alignCast(userdata));2453 const t: *Threaded = @ptrCast(@alignCast(userdata));
2454 const t_io = ioBasic(t);2454 _ = t;
24552455
2456 if (is_windows) @panic("TODO");2456 if (is_windows) @panic("TODO");
24572457
2458 const deadline = timeout.toDeadline(t_io) catch |err| switch (err) {
2459 error.UnsupportedClock, error.Unexpected => null,
2460 };
2461
2462 var poll_buffer: [100]posix.pollfd = undefined;2458 var poll_buffer: [100]posix.pollfd = undefined;
2463 var map_buffer: [poll_buffer.len]u8 = undefined; // poll_buffer index to operations index2459 var map_buffer: [poll_buffer.len]u8 = undefined; // poll_buffer index to operations index
2464 var poll_i: usize = 0;2460 var poll_i: usize = 0;
2465 var completed: usize = 0;
24662461
2467 // Put all the file reads with nonblocking enabled into the poll set.2462 // Put all the file reads with nonblocking enabled into the poll set.
2468 if (operations.len > poll_buffer.len) @panic("TODO");2463 if (operations.len > poll_buffer.len) @panic("TODO");
...@@ -2483,7 +2478,6 @@ fn operate(userdata: ?*anyopaque, operations: []Io.Operation, n_wait: usize, tim...@@ -2483,7 +2478,6 @@ fn operate(userdata: ?*anyopaque, operations: []Io.Operation, n_wait: usize, tim
2483 poll_i += 1;2478 poll_i += 1;
2484 } else {2479 } else {
2485 o.result = fileReadStreaming(o.file, o.data);2480 o.result = fileReadStreaming(o.file, o.data);
2486 completed += 1;
2487 }2481 }
2488 },2482 },
2489 };2483 };
...@@ -2493,41 +2487,42 @@ fn operate(userdata: ?*anyopaque, operations: []Io.Operation, n_wait: usize, tim...@@ -2493,41 +2487,42 @@ fn operate(userdata: ?*anyopaque, operations: []Io.Operation, n_wait: usize, tim
2493 return;2487 return;
2494 }2488 }
24952489
2496 const max_poll_ms = std.math.maxInt(i32);2490 while (true) {
24972491 const syscall = Syscall.start() catch |err| switch (err) {
2498 while (completed < n_wait) {2492 error.Canceled => {
2499 const timeout_ms: i32 = if (deadline) |d| t: {2493 for (map_buffer[0..poll_i]) |operation_index| {
2500 const duration = d.durationFromNow(t_io) catch @panic("TODO make this unreachable");2494 switch (operations[operation_index]) {
2501 if (duration.raw.nanoseconds <= 0) return error.Timeout;2495 .noop => unreachable,
2502 break :t @intCast(@min(max_poll_ms, duration.raw.toMilliseconds()));2496 inline else => |*o| o.result = error.Canceled,
2503 } else -1;2497 }
2504 const syscall = try Syscall.start();2498 }
2505 const poll_rc = posix.system.poll(&poll_buffer, poll_i, timeout_ms);2499 return;
2500 },
2501 };
2502 const poll_rc = posix.system.poll(&poll_buffer, poll_i, -1);
2506 syscall.finish();2503 syscall.finish();
2507 switch (posix.errno(poll_rc)) {2504 switch (posix.errno(poll_rc)) {
2508 .SUCCESS => {2505 .SUCCESS => {
2509 if (poll_rc == 0) {2506 if (poll_rc == 0) {
2510 // Although spurious timeouts are OK, when no deadline2507 // Spurious timeout; handle same as INTR.
2511 // is passed we must not return `error.Timeout`.2508 continue;
2512 if (deadline == null) continue;
2513 return error.Timeout;
2514 }
2515 for (poll_buffer[0..poll_i], map_buffer[0..poll_i]) |*poll_fd, operation_index| {
2516 if (poll_fd.revents == 0) continue;
2517 poll_fd.fd = -1; // Disarm this operation.
2518 switch (operations[operation_index]) {
2519 .noop => unreachable,
2520 .file_read_streaming => |*o| {
2521 o.result = fileReadStreaming(o.file, o.data);
2522 completed += 1;
2523 },
2524 }
2525 }2509 }
2510 break;
2526 },2511 },
2527 .INTR => continue,2512 .INTR => continue,
2528 else => @panic("TODO handle unexpected error from poll()"),2513 else => @panic("TODO handle unexpected error from poll()"),
2529 }2514 }
2530 }2515 }
2516
2517 for (poll_buffer[0..poll_i], map_buffer[0..poll_i]) |*poll_fd, operation_index| {
2518 if (poll_fd.revents == 0) continue;
2519 switch (operations[operation_index]) {
2520 .noop => unreachable,
2521 .file_read_streaming => |*o| {
2522 o.result = fileReadStreaming(o.file, o.data);
2523 },
2524 }
2525 }
2531}2526}
25322527
2533const dirCreateDir = switch (native_os) {2528const dirCreateDir = switch (native_os) {
lib/std/process.zig-2
...@@ -490,7 +490,6 @@ pub const RunOptions = struct {...@@ -490,7 +490,6 @@ pub const RunOptions = struct {
490 create_no_window: bool = true,490 create_no_window: bool = true,
491 /// Darwin-only. Disable ASLR for the child process.491 /// Darwin-only. Disable ASLR for the child process.
492 disable_aslr: bool = false,492 disable_aslr: bool = false,
493 timeout: Io.Timeout = .none,
494};493};
495494
496pub const RunResult = struct {495pub const RunResult = struct {
...@@ -532,7 +531,6 @@ pub fn run(gpa: Allocator, io: Io, options: RunOptions) RunError!RunResult {...@@ -532,7 +531,6 @@ pub fn run(gpa: Allocator, io: Io, options: RunOptions) RunError!RunResult {
532 .stderr = &stderr,531 .stderr = &stderr,
533 .stdout_limit = options.stdout_limit,532 .stdout_limit = options.stdout_limit,
534 .stderr_limit = options.stderr_limit,533 .stderr_limit = options.stderr_limit,
535 .timeout = options.timeout,
536 });534 });
537535
538 const term = try child.wait(io);536 const term = try child.wait(io);
lib/std/process/Child.zig+1-3
...@@ -138,7 +138,6 @@ pub const CollectOutputOptions = struct {...@@ -138,7 +138,6 @@ pub const CollectOutputOptions = struct {
138 allocator: ?Allocator = null,138 allocator: ?Allocator = null,
139 stdout_limit: Io.Limit = .unlimited,139 stdout_limit: Io.Limit = .unlimited,
140 stderr_limit: Io.Limit = .unlimited,140 stderr_limit: Io.Limit = .unlimited,
141 timeout: Io.Timeout = .none,
142};141};
143142
144/// Collect the output from the process's stdout and stderr. Will return once143/// Collect the output from the process's stdout and stderr. Will return once
...@@ -174,7 +173,7 @@ pub fn collectOutput(child: *const Child, io: Io, options: CollectOutputOptions)...@@ -174,7 +173,7 @@ pub fn collectOutput(child: *const Child, io: Io, options: CollectOutputOptions)
174 var all_done = true;173 var all_done = true;
175 var any_canceled = false;174 var any_canceled = false;
176 var other_err: (error{StreamTooLong} || Io.File.Reader.Error)!void = {};175 var other_err: (error{StreamTooLong} || Io.File.Reader.Error)!void = {};
177 const op_result = io.vtable.operate(io.userdata, &reads, 1, options.timeout);176 io.vtable.operate(io.userdata, &reads);
178 for (&reads, &lists, &limits, &dones) |*read, list, limit, *done| {177 for (&reads, &lists, &limits, &dones) |*read, list, limit, *done| {
179 if (done.*) continue;178 if (done.*) continue;
180 const n = read.file_read_streaming.result catch |err| switch (err) {179 const n = read.file_read_streaming.result catch |err| switch (err) {
...@@ -197,7 +196,6 @@ pub fn collectOutput(child: *const Child, io: Io, options: CollectOutputOptions)...@@ -197,7 +196,6 @@ pub fn collectOutput(child: *const Child, io: Io, options: CollectOutputOptions)
197 if (list.items.len > @intFromEnum(limit)) other_err = error.StreamTooLong;196 if (list.items.len > @intFromEnum(limit)) other_err = error.StreamTooLong;
198 }197 }
199 if (any_canceled) return error.Canceled;198 if (any_canceled) return error.Canceled;
200 try op_result; // could be error.Canceled
201 try other_err;199 try other_err;
202 if (all_done) return;200 if (all_done) return;
203 }201 }