diff --git a/lib/std/Io/Threaded.zig b/lib/std/Io/Threaded.zig index 96a5d648eedfec47b31715d804c3e424cc76d3de..d7b8b71c110c85a7d63717d5d0c82d1c48bcb6a5 100644 --- a/lib/std/Io/Threaded.zig +++ b/lib/std/Io/Threaded.zig @@ -339,8 +339,8 @@ const Group = struct { .canceled => true, .parked => unreachable, .blocked => unreachable, - .blocked_apc => unreachable, - .blocked_windows_dns => unreachable, + .blocked_alertable => unreachable, + .blocked_alertable_canceling => unreachable, .blocked_canceling => unreachable, }; if (result) { @@ -379,10 +379,7 @@ const Group = struct { while (it) |thread| : (it = thread.next) { // This non-mutating RMW exists for ordering reasons: see comment in `Group.Task.start` for reasons. _ = thread.status.fetchOr(.{ .cancelation = @enumFromInt(0), .awaitable = .null }, .release); - if (thread.cancelAwaitable(.fromGroup(g.ptr))) |method| { - thread.interrupt_method = method; - any_blocked = true; - } + if (thread.cancelAwaitable(.fromGroup(g.ptr))) any_blocked = true; } return any_blocked; } @@ -394,7 +391,7 @@ const Group = struct { var any_signaled = false; var it = t.worker_threads.load(.acquire); // acquire `Thread` values while (it) |thread| : (it = thread.next) { - if (thread.signalCanceledSyscall(t, .fromGroup(g.ptr), thread.interrupt_method)) any_signaled = true; + if (thread.signalCanceledSyscall(t, .fromGroup(g.ptr))) any_signaled = true; } return any_signaled; } @@ -546,8 +543,8 @@ const Future = struct { .canceled => true, .parked => unreachable, .blocked => unreachable, - .blocked_apc => unreachable, - .blocked_windows_dns => unreachable, + .blocked_alertable => unreachable, + .blocked_alertable_canceling => unreachable, .blocked_canceling => unreachable, }; thread.status.store(.{ .cancelation = .none, .awaitable = .null }, .monotonic); @@ -576,15 +573,11 @@ const Future = struct { num_completed: *std.atomic.Value(u32), thread: ?*Thread, ) void { - var interrupt_method: ?Thread.InterruptMethod = - if (thread) |th| th.cancelAwaitable(.fromFuture(future)) else null; + var need_signal: bool = if (thread) |th| th.cancelAwaitable(.fromFuture(future)) else false; var timeout_ns: u64 = 1 << 10; while (true) { - if (interrupt_method) |method| { - if (!thread.?.signalCanceledSyscall(t, .fromFuture(future), method)) - interrupt_method = null; - } - Thread.futexWaitUncancelable(&num_completed.raw, 0, if (interrupt_method != null) timeout_ns else null); + need_signal = need_signal and thread.?.signalCanceledSyscall(t, .fromFuture(future)); + Thread.futexWaitUncancelable(&num_completed.raw, 0, if (need_signal) timeout_ns else null); switch (num_completed.load(.acquire)) { // acquire task results 0 => {}, 1 => break, @@ -632,17 +625,9 @@ const Thread = struct { cancel_protection: Io.CancelProtection, /// Always released when `Status.cancelation` is set to `.parked`. futex_waiter: if (use_parking_futex) ?*parking_futex.Waiter else ?noreturn, - apc: Apc, - /// Used only by group cancelation code for temporary storage. - interrupt_method: InterruptMethod, csprng: Csprng, - const Apc = if (is_windows) struct { - handle: windows.HANDLE, - iosb: ?*windows.IO_STATUS_BLOCK, - } else void; - const Handle = Handle: { if (std.Thread.use_pthreads) break :Handle std.c.pthread_t; if (is_windows) break :Handle windows.HANDLE; @@ -667,13 +652,11 @@ const Thread = struct { /// To request cancelation, set the status to `.blocked_canceling` and repeatedly interrupt the system call until the status changes. blocked = 0b011, - /// Windows-only: the thread is blocked in a call to `NtDelayExecution`. - /// To request cancelation, set the status to `.canceling` and call `NtCancelIoFileEx`. - blocked_apc = 0b100, - - /// Windows-only: the thread is blocked in a call to `GetAddrInfoExW`. - /// To request cancelation, set the status to `.canceling` and call `GetAddrInfoExCancel`. - blocked_windows_dns = 0b010, + /// Windows-only: the thread is blocked in an alertable wait via + /// `NtDelayExecution`. To request cancelation, set the status to + /// `blocked_alertable_canceling` and repeatedly alert the thread + /// until the status changes. + blocked_alertable = 0b010, /// The thread has an outstanding cancelation request but is not in a cancelable operation. /// When it acknowledges the cancelation, it will set the status to `.canceled`. @@ -722,8 +705,8 @@ const Thread = struct { switch (status.cancelation) { .parked => unreachable, .blocked => unreachable, - .blocked_apc => unreachable, - .blocked_windows_dns => unreachable, + .blocked_alertable => unreachable, + .blocked_alertable_canceling => unreachable, .blocked_canceling => unreachable, .none, .canceled => {}, .canceling => { @@ -1003,17 +986,17 @@ const Thread = struct { /// It is possible that `thread` gets canceled by this function, but is blocked in a syscall. In /// that case, the thread may need to be sent a signal to interrupt the call. This function will /// return `true` to indicate this, in which case the caller must call `signalCanceledSyscall`. - fn cancelAwaitable(thread: *Thread, awaitable: AwaitableId) ?InterruptMethod { + fn cancelAwaitable(thread: *Thread, awaitable: AwaitableId) bool { var status = thread.status.load(.monotonic); while (true) { - if (status.awaitable != awaitable) return null; // thread is working on something else + if (status.awaitable != awaitable) return false; // thread is working on something else status = switch (status.cancelation) { .none => thread.status.cmpxchgWeak( .{ .cancelation = .none, .awaitable = awaitable }, .{ .cancelation = .canceling, .awaitable = awaitable }, .monotonic, .monotonic, - ) orelse return null, + ) orelse return false, .parked => thread.status.cmpxchgWeak( .{ .cancelation = .parked, .awaitable = awaitable }, @@ -1026,7 +1009,7 @@ const Thread = struct { parking_futex.removeCanceledWaiter(futex_waiter); } unpark(&.{thread.id}, null); - return null; + return false; }, .blocked => thread.status.cmpxchgWeak( @@ -1034,17 +1017,7 @@ const Thread = struct { .{ .cancelation = .blocked_canceling, .awaitable = awaitable }, .monotonic, .monotonic, - ) orelse return .sync, - - .blocked_apc => thread.status.cmpxchgWeak( - .{ .cancelation = .blocked_apc, .awaitable = awaitable }, - .{ .cancelation = .canceling, .awaitable = awaitable }, - .monotonic, - .monotonic, - ) orelse { - if (!is_windows) unreachable; - return .apc; - }, + ) orelse return true, .blocked_alertable => thread.status.cmpxchgWeak( .{ .cancelation = .blocked_alertable, .awaitable = awaitable }, @@ -1053,14 +1026,14 @@ const Thread = struct { .monotonic, ) orelse { if (!is_windows) unreachable; - return .dns; + return true; }, .canceling, .canceled => { // This can happen when the task start raced with the cancelation, so the thread // saw the cancelation on the future/group *and* we are trying to signal the // thread here. - return null; + return false; }, .blocked_canceling => unreachable, // `awaitable` has not been canceled before now @@ -1069,11 +1042,6 @@ const Thread = struct { } } - const InterruptMethod = switch (native_os) { - .windows => enum { sync, dns, apc }, - else => enum { sync }, - }; - /// Sends a signal to `thread` if it is still blocked in a syscall (i.e. has not yet observed /// the cancelation request from `cancelAwaitable`). /// @@ -1083,21 +1051,24 @@ const Thread = struct { /// the thread is still blocked. For the implementation, `Future.waitForCancelWithSignaling` and /// `Group.waitForCancelWithSignaling`: they use exponential backoff starting at a 1us delay and /// doubling each call. In practice, it is rare to send more than one signal. - fn signalCanceledSyscall(thread: *Thread, t: *Threaded, awaitable: AwaitableId, method: InterruptMethod) bool { - const bad_status: Status = .{ .cancelation = .blocked_canceling, .awaitable = awaitable }; - if (thread.status.load(.monotonic) != bad_status) return false; + fn signalCanceledSyscall(thread: *Thread, t: *Threaded, awaitable: AwaitableId) bool { + const status = thread.status.load(.monotonic); + if (status.awaitable != awaitable) { + // The thread has moved on and is working on something totally different. + return false; + } // The thread ID and/or handle can be read non-atomically because they never change and were // released by the store that made `thread` available to us. - if (std.Thread.use_pthreads) switch (method) { - .sync => return switch (std.c.pthread_kill(thread.handle, .IO)) { - 0 => true, - else => false, - }, - } else switch (native_os) { - .linux => switch (method) { - .sync => { + switch (status.cancelation) { + .blocked_canceling => if (std.Thread.use_pthreads) { + return switch (std.c.pthread_kill(thread.handle, .IO)) { + 0 => true, + else => false, + }; + } else switch (native_os) { + .linux => { const pid: posix.pid_t = pid: { const cached_pid = @atomicLoad(Pid, &t.pid, .monotonic); if (cached_pid != .unknown) break :pid @intFromEnum(cached_pid); @@ -1110,9 +1081,7 @@ const Thread = struct { else => false, }; }, - }, - .windows => switch (method) { - .sync => { + .windows => { var iosb: windows.IO_STATUS_BLOCK = undefined; return switch (windows.ntdll.NtCancelSynchronousIoFile(thread.handle, null, &iosb)) { .NOT_FOUND => true, // this might mean the operation hasn't started yet @@ -1120,15 +1089,15 @@ const Thread = struct { else => false, }; }, - .dns => @panic("TODO call GetAddrInfoExCancel"), - .apc => { - var iosb: windows.IO_STATUS_BLOCK = undefined; - return switch (windows.ntdll.NtCancelIoFileEx(thread.apc.handle, thread.apc.iosb, &iosb)) { - .NOT_FOUND => true, // this might mean the operation hasn't started yet - .SUCCESS => false, // the OS confirmed that our cancelation worked - else => false, - }; - }, + else => return false, + }, + + .blocked_alertable_canceling => { + if (!is_windows) unreachable; + return switch (windows.ntdll.NtAlertThread(thread.handle)) { + .SUCCESS => true, + else => false, + }; }, else => { @@ -1176,8 +1145,8 @@ const Syscall = struct { }, .monotonic).cancelation) { .parked => unreachable, .blocked => unreachable, - .blocked_apc => unreachable, - .blocked_windows_dns => unreachable, + .blocked_alertable => unreachable, + .blocked_alertable_canceling => unreachable, .blocked_canceling => unreachable, .none => return .{ .thread = thread }, // new status is `.blocked` .canceling => return error.Canceled, // new status is `.canceled` @@ -1196,8 +1165,8 @@ const Syscall = struct { }, .monotonic).cancelation) { .none => unreachable, .parked => unreachable, - .blocked_apc => unreachable, - .blocked_windows_dns => unreachable, + .blocked_alertable => unreachable, + .blocked_alertable_canceling => unreachable, .canceling => unreachable, .canceled => unreachable, .blocked => {}, // new status is `.blocked` (unchanged) @@ -1213,8 +1182,8 @@ const Syscall = struct { }, .monotonic).cancelation) { .none => unreachable, .parked => unreachable, - .blocked_apc => unreachable, - .blocked_windows_dns => unreachable, + .blocked_alertable => unreachable, + .blocked_alertable_canceling => unreachable, .canceling => unreachable, .canceled => unreachable, .blocked => {}, // new status is `.none` @@ -1222,25 +1191,25 @@ const Syscall = struct { } } /// Indicates instead of `NtCancelSynchronousIoFile` we need to use - /// `NtCancelIoFileEx` to interrupt the wait. + /// `NtAlertThread` to interrupt the wait. /// /// Windows only, called from blocked state only. - fn toApc(s: Syscall, apc: Thread.Apc) Io.Cancelable!void { - const thread = s.thread orelse return; - thread.apc = apc; + fn toAlertable(s: Syscall) Io.Cancelable!AlertableSyscall { + comptime assert(is_windows); + const thread = s.thread orelse return .{ .thread = null }; var prev = thread.status.load(.monotonic); while (true) prev = switch (prev.cancelation) { .none => unreachable, .parked => unreachable, - .blocked_apc => unreachable, - .blocked_windows_dns => unreachable, + .blocked_alertable => unreachable, + .blocked_alertable_canceling => unreachable, .canceling => unreachable, .canceled => unreachable, .blocked => thread.status.cmpxchgWeak(prev, .{ - .cancelation = .blocked_apc, + .cancelation = .blocked_alertable, .awaitable = prev.awaitable, - }, .monotonic, .monotonic) orelse return, + }, .monotonic, .monotonic) orelse return .{ .thread = thread }, .blocked_canceling => thread.status.cmpxchgWeak(prev, .{ .cancelation = .canceled, @@ -1248,45 +1217,6 @@ const Syscall = struct { }, .monotonic, .monotonic) orelse return error.Canceled, }; } - /// Windows only, called from blocked_apc state only. - fn checkCancelApc(s: Syscall) Io.Cancelable!void { - const thread = s.thread orelse return; - var prev = thread.status.load(.monotonic); - while (true) prev = switch (prev.cancelation) { - .none => unreachable, - .parked => unreachable, - .blocked_windows_dns => unreachable, - .blocked => unreachable, - .canceling => unreachable, - .canceled => unreachable, - .blocked_apc => return, - .blocked_canceling => thread.status.cmpxchgWeak(prev, .{ - .cancelation = .canceled, - .awaitable = prev.awaitable, - }, .monotonic, .monotonic) orelse return error.Canceled, - }; - } - /// Windows only, called from blocked_apc state only. - fn finishApc(s: Syscall) void { - const thread = s.thread orelse return; - var prev = thread.status.load(.monotonic); - while (true) prev = switch (prev.cancelation) { - .none => unreachable, - .parked => unreachable, - .blocked_windows_dns => unreachable, - .blocked => unreachable, - .canceling => unreachable, - .canceled => unreachable, - .blocked_apc => thread.status.cmpxchgWeak(prev, .{ - .cancelation = .none, - .awaitable = prev.awaitable, - }, .monotonic, .monotonic) orelse return, - .blocked_canceling => thread.status.cmpxchgWeak(prev, .{ - .cancelation = .canceling, - .awaitable = prev.awaitable, - }, .monotonic, .monotonic) orelse return, - }; - } /// Convenience wrapper which calls `finish`, then returns `err`. fn fail(s: Syscall, err: anytype) @TypeOf(err) { s.finish(); @@ -1566,8 +1496,6 @@ fn worker(t: *Threaded) void { .cancel_protection = .unblocked, .futex_waiter = undefined, .csprng = .{}, - .apc = undefined, - .interrupt_method = undefined, }; Thread.current = &thread; @@ -2172,8 +2100,8 @@ fn groupAsyncEager( .canceled => true, .parked => unreachable, .blocked => unreachable, - .blocked_apc => unreachable, - .blocked_windows_dns => unreachable, + .blocked_alertable => unreachable, + .blocked_alertable_canceling => unreachable, .blocked_canceling => unreachable, }; } else false; @@ -2184,8 +2112,8 @@ fn groupAsyncEager( .canceled => true, .parked => unreachable, .blocked => unreachable, - .blocked_apc => unreachable, - .blocked_windows_dns => unreachable, + .blocked_alertable => unreachable, + .blocked_alertable_canceling => unreachable, .blocked_canceling => unreachable, }; } else false; @@ -2364,8 +2292,8 @@ fn recancelInner() void { .canceling => unreachable, // called `recancel` but cancelation was already pending .parked => unreachable, .blocked => unreachable, - .blocked_apc => unreachable, - .blocked_windows_dns => unreachable, + .blocked_alertable => unreachable, + .blocked_alertable_canceling => unreachable, .blocked_canceling => unreachable, } } @@ -8324,36 +8252,37 @@ fn fileReadStreamingWindows(file: File, data: []const []u8) File.Reader.Error!us continue; }, .INVALID_PARAMETER => |err| return syscall.ntstatusBug(err), // streaming read of async mode file - else => |status| std.debug.panic("fileReadStreamingWindows NtReadFile returned {t}", .{status}), - //else => |status| return syscall.unexpectedNtstatus(status), + else => |status| return syscall.unexpectedNtstatus(status), } } - try syscall.toApc(.{ .handle = file.handle, .iosb = &io_status_block }); - while (true) { - switch (windows.ntdll.NtDelayExecution(1, &infinite)) { - .USER_APC => { - if (!done) { - // Other APC work was queued before calling into this function. - try syscall.checkCancelApc(); - continue; - } - break syscall.finishApc(); + // Once we get here we received PENDING so we must not return from the + // function until the operation completes. + defer while (!done) { + _ = windows.ntdll.NtDelayExecution(1, &infinite); + }; + + const alertable_syscall = syscall.toAlertable() catch |err| switch (err) { + error.Canceled => |e| { + _ = windows.ntdll.NtCancelIoFile(file.handle, &io_status_block); + return e; + }, + }; + defer alertable_syscall.finish(); + while (!done) { + _ = windows.ntdll.NtDelayExecution(1, &infinite); + alertable_syscall.checkCancel() catch |err| switch (err) { + error.Canceled => |e| { + _ = windows.ntdll.NtCancelIoFile(file.handle, &io_status_block); + return e; }, - .SUCCESS, .CANCELLED, .TIMEOUT, .ALERTED => { - try syscall.checkCancelApc(); - continue; - }, - else => |status| std.debug.panic("fileReadStreamingWindows NtDelayExecution returned {t}", .{status}), - //else => |status| return syscall.unexpectedNtstatus(status), - } + }; } } switch (io_status_block.u.Status) { .SUCCESS, .END_OF_FILE, .PIPE_BROKEN => {}, .ACCESS_DENIED => return error.AccessDenied, - else => |status| std.debug.panic("fileReadStreamingWindows IO_STATUS_BLOCK returned {t}", .{status}), - //else => |status| return windows.unexpectedStatus(status), + else => |status| return windows.unexpectedStatus(status), } return io_status_block.Information; } @@ -12331,7 +12260,7 @@ fn netLookupFallible( var res: *ws2_32.ADDRINFOEXW = undefined; const timeout: ?*ws2_32.timeval = null; while (true) { - // TODO: hook this up to cancelation with `Thread.Status.cancelation.blocked_windows_dns`. + // TODO: hook this up to cancelation with `NtDelayExecution` and APC callbacks. try Thread.checkCancel(); // TODO make this append to the queue eagerly rather than blocking until the whole thing finishes const rc: ws2_32.WinsockError = @enumFromInt(ws2_32.GetAddrInfoExW(name_w, port_w, .DNS, null, &hints, &res, timeout, null, null, null)); @@ -16014,8 +15943,8 @@ const parking_futex = struct { .canceled => break :cancelable, // status is still `.canceled` .parked => unreachable, .blocked => unreachable, - .blocked_apc => unreachable, - .blocked_windows_dns => unreachable, + .blocked_alertable => unreachable, + .blocked_alertable_canceling => unreachable, .blocked_canceling => unreachable, } // We could now be unparked for a cancelation at any time! @@ -16066,8 +15995,8 @@ const parking_futex = struct { }, .canceled => unreachable, .blocked => unreachable, - .blocked_apc => unreachable, - .blocked_windows_dns => unreachable, + .blocked_alertable => unreachable, + .blocked_alertable_canceling => unreachable, .blocked_canceling => unreachable, }, } @@ -16108,8 +16037,8 @@ const parking_futex = struct { .canceling => continue, // race with a canceler who hasn't called `removeCanceledWaiter` yet .canceled => unreachable, .blocked => unreachable, - .blocked_apc => unreachable, - .blocked_windows_dns => unreachable, + .blocked_alertable => unreachable, + .blocked_alertable_canceling => unreachable, .blocked_canceling => unreachable, } // We're waking this waiter. Remove them from the bucket and add them to our local list. @@ -16175,8 +16104,8 @@ const parking_sleep = struct { .canceled => break :cancelable, // status is still `.canceled` .parked => unreachable, .blocked => unreachable, - .blocked_apc => unreachable, - .blocked_windows_dns => unreachable, + .blocked_alertable => unreachable, + .blocked_alertable_canceling => unreachable, .blocked_canceling => unreachable, } while (park(deadline, null)) { @@ -16194,8 +16123,8 @@ const parking_sleep = struct { .none => unreachable, .canceled => unreachable, .blocked => unreachable, - .blocked_apc => unreachable, - .blocked_windows_dns => unreachable, + .blocked_alertable => unreachable, + .blocked_alertable_canceling => unreachable, .blocked_canceling => unreachable, } } else |err| switch (err) { @@ -16214,8 +16143,8 @@ const parking_sleep = struct { .none => unreachable, .canceled => unreachable, .blocked => unreachable, - .blocked_apc => unreachable, - .blocked_windows_dns => unreachable, + .blocked_alertable => unreachable, + .blocked_alertable_canceling => unreachable, .blocked_canceling => unreachable, }, } diff --git a/lib/std/os/windows/ntdll.zig b/lib/std/os/windows/ntdll.zig index c9d9b5ef0ebe5709046de05bf2ea4ea770e3b7a0..0d5152513b0276e11af3545909a38c5c3f0de171 100644 --- a/lib/std/os/windows/ntdll.zig +++ b/lib/std/os/windows/ntdll.zig @@ -596,7 +596,7 @@ pub extern "ntdll" fn NtDelayExecution( pub extern "ntdll" fn NtCancelIoFileEx( FileHandle: HANDLE, - IoRequestToCancel: ?*IO_STATUS_BLOCK, + IoRequestToCancel: *const IO_STATUS_BLOCK, IoStatusBlock: *IO_STATUS_BLOCK, ) callconv(.winapi) NTSTATUS;