authorgravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2024-05-23 14:10:03-07:00
committergravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2024-05-27 20:56:48-07:00
logf07116404ae323efceb57cc48459f62e7a4d6f81
treec85fe6443fe00f885e7cbf0abeeda8931f68da81
parented36470af1c71a254bbe535bf70220aa27370f89

std.Progress: child process sends updates via IPC

3 files changed, 241 insertions(+), 73 deletions(-)

lib/std/Progress.zig+127-56
......@@ -74,7 +74,7 @@ pub const Options = struct {
7474pub const Node = struct {
7575 index: OptionalIndex,
7676
77 pub const max_name_len = 38;
77 pub const max_name_len = 40;
7878
7979 const Storage = extern struct {
8080 /// Little endian.
......@@ -268,17 +268,7 @@ var node_freelist_buffer: [default_node_storage_buffer_len]Node.OptionalIndex =
268268pub fn start(options: Options) Node {
269269 // Ensure there is only 1 global Progress object.
270270 assert(global_progress.node_end_index == 0);
271 const stderr = std.io.getStdErr();
272 if (stderr.supportsAnsiEscapeCodes()) {
273 global_progress.terminal = stderr;
274 global_progress.supports_ansi_escape_codes = true;
275 } else if (builtin.os.tag == .windows and stderr.isTty()) {
276 global_progress.is_windows_terminal = true;
277 global_progress.terminal = stderr;
278 } else if (builtin.os.tag != .windows) {
279 // we are in a "dumb" terminal like in acme or writing to a file
280 global_progress.terminal = stderr;
281 }
271
282272 @memset(global_progress.node_parents, .unused);
283273 const root_node = Node.init(@enumFromInt(0), .none, options.root_name, options.estimated_total_items);
284274 global_progress.done = false;
......@@ -289,22 +279,51 @@ pub fn start(options: Options) Node {
289279 global_progress.refresh_rate_ns = options.refresh_rate_ns;
290280 global_progress.initial_delay_ns = options.initial_delay_ns;
291281
292 var act: posix.Sigaction = .{
293 .handler = .{ .sigaction = handleSigWinch },
294 .mask = posix.empty_sigset,
295 .flags = (posix.SA.SIGINFO | posix.SA.RESTART),
296 };
297 posix.sigaction(posix.SIG.WINCH, &act, null) catch {
298 global_progress.terminal = null;
299 return root_node;
300 };
301
302 if (global_progress.terminal != null) {
303 if (std.Thread.spawn(.{}, updateThreadRun, .{})) |thread| {
282 if (std.process.parseEnvVarInt("ZIG_PROGRESS", u31, 10)) |ipc_fd| {
283 if (std.Thread.spawn(.{}, ipcThreadRun, .{ipc_fd})) |thread| {
304284 global_progress.update_thread = thread;
305 } else |_| {
306 global_progress.terminal = null;
285 } else |err| {
286 std.log.warn("failed to spawn IPC thread for communicating progress to parent: {s}", .{@errorName(err)});
287 return .{ .index = .none };
307288 }
289 } else |env_err| switch (env_err) {
290 error.EnvironmentVariableNotFound => {
291 const stderr = std.io.getStdErr();
292 if (stderr.supportsAnsiEscapeCodes()) {
293 global_progress.terminal = stderr;
294 global_progress.supports_ansi_escape_codes = true;
295 } else if (builtin.os.tag == .windows and stderr.isTty()) {
296 global_progress.is_windows_terminal = true;
297 global_progress.terminal = stderr;
298 } else if (builtin.os.tag != .windows) {
299 // we are in a "dumb" terminal like in acme or writing to a file
300 global_progress.terminal = stderr;
301 }
302
303 if (global_progress.terminal == null) {
304 return .{ .index = .none };
305 }
306
307 var act: posix.Sigaction = .{
308 .handler = .{ .sigaction = handleSigWinch },
309 .mask = posix.empty_sigset,
310 .flags = (posix.SA.SIGINFO | posix.SA.RESTART),
311 };
312 posix.sigaction(posix.SIG.WINCH, &act, null) catch |err| {
313 std.log.warn("failed to install SIGWINCH signal handler for noticing terminal resizes: {s}", .{@errorName(err)});
314 };
315
316 if (std.Thread.spawn(.{}, updateThreadRun, .{})) |thread| {
317 global_progress.update_thread = thread;
318 } else |err| {
319 std.log.warn("unable to spawn thread for printing progress to terminal: {s}", .{@errorName(err)});
320 return .{ .index = .none };
321 }
322 },
323 else => |e| {
324 std.log.warn("invalid ZIG_PROGRESS file descriptor integer: {s}", .{@errorName(e)});
325 return .{ .index = .none };
326 },
308327 }
309328
310329 return root_node;
......@@ -326,12 +345,10 @@ fn updateThreadRun() void {
326345 const resize_flag = wait(global_progress.initial_delay_ns);
327346 maybeUpdateSize(resize_flag);
328347
329 const buffer = b: {
330 if (@atomicLoad(bool, &global_progress.done, .seq_cst))
331 return clearTerminal();
348 if (@atomicLoad(bool, &global_progress.done, .seq_cst))
349 return clearTerminal();
332350
333 break :b computeRedraw();
334 };
351 const buffer = computeRedraw();
335352 write(buffer);
336353 }
337354
......@@ -339,16 +356,36 @@ fn updateThreadRun() void {
339356 const resize_flag = wait(global_progress.refresh_rate_ns);
340357 maybeUpdateSize(resize_flag);
341358
342 const buffer = b: {
343 if (@atomicLoad(bool, &global_progress.done, .seq_cst))
344 return clearTerminal();
359 if (@atomicLoad(bool, &global_progress.done, .seq_cst))
360 return clearTerminal();
345361
346 break :b computeRedraw();
347 };
362 const buffer = computeRedraw();
348363 write(buffer);
349364 }
350365}
351366
367fn ipcThreadRun(fd: posix.fd_t) void {
368 {
369 _ = wait(global_progress.initial_delay_ns);
370
371 if (@atomicLoad(bool, &global_progress.done, .seq_cst))
372 return;
373
374 const serialized = serialize();
375 writeIpc(fd, serialized);
376 }
377
378 while (true) {
379 _ = wait(global_progress.refresh_rate_ns);
380
381 if (@atomicLoad(bool, &global_progress.done, .seq_cst))
382 return clearTerminal();
383
384 const serialized = serialize();
385 writeIpc(fd, serialized);
386 }
387}
388
352389const start_sync = "\x1b[?2026h";
353390const up_one_line = "\x1bM";
354391const clear = "\x1b[J";
......@@ -400,11 +437,17 @@ const Children = struct {
400437 sibling: Node.OptionalIndex,
401438};
402439
403fn computeRedraw() []u8 {
404 // TODO make this configurable
405 var serialized_node_parents_buffer: [default_node_storage_buffer_len]Node.Parent = undefined;
406 var serialized_node_storage_buffer: [default_node_storage_buffer_len]Node.Storage = undefined;
407 var serialized_node_map_buffer: [default_node_storage_buffer_len]Node.Index = undefined;
440// TODO make this configurable
441var serialized_node_parents_buffer: [default_node_storage_buffer_len]Node.Parent = undefined;
442var serialized_node_storage_buffer: [default_node_storage_buffer_len]Node.Storage = undefined;
443var serialized_node_map_buffer: [default_node_storage_buffer_len]Node.Index = undefined;
444
445const Serialized = struct {
446 parents: []Node.Parent,
447 storage: []Node.Storage,
448};
449
450fn serialize() Serialized {
408451 var serialized_len: usize = 0;
409452
410453 // Iterate all of the nodes and construct a serializable copy of the state that can be examined
......@@ -447,12 +490,21 @@ fn computeRedraw() []u8 {
447490 };
448491 }
449492
493 return .{
494 .parents = serialized_node_parents,
495 .storage = serialized_node_storage,
496 };
497}
498
499fn computeRedraw() []u8 {
500 const serialized = serialize();
501
450502 var children_buffer: [default_node_storage_buffer_len]Children = undefined;
451 const children = children_buffer[0..serialized_len];
503 const children = children_buffer[0..serialized.parents.len];
452504
453505 @memset(children, .{ .child = .none, .sibling = .none });
454506
455 for (serialized_node_parents, 0..) |parent, child_index_usize| {
507 for (serialized.parents, 0..) |parent, child_index_usize| {
456508 const child_index: Node.Index = @enumFromInt(child_index_usize);
457509 assert(parent != .unused);
458510 const parent_index = parent.unwrap() orelse continue;
......@@ -478,7 +530,7 @@ fn computeRedraw() []u8 {
478530 i = computeClear(buf, i);
479531
480532 const root_node_index: Node.Index = @enumFromInt(0);
481 i = computeNode(buf, i, serialized_node_storage, serialized_node_parents, children, root_node_index);
533 i = computeNode(buf, i, serialized, children, root_node_index);
482534
483535 // Truncate trailing newline.
484536 if (buf[i - 1] == '\n') i -= 1;
......@@ -492,15 +544,14 @@ fn computeRedraw() []u8 {
492544fn computePrefix(
493545 buf: []u8,
494546 start_i: usize,
495 serialized_node_storage: []const Node.Storage,
496 serialized_node_parents: []const Node.Parent,
547 serialized: Serialized,
497548 children: []const Children,
498549 node_index: Node.Index,
499550) usize {
500551 var i = start_i;
501 const parent_index = serialized_node_parents[@intFromEnum(node_index)].unwrap() orelse return i;
502 if (serialized_node_parents[@intFromEnum(parent_index)] == .none) return i;
503 i = computePrefix(buf, i, serialized_node_storage, serialized_node_parents, children, parent_index);
552 const parent_index = serialized.parents[@intFromEnum(node_index)].unwrap() orelse return i;
553 if (serialized.parents[@intFromEnum(parent_index)] == .none) return i;
554 i = computePrefix(buf, i, serialized, children, parent_index);
504555 if (children[@intFromEnum(parent_index)].sibling == .none) {
505556 buf[i..][0..3].* = " ".*;
506557 i += 3;
......@@ -514,19 +565,18 @@ fn computePrefix(
514565fn computeNode(
515566 buf: []u8,
516567 start_i: usize,
517 serialized_node_storage: []const Node.Storage,
518 serialized_node_parents: []const Node.Parent,
568 serialized: Serialized,
519569 children: []const Children,
520570 node_index: Node.Index,
521571) usize {
522572 var i = start_i;
523 i = computePrefix(buf, i, serialized_node_storage, serialized_node_parents, children, node_index);
573 i = computePrefix(buf, i, serialized, children, node_index);
524574
525 const storage = &serialized_node_storage[@intFromEnum(node_index)];
575 const storage = &serialized.storage[@intFromEnum(node_index)];
526576 const estimated_total = storage.estimated_total_count;
527577 const completed_items = storage.completed_count;
528578 const name = if (std.mem.indexOfScalar(u8, &storage.name, 0)) |end| storage.name[0..end] else &storage.name;
529 const parent = serialized_node_parents[@intFromEnum(node_index)];
579 const parent = serialized.parents[@intFromEnum(node_index)];
530580
531581 if (parent != .none) {
532582 if (children[@intFromEnum(node_index)].sibling == .none) {
......@@ -555,11 +605,11 @@ fn computeNode(
555605 global_progress.newline_count += 1;
556606
557607 if (children[@intFromEnum(node_index)].child.unwrap()) |child| {
558 i = computeNode(buf, i, serialized_node_storage, serialized_node_parents, children, child);
608 i = computeNode(buf, i, serialized, children, child);
559609 }
560610
561611 if (children[@intFromEnum(node_index)].sibling.unwrap()) |sibling| {
562 i = computeNode(buf, i, serialized_node_storage, serialized_node_parents, children, sibling);
612 i = computeNode(buf, i, serialized, children, sibling);
563613 }
564614
565615 return i;
......@@ -572,6 +622,27 @@ fn write(buf: []const u8) void {
572622 };
573623}
574624
625fn writeIpc(fd: posix.fd_t, serialized: Serialized) void {
626 assert(serialized.parents.len == serialized.storage.len);
627 const header = std.mem.asBytes(&serialized.parents.len);
628 const storage = std.mem.sliceAsBytes(serialized.storage);
629 const parents = std.mem.sliceAsBytes(serialized.parents);
630
631 var vecs: [3]std.posix.iovec_const = .{
632 .{ .base = header.ptr, .len = header.len },
633 .{ .base = storage.ptr, .len = storage.len },
634 .{ .base = parents.ptr, .len = parents.len },
635 };
636
637 // TODO: if big endian, byteswap
638 // this is needed because the parent or child process might be running in qemu
639
640 const file: std.fs.File = .{ .handle = fd };
641 file.writevAll(&vecs) catch |err| {
642 std.log.warn("failed to send progress to parent process: {s}", .{@errorName(err)});
643 };
644}
645
575646fn maybeUpdateSize(resize_flag: bool) void {
576647 if (!resize_flag) return;
577648
lib/std/process.zig+70-10
......@@ -431,6 +431,29 @@ pub fn hasEnvVarConstant(comptime key: []const u8) bool {
431431 }
432432}
433433
434pub const ParseEnvVarIntError = std.fmt.ParseIntError || error{EnvironmentVariableNotFound};
435
436/// Parses an environment variable as an integer.
437///
438/// Since the key is comptime-known, no allocation is needed.
439///
440/// On Windows, `key` must be valid UTF-8.
441pub fn parseEnvVarInt(comptime key: []const u8, comptime I: type, base: u8) ParseEnvVarIntError!I {
442 if (native_os == .windows) {
443 const key_w = comptime std.unicode.utf8ToUtf16LeStringLiteral(key);
444 const text = getenvW(key_w) orelse return error.EnvironmentVariableNotFound;
445 // For this implementation perhaps std.fmt.parseInt can be expanded to be generic across
446 // []u8 and []u16 like how many std.mem functions work.
447 _ = text;
448 @compileError("TODO implement this");
449 } else if (native_os == .wasi and !builtin.link_libc) {
450 @compileError("parseEnvVarInt is not supported for WASI without libc");
451 } else {
452 const text = posix.getenv(key) orelse return error.EnvironmentVariableNotFound;
453 return std.fmt.parseInt(I, text, base);
454 }
455}
456
434457pub const HasEnvVarError = error{
435458 OutOfMemory,
436459
......@@ -1790,24 +1813,61 @@ test raiseFileDescriptorLimit {
17901813 raiseFileDescriptorLimit();
17911814}
17921815
1793pub fn createNullDelimitedEnvMap(arena: mem.Allocator, env_map: *const EnvMap) ![:null]?[*:0]u8 {
1794 const envp_count = env_map.count();
1816pub const CreateEnvironOptions = struct {
1817 env_map: ?*const EnvMap = null,
1818 existing: ?[*:null]const ?[*:0]const u8 = null,
1819 extra_usizes: []const ExtraUsize = &.{},
1820
1821 pub const ExtraUsize = struct {
1822 name: []const u8,
1823 value: usize,
1824 };
1825};
1826
1827/// Creates a null-deliminated environment variable block in the format
1828/// expected by POSIX, by combining all the sources of key-value pairs together
1829/// from `options`.
1830pub fn createEnviron(arena: Allocator, options: CreateEnvironOptions) Allocator.Error![:null]?[*:0]u8 {
1831 const envp_count = c: {
1832 var count: usize = 0;
1833 if (options.existing) |env| {
1834 while (env[count]) |_| : (count += 1) {}
1835 }
1836 if (options.env_map) |env_map| {
1837 count += env_map.count();
1838 }
1839 count += options.extra_usizes.len;
1840 break :c count;
1841 };
17951842 const envp_buf = try arena.allocSentinel(?[*:0]u8, envp_count, null);
1796 {
1843 var i: usize = 0;
1844
1845 if (options.existing) |env| {
1846 while (env[i]) |line| : (i += 1) {
1847 envp_buf[i] = try arena.dupeZ(u8, mem.span(line));
1848 }
1849 }
1850
1851 for (options.extra_usizes, envp_buf[i..][0..options.extra_usizes.len]) |extra_usize, *out| {
1852 out.* = try std.fmt.allocPrintZ(arena, "{s}={d}", .{ extra_usize.name, extra_usize.value });
1853 }
1854 i += options.extra_usizes.len;
1855
1856 if (options.env_map) |env_map| {
17971857 var it = env_map.iterator();
1798 var i: usize = 0;
17991858 while (it.next()) |pair| : (i += 1) {
1800 const env_buf = try arena.allocSentinel(u8, pair.key_ptr.len + pair.value_ptr.len + 1, 0);
1801 @memcpy(env_buf[0..pair.key_ptr.len], pair.key_ptr.*);
1802 env_buf[pair.key_ptr.len] = '=';
1803 @memcpy(env_buf[pair.key_ptr.len + 1 ..][0..pair.value_ptr.len], pair.value_ptr.*);
1804 envp_buf[i] = env_buf.ptr;
1859 envp_buf[i] = try std.fmt.allocPrintZ(arena, "{s}={s}", .{ pair.key_ptr.*, pair.value_ptr.* });
18051860 }
1806 assert(i == envp_count);
18071861 }
1862
1863 assert(i == envp_count);
18081864 return envp_buf;
18091865}
18101866
1867pub fn createNullDelimitedEnvMap(arena: mem.Allocator, env_map: *const EnvMap) ![:null]?[*:0]u8 {
1868 return createEnviron(arena, .{ .env_map = env_map });
1869}
1870
18111871test createNullDelimitedEnvMap {
18121872 const allocator = testing.allocator;
18131873 var envmap = EnvMap.init(allocator);
lib/std/process/Child.zig+44-7
......@@ -12,6 +12,7 @@ const EnvMap = std.process.EnvMap;
1212const maxInt = std.math.maxInt;
1313const assert = std.debug.assert;
1414const native_os = builtin.os.tag;
15const Allocator = std.mem.Allocator;
1516const ChildProcess = @This();
1617
1718pub const Id = switch (native_os) {
......@@ -92,6 +93,13 @@ request_resource_usage_statistics: bool = false,
9293/// `spawn`.
9394resource_usage_statistics: ResourceUsageStatistics = .{},
9495
96/// When populated, a pipe will be created for the child process to
97/// communicate progress back to the parent. The file descriptor of the
98/// write end of the pipe will be specified in the `ZIG_PROGRESS`
99/// environment variable inside the child process. The progress reported by
100/// the child will be attached to this progress node in the parent process.
101parent_progress_node: std.Progress.Node = .{ .index = .none },
102
95103pub const ResourceUsageStatistics = struct {
96104 rusage: @TypeOf(rusage_init) = rusage_init,
97105
......@@ -572,6 +580,16 @@ fn spawnPosix(self: *ChildProcess) SpawnError!void {
572580 if (any_ignore) posix.close(dev_null_fd);
573581 }
574582
583 const prog_pipe: [2]posix.fd_t = p: {
584 if (self.parent_progress_node.index == .none) {
585 break :p .{ -1, -1 };
586 } else {
587 // No CLOEXEC because the child needs access to this file descriptor.
588 break :p try posix.pipe2(.{});
589 }
590 };
591 errdefer destroyPipe(prog_pipe);
592
575593 var arena_allocator = std.heap.ArenaAllocator.init(self.allocator);
576594 defer arena_allocator.deinit();
577595 const arena = arena_allocator.allocator();
......@@ -588,16 +606,35 @@ fn spawnPosix(self: *ChildProcess) SpawnError!void {
588606 const argv_buf = try arena.allocSentinel(?[*:0]const u8, self.argv.len, null);
589607 for (self.argv, 0..) |arg, i| argv_buf[i] = (try arena.dupeZ(u8, arg)).ptr;
590608
591 const envp = m: {
609 const envp: [*:null]const ?[*:0]const u8 = m: {
610 const extra_usizes: []const process.CreateEnvironOptions.ExtraUsize = if (prog_pipe[1] == -1) &.{} else &.{
611 .{ .name = "ZIG_PROGRESS", .value = @intCast(prog_pipe[1]) },
612 };
592613 if (self.env_map) |env_map| {
593 const envp_buf = try process.createNullDelimitedEnvMap(arena, env_map);
594 break :m envp_buf.ptr;
614 break :m (try process.createEnviron(arena, .{
615 .env_map = env_map,
616 .extra_usizes = extra_usizes,
617 })).ptr;
595618 } else if (builtin.link_libc) {
596 break :m std.c.environ;
619 if (extra_usizes.len == 0) {
620 break :m std.c.environ;
621 } else {
622 break :m (try process.createEnviron(arena, .{
623 .existing = std.c.environ,
624 .extra_usizes = extra_usizes,
625 })).ptr;
626 }
597627 } else if (builtin.output_mode == .Exe) {
598628 // Then we have Zig start code and this works.
599 // TODO type-safety for null-termination of `os.environ`.
600 break :m @as([*:null]const ?[*:0]const u8, @ptrCast(std.os.environ.ptr));
629 if (extra_usizes.len == 0) {
630 break :m @ptrCast(std.os.environ.ptr);
631 } else {
632 break :m (try process.createEnviron(arena, .{
633 // TODO type-safety for null-termination of `os.environ`.
634 .existing = @ptrCast(std.os.environ.ptr),
635 .extra_usizes = extra_usizes,
636 })).ptr;
637 }
601638 } else {
602639 // TODO come up with a solution for this.
603640 @compileError("missing std lib enhancement: ChildProcess implementation has no way to collect the environment variables to forward to the child process");
......@@ -962,7 +999,7 @@ fn setUpChildIo(stdio: StdIo, pipe_fd: i32, std_fileno: i32, dev_null_fd: i32) !
962999}
9631000
9641001fn destroyPipe(pipe: [2]posix.fd_t) void {
965 posix.close(pipe[0]);
1002 if (pipe[0] != -1) posix.close(pipe[0]);
9661003 if (pipe[0] != pipe[1]) posix.close(pipe[1]);
9671004}
9681005