lib/sys/src/thread.zig

daab053ee43316e1809a84551d573ddd1e5bf3d2

  1 const std = @import("std");
  2 const builtin = @import("builtin");
  3 const sys_allocator = @import("allocator.zig");
  4 const capabilities = @import("capabilities.zig");
  5 
  6 const linux = std.os.linux;
  7 const native_os = builtin.os.tag;
  8 const posix = std.posix;
  9 
 10 pub const required_capabilities = capabilities.host(&.{.threads});
 11 
 12 pub const ThreadSpecificError = error{
 13     UnsupportedPlatform,
 14     SystemResources,
 15 };
 16 
 17 pub const SpawnError = error{
 18     UnsupportedPlatform,
 19     OutOfMemory,
 20     SystemResources,
 21 };
 22 
 23 pub const Placement = struct {
 24     cpu_id: usize,
 25     numa_node: usize,
 26 };
 27 
 28 pub const PlacementError = error{ UnsupportedPlatform, QueryFailed };
 29 pub const CpuSet = linux.cpu_set_t;
 30 pub const AffinityError = error{ UnsupportedPlatform, QueryFailed, ApplyFailed };
 31 
 32 pub const ThreadSpecificDestructor = *const fn (value: *anyopaque) callconv(.c) void;
 33 pub const Id = std.Thread.Id;
 34 pub const ThreadedIo = std.Io.Threaded;
 35 pub const ThreadedIoOptions = std.Io.Threaded.InitOptions;
 36 
 37 pub const ThreadSpecificDestructorPolicy = enum {
 38     unsupported,
 39     native_spawn_exit,
 40 };
 41 
 42 fn threadedIo() std.Io {
 43     return std.Io.Threaded.global_single_threaded.io();
 44 }
 45 
 46 pub fn initThreadedIo(allocator: std.mem.Allocator, options: ThreadedIoOptions) ThreadedIo {
 47     return ThreadedIo.init(allocator, options);
 48 }
 49 
 50 pub fn initProcessThreadedIo(allocator: std.mem.Allocator) ThreadedIo {
 51     return ThreadedIo.init(allocator, .{
 52         .environ = @import("env.zig").processEnviron(),
 53     });
 54 }
 55 
 56 pub const Mutex = struct {
 57     raw: std.Io.Mutex = .init,
 58 
 59     pub fn tryLock(self: *Mutex) bool {
 60         return self.raw.tryLock();
 61     }
 62 
 63     pub fn lock(self: *Mutex) void {
 64         std.Io.Threaded.mutexLock(&self.raw);
 65     }
 66 
 67     pub fn unlock(self: *Mutex) void {
 68         std.Io.Threaded.mutexUnlock(&self.raw);
 69     }
 70 };
 71 
 72 pub const Condition = struct {
 73     raw: std.Io.Condition = .init,
 74 
 75     pub fn wait(self: *Condition, mutex: *Mutex) void {
 76         self.raw.waitUncancelable(threadedIo(), &mutex.raw);
 77     }
 78 
 79     pub fn signal(self: *Condition) void {
 80         self.raw.signal(threadedIo());
 81     }
 82 
 83     pub fn broadcast(self: *Condition) void {
 84         self.raw.broadcast(threadedIo());
 85     }
 86 };
 87 
 88 pub const Semaphore = struct {
 89     raw: std.Io.Semaphore = .{},
 90 
 91     pub fn post(self: *Semaphore) void {
 92         self.raw.post(threadedIo());
 93     }
 94 
 95     pub fn wait(self: *Semaphore) void {
 96         self.raw.waitUncancelable(threadedIo());
 97     }
 98 };
 99 
