authorgravatar for mlugg@mlugg.co.ukMatthew Lugg <mlugg@mlugg.co.uk> 2024-09-29 00:03:46+01:00
committergravatar for mlugg@mlugg.co.ukMatthew Lugg <mlugg@mlugg.co.uk> 2024-10-06 07:01:12+01:00
log90db7677212f8331733a661615490d37c7bf75d2
treee208936079ab24bcc0583820369c28c6ce8a3a59
parentada60616b37b76004237f4e13de8b552c16dc773
signaturelock-open Commit is signed but in an unrecognized format.

std: async read into small temporary buffer between `poll` calls on Windows

This commit changes how `std.io.poll` is implemented on Windows. The new implementation unfortunately incurs a little extra system call overhead, but fixes several bugs in the old implementation: * The `lpNumberOfBytesRead` parameter of `ReadFile` was used with overlapped I/O. This is explicitly disallowed by the documentation, as the value written to this pointer is "potentially erroneous"; instead, `GetOverlappedResult` must always be used, even if the operation immediately returns. Documentation states that `lpNumberOfBytesRead` cannot be passed as null on Windows 7, so for compatibility, the parameter is passed as a pointer to a dummy global. * If the initial `ReadFile` returned data, and the next read returned `BROKEN_PIPE`, the received data was silently ignored in the sense that `pollWindows` did not `return`, instead waiting for data to come in on another file (or for all files to close). * The asynchronous `ReadFile` calls which were left pending between calls to `pollWindows` pointed to a potentially unstable buffer, since the user of `poll` may use part of the `LinearFifo` API which rotate its ring buffer. This race condition was causing CI failures in some uses of the compiler server protocol. These issues are all resolved. Now, `pollWindows` will queue an initial read to a small (128-byte) stable buffer per file. When this read is completed, reads directly into the FIFO's writable slice are performed until one is left pending, at which point that read is cancelled (with a check to see if it was completed between the `ReadFile` and `CancelIo` calls) and the next read into the small stable buffer is queued. These small buffer reads are the ones left pending between `pollWindows` calls, avoiding the race condition described above. Related: #21565

1 files changed, 162 insertions(+), 30 deletions(-)

