From b47080026508f411c9fbda53d62d15dc5e90c2f9 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 a9a8586da682b3423b177509145a83a9e11405ee..3d7520dbe375716629ad6447ba92c2959230289c 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; @@ -2176,8 +2104,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; @@ -2188,8 +2116,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; @@ -2368,8 +2296,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, } } @@ -8467,36 +8395,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; } @@ -12519,7 +12448,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)); @@ -16253,8 +16182,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! @@ -16305,8 +16234,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, }, } @@ -16347,8 +16276,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. @@ -16414,8 +16343,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)) { @@ -16433,8 +16362,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) { @@ -16453,8 +16382,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 04b0a288debe648de202ed988570f1bb97bbaa6f..d68cd1494b8392381187de4316a7155351687eae 100644 --- a/lib/std/os/windows/ntdll.zig +++ b/lib/std/os/windows/ntdll.zig @@ -601,7 +601,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