From e0d9c04e35bf122b362697b543f5fb917a561182 Mon Sep 17 00:00:00 2001 From: Andrew Kelley Date: Thu, 22 Jan 2026 14:32:39 -0800 Subject: [PATCH] std.Io.Threaded: avoid extra fields of Thread As mlugg pointed out those race when a thread finishes an operation just after it is canceled and then that thread to picks up another task, resulting in these fields being potentially overwritten. This updates fileReadStreaming on Windows to handle being alerted, and then manage its own cancelation of the file I/O. --- lib/std/Io/Threaded.zig | 275 +++++++++++++---------------------- lib/std/os/windows/ntdll.zig | 2 +- 2 files changed, 103 insertions(+), 174 deletions(-) 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; -- 2.54.0