lib/std/io.zig+162-30
...@@ -442,6 +442,7 @@ pub fn poll(...@@ -442,6 +442,7 @@ pub fn poll(
442 .overlapped = [1]windows.OVERLAPPED{442 .overlapped = [1]windows.OVERLAPPED{
443 mem.zeroes(windows.OVERLAPPED),443 mem.zeroes(windows.OVERLAPPED),
444 } ** enum_fields.len,444 } ** enum_fields.len,
445 .small_bufs = undefined,
445 .active = .{446 .active = .{
446 .count = 0,447 .count = 0,
447 .handles_buf = undefined,448 .handles_buf = undefined,
...@@ -481,6 +482,7 @@ pub fn Poller(comptime StreamEnum: type) type {...@@ -481,6 +482,7 @@ pub fn Poller(comptime StreamEnum: type) type {
481 windows: if (is_windows) struct {482 windows: if (is_windows) struct {
482 first_read_done: bool,483 first_read_done: bool,
483 overlapped: [enum_fields.len]windows.OVERLAPPED,484 overlapped: [enum_fields.len]windows.OVERLAPPED,
485 small_bufs: [enum_fields.len][128]u8,
484 active: struct {486 active: struct {
485 count: math.IntFittingRange(0, enum_fields.len),487 count: math.IntFittingRange(0, enum_fields.len),
486 handles_buf: [enum_fields.len]windows.HANDLE,488 handles_buf: [enum_fields.len]windows.HANDLE,
...@@ -534,24 +536,31 @@ pub fn Poller(comptime StreamEnum: type) type {...@@ -534,24 +536,31 @@ pub fn Poller(comptime StreamEnum: type) type {
534 const bump_amt = 512;536 const bump_amt = 512;
535537
536 if (!self.windows.first_read_done) {538 if (!self.windows.first_read_done) {
537 // Windows Async IO requires an initial call to ReadFile before waiting on the handle539 var already_read_data = false;
538 for (0..enum_fields.len) |i| {540 for (0..enum_fields.len) |i| {
539 const handle = self.windows.active.handles_buf[i];541 const handle = self.windows.active.handles_buf[i];
540 switch (try windowsAsyncRead(542 switch (try windowsAsyncReadToFifoAndQueueSmallRead(
541 handle,543 handle,
542 &self.windows.overlapped[i],544 &self.windows.overlapped[i],
543 &self.fifos[i],545 &self.fifos[i],
546 &self.windows.small_bufs[i],
544 bump_amt,547 bump_amt,
545 )) {548 )) {
546 .pending => {549 .populated, .empty => |state| {
550 if (state == .populated) already_read_data = true;
547 self.windows.active.handles_buf[self.windows.active.count] = handle;551 self.windows.active.handles_buf[self.windows.active.count] = handle;
548 self.windows.active.stream_map[self.windows.active.count] = @as(StreamEnum, @enumFromInt(i));552 self.windows.active.stream_map[self.windows.active.count] = @as(StreamEnum, @enumFromInt(i));
549 self.windows.active.count += 1;553 self.windows.active.count += 1;
550 },554 },
551 .closed => {}, // don't add to the wait_objects list555 .closed => {}, // don't add to the wait_objects list
556 .closed_populated => {
557 // don't add to the wait_objects list, but we did already get data
558 already_read_data = true;
559 },
552 }560 }
553 }561 }
554 self.windows.first_read_done = true;562 self.windows.first_read_done = true;
563 if (already_read_data) return true;
555 }564 }
556565
557 while (true) {566 while (true) {
...@@ -576,32 +585,35 @@ pub fn Poller(comptime StreamEnum: type) type {...@@ -576,32 +585,35 @@ pub fn Poller(comptime StreamEnum: type) type {
576585
577 const active_idx = status - windows.WAIT_OBJECT_0;586 const active_idx = status - windows.WAIT_OBJECT_0;
578587
579 const handle = self.windows.active.handles_buf[active_idx];
580 const stream_idx = @intFromEnum(self.windows.active.stream_map[active_idx]);588 const stream_idx = @intFromEnum(self.windows.active.stream_map[active_idx]);
581 var read_bytes: u32 = undefined;589 const handle = self.windows.active.handles_buf[active_idx];
582 if (0 == windows.kernel32.GetOverlappedResult(590
583 handle,591 const overlapped = &self.windows.overlapped[stream_idx];
584 &self.windows.overlapped[stream_idx],592 const stream_fifo = &self.fifos[stream_idx];
585 &read_bytes,593 const small_buf = &self.windows.small_bufs[stream_idx];
586 0,594
587 )) switch (windows.GetLastError()) {595 const num_bytes_read = switch (try windowsGetReadResult(handle, overlapped, false)) {
588 .BROKEN_PIPE => {596 .success => |n| n,
597 .closed => {
589 self.windows.active.removeAt(active_idx);598 self.windows.active.removeAt(active_idx);
590 continue;599 continue;
591 },600 },
592 else => |err| return windows.unexpectedError(err),601 .aborted => unreachable,
593 };602 };
603 try stream_fifo.write(small_buf[0..num_bytes_read]);
594604
595 self.fifos[stream_idx].update(read_bytes);605 switch (try windowsAsyncReadToFifoAndQueueSmallRead(
596
597 switch (try windowsAsyncRead(
598 handle,606 handle,
599 &self.windows.overlapped[stream_idx],607 overlapped,
600 &self.fifos[stream_idx],608 stream_fifo,
609 small_buf,
601 bump_amt,610 bump_amt,
602 )) {611 )) {
603 .pending => {},612 .empty => {}, // irrelevant, we already got data from the small buffer
604 .closed => self.windows.active.removeAt(active_idx),613 .populated => {},
614 .closed,
615 .closed_populated, // identical, since we already got data from the small buffer
616 => self.windows.active.removeAt(active_idx),
605 }617 }
606 return true;618 return true;
607 }619 }
...@@ -654,25 +666,145 @@ pub fn Poller(comptime StreamEnum: type) type {...@@ -654,25 +666,145 @@ pub fn Poller(comptime StreamEnum: type) type {
654 };666 };
655}667}
656668
657fn windowsAsyncRead(669/// The `ReadFile` docuementation states that `lpNumberOfBytesRead` does not have a meaningful
670/// result when using overlapped I/O, but also that it cannot be `null` on Windows 7. For
671/// compatibility, we point it to this dummy variables, which we never otherwise access.
672/// See: https://learn.microsoft.com/en-us/windows/win32/api/fileapi/nf-fileapi-readfile
673var win_dummy_bytes_read: u32 = undefined;
674
675/// Read as much data as possible from `handle` with `overlapped`, and write it to the FIFO. Before
676/// returning, queue a read into `small_buf` so that `WaitForMultipleObjects` returns when more data
677/// is available. `handle` must have no pending asynchronous operation.
678fn windowsAsyncReadToFifoAndQueueSmallRead(
658 handle: windows.HANDLE,679 handle: windows.HANDLE,
659 overlapped: *windows.OVERLAPPED,680 overlapped: *windows.OVERLAPPED,
660 fifo: *PollFifo,681 fifo: *PollFifo,
682 small_buf: *[128]u8,
661 bump_amt: usize,683 bump_amt: usize,
662) !enum { pending, closed } {684) !enum { empty, populated, closed_populated, closed } {
685 var read_any_data = false;
663 while (true) {686 while (true) {
664 const buf = try fifo.writableWithSize(bump_amt);687 const fifo_read_pending = while (true) {
665 var read_bytes: u32 = undefined;688 const buf = try fifo.writableWithSize(bump_amt);
666 const read_result = windows.kernel32.ReadFile(handle, buf.ptr, math.cast(u32, buf.len) orelse math.maxInt(u32), &read_bytes, overlapped);689 const buf_len = math.cast(u32, buf.len) orelse math.maxInt(u32);
667 if (read_result == 0) return switch (windows.GetLastError()) {690
668 .IO_PENDING => .pending,691 if (0 == windows.kernel32.ReadFile(
669 .BROKEN_PIPE => .closed,692 handle,
670 else => |err| windows.unexpectedError(err),693 buf.ptr,
694 buf_len,
695 &win_dummy_bytes_read,
696 overlapped,
697 )) switch (windows.GetLastError()) {
698 .IO_PENDING => break true,
699 .BROKEN_PIPE => return if (read_any_data) .closed_populated else .closed,
700 else => |err| return windows.unexpectedError(err),
701 };
702
703 const num_bytes_read = switch (try windowsGetReadResult(handle, overlapped, false)) {
704 .success => |n| n,
705 .closed => return if (read_any_data) .closed_populated else .closed,
706 .aborted => unreachable,
707 };
708
709 read_any_data = true;
710 fifo.update(num_bytes_read);
711
712 if (num_bytes_read == buf_len) {
713 // We filled the buffer, so there's probably more data available.
714 continue;
715 } else {
716 // We didn't fill the buffer, so assume we're out of data.
717 // There is no pending read.
718 break false;
719 }
671 };720 };
672 fifo.update(read_bytes);721
722 if (fifo_read_pending) cancel_read: {
723 // Cancel the pending read into the FIFO.
724 _ = windows.kernel32.CancelIo(handle);
725
726 // We have to wait for the handle to be signalled, i.e. for the cancellation to complete.
727 switch (windows.kernel32.WaitForSingleObject(handle, windows.INFINITE)) {
728 windows.WAIT_OBJECT_0 => {},
729 windows.WAIT_FAILED => return windows.unexpectedError(windows.GetLastError()),
730 else => unreachable,
731 }
732
733 // If it completed before we canceled, make sure to tell the FIFO!
734 const num_bytes_read = switch (try windowsGetReadResult(handle, overlapped, true)) {
735 .success => |n| n,
736 .closed => return if (read_any_data) .closed_populated else .closed,
737 .aborted => break :cancel_read,
738 };
739 read_any_data = true;
740 fifo.update(num_bytes_read);
741 }
742
743 // Try to queue the 1-byte read.
744 if (0 == windows.kernel32.ReadFile(
745 handle,
746 small_buf,
747 small_buf.len,
748 &win_dummy_bytes_read,
749 overlapped,
750 )) switch (windows.GetLastError()) {
751 .IO_PENDING => {
752 // 1-byte read pending as intended
753 return if (read_any_data) .populated else .empty;
754 },
755 .BROKEN_PIPE => return if (read_any_data) .closed_populated else .closed,
756 else => |err| return windows.unexpectedError(err),
757 };
758
759 // We got data back this time. Write it to the FIFO and run the main loop again.
760 const num_bytes_read = switch (try windowsGetReadResult(handle, overlapped, false)) {
761 .success => |n| n,
762 .closed => return if (read_any_data) .closed_populated else .closed,
763 .aborted => unreachable,
764 };
765 try fifo.write(small_buf[0..num_bytes_read]);
766 read_any_data = true;
673 }767 }
674}768}
675769
770/// Simple wrapper around `GetOverlappedResult` to determine the result of a `ReadFile` operation.
771/// If `!allow_aborted`, then `aborted` is never returned (`OPERATION_ABORTED` is considered unexpected).
772///
773/// The `ReadFile` documentation states that the number of bytes read by an overlapped `ReadFile` must be determined using `GetOverlappedResult`, even if the
774/// operation immediately returns data:
775/// "Use NULL for [lpNumberOfBytesRead] if this is an asynchronous operation to avoid potentially
776/// erroneous results."
777/// "If `hFile` was opened with `FILE_FLAG_OVERLAPPED`, the following conditions are in effect: [...]
778/// The lpNumberOfBytesRead parameter should be set to NULL. Use the GetOverlappedResult function to
779/// get the actual number of bytes read."
780/// See: https://learn.microsoft.com/en-us/windows/win32/api/fileapi/nf-fileapi-readfile
781fn windowsGetReadResult(
782 handle: windows.HANDLE,
783 overlapped: *windows.OVERLAPPED,
784 allow_aborted: bool,
785) !union(enum) {
786 success: u32,
787 closed,
788 aborted,
789} {
790 var num_bytes_read: u32 = undefined;
791 if (0 == windows.kernel32.GetOverlappedResult(
792 handle,
793 overlapped,
794 &num_bytes_read,
795 0,
796 )) switch (windows.GetLastError()) {
797 .BROKEN_PIPE => return .closed,
798 .OPERATION_ABORTED => |err| if (allow_aborted) {
799 return .aborted;
800 } else {
801 return windows.unexpectedError(err);
802 },
803 else => |err| return windows.unexpectedError(err),
804 };
805 return .{ .success = num_bytes_read };
806}
807
676/// Given an enum, returns a struct with fields of that enum, each field808/// Given an enum, returns a struct with fields of that enum, each field
677/// representing an I/O stream for polling.809/// representing an I/O stream for polling.
678pub fn PollFiles(comptime StreamEnum: type) type {810pub fn PollFiles(comptime StreamEnum: type) type {