| ... | @@ -125,15 +125,14 @@ pub fn wait(child: *Child, io: Io) WaitError!Term { | ... | @@ -125,15 +125,14 @@ 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 | } |
| 127 | | 127 | |
| 128 | pub const CollectOutputError = error{ | 128 | pub const CollectOutputError = error{StreamTooLong} || Allocator.Error || Io.File.Reader.Error; |
| 129 | StreamTooLong, | | |
| 130 | ConcurrencyUnavailable, | | |
| 131 | } || Allocator.Error || Io.File.Reader.Error; | | |
| 132 | | 129 | |
| 133 | pub const CollectOutputOptions = struct { | 130 | pub const CollectOutputOptions = struct { |
| 134 | stdout: *std.ArrayList(u8), | 131 | stdout: *std.ArrayList(u8), |
| 135 | stderr: *std.ArrayList(u8), | 132 | stderr: *std.ArrayList(u8), |
| 136 | allocator: Allocator, | 133 | /// Used for `stdout` and `stderr`. If not provided, only the existing |
| | 134 | /// capacity will be used. |
| | 135 | allocator: ?Allocator = null, |
| 137 | stdout_limit: Io.Limit = .unlimited, | 136 | stdout_limit: Io.Limit = .unlimited, |
| 138 | stderr_limit: Io.Limit = .unlimited, | 137 | stderr_limit: Io.Limit = .unlimited, |
| 139 | }; | 138 | }; |
| ... | @@ -145,24 +144,56 @@ pub const CollectOutputOptions = struct { | ... | @@ -145,24 +144,56 @@ pub const CollectOutputOptions = struct { |
| 145 | /// The process must have been started with stdout and stderr set to | 144 | /// The process must have been started with stdout and stderr set to |
| 146 | /// `process.SpawnOptions.StdIo.pipe`. | 145 | /// `process.SpawnOptions.StdIo.pipe`. |
| 147 | pub fn collectOutput(child: *const Child, io: Io, options: CollectOutputOptions) CollectOutputError!void { | 146 | pub fn collectOutput(child: *const Child, io: Io, options: CollectOutputOptions) CollectOutputError!void { |
| 148 | var stdout = try io.concurrent(collectStream, .{ | 147 | const files: [2]Io.File = .{ child.stdout.?, child.stderr.? }; |
| 149 | io, options.allocator, child.stdout.?, options.stdout, options.stdout_limit, | 148 | const lists: [2]*std.ArrayList(u8) = .{ options.stdout, options.stderr }; |
| 150 | }); | 149 | const limits: [2]Io.Limit = .{ options.stdout_limit, options.stderr_limit }; |
| 151 | defer stdout.cancel(io) catch {}; | 150 | var dones: [2]bool = .{ false, false }; |
| 152 | | 151 | var reads: [2]Io.Operation = undefined; |
| 153 | var stderr = try io.concurrent(collectStream, .{ | 152 | var vecs: [2][1][]u8 = undefined; |
| 154 | io, options.allocator, child.stderr.?, options.stderr, options.stderr_limit, | 153 | while (true) { |
| 155 | }); | 154 | for (&reads, &lists, &files, dones, &vecs) |*read, list, file, done, *vec| { |
| 156 | defer stderr.cancel(io) catch {}; | 155 | if (done) { |
| 157 | | 156 | read.* = .noop; |
| 158 | try stdout.await(io); | 157 | continue; |
| 159 | try stderr.await(io); | 158 | } |
| 160 | } | 159 | if (options.allocator) |gpa| try list.ensureUnusedCapacity(gpa, 1); |
| 161 | | 160 | const cap = list.unusedCapacitySlice(); |
| 162 | fn collectStream(io: Io, gpa: Allocator, file: File, list: *std.ArrayList(u8), limit: Io.Limit) CollectOutputError!void { | 161 | if (cap.len == 0) return error.StreamTooLong; |
| 163 | var fr = file.readerStreaming(io, &.{}); | 162 | vec[0] = cap; |
| 164 | fr.interface.appendRemaining(gpa, list, limit) catch |err| switch (err) { | 163 | read.* = .{ .file_read_streaming = .{ |
| 165 | error.ReadFailed => return fr.err.?, | 164 | .file = file, |
| 166 | else => |e| return e, | 165 | .data = vec, |
| 167 | }; | 166 | .nonblocking = true, |
| | 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 | } |
| 168 | } | 199 | } |
Andrew Kelley <andrew@ziglang.org> 2026-01-08 16:55:51-08:00
lib