Skip to content

Commit b8955a2

Browse files
andrewrkandrewrk
committed
std.Io.poll: update to new I/O API
1 parent bc8e1a7 commit b8955a2

File tree

9 files changed

+387
-380
lines changed

9 files changed

+387
-380
lines changed

‎lib/std/Build/Fuzz/WebServer.zig

Lines changed: 9 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -273,21 +273,17 @@ fn buildWasmBinary(
273273
try sendMessage(child.stdin.?, .update);
274274
try sendMessage(child.stdin.?, .exit);
275275

276-
const Header = std.zig.Server.Message.Header;
277276
var result: ?Path = null;
278277
var result_error_bundle = std.zig.ErrorBundle.empty;
279278

280-
const stdout = poller.fifo(.stdout);
279+
const stdout = poller.reader(.stdout);
281280

282281
poll: while (true) {
283-
while (stdout.readableLength() < @sizeOf(Header)) {
284-
if (!(try poller.poll())) break :poll;
285-
}
286-
const header = stdout.reader().readStruct(Header) catch unreachable;
287-
while (stdout.readableLength() < header.bytes_len) {
288-
if (!(try poller.poll())) break :poll;
289-
}
290-
const body = stdout.readableSliceOfLen(header.bytes_len);
282+
const Header = std.zig.Server.Message.Header;
283+
while (stdout.buffered().len < @sizeOf(Header)) if (!try poller.poll()) break :poll;
284+
const header = stdout.takeStruct(Header, .little) catch unreachable;
285+
while (stdout.buffered().len < header.bytes_len) if (!try poller.poll()) break :poll;
286+
const body = stdout.take(header.bytes_len) catch unreachable;
291287

292288
switch (header.tag) {
293289
.zig_version => {
@@ -325,15 +321,11 @@ fn buildWasmBinary(
325321
},
326322
else => {}, // ignore other messages
327323
}
328-
329-
stdout.discard(body.len);
330324
}
331325

332-
const stderr = poller.fifo(.stderr);
333-
if (stderr.readableLength() > 0) {
334-
const owned_stderr = try stderr.toOwnedSlice();
335-
defer gpa.free(owned_stderr);
336-
std.debug.print("{s}", .{owned_stderr});
326+
const stderr_contents = try poller.toOwnedSlice(.stderr);
327+
if (stderr_contents.len > 0) {
328+
std.debug.print("{s}", .{stderr_contents});
337329
}
338330

339331
// Send EOF to stdin.

‎lib/std/Build/Step.zig

Lines changed: 25 additions & 34 deletions
Original file line numberDiff line numberDiff line change
@@ -286,7 +286,7 @@ pub fn cast(step: *Step, comptime T: type) ?*T {
286286
}
287287

288288
/// For debugging purposes, prints identifying information about this Step.
289-
pub fn dump(step: *Step, w: *std.io.Writer, tty_config: std.io.tty.Config) void {
289+
pub fn dump(step: *Step, w: *std.Io.Writer, tty_config: std.Io.tty.Config) void {
290290
const debug_info = std.debug.getSelfDebugInfo() catch |err| {
291291
w.print("Unable to dump stack trace: Unable to open debug info: {s}\n", .{
292292
@errorName(err),
@@ -359,7 +359,7 @@ pub fn addError(step: *Step, comptime fmt: []const u8, args: anytype) error{OutO
359359

360360
pub const ZigProcess = struct {
361361
child: std.process.Child,
362-
poller: std.io.Poller(StreamEnum),
362+
poller: std.Io.Poller(StreamEnum),
363363
progress_ipc_fd: if (std.Progress.have_ipc) ?std.posix.fd_t else void,
364364

365365
pub const StreamEnum = enum { stdout, stderr };
@@ -428,7 +428,7 @@ pub fn evalZigProcess(
428428
const zp = try gpa.create(ZigProcess);
429429
zp.* = .{
430430
.child = child,
431-
.poller = std.io.poll(gpa, ZigProcess.StreamEnum, .{
431+
.poller = std.Io.poll(gpa, ZigProcess.StreamEnum, .{
432432
.stdout = child.stdout.?,
433433
.stderr = child.stderr.?,
434434
}),
@@ -508,20 +508,16 @@ fn zigProcessUpdate(s: *Step, zp: *ZigProcess, watch: bool) !?Path {
508508
try sendMessage(zp.child.stdin.?, .update);
509509
if (!watch) try sendMessage(zp.child.stdin.?, .exit);
510510

511-
const Header = std.zig.Server.Message.Header;
512511
var result: ?Path = null;
513512

514-
const stdout = zp.poller.fifo(.stdout);
513+
const stdout = zp.poller.reader(.stdout);
515514

516515
poll: while (true) {
517-
while (stdout.readableLength() < @sizeOf(Header)) {
518-
if (!(try zp.poller.poll())) break :poll;
519-
}
520-
const header = stdout.reader().readStruct(Header) catch unreachable;
521-
while (stdout.readableLength() < header.bytes_len) {
522-
if (!(try zp.poller.poll())) break :poll;
523-
}
524-
const body = stdout.readableSliceOfLen(header.bytes_len);
516+
const Header = std.zig.Server.Message.Header;
517+
while (stdout.buffered().len < @sizeOf(Header)) if (!try zp.poller.poll()) break :poll;
518+
const header = stdout.takeStruct(Header, .little) catch unreachable;
519+
while (stdout.buffered().len < header.bytes_len) if (!try zp.poller.poll()) break :poll;
520+
const body = stdout.take(header.bytes_len) catch unreachable;
525521

526522
switch (header.tag) {
527523
.zig_version => {
@@ -547,11 +543,8 @@ fn zigProcessUpdate(s: *Step, zp: *ZigProcess, watch: bool) !?Path {
547543
.string_bytes = try arena.dupe(u8, string_bytes),
548544
.extra = extra_array,
549545
};
550-
if (watch) {
551-
// This message indicates the end of the update.
552-
stdout.discard(body.len);
553-
break;
554-
}
546+
// This message indicates the end of the update.
547+
if (watch) break :poll;
555548
},
556549
.emit_digest => {
557550
const EmitDigest = std.zig.Server.Message.EmitDigest;
@@ -611,15 +604,13 @@ fn zigProcessUpdate(s: *Step, zp: *ZigProcess, watch: bool) !?Path {
611604
},
612605
else => {}, // ignore other messages
613606
}
614-
615-
stdout.discard(body.len);
616607
}
617608

618609
s.result_duration_ns = timer.read();
619610

620-
const stderr = zp.poller.fifo(.stderr);
621-
if (stderr.readableLength() > 0) {
622-
try s.result_error_msgs.append(arena, try stderr.toOwnedSlice());
611+
const stderr_contents = try zp.poller.toOwnedSlice(.stderr);
612+
if (stderr_contents.len > 0) {
613+
try s.result_error_msgs.append(arena, try arena.dupe(u8, stderr_contents));
623614
}
624615

625616
return result;
@@ -736,7 +727,7 @@ pub fn allocPrintCmd2(
736727
argv: []const []const u8,
737728
) Allocator.Error![]u8 {
738729
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 {
740731
for (string) |c| {
741732
if (switch (c) {
742733
else => true,
@@ -770,9 +761,9 @@ pub fn allocPrintCmd2(
770761
}
771762
};
772763

773-
var buf: std.ArrayListUnmanaged(u8) = .empty;
774-
const writer = buf.writer(arena);
775-
if (opt_cwd) |cwd| try writer.print("cd {s} && ", .{cwd});
764+
var aw: std.Io.Writer.Allocating = .init(arena);
765+
const writer = &aw.writer;
766+
if (opt_cwd) |cwd| writer.print("cd {s} && ", .{cwd}) catch return error.OutOfMemory;
776767
if (opt_env) |env| {
777768
const process_env_map = std.process.getEnvMap(arena) catch std.process.EnvMap.init(arena);
778769
var it = env.iterator();
@@ -782,17 +773,17 @@ pub fn allocPrintCmd2(
782773
if (process_env_map.get(key)) |process_value| {
783774
if (std.mem.eql(u8, value, process_value)) continue;
784775
}
785-
try writer.print("{s}=", .{key});
786-
try shell.escape(writer, value, false);
787-
try writer.writeByte(' ');
776+
writer.print("{s}=", .{key}) catch return error.OutOfMemory;
777+
shell.escape(writer, value, false) catch return error.OutOfMemory;
778+
writer.writeByte(' ') catch return error.OutOfMemory;
788779
}
789780
}
790-
try shell.escape(writer, argv[0], true);
781+
shell.escape(writer, argv[0], true) catch return error.OutOfMemory;
791782
for (argv[1..]) |arg| {
792-
try writer.writeByte(' ');
793-
try shell.escape(writer, arg, false);
783+
writer.writeByte(' ') catch return error.OutOfMemory;
784+
shell.escape(writer, arg, false) catch return error.OutOfMemory;
794785
}
795-
return buf.toOwnedSlice(arena);
786+
return aw.toOwnedSlice();
796787
}
797788

798789
/// Prefer `cacheHitAndWatch` unless you already added watch inputs

‎lib/std/Build/Step/Run.zig

Lines changed: 44 additions & 34 deletions
Original file line numberDiff line numberDiff line change
@@ -73,9 +73,12 @@ skip_foreign_checks: bool,
7373
/// external executor (such as qemu) but not fail if the executor is unavailable.
7474
failing_to_execute_foreign_is_an_error: bool,
7575

76+
/// Deprecated in favor of `stdio_limit`.
77+
max_stdio_size: usize,
78+
7679
/// If stderr or stdout exceeds this amount, the child process is killed and
7780
/// the step fails.
78-
max_stdio_size: usize,
81+
stdio_limit: std.Io.Limit,
7982

8083
captured_stdout: ?*Output,
8184
captured_stderr: ?*Output,
@@ -186,6 +189,7 @@ pub fn create(owner: *std.Build, name: []const u8) *Run {
186189
.skip_foreign_checks = false,
187190
.failing_to_execute_foreign_is_an_error = true,
188191
.max_stdio_size = 10 * 1024 * 1024,
192+
.stdio_limit = .unlimited,
189193
.captured_stdout = null,
190194
.captured_stderr = null,
191195
.dep_output_file = null,
@@ -1011,7 +1015,7 @@ fn populateGeneratedPaths(
10111015
}
10121016
}
10131017

1014-
fn formatTerm(term: ?std.process.Child.Term, w: *std.io.Writer) std.io.Writer.Error!void {
1018+
fn formatTerm(term: ?std.process.Child.Term, w: *std.Io.Writer) std.Io.Writer.Error!void {
10151019
if (term) |t| switch (t) {
10161020
.Exited => |code| try w.print("exited with code {d}", .{code}),
10171021
.Signal => |sig| try w.print("terminated with signal {d}", .{sig}),
@@ -1500,7 +1504,7 @@ fn evalZigTest(
15001504
const gpa = run.step.owner.allocator;
15011505
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 }, .{
15041508
.stdout = child.stdout.?,
15051509
.stderr = child.stderr.?,
15061510
});
@@ -1524,11 +1528,6 @@ fn evalZigTest(
15241528
break :failed false;
15251529
};
15261530

1527-
const Header = std.zig.Server.Message.Header;
1528-
1529-
const stdout = poller.fifo(.stdout);
1530-
const stderr = poller.fifo(.stderr);
1531-
15321531
var fail_count: u32 = 0;
15331532
var skip_count: u32 = 0;
15341533
var leak_count: u32 = 0;
@@ -1541,16 +1540,14 @@ fn evalZigTest(
15411540
var sub_prog_node: ?std.Progress.Node = null;
15421541
defer if (sub_prog_node) |n| n.end();
15431542

1543+
const stdout = poller.reader(.stdout);
1544+
const stderr = poller.reader(.stderr);
15441545
const any_write_failed = first_write_failed or poll: while (true) {
1545-
while (stdout.readableLength() < @sizeOf(Header)) {
1546-
if (!(try poller.poll())) break :poll false;
1547-
}
1548-
const header = stdout.reader().readStruct(Header) catch unreachable;
1549-
while (stdout.readableLength() < header.bytes_len) {
1550-
if (!(try poller.poll())) break :poll false;
1551-
}
1552-
const body = stdout.readableSliceOfLen(header.bytes_len);
1553-
1546+
const Header = std.zig.Server.Message.Header;
1547+
while (stdout.buffered().len < @sizeOf(Header)) if (!try poller.poll()) break :poll false;
1548+
const header = stdout.takeStruct(Header, .little) catch unreachable;
1549+
while (stdout.buffered().len < header.bytes_len) if (!try poller.poll()) break :poll false;
1550+
const body = stdout.take(header.bytes_len) catch unreachable;
15541551
switch (header.tag) {
15551552
.zig_version => {
15561553
if (!std.mem.eql(u8, builtin.zig_version_string, body)) {
@@ -1607,9 +1604,9 @@ fn evalZigTest(
16071604

16081605
if (tr_hdr.flags.fail or tr_hdr.flags.leak or tr_hdr.flags.log_err_count > 0) {
16091606
const name = std.mem.sliceTo(md.string_bytes[md.names[tr_hdr.index]..], 0);
1610-
const orig_msg = stderr.readableSlice(0);
1611-
defer stderr.discard(orig_msg.len);
1612-
const msg = std.mem.trim(u8, orig_msg, "\n");
1607+
const stderr_contents = stderr.buffered();
1608+
stderr.toss(stderr_contents.len);
1609+
const msg = std.mem.trim(u8, stderr_contents, "\n");
16131610
const label = if (tr_hdr.flags.fail)
16141611
"failed"
16151612
else if (tr_hdr.flags.leak)
@@ -1660,8 +1657,6 @@ fn evalZigTest(
16601657
},
16611658
else => {}, // ignore other messages
16621659
}
1663-
1664-
stdout.discard(body.len);
16651660
};
16661661

16671662
if (any_write_failed) {
@@ -1670,9 +1665,9 @@ fn evalZigTest(
16701665
while (try poller.poll()) {}
16711666
}
16721667

1673-
if (stderr.readableLength() > 0) {
1674-
const msg = std.mem.trim(u8, try stderr.toOwnedSlice(), "\n");
1675-
if (msg.len > 0) run.step.result_stderr = msg;
1668+
const stderr_contents = std.mem.trim(u8, stderr.buffered(), "\n");
1669+
if (stderr_contents.len > 0) {
1670+
run.step.result_stderr = try arena.dupe(u8, stderr_contents);
16761671
}
16771672

16781673
// Send EOF to stdin.
@@ -1795,28 +1790,43 @@ fn evalGeneric(run: *Run, child: *std.process.Child) !StdIoResult {
17951790
var stdout_bytes: ?[]const u8 = null;
17961791
var stderr_bytes: ?[]const u8 = null;
17971792

1793+
run.stdio_limit = run.stdio_limit.min(.limited(run.max_stdio_size));
17981794
if (child.stdout) |stdout| {
17991795
if (child.stderr) |stderr| {
1800-
var poller = std.io.poll(arena, enum { stdout, stderr }, .{
1796+
var poller = std.Io.poll(arena, enum { stdout, stderr }, .{
18011797
.stdout = stdout,
18021798
.stderr = stderr,
18031799
});
18041800
defer poller.deinit();
18051801

18061802
while (try poller.poll()) {
1807-
if (poller.fifo(.stdout).count > run.max_stdio_size)
1808-
return error.StdoutStreamTooLong;
1809-
if (poller.fifo(.stderr).count > run.max_stdio_size)
1810-
return error.StderrStreamTooLong;
1803+
if (run.stdio_limit.toInt()) |limit| {
1804+
if (poller.reader(.stderr).buffered().len > limit)
1805+
return error.StdoutStreamTooLong;
1806+
if (poller.reader(.stderr).buffered().len > limit)
1807+
return error.StderrStreamTooLong;
1808+
}
18111809
}
18121810

1813-
stdout_bytes = try poller.fifo(.stdout).toOwnedSlice();
1814-
stderr_bytes = try poller.fifo(.stderr).toOwnedSlice();
1811+
stdout_bytes = try poller.toOwnedSlice(.stdout);
1812+
stderr_bytes = try poller.toOwnedSlice(.stderr);
18151813
} 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+
};
18171821
}
18181822
} 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+
};
18201830
}
18211831

18221832
if (stderr_bytes) |bytes| if (bytes.len > 0) {

0 commit comments

Comments
 (0)