authorgravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2026-01-08 16:55:26-08:00
committergravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2026-01-09 20:47:25-08:00
loge1f7844616b8613dbacfbcc4fa7fff8a3d0d0e82
treef3bbd20c6ed89993ca22a8ed0f81a3547622c0d2
parentf8e1257b62b1ea5fac700e55b2597a38af82f75a

std.process.Child: rewrite using concurrent

I plan to immediately revert this, but here's a commit for posterity

2 files changed, 26 insertions(+), 56 deletions(-)

lib/std/process.zig+1
...@@ -467,6 +467,7 @@ pub fn spawnPath(io: Io, dir: Io.Dir, options: SpawnOptions) SpawnError!Child {...@@ -467,6 +467,7 @@ pub fn spawnPath(io: Io, dir: Io.Dir, options: SpawnOptions) SpawnError!Child {
467467
468pub const RunError = posix.GetCwdError || posix.ReadError || SpawnError || posix.PollError || error{468pub const RunError = posix.GetCwdError || posix.ReadError || SpawnError || posix.PollError || error{
469 StreamTooLong,469 StreamTooLong,
470 ConcurrencyUnavailable,
470};471};
471472
472pub const RunOptions = struct {473pub const RunOptions = struct {
lib/std/process/Child.zig+25-56
...@@ -125,14 +125,15 @@ pub fn wait(child: *Child, io: Io) WaitError!Term {...@@ -125,14 +125,15 @@ pub fn wait(child: *Child, io: Io) WaitError!Term {
125 return io.vtable.childWait(io.userdata, child);125 return io.vtable.childWait(io.userdata, child);
126}126}
127127
128pub const CollectOutputError = error{StreamTooLong} || Allocator.Error || Io.File.Reader.Error;128pub const CollectOutputError = error{
129 StreamTooLong,
130 ConcurrencyUnavailable,
131} || Allocator.Error || Io.File.Reader.Error;
129132
130pub const CollectOutputOptions = struct {133pub const CollectOutputOptions = struct {
131 stdout: *std.ArrayList(u8),134 stdout: *std.ArrayList(u8),
132 stderr: *std.ArrayList(u8),135 stderr: *std.ArrayList(u8),
133 /// Used for `stdout` and `stderr`. If not provided, only the existing136 allocator: Allocator,
134 /// capacity will be used.
135 allocator: ?Allocator = null,
136 stdout_limit: Io.Limit = .unlimited,137 stdout_limit: Io.Limit = .unlimited,
137 stderr_limit: Io.Limit = .unlimited,138 stderr_limit: Io.Limit = .unlimited,
138};139};
...@@ -144,56 +145,24 @@ pub const CollectOutputOptions = struct {...@@ -144,56 +145,24 @@ pub const CollectOutputOptions = struct {
144/// The process must have been started with stdout and stderr set to145/// The process must have been started with stdout and stderr set to
145/// `process.SpawnOptions.StdIo.pipe`.146/// `process.SpawnOptions.StdIo.pipe`.
146pub fn collectOutput(child: *const Child, io: Io, options: CollectOutputOptions) CollectOutputError!void {147pub fn collectOutput(child: *const Child, io: Io, options: CollectOutputOptions) CollectOutputError!void {
147 const files: [2]Io.File = .{ child.stdout.?, child.stderr.? };148 var stdout = try io.concurrent(collectStream, .{
148 const lists: [2]*std.ArrayList(u8) = .{ options.stdout, options.stderr };149 io, options.allocator, child.stdout.?, options.stdout, options.stdout_limit,
149 const limits: [2]Io.Limit = .{ options.stdout_limit, options.stderr_limit };150 });
150 var dones: [2]bool = .{ false, false };151 defer stdout.cancel(io) catch {};
151 var reads: [2]Io.Operation = undefined;152
152 var vecs: [2][1][]u8 = undefined;153 var stderr = try io.concurrent(collectStream, .{
153 while (true) {154 io, options.allocator, child.stderr.?, options.stderr, options.stderr_limit,
154 for (&reads, &lists, &files, dones, &vecs) |*read, list, file, done, *vec| {155 });
155 if (done) {156 defer stderr.cancel(io) catch {};
156 read.* = .noop;157
157 continue;158 try stdout.await(io);
158 }159 try stderr.await(io);
159 if (options.allocator) |gpa| try list.ensureUnusedCapacity(gpa, 1);160}
160 const cap = list.unusedCapacitySlice();161
161 if (cap.len == 0) return error.StreamTooLong;162fn collectStream(io: Io, gpa: Allocator, file: File, list: *std.ArrayList(u8), limit: Io.Limit) CollectOutputError!void {
162 vec[0] = cap;163 var fr = file.readerStreaming(io, &.{});
163 read.* = .{ .file_read_streaming = .{164 fr.interface.appendRemaining(gpa, list, limit) catch |err| switch (err) {
164 .file = file,165 error.ReadFailed => return fr.err.?,
165 .data = vec,166 else => |e| return e,
166 .nonblocking = true,167 };
167 .result = undefined,
168 } };
169 }
170 var all_done = true;
171 var any_canceled = false;
172 var other_err: (error{StreamTooLong} || Io.File.Reader.Error)!void = {};
173 io.vtable.operate(io.userdata, &reads);
174 for (&reads, &lists, &limits, &dones) |*read, list, limit, *done| {
175 if (done.*) continue;
176 const n = read.file_read_streaming.result catch |err| switch (err) {
177 error.Canceled => {
178 any_canceled = true;
179 continue;
180 },
181 error.WouldBlock => continue,
182 else => |e| {
183 other_err = e;
184 continue;
185 },
186 };
187 if (n == 0) {
188 done.* = true;
189 } else {
190 all_done = false;
191 }
192 list.items.len += n;
193 if (list.items.len > @intFromEnum(limit)) other_err = error.StreamTooLong;
194 }
195 if (any_canceled) return error.Canceled;
196 try other_err;
197 if (all_done) return;
198 }
199}168}