tiny.sys.thread

1pub const thread = @import("thread.zig");

1const std = @import("std");2const builtin = @import("builtin");3const sys_allocator = @import("allocator.zig");4const capabilities = @import("capabilities.zig");56const linux = std.os.linux;7const native_os = builtin.os.tag;8const posix = std.posix;910pub const required_capabilities = capabilities.host(&.{.threads});1112pub const ThreadSpecificError = error{13    UnsupportedPlatform,14    SystemResources,15};1617pub const SpawnError = error{18    UnsupportedPlatform,19    OutOfMemory,20    SystemResources,21};2223pub const Placement = struct {24    cpu_id: usize,25    numa_node: usize,26};2728pub const PlacementError = error{ UnsupportedPlatform, QueryFailed };29pub const CpuSet = linux.cpu_set_t;30pub const AffinityError = error{ UnsupportedPlatform, QueryFailed, ApplyFailed };3132pub const ThreadSpecificDestructor = *const fn (value: *anyopaque) callconv(.c) void;33pub const Id = std.Thread.Id;34pub const ThreadedIo = std.Io.Threaded;35pub const ThreadedIoOptions = std.Io.Threaded.InitOptions;3637pub const ThreadSpecificDestructorPolicy = enum {38    unsupported,39    native_spawn_exit,40};4142fn threadedIo() std.Io {43    return std.Io.Threaded.global_single_threaded.io();44}4546pub fn initThreadedIo(allocator: std.mem.Allocator, options: ThreadedIoOptions) ThreadedIo {47    return ThreadedIo.init(allocator, options);48}4950pub fn initProcessThreadedIo(allocator: std.mem.Allocator) ThreadedIo {51    return ThreadedIo.init(allocator, .{52        .environ = @import("env.zig").processEnviron(),53    });54}5556pub const Mutex = struct {57    raw: std.Io.Mutex = .init,5859    pub fn tryLock(self: *Mutex) bool {60        return self.raw.tryLock();61    }6263    pub fn lock(self: *Mutex) void {64        std.Io.Threaded.mutexLock(&self.raw);65    }6667    pub fn unlock(self: *Mutex) void {68        std.Io.Threaded.mutexUnlock(&self.raw);69    }70};7172pub const Condition = struct {73    raw: std.Io.Condition = .init,7475    pub fn wait(self: *Condition, mutex: *Mutex) void {76        self.raw.waitUncancelable(threadedIo(), &mutex.raw);77    }7879    pub fn signal(self: *Condition) void {80        self.raw.signal(threadedIo());81    }8283    pub fn broadcast(self: *Condition) void {84        self.raw.broadcast(threadedIo());85    }86};8788pub const Semaphore = struct {89    raw: std.Io.Semaphore = .{},9091    pub fn post(self: *Semaphore) void {92        self.raw.post(threadedIo());93    }9495    pub fn wait(self: *Semaphore) void {96        self.raw.waitUncancelable(threadedIo());97    }98};99100pub const ThreadSpecificKey = struct {101    index: usize = invalid_thread_specific_key,102};103104const invalid_thread_specific_key = std.math.maxInt(usize);105const native_thread_specific_destructor_iterations = 4;106107const NativeThreadSpecificRegistry = struct {108    mutex: std.atomic.Mutex = .unlocked,109    destructors: std.ArrayList(ThreadSpecificDestructor) = .empty,110111    fn createKey(self: *NativeThreadSpecificRegistry, allocator: std.mem.Allocator, callback: ThreadSpecificDestructor) ThreadSpecificError!ThreadSpecificKey {112        self.lock();113        defer self.mutex.unlock();114        const index = self.destructors.items.len;115        if (index == invalid_thread_specific_key) return error.SystemResources;116        self.destructors.append(allocator, callback) catch return error.SystemResources;117        return .{ .index = index };118    }119120    fn destructor(self: *NativeThreadSpecificRegistry, key_index: usize) ?ThreadSpecificDestructor {121        self.lock();122        defer self.mutex.unlock();123        if (key_index >= self.destructors.items.len) return null;124        return self.destructors.items[key_index];125    }126127    fn lock(self: *NativeThreadSpecificRegistry) void {128        while (!self.mutex.tryLock()) std.atomic.spinLoopHint();129    }130};131132const NativeThreadSpecificValues = struct {133    values: std.ArrayList(?*anyopaque) = .empty,134135    fn set(self: *NativeThreadSpecificValues, allocator: std.mem.Allocator, key: ThreadSpecificKey, value: ?*anyopaque) bool {136        if (key.index == invalid_thread_specific_key) return false;137        if (value == null and key.index >= self.values.items.len) return true;138        while (self.values.items.len <= key.index) {139            self.values.append(allocator, null) catch return false;140        }141        self.values.items[key.index] = value;142        return true;143    }144145    fn get(self: *NativeThreadSpecificValues, key: ThreadSpecificKey) ?*anyopaque {146        if (key.index == invalid_thread_specific_key or key.index >= self.values.items.len) return null;147        return self.values.items[key.index];148    }149150    fn runDestructors(self: *NativeThreadSpecificValues, allocator: std.mem.Allocator) void {151        defer {152            self.values.deinit(allocator);153            self.* = .{};154        }155156        var iteration: usize = 0;157        while (iteration < native_thread_specific_destructor_iterations) : (iteration += 1) {158            var called = false;159            var index: usize = 0;160            while (index < self.values.items.len) : (index += 1) {161                const value = self.values.items[index] orelse continue;162                const destructor = native_thread_specific_registry.destructor(index) orelse continue;163                self.values.items[index] = null;164                called = true;165                destructor(value);166            }167            if (!called) break;168        }169    }170};171172var native_thread_specific_registry: NativeThreadSpecificRegistry = .{};173threadlocal var native_thread_specific_values: NativeThreadSpecificValues = .{};174175pub const JoinHandle = struct {176    raw: std.Thread,177178    pub fn join(self: JoinHandle) void {179        self.raw.join();180    }181182    pub fn detach(self: JoinHandle) void {183        self.raw.detach();184    }185186    pub fn setName(187        self: JoinHandle,188        name: []const u8,189    ) std.Thread.SetNameError!void {190        try self.raw.setName(std.Options.debug_io, name);191    }192};193194pub fn blockUntilDifferent(195    previous: u32,196    current: *const std.atomic.Value(u32),197) u32 {198    while (true) {199        const next = current.load(.acquire);200        if (next != previous) return next;201        std.Io.futexWaitUncancelable(threadedIo(), u32, &current.raw, previous);202    }203}204205pub fn wakeAll(current: *std.atomic.Value(u32)) void {206    std.Io.futexWake(threadedIo(), u32, &current.raw, std.math.maxInt(u32));207}208209pub fn threadsSupported() bool {210    return comptime capabilities.current.supportsThreads();211}212213pub fn spawn(comptime func: anytype, args: anytype) SpawnError!JoinHandle {214    if (comptime !threadsSupported()) return error.UnsupportedPlatform;215    const Runner = struct {216        fn run(captured_args: @TypeOf(args)) void {217            defer runNativeThreadSpecificDestructors();218            _ = @call(.auto, func, captured_args);219        }220    };221    return .{ .raw = std.Thread.spawn(.{}, Runner.run, .{args}) catch |err| switch (err) {222        error.OutOfMemory => return error.OutOfMemory,223        else => return error.SystemResources,224    } };225}226227pub fn yield() void {228    if (comptime !threadsSupported()) return;229    std.Thread.yield() catch {};230}231232pub fn cpuCount() usize {233    if (comptime !threadsSupported()) return 1;234    return std.Thread.getCpuCount() catch 1;235}236237pub fn currentId() Id {238    if (comptime !threadsSupported()) return 0;239    return std.Thread.getCurrentId();240}241242pub fn currentPlacement() PlacementError!Placement {243    if (comptime native_os != .linux) return error.UnsupportedPlatform;244    var cpu_id: usize = 0;245    var numa_node: usize = 0;246    return switch (linux.errno(linux.getcpu(&cpu_id, &numa_node))) {247        .SUCCESS => .{ .cpu_id = cpu_id, .numa_node = numa_node },248        .NOSYS => error.UnsupportedPlatform,249        else => error.QueryFailed,250    };251}252253pub fn currentAffinity() AffinityError!CpuSet {254    if (comptime native_os != .linux) return error.UnsupportedPlatform;255    return posix.sched_getaffinity(0) catch return error.QueryFailed;256}257258pub fn setCurrentAffinity(set: *const CpuSet) AffinityError!void {259    if (comptime native_os != .linux) return error.UnsupportedPlatform;260    linux.sched_setaffinity(0, set) catch return error.ApplyFailed;261}262263pub fn threadSpecificDestructorPolicy() ThreadSpecificDestructorPolicy {264    if (comptime threadsSupported()) return .native_spawn_exit;265    return .unsupported;266}267268pub fn threadSpecificDestructorsSupported() bool {269    return threadSpecificDestructorPolicy() != .unsupported;270}271272pub fn createThreadSpecificKey(destructor: ThreadSpecificDestructor) ThreadSpecificError!ThreadSpecificKey {273    if (comptime !threadSpecificDestructorsSupported()) return error.UnsupportedPlatform;274    return native_thread_specific_registry.createKey(sys_allocator.processAllocator(), destructor);275}276277pub fn setThreadSpecificValue(key: ThreadSpecificKey, value: ?*anyopaque) bool {278    if (comptime !threadSpecificDestructorsSupported()) return false;279    return native_thread_specific_values.set(sys_allocator.processAllocator(), key, value);280}281282pub fn getThreadSpecificValue(key: ThreadSpecificKey) ?*anyopaque {283    if (comptime !threadSpecificDestructorsSupported()) return null;284    return native_thread_specific_values.get(key);285}286287fn runNativeThreadSpecificDestructors() void {288    if (comptime !threadSpecificDestructorsSupported()) return;289    native_thread_specific_values.runDestructors(sys_allocator.processAllocator());290}291292test "thread-specific destructor support uses sys-owned thread exit policy" {293    const expected: ThreadSpecificDestructorPolicy = if (comptime threadsSupported()) .native_spawn_exit else .unsupported;294    try std.testing.expectEqual(expected, threadSpecificDestructorPolicy());295    try std.testing.expectEqual(expected != .unsupported, threadSpecificDestructorsSupported());296}297298test "thread support follows host capabilities" {299    try std.testing.expectEqual(capabilities.current.supportsThreads(), threadsSupported());300    try std.testing.expect(cpuCount() >= 1);301    try std.testing.expectEqual(currentId(), currentId());302}303304test "current placement translates the host query" {305    if (native_os != .linux) {306        try std.testing.expectError(error.UnsupportedPlatform, currentPlacement());307        return;308    }309    const placement = try currentPlacement();310    try std.testing.expect(placement.cpu_id < std.math.maxInt(usize));311    try std.testing.expect(placement.numa_node < std.math.maxInt(usize));312    _ = try currentAffinity();313}314315test "semaphore routes through sys thread io policy" {316    var semaphore: Semaphore = .{};317    semaphore.post();318    semaphore.wait();319}320321const FutexProbe = struct {322    current: *std.atomic.Value(u32),323    ready: *std.atomic.Value(bool),324    observed: *u32,325326    fn wait(self: FutexProbe) void {327        self.ready.store(true, .release);328        self.observed.* = blockUntilDifferent(7, self.current);329    }330};331332test "futex boundary waits on and wakes an atomic word" {333    if (!threadsSupported()) return error.SkipZigTest;334    var current = std.atomic.Value(u32).init(7);335    var ready = std.atomic.Value(bool).init(false);336    var observed: u32 = 0;337    const handle = try spawn(FutexProbe.wait, .{338        FutexProbe{339            .current = &current,340            .ready = &ready,341            .observed = &observed,342        },343    });344    while (!ready.load(.acquire)) std.atomic.spinLoopHint();345    current.store(11, .release);346    wakeAll(&current);347    handle.join();348    try std.testing.expectEqual(@as(u32, 11), observed);349}350351const ThreadSpecificProbe = struct {352    fn destroy(value: *anyopaque) callconv(.c) void {353        const counter: *std.atomic.Value(usize) = @ptrCast(@alignCast(value));354        _ = counter.fetchAdd(1, .monotonic);355    }356357    fn setValue(key: ThreadSpecificKey, counter: *std.atomic.Value(usize)) void {358        if (!setThreadSpecificValue(key, counter)) @panic("thread specific set failed");359    }360361    fn isolate(key: ThreadSpecificKey, expected_missing: *bool, value: *usize, result: *usize) void {362        expected_missing.* = getThreadSpecificValue(key) == null;363        if (!setThreadSpecificValue(key, value)) @panic("thread specific set failed");364        result.* = @intFromPtr(getThreadSpecificValue(key) orelse return);365    }366};367368test "native thread-specific destructor runs when sys-spawned thread exits" {369    if (comptime threadSpecificDestructorPolicy() != .native_spawn_exit) return error.SkipZigTest;370371    const key = try createThreadSpecificKey(ThreadSpecificProbe.destroy);372    var counter = std.atomic.Value(usize).init(0);373    const thread = try spawn(ThreadSpecificProbe.setValue, .{ key, &counter });374    thread.join();375376    try std.testing.expectEqual(@as(usize, 1), counter.load(.monotonic));377}378379test "native thread-specific values are isolated per sys thread" {380    if (comptime threadSpecificDestructorPolicy() != .native_spawn_exit) return error.SkipZigTest;381382    const key = try createThreadSpecificKey(ThreadSpecificProbe.destroy);383    var main_counter = std.atomic.Value(usize).init(0);384    try std.testing.expect(setThreadSpecificValue(key, &main_counter));385    defer _ = setThreadSpecificValue(key, null);386387    var thread_missing_main = false;388    var thread_value: usize = 0;389    var thread_result: usize = 0;390    const thread = try spawn(ThreadSpecificProbe.isolate, .{ key, &thread_missing_main, &thread_value, &thread_result });391    thread.join();392393    try std.testing.expect(thread_missing_main);394    try std.testing.expectEqual(@intFromPtr(&thread_value), thread_result);395    try std.testing.expectEqual(@intFromPtr(&main_counter), @intFromPtr(getThreadSpecificValue(key) orelse return error.MainThreadSpecificValueMissing));396}