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-09 20:47:24-08:00
logd63172a35da5d493aef72aa074afa90ac5bb619a
tree8b7254d33b34e466eaf59cdbd5f83491d401250a
parent31fadc0eee2dc25983f493eca7e843629b917b2b

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, 274 insertions(+), 80 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
...@@ -148,6 +148,8 @@ pub const VTable = struct {...@@ -148,6 +148,8 @@ pub const VTable = struct {
148 futexWaitUncancelable: *const fn (?*anyopaque, ptr: *const u32, expected: u32) void,148 futexWaitUncancelable: *const fn (?*anyopaque, ptr: *const u32, expected: u32) void,
149 futexWake: *const fn (?*anyopaque, ptr: *const u32, max_waiters: u32) void,149 futexWake: *const fn (?*anyopaque, ptr: *const u32, max_waiters: u32) void,
150150
151 operate: *const fn (?*anyopaque, []Operation, n_wait: usize, Timeout) OperateError!void,
152
151 dirCreateDir: *const fn (?*anyopaque, Dir, []const u8, Dir.Permissions) Dir.CreateDirError!void,153 dirCreateDir: *const fn (?*anyopaque, Dir, []const u8, Dir.Permissions) Dir.CreateDirError!void,
152 dirCreateDirPath: *const fn (?*anyopaque, Dir, []const u8, Dir.Permissions) Dir.CreateDirPathError!Dir.CreatePathStatus,154 dirCreateDirPath: *const fn (?*anyopaque, Dir, []const u8, Dir.Permissions) Dir.CreateDirPathError!Dir.CreatePathStatus,
153 dirCreateDirPathOpen: *const fn (?*anyopaque, Dir, []const u8, Dir.Permissions, Dir.OpenOptions) Dir.CreateDirPathOpenError!Dir,155 dirCreateDirPathOpen: *const fn (?*anyopaque, Dir, []const u8, Dir.Permissions, Dir.OpenOptions) Dir.CreateDirPathOpenError!Dir,
...@@ -183,8 +185,6 @@ pub const VTable = struct {...@@ -183,8 +185,6 @@ pub const VTable = struct {
183 fileWriteFileStreaming: *const fn (?*anyopaque, File, header: []const u8, *Io.File.Reader, Io.Limit) File.Writer.WriteFileError!usize,185 fileWriteFileStreaming: *const fn (?*anyopaque, File, header: []const u8, *Io.File.Reader, Io.Limit) File.Writer.WriteFileError!usize,
184 fileWriteFilePositional: *const fn (?*anyopaque, File, header: []const u8, *Io.File.Reader, Io.Limit, offset: u64) File.WriteFilePositionalError!usize,186 fileWriteFilePositional: *const fn (?*anyopaque, File, header: []const u8, *Io.File.Reader, Io.Limit, offset: u64) File.WriteFilePositionalError!usize,
185 /// Returns 0 on end of stream.187 /// Returns 0 on end of stream.
186 fileReadStreaming: *const fn (?*anyopaque, File, data: []const []u8) File.Reader.Error!usize,
187 /// Returns 0 on end of stream.
188 fileReadPositional: *const fn (?*anyopaque, File, data: []const []u8, offset: u64) File.ReadPositionalError!usize,188 fileReadPositional: *const fn (?*anyopaque, File, data: []const []u8, offset: u64) File.ReadPositionalError!usize,
189 fileSeekBy: *const fn (?*anyopaque, File, relative_offset: i64) File.SeekError!void,189 fileSeekBy: *const fn (?*anyopaque, File, relative_offset: i64) File.SeekError!void,
190 fileSeekTo: *const fn (?*anyopaque, File, absolute_offset: u64) File.SeekError!void,190 fileSeekTo: *const fn (?*anyopaque, File, absolute_offset: u64) File.SeekError!void,
...@@ -243,6 +243,38 @@ pub const VTable = struct {...@@ -243,6 +243,38 @@ pub const VTable = struct {
243 netLookup: *const fn (?*anyopaque, net.HostName, *Queue(net.HostName.LookupResult), net.HostName.LookupOptions) net.HostName.LookupError!void,243 netLookup: *const fn (?*anyopaque, net.HostName, *Queue(net.HostName.LookupResult), net.HostName.LookupOptions) net.HostName.LookupError!void,
244};244};
245245
246pub const Operation = union(enum) {
247 noop,
248 file_read_streaming: FileReadStreaming,
249
250 pub const FileReadStreaming = struct {
251 file: File,
252 data: []const []u8,
253 /// Causes `result` to return `error.WouldBlock` instead of blocking.
254 nonblocking: bool = false,
255 /// Returns 0 on end of stream.
256 result: File.Reader.Error!usize,
257 };
258};
259
260pub const OperateError = error{ Canceled, Timeout };
261
262/// Performs all `operations` in a non-deterministic order. Returns after all
263/// `operations` have been attempted. The degree to which the operations are
264/// performed concurrently is determined by the `Io` implementation.
265///
266/// `n_wait` is an amount of operations between `0` and `operations.len` that
267/// determines how many attempted operations must complete before `operate`
268/// returns. Operation completion is defined by returning a value other than
269/// `error.WouldBlock`. If the operation cannot return `error.WouldBlock`, it
270/// always counts as completing.
271///
272/// In the event `error.Canceled` is returned, any number of `operations` may
273/// still have been completed successfully.
274pub fn operate(io: Io, operations: []Operation, n_wait: usize, timeout: Timeout) OperateError!void {
275 return io.vtable.operate(io.userdata, operations, n_wait, timeout);
276}
277
246pub const Limit = enum(usize) {278pub const Limit = enum(usize) {
247 nothing = 0,279 nothing = 0,
248 unlimited = math.maxInt(usize),280 unlimited = math.maxInt(usize),
lib/std/Io/File.zig+7-1
...@@ -526,7 +526,13 @@ pub fn setTimestampsNow(file: File, io: Io) SetTimestampsError!void {...@@ -526,7 +526,13 @@ pub fn setTimestampsNow(file: File, io: Io) SetTimestampsError!void {
526/// See also:526/// See also:
527/// * `reader`527/// * `reader`
528pub fn readStreaming(file: File, io: Io, buffer: []const []u8) Reader.Error!usize {528pub fn readStreaming(file: File, io: Io, buffer: []const []u8) Reader.Error!usize {
529 return io.vtable.fileReadStreaming(io.userdata, file, buffer);529 var operation: Io.Operation = .{ .file_read_streaming = .{
530 .file = file,
531 .data = buffer,
532 .result = undefined,
533 } };
534 io.vtable.operate(io.userdata, (&operation)[0..1], 1, .none) catch unreachable;
535 return operation.file_read_streaming.result;
530}536}
531537
532pub const ReadPositionalError = Reader.Error || error{Unseekable};538pub const ReadPositionalError = Reader.Error || error{Unseekable};
lib/std/Io/File/Reader.zig+2-2
...@@ -301,7 +301,7 @@ fn readVecStreaming(r: *Reader, data: [][]u8) Io.Reader.Error!usize {...@@ -301,7 +301,7 @@ fn readVecStreaming(r: *Reader, data: [][]u8) Io.Reader.Error!usize {
301 const dest_n, const data_size = try r.interface.writableVector(&iovecs_buffer, data);301 const dest_n, const data_size = try r.interface.writableVector(&iovecs_buffer, data);
302 const dest = iovecs_buffer[0..dest_n];302 const dest = iovecs_buffer[0..dest_n];
303 assert(dest[0].len > 0);303 assert(dest[0].len > 0);
304 const n = io.vtable.fileReadStreaming(io.userdata, r.file, dest) catch |err| {304 const n = r.file.readStreaming(io, dest) catch |err| {
305 r.err = err;305 r.err = err;
306 return error.ReadFailed;306 return error.ReadFailed;
307 };307 };
...@@ -356,7 +356,7 @@ fn discard(io_reader: *Io.Reader, limit: Io.Limit) Io.Reader.Error!usize {...@@ -356,7 +356,7 @@ fn discard(io_reader: *Io.Reader, limit: Io.Limit) Io.Reader.Error!usize {
356 const dest_n, const data_size = try r.interface.writableVector(&iovecs_buffer, &data);356 const dest_n, const data_size = try r.interface.writableVector(&iovecs_buffer, &data);
357 const dest = iovecs_buffer[0..dest_n];357 const dest = iovecs_buffer[0..dest_n];
358 assert(dest[0].len > 0);358 assert(dest[0].len > 0);
359 const n = io.vtable.fileReadStreaming(io.userdata, file, dest) catch |err| {359 const n = file.readStreaming(io, dest) catch |err| {
360 r.err = err;360 r.err = err;
361 return error.ReadFailed;361 return error.ReadFailed;
362 };362 };
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+87-10
...@@ -1437,6 +1437,8 @@ pub fn io(t: *Threaded) Io {...@@ -1437,6 +1437,8 @@ pub fn io(t: *Threaded) Io {
1437 .futexWaitUncancelable = futexWaitUncancelable,1437 .futexWaitUncancelable = futexWaitUncancelable,
1438 .futexWake = futexWake,1438 .futexWake = futexWake,
14391439
1440 .operate = operate,
1441
1440 .dirCreateDir = dirCreateDir,1442 .dirCreateDir = dirCreateDir,
1441 .dirCreateDirPath = dirCreateDirPath,1443 .dirCreateDirPath = dirCreateDirPath,
1442 .dirCreateDirPathOpen = dirCreateDirPathOpen,1444 .dirCreateDirPathOpen = dirCreateDirPathOpen,
...@@ -1471,7 +1473,6 @@ pub fn io(t: *Threaded) Io {...@@ -1471,7 +1473,6 @@ pub fn io(t: *Threaded) Io {
1471 .fileWritePositional = fileWritePositional,1473 .fileWritePositional = fileWritePositional,
1472 .fileWriteFileStreaming = fileWriteFileStreaming,1474 .fileWriteFileStreaming = fileWriteFileStreaming,
1473 .fileWriteFilePositional = fileWriteFilePositional,1475 .fileWriteFilePositional = fileWriteFilePositional,
1474 .fileReadStreaming = fileReadStreaming,
1475 .fileReadPositional = fileReadPositional,1476 .fileReadPositional = fileReadPositional,
1476 .fileSeekBy = fileSeekBy,1477 .fileSeekBy = fileSeekBy,
1477 .fileSeekTo = fileSeekTo,1478 .fileSeekTo = fileSeekTo,
...@@ -1589,6 +1590,8 @@ pub fn ioBasic(t: *Threaded) Io {...@@ -1589,6 +1590,8 @@ pub fn ioBasic(t: *Threaded) Io {
1589 .futexWaitUncancelable = futexWaitUncancelable,1590 .futexWaitUncancelable = futexWaitUncancelable,
1590 .futexWake = futexWake,1591 .futexWake = futexWake,
15911592
1593 .operate = operate,
1594
1592 .dirCreateDir = dirCreateDir,1595 .dirCreateDir = dirCreateDir,
1593 .dirCreateDirPath = dirCreateDirPath,1596 .dirCreateDirPath = dirCreateDirPath,
1594 .dirCreateDirPathOpen = dirCreateDirPathOpen,1597 .dirCreateDirPathOpen = dirCreateDirPathOpen,
...@@ -1623,7 +1626,6 @@ pub fn ioBasic(t: *Threaded) Io {...@@ -1623,7 +1626,6 @@ pub fn ioBasic(t: *Threaded) Io {
1623 .fileWritePositional = fileWritePositional,1626 .fileWritePositional = fileWritePositional,
1624 .fileWriteFileStreaming = fileWriteFileStreaming,1627 .fileWriteFileStreaming = fileWriteFileStreaming,
1625 .fileWriteFilePositional = fileWriteFilePositional,1628 .fileWriteFilePositional = fileWriteFilePositional,
1626 .fileReadStreaming = fileReadStreaming,
1627 .fileReadPositional = fileReadPositional,1629 .fileReadPositional = fileReadPositional,
1628 .fileSeekBy = fileSeekBy,1630 .fileSeekBy = fileSeekBy,
1629 .fileSeekTo = fileSeekTo,1631 .fileSeekTo = fileSeekTo,
...@@ -2266,6 +2268,87 @@ fn futexWake(userdata: ?*anyopaque, ptr: *const u32, max_waiters: u32) void {...@@ -2266,6 +2268,87 @@ fn futexWake(userdata: ?*anyopaque, ptr: *const u32, max_waiters: u32) void {
2266 Thread.futexWake(ptr, max_waiters);2268 Thread.futexWake(ptr, max_waiters);
2267}2269}
22682270
2271fn operate(userdata: ?*anyopaque, operations: []Io.Operation, n_wait: usize, timeout: Io.Timeout) Io.OperateError!void {
2272 const t: *Threaded = @ptrCast(@alignCast(userdata));
2273 const t_io = ioBasic(t);
2274
2275 if (is_windows) @panic("TODO");
2276
2277 const deadline = timeout.toDeadline(t_io) catch |err| switch (err) {
2278 error.UnsupportedClock, error.Unexpected => null,
2279 };
2280
2281 var poll_buffer: [100]posix.pollfd = undefined;
2282 var map_buffer: [poll_buffer.len]u8 = undefined; // poll_buffer index to operations index
2283 var poll_i: usize = 0;
2284 var completed: usize = 0;
2285
2286 // Put all the file reads with nonblocking enabled into the poll set.
2287 if (operations.len > poll_buffer.len) @panic("TODO");
2288
2289 // TODO if any operation is canceled, cancel the rest
2290
2291 for (operations, 0..) |*operation, operation_index| switch (operation.*) {
2292 .noop => continue,
2293 .file_read_streaming => |*o| {
2294 if (o.nonblocking) {
2295 o.result = error.WouldBlock;
2296 poll_buffer[poll_i] = .{
2297 .fd = o.file.handle,
2298 .events = posix.POLL.IN,
2299 .revents = undefined,
2300 };
2301 map_buffer[poll_i] = @intCast(operation_index);
2302 poll_i += 1;
2303 } else {
2304 o.result = fileReadStreaming(o.file, o.data);
2305 completed += 1;
2306 }
2307 },
2308 };
2309
2310 if (poll_i == 0) {
2311 @branchHint(.likely);
2312 return;
2313 }
2314
2315 const max_poll_ms = std.math.maxInt(i32);
2316
2317 while (completed < n_wait) {
2318 const timeout_ms: i32 = if (deadline) |d| t: {
2319 const duration = d.durationFromNow(t_io) catch @panic("TODO make this unreachable");
2320 if (duration.raw.nanoseconds <= 0) return error.Timeout;
2321 break :t @intCast(@min(max_poll_ms, duration.raw.toMilliseconds()));
2322 } else -1;
2323 const syscall = try Syscall.start();
2324 const poll_rc = posix.system.poll(&poll_buffer, poll_i, timeout_ms);
2325 syscall.finish();
2326 switch (posix.errno(poll_rc)) {
2327 .SUCCESS => {
2328 if (poll_rc == 0) {
2329 // Although spurious timeouts are OK, when no deadline
2330 // is passed we must not return `error.Timeout`.
2331 if (deadline == null) continue;
2332 return error.Timeout;
2333 }
2334 for (poll_buffer[0..poll_i], map_buffer[0..poll_i]) |*poll_fd, operation_index| {
2335 if (poll_fd.revents == 0) continue;
2336 poll_fd.fd = -1; // Disarm this operation.
2337 switch (operations[operation_index]) {
2338 .noop => unreachable,
2339 .file_read_streaming => |*o| {
2340 o.result = fileReadStreaming(o.file, o.data);
2341 completed += 1;
2342 },
2343 }
2344 }
2345 },
2346 .INTR => continue,
2347 else => @panic("TODO handle unexpected error from poll()"),
2348 }
2349 }
2350}
2351
2269const dirCreateDir = switch (native_os) {2352const dirCreateDir = switch (native_os) {
2270 .windows => dirCreateDirWindows,2353 .windows => dirCreateDirWindows,
2271 .wasi => dirCreateDirWasi,2354 .wasi => dirCreateDirWasi,
...@@ -7865,10 +7948,7 @@ const fileReadStreaming = switch (native_os) {...@@ -7865,10 +7948,7 @@ const fileReadStreaming = switch (native_os) {
7865 else => fileReadStreamingPosix,7948 else => fileReadStreamingPosix,
7866};7949};
78677950
7868fn fileReadStreamingPosix(userdata: ?*anyopaque, file: File, data: []const []u8) File.Reader.Error!usize {7951fn fileReadStreamingPosix(file: File, data: []const []u8) File.Reader.Error!usize {
7869 const t: *Threaded = @ptrCast(@alignCast(userdata));
7870 _ = t;
7871
7872 var iovecs_buffer: [max_iovecs_len]posix.iovec = undefined;7952 var iovecs_buffer: [max_iovecs_len]posix.iovec = undefined;
7873 var i: usize = 0;7953 var i: usize = 0;
7874 for (data) |buf| {7954 for (data) |buf| {
...@@ -7952,10 +8032,7 @@ fn fileReadStreamingPosix(userdata: ?*anyopaque, file: File, data: []const []u8)...@@ -7952,10 +8032,7 @@ fn fileReadStreamingPosix(userdata: ?*anyopaque, file: File, data: []const []u8)
7952 }8032 }
7953}8033}
79548034
7955fn fileReadStreamingWindows(userdata: ?*anyopaque, file: File, data: []const []u8) File.Reader.Error!usize {8035fn fileReadStreamingWindows(file: File, data: []const []u8) File.Reader.Error!usize {
7956 const t: *Threaded = @ptrCast(@alignCast(userdata));
7957 _ = t;
7958
7959 const DWORD = windows.DWORD;8036 const DWORD = windows.DWORD;
7960 var index: usize = 0;8037 var index: usize = 0;
7961 while (index < data.len and data[index].len == 0) index += 1;8038 while (index < data.len and data[index].len == 0) index += 1;
lib/std/process.zig+19-4
...@@ -466,13 +466,17 @@ pub fn spawnPath(io: Io, dir: Io.Dir, options: SpawnOptions) SpawnError!Child {...@@ -466,13 +466,17 @@ pub fn spawnPath(io: Io, dir: Io.Dir, options: SpawnOptions) SpawnError!Child {
466}466}
467467
468pub const RunError = posix.GetCwdError || posix.ReadError || SpawnError || posix.PollError || error{468pub const RunError = posix.GetCwdError || posix.ReadError || SpawnError || posix.PollError || error{
469 StdoutStreamTooLong,469 StreamTooLong,
470 StderrStreamTooLong,
471};470};
472471
473pub const RunOptions = struct {472pub const RunOptions = struct {
474 argv: []const []const u8,473 argv: []const []const u8,
475 max_output_bytes: usize = 50 * 1024,474 stderr_limit: Io.Limit = .unlimited,
475 stdout_limit: Io.Limit = .unlimited,
476 /// How many bytes to initially allocate for stderr.
477 stderr_reserve_amount: usize = 1,
478 /// How many bytes to initially allocate for stdout.
479 stdout_reserve_amount: usize = 1,
476480
477 /// Set to change the current working directory when spawning the child process.481 /// Set to change the current working directory when spawning the child process.
478 cwd: ?[]const u8 = null,482 cwd: ?[]const u8 = null,
...@@ -498,6 +502,7 @@ pub const RunOptions = struct {...@@ -498,6 +502,7 @@ pub const RunOptions = struct {
498 create_no_window: bool = true,502 create_no_window: bool = true,
499 /// Darwin-only. Disable ASLR for the child process.503 /// Darwin-only. Disable ASLR for the child process.
500 disable_aslr: bool = false,504 disable_aslr: bool = false,
505 timeout: Io.Timeout = .none,
501};506};
502507
503pub const RunResult = struct {508pub const RunResult = struct {
...@@ -530,7 +535,17 @@ pub fn run(gpa: Allocator, io: Io, options: RunOptions) RunError!RunResult {...@@ -530,7 +535,17 @@ pub fn run(gpa: Allocator, io: Io, options: RunOptions) RunError!RunResult {
530 var stderr: std.ArrayList(u8) = .empty;535 var stderr: std.ArrayList(u8) = .empty;
531 defer stderr.deinit(gpa);536 defer stderr.deinit(gpa);
532537
533 try child.collectOutput(gpa, &stdout, &stderr, options.max_output_bytes);538 try stdout.ensureUnusedCapacity(gpa, options.stdout_reserve_amount);
539 try stderr.ensureUnusedCapacity(gpa, options.stderr_reserve_amount);
540
541 try child.collectOutput(io, .{
542 .allocator = gpa,
543 .stdout = &stdout,
544 .stderr = &stderr,
545 .stdout_limit = options.stdout_limit,
546 .stderr_limit = options.stderr_limit,
547 .timeout = options.timeout,
548 });
534549
535 return .{550 return .{
536 .stdout = try stdout.toOwnedSlice(gpa),551 .stdout = try stdout.toOwnedSlice(gpa),
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}