authorgravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2026-01-07 18:38:19-08:00
committergravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2026-01-30 12:10:01-08:00
log6d22f7b4d7bd6d29ecbc5e3bd4a2e1c085f293a3
tree4c506ee95d083d90d793c2ed103f2d95307c24aa
parentdef22c2b63022b0c06dac8b5f7554deb644ca089

std.Io: proof-of-concept "operations" API

This commit shows a proof-of-concept direction for std.Io.VTable to go, which is to have general support for batching, timeouts, and non-blocking. I'm not sure if this is a good idea or not so I'm putting it up for scrutiny. This commit introduces `std.Io.operate`, `std.Io.Operation`, and implements it experimentally for `FileReadStreaming`. In `std.Io.Threaded`, the implementation is based on poll(). This commit shows how it can be used in `std.process.run` to collect both stdout and stderr in a single-threaded program using `std.Threaded.Io`. It also demonstrates how to upgrade code that was previously using `std.Io.poll` (*not* integrated with the interface!) using concurrency. This may not be ideal since it makes the build runner no longer support single-threaded mode. There is still a needed abstraction for conveniently reading multiple File streams concurrently without io.concurrent, but this commit demonstrates that such an API can be built on top of the new `std.Io.operate` functionality.

8 files changed, 272 insertions(+), 72 deletions(-)

