Implement Progress IPC for Windows (#20105)

skyfex · skyfex · commit cb1d9680cd59 · 2024-11-16T20:34:43.000+01:00
diff --git a/lib/std/Progress.zig b/lib/std/Progress.zig
@@ -103,9 +103,12 @@ pub const Node = struct {
 
         /// Thread-safe.
         fn setIpcFd(s: *Storage, fd: posix.fd_t) void {
+            // NOTE: On Windows we end up casting a HANDLE (potential 64-bit opaque value) to i32,
+            // but it seems the handle values are actually small values, so OK for now, but should perhaps
+            // be handled better
             const integer: u32 = switch (@typeInfo(posix.fd_t)) {
                 .int => @bitCast(fd),
-                .pointer => @intFromPtr(fd),
+                .pointer => @intCast(@intFromPtr(fd)),
                 else => @compileError("unsupported fd_t of " ++ @typeName(posix.fd_t)),
             };
             // `estimated_total_count` max int indicates the special state that
@@ -263,10 +266,12 @@ pub const Node = struct {
     /// Posix-only. Used by `std.process.Child`. Thread-safe.
     pub fn setIpcFd(node: Node, fd: posix.fd_t) void {
         const index = node.index.unwrap() orelse return;
-        assert(fd >= 0);
-        assert(fd != posix.STDOUT_FILENO);
-        assert(fd != posix.STDIN_FILENO);
-        assert(fd != posix.STDERR_FILENO);
+        if (!is_windows) {
+            assert(fd >= 0);
+            assert(fd != posix.STDOUT_FILENO);
+            assert(fd != posix.STDIN_FILENO);
+            assert(fd != posix.STDERR_FILENO);
+        }
         storageByIndex(index).setIpcFd(fd);
     }
 
@@ -379,7 +384,8 @@ pub fn start(options: Options) Node {
     if (noop_impl)
         return Node.none;
 
-    if (std.process.parseEnvVarInt("ZIG_PROGRESS", u31, 10)) |ipc_fd| {
+    const zp_env_type = if (is_windows) usize else u31;
+    if (std.process.parseEnvVarInt("ZIG_PROGRESS", zp_env_type, 10)) |ipc_fd| {
         global_progress.update_thread = std.Thread.spawn(.{}, ipcThreadRun, .{
             @as(posix.fd_t, switch (@typeInfo(posix.fd_t)) {
                 .int => ipc_fd,
@@ -890,6 +896,7 @@ fn serializeIpc(start_serialized_len: usize, serialized_buffer: *Serialized.Buff
             }
             bytes_read += n;
         }
+
         // Ignore all but the last message on the pipe.
         var input: []u8 = pipe_buf[0..bytes_read];
         if (input.len == 0) {
diff --git a/lib/std/os/windows.zig b/lib/std/os/windows.zig
@@ -631,6 +631,7 @@ pub fn ReadFile(in_hFile: HANDLE, buffer: []u8, offset: ?u64) ReadFileError!usiz
                 .OPERATION_ABORTED => continue,
                 .BROKEN_PIPE => return 0,
                 .HANDLE_EOF => return 0,
+                .NO_DATA => return 0,
                 .NETNAME_DELETED => return error.ConnectionResetByPeer,
                 .LOCK_VIOLATION => return error.LockViolation,
                 else => |err| return unexpectedError(err),
diff --git a/lib/std/process.zig b/lib/std/process.zig
@@ -1831,7 +1831,7 @@ pub const CreateEnvironOptions = struct {
     /// `null` means to leave the `ZIG_PROGRESS` environment variable unmodified.
     /// If non-null, negative means to remove the environment variable, and >= 0
     /// means to provide it with the given integer.
-    zig_progress_fd: ?i32 = null,
+    zig_progress_fd: if (native_os == .windows) ?usize else ?i32 = null,
 };
 
 /// Creates a null-delimited environment variable block in the format
@@ -1997,7 +1997,22 @@ test createNullDelimitedEnvMap {
 }
 
 /// Caller must free result.
-pub fn createWindowsEnvBlock(allocator: mem.Allocator, env_map: *const EnvMap) ![]u16 {
+pub fn createWindowsEnvBlock(allocator: mem.Allocator, env_map: *const EnvMap, options: CreateEnvironOptions) ![]u16 {
+    var zig_progress_value: ?[]const u8 = env_map.get("ZIG_PROGRESS");
+
+    const ZigProgressAction = enum { nothing, edit, delete, add };
+    const zig_progress_action: ZigProgressAction = a: {
+        const fd = options.zig_progress_fd orelse break :a .nothing;
+        const contains = zig_progress_value != null;
+        if (fd >= 0) {
+            zig_progress_value = try std.fmt.allocPrintZ(allocator, "{d}", .{fd});
+            break :a if (contains) .edit else .add;
+        } else {
+            if (contains) break :a .delete;
+        }
+        break :a .nothing;
+    };
+
     // count bytes needed
     const max_chars_needed = x: {
         var max_chars_needed: usize = 4; // 4 for the final 4 null bytes
@@ -2007,18 +2022,41 @@ pub fn createWindowsEnvBlock(allocator: mem.Allocator, env_map: *const EnvMap) !
             // +1 for null byte
             max_chars_needed += pair.key_ptr.len + pair.value_ptr.len + 2;
         }
+        switch (zig_progress_action) {
+            .add => max_chars_needed += "ZIG_PROGRESS".len + (zig_progress_value.?).len + 2,
+            .delete, .nothing, .edit => {},
+        }
         break :x max_chars_needed;
     };
     const result = try allocator.alloc(u16, max_chars_needed);
     errdefer allocator.free(result);
 
     var it = env_map.iterator();
     var i: usize = 0;
+
+    if (zig_progress_action == .add) {
+        i += try unicode.wtf8ToWtf16Le(result[i..], "ZIG_PROGRESS");
+        result[i] = '=';
+        i += 1;
+        i += try unicode.wtf8ToWtf16Le(result[i..], zig_progress_value.?);
+        result[i] = 0;
+        i += 1;
+    }
+
     while (it.next()) |pair| {
+        var value = pair.value_ptr.*;
+        if (mem.eql(u8, pair.key_ptr.*, "ZIG_PROGRESS")) switch (zig_progress_action) {
+            .add => unreachable,
+            .delete => continue,
+            .edit => {
+                value = zig_progress_value.?;
+            },
+            .nothing => {},
+        };
         i += try unicode.wtf8ToWtf16Le(result[i..], pair.key_ptr.*);
         result[i] = '=';
         i += 1;
-        i += try unicode.wtf8ToWtf16Le(result[i..], pair.value_ptr.*);
+        i += try unicode.wtf8ToWtf16Le(result[i..], value);
         result[i] = 0;
         i += 1;
     }
diff --git a/lib/std/process/Child.zig b/lib/std/process/Child.zig
@@ -820,6 +820,12 @@ fn spawnWindows(self: *ChildProcess) SpawnError!void {
         windowsDestroyPipe(g_hChildStd_ERR_Rd, g_hChildStd_ERR_Wr);
     };
 
+    var prog_pipe_rd: ?windows.HANDLE = null;
+    var prog_pipe_wr: ?windows.HANDLE = null;
+    if (self.progress_node.index != .none) {
+        try windowsMakeProgressPipe(&prog_pipe_rd, &prog_pipe_wr, &saAttr);
+        errdefer windowsDestroyPipe(prog_pipe_rd, prog_pipe_wr);
+    }
     var siStartInfo = windows.STARTUPINFOW{
         .cb = @sizeOf(windows.STARTUPINFOW),
         .hStdError = g_hChildStd_ERR_Wr,
@@ -847,9 +853,19 @@ fn spawnWindows(self: *ChildProcess) SpawnError!void {
     defer if (cwd_w) |cwd| self.allocator.free(cwd);
     const cwd_w_ptr = if (cwd_w) |cwd| cwd.ptr else null;
 
-    const maybe_envp_buf = if (self.env_map) |env_map| try process.createWindowsEnvBlock(self.allocator, env_map) else null;
-    defer if (maybe_envp_buf) |envp_buf| self.allocator.free(envp_buf);
-    const envp_ptr = if (maybe_envp_buf) |envp_buf| envp_buf.ptr else null;
+    const maybe_envp_buf = m: {
+        const prog_fd: ?usize = if (prog_pipe_wr) |h| @intFromPtr(h) else null;
+        if (self.env_map) |env_map| {
+            break :m try process.createWindowsEnvBlock(self.allocator, env_map, .{ .zig_progress_fd = prog_fd });
+        } else {
+            const env_map = try process.getEnvMap(self.allocator);
+            break :m try process.createWindowsEnvBlock(self.allocator, &env_map, .{ .zig_progress_fd = prog_fd });
+        }
+    };
+    // defer if (maybe_envp_buf) |envp_buf| self.allocator.free(envp_buf);
+    // const envp_ptr = if (maybe_envp_buf) |envp_buf| envp_buf.ptr else null;
+    defer self.allocator.free(maybe_envp_buf);
+    const envp_ptr = maybe_envp_buf.ptr;
 
     const app_name_wtf8 = self.argv[0];
     const app_name_is_absolute = fs.path.isAbsolute(app_name_wtf8);
@@ -998,6 +1014,7 @@ fn spawnWindows(self: *ChildProcess) SpawnError!void {
     if (self.stdout_behavior == StdIo.Pipe) {
         posix.close(g_hChildStd_OUT_Wr.?);
     }
+    if (prog_pipe_rd) |fd| self.progress_node.setIpcFd(fd);
 }
 
 fn setUpChildIo(stdio: StdIo, pipe_fd: i32, std_fileno: i32, dev_null_fd: i32) !void {
@@ -1394,6 +1411,70 @@ fn windowsMakeAsyncPipe(rd: *?windows.HANDLE, wr: *?windows.HANDLE, sattr: *cons
     wr.* = write_handle;
 }
 
+fn windowsMakeProgressPipe(rd: *?windows.HANDLE, wr: *?windows.HANDLE, sattr: *const windows.SECURITY_ATTRIBUTES) !void {
+    var tmp_bufw: [128]u16 = undefined;
+
+    // NOTE: Only difference with windowsMakeAsyncPipe now is windows.PIPE_NOWAIT flag and
+    // pipe name
+
+    // Anonymous pipes are built upon Named pipes.
+    // https://docs.microsoft.com/en-us/windows/win32/api/namedpipeapi/nf-namedpipeapi-createpipe
+    // Asynchronous (overlapped) read and write operations are not supported by anonymous pipes.
+    // https://docs.microsoft.com/en-us/windows/win32/ipc/anonymous-pipe-operations
+    const pipe_path = blk: {
+        var tmp_buf: [128]u8 = undefined;
+        // Forge a random path for the pipe.
+        const pipe_path = std.fmt.bufPrintZ(
+            &tmp_buf,
+            "\\\\.\\pipe\\zig-progress-{d}-{d}",
+            .{ windows.GetCurrentProcessId(), pipe_name_counter.fetchAdd(1, .monotonic) },
+        ) catch unreachable;
+        const len = std.unicode.wtf8ToWtf16Le(&tmp_bufw, pipe_path) catch unreachable;
+        tmp_bufw[len] = 0;
+        break :blk tmp_bufw[0..len :0];
+    };
+
+    // Create the read handle that can be used with overlapped IO ops.
+    const read_handle = windows.kernel32.CreateNamedPipeW(
+        pipe_path.ptr,
+        windows.PIPE_ACCESS_INBOUND | windows.FILE_FLAG_OVERLAPPED,
+        windows.PIPE_TYPE_BYTE | windows.PIPE_NOWAIT,
+        1,
+        4096,
+        4096,
+        0,
+        sattr,
+    );
+    if (read_handle == windows.INVALID_HANDLE_VALUE) {
+        switch (windows.GetLastError()) {
+            else => |err| return windows.unexpectedError(err),
+        }
+    }
+    errdefer posix.close(read_handle);
+
+    var sattr_copy = sattr.*;
+    const write_handle = windows.kernel32.CreateFileW(
+        pipe_path.ptr,
+        windows.GENERIC_WRITE,
+        0,
+        &sattr_copy,
+        windows.OPEN_EXISTING,
+        windows.FILE_ATTRIBUTE_NORMAL,
+        null,
+    );
+    if (write_handle == windows.INVALID_HANDLE_VALUE) {
+        switch (windows.GetLastError()) {
+            else => |err| return windows.unexpectedError(err),
+        }
+    }
+    errdefer posix.close(write_handle);
+
+    try windows.SetHandleInformation(read_handle, windows.HANDLE_FLAG_INHERIT, 0);
+
+    rd.* = read_handle;
+    wr.* = write_handle;
+}
+
 var pipe_name_counter = std.atomic.Value(u32).init(1);
 
 /// File name extensions supported natively by `CreateProcess()` on Windows.
