authorgravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2026-01-12 23:21:55-08:00
committergravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2026-01-30 12:10:01-08:00
log1a168f08b51572ab136e2a8fcda35ce3c68c89bb
treea4b70effc0e3ebc49ade76209f86f5546c136ab0
parenta1b3d9e447106a685857c6b4e51c96bde2519fb7

std.Io.File: introduce MultiReader

Concurrently read from multiple file streams, eliminating risk of deadlocking.

5 files changed, 256 insertions(+), 28 deletions(-)

lib/std/Build/Step.zig+8-21
...@@ -527,9 +527,6 @@ fn zigProcessUpdate(s: *Step, zp: *ZigProcess, watch: bool, web_server: ?*Build....@@ -527,9 +527,6 @@ fn zigProcessUpdate(s: *Step, zp: *ZigProcess, watch: bool, web_server: ?*Build.
527 const arena = b.allocator;527 const arena = b.allocator;
528 const io = b.graph.io;528 const io = b.graph.io;
529529
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
533 var timer = try std.time.Timer.start();530 var timer = try std.time.Timer.start();
534531
535 try sendMessage(io, zp.child.stdin.?, .update);532 try sendMessage(io, zp.child.stdin.?, .update);
...@@ -537,19 +534,18 @@ fn zigProcessUpdate(s: *Step, zp: *ZigProcess, watch: bool, web_server: ?*Build....@@ -537,19 +534,18 @@ fn zigProcessUpdate(s: *Step, zp: *ZigProcess, watch: bool, web_server: ?*Build.
537534
538 var result: ?Path = null;535 var result: ?Path = null;
539536
540 var stdout_buffer: [512]u8 = undefined;537 var multi_reader_buffer: Io.File.MultiReader.Buffer(2) = undefined;
541 var stdout_reader: Io.File.Reader = .initStreaming(zp.child.stdout.?, io, &stdout_buffer);538 var multi_reader: Io.File.MultiReader = undefined;
542 const stdout = &stdout_reader.interface;539 multi_reader.init(gpa, io, multi_reader_buffer.toStreams(), &.{ zp.child.stdout.?, zp.child.stderr.? });
540 defer multi_reader.deinit();
543541
544 var body_buffer: std.ArrayList(u8) = .empty;542 const stdout = multi_reader.reader(0);
545 defer body_buffer.deinit(gpa);543 const stderr = multi_reader.reader(1);
546544
547 while (true) {545 while (true) {
548 const Header = std.zig.Server.Message.Header;546 const Header = std.zig.Server.Message.Header;
549 const header = try stdout.takeStruct(Header, .little);547 const header = try stdout.takeStruct(Header, .little);
550 body_buffer.clearRetainingCapacity();548 const body = try stdout.take(header.bytes_len);
551 try stdout.appendExact(gpa, &body_buffer, header.bytes_len);
552 const body = body_buffer.items;
553 switch (header.tag) {549 switch (header.tag) {
554 .zig_version => {550 .zig_version => {
555 if (!std.mem.eql(u8, builtin.zig_version_string, body)) {551 if (!std.mem.eql(u8, builtin.zig_version_string, body)) {
...@@ -640,8 +636,7 @@ fn zigProcessUpdate(s: *Step, zp: *ZigProcess, watch: bool, web_server: ?*Build....@@ -640,8 +636,7 @@ fn zigProcessUpdate(s: *Step, zp: *ZigProcess, watch: bool, web_server: ?*Build.
640636
641 s.result_duration_ns = timer.read();637 s.result_duration_ns = timer.read();
642638
643 const stderr_contents = try stderr_task.await(io);639 const stderr_contents = stderr.buffered();
644 defer gpa.free(stderr_contents);
645 if (stderr_contents.len > 0) {640 if (stderr_contents.len > 0) {
646 try s.result_error_msgs.append(arena, try arena.dupe(u8, stderr_contents));641 try s.result_error_msgs.append(arena, try arena.dupe(u8, stderr_contents));
647 }642 }
...@@ -649,14 +644,6 @@ fn zigProcessUpdate(s: *Step, zp: *ZigProcess, watch: bool, web_server: ?*Build....@@ -649,14 +644,6 @@ fn zigProcessUpdate(s: *Step, zp: *ZigProcess, watch: bool, web_server: ?*Build.
649 return result;644 return result;
650}645}
651646
652fn readStreamAlloc(gpa: Allocator, io: Io, file: Io.File, limit: Io.Limit) ![]u8 {
653 var file_reader: Io.File.Reader = .initStreaming(file, io, &.{});
654 return file_reader.interface.allocRemaining(gpa, limit) catch |err| switch (err) {
655 error.ReadFailed => return file_reader.err.?,
656 else => |e| return e,
657 };
658}
659
660pub fn getZigProcess(s: *Step) ?*ZigProcess {647pub fn getZigProcess(s: *Step) ?*ZigProcess {
661 return switch (s.id) {648 return switch (s.id) {
662 .compile => s.cast(Compile).?.zig_process,649 .compile => s.cast(Compile).?.zig_process,
lib/std/Io.zig+2-2
...@@ -350,8 +350,6 @@ pub const Batch = struct {...@@ -350,8 +350,6 @@ pub const Batch = struct {
350 }350 }
351 };351 };
352352
353 pub const WaitError = ConcurrentError || Cancelable || Timeout.Error;
354
355 pub fn init(operations: []Operation, ring: []u32) Batch {353 pub fn init(operations: []Operation, ring: []u32) Batch {
356 const len: u31 = @intCast(operations.len);354 const len: u31 = @intCast(operations.len);
357 assert(ring.len == len);355 assert(ring.len == len);
...@@ -405,6 +403,8 @@ pub const Batch = struct {...@@ -405,6 +403,8 @@ pub const Batch = struct {
405 return b.ring[0..len][head.index(len)];403 return b.ring[0..len][head.index(len)];
406 }404 }
407405
406 pub const WaitError = ConcurrentError || Cancelable || Timeout.Error;
407
408 /// Starts work on any submitted operations and returns when at least one has completeed.408 /// Starts work on any submitted operations and returns when at least one has completeed.
409 ///409 ///
410 /// Returns `error.Timeout` if `timeout` expires first.410 /// Returns `error.Timeout` if `timeout` expires first.
lib/std/Io/File.zig+3
...@@ -18,6 +18,9 @@ pub const Writer = @import("File/Writer.zig");...@@ -18,6 +18,9 @@ pub const Writer = @import("File/Writer.zig");
18pub const Atomic = @import("File/Atomic.zig");18pub const Atomic = @import("File/Atomic.zig");
19/// Memory intended to remain consistent with file contents.19/// Memory intended to remain consistent with file contents.
20pub const MemoryMap = @import("File/MemoryMap.zig");20pub const MemoryMap = @import("File/MemoryMap.zig");
21/// Concurrently read from multiple file streams, eliminating risk of
22/// deadlocking.
23pub const MultiReader = @import("File/MultiReader.zig");
2124
22pub const INode = std.posix.ino_t;25pub const INode = std.posix.ino_t;
23pub const NLink = std.posix.nlink_t;26pub const NLink = std.posix.nlink_t;
lib/std/Io/File/MultiReader.zig created+240
...@@ -0,0 +1,240 @@
1const MultiReader = @This();
2
3const std = @import("../../std.zig");
4const Io = std.Io;
5const File = Io.File;
6const Allocator = std.mem.Allocator;
7const assert = std.debug.assert;
8
9gpa: Allocator,
10streams: *Streams,
11batch: Io.Batch,
12
13pub const Context = struct {
14 mr: *MultiReader,
15 fr: File.Reader,
16 vec: [1][]u8,
17 err: ?Error,
18 eos: bool,
19};
20
21pub const Error = Allocator.Error || File.Reader.Error || Io.ConcurrentError;
22
23/// Trailing:
24/// * `contexts: [len]Context`
25/// * `ring: [len]u32`
26/// * `operations: [len]Io.Operation`
27pub const Streams = extern struct {
28 len: u32,
29
30 pub fn contexts(s: *Streams) []Context {
31 _ = s;
32 @panic("TODO");
33 }
34
35 pub fn ring(s: *Streams) []u32 {
36 _ = s;
37 @panic("TODO");
38 }
39
40 pub fn operations(s: *Streams) []Io.Operation {
41 _ = s;
42 @panic("TODO");
43 }
44};
45
46pub fn Buffer(comptime n: usize) type {
47 return extern struct {
48 len: u32,
49 contexts: [n][@sizeOf(Context)]u8 align(@alignOf(Context)),
50 ring: [n]u32,
51 operations: [n][@sizeOf(Io.Operation)]u8 align(@alignOf(Io.Operation)),
52
53 pub fn toStreams(b: *@This()) *Streams {
54 return @ptrCast(b);
55 }
56 };
57}
58
59/// See `Streams.Buffer` for convenience API to obtain the `streams` parameter.
60pub fn init(mr: *MultiReader, gpa: Allocator, io: Io, streams: *Streams, files: []const File) void {
61 const contexts = streams.contexts();
62 for (contexts, files) |*context, file| context.* = .{
63 .mr = mr,
64 .fr = .{
65 .io = io,
66 .file = file,
67 .mode = .streaming,
68 .interface = .{
69 .vtable = &.{
70 .stream = stream,
71 .discard = discard,
72 .readVec = readVec,
73 .rebase = rebase,
74 },
75 .buffer = &.{},
76 .seek = 0,
77 .end = 0,
78 },
79 },
80 .vec = .{&.{}},
81 .err = null,
82 .eos = false,
83 };
84 const operations = streams.operations();
85 const ring = streams.ring();
86 mr.* = .{
87 .gpa = gpa,
88 .streams = streams,
89 .batch = .init(operations, ring),
90 };
91 for (operations, contexts, files, 0..) |*op, *context, file, i| {
92 const r = &context.fr.interface;
93 op.* = .{ .file_read_streaming = .{
94 .file = file,
95 .data = &context.vec,
96 } };
97 rebaseGrowing(mr, context, 1) catch |err| {
98 context.err = err;
99 continue;
100 };
101 context.vec[0] = r.buffer;
102 mr.batch.add(i);
103 }
104}
105
106pub fn deinit(mr: *MultiReader) void {
107 const gpa = mr.gpa;
108 const contexts = mr.streams.contexts();
109 const io = contexts[0].fr.io;
110 mr.batch.cancel(io);
111 for (contexts) |*context| {
112 gpa.free(context.fr.interface.buffer);
113 }
114}
115
116pub fn reader(mr: *MultiReader, index: usize) *Io.Reader {
117 return &mr.streams.contexts()[index].fr.interface;
118}
119
120pub fn toOwnedSlice(mr: *MultiReader, index: usize) Allocator.Error![]u8 {
121 const gpa = mr.gpa;
122 const r: *Io.Reader = reader(mr, index);
123 if (r.seek == 0) {
124 const new = try gpa.realloc(r.buffer, r.end);
125 r.buffer = &.{};
126 r.end = 0;
127 return new;
128 }
129 const new = try gpa.dupe(u8, r.buffered());
130 gpa.free(r.buffer);
131 r.buffer = &.{};
132 r.seek = 0;
133 r.end = 0;
134 return new;
135}
136
137fn stream(r: *Io.Reader, w: *Io.Writer, limit: Io.Limit) Io.Reader.StreamError!usize {
138 _ = limit;
139 _ = w;
140 const fr: *File.Reader = @alignCast(@fieldParentPtr("interface", r));
141 const context: *Context = @fieldParentPtr("fr", fr);
142 const mr = context.mr;
143 return fill(mr, context);
144}
145
146fn discard(r: *Io.Reader, limit: Io.Limit) Io.Reader.Error!usize {
147 _ = limit;
148 const fr: *File.Reader = @alignCast(@fieldParentPtr("interface", r));
149 const context: *Context = @fieldParentPtr("fr", fr);
150 const mr = context.mr;
151 return fill(mr, context);
152}
153
154fn readVec(r: *Io.Reader, data: [][]u8) Io.Reader.Error!usize {
155 _ = data;
156 const fr: *File.Reader = @alignCast(@fieldParentPtr("interface", r));
157 const context: *Context = @fieldParentPtr("fr", fr);
158 const mr = context.mr;
159 return fill(mr, context);
160}
161
162fn rebase(r: *Io.Reader, capacity: usize) Io.Reader.RebaseError!void {
163 const fr: *File.Reader = @alignCast(@fieldParentPtr("interface", r));
164 const context: *Context = @fieldParentPtr("fr", fr);
165 const mr = context.mr;
166
167 return rebaseGrowing(mr, context, capacity) catch |err| {
168 context.err = err;
169 return error.ReadFailed;
170 };
171}
172
173fn rebaseGrowing(mr: *MultiReader, context: *Context, capacity: usize) Allocator.Error!void {
174 const gpa = mr.gpa;
175 const r = &context.fr.interface;
176 if (r.buffer.len >= capacity) {
177 const data = r.buffer[r.seek..r.end];
178 @memmove(r.buffer[0..data.len], data);
179 r.seek = 0;
180 r.end = data.len;
181 } else {
182 const adjusted_capacity = std.ArrayList(u8).growCapacity(capacity);
183
184 if (r.seek == 0) {
185 if (gpa.remap(r.buffer, adjusted_capacity)) |new_memory| {
186 r.buffer = new_memory;
187 return;
188 }
189 }
190
191 const data = r.buffer[r.seek..r.end];
192 const new = try gpa.alloc(u8, adjusted_capacity);
193 @memcpy(new[0..data.len], data);
194 r.seek = 0;
195 r.end = data.len;
196 }
197}
198
199fn fill(mr: *MultiReader, original_context: *Context) Io.Reader.Error!usize {
200 const contexts = mr.streams.contexts();
201 const operations = mr.streams.operations();
202 const io = contexts[0].fr.io;
203
204 mr.batch.wait(io, .none) catch |err| switch (err) {
205 error.Timeout, error.UnsupportedClock => unreachable,
206 else => |e| {
207 original_context.err = e;
208 return error.ReadFailed;
209 },
210 };
211
212 while (mr.batch.next()) |i| {
213 const context = &contexts[i];
214 const operation = &operations[i];
215 const n = operation.file_read_streaming.status.result catch |err| {
216 context.err = err;
217 continue;
218 };
219 if (n == 0) {
220 context.eos = true;
221 continue;
222 }
223 const r = &context.fr.interface;
224 r.end += n;
225 if (r.buffer.len - r.end == 0) {
226 rebaseGrowing(mr, context, r.bufferedLen() + 1) catch |err| {
227 context.err = err;
228 continue;
229 };
230 assert(r.seek == 0);
231 context.vec[0] = r.buffer;
232 }
233 operation.file_read_streaming.status = .{ .unstarted = {} };
234 mr.batch.add(i);
235 }
236
237 if (original_context.err != null) return error.ReadFailed;
238 if (original_context.eos) return error.EndOfStream;
239 return 0;
240}
lib/std/Io/Reader.zig+3-5
...@@ -127,9 +127,7 @@ pub const ShortError = error{...@@ -127,9 +127,7 @@ pub const ShortError = error{
127 ReadFailed,127 ReadFailed,
128};128};
129129
130pub const RebaseError = error{130pub const RebaseError = Error;
131 EndOfStream,
132};
133131
134pub const failing: Reader = .{132pub const failing: Reader = .{
135 .vtable = &.{133 .vtable = &.{
...@@ -1402,7 +1400,7 @@ pub fn takeLeb128(r: *Reader, comptime T: type) TakeLeb128Error!T {...@@ -1402,7 +1400,7 @@ pub fn takeLeb128(r: *Reader, comptime T: type) TakeLeb128Error!T {
1402}1400}
14031401
1404/// Ensures `capacity` data can be buffered without rebasing.1402/// Ensures `capacity` data can be buffered without rebasing.
1405pub fn rebase(r: *Reader, capacity: usize) RebaseError!void {1403pub fn rebase(r: *Reader, capacity: usize) Error!void {
1406 if (r.buffer.len - r.seek >= capacity) {1404 if (r.buffer.len - r.seek >= capacity) {
1407 @branchHint(.likely);1405 @branchHint(.likely);
1408 return;1406 return;
...@@ -1410,7 +1408,7 @@ pub fn rebase(r: *Reader, capacity: usize) RebaseError!void {...@@ -1410,7 +1408,7 @@ pub fn rebase(r: *Reader, capacity: usize) RebaseError!void {
1410 return r.vtable.rebase(r, capacity);1408 return r.vtable.rebase(r, capacity);
1411}1409}
14121410
1413pub fn defaultRebase(r: *Reader, capacity: usize) RebaseError!void {1411pub fn defaultRebase(r: *Reader, capacity: usize) Error!void {
1414 assert(r.buffer.len - r.seek < capacity);1412 assert(r.buffer.len - r.seek < capacity);
1415 const data = r.buffer[r.seek..r.end];1413 const data = r.buffer[r.seek..r.end];
1416 @memmove(r.buffer[0..data.len], data);1414 @memmove(r.buffer[0..data.len], data);