lib/std/Build/Step.zig+32-15
...@@ -381,10 +381,15 @@ pub fn addError(step: *Step, comptime fmt: []const u8, args: anytype) error{OutO...@@ -381,10 +381,15 @@ pub fn addError(step: *Step, comptime fmt: []const u8, args: anytype) error{OutO
381381
382pub const ZigProcess = struct {382pub const ZigProcess = struct {
383 child: std.process.Child,383 child: std.process.Child,
384 poller: Io.Poller(StreamEnum),
385 progress_ipc_fd: if (std.Progress.have_ipc) ?std.posix.fd_t else void,384 progress_ipc_fd: if (std.Progress.have_ipc) ?std.posix.fd_t else void,
386385
387 pub const StreamEnum = enum { stdout, stderr };386 pub const StreamEnum = enum { stdout, stderr };
387
388 pub fn deinit(zp: *ZigProcess, gpa: Allocator, io: Io) void {
389 _ = gpa;
390 zp.child.kill(io);
391 zp.* = undefined;
392 }
388};393};
389394
390/// Assumes that argv contains `--listen=-` and that the process being spawned395/// Assumes that argv contains `--listen=-` and that the process being spawned
...@@ -459,14 +464,10 @@ pub fn evalZigProcess(...@@ -459,14 +464,10 @@ pub fn evalZigProcess(
459464
460 zp.* = .{465 zp.* = .{
461 .child = zp.child,466 .child = zp.child,
462 .poller = Io.poll(gpa, ZigProcess.StreamEnum, .{
463 .stdout = zp.child.stdout.?,
464 .stderr = zp.child.stderr.?,
465 }),
466 .progress_ipc_fd = if (std.Progress.have_ipc) prog_node.getIpcFd() else {},467 .progress_ipc_fd = if (std.Progress.have_ipc) prog_node.getIpcFd() else {},
467 };468 };
468 if (watch) s.setZigProcess(zp);469 if (watch) s.setZigProcess(zp);
469 defer if (!watch) zp.poller.deinit();470 defer if (!watch) zp.deinit(gpa, io);
470471
471 const result = try zigProcessUpdate(s, zp, watch, web_server, gpa);472 const result = try zigProcessUpdate(s, zp, watch, web_server, gpa);
472473
...@@ -526,6 +527,9 @@ fn zigProcessUpdate(s: *Step, zp: *ZigProcess, watch: bool, web_server: ?*Build....@@ -526,6 +527,9 @@ fn zigProcessUpdate(s: *Step, zp: *ZigProcess, watch: bool, web_server: ?*Build.
526 const arena = b.allocator;527 const arena = b.allocator;
527 const io = b.graph.io;528 const io = b.graph.io;
528529
530 var stderr_task = try io.concurrent(readStreamAlloc, .{ gpa, io, zp.child.stderr.?, .unlimited });
531 defer if (stderr_task.cancel(io)) |slice| gpa.free(slice) else |_| {};
532
529 var timer = try std.time.Timer.start();533 var timer = try std.time.Timer.start();
530534
531 try sendMessage(io, zp.child.stdin.?, .update);535 try sendMessage(io, zp.child.stdin.?, .update);
...@@ -533,14 +537,18 @@ fn zigProcessUpdate(s: *Step, zp: *ZigProcess, watch: bool, web_server: ?*Build....@@ -533,14 +537,18 @@ fn zigProcessUpdate(s: *Step, zp: *ZigProcess, watch: bool, web_server: ?*Build.
533537
534 var result: ?Path = null;538 var result: ?Path = null;
535539
536 const stdout = zp.poller.reader(.stdout);540 var stdout_buffer: [512]u8 = undefined;
541 var stdout_reader: Io.File.Reader = .initStreaming(zp.child.stdout.?, io, &stdout_buffer);
542 const stdout = &stdout_reader.interface;
543
544 var body_buffer: std.ArrayList(u8) = .empty;
537545
538 poll: while (true) {546 while (true) {
539 const Header = std.zig.Server.Message.Header;547 const Header = std.zig.Server.Message.Header;
540 while (stdout.buffered().len < @sizeOf(Header)) if (!try zp.poller.poll()) break :poll;548 const header = try stdout.takeStruct(Header, .little);
541 const header = stdout.takeStruct(Header, .little) catch unreachable;549 body_buffer.clearRetainingCapacity();
542 while (stdout.buffered().len < header.bytes_len) if (!try zp.poller.poll()) break :poll;550 try stdout.appendExact(gpa, &body_buffer, header.bytes_len);
543 const body = stdout.take(header.bytes_len) catch unreachable;551 const body = body_buffer.items;
544 switch (header.tag) {552 switch (header.tag) {
545 .zig_version => {553 .zig_version => {
546 if (!std.mem.eql(u8, builtin.zig_version_string, body)) {554 if (!std.mem.eql(u8, builtin.zig_version_string, body)) {
...@@ -553,11 +561,11 @@ fn zigProcessUpdate(s: *Step, zp: *ZigProcess, watch: bool, web_server: ?*Build....@@ -553,11 +561,11 @@ fn zigProcessUpdate(s: *Step, zp: *ZigProcess, watch: bool, web_server: ?*Build.
553 .error_bundle => {561 .error_bundle => {
554 s.result_error_bundle = try std.zig.Server.allocErrorBundle(gpa, body);562 s.result_error_bundle = try std.zig.Server.allocErrorBundle(gpa, body);
555 // This message indicates the end of the update.563 // This message indicates the end of the update.
556 if (watch) break :poll;564 if (watch) break;
557 },565 },
558 .emit_digest => {566 .emit_digest => {
559 const EmitDigest = std.zig.Server.Message.EmitDigest;567 const EmitDigest = std.zig.Server.Message.EmitDigest;
560 const emit_digest = @as(*align(1) const EmitDigest, @ptrCast(body));568 const emit_digest: *align(1) const EmitDigest = @ptrCast(body);
561 s.result_cached = emit_digest.flags.cache_hit;569 s.result_cached = emit_digest.flags.cache_hit;
562 const digest = body[@sizeOf(EmitDigest)..][0..Cache.bin_digest_len];570 const digest = body[@sizeOf(EmitDigest)..][0..Cache.bin_digest_len];
563 result = .{571 result = .{
...@@ -631,7 +639,8 @@ fn zigProcessUpdate(s: *Step, zp: *ZigProcess, watch: bool, web_server: ?*Build....@@ -631,7 +639,8 @@ fn zigProcessUpdate(s: *Step, zp: *ZigProcess, watch: bool, web_server: ?*Build.
631639
632 s.result_duration_ns = timer.read();640 s.result_duration_ns = timer.read();
633641
634 const stderr_contents = try zp.poller.toOwnedSlice(.stderr);642 const stderr_contents = try stderr_task.await(io);
643 defer gpa.free(stderr_contents);
635 if (stderr_contents.len > 0) {644 if (stderr_contents.len > 0) {
636 try s.result_error_msgs.append(arena, try arena.dupe(u8, stderr_contents));645 try s.result_error_msgs.append(arena, try arena.dupe(u8, stderr_contents));
637 }646 }
...@@ -639,6 +648,14 @@ fn zigProcessUpdate(s: *Step, zp: *ZigProcess, watch: bool, web_server: ?*Build....@@ -639,6 +648,14 @@ fn zigProcessUpdate(s: *Step, zp: *ZigProcess, watch: bool, web_server: ?*Build.
639 return result;648 return result;
640}649}
641650
651fn readStreamAlloc(gpa: Allocator, io: Io, file: Io.File, limit: Io.Limit) ![]u8 {
652 var file_reader: Io.File.Reader = .initStreaming(file, io, &.{});
653 return file_reader.interface.allocRemaining(gpa, limit) catch |err| switch (err) {
654 error.ReadFailed => return file_reader.err.?,
655 else => |e| return e,
656 };
657}
658
642pub fn getZigProcess(s: *Step) ?*ZigProcess {659pub fn getZigProcess(s: *Step) ?*ZigProcess {
643 return switch (s.id) {660 return switch (s.id) {
644 .compile => s.cast(Compile).?.zig_process,661 .compile => s.cast(Compile).?.zig_process,
lib/std/Io.zig+34-2
...@@ -149,6 +149,8 @@ pub const VTable = struct {...@@ -149,6 +149,8 @@ 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,
153
152 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,
153 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,
154 dirCreateDirPathOpen: *const fn (?*anyopaque, Dir, []const u8, Dir.Permissions, Dir.OpenOptions) Dir.CreateDirPathOpenError!Dir,156 dirCreateDirPathOpen: *const fn (?*anyopaque, Dir, []const u8, Dir.Permissions, Dir.OpenOptions) Dir.CreateDirPathOpenError!Dir,
...@@ -184,8 +186,6 @@ pub const VTable = struct {...@@ -184,8 +186,6 @@ pub const VTable = struct {
184 fileWriteFileStreaming: *const fn (?*anyopaque, File, header: []const u8, *Io.File.Reader, Io.Limit) File.Writer.WriteFileError!usize,186 fileWriteFileStreaming: *const fn (?*anyopaque, File, header: []const u8, *Io.File.Reader, Io.Limit) File.Writer.WriteFileError!usize,
185 fileWriteFilePositional: *const fn (?*anyopaque, File, header: []const u8, *Io.File.Reader, Io.Limit, offset: u64) File.WriteFilePositionalError!usize,187 fileWriteFilePositional: *const fn (?*anyopaque, File, header: []const u8, *Io.File.Reader, Io.Limit, offset: u64) File.WriteFilePositionalError!usize,
186 /// Returns 0 on end of stream.188 /// Returns 0 on end of stream.
187 fileReadStreaming: *const fn (?*anyopaque, File, data: []const []u8) File.Reader.Error!usize,
188 /// Returns 0 on end of stream.
189 fileReadPositional: *const fn (?*anyopaque, File, data: []const []u8, offset: u64) File.ReadPositionalError!usize,189 fileReadPositional: *const fn (?*anyopaque, File, data: []const []u8, offset: u64) File.ReadPositionalError!usize,
190 fileSeekBy: *const fn (?*anyopaque, File, relative_offset: i64) File.SeekError!void,190 fileSeekBy: *const fn (?*anyopaque, File, relative_offset: i64) File.SeekError!void,
191 fileSeekTo: *const fn (?*anyopaque, File, absolute_offset: u64) File.SeekError!void,191 fileSeekTo: *const fn (?*anyopaque, File, absolute_offset: u64) File.SeekError!void,
...@@ -252,6 +252,38 @@ pub const VTable = struct {...@@ -252,6 +252,38 @@ pub const VTable = struct {
252 netLookup: *const fn (?*anyopaque, net.HostName, *Queue(net.HostName.LookupResult), net.HostName.LookupOptions) net.HostName.LookupError!void,252 netLookup: *const fn (?*anyopaque, net.HostName, *Queue(net.HostName.LookupResult), net.HostName.LookupOptions) net.HostName.LookupError!void,
253};253};
254254
255pub const Operation = union(enum) {
256 noop,
257 file_read_streaming: FileReadStreaming,
258
259 pub const FileReadStreaming = struct {
260 file: File,
261 data: []const []u8,
262 /// Causes `result` to return `error.WouldBlock` instead of blocking.
263 nonblocking: bool = false,
264 /// Returns 0 on end of stream.
265 result: File.Reader.Error!usize,
266 };
267};
268
269pub const OperateError = error{ Canceled, Timeout };
270
271/// Performs all `operations` in a non-deterministic order. Returns after all
272/// `operations` have been attempted. The degree to which the operations are
273/// performed concurrently is determined by the `Io` implementation.
274///
275/// `n_wait` is an amount of operations between `0` and `operations.len` that
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}
286
255pub const Limit = enum(usize) {287pub const Limit = enum(usize) {
256 nothing = 0,288 nothing = 0,
257 unlimited = math.maxInt(usize),289 unlimited = math.maxInt(usize),
lib/std/Io/File.zig+7-1
...@@ -554,7 +554,13 @@ pub fn setTimestampsNow(file: File, io: Io) SetTimestampsError!void {...@@ -554,7 +554,13 @@ pub fn setTimestampsNow(file: File, io: Io) SetTimestampsError!void {
554/// See also:554/// See also:
555/// * `reader`555/// * `reader`
556pub fn readStreaming(file: File, io: Io, buffer: []const []u8) Reader.Error!usize {556pub fn readStreaming(file: File, io: Io, buffer: []const []u8) Reader.Error!usize {
557 return io.vtable.fileReadStreaming(io.userdata, file, buffer);557 var operation: Io.Operation = .{ .file_read_streaming = .{
558 .file = file,
559 .data = buffer,
560 .result = undefined,
561 } };
562 io.vtable.operate(io.userdata, (&operation)[0..1], 1, .none) catch unreachable;
563 return operation.file_read_streaming.result;
558}564}
559565
560pub const ReadPositionalError = error{566pub const ReadPositionalError = error{
lib/std/Io/File/Reader.zig+2-2
...@@ -300,7 +300,7 @@ fn readVecStreaming(r: *Reader, data: [][]u8) Io.Reader.Error!usize {...@@ -300,7 +300,7 @@ fn readVecStreaming(r: *Reader, data: [][]u8) Io.Reader.Error!usize {
300 const dest_n, const data_size = try r.interface.writableVector(&iovecs_buffer, data);300 const dest_n, const data_size = try r.interface.writableVector(&iovecs_buffer, data);
301 const dest = iovecs_buffer[0..dest_n];301 const dest = iovecs_buffer[0..dest_n];
302 assert(dest[0].len > 0);302 assert(dest[0].len > 0);
303 const n = io.vtable.fileReadStreaming(io.userdata, r.file, dest) catch |err| {303 const n = r.file.readStreaming(io, dest) catch |err| {
304 r.err = err;304 r.err = err;
305 return error.ReadFailed;305 return error.ReadFailed;
306 };306 };
...@@ -355,7 +355,7 @@ fn discard(io_reader: *Io.Reader, limit: Io.Limit) Io.Reader.Error!usize {...@@ -355,7 +355,7 @@ fn discard(io_reader: *Io.Reader, limit: Io.Limit) Io.Reader.Error!usize {
355 const dest_n, const data_size = try r.interface.writableVector(&iovecs_buffer, &data);355 const dest_n, const data_size = try r.interface.writableVector(&iovecs_buffer, &data);
356 const dest = iovecs_buffer[0..dest_n];356 const dest = iovecs_buffer[0..dest_n];
357 assert(dest[0].len > 0);357 assert(dest[0].len > 0);
358 const n = io.vtable.fileReadStreaming(io.userdata, file, dest) catch |err| {358 const n = file.readStreaming(io, dest) catch |err| {
359 r.err = err;359 r.err = err;
360 return error.ReadFailed;360 return error.ReadFailed;
361 };361 };
lib/std/Io/Reader.zig+21
...@@ -315,6 +315,27 @@ pub fn allocRemainingAlignedSentinel(...@@ -315,6 +315,27 @@ pub fn allocRemainingAlignedSentinel(
315 }315 }
316}316}
317317
318pub const AppendExactError = Allocator.Error || Error;
319
320/// Transfers exactly `n` bytes from the reader to the `ArrayList`.
321///
322/// See also:
323/// * `appendRemaining`
324pub fn appendExact(
325 r: *Reader,
326 gpa: Allocator,
327 list: *ArrayList(u8),
328 n: usize,
329) AppendExactError!void {
330 try list.ensureUnusedCapacity(gpa, n);
331 var a = std.Io.Writer.Allocating.fromArrayList(gpa, list);
332 defer list.* = a.toArrayList();
333 streamExact(r, &a.writer, n) catch |err| switch (err) {
334 error.ReadFailed, error.EndOfStream => |e| return e,
335 error.WriteFailed => unreachable,
336 };
337}
338
318/// Transfers all bytes from the current position to the end of the stream, up339/// Transfers all bytes from the current position to the end of the stream, up
319/// to `limit`, appending them to `list`.340/// to `limit`, appending them to `list`.
320///341///
lib/std/Io/Threaded.zig+85-2
...@@ -1586,6 +1586,8 @@ pub fn io(t: *Threaded) Io {...@@ -1586,6 +1586,8 @@ pub fn io(t: *Threaded) Io {
1586 .futexWaitUncancelable = futexWaitUncancelable,1586 .futexWaitUncancelable = futexWaitUncancelable,
1587 .futexWake = futexWake,1587 .futexWake = futexWake,
15881588
1589 .operate = operate,
1590
1589 .dirCreateDir = dirCreateDir,1591 .dirCreateDir = dirCreateDir,
1590 .dirCreateDirPath = dirCreateDirPath,1592 .dirCreateDirPath = dirCreateDirPath,
1591 .dirCreateDirPathOpen = dirCreateDirPathOpen,1593 .dirCreateDirPathOpen = dirCreateDirPathOpen,
...@@ -1620,7 +1622,6 @@ pub fn io(t: *Threaded) Io {...@@ -1620,7 +1622,6 @@ pub fn io(t: *Threaded) Io {
1620 .fileWritePositional = fileWritePositional,1622 .fileWritePositional = fileWritePositional,
1621 .fileWriteFileStreaming = fileWriteFileStreaming,1623 .fileWriteFileStreaming = fileWriteFileStreaming,
1622 .fileWriteFilePositional = fileWriteFilePositional,1624 .fileWriteFilePositional = fileWriteFilePositional,
1623 .fileReadStreaming = fileReadStreaming,
1624 .fileReadPositional = fileReadPositional,1625 .fileReadPositional = fileReadPositional,
1625 .fileSeekBy = fileSeekBy,1626 .fileSeekBy = fileSeekBy,
1626 .fileSeekTo = fileSeekTo,1627 .fileSeekTo = fileSeekTo,
...@@ -1746,6 +1747,8 @@ pub fn ioBasic(t: *Threaded) Io {...@@ -1746,6 +1747,8 @@ pub fn ioBasic(t: *Threaded) Io {
1746 .futexWaitUncancelable = futexWaitUncancelable,1747 .futexWaitUncancelable = futexWaitUncancelable,
1747 .futexWake = futexWake,1748 .futexWake = futexWake,
17481749
1750 .operate = operate,
1751
1749 .dirCreateDir = dirCreateDir,1752 .dirCreateDir = dirCreateDir,
1750 .dirCreateDirPath = dirCreateDirPath,1753 .dirCreateDirPath = dirCreateDirPath,
1751 .dirCreateDirPathOpen = dirCreateDirPathOpen,1754 .dirCreateDirPathOpen = dirCreateDirPathOpen,
...@@ -1780,7 +1783,6 @@ pub fn ioBasic(t: *Threaded) Io {...@@ -1780,7 +1783,6 @@ pub fn ioBasic(t: *Threaded) Io {
1780 .fileWritePositional = fileWritePositional,1783 .fileWritePositional = fileWritePositional,
1781 .fileWriteFileStreaming = fileWriteFileStreaming,1784 .fileWriteFileStreaming = fileWriteFileStreaming,
1782 .fileWriteFilePositional = fileWriteFilePositional,1785 .fileWriteFilePositional = fileWriteFilePositional,
1783 .fileReadStreaming = fileReadStreaming,
1784 .fileReadPositional = fileReadPositional,1786 .fileReadPositional = fileReadPositional,
1785 .fileSeekBy = fileSeekBy,1787 .fileSeekBy = fileSeekBy,
1786 .fileSeekTo = fileSeekTo,1788 .fileSeekTo = fileSeekTo,
...@@ -2447,6 +2449,87 @@ fn futexWake(userdata: ?*anyopaque, ptr: *const u32, max_waiters: u32) void {...@@ -2447,6 +2449,87 @@ fn futexWake(userdata: ?*anyopaque, ptr: *const u32, max_waiters: u32) void {
2447 Thread.futexWake(ptr, max_waiters);2449 Thread.futexWake(ptr, max_waiters);
2448}2450}
24492451
2452fn operate(userdata: ?*anyopaque, operations: []Io.Operation, n_wait: usize, timeout: Io.Timeout) Io.OperateError!void {
2453 const t: *Threaded = @ptrCast(@alignCast(userdata));
2454 const t_io = ioBasic(t);
2455
2456 if (is_windows) @panic("TODO");
2457
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;
2463 var map_buffer: [poll_buffer.len]u8 = undefined; // poll_buffer index to operations index
2464 var poll_i: usize = 0;
2465 var completed: usize = 0;
2466
2467 // Put all the file reads with nonblocking enabled into the poll set.
2468 if (operations.len > poll_buffer.len) @panic("TODO");
2469
2470 // TODO if any operation is canceled, cancel the rest
2471
2472 for (operations, 0..) |*operation, operation_index| switch (operation.*) {
2473 .noop => continue,
2474 .file_read_streaming => |*o| {
2475 if (o.nonblocking) {
2476 o.result = error.WouldBlock;
2477 poll_buffer[poll_i] = .{
2478 .fd = o.file.handle,
2479 .events = posix.POLL.IN,
2480 .revents = undefined,
2481 };
2482 map_buffer[poll_i] = @intCast(operation_index);
2483 poll_i += 1;
2484 } else {
2485 o.result = fileReadStreaming(o.file, o.data);
2486 completed += 1;
2487 }
2488 },
2489 };
2490
2491 if (poll_i == 0) {
2492 @branchHint(.likely);
2493 return;
2494 }
2495
2496 const max_poll_ms = std.math.maxInt(i32);
2497
2498 while (completed < n_wait) {
2499 const timeout_ms: i32 = if (deadline) |d| t: {
2500 const duration = d.durationFromNow(t_io) catch @panic("TODO make this unreachable");
2501 if (duration.raw.nanoseconds <= 0) return error.Timeout;
2502 break :t @intCast(@min(max_poll_ms, duration.raw.toMilliseconds()));
2503 } else -1;
2504 const syscall = try Syscall.start();
2505 const poll_rc = posix.system.poll(&poll_buffer, poll_i, timeout_ms);
2506 syscall.finish();
2507 switch (posix.errno(poll_rc)) {
2508 .SUCCESS => {
2509 if (poll_rc == 0) {
2510 // Although spurious timeouts are OK, when no deadline
2511 // is passed we must not return `error.Timeout`.
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 }
2526 },
2527 .INTR => continue,
2528 else => @panic("TODO handle unexpected error from poll()"),
2529 }
2530 }
2531}
2532
2450const dirCreateDir = switch (native_os) {2533const dirCreateDir = switch (native_os) {
2451 .windows => dirCreateDirWindows,2534 .windows => dirCreateDirWindows,
2452 .wasi => dirCreateDirWasi,2535 .wasi => dirCreateDirWasi,
lib/std/process.zig+19-4
...@@ -454,13 +454,17 @@ pub fn spawnPath(io: Io, dir: Io.Dir, options: SpawnOptions) SpawnError!Child {...@@ -454,13 +454,17 @@ pub fn spawnPath(io: Io, dir: Io.Dir, options: SpawnOptions) SpawnError!Child {
454}454}
455455
456pub const RunError = CurrentPathError || posix.ReadError || SpawnError || posix.PollError || error{456pub const RunError = CurrentPathError || posix.ReadError || SpawnError || posix.PollError || error{
457 StdoutStreamTooLong,457 StreamTooLong,
458 StderrStreamTooLong,
459};458};
460459
461pub const RunOptions = struct {460pub const RunOptions = struct {
462 argv: []const []const u8,461 argv: []const []const u8,
463 max_output_bytes: usize = 50 * 1024,462 stderr_limit: Io.Limit = .unlimited,
463 stdout_limit: Io.Limit = .unlimited,
464 /// How many bytes to initially allocate for stderr.
465 stderr_reserve_amount: usize = 1,
466 /// How many bytes to initially allocate for stdout.
467 stdout_reserve_amount: usize = 1,
464468
465 /// Set to change the current working directory when spawning the child process.469 /// Set to change the current working directory when spawning the child process.
466 cwd: ?[]const u8 = null,470 cwd: ?[]const u8 = null,
...@@ -486,6 +490,7 @@ pub const RunOptions = struct {...@@ -486,6 +490,7 @@ pub const RunOptions = struct {
486 create_no_window: bool = true,490 create_no_window: bool = true,
487 /// Darwin-only. Disable ASLR for the child process.491 /// Darwin-only. Disable ASLR for the child process.
488 disable_aslr: bool = false,492 disable_aslr: bool = false,
493 timeout: Io.Timeout = .none,
489};494};
490495
491pub const RunResult = struct {496pub const RunResult = struct {
...@@ -518,7 +523,17 @@ pub fn run(gpa: Allocator, io: Io, options: RunOptions) RunError!RunResult {...@@ -518,7 +523,17 @@ pub fn run(gpa: Allocator, io: Io, options: RunOptions) RunError!RunResult {
518 var stderr: std.ArrayList(u8) = .empty;523 var stderr: std.ArrayList(u8) = .empty;
519 defer stderr.deinit(gpa);524 defer stderr.deinit(gpa);
520525
521 try child.collectOutput(gpa, &stdout, &stderr, options.max_output_bytes);526 try stdout.ensureUnusedCapacity(gpa, options.stdout_reserve_amount);
527 try stderr.ensureUnusedCapacity(gpa, options.stderr_reserve_amount);
528
529 try child.collectOutput(io, .{
530 .allocator = gpa,
531 .stdout = &stdout,
532 .stderr = &stderr,
533 .stdout_limit = options.stdout_limit,
534 .stderr_limit = options.stderr_limit,
535 .timeout = options.timeout,
536 });
522537
523 const term = try child.wait(io);538 const term = try child.wait(io);
524539
lib/std/process/Child.zig+72-46
...@@ -9,7 +9,6 @@ const process = std.process;...@@ -9,7 +9,6 @@ const process = std.process;
9const File = std.Io.File;9const File = std.Io.File;
10const assert = std.debug.assert;10const assert = std.debug.assert;
11const Allocator = std.mem.Allocator;11const Allocator = std.mem.Allocator;
12const ArrayList = std.ArrayList;
1312
14pub const Id = switch (native_os) {13pub const Id = switch (native_os) {
15 .windows => std.os.windows.HANDLE,14 .windows => std.os.windows.HANDLE,
...@@ -126,53 +125,80 @@ pub fn wait(child: *Child, io: Io) WaitError!Term {...@@ -126,53 +125,80 @@ pub fn wait(child: *Child, io: Io) WaitError!Term {
126 return io.vtable.childWait(io.userdata, child);125 return io.vtable.childWait(io.userdata, child);
127}126}
128127
129/// Collect the output from the process's stdout and stderr. Will return once all output128pub const CollectOutputError = error{
130/// has been collected. This does not mean that the process has ended. `wait` should still129 Timeout,
131/// be called to wait for and clean up the process.130 StreamTooLong,
131} || Allocator.Error || Io.File.Reader.Error;
132
133pub const CollectOutputOptions = struct {
134 stdout: *std.ArrayList(u8),
135 stderr: *std.ArrayList(u8),
136 /// Used for `stdout` and `stderr`. If not provided, only the existing
137 /// capacity will be used.
138 allocator: ?Allocator = null,
139 stdout_limit: Io.Limit = .unlimited,
140 stderr_limit: Io.Limit = .unlimited,
141 timeout: Io.Timeout = .none,
142};
143
144/// Collect the output from the process's stdout and stderr. Will return once
145/// all output has been collected. This does not mean that the process has
146/// ended. `wait` should still be called to wait for and clean up the process.
132///147///
133/// The process must have been started with stdout and stderr set to148/// The process must have been started with stdout and stderr set to
134/// `process.SpawnOptions.StdIo.pipe`.149/// `process.SpawnOptions.StdIo.pipe`.
135pub fn collectOutput(150pub fn collectOutput(child: *const Child, io: Io, options: CollectOutputOptions) CollectOutputError!void {
136 child: *const Child,151 const files: [2]Io.File = .{ child.stdout.?, child.stderr.? };
137 /// Used for `stdout` and `stderr`.152 const lists: [2]*std.ArrayList(u8) = .{ options.stdout, options.stderr };
138 allocator: Allocator,153 const limits: [2]Io.Limit = .{ options.stdout_limit, options.stderr_limit };
139 stdout: *ArrayList(u8),154 var dones: [2]bool = .{ false, false };
140 stderr: *ArrayList(u8),155 var reads: [2]Io.Operation = undefined;
141 max_output_bytes: usize,156 var vecs: [2][1][]u8 = undefined;
142) !void {157 while (true) {
143 var poller = std.Io.poll(allocator, enum { stdout, stderr }, .{158 for (&reads, &lists, &files, dones, &vecs) |*read, list, file, done, *vec| {
144 .stdout = child.stdout.?,159 if (done) {
145 .stderr = child.stderr.?,160 read.* = .noop;
146 });161 continue;
147 defer poller.deinit();162 }
148163 if (options.allocator) |gpa| try list.ensureUnusedCapacity(gpa, 1);
149 const stdout_r = poller.reader(.stdout);164 const cap = list.unusedCapacitySlice();
150 stdout_r.buffer = stdout.allocatedSlice();165 if (cap.len == 0) return error.StreamTooLong;
151 stdout_r.seek = 0;166 vec[0] = cap;
152 stdout_r.end = stdout.items.len;167 read.* = .{ .file_read_streaming = .{
153168 .file = file,
154 const stderr_r = poller.reader(.stderr);169 .data = vec,
155 stderr_r.buffer = stderr.allocatedSlice();170 .nonblocking = true,
156 stderr_r.seek = 0;171 .result = undefined,
157 stderr_r.end = stderr.items.len;172 } };
158173 }
159 defer {174 var all_done = true;
160 stdout.* = .{175 var any_canceled = false;
161 .items = stdout_r.buffer[0..stdout_r.end],176 var other_err: (error{StreamTooLong} || Io.File.Reader.Error)!void = {};
162 .capacity = stdout_r.buffer.len,177 const op_result = io.vtable.operate(io.userdata, &reads, 1, options.timeout);
163 };178 for (&reads, &lists, &limits, &dones) |*read, list, limit, *done| {
164 stderr.* = .{179 if (done.*) continue;
165 .items = stderr_r.buffer[0..stderr_r.end],180 const n = read.file_read_streaming.result catch |err| switch (err) {
166 .capacity = stderr_r.buffer.len,181 error.Canceled => {
167 };182 any_canceled = true;
168 stdout_r.buffer = &.{};183 continue;
169 stderr_r.buffer = &.{};184 },
170 }185 error.WouldBlock => continue,
171186 else => |e| {
172 while (try poller.poll()) {187 other_err = e;
173 if (stdout_r.bufferedLen() > max_output_bytes)188 continue;
174 return error.StdoutStreamTooLong;189 },
175 if (stderr_r.bufferedLen() > max_output_bytes)190 };
176 return error.StderrStreamTooLong;191 if (n == 0) {
192 done.* = true;
193 } else {
194 all_done = false;
195 }
196 list.items.len += n;
197 if (list.items.len > @intFromEnum(limit)) other_err = error.StreamTooLong;
198 }
199 if (any_canceled) return error.Canceled;
200 try op_result; // could be error.Canceled
201 try other_err;
202 if (all_done) return;
177 }203 }
178}204}