const Thread = structconst Thread = struct {
next: ?*Thread,
id: std.Thread.Id,
handle: Handle,
status: std.atomic.Value(Status),
cancel_protection: Io.CancelProtection,
/// Always released when `Status.cancelation` is set to `.parked`.
futex_waiter: if (use_parking_futex) ?*parking_futex.Waiter else ?noreturn,
unpark_flag: UnparkFlag,
csprng: Csprng,
const Handle = Handle: {
if (std.Thread.use_pthreads) break :Handle std.c.pthread_t;
if (is_windows) break :Handle windows.HANDLE;
break :Handle void;
};
const Status = packed struct(usize) {
/// The specific values of these enum fields are chosen to simplify the implementation of
/// the transformations we need to apply to this state.
cancelation: enum(u3) {
/// The thread has not yet been canceled, and is not in a cancelable operation.
/// To request cancelation, just set the status to `.canceling`.
none = 0b000,
/// The thread is parked in a cancelable futex wait or sleep.
/// Only applicable if `use_parking_futex` or `use_parking_sleep`.
/// To request cancelation, set the status to `.canceling` and unpark the thread.
/// To unpark for another reason (futex wake), set the status to `.none` and unpark the thread.
parked = 0b001,
/// The thread is blocked in a cancelable system call.
/// 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 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`.
canceling = 0b110,
/// The thread has received and acknowledged a cancelation request.
/// If `recancel` is called, the status will revert to `.canceling`, but otherwise, the status
/// will not change for the remainder of this task's execution.
canceled = 0b111,
/// The thread is blocked in a cancelable system call, and is being
/// canceled. The thread which triggered the cancelation will send
/// signals to this thread until its status changes.
blocked_canceling = 0b101,
/// Windows-only: the thread is blocked in an alertable wait via
/// `NtDelayExecution`, and is being canceled. The thread which
/// triggered the cancelation will send signals to this thread
/// until its status changes.
blocked_alertable_canceling = 0b100,
},
/// We cannot turn this value back into a pointer. Instead, it exists so that a task can be
/// canceled by a cmpxchg on thread status: if it is running the task we want to cancel,
/// then update the `cancelation` field.
awaitable: AwaitableId,
};
const SignaleeId = if (std.Thread.use_pthreads) std.c.pthread_t else std.Thread.Id;
threadlocal var current: ?*Thread = null;
/// A value that does not alias any other thread id.
const invalid_id: std.Thread.Id = std.math.maxInt(std.Thread.Id);
fn currentId() std.Thread.Id {
return if (current) |t| t.id else std.Thread.getCurrentId();
}
/// The thread is neither in a syscall nor entering one, but we want to check for cancelation
/// anyway. If there is a pending cancel request, acknowledge it and return `error.Canceled`.
fn checkCancel() Io.Cancelable!void {
const thread = Thread.current orelse return;
switch (thread.cancel_protection) {
.blocked => return,
.unblocked => {},
}
// Here, unlike `Syscall.checkCancel`, it's not particularly likely that we're canceled, so
// it seems preferable to do a cheap atomic load and, in the unlikely case, a separate store
// to acknowledge. Besides, the state transitions we need here can't be done with one atomic
// OR/AND/XOR on `Status.cancelation`, so we don't actually have any other option.
const status = thread.status.load(.monotonic);
switch (status.cancelation) {
.parked => unreachable,
.blocked => unreachable,
.blocked_alertable => unreachable,
.blocked_alertable_canceling => unreachable,
.blocked_canceling => unreachable,
.none, .canceled => {},
.canceling => {
thread.status.store(.{
.cancelation = .canceled,
.awaitable = status.awaitable,
}, .monotonic);
return error.Canceled;
},
}
}
fn futexWaitUncancelable(ptr: *const u32, expect: u32, timeout_ns: ?u64) void {
return Thread.futexWaitInner(ptr, expect, true, timeout_ns) catch unreachable;
}
fn futexWait(ptr: *const u32, expect: u32, timeout_ns: ?u64) Io.Cancelable!void {
return Thread.futexWaitInner(ptr, expect, false, timeout_ns);
}
fn futexWaitInner(ptr: *const u32, expect: u32, uncancelable: bool, timeout_ns: ?u64) Io.Cancelable!void {
@branchHint(.cold);
if (builtin.single_threaded) unreachable; // nobody would ever wake us
if (use_parking_futex) {
return parking_futex.wait(
ptr,
expect,
uncancelable,
if (timeout_ns) |ns| .{ .duration = .{
.raw = .fromNanoseconds(ns),
.clock = .boot,
} } else .none,
);
} else if (builtin.cpu.arch.isWasm()) {
comptime assert(builtin.cpu.has(.wasm, .atomics));
// TODO implement cancelation for WASM futex waits by signaling the futex
if (!uncancelable) try Thread.checkCancel();
const to: i64 = if (timeout_ns) |ns| std.math.cast(i64, ns) orelse std.math.maxInt(i64) else -1;
const signed_expect: i32 = @bitCast(expect);
const result = asm volatile (
\\local.get %[ptr]
\\local.get %[expected]
\\local.get %[timeout]
\\memory.atomic.wait32 0
\\local.set %[ret]
: [ret] "=r" (-> u32),
: [ptr] "r" (ptr),
[expected] "r" (signed_expect),
[timeout] "r" (to),
);
switch (result) {
0 => {}, // ok
1 => {}, // expected != loaded
2 => {}, // timeout
else => assert(!is_debug),
}
} else switch (native_os) {
.linux => {
const linux = std.os.linux;
var ts_buffer: linux.timespec = undefined;
const ts: ?*linux.timespec = if (timeout_ns) |ns| ts: {
ts_buffer = timestampToPosix(ns);
break :ts &ts_buffer;
} else null;
const syscall: Syscall = if (uncancelable) .{ .thread = null } else try .start();
const rc = linux.futex_4arg(ptr, .{ .cmd = .WAIT, .private = true }, expect, ts);
syscall.finish();
switch (linux.errno(rc)) {
.SUCCESS => {}, // notified by `wake()`
.INTR => {}, // caller's responsibility to retry
.AGAIN => {}, // ptr.* != expect
.INVAL => {}, // possibly timeout overflow
.TIMEDOUT => {},
.FAULT => recoverableOsBugDetected(), // ptr was invalid
else => recoverableOsBugDetected(),
}
},
.driverkit, .ios, .maccatalyst, .macos, .tvos, .visionos, .watchos => {
const c = std.c;
const flags: c.UL = .{
.op = .COMPARE_AND_WAIT,
.NO_ERRNO = true,
};
const syscall: Syscall = if (uncancelable) .{ .thread = null } else try .start();
const status = switch (darwin_supports_ulock_wait2) {
true => c.__ulock_wait2(flags, ptr, expect, ns: {
const ns = timeout_ns orelse break :ns 0;
if (ns == 0) break :ns 1;
break :ns ns;
}, 0),
false => c.__ulock_wait(flags, ptr, expect, us: {
const ns = timeout_ns orelse break :us 0;
const us = std.math.lossyCast(u32, ns / std.time.ns_per_us);
if (us == 0) break :us 1;
break :us us;
}),
};
syscall.finish();
if (status >= 0) return;
switch (@as(c.E, @fromBackingInt(@intCast(-status)))) {
.INTR => {}, // spurious wake
// Address of the futex was paged out. This is unlikely, but possible in theory, and
// pthread/libdispatch on darwin bother to handle it. In this case we'll return
// without waiting, but the caller should retry anyway.
.FAULT => {},
.TIMEDOUT => {}, // timeout
else => recoverableOsBugDetected(),
}
},
.freebsd => {
const flags = @backingInt(std.c.UMTX_OP.WAIT_UINT_PRIVATE);
var tm_size: usize = 0;
var tm: std.c._umtx_time = undefined;
var tm_ptr: ?*const std.c._umtx_time = null;
if (timeout_ns) |ns| {
tm_ptr = &tm;
tm_size = @sizeOf(@TypeOf(tm));
tm.flags = 0; // use relative time not UMTX_ABSTIME
tm.clockid = .MONOTONIC;
tm.timeout = timestampToPosix(ns);
}
const syscall: Syscall = if (uncancelable) .{ .thread = null } else try .start();
const rc = std.c._umtx_op(@intFromPtr(ptr), flags, @as(c_ulong, expect), tm_size, @intFromPtr(tm_ptr));
syscall.finish();
if (is_debug) switch (posix.errno(rc)) {
.SUCCESS => {},
.FAULT => unreachable, // one of the args points to invalid memory
.INVAL => unreachable, // arguments should be correct
.TIMEDOUT => {}, // timeout
.INTR => {}, // spurious wake
else => unreachable,
};
},
.openbsd => {
var tm: std.c.timespec = undefined;
var tm_ptr: ?*const std.c.timespec = null;
if (timeout_ns) |ns| {
tm_ptr = &tm;
tm = timestampToPosix(ns);
}
const syscall: Syscall = if (uncancelable) .{ .thread = null } else try .start();
const rc = std.c.futex(
ptr,
std.c.FUTEX.WAIT | std.c.FUTEX.PRIVATE_FLAG,
@as(c_int, @bitCast(expect)),
tm_ptr,
null, // uaddr2 is ignored
);
syscall.finish();
if (is_debug) switch (posix.errno(rc)) {
.SUCCESS => {},
.NOSYS => unreachable, // constant op known good value
.AGAIN => {}, // contents of uaddr != val
.INVAL => unreachable, // invalid timeout
.TIMEDOUT => {}, // timeout
.INTR => {}, // a signal arrived
.CANCELED => {}, // a signal arrived and SA_RESTART was set
else => unreachable,
};
},
.dragonfly => {
var timeout_us: c_int = undefined;
if (timeout_ns) |ns| {
timeout_us = std.math.cast(c_int, ns / std.time.ns_per_us) orelse std.math.maxInt(c_int);
} else {
timeout_us = 0;
}
const syscall: Syscall = if (uncancelable) .{ .thread = null } else try .start();
const rc = std.c.umtx_sleep(@ptrCast(ptr), @bitCast(expect), timeout_us);
syscall.finish();
if (is_debug) switch (std.posix.errno(rc)) {
.SUCCESS => {},
.BUSY => {}, // ptr != expect
.AGAIN => {}, // maybe timed out, or paged out, or hit 2s kernel refresh
.INTR => {}, // spurious wake
.INVAL => unreachable, // invalid timeout
else => unreachable,
};
},
else => @compileError("unimplemented: futexWait"),
}
}
fn futexWake(ptr: *const u32, max_waiters: u32) void {
@branchHint(.cold);
assert(max_waiters != 0);
if (builtin.single_threaded) return; // nothing to wake up
if (use_parking_futex) {
return parking_futex.wake(ptr, max_waiters);
} else if (builtin.cpu.arch.isWasm()) {
comptime assert(builtin.cpu.has(.wasm, .atomics));
const woken_count = asm volatile (
\\local.get %[ptr]
\\local.get %[waiters]
\\memory.atomic.notify 0
\\local.set %[ret]
: [ret] "=r" (-> u32),
: [ptr] "r" (ptr),
[waiters] "r" (max_waiters),
);
_ = woken_count; // can be 0 when linker flag 'shared-memory' is not enabled
} else switch (native_os) {
.linux => {
const linux = std.os.linux;
switch (linux.errno(linux.futex_3arg(
ptr,
.{ .cmd = .WAKE, .private = true },
@min(max_waiters, std.math.maxInt(i32)),
))) {
.SUCCESS => return, // successful wake up
.INVAL => return, // invalid futex_wait() on ptr done elsewhere
.FAULT => return, // pointer became invalid while doing the wake
else => return recoverableOsBugDetected(), // deadlock due to operating system bug
}
},
.driverkit, .ios, .maccatalyst, .macos, .tvos, .visionos, .watchos => {
const c = std.c;
const flags: c.UL = .{
.op = .COMPARE_AND_WAIT,
.NO_ERRNO = true,
.WAKE_ALL = max_waiters > 1,
};
while (true) {
const status = c.__ulock_wake(flags, ptr, 0);
if (status >= 0) return;
switch (@as(c.E, @fromBackingInt(@intCast(-status)))) {
.INTR, .CANCELED => continue, // spurious wake()
.FAULT => unreachable, // __ulock_wake doesn't generate EFAULT according to darwin pthread_cond_t
.NOENT => return, // nothing was woken up
.ALREADY => unreachable, // only for UL.Op.WAKE_THREAD
else => unreachable, // deadlock due to operating system bug
}
}
},
.freebsd => {
const rc = std.c._umtx_op(
@intFromPtr(ptr),
@backingInt(std.c.UMTX_OP.WAKE_PRIVATE),
@as(c_ulong, @min(max_waiters, std.math.maxInt(c_int))),
0, // there is no timeout struct
0, // there is no timeout struct pointer
);
switch (posix.errno(rc)) {
.SUCCESS => {},
.FAULT => {}, // it's ok if the ptr doesn't point to valid memory
.INVAL => unreachable, // arguments should be correct
else => unreachable, // deadlock due to operating system bug
}
},
.openbsd => {
const rc = std.c.futex(
ptr,
std.c.FUTEX.WAKE | std.c.FUTEX.PRIVATE_FLAG,
@min(max_waiters, std.math.maxInt(c_int)),
null, // timeout is ignored
null, // uaddr2 is ignored
);
assert(rc >= 0);
},
.dragonfly => {
// will generally return 0 unless the address is bad
_ = std.c.umtx_wakeup(
@ptrCast(ptr),
@min(max_waiters, std.math.maxInt(c_int)),
);
},
else => @compileError("unimplemented: futexWake"),
}
}
/// Cancels `thread` if it is working on `awaitable`.
///
/// 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) bool {
var status = thread.status.load(.monotonic);
while (true) {
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 false,
.parked => thread.status.cmpxchgWeak(
.{ .cancelation = .parked, .awaitable = awaitable },
.{ .cancelation = .canceling, .awaitable = awaitable },
.acquire, // acquire `thread.futex_waiter`
.monotonic,
) orelse {
if (!use_parking_futex and !use_parking_sleep) unreachable;
if (thread.futex_waiter) |futex_waiter| {
parking_futex.removeCanceledWaiter(futex_waiter);
}
if (need_unpark_flag) setUnparkFlag(&thread.unpark_flag);
unpark(&.{thread.id}, null);
return false;
},
.blocked => thread.status.cmpxchgWeak(
.{ .cancelation = .blocked, .awaitable = awaitable },
.{ .cancelation = .blocked_canceling, .awaitable = awaitable },
.monotonic,
.monotonic,
) orelse return true,
.blocked_alertable => thread.status.cmpxchgWeak(
.{ .cancelation = .blocked_alertable, .awaitable = awaitable },
.{ .cancelation = .blocked_alertable_canceling, .awaitable = awaitable },
.monotonic,
.monotonic,
) orelse {
if (!is_windows) unreachable;
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 false;
},
.blocked_canceling => unreachable, // `awaitable` has not been canceled before now
.blocked_alertable_canceling => unreachable, // `awaitable` has not been canceled before now
};
}
}
/// Sends a signal to `thread` if it is still blocked in a syscall (i.e. has not yet observed
/// the cancelation request from `cancelAwaitable`).
///
/// Unfortunately, the signal could arrive before the syscall actually starts, so the interrupt
/// is missed. To handle this, we may need to send multiple signals. As such, if this function
/// returns `true`, then it should be called again after a short delay to send another signal if
/// 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) 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.
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 @backingInt(cached_pid);
const pid = std.os.linux.getpid();
@atomicStore(Pid, &t.pid, @fromBackingInt(@intCast(pid)), .monotonic);
break :pid pid;
};
return switch (std.os.linux.tgkill(pid, @bitCast(thread.id), .IO)) {
0 => true,
else => false,
};
},
.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
.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 => {
// The thread is working on `awaitable`, but no longer needs signaling (they already
// woke up and saw the cancelation).
return false;
},
}
}
/// Like a `*Thread`, but 2 bits smaller than a pointer (because the LSBs are always 0 due to
/// alignment) so that those two bits can be used in a `packed struct`.
const PackedPtr = enum(@Int(.unsigned, @bitSizeOf(usize) - 2)) {
null = 0,
all_ones = std.math.maxInt(@Int(.unsigned, @bitSizeOf(usize) - 2)),
_,
const Split = packed struct(usize) { low: u2, high: PackedPtr };
fn pack(ptr: *Thread) PackedPtr {
const split: Split = @bitCast(@intFromPtr(ptr));
assert(split.low == 0);
return split.high;
}
fn unpack(ptr: PackedPtr) ?*Thread {
const split: Split = .{ .low = 0, .high = ptr };
return @ptrFromInt(@as(usize, @bitCast(split)));
}
};
}