| ... | ... | @@ -628,6 +628,7 @@ const Thread = struct { |
| 628 | 628 | cancel_protection: Io.CancelProtection, |
| 629 | 629 | /// Always released when `Status.cancelation` is set to `.parked`. |
| 630 | 630 | futex_waiter: if (use_parking_futex) ?*parking_futex.Waiter else ?noreturn, |
| 631 | unpark_flag: UnparkFlag, |
| 631 | 632 | |
| 632 | 633 | csprng: Csprng, |
| 633 | 634 | |
| ... | ... | @@ -1018,6 +1019,7 @@ const Thread = struct { |
| 1018 | 1019 | if (thread.futex_waiter) |futex_waiter| { |
| 1019 | 1020 | parking_futex.removeCanceledWaiter(futex_waiter); |
| 1020 | 1021 | } |
| 1022 | if (need_unpark_flag) setUnparkFlag(&thread.unpark_flag); |
| 1021 | 1023 | unpark(&.{thread.id}, null); |
| 1022 | 1024 | return false; |
| 1023 | 1025 | }, |
| ... | ... | @@ -1559,6 +1561,7 @@ fn worker(t: *Threaded) void { |
| 1559 | 1561 | }), |
| 1560 | 1562 | .cancel_protection = .unblocked, |
| 1561 | 1563 | .futex_waiter = undefined, |
| 1564 | .unpark_flag = unpark_flag_init, |
| 1562 | 1565 | .csprng = .{}, |
| 1563 | 1566 | }; |
| 1564 | 1567 | Thread.current = &thread; |
| ... | ... | @@ -17007,6 +17010,7 @@ const parking_futex = struct { |
| 17007 | 17010 | /// * Unparking the thread (*after* the above, so that the `Waiter` does not go out of scope |
| 17008 | 17011 | /// while it is still in the `Bucket`). |
| 17009 | 17012 | thread_status: *std.atomic.Value(Thread.Status), |
| 17013 | unpark_flag: if (need_unpark_flag) *UnparkFlag else void, |
| 17010 | 17014 | }; |
| 17011 | 17015 | |
| 17012 | 17016 | fn bucketForAddress(address: usize) *Bucket { |
| ... | ... | @@ -17045,9 +17049,11 @@ const parking_futex = struct { |
| 17045 | 17049 | .address = @intFromPtr(ptr), |
| 17046 | 17050 | .tid = self_tid, |
| 17047 | 17051 | .thread_status = undefined, // populated in critical section |
| 17052 | .unpark_flag = undefined, // populated in critical section |
| 17048 | 17053 | }; |
| 17049 | 17054 | |
| 17050 | 17055 | var status_buf: std.atomic.Value(Thread.Status) = undefined; |
| 17056 | var unpark_flag_buf: UnparkFlag = unpark_flag_init; |
| 17051 | 17057 | |
| 17052 | 17058 | { |
| 17053 | 17059 | bucket.mutex.lock(); |
| ... | ... | @@ -17062,7 +17068,7 @@ const parking_futex = struct { |
| 17062 | 17068 | |
| 17063 | 17069 | // This is in the critical section to avoid marking the thread as parked until we're |
| 17064 | 17070 | // certain that we're actually going to park. |
| 17065 | | waiter.thread_status = status: { |
| 17071 | waiter.thread_status, waiter.unpark_flag = status: { |
| 17066 | 17072 | cancelable: { |
| 17067 | 17073 | if (uncancelable) break :cancelable; |
| 17068 | 17074 | const thread = opt_thread orelse break :cancelable; |
| ... | ... | @@ -17090,19 +17096,19 @@ const parking_futex = struct { |
| 17090 | 17096 | .blocked_canceling => unreachable, |
| 17091 | 17097 | } |
| 17092 | 17098 | // We could now be unparked for a cancelation at any time! |
| 17093 | | break :status &thread.status; |
| 17099 | break :status .{ &thread.status, if (need_unpark_flag) &thread.unpark_flag }; |
| 17094 | 17100 | } |
| 17095 | 17101 | // This is an uncancelable wait, so just use `status_buf`. Note that the value of |
| 17096 | 17102 | // `status_buf.awaitable` is irrelevant because this is only visible to futex code, |
| 17097 | 17103 | // while only cancelation cares about `awaitable`. |
| 17098 | 17104 | status_buf.raw = .{ .cancelation = .parked, .awaitable = .null }; |
| 17099 | | break :status &status_buf; |
| 17105 | break :status .{ &status_buf, if (need_unpark_flag) &unpark_flag_buf }; |
| 17100 | 17106 | }; |
| 17101 | 17107 | |
| 17102 | 17108 | bucket.waiters.append(&waiter.node); |
| 17103 | 17109 | } |
| 17104 | 17110 | |
| 17105 | | if (park(timeout, ptr, waiter.thread_status)) { |
| 17111 | if (park(timeout, ptr, waiter.unpark_flag)) { |
| 17106 | 17112 | // We were unparked by either `wake` or cancelation, so our current status is either |
| 17107 | 17113 | // `.none` or `.canceling`. In either case, they've already removed `waiter` from |
| 17108 | 17114 | // `bucket`, so we have nothing more to do! |
| ... | ... | @@ -17127,7 +17133,7 @@ const parking_futex = struct { |
| 17127 | 17133 | // to unpark us. Whoever did that will remove us from `bucket`. Wait for |
| 17128 | 17134 | // that (and drop the unpark request in doing so). |
| 17129 | 17135 | // New status is `.none` or `.canceling` respectively. |
| 17130 | | park(.none, ptr, waiter.thread_status) catch |e| switch (e) { |
| 17136 | park(.none, ptr, waiter.unpark_flag) catch |e| switch (e) { |
| 17131 | 17137 | error.Timeout => unreachable, |
| 17132 | 17138 | }; |
| 17133 | 17139 | }, |
| ... | ... | @@ -17201,6 +17207,7 @@ const parking_futex = struct { |
| 17201 | 17207 | waking_head = node.next; |
| 17202 | 17208 | const waiter: *Waiter = @fieldParentPtr("node", node); |
| 17203 | 17209 | unpark_buf[unpark_len] = waiter.tid; |
| 17210 | if (need_unpark_flag) setUnparkFlag(waiter.unpark_flag); |
| 17204 | 17211 | unpark_len += 1; |
| 17205 | 17212 | if (unpark_len == unpark_buf.len) { |
| 17206 | 17213 | unpark(&unpark_buf, ptr); |
| ... | ... | @@ -17249,7 +17256,7 @@ const parking_sleep = struct { |
| 17249 | 17256 | .blocked_canceling => unreachable, |
| 17250 | 17257 | } |
| 17251 | 17258 | } |
| 17252 | | if (park(timeout, null, &thread.status)) { |
| 17259 | if (park(timeout, null, if (need_unpark_flag) &thread.unpark_flag)) { |
| 17253 | 17260 | // The only reason this could possibly happen is cancelation. |
| 17254 | 17261 | const old_status = thread.status.load(.monotonic); |
| 17255 | 17262 | assert(old_status.cancelation == .canceling); |
| ... | ... | @@ -17272,7 +17279,7 @@ const parking_sleep = struct { |
| 17272 | 17279 | // us for a cancelation. Whoever did that will have called `unpark`, so |
| 17273 | 17280 | // drop that unpark request by waiting for it. |
| 17274 | 17281 | // Status is still `.canceling`. |
| 17275 | | park(.none, null, &thread.status) catch |e| switch (e) { |
| 17282 | park(.none, null, if (need_unpark_flag) &thread.unpark_flag) catch |e| switch (e) { |
| 17276 | 17283 | error.Timeout => unreachable, |
| 17277 | 17284 | }; |
| 17278 | 17285 | return; |
| ... | ... | @@ -17288,8 +17295,8 @@ const parking_sleep = struct { |
| 17288 | 17295 | } |
| 17289 | 17296 | } |
| 17290 | 17297 | // Uncancelable sleep; we expect not to be manually unparked. |
| 17291 | | var dummy_status: std.atomic.Value(Thread.Status) = .init(.{ .cancelation = .parked, .awaitable = .null }); |
| 17292 | | if (park(timeout, null, &dummy_status)) { |
| 17298 | var dummy_flag: UnparkFlag = unpark_flag_init; |
| 17299 | if (park(timeout, null, if (need_unpark_flag) &dummy_flag)) { |
| 17293 | 17300 | unreachable; // unexpected unpark |
| 17294 | 17301 | } else |err| switch (err) { |
| 17295 | 17302 | error.Timeout => return, |
| ... | ... | @@ -17322,7 +17329,7 @@ const ParkingMutex = struct { |
| 17322 | 17329 | } |
| 17323 | 17330 | }; |
| 17324 | 17331 | const Waiter = struct { |
| 17325 | | status: std.atomic.Value(Thread.Status), |
| 17332 | unpark_flag: UnparkFlag, |
| 17326 | 17333 | /// Never modified once the `Waiter` is in the linked list. |
| 17327 | 17334 | next: ?*Waiter, |
| 17328 | 17335 | /// Never modified once the `Waiter` is in the linked list. |
| ... | ... | @@ -17345,7 +17352,7 @@ const ParkingMutex = struct { |
| 17345 | 17352 | const self_tid = if (Thread.current) |t| t.id else std.Thread.getCurrentId(); |
| 17346 | 17353 | var waiter: Waiter = .{ |
| 17347 | 17354 | .next = old_waiter, |
| 17348 | | .status = .init(.{ .cancelation = .parked, .awaitable = .null }), |
| 17355 | .unpark_flag = unpark_flag_init, |
| 17349 | 17356 | .tid = self_tid, |
| 17350 | 17357 | }; |
| 17351 | 17358 | if (m.state.cmpxchgWeak( |
| ... | ... | @@ -17357,11 +17364,9 @@ const ParkingMutex = struct { |
| 17357 | 17364 | continue :state new_state; |
| 17358 | 17365 | } |
| 17359 | 17366 | // We're now in the list of waiters---park until we're given the lock. |
| 17360 | | park(.none, m, &waiter.status) catch |err| switch (err) { |
| 17367 | park(.none, m, if (need_unpark_flag) &waiter.unpark_flag) catch |err| switch (err) { |
| 17361 | 17368 | error.Timeout => unreachable, |
| 17362 | 17369 | }; |
| 17363 | | // We now hold the lock. |
| 17364 | | assert(waiter.status.load(.monotonic).cancelation == .none); |
| 17365 | 17370 | return; |
| 17366 | 17371 | }, |
| 17367 | 17372 | } |
| ... | ... | @@ -17383,7 +17388,7 @@ const ParkingMutex = struct { |
| 17383 | 17388 | _ => |last_state| { |
| 17384 | 17389 | // The logic here does not have ABA problems, and does some accesses non-atomically, |
| 17385 | 17390 | // because `Waiter.next` is owned by the lock holder (that's us!) once the waiter is |
| 17386 | | // in the linked list, up until we set `Waiter.status` to `.none`. |
| 17391 | // in the linked list, up until we unpark the waiter. |
| 17387 | 17392 | |
| 17388 | 17393 | // Run through the waiter list to the end to ensure fairness. This is obviously not |
| 17389 | 17394 | // ideal, but it shouldn't be a big deal in practice provided the critical section |
| ... | ... | @@ -17412,8 +17417,8 @@ const ParkingMutex = struct { |
| 17412 | 17417 | } |
| 17413 | 17418 | } |
| 17414 | 17419 | // Now we're ready to actually hand the lock over to them. |
| 17415 | | const tid = waiter.tid; // load this before the store below potentially invalidates `waiter` |
| 17416 | | waiter.status.store(.{ .cancelation = .none, .awaitable = .null }, .release); // release lock |
| 17420 | const tid = waiter.tid; // load before the unpark below potentially invalidates `waiter` |
| 17421 | if (need_unpark_flag) setUnparkFlag(&waiter.unpark_flag); |
| 17417 | 17422 | unpark(&.{tid}, m); |
| 17418 | 17423 | return; |
| 17419 | 17424 | }, |
| ... | ... | @@ -17451,15 +17456,34 @@ fn timeoutToWindowsInterval(timeout: Io.Timeout) ?windows.LARGE_INTEGER { |
| 17451 | 17456 | } |
| 17452 | 17457 | } |
| 17453 | 17458 | |
| 17459 | /// The API on NetBSD and Illumos sucks and can unpark spuriously (well, it *can't*, but signals |
| 17460 | /// cause an indistinguishable unblock, and libpthread really likes to leave unparks pending). |
| 17461 | /// As such, on these targets only, we need to pass around a flag to track whether a thread is |
| 17462 | /// "actually" being unparked. |
| 17463 | const need_unpark_flag = switch (native_os) { |
| 17464 | .netbsd, .illumos => true, |
| 17465 | else => false, |
| 17466 | }; |
| 17467 | const UnparkFlag = if (need_unpark_flag) std.atomic.Value(bool) else void; |
| 17468 | const unpark_flag_init: UnparkFlag = if (need_unpark_flag) .init(false); |
| 17469 | /// Must be called before `unpark`. After this function is called, the thread may be unparked at any |
| 17470 | /// time, so the caller must not reference values on its stack. |
| 17471 | fn setUnparkFlag(f: *UnparkFlag) void { |
| 17472 | f.store(true, .release); |
| 17473 | } |
| 17474 | |
| 17475 | /// The type passed into `unpark` for the thread ID. You'd think this was just a `std.Thread.Id`, |
| 17476 | /// but it seems that someone at Microsoft forgot how big their TIDs are supposed to be. |
| 17477 | const UnparkTid = switch (native_os) { |
| 17478 | .windows => usize, |
| 17479 | else => std.Thread.Id, |
| 17480 | }; |
| 17481 | |
| 17454 | 17482 | fn park( |
| 17455 | 17483 | timeout: Io.Timeout, |
| 17456 | 17484 | /// This value has no semantic effect, but may allow the OS to optimize the operation. |
| 17457 | 17485 | addr_hint: ?*const anyopaque, |
| 17458 | | /// The API on NetBSD and Illumos sucks and can unpark spuriously (well, it *can't*, but signals |
| 17459 | | /// cause an indistinguishable unblock, and libpthread really likes to leave unparks pending). |
| 17460 | | /// As such, on these targets only, this `status` is checked to determine if an unpark is real. |
| 17461 | | /// no way to differentiate |
| 17462 | | status: *std.atomic.Value(Thread.Status), |
| 17486 | unpark_flag: if (need_unpark_flag) *UnparkFlag else void, |
| 17463 | 17487 | ) error{Timeout}!void { |
| 17464 | 17488 | comptime assert(use_parking_futex or use_parking_sleep); |
| 17465 | 17489 | switch (native_os) { |
| ... | ... | @@ -17502,7 +17526,7 @@ fn park( |
| 17502 | 17526 | }; |
| 17503 | 17527 | // It's okay to pass the same timeout in a loop. If it's a duration, the OS actually |
| 17504 | 17528 | // writes the remaining time into the buffer when the syscall returns. |
| 17505 | | while (status.load(.monotonic).cancelation == .parked) { |
| 17529 | while (!unpark_flag.swap(false, .acquire)) { |
| 17506 | 17530 | switch (posix.errno(std.c._lwp_park( |
| 17507 | 17531 | if (clock_real) .REALTIME else .MONOTONIC, |
| 17508 | 17532 | .{ .ABSTIME = abstime }, |
| ... | ... | @@ -17523,12 +17547,6 @@ fn park( |
| 17523 | 17547 | else => comptime unreachable, |
| 17524 | 17548 | } |
| 17525 | 17549 | } |
| 17526 | | |
| 17527 | | const UnparkTid = switch (native_os) { |
| 17528 | | // `NtAlertMultipleThreadByThreadId` is weird and wants 64-bit thread IDs? |
| 17529 | | .windows => usize, |
| 17530 | | else => std.Thread.Id, |
| 17531 | | }; |
| 17532 | 17550 | /// `addr_hint` has no semantic effect, but may allow the OS to optimize this operation. |
| 17533 | 17551 | fn unpark(tids: []const UnparkTid, addr_hint: ?*const anyopaque) void { |
| 17534 | 17552 | comptime assert(use_parking_futex or use_parking_sleep); |
| ... | ... | @@ -17548,8 +17566,8 @@ fn unpark(tids: []const UnparkTid, addr_hint: ?*const anyopaque) void { |
| 17548 | 17566 | switch (posix.errno(std.c._lwp_unpark_all(@ptrCast(tids.ptr), tids.len, addr_hint))) { |
| 17549 | 17567 | .SUCCESS => return, |
| 17550 | 17568 | // For errors, fall through to a loop over `tids`, though this is only expected to |
| 17551 | | // be possible for ENOMEM (and even that is questionable). |
| 17552 | | .SRCH => recoverableOsBugDetected(), |
| 17569 | // be possible for ENOMEM (even that is questionable) and ESRCH (see comment below). |
| 17570 | .SRCH => {}, |
| 17553 | 17571 | .FAULT => recoverableOsBugDetected(), |
| 17554 | 17572 | .INVAL => recoverableOsBugDetected(), |
| 17555 | 17573 | .NOMEM => {}, |
| ... | ... | @@ -17558,7 +17576,11 @@ fn unpark(tids: []const UnparkTid, addr_hint: ?*const anyopaque) void { |
| 17558 | 17576 | for (tids) |tid| { |
| 17559 | 17577 | switch (posix.errno(std.c._lwp_unpark(@bitCast(tid), addr_hint))) { |
| 17560 | 17578 | .SUCCESS => {}, |
| 17561 | | .SRCH => recoverableOsBugDetected(), |
| 17579 | .SRCH => { |
| 17580 | // This can happen in a rare race: the thread might have been spuriously |
| 17581 | // unparked, so already observed the changing status, and from there have |
| 17582 | // exited. That's okay, because the thread has woken up like we wanted. |
| 17583 | }, |
| 17562 | 17584 | else => recoverableOsBugDetected(), |
| 17563 | 17585 | } |
| 17564 | 17586 | } |
Alex Rønne Petersen <alex@alexrp.com> 2026-02-04 20:44:45+01:00
Signed by SSH key SHA256:7B/LJ7bpR1eX8aCXSr4mtd5M45VMPKcx9zY8e95b5QM