100 pub const ThreadSpecificKey = struct {
101     index: usize = invalid_thread_specific_key,
102 };
103 
104 const invalid_thread_specific_key = std.math.maxInt(usize);
105 const native_thread_specific_destructor_iterations = 4;
106 
107 const NativeThreadSpecificRegistry = struct {
108     mutex: std.atomic.Mutex = .unlocked,
109     destructors: std.ArrayList(ThreadSpecificDestructor) = .empty,
110 
111     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     }
119 
120     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     }
126 
127     fn lock(self: *NativeThreadSpecificRegistry) void {
128         while (!self.mutex.tryLock()) std.atomic.spinLoopHint();
129     }
130 };
131 
132 const NativeThreadSpecificValues = struct {
133     values: std.ArrayList(?*anyopaque) = .empty,
134 
135     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     }
144 
145     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     }
149 
150     fn runDestructors(self: *NativeThreadSpecificValues, allocator: std.mem.Allocator) void {
151         defer {
152             self.values.deinit(allocator);
153             self.* = .{};
154         }
155 
156         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 };
171 
172 var native_thread_specific_registry: NativeThreadSpecificRegistry = .{};
173 threadlocal var native_thread_specific_values: NativeThreadSpecificValues = .{};
174 
175 pub const JoinHandle = struct {
176     raw: std.Thread,
177 
178     pub fn join(self: JoinHandle) void {
179         self.raw.join();
180     }
181 
182     pub fn detach(self: JoinHandle) void {
183         self.raw.detach();
184     }
185 
186     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 };
193 
194 pub 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 }
204 
205 pub fn wakeAll(current: *std.atomic.Value(u32)) void {
206     std.Io.futexWake(threadedIo(), u32, &current.raw, std.math.maxInt(u32));
207 }
208 
209 pub fn threadsSupported() bool {
210     return comptime capabilities.current.supportsThreads();
211 }
212 
213 pub 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 }
226 
227 pub fn yield() void {
228     if (comptime !threadsSupported()) return;
229     std.Thread.yield() catch {};
230 }
231 
232 pub fn cpuCount() usize {
233     if (comptime !threadsSupported()) return 1;
234     return std.Thread.getCpuCount() catch 1;
235 }
236 
237 pub fn currentId() Id {
238     if (comptime !threadsSupported()) return 0;
239     return std.Thread.getCurrentId();
240 }
241 
242 pub 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 }
252 
253 pub fn currentAffinity() AffinityError!CpuSet {
254     if (comptime native_os != .linux) return error.UnsupportedPlatform;
255     return posix.sched_getaffinity(0) catch return error.QueryFailed;
256 }
257 
258 pub 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 }
262 
263 pub fn threadSpecificDestructorPolicy() ThreadSpecificDestructorPolicy {
264     if (comptime threadsSupported()) return .native_spawn_exit;
265     return .unsupported;
266 }
267 
268 pub fn threadSpecificDestructorsSupported() bool {
269     return threadSpecificDestructorPolicy() != .unsupported;
270 }
271 
272 pub 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 }
276 
277 pub 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 }
281 
282 pub fn getThreadSpecificValue(key: ThreadSpecificKey) ?*anyopaque {
283     if (comptime !threadSpecificDestructorsSupported()) return null;
284     return native_thread_specific_values.get(key);
285 }
286 
287 fn runNativeThreadSpecificDestructors() void {
288     if (comptime !threadSpecificDestructorsSupported()) return;
289     native_thread_specific_values.runDestructors(sys_allocator.processAllocator());
290 }
291 
292 test "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 }
297 
298 test "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 }
303 
304 test "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 }
314 
315 test "semaphore routes through sys thread io policy" {
316     var semaphore: Semaphore = .{};
317     semaphore.post();
318     semaphore.wait();
319 }
320 
321 const FutexProbe = struct {
322     current: *std.atomic.Value(u32),
323     ready: *std.atomic.Value(bool),
324     observed: *u32,
325 
326     fn wait(self: FutexProbe) void {
327         self.ready.store(true, .release);
328         self.observed.* = blockUntilDifferent(7, self.current);
329     }
330 };
331 
332 test "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 }
350 
351 const 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     }
356 
357     fn setValue(key: ThreadSpecificKey, counter: *std.atomic.Value(usize)) void {
358         if (!setThreadSpecificValue(key, counter)) @panic("thread specific set failed");
359     }
360 
361     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 };
367 
368 test "native thread-specific destructor runs when sys-spawned thread exits" {
369     if (comptime threadSpecificDestructorPolicy() != .native_spawn_exit) return error.SkipZigTest;
370 
371     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();
375 
376     try std.testing.expectEqual(@as(usize, 1), counter.load(.monotonic));
377 }
378 
379 test "native thread-specific values are isolated per sys thread" {
380     if (comptime threadSpecificDestructorPolicy() != .native_spawn_exit) return error.SkipZigTest;
381 
382     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);
386 
387     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();
392 
393     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 }