authorgravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2026-01-13 21:23:44-08:00
committergravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2026-01-30 12:10:01-08:00
log5a2cab21e1cf7c0c5dadb5e6c77c359b0b62ff55
treec2874224d4f29a53173277ae4053735e4ec665f6
parent0cca18e43c2685cfb4a72b3b265b68af1e85ca17

std.Io.File.MultiReader: implementation fixes

4 files changed, 87 insertions(+), 80 deletions(-)

lib/std/Build/Step.zig+20-14
...@@ -381,13 +381,15 @@ pub fn addError(step: *Step, comptime fmt: []const u8, args: anytype) error{OutO...@@ -381,13 +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 multi_reader_buffer: Io.File.MultiReader.Buffer(2),
385 multi_reader: Io.File.MultiReader,
384 progress_ipc_fd: if (std.Progress.have_ipc) ?std.posix.fd_t else void,386 progress_ipc_fd: if (std.Progress.have_ipc) ?std.posix.fd_t else void,
385387
386 pub const StreamEnum = enum { stdout, stderr };388 pub const StreamEnum = enum { stdout, stderr };
387389
388 pub fn deinit(zp: *ZigProcess, gpa: Allocator, io: Io) void {390 pub fn deinit(zp: *ZigProcess, io: Io) void {
389 _ = gpa;
390 zp.child.kill(io);391 zp.child.kill(io);
392 zp.multi_reader.deinit();
391 zp.* = undefined;393 zp.* = undefined;
392 }394 }
393};395};
...@@ -460,14 +462,18 @@ pub fn evalZigProcess(...@@ -460,14 +462,18 @@ pub fn evalZigProcess(
460 .request_resource_usage_statistics = true,462 .request_resource_usage_statistics = true,
461 .progress_node = prog_node,463 .progress_node = prog_node,
462 }) catch |err| return s.fail("failed to spawn zig compiler {s}: {t}", .{ argv[0], err });464 }) catch |err| return s.fail("failed to spawn zig compiler {s}: {t}", .{ argv[0], err });
463 defer if (!watch) zp.child.kill(io);
464465
465 zp.* = .{466 zp.* = .{
466 .child = zp.child,467 .child = zp.child,
468 .multi_reader_buffer = undefined,
469 .multi_reader = undefined,
467 .progress_ipc_fd = if (std.Progress.have_ipc) prog_node.getIpcFd() else {},470 .progress_ipc_fd = if (std.Progress.have_ipc) prog_node.getIpcFd() else {},
468 };471 };
472 zp.multi_reader.init(gpa, io, zp.multi_reader_buffer.toStreams(), &.{
473 zp.child.stdout.?, zp.child.stderr.?,
474 });
469 if (watch) s.setZigProcess(zp);475 if (watch) s.setZigProcess(zp);
470 defer if (!watch) zp.deinit(gpa, io);476 defer if (!watch) zp.deinit(io);
471477
472 const result = try zigProcessUpdate(s, zp, watch, web_server, gpa);478 const result = try zigProcessUpdate(s, zp, watch, web_server, gpa);
473479
...@@ -534,18 +540,18 @@ fn zigProcessUpdate(s: *Step, zp: *ZigProcess, watch: bool, web_server: ?*Build....@@ -534,18 +540,18 @@ fn zigProcessUpdate(s: *Step, zp: *ZigProcess, watch: bool, web_server: ?*Build.
534540
535 var result: ?Path = null;541 var result: ?Path = null;
536542
537 var multi_reader_buffer: Io.File.MultiReader.Buffer(2) = undefined;543 const stdout = zp.multi_reader.fileReader(0);
538 var multi_reader: Io.File.MultiReader = undefined;
539 multi_reader.init(gpa, io, multi_reader_buffer.toStreams(), &.{ zp.child.stdout.?, zp.child.stderr.? });
540 defer multi_reader.deinit();
541
542 const stdout = multi_reader.reader(0);
543 const stderr = multi_reader.reader(1);
544544
545 while (true) {545 while (true) {
546 const Header = std.zig.Server.Message.Header;546 const Header = std.zig.Server.Message.Header;
547 const header = try stdout.takeStruct(Header, .little);547 const header = stdout.interface.takeStruct(Header, .little) catch |err| switch (err) {
548 const body = try stdout.take(header.bytes_len);548 error.EndOfStream => break,
549 error.ReadFailed => return stdout.err.?,
550 };
551 const body = stdout.interface.take(header.bytes_len) catch |err| switch (err) {
552 error.EndOfStream => |e| return e,
553 error.ReadFailed => return stdout.err.?,
554 };
549 switch (header.tag) {555 switch (header.tag) {
550 .zig_version => {556 .zig_version => {
551 if (!std.mem.eql(u8, builtin.zig_version_string, body)) {557 if (!std.mem.eql(u8, builtin.zig_version_string, body)) {
...@@ -636,7 +642,7 @@ fn zigProcessUpdate(s: *Step, zp: *ZigProcess, watch: bool, web_server: ?*Build....@@ -636,7 +642,7 @@ fn zigProcessUpdate(s: *Step, zp: *ZigProcess, watch: bool, web_server: ?*Build.
636642
637 s.result_duration_ns = timer.read();643 s.result_duration_ns = timer.read();
638644
639 const stderr_contents = stderr.buffered();645 const stderr_contents = zp.multi_reader.reader(1).buffered();
640 if (stderr_contents.len > 0) {646 if (stderr_contents.len > 0) {
641 try s.result_error_msgs.append(arena, try arena.dupe(u8, stderr_contents));647 try s.result_error_msgs.append(arena, try arena.dupe(u8, stderr_contents));
642 }648 }
lib/std/Build/Step/Run.zig+5-7
...@@ -1695,11 +1695,9 @@ fn evalZigTest(...@@ -1695,11 +1695,9 @@ fn evalZigTest(
1695 // The runner unexpectedly closed a stdio pipe, which means a crash. Make sure we've captured1695 // The runner unexpectedly closed a stdio pipe, which means a crash. Make sure we've captured
1696 // all available stderr to make our error output as useful as possible.1696 // all available stderr to make our error output as useful as possible.
1697 const stderr_fr = multi_reader.fileReader(1);1697 const stderr_fr = multi_reader.fileReader(1);
1698 while (true) {1698 while (stderr_fr.interface.fillMore()) |_| {} else |e| switch (e) {
1699 stderr_fr.interface.fillMore() catch |e| switch (e) {1699 error.ReadFailed => return stderr_fr.err.?,
1700 error.ReadFailed => return stderr_fr.err.?,1700 error.EndOfStream => {},
1701 error.EndOfStream => break,
1702 };
1703 }1701 }
1704 run.step.result_stderr = try arena.dupe(u8, stderr_fr.interface.buffered());1702 run.step.result_stderr = try arena.dupe(u8, stderr_fr.interface.buffered());
17051703
...@@ -1905,7 +1903,7 @@ fn waitZigTest(...@@ -1905,7 +1903,7 @@ fn waitZigTest(
1905 .clock = .awake,1903 .clock = .awake,
1906 } } else .none;1904 } } else .none;
19071905
1908 multi_reader.fill(timeout) catch |err| switch (err) {1906 multi_reader.fill(64, timeout) catch |err| switch (err) {
1909 error.Timeout, error.EndOfStream => return .{ .no_poll = .{1907 error.Timeout, error.EndOfStream => return .{ .no_poll = .{
1910 .active_test_index = active_test_index,1908 .active_test_index = active_test_index,
1911 .ns_elapsed = if (timer) |*t| t.read() else 0,1909 .ns_elapsed = if (timer) |*t| t.read() else 0,
...@@ -2227,7 +2225,7 @@ fn evalGeneric(run: *Run, spawn_options: process.SpawnOptions) !EvalGenericResul...@@ -2227,7 +2225,7 @@ fn evalGeneric(run: *Run, spawn_options: process.SpawnOptions) !EvalGenericResul
2227 const stdout_reader = multi_reader.reader(0);2225 const stdout_reader = multi_reader.reader(0);
2228 const stderr_reader = multi_reader.reader(1);2226 const stderr_reader = multi_reader.reader(1);
22292227
2230 while (multi_reader.fill(.none)) |_| {2228 while (multi_reader.fill(64, .none)) |_| {
2231 if (run.stdio_limit.toInt()) |limit| {2229 if (run.stdio_limit.toInt()) |limit| {
2232 if (stdout_reader.buffered().len > limit)2230 if (stdout_reader.buffered().len > limit)
2233 return error.StdoutStreamTooLong;2231 return error.StdoutStreamTooLong;
lib/std/Io/File/MultiReader.zig+60-58
...@@ -28,18 +28,23 @@ pub const Streams = extern struct {...@@ -28,18 +28,23 @@ pub const Streams = extern struct {
28 len: u32,28 len: u32,
2929
30 pub fn contexts(s: *Streams) []Context {30 pub fn contexts(s: *Streams) []Context {
31 _ = s;31 const base: usize = @intFromPtr(s);
32 @panic("TODO");32 const ptr: [*]Context = @ptrFromInt(std.mem.alignForward(usize, base + @sizeOf(Streams), @alignOf(Context)));
33 return ptr[0..s.len];
33 }34 }
3435
35 pub fn ring(s: *Streams) []u32 {36 pub fn ring(s: *Streams) []u32 {
36 _ = s;37 const prev = contexts(s);
37 @panic("TODO");38 const end = prev.ptr + prev.len;
39 const ptr: [*]u32 = @ptrFromInt(std.mem.alignForward(usize, @intFromPtr(end), @alignOf(u32)));
40 return ptr[0..s.len];
38 }41 }
3942
40 pub fn operations(s: *Streams) []Io.Operation {43 pub fn operations(s: *Streams) []Io.Operation {
41 _ = s;44 const prev = ring(s);
42 @panic("TODO");45 const end = prev.ptr + prev.len;
46 const ptr: [*]Io.Operation = @ptrFromInt(std.mem.alignForward(usize, @intFromPtr(end), @alignOf(Io.Operation)));
47 return ptr[0..s.len];
43 }48 }
44};49};
4550
...@@ -51,6 +56,7 @@ pub fn Buffer(comptime n: usize) type {...@@ -51,6 +56,7 @@ pub fn Buffer(comptime n: usize) type {
51 operations: [n][@sizeOf(Io.Operation)]u8 align(@alignOf(Io.Operation)),56 operations: [n][@sizeOf(Io.Operation)]u8 align(@alignOf(Io.Operation)),
5257
53 pub fn toStreams(b: *@This()) *Streams {58 pub fn toStreams(b: *@This()) *Streams {
59 b.len = n;
54 return @ptrCast(b);60 return @ptrCast(b);
55 }61 }
56 };62 };
...@@ -157,61 +163,43 @@ fn stream(r: *Io.Reader, w: *Io.Writer, limit: Io.Limit) Io.Reader.StreamError!u...@@ -157,61 +163,43 @@ fn stream(r: *Io.Reader, w: *Io.Writer, limit: Io.Limit) Io.Reader.StreamError!u
157 _ = w;163 _ = w;
158 const fr: *File.Reader = @alignCast(@fieldParentPtr("interface", r));164 const fr: *File.Reader = @alignCast(@fieldParentPtr("interface", r));
159 const context: *Context = @fieldParentPtr("fr", fr);165 const context: *Context = @fieldParentPtr("fr", fr);
160 const mr = context.mr;166 try fillUntimed(context, 1);
161 return fillUntimed(mr, context);167 return 0;
162}168}
163169
164fn discard(r: *Io.Reader, limit: Io.Limit) Io.Reader.Error!usize {170fn discard(r: *Io.Reader, limit: Io.Limit) Io.Reader.Error!usize {
165 _ = limit;171 _ = limit;
166 const fr: *File.Reader = @alignCast(@fieldParentPtr("interface", r));172 const fr: *File.Reader = @alignCast(@fieldParentPtr("interface", r));
167 const context: *Context = @fieldParentPtr("fr", fr);173 const context: *Context = @fieldParentPtr("fr", fr);
168 const mr = context.mr;174 try fillUntimed(context, 1);
169 return fillUntimed(mr, context);175 return 0;
170}176}
171177
172fn readVec(r: *Io.Reader, data: [][]u8) Io.Reader.Error!usize {178fn readVec(r: *Io.Reader, data: [][]u8) Io.Reader.Error!usize {
173 _ = data;179 _ = data;
174 const fr: *File.Reader = @alignCast(@fieldParentPtr("interface", r));180 const fr: *File.Reader = @alignCast(@fieldParentPtr("interface", r));
175 const context: *Context = @fieldParentPtr("fr", fr);181 const context: *Context = @fieldParentPtr("fr", fr);
176 const mr = context.mr;182 try fillUntimed(context, 1);
177 return fillUntimed(mr, context);183 return 0;
178}184}
179185
180fn rebase(r: *Io.Reader, capacity: usize) Io.Reader.RebaseError!void {186fn rebase(r: *Io.Reader, capacity: usize) Io.Reader.RebaseError!void {
181 const fr: *File.Reader = @alignCast(@fieldParentPtr("interface", r));187 const fr: *File.Reader = @alignCast(@fieldParentPtr("interface", r));
182 const context: *Context = @fieldParentPtr("fr", fr);188 const context: *Context = @fieldParentPtr("fr", fr);
183 const mr = context.mr;189 try fillUntimed(context, capacity);
184
185 return rebaseGrowing(mr, context, capacity) catch |err| {
186 context.err = err;
187 return error.ReadFailed;
188 };
189}190}
190191
191fn rebaseGrowing(mr: *MultiReader, context: *Context, capacity: usize) Allocator.Error!void {192fn fillUntimed(context: *Context, capacity: usize) Io.Reader.Error!void {
192 const gpa = mr.gpa;193 fill(context.mr, capacity, .none) catch |err| switch (err) {
193 const r = &context.fr.interface;194 error.Timeout, error.UnsupportedClock => unreachable,
194 if (r.buffer.len >= capacity) {195 error.Canceled, error.ConcurrencyUnavailable => |e| {
195 const data = r.buffer[r.seek..r.end];196 context.err = e;
196 @memmove(r.buffer[0..data.len], data);197 return error.ReadFailed;
197 r.seek = 0;198 },
198 r.end = data.len;199 error.EndOfStream => |e| return e,
199 } else {200 };
200 const adjusted_capacity = std.ArrayList(u8).growCapacity(capacity);201 if (context.err != null) return error.ReadFailed;
201202 if (context.eos) return error.EndOfStream;
202 if (r.seek == 0) {
203 if (gpa.remap(r.buffer, adjusted_capacity)) |new_memory| {
204 r.buffer = new_memory;
205 return;
206 }
207 }
208
209 const data = r.buffer[r.seek..r.end];
210 const new = try gpa.alloc(u8, adjusted_capacity);
211 @memcpy(new[0..data.len], data);
212 r.seek = 0;
213 r.end = data.len;
214 }
215}203}
216204
217pub const FillError = Io.Batch.WaitError || error{205pub const FillError = Io.Batch.WaitError || error{
...@@ -221,7 +209,7 @@ pub const FillError = Io.Batch.WaitError || error{...@@ -221,7 +209,7 @@ pub const FillError = Io.Batch.WaitError || error{
221};209};
222210
223/// Wait until at least one stream receives more data.211/// Wait until at least one stream receives more data.
224pub fn fill(mr: *MultiReader, timeout: Io.Timeout) FillError!void {212pub fn fill(mr: *MultiReader, unused_capacity: usize, timeout: Io.Timeout) FillError!void {
225 const contexts = mr.streams.contexts();213 const contexts = mr.streams.contexts();
226 const operations = mr.streams.operations();214 const operations = mr.streams.operations();
227 const io = contexts[0].fr.io;215 const io = contexts[0].fr.io;
...@@ -243,14 +231,14 @@ pub fn fill(mr: *MultiReader, timeout: Io.Timeout) FillError!void {...@@ -243,14 +231,14 @@ pub fn fill(mr: *MultiReader, timeout: Io.Timeout) FillError!void {
243 }231 }
244 const r = &context.fr.interface;232 const r = &context.fr.interface;
245 r.end += n;233 r.end += n;
246 if (r.buffer.len - r.end == 0) {234 if (r.buffer.len - r.end < unused_capacity) {
247 rebaseGrowing(mr, context, r.bufferedLen() + 1) catch |err| {235 rebaseGrowing(mr, context, r.bufferedLen() + unused_capacity) catch |err| {
248 context.err = err;236 context.err = err;
249 continue;237 continue;
250 };238 };
251 assert(r.seek == 0);239 assert(r.seek == 0);
252 context.vec[0] = r.buffer;
253 }240 }
241 context.vec[0] = r.buffer[r.end..];
254 operation.file_read_streaming.status = .{ .unstarted = {} };242 operation.file_read_streaming.status = .{ .unstarted = {} };
255 mr.batch.add(i);243 mr.batch.add(i);
256 }244 }
...@@ -258,16 +246,30 @@ pub fn fill(mr: *MultiReader, timeout: Io.Timeout) FillError!void {...@@ -258,16 +246,30 @@ pub fn fill(mr: *MultiReader, timeout: Io.Timeout) FillError!void {
258 if (!any_completed) return error.EndOfStream;246 if (!any_completed) return error.EndOfStream;
259}247}
260248
261fn fillUntimed(mr: *MultiReader, context: *Context) Io.Reader.Error!usize {249fn rebaseGrowing(mr: *MultiReader, context: *Context, capacity: usize) Allocator.Error!void {
262 fill(mr, .none) catch |err| switch (err) {250 const gpa = mr.gpa;
263 error.Timeout, error.UnsupportedClock => unreachable,251 const r = &context.fr.interface;
264 error.Canceled, error.ConcurrencyUnavailable => |e| {252 if (r.buffer.len >= capacity) {
265 context.err = e;253 const data = r.buffer[r.seek..r.end];
266 return error.ReadFailed;254 @memmove(r.buffer[0..data.len], data);
267 },255 r.seek = 0;
268 error.EndOfStream => |e| return e,256 r.end = data.len;
269 };257 } else {
270 if (context.err != null) return error.ReadFailed;258 const adjusted_capacity = std.ArrayList(u8).growCapacity(capacity);
271 if (context.eos) return error.EndOfStream;259
272 return 0;260 if (r.seek == 0) {
261 if (gpa.remap(r.buffer, adjusted_capacity)) |new_memory| {
262 r.buffer = new_memory;
263 return;
264 }
265 }
266
267 const data = r.buffer[r.seek..r.end];
268 const new = try gpa.alloc(u8, adjusted_capacity);
269 @memcpy(new[0..data.len], data);
270 gpa.free(r.buffer);
271 r.buffer = new;
272 r.seek = 0;
273 r.end = data.len;
274 }
273}275}
lib/std/crypto/tls/Client.zig+2-1
...@@ -336,10 +336,11 @@ pub fn init(input: *Reader, output: *Writer, options: Options) InitError!Client...@@ -336,10 +336,11 @@ pub fn init(input: *Reader, output: *Writer, options: Options) InitError!Client
336 // Ensure the input buffer pointer is stable in this scope.336 // Ensure the input buffer pointer is stable in this scope.
337 input.rebase(tls.max_ciphertext_record_len) catch |err| switch (err) {337 input.rebase(tls.max_ciphertext_record_len) catch |err| switch (err) {
338 error.EndOfStream => {}, // We have assurance the remainder of stream can be buffered.338 error.EndOfStream => {}, // We have assurance the remainder of stream can be buffered.
339 error.ReadFailed => |e| return e,
339 };340 };
340 const record_header = input.peek(tls.record_header_len) catch |err| switch (err) {341 const record_header = input.peek(tls.record_header_len) catch |err| switch (err) {
341 error.EndOfStream => return error.TlsConnectionTruncated,342 error.EndOfStream => return error.TlsConnectionTruncated,
342 error.ReadFailed => return error.ReadFailed,343 error.ReadFailed => |e| return e,
343 };344 };
344 const record_ct = input.takeEnumNonexhaustive(tls.ContentType, .big) catch unreachable; // already peeked345 const record_ct = input.takeEnumNonexhaustive(tls.ContentType, .big) catch unreachable; // already peeked
345 input.toss(2); // legacy_version346 input.toss(2); // legacy_version