authorgravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2025-07-22 21:21:27-07:00
committergravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2025-07-23 21:25:34-07:00
logb8955a2e0aea692b6c76dc39962b3620649b6e1f
tree6c3e465eb83a9cceefdfd126ff932056cb4dfefc
parentbc8e1a74c514e487842c03ab32d7f6e49f42c529

std.Io.poll: update to new I/O API

9 files changed, 387 insertions(+), 380 deletions(-)

lib/std/Build/Fuzz/WebServer.zig+9-17
...@@ -273,21 +273,17 @@ fn buildWasmBinary(...@@ -273,21 +273,17 @@ fn buildWasmBinary(
273 try sendMessage(child.stdin.?, .update);273 try sendMessage(child.stdin.?, .update);
274 try sendMessage(child.stdin.?, .exit);274 try sendMessage(child.stdin.?, .exit);
275275
276 const Header = std.zig.Server.Message.Header;
277 var result: ?Path = null;276 var result: ?Path = null;
278 var result_error_bundle = std.zig.ErrorBundle.empty;277 var result_error_bundle = std.zig.ErrorBundle.empty;
279278
280 const stdout = poller.fifo(.stdout);279 const stdout = poller.reader(.stdout);
281280
282 poll: while (true) {281 poll: while (true) {
283 while (stdout.readableLength() < @sizeOf(Header)) {282 const Header = std.zig.Server.Message.Header;
284 if (!(try poller.poll())) break :poll;283 while (stdout.buffered().len < @sizeOf(Header)) if (!try poller.poll()) break :poll;
285 }284 const header = stdout.takeStruct(Header, .little) catch unreachable;
286 const header = stdout.reader().readStruct(Header) catch unreachable;285 while (stdout.buffered().len < header.bytes_len) if (!try poller.poll()) break :poll;
287 while (stdout.readableLength() < header.bytes_len) {286 const body = stdout.take(header.bytes_len) catch unreachable;
288 if (!(try poller.poll())) break :poll;
289 }
290 const body = stdout.readableSliceOfLen(header.bytes_len);
291287
292 switch (header.tag) {288 switch (header.tag) {
293 .zig_version => {289 .zig_version => {
...@@ -325,15 +321,11 @@ fn buildWasmBinary(...@@ -325,15 +321,11 @@ fn buildWasmBinary(
325 },321 },
326 else => {}, // ignore other messages322 else => {}, // ignore other messages
327 }323 }
328
329 stdout.discard(body.len);
330 }324 }
331325
332 const stderr = poller.fifo(.stderr);326 const stderr_contents = try poller.toOwnedSlice(.stderr);
333 if (stderr.readableLength() > 0) {327 if (stderr_contents.len > 0) {
334 const owned_stderr = try stderr.toOwnedSlice();328 std.debug.print("{s}", .{stderr_contents});
335 defer gpa.free(owned_stderr);
336 std.debug.print("{s}", .{owned_stderr});
337 }329 }
338330
339 // Send EOF to stdin.331 // Send EOF to stdin.
lib/std/Build/Step.zig+25-34
...@@ -286,7 +286,7 @@ pub fn cast(step: *Step, comptime T: type) ?*T {...@@ -286,7 +286,7 @@ pub fn cast(step: *Step, comptime T: type) ?*T {
286}286}
287287
288/// For debugging purposes, prints identifying information about this Step.288/// For debugging purposes, prints identifying information about this Step.
289pub fn dump(step: *Step, w: *std.io.Writer, tty_config: std.io.tty.Config) void {289pub fn dump(step: *Step, w: *std.Io.Writer, tty_config: std.Io.tty.Config) void {
290 const debug_info = std.debug.getSelfDebugInfo() catch |err| {290 const debug_info = std.debug.getSelfDebugInfo() catch |err| {
291 w.print("Unable to dump stack trace: Unable to open debug info: {s}\n", .{291 w.print("Unable to dump stack trace: Unable to open debug info: {s}\n", .{
292 @errorName(err),292 @errorName(err),
...@@ -359,7 +359,7 @@ pub fn addError(step: *Step, comptime fmt: []const u8, args: anytype) error{OutO...@@ -359,7 +359,7 @@ pub fn addError(step: *Step, comptime fmt: []const u8, args: anytype) error{OutO
359359
360pub const ZigProcess = struct {360pub const ZigProcess = struct {
361 child: std.process.Child,361 child: std.process.Child,
362 poller: std.io.Poller(StreamEnum),362 poller: std.Io.Poller(StreamEnum),
363 progress_ipc_fd: if (std.Progress.have_ipc) ?std.posix.fd_t else void,363 progress_ipc_fd: if (std.Progress.have_ipc) ?std.posix.fd_t else void,
364364
365 pub const StreamEnum = enum { stdout, stderr };365 pub const StreamEnum = enum { stdout, stderr };
...@@ -428,7 +428,7 @@ pub fn evalZigProcess(...@@ -428,7 +428,7 @@ pub fn evalZigProcess(
428 const zp = try gpa.create(ZigProcess);428 const zp = try gpa.create(ZigProcess);
429 zp.* = .{429 zp.* = .{
430 .child = child,430 .child = child,
431 .poller = std.io.poll(gpa, ZigProcess.StreamEnum, .{431 .poller = std.Io.poll(gpa, ZigProcess.StreamEnum, .{
432 .stdout = child.stdout.?,432 .stdout = child.stdout.?,
433 .stderr = child.stderr.?,433 .stderr = child.stderr.?,
434 }),434 }),
...@@ -508,20 +508,16 @@ fn zigProcessUpdate(s: *Step, zp: *ZigProcess, watch: bool) !?Path {...@@ -508,20 +508,16 @@ fn zigProcessUpdate(s: *Step, zp: *ZigProcess, watch: bool) !?Path {
508 try sendMessage(zp.child.stdin.?, .update);508 try sendMessage(zp.child.stdin.?, .update);
509 if (!watch) try sendMessage(zp.child.stdin.?, .exit);509 if (!watch) try sendMessage(zp.child.stdin.?, .exit);
510510
511 const Header = std.zig.Server.Message.Header;
512 var result: ?Path = null;511 var result: ?Path = null;
513512
514 const stdout = zp.poller.fifo(.stdout);513 const stdout = zp.poller.reader(.stdout);
515514
516 poll: while (true) {515 poll: while (true) {
517 while (stdout.readableLength() < @sizeOf(Header)) {516 const Header = std.zig.Server.Message.Header;
518 if (!(try zp.poller.poll())) break :poll;517 while (stdout.buffered().len < @sizeOf(Header)) if (!try zp.poller.poll()) break :poll;
519 }518 const header = stdout.takeStruct(Header, .little) catch unreachable;
520 const header = stdout.reader().readStruct(Header) catch unreachable;519 while (stdout.buffered().len < header.bytes_len) if (!try zp.poller.poll()) break :poll;
521 while (stdout.readableLength() < header.bytes_len) {520 const body = stdout.take(header.bytes_len) catch unreachable;
522 if (!(try zp.poller.poll())) break :poll;
523 }
524 const body = stdout.readableSliceOfLen(header.bytes_len);
525521
526 switch (header.tag) {522 switch (header.tag) {
527 .zig_version => {523 .zig_version => {
...@@ -547,11 +543,8 @@ fn zigProcessUpdate(s: *Step, zp: *ZigProcess, watch: bool) !?Path {...@@ -547,11 +543,8 @@ fn zigProcessUpdate(s: *Step, zp: *ZigProcess, watch: bool) !?Path {
547 .string_bytes = try arena.dupe(u8, string_bytes),543 .string_bytes = try arena.dupe(u8, string_bytes),
548 .extra = extra_array,544 .extra = extra_array,
549 };545 };
550 if (watch) {546 // This message indicates the end of the update.
551 // This message indicates the end of the update.547 if (watch) break :poll;
552 stdout.discard(body.len);
553 break;
554 }
555 },548 },
556 .emit_digest => {549 .emit_digest => {
557 const EmitDigest = std.zig.Server.Message.EmitDigest;550 const EmitDigest = std.zig.Server.Message.EmitDigest;
...@@ -611,15 +604,13 @@ fn zigProcessUpdate(s: *Step, zp: *ZigProcess, watch: bool) !?Path {...@@ -611,15 +604,13 @@ fn zigProcessUpdate(s: *Step, zp: *ZigProcess, watch: bool) !?Path {
611 },604 },
612 else => {}, // ignore other messages605 else => {}, // ignore other messages
613 }606 }
614
615 stdout.discard(body.len);
616 }607 }
617608
618 s.result_duration_ns = timer.read();609 s.result_duration_ns = timer.read();
619610
620 const stderr = zp.poller.fifo(.stderr);611 const stderr_contents = try zp.poller.toOwnedSlice(.stderr);
621 if (stderr.readableLength() > 0) {612 if (stderr_contents.len > 0) {
622 try s.result_error_msgs.append(arena, try stderr.toOwnedSlice());613 try s.result_error_msgs.append(arena, try arena.dupe(u8, stderr_contents));
623 }614 }
624615
625 return result;616 return result;
...@@ -736,7 +727,7 @@ pub fn allocPrintCmd2(...@@ -736,7 +727,7 @@ pub fn allocPrintCmd2(
736 argv: []const []const u8,727 argv: []const []const u8,
737) Allocator.Error![]u8 {728) Allocator.Error![]u8 {
738 const shell = struct {729 const shell = struct {
739 fn escape(writer: anytype, string: []const u8, is_argv0: bool) !void {730 fn escape(writer: *std.Io.Writer, string: []const u8, is_argv0: bool) !void {
740 for (string) |c| {731 for (string) |c| {
741 if (switch (c) {732 if (switch (c) {
742 else => true,733 else => true,
...@@ -770,9 +761,9 @@ pub fn allocPrintCmd2(...@@ -770,9 +761,9 @@ pub fn allocPrintCmd2(
770 }761 }
771 };762 };
772763
773 var buf: std.ArrayListUnmanaged(u8) = .empty;764 var aw: std.Io.Writer.Allocating = .init(arena);
774 const writer = buf.writer(arena);765 const writer = &aw.writer;
775 if (opt_cwd) |cwd| try writer.print("cd {s} && ", .{cwd});766 if (opt_cwd) |cwd| writer.print("cd {s} && ", .{cwd}) catch return error.OutOfMemory;
776 if (opt_env) |env| {767 if (opt_env) |env| {
777 const process_env_map = std.process.getEnvMap(arena) catch std.process.EnvMap.init(arena);768 const process_env_map = std.process.getEnvMap(arena) catch std.process.EnvMap.init(arena);
778 var it = env.iterator();769 var it = env.iterator();
...@@ -782,17 +773,17 @@ pub fn allocPrintCmd2(...@@ -782,17 +773,17 @@ pub fn allocPrintCmd2(
782 if (process_env_map.get(key)) |process_value| {773 if (process_env_map.get(key)) |process_value| {
783 if (std.mem.eql(u8, value, process_value)) continue;774 if (std.mem.eql(u8, value, process_value)) continue;
784 }775 }
785 try writer.print("{s}=", .{key});776 writer.print("{s}=", .{key}) catch return error.OutOfMemory;
786 try shell.escape(writer, value, false);777 shell.escape(writer, value, false) catch return error.OutOfMemory;
787 try writer.writeByte(' ');778 writer.writeByte(' ') catch return error.OutOfMemory;
788 }779 }
789 }780 }
790 try shell.escape(writer, argv[0], true);781 shell.escape(writer, argv[0], true) catch return error.OutOfMemory;
791 for (argv[1..]) |arg| {782 for (argv[1..]) |arg| {
792 try writer.writeByte(' ');783 writer.writeByte(' ') catch return error.OutOfMemory;
793 try shell.escape(writer, arg, false);784 shell.escape(writer, arg, false) catch return error.OutOfMemory;
794 }785 }
795 return buf.toOwnedSlice(arena);786 return aw.toOwnedSlice();
796}787}
797788
798/// Prefer `cacheHitAndWatch` unless you already added watch inputs789/// Prefer `cacheHitAndWatch` unless you already added watch inputs
lib/std/Build/Step/Run.zig+44-34
...@@ -73,9 +73,12 @@ skip_foreign_checks: bool,...@@ -73,9 +73,12 @@ skip_foreign_checks: bool,
73/// external executor (such as qemu) but not fail if the executor is unavailable.73/// external executor (such as qemu) but not fail if the executor is unavailable.
74failing_to_execute_foreign_is_an_error: bool,74failing_to_execute_foreign_is_an_error: bool,
7575
76/// Deprecated in favor of `stdio_limit`.
77max_stdio_size: usize,
78
76/// If stderr or stdout exceeds this amount, the child process is killed and79/// If stderr or stdout exceeds this amount, the child process is killed and
77/// the step fails.80/// the step fails.
78max_stdio_size: usize,81stdio_limit: std.Io.Limit,
7982
80captured_stdout: ?*Output,83captured_stdout: ?*Output,
81captured_stderr: ?*Output,84captured_stderr: ?*Output,
...@@ -186,6 +189,7 @@ pub fn create(owner: *std.Build, name: []const u8) *Run {...@@ -186,6 +189,7 @@ pub fn create(owner: *std.Build, name: []const u8) *Run {
186 .skip_foreign_checks = false,189 .skip_foreign_checks = false,
187 .failing_to_execute_foreign_is_an_error = true,190 .failing_to_execute_foreign_is_an_error = true,
188 .max_stdio_size = 10 * 1024 * 1024,191 .max_stdio_size = 10 * 1024 * 1024,
192 .stdio_limit = .unlimited,
189 .captured_stdout = null,193 .captured_stdout = null,
190 .captured_stderr = null,194 .captured_stderr = null,
191 .dep_output_file = null,195 .dep_output_file = null,
...@@ -1011,7 +1015,7 @@ fn populateGeneratedPaths(...@@ -1011,7 +1015,7 @@ fn populateGeneratedPaths(
1011 }1015 }
1012}1016}
10131017
1014fn formatTerm(term: ?std.process.Child.Term, w: *std.io.Writer) std.io.Writer.Error!void {1018fn formatTerm(term: ?std.process.Child.Term, w: *std.Io.Writer) std.Io.Writer.Error!void {
1015 if (term) |t| switch (t) {1019 if (term) |t| switch (t) {
1016 .Exited => |code| try w.print("exited with code {d}", .{code}),1020 .Exited => |code| try w.print("exited with code {d}", .{code}),
1017 .Signal => |sig| try w.print("terminated with signal {d}", .{sig}),1021 .Signal => |sig| try w.print("terminated with signal {d}", .{sig}),
...@@ -1500,7 +1504,7 @@ fn evalZigTest(...@@ -1500,7 +1504,7 @@ fn evalZigTest(
1500 const gpa = run.step.owner.allocator;1504 const gpa = run.step.owner.allocator;
1501 const arena = run.step.owner.allocator;1505 const arena = run.step.owner.allocator;
15021506
1503 var poller = std.io.poll(gpa, enum { stdout, stderr }, .{1507 var poller = std.Io.poll(gpa, enum { stdout, stderr }, .{
1504 .stdout = child.stdout.?,1508 .stdout = child.stdout.?,
1505 .stderr = child.stderr.?,1509 .stderr = child.stderr.?,
1506 });1510 });
...@@ -1524,11 +1528,6 @@ fn evalZigTest(...@@ -1524,11 +1528,6 @@ fn evalZigTest(
1524 break :failed false;1528 break :failed false;
1525 };1529 };
15261530
1527 const Header = std.zig.Server.Message.Header;
1528
1529 const stdout = poller.fifo(.stdout);
1530 const stderr = poller.fifo(.stderr);
1531
1532 var fail_count: u32 = 0;1531 var fail_count: u32 = 0;
1533 var skip_count: u32 = 0;1532 var skip_count: u32 = 0;
1534 var leak_count: u32 = 0;1533 var leak_count: u32 = 0;
...@@ -1541,16 +1540,14 @@ fn evalZigTest(...@@ -1541,16 +1540,14 @@ fn evalZigTest(
1541 var sub_prog_node: ?std.Progress.Node = null;1540 var sub_prog_node: ?std.Progress.Node = null;
1542 defer if (sub_prog_node) |n| n.end();1541 defer if (sub_prog_node) |n| n.end();
15431542
1543 const stdout = poller.reader(.stdout);
1544 const stderr = poller.reader(.stderr);
1544 const any_write_failed = first_write_failed or poll: while (true) {1545 const any_write_failed = first_write_failed or poll: while (true) {
1545 while (stdout.readableLength() < @sizeOf(Header)) {1546 const Header = std.zig.Server.Message.Header;
1546 if (!(try poller.poll())) break :poll false;1547 while (stdout.buffered().len < @sizeOf(Header)) if (!try poller.poll()) break :poll false;
1547 }1548 const header = stdout.takeStruct(Header, .little) catch unreachable;
1548 const header = stdout.reader().readStruct(Header) catch unreachable;1549 while (stdout.buffered().len < header.bytes_len) if (!try poller.poll()) break :poll false;
1549 while (stdout.readableLength() < header.bytes_len) {1550 const body = stdout.take(header.bytes_len) catch unreachable;
1550 if (!(try poller.poll())) break :poll false;
1551 }
1552 const body = stdout.readableSliceOfLen(header.bytes_len);
1553
1554 switch (header.tag) {1551 switch (header.tag) {
1555 .zig_version => {1552 .zig_version => {
1556 if (!std.mem.eql(u8, builtin.zig_version_string, body)) {1553 if (!std.mem.eql(u8, builtin.zig_version_string, body)) {
...@@ -1607,9 +1604,9 @@ fn evalZigTest(...@@ -1607,9 +1604,9 @@ fn evalZigTest(
16071604
1608 if (tr_hdr.flags.fail or tr_hdr.flags.leak or tr_hdr.flags.log_err_count > 0) {1605 if (tr_hdr.flags.fail or tr_hdr.flags.leak or tr_hdr.flags.log_err_count > 0) {
1609 const name = std.mem.sliceTo(md.string_bytes[md.names[tr_hdr.index]..], 0);1606 const name = std.mem.sliceTo(md.string_bytes[md.names[tr_hdr.index]..], 0);
1610 const orig_msg = stderr.readableSlice(0);1607 const stderr_contents = stderr.buffered();
1611 defer stderr.discard(orig_msg.len);1608 stderr.toss(stderr_contents.len);
1612 const msg = std.mem.trim(u8, orig_msg, "\n");1609 const msg = std.mem.trim(u8, stderr_contents, "\n");
1613 const label = if (tr_hdr.flags.fail)1610 const label = if (tr_hdr.flags.fail)
1614 "failed"1611 "failed"
1615 else if (tr_hdr.flags.leak)1612 else if (tr_hdr.flags.leak)
...@@ -1660,8 +1657,6 @@ fn evalZigTest(...@@ -1660,8 +1657,6 @@ fn evalZigTest(
1660 },1657 },
1661 else => {}, // ignore other messages1658 else => {}, // ignore other messages
1662 }1659 }
1663
1664 stdout.discard(body.len);
1665 };1660 };
16661661
1667 if (any_write_failed) {1662 if (any_write_failed) {
...@@ -1670,9 +1665,9 @@ fn evalZigTest(...@@ -1670,9 +1665,9 @@ fn evalZigTest(
1670 while (try poller.poll()) {}1665 while (try poller.poll()) {}
1671 }1666 }
16721667
1673 if (stderr.readableLength() > 0) {1668 const stderr_contents = std.mem.trim(u8, stderr.buffered(), "\n");
1674 const msg = std.mem.trim(u8, try stderr.toOwnedSlice(), "\n");1669 if (stderr_contents.len > 0) {
1675 if (msg.len > 0) run.step.result_stderr = msg;1670 run.step.result_stderr = try arena.dupe(u8, stderr_contents);
1676 }1671 }
16771672
1678 // Send EOF to stdin.1673 // Send EOF to stdin.
...@@ -1795,28 +1790,43 @@ fn evalGeneric(run: *Run, child: *std.process.Child) !StdIoResult {...@@ -1795,28 +1790,43 @@ fn evalGeneric(run: *Run, child: *std.process.Child) !StdIoResult {
1795 var stdout_bytes: ?[]const u8 = null;1790 var stdout_bytes: ?[]const u8 = null;
1796 var stderr_bytes: ?[]const u8 = null;1791 var stderr_bytes: ?[]const u8 = null;
17971792
1793 run.stdio_limit = run.stdio_limit.min(.limited(run.max_stdio_size));
1798 if (child.stdout) |stdout| {1794 if (child.stdout) |stdout| {
1799 if (child.stderr) |stderr| {1795 if (child.stderr) |stderr| {
1800 var poller = std.io.poll(arena, enum { stdout, stderr }, .{1796 var poller = std.Io.poll(arena, enum { stdout, stderr }, .{
1801 .stdout = stdout,1797 .stdout = stdout,
1802 .stderr = stderr,1798 .stderr = stderr,
1803 });1799 });
1804 defer poller.deinit();1800 defer poller.deinit();
18051801
1806 while (try poller.poll()) {1802 while (try poller.poll()) {
1807 if (poller.fifo(.stdout).count > run.max_stdio_size)1803 if (run.stdio_limit.toInt()) |limit| {
1808 return error.StdoutStreamTooLong;1804 if (poller.reader(.stderr).buffered().len > limit)
1809 if (poller.fifo(.stderr).count > run.max_stdio_size)1805 return error.StdoutStreamTooLong;
1810 return error.StderrStreamTooLong;1806 if (poller.reader(.stderr).buffered().len > limit)
1807 return error.StderrStreamTooLong;
1808 }
1811 }1809 }
18121810
1813 stdout_bytes = try poller.fifo(.stdout).toOwnedSlice();1811 stdout_bytes = try poller.toOwnedSlice(.stdout);
1814 stderr_bytes = try poller.fifo(.stderr).toOwnedSlice();1812 stderr_bytes = try poller.toOwnedSlice(.stderr);
1815 } else {1813 } else {
1816 stdout_bytes = try stdout.deprecatedReader().readAllAlloc(arena, run.max_stdio_size);1814 var small_buffer: [1]u8 = undefined;
1815 var stdout_reader = stdout.readerStreaming(&small_buffer);
1816 stdout_bytes = stdout_reader.interface.allocRemaining(arena, run.stdio_limit) catch |err| switch (err) {
1817 error.OutOfMemory => return error.OutOfMemory,
1818 error.ReadFailed => return stdout_reader.err.?,
1819 error.StreamTooLong => return error.StdoutStreamTooLong,
1820 };
1817 }1821 }
1818 } else if (child.stderr) |stderr| {1822 } else if (child.stderr) |stderr| {
1819 stderr_bytes = try stderr.deprecatedReader().readAllAlloc(arena, run.max_stdio_size);1823 var small_buffer: [1]u8 = undefined;
1824 var stderr_reader = stderr.readerStreaming(&small_buffer);
1825 stderr_bytes = stderr_reader.interface.allocRemaining(arena, run.stdio_limit) catch |err| switch (err) {
1826 error.OutOfMemory => return error.OutOfMemory,
1827 error.ReadFailed => return stderr_reader.err.?,
1828 error.StreamTooLong => return error.StderrStreamTooLong,
1829 };
1820 }1830 }
18211831
1822 if (stderr_bytes) |bytes| if (bytes.len > 0) {1832 if (stderr_bytes) |bytes| if (bytes.len > 0) {
lib/std/Io.zig+228-176
...@@ -1,16 +1,11 @@...@@ -1,16 +1,11 @@
1const std = @import("std.zig");
2const builtin = @import("builtin");1const builtin = @import("builtin");
3const root = @import("root");
4const c = std.c;
5const is_windows = builtin.os.tag == .windows;2const is_windows = builtin.os.tag == .windows;
3
4const std = @import("std.zig");
6const windows = std.os.windows;5const windows = std.os.windows;
7const posix = std.posix;6const posix = std.posix;
8const math = std.math;7const math = std.math;
9const assert = std.debug.assert;8const assert = std.debug.assert;
10const fs = std.fs;
11const mem = std.mem;
12const meta = std.meta;
13const File = std.fs.File;
14const Allocator = std.mem.Allocator;9const Allocator = std.mem.Allocator;
15const Alignment = std.mem.Alignment;10const Alignment = std.mem.Alignment;
1611
...@@ -493,54 +488,51 @@ test null_writer {...@@ -493,54 +488,51 @@ test null_writer {
493}488}
494489
495pub fn poll(490pub fn poll(
496 allocator: Allocator,491 gpa: Allocator,
497 comptime StreamEnum: type,492 comptime StreamEnum: type,
498 files: PollFiles(StreamEnum),493 files: PollFiles(StreamEnum),
499) Poller(StreamEnum) {494) Poller(StreamEnum) {
500 const enum_fields = @typeInfo(StreamEnum).@"enum".fields;495 const enum_fields = @typeInfo(StreamEnum).@"enum".fields;
501 var result: Poller(StreamEnum) = undefined;496 var result: Poller(StreamEnum) = .{
502497 .gpa = gpa,
503 if (is_windows) result.windows = .{498 .readers = @splat(.failing),
504 .first_read_done = false,499 .poll_fds = undefined,
505 .overlapped = [1]windows.OVERLAPPED{500 .windows = if (is_windows) .{
506 mem.zeroes(windows.OVERLAPPED),501 .first_read_done = false,
507 } ** enum_fields.len,502 .overlapped = [1]windows.OVERLAPPED{
508 .small_bufs = undefined,503 std.mem.zeroes(windows.OVERLAPPED),
509 .active = .{504 } ** enum_fields.len,
510 .count = 0,505 .small_bufs = undefined,
511 .handles_buf = undefined,506 .active = .{
512 .stream_map = undefined,507 .count = 0,
513 },508 .handles_buf = undefined,
509 .stream_map = undefined,
510 },
511 } else {},
514 };512 };
515513
516 inline for (0..enum_fields.len) |i| {514 inline for (enum_fields, 0..) |field, i| {
517 result.fifos[i] = .{
518 .allocator = allocator,
519 .buf = &.{},
520 .head = 0,
521 .count = 0,
522 };
523 if (is_windows) {515 if (is_windows) {
524 result.windows.active.handles_buf[i] = @field(files, enum_fields[i].name).handle;516 result.windows.active.handles_buf[i] = @field(files, field.name).handle;
525 } else {517 } else {
526 result.poll_fds[i] = .{518 result.poll_fds[i] = .{
527 .fd = @field(files, enum_fields[i].name).handle,519 .fd = @field(files, field.name).handle,
528 .events = posix.POLL.IN,520 .events = posix.POLL.IN,
529 .revents = undefined,521 .revents = undefined,
530 };522 };
531 }523 }
532 }524 }
525
533 return result;526 return result;
534}527}
535528
536pub const PollFifo = std.fifo.LinearFifo(u8, .Dynamic);
537
538pub fn Poller(comptime StreamEnum: type) type {529pub fn Poller(comptime StreamEnum: type) type {
539 return struct {530 return struct {
540 const enum_fields = @typeInfo(StreamEnum).@"enum".fields;531 const enum_fields = @typeInfo(StreamEnum).@"enum".fields;
541 const PollFd = if (is_windows) void else posix.pollfd;532 const PollFd = if (is_windows) void else posix.pollfd;
542533
543 fifos: [enum_fields.len]PollFifo,534 gpa: Allocator,
535 readers: [enum_fields.len]Reader,
544 poll_fds: [enum_fields.len]PollFd,536 poll_fds: [enum_fields.len]PollFd,
545 windows: if (is_windows) struct {537 windows: if (is_windows) struct {
546 first_read_done: bool,538 first_read_done: bool,
...@@ -552,7 +544,7 @@ pub fn Poller(comptime StreamEnum: type) type {...@@ -552,7 +544,7 @@ pub fn Poller(comptime StreamEnum: type) type {
552 stream_map: [enum_fields.len]StreamEnum,544 stream_map: [enum_fields.len]StreamEnum,
553545
554 pub fn removeAt(self: *@This(), index: u32) void {546 pub fn removeAt(self: *@This(), index: u32) void {
555 std.debug.assert(index < self.count);547 assert(index < self.count);
556 for (index + 1..self.count) |i| {548 for (index + 1..self.count) |i| {
557 self.handles_buf[i - 1] = self.handles_buf[i];549 self.handles_buf[i - 1] = self.handles_buf[i];
558 self.stream_map[i - 1] = self.stream_map[i];550 self.stream_map[i - 1] = self.stream_map[i];
...@@ -565,13 +557,14 @@ pub fn Poller(comptime StreamEnum: type) type {...@@ -565,13 +557,14 @@ pub fn Poller(comptime StreamEnum: type) type {
565 const Self = @This();557 const Self = @This();
566558
567 pub fn deinit(self: *Self) void {559 pub fn deinit(self: *Self) void {
560 const gpa = self.gpa;
568 if (is_windows) {561 if (is_windows) {
569 // cancel any pending IO to prevent clobbering OVERLAPPED value562 // cancel any pending IO to prevent clobbering OVERLAPPED value
570 for (self.windows.active.handles_buf[0..self.windows.active.count]) |h| {563 for (self.windows.active.handles_buf[0..self.windows.active.count]) |h| {
571 _ = windows.kernel32.CancelIo(h);564 _ = windows.kernel32.CancelIo(h);
572 }565 }
573 }566 }
574 inline for (&self.fifos) |*q| q.deinit();567 inline for (&self.readers) |*r| gpa.free(r.buffer);
575 self.* = undefined;568 self.* = undefined;
576 }569 }
577570
...@@ -591,21 +584,40 @@ pub fn Poller(comptime StreamEnum: type) type {...@@ -591,21 +584,40 @@ pub fn Poller(comptime StreamEnum: type) type {
591 }584 }
592 }585 }
593586
594 pub inline fn fifo(self: *Self, comptime which: StreamEnum) *PollFifo {587 pub fn reader(self: *Self, which: StreamEnum) *Reader {
595 return &self.fifos[@intFromEnum(which)];588 return &self.readers[@intFromEnum(which)];
589 }
590
591 pub fn toOwnedSlice(self: *Self, which: StreamEnum) error{OutOfMemory}![]u8 {
592 const gpa = self.gpa;
593 const r = reader(self, which);
594 if (r.seek == 0) {
595 const new = try gpa.realloc(r.buffer, r.end);
596 r.buffer = &.{};
597 r.end = 0;
598 return new;
599 }
600 const new = try gpa.dupe(u8, r.buffered());
601 gpa.free(r.buffer);
602 r.buffer = &.{};
603 r.seek = 0;
604 r.end = 0;
605 return new;
596 }606 }
597607
598 fn pollWindows(self: *Self, nanoseconds: ?u64) !bool {608 fn pollWindows(self: *Self, nanoseconds: ?u64) !bool {
599 const bump_amt = 512;609 const bump_amt = 512;
610 const gpa = self.gpa;
600611
601 if (!self.windows.first_read_done) {612 if (!self.windows.first_read_done) {
602 var already_read_data = false;613 var already_read_data = false;
603 for (0..enum_fields.len) |i| {614 for (0..enum_fields.len) |i| {
604 const handle = self.windows.active.handles_buf[i];615 const handle = self.windows.active.handles_buf[i];
605 switch (try windowsAsyncReadToFifoAndQueueSmallRead(616 switch (try windowsAsyncReadToFifoAndQueueSmallRead(
617 gpa,
606 handle,618 handle,
607 &self.windows.overlapped[i],619 &self.windows.overlapped[i],
608 &self.fifos[i],620 &self.readers[i],
609 &self.windows.small_bufs[i],621 &self.windows.small_bufs[i],
610 bump_amt,622 bump_amt,
611 )) {623 )) {
...@@ -652,7 +664,7 @@ pub fn Poller(comptime StreamEnum: type) type {...@@ -652,7 +664,7 @@ pub fn Poller(comptime StreamEnum: type) type {
652 const handle = self.windows.active.handles_buf[active_idx];664 const handle = self.windows.active.handles_buf[active_idx];
653665
654 const overlapped = &self.windows.overlapped[stream_idx];666 const overlapped = &self.windows.overlapped[stream_idx];
655 const stream_fifo = &self.fifos[stream_idx];667 const stream_reader = &self.readers[stream_idx];
656 const small_buf = &self.windows.small_bufs[stream_idx];668 const small_buf = &self.windows.small_bufs[stream_idx];
657669
658 const num_bytes_read = switch (try windowsGetReadResult(handle, overlapped, false)) {670 const num_bytes_read = switch (try windowsGetReadResult(handle, overlapped, false)) {
...@@ -663,12 +675,16 @@ pub fn Poller(comptime StreamEnum: type) type {...@@ -663,12 +675,16 @@ pub fn Poller(comptime StreamEnum: type) type {
663 },675 },
664 .aborted => unreachable,676 .aborted => unreachable,
665 };677 };
666 try stream_fifo.write(small_buf[0..num_bytes_read]);678 const buf = small_buf[0..num_bytes_read];
679 const dest = try writableSliceGreedyAlloc(stream_reader, gpa, buf.len);
680 @memcpy(dest[0..buf.len], buf);
681 advanceBufferEnd(stream_reader, buf.len);
667682
668 switch (try windowsAsyncReadToFifoAndQueueSmallRead(683 switch (try windowsAsyncReadToFifoAndQueueSmallRead(
684 gpa,
669 handle,685 handle,
670 overlapped,686 overlapped,
671 stream_fifo,687 stream_reader,
672 small_buf,688 small_buf,
673 bump_amt,689 bump_amt,
674 )) {690 )) {
...@@ -683,6 +699,7 @@ pub fn Poller(comptime StreamEnum: type) type {...@@ -683,6 +699,7 @@ pub fn Poller(comptime StreamEnum: type) type {
683 }699 }
684700
685 fn pollPosix(self: *Self, nanoseconds: ?u64) !bool {701 fn pollPosix(self: *Self, nanoseconds: ?u64) !bool {
702 const gpa = self.gpa;
686 // We ask for ensureUnusedCapacity with this much extra space. This703 // We ask for ensureUnusedCapacity with this much extra space. This
687 // has more of an effect on small reads because once the reads704 // has more of an effect on small reads because once the reads
688 // start to get larger the amount of space an ArrayList will705 // start to get larger the amount of space an ArrayList will
...@@ -702,18 +719,18 @@ pub fn Poller(comptime StreamEnum: type) type {...@@ -702,18 +719,18 @@ pub fn Poller(comptime StreamEnum: type) type {
702 }719 }
703720
704 var keep_polling = false;721 var keep_polling = false;
705 inline for (&self.poll_fds, &self.fifos) |*poll_fd, *q| {722 for (&self.poll_fds, &self.readers) |*poll_fd, *r| {
706 // Try reading whatever is available before checking the error723 // Try reading whatever is available before checking the error
707 // conditions.724 // conditions.
708 // It's still possible to read after a POLL.HUP is received,725 // It's still possible to read after a POLL.HUP is received,
709 // always check if there's some data waiting to be read first.726 // always check if there's some data waiting to be read first.
710 if (poll_fd.revents & posix.POLL.IN != 0) {727 if (poll_fd.revents & posix.POLL.IN != 0) {
711 const buf = try q.writableWithSize(bump_amt);728 const buf = try writableSliceGreedyAlloc(r, gpa, bump_amt);
712 const amt = posix.read(poll_fd.fd, buf) catch |err| switch (err) {729 const amt = posix.read(poll_fd.fd, buf) catch |err| switch (err) {
713 error.BrokenPipe => 0, // Handle the same as EOF.730 error.BrokenPipe => 0, // Handle the same as EOF.
714 else => |e| return e,731 else => |e| return e,
715 };732 };
716 q.update(amt);733 advanceBufferEnd(r, amt);
717 if (amt == 0) {734 if (amt == 0) {
718 // Remove the fd when the EOF condition is met.735 // Remove the fd when the EOF condition is met.
719 poll_fd.fd = -1;736 poll_fd.fd = -1;
...@@ -729,146 +746,181 @@ pub fn Poller(comptime StreamEnum: type) type {...@@ -729,146 +746,181 @@ pub fn Poller(comptime StreamEnum: type) type {
729 }746 }
730 return keep_polling;747 return keep_polling;
731 }748 }
732 };
733}
734749
735/// The `ReadFile` docuementation states that `lpNumberOfBytesRead` does not have a meaningful750 /// Returns a slice into the unused capacity of `buffer` with at least
736/// result when using overlapped I/O, but also that it cannot be `null` on Windows 7. For751 /// `min_len` bytes, extending `buffer` by resizing it with `gpa` as necessary.
737/// compatibility, we point it to this dummy variables, which we never otherwise access.752 ///
738/// See: https://learn.microsoft.com/en-us/windows/win32/api/fileapi/nf-fileapi-readfile753 /// After calling this function, typically the caller will follow up with a
739var win_dummy_bytes_read: u32 = undefined;754 /// call to `advanceBufferEnd` to report the actual number of bytes buffered.
740755 fn writableSliceGreedyAlloc(r: *Reader, allocator: Allocator, min_len: usize) Allocator.Error![]u8 {
741/// Read as much data as possible from `handle` with `overlapped`, and write it to the FIFO. Before756 {
742/// returning, queue a read into `small_buf` so that `WaitForMultipleObjects` returns when more data757 const unused = r.buffer[r.end..];
743/// is available. `handle` must have no pending asynchronous operation.758 if (unused.len >= min_len) return unused;
744fn windowsAsyncReadToFifoAndQueueSmallRead(759 }
745 handle: windows.HANDLE,760 if (r.seek > 0) r.rebase();
746 overlapped: *windows.OVERLAPPED,761 {
747 fifo: *PollFifo,762 var list: std.ArrayListUnmanaged(u8) = .{
748 small_buf: *[128]u8,763 .items = r.buffer[0..r.end],
749 bump_amt: usize,764 .capacity = r.buffer.len,
750) !enum { empty, populated, closed_populated, closed } {765 };
751 var read_any_data = false;766 defer r.buffer = list.allocatedSlice();
752 while (true) {767 try list.ensureUnusedCapacity(allocator, min_len);
753 const fifo_read_pending = while (true) {768 }
754 const buf = try fifo.writableWithSize(bump_amt);769 const unused = r.buffer[r.end..];
755 const buf_len = math.cast(u32, buf.len) orelse math.maxInt(u32);770 assert(unused.len >= min_len);
756771 return unused;
757 if (0 == windows.kernel32.ReadFile(772 }
758 handle,773
759 buf.ptr,774 /// After writing directly into the unused capacity of `buffer`, this function
760 buf_len,775 /// updates `end` so that users of `Reader` can receive the data.
761 &win_dummy_bytes_read,776 fn advanceBufferEnd(r: *Reader, n: usize) void {
762 overlapped,777 assert(n <= r.buffer.len - r.end);
763 )) switch (windows.GetLastError()) {778 r.end += n;
764 .IO_PENDING => break true,779 }
765 .BROKEN_PIPE => return if (read_any_data) .closed_populated else .closed,780
766 else => |err| return windows.unexpectedError(err),781 /// The `ReadFile` docuementation states that `lpNumberOfBytesRead` does not have a meaningful
767 };782 /// result when using overlapped I/O, but also that it cannot be `null` on Windows 7. For
783 /// compatibility, we point it to this dummy variables, which we never otherwise access.
784 /// See: https://learn.microsoft.com/en-us/windows/win32/api/fileapi/nf-fileapi-readfile
785 var win_dummy_bytes_read: u32 = undefined;
786
787 /// Read as much data as possible from `handle` with `overlapped`, and write it to the FIFO. Before
788 /// returning, queue a read into `small_buf` so that `WaitForMultipleObjects` returns when more data
789 /// is available. `handle` must have no pending asynchronous operation.
790 fn windowsAsyncReadToFifoAndQueueSmallRead(
791 gpa: Allocator,
792 handle: windows.HANDLE,
793 overlapped: *windows.OVERLAPPED,
794 r: *Reader,
795 small_buf: *[128]u8,
796 bump_amt: usize,
797 ) !enum { empty, populated, closed_populated, closed } {
798 var read_any_data = false;
799 while (true) {
800 const fifo_read_pending = while (true) {
801 const buf = try writableSliceGreedyAlloc(r, gpa, bump_amt);
802 const buf_len = math.cast(u32, buf.len) orelse math.maxInt(u32);
768803
769 const num_bytes_read = switch (try windowsGetReadResult(handle, overlapped, false)) {804 if (0 == windows.kernel32.ReadFile(
770 .success => |n| n,805 handle,
771 .closed => return if (read_any_data) .closed_populated else .closed,806 buf.ptr,
772 .aborted => unreachable,807 buf_len,
773 };808 &win_dummy_bytes_read,
809 overlapped,
810 )) switch (windows.GetLastError()) {
811 .IO_PENDING => break true,
812 .BROKEN_PIPE => return if (read_any_data) .closed_populated else .closed,
813 else => |err| return windows.unexpectedError(err),
814 };
774815
775 read_any_data = true;816 const num_bytes_read = switch (try windowsGetReadResult(handle, overlapped, false)) {
776 fifo.update(num_bytes_read);817 .success => |n| n,
818 .closed => return if (read_any_data) .closed_populated else .closed,
819 .aborted => unreachable,
820 };
777821
778 if (num_bytes_read == buf_len) {822 read_any_data = true;
779 // We filled the buffer, so there's probably more data available.823 advanceBufferEnd(r, num_bytes_read);
780 continue;
781 } else {
782 // We didn't fill the buffer, so assume we're out of data.
783 // There is no pending read.
784 break false;
785 }
786 };
787824
788 if (fifo_read_pending) cancel_read: {825 if (num_bytes_read == buf_len) {
789 // Cancel the pending read into the FIFO.826 // We filled the buffer, so there's probably more data available.
790 _ = windows.kernel32.CancelIo(handle);827 continue;
828 } else {
829 // We didn't fill the buffer, so assume we're out of data.
830 // There is no pending read.
831 break false;
832 }
833 };
791834
792 // We have to wait for the handle to be signalled, i.e. for the cancellation to complete.835 if (fifo_read_pending) cancel_read: {
793 switch (windows.kernel32.WaitForSingleObject(handle, windows.INFINITE)) {836 // Cancel the pending read into the FIFO.
794 windows.WAIT_OBJECT_0 => {},837 _ = windows.kernel32.CancelIo(handle);
795 windows.WAIT_FAILED => return windows.unexpectedError(windows.GetLastError()),
796 else => unreachable,
797 }
798838
799 // If it completed before we canceled, make sure to tell the FIFO!839 // We have to wait for the handle to be signalled, i.e. for the cancellation to complete.
800 const num_bytes_read = switch (try windowsGetReadResult(handle, overlapped, true)) {840 switch (windows.kernel32.WaitForSingleObject(handle, windows.INFINITE)) {
801 .success => |n| n,841 windows.WAIT_OBJECT_0 => {},
802 .closed => return if (read_any_data) .closed_populated else .closed,842 windows.WAIT_FAILED => return windows.unexpectedError(windows.GetLastError()),
803 .aborted => break :cancel_read,843 else => unreachable,
804 };844 }
805 read_any_data = true;
806 fifo.update(num_bytes_read);
807 }
808
809 // Try to queue the 1-byte read.
810 if (0 == windows.kernel32.ReadFile(
811 handle,
812 small_buf,
813 small_buf.len,
814 &win_dummy_bytes_read,
815 overlapped,
816 )) switch (windows.GetLastError()) {
817 .IO_PENDING => {
818 // 1-byte read pending as intended
819 return if (read_any_data) .populated else .empty;
820 },
821 .BROKEN_PIPE => return if (read_any_data) .closed_populated else .closed,
822 else => |err| return windows.unexpectedError(err),
823 };
824845
825 // We got data back this time. Write it to the FIFO and run the main loop again.846 // If it completed before we canceled, make sure to tell the FIFO!
826 const num_bytes_read = switch (try windowsGetReadResult(handle, overlapped, false)) {847 const num_bytes_read = switch (try windowsGetReadResult(handle, overlapped, true)) {
827 .success => |n| n,848 .success => |n| n,
828 .closed => return if (read_any_data) .closed_populated else .closed,849 .closed => return if (read_any_data) .closed_populated else .closed,
829 .aborted => unreachable,850 .aborted => break :cancel_read,
830 };851 };
831 try fifo.write(small_buf[0..num_bytes_read]);852 read_any_data = true;
832 read_any_data = true;853 advanceBufferEnd(r, num_bytes_read);
833 }854 }
834}
835855
836/// Simple wrapper around `GetOverlappedResult` to determine the result of a `ReadFile` operation.856 // Try to queue the 1-byte read.
837/// If `!allow_aborted`, then `aborted` is never returned (`OPERATION_ABORTED` is considered unexpected).857 if (0 == windows.kernel32.ReadFile(
838///858 handle,
839/// The `ReadFile` documentation states that the number of bytes read by an overlapped `ReadFile` must be determined using `GetOverlappedResult`, even if the859 small_buf,
840/// operation immediately returns data:860 small_buf.len,
841/// "Use NULL for [lpNumberOfBytesRead] if this is an asynchronous operation to avoid potentially861 &win_dummy_bytes_read,
842/// erroneous results."862 overlapped,
843/// "If `hFile` was opened with `FILE_FLAG_OVERLAPPED`, the following conditions are in effect: [...]863 )) switch (windows.GetLastError()) {
844/// The lpNumberOfBytesRead parameter should be set to NULL. Use the GetOverlappedResult function to864 .IO_PENDING => {
845/// get the actual number of bytes read."865 // 1-byte read pending as intended
846/// See: https://learn.microsoft.com/en-us/windows/win32/api/fileapi/nf-fileapi-readfile866 return if (read_any_data) .populated else .empty;
847fn windowsGetReadResult(867 },
848 handle: windows.HANDLE,868 .BROKEN_PIPE => return if (read_any_data) .closed_populated else .closed,
849 overlapped: *windows.OVERLAPPED,869 else => |err| return windows.unexpectedError(err),
850 allow_aborted: bool,870 };
851) !union(enum) {871
852 success: u32,872 // We got data back this time. Write it to the FIFO and run the main loop again.
853 closed,873 const num_bytes_read = switch (try windowsGetReadResult(handle, overlapped, false)) {
854 aborted,874 .success => |n| n,
855} {875 .closed => return if (read_any_data) .closed_populated else .closed,
856 var num_bytes_read: u32 = undefined;876 .aborted => unreachable,
857 if (0 == windows.kernel32.GetOverlappedResult(877 };
858 handle,878 const buf = small_buf[0..num_bytes_read];
859 overlapped,879 const dest = try writableSliceGreedyAlloc(r, gpa, buf.len);
860 &num_bytes_read,880 @memcpy(dest[0..buf.len], buf);
861 0,881 advanceBufferEnd(r, buf.len);
862 )) switch (windows.GetLastError()) {882 read_any_data = true;
863 .BROKEN_PIPE => return .closed,883 }
864 .OPERATION_ABORTED => |err| if (allow_aborted) {884 }
865 return .aborted;885
866 } else {886 /// Simple wrapper around `GetOverlappedResult` to determine the result of a `ReadFile` operation.
867 return windows.unexpectedError(err);887 /// If `!allow_aborted`, then `aborted` is never returned (`OPERATION_ABORTED` is considered unexpected).
868 },888 ///
869 else => |err| return windows.unexpectedError(err),889 /// The `ReadFile` documentation states that the number of bytes read by an overlapped `ReadFile` must be determined using `GetOverlappedResult`, even if the
890 /// operation immediately returns data:
891 /// "Use NULL for [lpNumberOfBytesRead] if this is an asynchronous operation to avoid potentially
892 /// erroneous results."
893 /// "If `hFile` was opened with `FILE_FLAG_OVERLAPPED`, the following conditions are in effect: [...]
894 /// The lpNumberOfBytesRead parameter should be set to NULL. Use the GetOverlappedResult function to
895 /// get the actual number of bytes read."
896 /// See: https://learn.microsoft.com/en-us/windows/win32/api/fileapi/nf-fileapi-readfile
897 fn windowsGetReadResult(
898 handle: windows.HANDLE,
899 overlapped: *windows.OVERLAPPED,
900 allow_aborted: bool,
901 ) !union(enum) {
902 success: u32,
903 closed,
904 aborted,
905 } {
906 var num_bytes_read: u32 = undefined;
907 if (0 == windows.kernel32.GetOverlappedResult(
908 handle,
909 overlapped,
910 &num_bytes_read,
911 0,
912 )) switch (windows.GetLastError()) {
913 .BROKEN_PIPE => return .closed,
914 .OPERATION_ABORTED => |err| if (allow_aborted) {
915 return .aborted;
916 } else {
917 return windows.unexpectedError(err);
918 },
919 else => |err| return windows.unexpectedError(err),
920 };
921 return .{ .success = num_bytes_read };
922 }
870 };923 };
871 return .{ .success = num_bytes_read };
872}924}
873925
874/// Given an enum, returns a struct with fields of that enum, each field926/// Given an enum, returns a struct with fields of that enum, each field
...@@ -879,10 +931,10 @@ pub fn PollFiles(comptime StreamEnum: type) type {...@@ -879,10 +931,10 @@ pub fn PollFiles(comptime StreamEnum: type) type {
879 for (&struct_fields, enum_fields) |*struct_field, enum_field| {931 for (&struct_fields, enum_fields) |*struct_field, enum_field| {
880 struct_field.* = .{932 struct_field.* = .{
881 .name = enum_field.name,933 .name = enum_field.name,
882 .type = fs.File,934 .type = std.fs.File,
883 .default_value_ptr = null,935 .default_value_ptr = null,
884 .is_comptime = false,936 .is_comptime = false,
885 .alignment = @alignOf(fs.File),937 .alignment = @alignOf(std.fs.File),
886 };938 };
887 }939 }
888 return @Type(.{ .@"struct" = .{940 return @Type(.{ .@"struct" = .{
lib/std/Io/Reader.zig-31
...@@ -1241,37 +1241,6 @@ pub fn fillAlloc(r: *Reader, allocator: Allocator, n: usize) FillAllocError!void...@@ -1241,37 +1241,6 @@ pub fn fillAlloc(r: *Reader, allocator: Allocator, n: usize) FillAllocError!void
1241 return fill(r, n);1241 return fill(r, n);
1242}1242}
12431243
1244/// Returns a slice into the unused capacity of `buffer` with at least
1245/// `min_len` bytes, extending `buffer` by resizing it with `gpa` as necessary.
1246///
1247/// After calling this function, typically the caller will follow up with a
1248/// call to `advanceBufferEnd` to report the actual number of bytes buffered.
1249pub fn writableSliceGreedyAlloc(r: *Reader, allocator: Allocator, min_len: usize) Allocator.Error![]u8 {
1250 {
1251 const unused = r.buffer[r.end..];
1252 if (unused.len >= min_len) return unused;
1253 }
1254 if (r.seek > 0) rebase(r);
1255 {
1256 var list: ArrayList(u8) = .{
1257 .items = r.buffer[0..r.end],
1258 .capacity = r.buffer.len,
1259 };
1260 defer r.buffer = list.allocatedSlice();
1261 try list.ensureUnusedCapacity(allocator, min_len);
1262 }
1263 const unused = r.buffer[r.end..];
1264 assert(unused.len >= min_len);
1265 return unused;
1266}
1267
1268/// After writing directly into the unused capacity of `buffer`, this function
1269/// updates `end` so that users of `Reader` can receive the data.
1270pub fn advanceBufferEnd(r: *Reader, n: usize) void {
1271 assert(n <= r.buffer.len - r.end);
1272 r.end += n;
1273}
1274
1275fn takeMultipleOf7Leb128(r: *Reader, comptime Result: type) TakeLeb128Error!Result {1244fn takeMultipleOf7Leb128(r: *Reader, comptime Result: type) TakeLeb128Error!Result {
1276 const result_info = @typeInfo(Result).int;1245 const result_info = @typeInfo(Result).int;
1277 comptime assert(result_info.bits % 7 == 0);1246 comptime assert(result_info.bits % 7 == 0);
lib/std/process/Child.zig+46-35
...@@ -14,6 +14,7 @@ const assert = std.debug.assert;...@@ -14,6 +14,7 @@ const assert = std.debug.assert;
14const native_os = builtin.os.tag;14const native_os = builtin.os.tag;
15const Allocator = std.mem.Allocator;15const Allocator = std.mem.Allocator;
16const ChildProcess = @This();16const ChildProcess = @This();
17const ArrayList = std.ArrayListUnmanaged;
1718
18pub const Id = switch (native_os) {19pub const Id = switch (native_os) {
19 .windows => windows.HANDLE,20 .windows => windows.HANDLE,
...@@ -348,19 +349,6 @@ pub const RunResult = struct {...@@ -348,19 +349,6 @@ pub const RunResult = struct {
348 stderr: []u8,349 stderr: []u8,
349};350};
350351
351fn writeFifoDataToArrayList(allocator: Allocator, list: *std.ArrayListUnmanaged(u8), fifo: *std.io.PollFifo) !void {
352 if (fifo.head != 0) fifo.realign();
353 if (list.capacity == 0) {
354 list.* = .{
355 .items = fifo.buf[0..fifo.count],
356 .capacity = fifo.buf.len,
357 };
358 fifo.* = std.io.PollFifo.init(fifo.allocator);
359 } else {
360 try list.appendSlice(allocator, fifo.buf[0..fifo.count]);
361 }
362}
363
364/// Collect the output from the process's stdout and stderr. Will return once all output352/// Collect the output from the process's stdout and stderr. Will return once all output
365/// has been collected. This does not mean that the process has ended. `wait` should still353/// has been collected. This does not mean that the process has ended. `wait` should still
366/// be called to wait for and clean up the process.354/// be called to wait for and clean up the process.
...@@ -370,28 +358,48 @@ pub fn collectOutput(...@@ -370,28 +358,48 @@ pub fn collectOutput(
370 child: ChildProcess,358 child: ChildProcess,
371 /// Used for `stdout` and `stderr`.359 /// Used for `stdout` and `stderr`.
372 allocator: Allocator,360 allocator: Allocator,
373 stdout: *std.ArrayListUnmanaged(u8),361 stdout: *ArrayList(u8),
374 stderr: *std.ArrayListUnmanaged(u8),362 stderr: *ArrayList(u8),
375 max_output_bytes: usize,363 max_output_bytes: usize,
376) !void {364) !void {
377 assert(child.stdout_behavior == .Pipe);365 assert(child.stdout_behavior == .Pipe);
378 assert(child.stderr_behavior == .Pipe);366 assert(child.stderr_behavior == .Pipe);
379367
380 var poller = std.io.poll(allocator, enum { stdout, stderr }, .{368 var poller = std.Io.poll(allocator, enum { stdout, stderr }, .{
381 .stdout = child.stdout.?,369 .stdout = child.stdout.?,
382 .stderr = child.stderr.?,370 .stderr = child.stderr.?,
383 });371 });
384 defer poller.deinit();372 defer poller.deinit();
385373
374 const stdout_r = poller.reader(.stdout);
375 stdout_r.buffer = stdout.allocatedSlice();
376 stdout_r.seek = 0;
377 stdout_r.end = stdout.items.len;
378
379 const stderr_r = poller.reader(.stderr);
380 stderr_r.buffer = stderr.allocatedSlice();
381 stderr_r.seek = 0;
382 stderr_r.end = stderr.items.len;
383
384 defer {
385 stdout.* = .{
386 .items = stdout_r.buffer[0..stdout_r.end],
387 .capacity = stdout_r.buffer.len,
388 };
389 stderr.* = .{
390 .items = stderr_r.buffer[0..stderr_r.end],
391 .capacity = stderr_r.buffer.len,
392 };
393 stdout_r.buffer = &.{};
394 stderr_r.buffer = &.{};
395 }
396
386 while (try poller.poll()) {397 while (try poller.poll()) {
387 if (poller.fifo(.stdout).count > max_output_bytes)398 if (stdout_r.bufferedLen() > max_output_bytes)
388 return error.StdoutStreamTooLong;399 return error.StdoutStreamTooLong;
389 if (poller.fifo(.stderr).count > max_output_bytes)400 if (stderr_r.bufferedLen() > max_output_bytes)
390 return error.StderrStreamTooLong;401 return error.StderrStreamTooLong;
391 }402 }
392
393 try writeFifoDataToArrayList(allocator, stdout, poller.fifo(.stdout));
394 try writeFifoDataToArrayList(allocator, stderr, poller.fifo(.stderr));
395}403}
396404
397pub const RunError = posix.GetCwdError || posix.ReadError || SpawnError || posix.PollError || error{405pub const RunError = posix.GetCwdError || posix.ReadError || SpawnError || posix.PollError || error{
...@@ -421,10 +429,10 @@ pub fn run(args: struct {...@@ -421,10 +429,10 @@ pub fn run(args: struct {
421 child.expand_arg0 = args.expand_arg0;429 child.expand_arg0 = args.expand_arg0;
422 child.progress_node = args.progress_node;430 child.progress_node = args.progress_node;
423431
424 var stdout: std.ArrayListUnmanaged(u8) = .empty;432 var stdout: ArrayList(u8) = .empty;
425 errdefer stdout.deinit(args.allocator);433 defer stdout.deinit(args.allocator);
426 var stderr: std.ArrayListUnmanaged(u8) = .empty;434 var stderr: ArrayList(u8) = .empty;
427 errdefer stderr.deinit(args.allocator);435 defer stderr.deinit(args.allocator);
428436
429 try child.spawn();437 try child.spawn();
430 errdefer {438 errdefer {
...@@ -432,7 +440,7 @@ pub fn run(args: struct {...@@ -432,7 +440,7 @@ pub fn run(args: struct {
432 }440 }
433 try child.collectOutput(args.allocator, &stdout, &stderr, args.max_output_bytes);441 try child.collectOutput(args.allocator, &stdout, &stderr, args.max_output_bytes);
434442
435 return RunResult{443 return .{
436 .stdout = try stdout.toOwnedSlice(args.allocator),444 .stdout = try stdout.toOwnedSlice(args.allocator),
437 .stderr = try stderr.toOwnedSlice(args.allocator),445 .stderr = try stderr.toOwnedSlice(args.allocator),
438 .term = try child.wait(),446 .term = try child.wait(),
...@@ -878,12 +886,12 @@ fn spawnWindows(self: *ChildProcess) SpawnError!void {...@@ -878,12 +886,12 @@ fn spawnWindows(self: *ChildProcess) SpawnError!void {
878 var cmd_line_cache = WindowsCommandLineCache.init(self.allocator, self.argv);886 var cmd_line_cache = WindowsCommandLineCache.init(self.allocator, self.argv);
879 defer cmd_line_cache.deinit();887 defer cmd_line_cache.deinit();
880888
881 var app_buf: std.ArrayListUnmanaged(u16) = .empty;889 var app_buf: ArrayList(u16) = .empty;
882 defer app_buf.deinit(self.allocator);890 defer app_buf.deinit(self.allocator);
883891
884 try app_buf.appendSlice(self.allocator, app_name_w);892 try app_buf.appendSlice(self.allocator, app_name_w);
885893
886 var dir_buf: std.ArrayListUnmanaged(u16) = .empty;894 var dir_buf: ArrayList(u16) = .empty;
887 defer dir_buf.deinit(self.allocator);895 defer dir_buf.deinit(self.allocator);
888896
889 if (cwd_path_w.len > 0) {897 if (cwd_path_w.len > 0) {
...@@ -1003,13 +1011,16 @@ fn forkChildErrReport(fd: i32, err: ChildProcess.SpawnError) noreturn {...@@ -1003,13 +1011,16 @@ fn forkChildErrReport(fd: i32, err: ChildProcess.SpawnError) noreturn {
1003}1011}
10041012
1005fn writeIntFd(fd: i32, value: ErrInt) !void {1013fn writeIntFd(fd: i32, value: ErrInt) !void {
1006 const file: File = .{ .handle = fd };1014 var buffer: [8]u8 = undefined;
1007 file.deprecatedWriter().writeInt(u64, @intCast(value), .little) catch return error.SystemResources;1015 var fw: std.fs.File.Writer = .initMode(.{ .handle = fd }, &buffer, .streaming);
1016 fw.interface.writeInt(u64, value, .little) catch unreachable;
1017 fw.interface.flush() catch return error.SystemResources;
1008}1018}
10091019
1010fn readIntFd(fd: i32) !ErrInt {1020fn readIntFd(fd: i32) !ErrInt {
1011 const file: File = .{ .handle = fd };1021 var buffer: [8]u8 = undefined;
1012 return @intCast(file.deprecatedReader().readInt(u64, .little) catch return error.SystemResources);1022 var fr: std.fs.File.Reader = .initMode(.{ .handle = fd }, &buffer, .streaming);
1023 return @intCast(fr.interface.takeInt(u64, .little) catch return error.SystemResources);
1013}1024}
10141025
1015const ErrInt = std.meta.Int(.unsigned, @sizeOf(anyerror) * 8);1026const ErrInt = std.meta.Int(.unsigned, @sizeOf(anyerror) * 8);
...@@ -1020,8 +1031,8 @@ const ErrInt = std.meta.Int(.unsigned, @sizeOf(anyerror) * 8);...@@ -1020,8 +1031,8 @@ const ErrInt = std.meta.Int(.unsigned, @sizeOf(anyerror) * 8);
1020/// Note: If the dir is the cwd, dir_buf should be empty (len = 0).1031/// Note: If the dir is the cwd, dir_buf should be empty (len = 0).
1021fn windowsCreateProcessPathExt(1032fn windowsCreateProcessPathExt(
1022 allocator: mem.Allocator,1033 allocator: mem.Allocator,
1023 dir_buf: *std.ArrayListUnmanaged(u16),1034 dir_buf: *ArrayList(u16),
1024 app_buf: *std.ArrayListUnmanaged(u16),1035 app_buf: *ArrayList(u16),
1025 pathext: [:0]const u16,1036 pathext: [:0]const u16,
1026 cmd_line_cache: *WindowsCommandLineCache,1037 cmd_line_cache: *WindowsCommandLineCache,
1027 envp_ptr: ?[*]u16,1038 envp_ptr: ?[*]u16,
...@@ -1504,7 +1515,7 @@ const WindowsCommandLineCache = struct {...@@ -1504,7 +1515,7 @@ const WindowsCommandLineCache = struct {
1504/// Returns the absolute path of `cmd.exe` within the Windows system directory.1515/// Returns the absolute path of `cmd.exe` within the Windows system directory.
1505/// The caller owns the returned slice.1516/// The caller owns the returned slice.
1506fn windowsCmdExePath(allocator: mem.Allocator) error{ OutOfMemory, Unexpected }![:0]u16 {1517fn windowsCmdExePath(allocator: mem.Allocator) error{ OutOfMemory, Unexpected }![:0]u16 {
1507 var buf = try std.ArrayListUnmanaged(u16).initCapacity(allocator, 128);1518 var buf = try ArrayList(u16).initCapacity(allocator, 128);
1508 errdefer buf.deinit(allocator);1519 errdefer buf.deinit(allocator);
1509 while (true) {1520 while (true) {
1510 const unused_slice = buf.unusedCapacitySlice();1521 const unused_slice = buf.unusedCapacitySlice();
src/Compilation.zig+11-13
...@@ -6215,19 +6215,20 @@ fn spawnZigRc(...@@ -6215,19 +6215,20 @@ fn spawnZigRc(
6215 return comp.failWin32Resource(win32_resource, "unable to spawn {s} rc: {s}", .{ argv[0], @errorName(err) });6215 return comp.failWin32Resource(win32_resource, "unable to spawn {s} rc: {s}", .{ argv[0], @errorName(err) });
6216 };6216 };
62176217
6218 var poller = std.io.poll(comp.gpa, enum { stdout }, .{6218 var poller = std.Io.poll(comp.gpa, enum { stdout, stderr }, .{
6219 .stdout = child.stdout.?,6219 .stdout = child.stdout.?,
6220 .stderr = child.stderr.?,
6220 });6221 });
6221 defer poller.deinit();6222 defer poller.deinit();
62226223
6223 const stdout = poller.fifo(.stdout);6224 const stdout = poller.reader(.stdout);
62246225
6225 poll: while (true) {6226 poll: while (true) {
6226 while (stdout.readableLength() < @sizeOf(std.zig.Server.Message.Header)) if (!try poller.poll()) break :poll;6227 const MessageHeader = std.zig.Server.Message.Header;
6227 var header: std.zig.Server.Message.Header = undefined;6228 while (stdout.buffered().len < @sizeOf(MessageHeader)) if (!try poller.poll()) break :poll;
6228 assert(stdout.read(std.mem.asBytes(&header)) == @sizeOf(std.zig.Server.Message.Header));6229 const header = stdout.takeStruct(MessageHeader, .little) catch unreachable;
6229 while (stdout.readableLength() < header.bytes_len) if (!try poller.poll()) break :poll;6230 while (stdout.buffered().len < header.bytes_len) if (!try poller.poll()) break :poll;
6230 const body = stdout.readableSliceOfLen(header.bytes_len);6231 const body = stdout.take(header.bytes_len) catch unreachable;
62316232
6232 switch (header.tag) {6233 switch (header.tag) {
6233 // We expect exactly one ErrorBundle, and if any error_bundle header is6234 // We expect exactly one ErrorBundle, and if any error_bundle header is
...@@ -6250,13 +6251,10 @@ fn spawnZigRc(...@@ -6250,13 +6251,10 @@ fn spawnZigRc(
6250 },6251 },
6251 else => {}, // ignore other messages6252 else => {}, // ignore other messages
6252 }6253 }
6253
6254 stdout.discard(body.len);
6255 }6254 }
62566255
6257 // Just in case there's a failure that didn't send an ErrorBundle (e.g. an error return trace)6256 // Just in case there's a failure that didn't send an ErrorBundle (e.g. an error return trace)
6258 const stderr_reader = child.stderr.?.deprecatedReader();6257 const stderr = poller.reader(.stderr);
6259 const stderr = try stderr_reader.readAllAlloc(arena, 10 * 1024 * 1024);
62606258
6261 const term = child.wait() catch |err| {6259 const term = child.wait() catch |err| {
6262 return comp.failWin32Resource(win32_resource, "unable to wait for {s} rc: {s}", .{ argv[0], @errorName(err) });6260 return comp.failWin32Resource(win32_resource, "unable to wait for {s} rc: {s}", .{ argv[0], @errorName(err) });
...@@ -6265,12 +6263,12 @@ fn spawnZigRc(...@@ -6265,12 +6263,12 @@ fn spawnZigRc(
6265 switch (term) {6263 switch (term) {
6266 .Exited => |code| {6264 .Exited => |code| {
6267 if (code != 0) {6265 if (code != 0) {
6268 log.err("zig rc failed with stderr:\n{s}", .{stderr});6266 log.err("zig rc failed with stderr:\n{s}", .{stderr.buffered()});
6269 return comp.failWin32Resource(win32_resource, "zig rc exited with code {d}", .{code});6267 return comp.failWin32Resource(win32_resource, "zig rc exited with code {d}", .{code});
6270 }6268 }
6271 },6269 },
6272 else => {6270 else => {
6273 log.err("zig rc terminated with stderr:\n{s}", .{stderr});6271 log.err("zig rc terminated with stderr:\n{s}", .{stderr.buffered()});
6274 return comp.failWin32Resource(win32_resource, "zig rc terminated unexpectedly", .{});6272 return comp.failWin32Resource(win32_resource, "zig rc terminated unexpectedly", .{});
6275 },6273 },
6276 }6274 }
tools/docgen.zig-1
...@@ -3,7 +3,6 @@ const builtin = @import("builtin");...@@ -3,7 +3,6 @@ const builtin = @import("builtin");
3const io = std.io;3const io = std.io;
4const fs = std.fs;4const fs = std.fs;
5const process = std.process;5const process = std.process;
6const ChildProcess = std.process.Child;
7const Progress = std.Progress;6const Progress = std.Progress;
8const print = std.debug.print;7const print = std.debug.print;
9const mem = std.mem;8const mem = std.mem;
tools/incr-check.zig+24-39
...@@ -186,7 +186,7 @@ pub fn main() !void {...@@ -186,7 +186,7 @@ pub fn main() !void {
186186
187 try child.spawn();187 try child.spawn();
188188
189 var poller = std.io.poll(arena, Eval.StreamEnum, .{189 var poller = std.Io.poll(arena, Eval.StreamEnum, .{
190 .stdout = child.stdout.?,190 .stdout = child.stdout.?,
191 .stderr = child.stderr.?,191 .stderr = child.stderr.?,
192 });192 });
...@@ -247,19 +247,15 @@ const Eval = struct {...@@ -247,19 +247,15 @@ const Eval = struct {
247247
248 fn check(eval: *Eval, poller: *Poller, update: Case.Update, prog_node: std.Progress.Node) !void {248 fn check(eval: *Eval, poller: *Poller, update: Case.Update, prog_node: std.Progress.Node) !void {
249 const arena = eval.arena;249 const arena = eval.arena;
250 const Header = std.zig.Server.Message.Header;250 const stdout = poller.reader(.stdout);
251 const stdout = poller.fifo(.stdout);251 const stderr = poller.reader(.stderr);
252 const stderr = poller.fifo(.stderr);
253252
254 poll: while (true) {253 poll: while (true) {
255 while (stdout.readableLength() < @sizeOf(Header)) {254 const Header = std.zig.Server.Message.Header;
256 if (!(try poller.poll())) break :poll;255 while (stdout.buffered().len < @sizeOf(Header)) if (!try poller.poll()) break :poll;
257 }256 const header = stdout.takeStruct(Header, .little) catch unreachable;
258 const header = stdout.reader().readStruct(Header) catch unreachable;257 while (stdout.buffered().len < header.bytes_len) if (!try poller.poll()) break :poll;
259 while (stdout.readableLength() < header.bytes_len) {258 const body = stdout.take(header.bytes_len) catch unreachable;
260 if (!(try poller.poll())) break :poll;
261 }
262 const body = stdout.readableSliceOfLen(header.bytes_len);
263259
264 switch (header.tag) {260 switch (header.tag) {
265 .error_bundle => {261 .error_bundle => {
...@@ -277,8 +273,8 @@ const Eval = struct {...@@ -277,8 +273,8 @@ const Eval = struct {
277 .string_bytes = try arena.dupe(u8, string_bytes),273 .string_bytes = try arena.dupe(u8, string_bytes),
278 .extra = extra_array,274 .extra = extra_array,
279 };275 };
280 if (stderr.readableLength() > 0) {276 if (stderr.bufferedLen() > 0) {
281 const stderr_data = try stderr.toOwnedSlice();277 const stderr_data = try poller.toOwnedSlice(.stderr);
282 if (eval.allow_stderr) {278 if (eval.allow_stderr) {
283 std.log.info("error_bundle included stderr:\n{s}", .{stderr_data});279 std.log.info("error_bundle included stderr:\n{s}", .{stderr_data});
284 } else {280 } else {
...@@ -289,15 +285,14 @@ const Eval = struct {...@@ -289,15 +285,14 @@ const Eval = struct {
289 try eval.checkErrorOutcome(update, result_error_bundle);285 try eval.checkErrorOutcome(update, result_error_bundle);
290 }286 }
291 // This message indicates the end of the update.287 // This message indicates the end of the update.
292 stdout.discard(body.len);
293 return;288 return;
294 },289 },
295 .emit_digest => {290 .emit_digest => {
296 const EbpHdr = std.zig.Server.Message.EmitDigest;291 const EbpHdr = std.zig.Server.Message.EmitDigest;
297 const ebp_hdr = @as(*align(1) const EbpHdr, @ptrCast(body));292 const ebp_hdr = @as(*align(1) const EbpHdr, @ptrCast(body));
298 _ = ebp_hdr;293 _ = ebp_hdr;
299 if (stderr.readableLength() > 0) {294 if (stderr.bufferedLen() > 0) {
300 const stderr_data = try stderr.toOwnedSlice();295 const stderr_data = try poller.toOwnedSlice(.stderr);
301 if (eval.allow_stderr) {296 if (eval.allow_stderr) {
302 std.log.info("emit_digest included stderr:\n{s}", .{stderr_data});297 std.log.info("emit_digest included stderr:\n{s}", .{stderr_data});
303 } else {298 } else {
...@@ -308,7 +303,6 @@ const Eval = struct {...@@ -308,7 +303,6 @@ const Eval = struct {
308 if (eval.target.backend == .sema) {303 if (eval.target.backend == .sema) {
309 try eval.checkSuccessOutcome(update, null, prog_node);304 try eval.checkSuccessOutcome(update, null, prog_node);
310 // This message indicates the end of the update.305 // This message indicates the end of the update.
311 stdout.discard(body.len);
312 }306 }
313307
314 const digest = body[@sizeOf(EbpHdr)..][0..Cache.bin_digest_len];308 const digest = body[@sizeOf(EbpHdr)..][0..Cache.bin_digest_len];
...@@ -323,21 +317,18 @@ const Eval = struct {...@@ -323,21 +317,18 @@ const Eval = struct {
323317
324 try eval.checkSuccessOutcome(update, bin_path, prog_node);318 try eval.checkSuccessOutcome(update, bin_path, prog_node);
325 // This message indicates the end of the update.319 // This message indicates the end of the update.
326 stdout.discard(body.len);
327 },320 },
328 else => {321 else => {
329 // Ignore other messages.322 // Ignore other messages.
330 stdout.discard(body.len);
331 },323 },
332 }324 }
333 }325 }
334326
335 if (stderr.readableLength() > 0) {327 if (stderr.bufferedLen() > 0) {
336 const stderr_data = try stderr.toOwnedSlice();
337 if (eval.allow_stderr) {328 if (eval.allow_stderr) {
338 std.log.info("update '{s}' included stderr:\n{s}", .{ update.name, stderr_data });329 std.log.info("update '{s}' included stderr:\n{s}", .{ update.name, stderr.buffered() });
339 } else {330 } else {
340 eval.fatal("update '{s}' failed:\n{s}", .{ update.name, stderr_data });331 eval.fatal("update '{s}' failed:\n{s}", .{ update.name, stderr.buffered() });
341 }332 }
342 }333 }
343334
...@@ -537,25 +528,19 @@ const Eval = struct {...@@ -537,25 +528,19 @@ const Eval = struct {
537 fn end(eval: *Eval, poller: *Poller) !void {528 fn end(eval: *Eval, poller: *Poller) !void {
538 requestExit(eval.child, eval);529 requestExit(eval.child, eval);
539530
540 const Header = std.zig.Server.Message.Header;531 const stdout = poller.reader(.stdout);
541 const stdout = poller.fifo(.stdout);532 const stderr = poller.reader(.stderr);
542 const stderr = poller.fifo(.stderr);
543533
544 poll: while (true) {534 poll: while (true) {
545 while (stdout.readableLength() < @sizeOf(Header)) {535 const Header = std.zig.Server.Message.Header;
546 if (!(try poller.poll())) break :poll;536 while (stdout.buffered().len < @sizeOf(Header)) if (!try poller.poll()) break :poll;
547 }537 const header = stdout.takeStruct(Header, .little) catch unreachable;
548 const header = stdout.reader().readStruct(Header) catch unreachable;538 while (stdout.buffered().len < header.bytes_len) if (!try poller.poll()) break :poll;
549 while (stdout.readableLength() < header.bytes_len) {539 stdout.toss(header.bytes_len);
550 if (!(try poller.poll())) break :poll;
551 }
552 const body = stdout.readableSliceOfLen(header.bytes_len);
553 stdout.discard(body.len);
554 }540 }
555541
556 if (stderr.readableLength() > 0) {542 if (stderr.bufferedLen() > 0) {
557 const stderr_data = try stderr.toOwnedSlice();543 eval.fatal("unexpected stderr:\n{s}", .{stderr.buffered()});
558 eval.fatal("unexpected stderr:\n{s}", .{stderr_data});
559 }544 }
560 }545 }
561546