authorgravatar for johnnymarler@gmail.comJonathan Marler <johnnymarler@gmail.com> 2021-06-17 17:36:42-06:00
committergravatar for johnnymarler@gmail.comJonathan Marler <johnnymarler@gmail.com> 2021-06-18 08:26:22-06:00
log9e0338b82e45f672975bf7daa1ade5f4b2de4c01
tree3fbe11f9e77dabc5dbe089a93f238d15a23a07df
parent1e0d68e6fbfe6d5d759ad773b656c5a5acba7a5e

finish ChildProcess collectOutputWindows

This finishes LemonBoy's Draft PR ziglang#6750. It updates ChildProcess to collect the output from stdout/stderr asynchronously using Overlapped IO and named pipes.

2 files changed, 118 insertions(+), 121 deletions(-)

lib/std/child_process.zig+104-88
...@@ -13,6 +13,7 @@ const process = std.process;...@@ -13,6 +13,7 @@ const process = std.process;
13const File = std.fs.File;13const File = std.fs.File;
14const windows = os.windows;14const windows = os.windows;
15const mem = std.mem;15const mem = std.mem;
16const math = std.math;
16const debug = std.debug;17const debug = std.debug;
17const BufMap = std.BufMap;18const BufMap = std.BufMap;
18const builtin = std.builtin;19const builtin = std.builtin;
...@@ -257,58 +258,76 @@ pub const ChildProcess = struct {...@@ -257,58 +258,76 @@ pub const ChildProcess = struct {
257 }258 }
258 }259 }
259260
260 fn collectOutputWindows(child: *const ChildProcess, stdout: *std.ArrayList(u8), stderr: *std.ArrayList(u8), max_output_bytes: usize) !void {261 fn collectOutputWindows(child: *const ChildProcess, outs: [2]*std.ArrayList(u8), max_output_bytes: usize) !void {
261 var wait_objects = [_]windows.kernel32.HANDLE{262 const bump_amt = 512;
262 child.stdout.?.handle, child.stderr.?.handle,263 const handles = [_]windows.HANDLE{
264 child.stdout.?.handle,
265 child.stderr.?.handle,
263 };266 };
264 var waiting_objects: u32 = wait_objects.len;
265267
266 // XXX: Calling zeroes([2]windows.OVERLAPPED) causes the stage1 compiler
267 // to crash and burn.
268 var overlapped = [_]windows.OVERLAPPED{268 var overlapped = [_]windows.OVERLAPPED{
269 mem.zeroes(windows.OVERLAPPED),269 mem.zeroes(windows.OVERLAPPED),
270 mem.zeroes(windows.OVERLAPPED),270 mem.zeroes(windows.OVERLAPPED),
271 };271 };
272 var temp_buf: [2][4096]u8 = undefined;272
273273 var wait_objects: [2]windows.HANDLE = undefined;
274 // Kickstart the loop by issuing two async reads.274 var wait_object_count: u2 = 0;
275 // ReadFile returns false and GetLastError returns ERROR_IO_PENDING if275
276 // everything is ok.276 // we need to cancel all pending IO before returning so our OVERLAPPED values don't go out of scope
277 _ = windows.kernel32.ReadFile(wait_objects[0], &temp_buf[0], temp_buf[0].len, null, &overlapped[0]);277 defer for (wait_objects[0..wait_object_count]) |o| {
278 _ = windows.kernel32.ReadFile(wait_objects[1], &temp_buf[1], temp_buf[1].len, null, &overlapped[1]);278 _ = windows.kernel32.CancelIo(o);
279279 };
280 poll: while (waiting_objects > 0) {280
281 const status = windows.kernel32.WaitForMultipleObjects(waiting_objects, &wait_objects, 0, windows.INFINITE);281 // Windows Async IO requires an initial call to ReadFile before waiting on the handle
282 switch (status) {282 for ([_]u1{ 0, 1 }) |i| {
283 windows.WAIT_OBJECT_0 + 0...windows.WAIT_OBJECT_0 + 1 => {283 try outs[i].ensureCapacity(bump_amt);
284 // stdout (or stderr) is ready.284 const buf = outs[i].unusedCapacitySlice();
285 const object = status - windows.WAIT_OBJECT_0;285 _ = windows.kernel32.ReadFile(handles[i], buf.ptr, math.cast(u32, buf.len) catch maxInt(u32), null, &overlapped[i]);
286286 wait_objects[wait_object_count] = handles[i];
287 var read_bytes: u32 = undefined;287 wait_object_count += 1;
288 if (windows.kernel32.GetOverlappedResult(wait_objects[object], &overlapped[object], &read_bytes, 0) == 0) {288 }
289 switch (windows.kernel32.GetLastError()) {289
290 .BROKEN_PIPE => {290 while (true) {
291 // Move it to the end to remove it.291 const status = windows.kernel32.WaitForMultipleObjects(wait_object_count, &wait_objects, 0, windows.INFINITE);
292 if (object != waiting_objects - 1)292 if (status == windows.WAIT_FAILED) {
293 mem.swap(windows.kernel32.HANDLE, &wait_objects[object], &wait_objects[waiting_objects - 1]);293 switch (windows.kernel32.GetLastError()) {
294 waiting_objects -= 1;294 else => |err| return windows.unexpectedError(err),
295 continue :poll;295 }
296 },
297 else => |err| return windows.unexpectedError(err),
298 }
299 }
300 try stdout.appendSlice(temp_buf[object][0..read_bytes]);
301 _ = windows.kernel32.ReadFile(wait_objects[object], &temp_buf[object], temp_buf[object].len, null, &overlapped[object]);
302 },
303 windows.WAIT_FAILED => {
304 switch (windows.kernel32.GetLastError()) {
305 else => |err| return windows.unexpectedError(err),
306 }
307 },
308 // We're waiting with an infinite timeout
309 windows.WAIT_TIMEOUT => unreachable,
310 else => unreachable,
311 }296 }
297 if (status < windows.WAIT_OBJECT_0 or status > windows.WAIT_OBJECT_0 + wait_object_count - 1)
298 unreachable;
299
300 const wait_idx = status - windows.WAIT_OBJECT_0;
301
302 // this extra `i` index is needed to map the wait handle back to the stdout or stderr
303 // values since the wait_idx can change which handle it corresponds with
304 const i: u1 = if (wait_objects[wait_idx] == handles[0]) 0 else 1;
305
306 // remove completed event from the wait list
307 wait_object_count -= 1;
308 if (wait_idx == 0)
309 wait_objects[0] = wait_objects[1];
310
311 var read_bytes: u32 = undefined;
312 if (windows.kernel32.GetOverlappedResult(handles[i], &overlapped[i], &read_bytes, 0) == 0) {
313 switch (windows.kernel32.GetLastError()) {
314 .BROKEN_PIPE => {
315 if (wait_object_count == 0)
316 break;
317 continue;
318 },
319 else => |err| return windows.unexpectedError(err),
320 }
321 }
322
323 outs[i].items.len += read_bytes;
324 const new_capacity = std.math.min(outs[i].items.len + bump_amt, max_output_bytes);
325 try outs[i].ensureCapacity(new_capacity);
326 const buf = outs[i].unusedCapacitySlice();
327 if (buf.len == 0) return if (i == 0) error.StdoutStreamTooLong else error.StderrStreamTooLong;
328 _ = windows.kernel32.ReadFile(handles[i], buf.ptr, math.cast(u32, buf.len) catch maxInt(u32), null, &overlapped[i]);
329 wait_objects[wait_object_count] = handles[i];
330 wait_object_count += 1;
312 }331 }
313 }332 }
314333
...@@ -361,12 +380,8 @@ pub const ChildProcess = struct {...@@ -361,12 +380,8 @@ pub const ChildProcess = struct {
361 stderr.deinit();380 stderr.deinit();
362 }381 }
363382
364 try collectOutputPosix(child, &stdout, &stderr, args.max_output_bytes);
365
366 // XXX: Respect max_output_bytes
367 // XXX: Smarter reading logic, read directly into the ArrayList
368 if (builtin.os.tag == .windows) {383 if (builtin.os.tag == .windows) {
369 try collectOutputWindows(child, &stdout, &stderr, args.max_output_bytes);384 try collectOutputWindows(child, [_]*std.ArrayList(u8){ &stdout, &stderr }, args.max_output_bytes);
370 } else {385 } else {
371 try collectOutputPosix(child, &stdout, &stderr, args.max_output_bytes);386 try collectOutputPosix(child, &stdout, &stderr, args.max_output_bytes);
372 }387 }
...@@ -707,7 +722,7 @@ pub const ChildProcess = struct {...@@ -707,7 +722,7 @@ pub const ChildProcess = struct {
707 var g_hChildStd_OUT_Wr: ?windows.HANDLE = null;722 var g_hChildStd_OUT_Wr: ?windows.HANDLE = null;
708 switch (self.stdout_behavior) {723 switch (self.stdout_behavior) {
709 StdIo.Pipe => {724 StdIo.Pipe => {
710 try windowsMakePipe(&g_hChildStd_OUT_Rd, &g_hChildStd_OUT_Wr, &saAttr);725 try windowsMakeAsyncPipe(&g_hChildStd_OUT_Rd, &g_hChildStd_OUT_Wr, &saAttr);
711 },726 },
712 StdIo.Ignore => {727 StdIo.Ignore => {
713 g_hChildStd_OUT_Wr = nul_handle;728 g_hChildStd_OUT_Wr = nul_handle;
...@@ -727,7 +742,7 @@ pub const ChildProcess = struct {...@@ -727,7 +742,7 @@ pub const ChildProcess = struct {
727 var g_hChildStd_ERR_Wr: ?windows.HANDLE = null;742 var g_hChildStd_ERR_Wr: ?windows.HANDLE = null;
728 switch (self.stderr_behavior) {743 switch (self.stderr_behavior) {
729 StdIo.Pipe => {744 StdIo.Pipe => {
730 try windowsMakePipe(&g_hChildStd_ERR_Rd, &g_hChildStd_ERR_Wr, &saAttr);745 try windowsMakeAsyncPipe(&g_hChildStd_ERR_Rd, &g_hChildStd_ERR_Wr, &saAttr);
731 },746 },
732 StdIo.Ignore => {747 StdIo.Ignore => {
733 g_hChildStd_ERR_Wr = nul_handle;748 g_hChildStd_ERR_Wr = nul_handle;
...@@ -960,25 +975,43 @@ fn windowsDestroyPipe(rd: ?windows.HANDLE, wr: ?windows.HANDLE) void {...@@ -960,25 +975,43 @@ fn windowsDestroyPipe(rd: ?windows.HANDLE, wr: ?windows.HANDLE) void {
960 if (wr) |h| os.close(h);975 if (wr) |h| os.close(h);
961}976}
962977
963var pipe_name_counter = std.atomic.Int(u32).init(1);978fn windowsMakePipeIn(rd: *?windows.HANDLE, wr: *?windows.HANDLE, sattr: *const windows.SECURITY_ATTRIBUTES) !void {
979 var rd_h: windows.HANDLE = undefined;
980 var wr_h: windows.HANDLE = undefined;
981 try windows.CreatePipe(&rd_h, &wr_h, sattr);
982 errdefer windowsDestroyPipe(rd_h, wr_h);
983 try windows.SetHandleInformation(wr_h, windows.HANDLE_FLAG_INHERIT, 0);
984 rd.* = rd_h;
985 wr.* = wr_h;
986}
964987
965fn windowsMakePipe(rd: *?windows.HANDLE, wr: *?windows.HANDLE, sattr: *const windows.SECURITY_ATTRIBUTES) !void {988var pipe_name_counter = std.atomic.Atomic(u32).init(1);
966 var tmp_buf: [128]u8 = undefined;989
967 // Forge a random path for the pipe.990fn windowsMakeAsyncPipe(rd: *?windows.HANDLE, wr: *?windows.HANDLE, sattr: *const windows.SECURITY_ATTRIBUTES) !void {
968 const pipe_path = std.fmt.bufPrintZ(991 var tmp_bufw: [128]u16 = undefined;
969 &tmp_buf,992
970 "\\\\.\\pipe\\zig-childprocess-{d}-{d}",993 // We must make a named pipe on windows because anonymous pipes do not support async IO
971 .{ windows.kernel32.GetCurrentProcessId(), pipe_name_counter.fetchAdd(1) },994 const pipe_path = blk: {
972 ) catch unreachable;995 var tmp_buf: [128]u8 = undefined;
996 // Forge a random path for the pipe.
997 const pipe_path = std.fmt.bufPrintZ(
998 &tmp_buf,
999 "\\\\.\\pipe\\zig-childprocess-{d}-{d}",
1000 .{ windows.kernel32.GetCurrentProcessId(), pipe_name_counter.fetchAdd(1, .Monotonic) },
1001 ) catch unreachable;
1002 const len = std.unicode.utf8ToUtf16Le(&tmp_bufw, pipe_path) catch unreachable;
1003 tmp_bufw[len] = 0;
1004 break :blk tmp_bufw[0..len :0];
1005 };
9731006
974 // Create the read handle that can be used with overlapped IO ops.1007 // Create the read handle that can be used with overlapped IO ops.
975 const read_handle = windows.kernel32.CreateNamedPipeA(1008 const read_handle = windows.kernel32.CreateNamedPipeW(
976 pipe_path,1009 pipe_path.ptr,
977 windows.PIPE_ACCESS_INBOUND | windows.FILE_FLAG_OVERLAPPED,1010 windows.PIPE_ACCESS_INBOUND | windows.FILE_FLAG_OVERLAPPED,
978 windows.PIPE_TYPE_BYTE,1011 windows.PIPE_TYPE_BYTE,
979 1,1012 1,
980 0x1000,1013 4096,
981 0x1000,1014 4096,
982 0,1015 0,
983 sattr,1016 sattr,
984 );1017 );
...@@ -987,12 +1020,14 @@ fn windowsMakePipe(rd: *?windows.HANDLE, wr: *?windows.HANDLE, sattr: *const win...@@ -987,12 +1020,14 @@ fn windowsMakePipe(rd: *?windows.HANDLE, wr: *?windows.HANDLE, sattr: *const win
987 else => |err| return windows.unexpectedError(err),1020 else => |err| return windows.unexpectedError(err),
988 }1021 }
989 }1022 }
1023 errdefer os.close(read_handle);
9901024
991 const write_handle = windows.kernel32.CreateFileA(1025 var sattr_copy = sattr.*;
992 pipe_path,1026 const write_handle = windows.kernel32.CreateFileW(
1027 pipe_path.ptr,
993 windows.GENERIC_WRITE,1028 windows.GENERIC_WRITE,
994 0,1029 0,
995 sattr,1030 &sattr_copy,
996 windows.OPEN_EXISTING,1031 windows.OPEN_EXISTING,
997 windows.FILE_ATTRIBUTE_NORMAL,1032 windows.FILE_ATTRIBUTE_NORMAL,
998 null,1033 null,
...@@ -1002,6 +1037,7 @@ fn windowsMakePipe(rd: *?windows.HANDLE, wr: *?windows.HANDLE, sattr: *const win...@@ -1002,6 +1037,7 @@ fn windowsMakePipe(rd: *?windows.HANDLE, wr: *?windows.HANDLE, sattr: *const win
1002 else => |err| return windows.unexpectedError(err),1037 else => |err| return windows.unexpectedError(err),
1003 }1038 }
1004 }1039 }
1040 errdefer os.close(write_handle);
10051041
1006 try windows.SetHandleInformation(read_handle, windows.HANDLE_FLAG_INHERIT, 0);1042 try windows.SetHandleInformation(read_handle, windows.HANDLE_FLAG_INHERIT, 0);
10071043
...@@ -1009,26 +1045,6 @@ fn windowsMakePipe(rd: *?windows.HANDLE, wr: *?windows.HANDLE, sattr: *const win...@@ -1009,26 +1045,6 @@ fn windowsMakePipe(rd: *?windows.HANDLE, wr: *?windows.HANDLE, sattr: *const win
1009 wr.* = write_handle;1045 wr.* = write_handle;
1010}1046}
10111047
1012fn windowsMakePipeIn(rd: *?windows.HANDLE, wr: *?windows.HANDLE, sattr: *const windows.SECURITY_ATTRIBUTES) !void {
1013 var rd_h: windows.HANDLE = undefined;
1014 var wr_h: windows.HANDLE = undefined;
1015 try windows.CreatePipe(&rd_h, &wr_h, sattr);
1016 errdefer windowsDestroyPipe(rd_h, wr_h);
1017 try windows.SetHandleInformation(wr_h, windows.HANDLE_FLAG_INHERIT, 0);
1018 rd.* = rd_h;
1019 wr.* = wr_h;
1020}
1021
1022fn windowsMakePipeOut(rd: *?windows.HANDLE, wr: *?windows.HANDLE, sattr: *const windows.SECURITY_ATTRIBUTES) !void {
1023 var rd_h: windows.HANDLE = undefined;
1024 var wr_h: windows.HANDLE = undefined;
1025 try windows.CreatePipe(&rd_h, &wr_h, sattr);
1026 errdefer windowsDestroyPipe(rd_h, wr_h);
1027 try windows.SetHandleInformation(rd_h, windows.HANDLE_FLAG_INHERIT, 0);
1028 rd.* = rd_h;
1029 wr.* = wr_h;
1030}
1031
1032fn destroyPipe(pipe: [2]os.fd_t) void {1048fn destroyPipe(pipe: [2]os.fd_t) void {
1033 os.close(pipe[0]);1049 os.close(pipe[0]);
1034 if (pipe[0] != pipe[1]) os.close(pipe[1]);1050 if (pipe[0] != pipe[1]) os.close(pipe[1]);
lib/std/os/windows/kernel32.zig+14-33
...@@ -8,6 +8,7 @@ usingnamespace @import("bits.zig");...@@ -8,6 +8,7 @@ usingnamespace @import("bits.zig");
8pub extern "kernel32" fn AddVectoredExceptionHandler(First: c_ulong, Handler: ?VECTORED_EXCEPTION_HANDLER) callconv(WINAPI) ?*c_void;8pub extern "kernel32" fn AddVectoredExceptionHandler(First: c_ulong, Handler: ?VECTORED_EXCEPTION_HANDLER) callconv(WINAPI) ?*c_void;
9pub extern "kernel32" fn RemoveVectoredExceptionHandler(Handle: HANDLE) callconv(WINAPI) c_ulong;9pub extern "kernel32" fn RemoveVectoredExceptionHandler(Handle: HANDLE) callconv(WINAPI) c_ulong;
1010
11pub extern "kernel32" fn CancelIo(hFile: HANDLE) callconv(WINAPI) BOOL;
11pub extern "kernel32" fn CancelIoEx(hFile: HANDLE, lpOverlapped: ?LPOVERLAPPED) callconv(WINAPI) BOOL;12pub extern "kernel32" fn CancelIoEx(hFile: HANDLE, lpOverlapped: ?LPOVERLAPPED) callconv(WINAPI) BOOL;
1213
13pub extern "kernel32" fn CloseHandle(hObject: HANDLE) callconv(WINAPI) BOOL;14pub extern "kernel32" fn CloseHandle(hObject: HANDLE) callconv(WINAPI) BOOL;
...@@ -15,29 +16,6 @@ pub extern "kernel32" fn CloseHandle(hObject: HANDLE) callconv(WINAPI) BOOL;...@@ -15,29 +16,6 @@ pub extern "kernel32" fn CloseHandle(hObject: HANDLE) callconv(WINAPI) BOOL;
15pub extern "kernel32" fn CreateDirectoryW(lpPathName: [*:0]const u16, lpSecurityAttributes: ?*SECURITY_ATTRIBUTES) callconv(WINAPI) BOOL;16pub extern "kernel32" fn CreateDirectoryW(lpPathName: [*:0]const u16, lpSecurityAttributes: ?*SECURITY_ATTRIBUTES) callconv(WINAPI) BOOL;
16pub extern "kernel32" fn SetEndOfFile(hFile: HANDLE) callconv(WINAPI) BOOL;17pub extern "kernel32" fn SetEndOfFile(hFile: HANDLE) callconv(WINAPI) BOOL;
1718
18pub extern "kernel32" fn GetCurrentProcessId() callconv(WINAPI) DWORD;
19
20pub extern "kernel32" fn CreateNamedPipeA(
21 lpName: [*:0]const u8,
22 dwOpenMode: DWORD,
23 dwPipeMode: DWORD,
24 nMaxInstances: DWORD,
25 nOutBufferSize: DWORD,
26 nInBufferSize: DWORD,
27 nDefaultTimeOut: DWORD,
28 lpSecurityAttributes: ?*const SECURITY_ATTRIBUTES,
29) callconv(WINAPI) HANDLE;
30pub extern "kernel32" fn CreateNamedPipeW(
31 lpName: LPCWSTR,
32 dwOpenMode: DWORD,
33 dwPipeMode: DWORD,
34 nMaxInstances: DWORD,
35 nOutBufferSize: DWORD,
36 nInBufferSize: DWORD,
37 nDefaultTimeOut: DWORD,
38 lpSecurityAttributes: ?*const SECURITY_ATTRIBUTES,
39) callconv(WINAPI) HANDLE;
40
41pub extern "kernel32" fn CreateEventExW(19pub extern "kernel32" fn CreateEventExW(
42 lpEventAttributes: ?*SECURITY_ATTRIBUTES,20 lpEventAttributes: ?*SECURITY_ATTRIBUTES,
43 lpName: [*:0]const u16,21 lpName: [*:0]const u16,
...@@ -55,16 +33,6 @@ pub extern "kernel32" fn CreateFileW(...@@ -55,16 +33,6 @@ pub extern "kernel32" fn CreateFileW(
55 hTemplateFile: ?HANDLE,33 hTemplateFile: ?HANDLE,
56) callconv(WINAPI) HANDLE;34) callconv(WINAPI) HANDLE;
5735
58pub extern "kernel32" fn CreateFileA(
59 lpFileName: [*:0]const u8,
60 dwDesiredAccess: DWORD,
61 dwShareMode: DWORD,
62 lpSecurityAttributes: ?*const SECURITY_ATTRIBUTES,
63 dwCreationDisposition: DWORD,
64 dwFlagsAndAttributes: DWORD,
65 hTemplateFile: ?HANDLE,
66) callconv(WINAPI) HANDLE;
67
68pub extern "kernel32" fn CreatePipe(36pub extern "kernel32" fn CreatePipe(
69 hReadPipe: *HANDLE,37 hReadPipe: *HANDLE,
70 hWritePipe: *HANDLE,38 hWritePipe: *HANDLE,
...@@ -72,6 +40,17 @@ pub extern "kernel32" fn CreatePipe(...@@ -72,6 +40,17 @@ pub extern "kernel32" fn CreatePipe(
72 nSize: DWORD,40 nSize: DWORD,
73) callconv(WINAPI) BOOL;41) callconv(WINAPI) BOOL;
7442
43pub extern "kernel32" fn CreateNamedPipeW(
44 lpName: LPCWSTR,
45 dwOpenMode: DWORD,
46 dwPipeMode: DWORD,
47 nMaxInstances: DWORD,
48 nOutBufferSize: DWORD,
49 nInBufferSize: DWORD,
50 nDefaultTimeOut: DWORD,
51 lpSecurityAttributes: ?*const SECURITY_ATTRIBUTES,
52) callconv(WINAPI) HANDLE;
53
75pub extern "kernel32" fn CreateProcessW(54pub extern "kernel32" fn CreateProcessW(
76 lpApplicationName: ?LPWSTR,55 lpApplicationName: ?LPWSTR,
77 lpCommandLine: LPWSTR,56 lpCommandLine: LPWSTR,
...@@ -132,6 +111,8 @@ pub extern "kernel32" fn GetCurrentDirectoryW(nBufferLength: DWORD, lpBuffer: ?[...@@ -132,6 +111,8 @@ pub extern "kernel32" fn GetCurrentDirectoryW(nBufferLength: DWORD, lpBuffer: ?[
132pub extern "kernel32" fn GetCurrentThread() callconv(WINAPI) HANDLE;111pub extern "kernel32" fn GetCurrentThread() callconv(WINAPI) HANDLE;
133pub extern "kernel32" fn GetCurrentThreadId() callconv(WINAPI) DWORD;112pub extern "kernel32" fn GetCurrentThreadId() callconv(WINAPI) DWORD;
134113
114pub extern "kernel32" fn GetCurrentProcessId() callconv(WINAPI) DWORD;
115
135pub extern "kernel32" fn GetCurrentProcess() callconv(WINAPI) HANDLE;116pub extern "kernel32" fn GetCurrentProcess() callconv(WINAPI) HANDLE;
136117
137pub extern "kernel32" fn GetEnvironmentStringsW() callconv(WINAPI) ?[*:0]u16;118pub extern "kernel32" fn GetEnvironmentStringsW() callconv(WINAPI) ?[*:0]u16;