authorgravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2025-11-21 08:04:52-08:00
committergravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2025-11-21 08:04:52-08:00
logc6e7c2cb80e9362da1f16ab3a229e40157e583eb
tree1ee12b497607edcedded509e0c236f4e5290907f
parentc91dec3b6fee7f549eb91e67ea8a0d81cbdbfa4d

WIP

3 files changed, 818 insertions(+), 557 deletions(-)

lib/std/Io.zig-5
......@@ -633,11 +633,6 @@ pub const VTable = struct {
633633 result: []u8,
634634 result_alignment: std.mem.Alignment,
635635 ) void,
636 /// Returns whether the current thread of execution is known to have
637 /// been requested to cancel.
638 ///
639 /// Thread-safe.
640 cancelRequested: *const fn (?*anyopaque) bool,
641636
642637 /// Executes `start` asynchronously in a manner such that it cleans itself
643638 /// up. This mode does not support results, await, or cancel.
lib/std/Io/Threaded.zig+812-239
......@@ -12,22 +12,16 @@ const Io = std.Io;
1212const net = std.Io.net;
1313const HostName = std.Io.net.HostName;
1414const IpAddress = std.Io.net.IpAddress;
15const Allocator = std.mem.Allocator;
1615const Alignment = std.mem.Alignment;
1716const assert = std.debug.assert;
1817const posix = std.posix;
1918
20/// Thread-safe.
21allocator: Allocator,
22mutex: std.Thread.Mutex = .{},
23cond: std.Thread.Condition = .{},
24run_queue: std.SinglyLinkedList = .{},
25join_requested: bool = false,
26threads: std.ArrayList(std.Thread),
27stack_size: usize,
28thread_capacity: std.atomic.Value(ThreadCapacity),
29thread_capacity_error: ?std.Thread.CpuCountError,
30concurrent_count: usize,
19main_thread: Thread,
20stack_size: usize = default_stack_size,
21capacity: std.atomic.Value(Capacity),
22capacity_error: ?std.Thread.CpuCountError,
23concurrent_limit: Io.Limit = .unlimited,
24pid: Pid = .unknown,
3125
3226wsa: if (is_windows) Wsa else struct {} = .{},
3327
......@@ -35,22 +29,626 @@ have_signal_handler: bool,
3529old_sig_io: if (have_sig_io) posix.Sigaction else void,
3630old_sig_pipe: if (have_sig_pipe) posix.Sigaction else void,
3731
38pub const ThreadCapacity = enum(usize) {
32pub const Pid = enum(if (posix.pid_t == void) u0 else posix.pid_t) {
3933 unknown = 0,
4034 _,
35};
36
37pub const Thread = struct {
38 /// The value that needs to be passed to pthread_kill or tgkill in order to
39 /// send a signal.
40 signal_id: SignalId,
41 /// Points to the next thread in the list. Singly-linked so that
42 /// it can be updated lock-free.
43 list_node: std.SinglyLinkedList.Node = .{},
44 run_queue: std.SinglyLinkedList.Node = .{},
45 current_closure: ?*Closure = null,
46 completion: Completion,
47 mutex: std.Thread.Mutex,
48 cond: std.Thread.Condition,
49 join_requested: bool,
50
51 threadlocal var current: *Thread = undefined;
52
53 const SignalId = if (use_pthreads) std.c.pthread_t else std.Thread.Id;
54
55 const Completion = switch (native_os) {
56 .windows => @compileError("TODO"),
57 .linux => struct {
58 state: State = State.init(.running),
59 child_tid: std.atomic.Value(i32) = std.atomic.Value(i32).init(1),
60 parent_tid: i32 = undefined,
61 mapped: []align(std.heap.page_size_min) u8,
62
63 /// State to synchronize detachment of spawner thread to spawned thread
64 const State = std.atomic.Value(enum(switch (builtin.zig_backend) {
65 .stage2_riscv64 => u32,
66 else => u8,
67 }) {
68 running,
69 detached,
70 completed,
71 });
72
73
74 /// Calls `munmap(mapped.ptr, mapped.len)` then `exit(1)` without touching the stack (which lives in `mapped.ptr`).
75 /// Ported over from musl libc's pthread detached implementation:
76 /// https://github.com/ifduyue/musl/search?q=__unmapself
77 fn freeAndExit(self: *Completion) noreturn {
78 switch (builtin.target.cpu.arch) {
79 .x86 => asm volatile (
80 \\ movl $91, %%eax # SYS_munmap
81 \\ movl %[ptr], %%ebx
82 \\ movl %[len], %%ecx
83 \\ int $128
84 \\ movl $1, %%eax # SYS_exit
85 \\ movl $0, %%ebx
86 \\ int $128
87 :
88 : [ptr] "r" (@intFromPtr(self.mapped.ptr)),
89 [len] "r" (self.mapped.len),
90 : .{ .memory = true }),
91 .x86_64 => asm volatile (switch (builtin.target.abi) {
92 .gnux32, .muslx32 =>
93 \\ movl $0x4000000b, %%eax # SYS_munmap
94 \\ syscall
95 \\ movl $0x4000003c, %%eax # SYS_exit
96 \\ xor %%rdi, %%rdi
97 \\ syscall
98 ,
99 else =>
100 \\ movl $11, %%eax # SYS_munmap
101 \\ syscall
102 \\ movl $60, %%eax # SYS_exit
103 \\ xor %%rdi, %%rdi
104 \\ syscall
105 ,
106 }
107 :
108 : [ptr] "{rdi}" (@intFromPtr(self.mapped.ptr)),
109 [len] "{rsi}" (self.mapped.len),
110 ),
111 .arm, .armeb, .thumb, .thumbeb => asm volatile (
112 \\ mov r7, #91 // SYS_munmap
113 \\ mov r0, %[ptr]
114 \\ mov r1, %[len]
115 \\ svc 0
116 \\ mov r7, #1 // SYS_exit
117 \\ mov r0, #0
118 \\ svc 0
119 :
120 : [ptr] "r" (@intFromPtr(self.mapped.ptr)),
121 [len] "r" (self.mapped.len),
122 : .{ .memory = true }),
123 .aarch64, .aarch64_be => asm volatile (
124 \\ mov x8, #215 // SYS_munmap
125 \\ mov x0, %[ptr]
126 \\ mov x1, %[len]
127 \\ svc 0
128 \\ mov x8, #93 // SYS_exit
129 \\ mov x0, #0
130 \\ svc 0
131 :
132 : [ptr] "r" (@intFromPtr(self.mapped.ptr)),
133 [len] "r" (self.mapped.len),
134 : .{ .memory = true }),
135 .alpha => asm volatile (
136 \\ ldi $0, 73 # SYS_munmap
137 \\ mov %[ptr], $16
138 \\ mov %[len], $17
139 \\ callsys
140 \\ ldi $0, 1 # SYS_exit
141 \\ ldi $16, 0
142 \\ callsys
143 :
144 : [ptr] "r" (@intFromPtr(self.mapped.ptr)),
145 [len] "r" (self.mapped.len),
146 : .{ .memory = true }),
147 .hexagon => asm volatile (
148 \\ r6 = #215 // SYS_munmap
149 \\ r0 = %[ptr]
150 \\ r1 = %[len]
151 \\ trap0(#1)
152 \\ r6 = #93 // SYS_exit
153 \\ r0 = #0
154 \\ trap0(#1)
155 :
156 : [ptr] "r" (@intFromPtr(self.mapped.ptr)),
157 [len] "r" (self.mapped.len),
158 : .{ .memory = true }),
159 .hppa => asm volatile (
160 \\ ldi 91, %%r20 /* SYS_munmap */
161 \\ copy %[ptr], %%r26
162 \\ copy %[len], %%r25
163 \\ ble 0x100(%%sr2, %%r0)
164 \\ ldi 1, %%r20 /* SYS_exit */
165 \\ ldi 0, %%r26
166 \\ ble 0x100(%%sr2, %%r0)
167 :
168 : [ptr] "r" (@intFromPtr(self.mapped.ptr)),
169 [len] "r" (self.mapped.len),
170 : .{ .memory = true }),
171 .m68k => asm volatile (
172 \\ move.l #91, %%d0 // SYS_munmap
173 \\ move.l %[ptr], %%d1
174 \\ move.l %[len], %%d2
175 \\ trap #0
176 \\ move.l #1, %%d0 // SYS_exit
177 \\ move.l #0, %%d1
178 \\ trap #0
179 :
180 : [ptr] "r" (@intFromPtr(self.mapped.ptr)),
181 [len] "r" (self.mapped.len),
182 : .{ .memory = true }),
183 .microblaze, .microblazeel => asm volatile (
184 \\ ori r12, r0, 91 # SYS_munmap
185 \\ ori r5, %[ptr], 0
186 \\ ori r6, %[len], 0
187 \\ brki r14, 0x8
188 \\ ori r12, r0, 1 # SYS_exit
189 \\ or r5, r0, r0
190 \\ brki r14, 0x8
191 :
192 : [ptr] "r" (@intFromPtr(self.mapped.ptr)),
193 [len] "r" (self.mapped.len),
194 : .{ .memory = true }),
195 // We set `sp` to the address of the current function as a workaround for a Linux
196 // kernel bug that caused syscalls to return EFAULT if the stack pointer is invalid.
197 // The bug was introduced in 46e12c07b3b9603c60fc1d421ff18618241cb081 and fixed in
198 // 7928eb0370d1133d0d8cd2f5ddfca19c309079d5.
199 .mips, .mipsel => asm volatile (
200 \\ move $sp, $t9
201 \\ li $v0, 4091 # SYS_munmap
202 \\ move $a0, %[ptr]
203 \\ move $a1, %[len]
204 \\ syscall
205 \\ li $v0, 4001 # SYS_exit
206 \\ li $a0, 0
207 \\ syscall
208 :
209 : [ptr] "r" (@intFromPtr(self.mapped.ptr)),
210 [len] "r" (self.mapped.len),
211 : .{ .memory = true }),
212 .mips64, .mips64el => asm volatile (switch (builtin.target.abi) {
213 .gnuabin32, .muslabin32 =>
214 \\ li $v0, 6011 # SYS_munmap
215 \\ move $a0, %[ptr]
216 \\ move $a1, %[len]
217 \\ syscall
218 \\ li $v0, 6058 # SYS_exit
219 \\ li $a0, 0
220 \\ syscall
221 ,
222 else =>
223 \\ li $v0, 5011 # SYS_munmap
224 \\ move $a0, %[ptr]
225 \\ move $a1, %[len]
226 \\ syscall
227 \\ li $v0, 5058 # SYS_exit
228 \\ li $a0, 0
229 \\ syscall
230 ,
231 }
232 :
233 : [ptr] "r" (@intFromPtr(self.mapped.ptr)),
234 [len] "r" (self.mapped.len),
235 : .{ .memory = true }),
236 .or1k => asm volatile (
237 \\ l.ori r11, r0, 215 # SYS_munmap
238 \\ l.ori r3, %[ptr]
239 \\ l.ori r4, %[len]
240 \\ l.sys 1
241 \\ l.ori r11, r0, 93 # SYS_exit
242 \\ l.ori r3, r0, r0
243 \\ l.sys 1
244 :
245 : [ptr] "r" (@intFromPtr(self.mapped.ptr)),
246 [len] "r" (self.mapped.len),
247 : .{ .memory = true }),
248 .powerpc, .powerpcle, .powerpc64, .powerpc64le => asm volatile (
249 \\ li 0, 91 # SYS_munmap
250 \\ mr 3, %[ptr]
251 \\ mr 4, %[len]
252 \\ sc
253 \\ li 0, 1 # SYS_exit
254 \\ li 3, 0
255 \\ sc
256 \\ blr
257 :
258 : [ptr] "r" (@intFromPtr(self.mapped.ptr)),
259 [len] "r" (self.mapped.len),
260 : .{ .memory = true }),
261 .riscv32, .riscv64 => asm volatile (
262 \\ li a7, 215 # SYS_munmap
263 \\ mv a0, %[ptr]
264 \\ mv a1, %[len]
265 \\ ecall
266 \\ li a7, 93 # SYS_exit
267 \\ mv a0, zero
268 \\ ecall
269 :
270 : [ptr] "r" (@intFromPtr(self.mapped.ptr)),
271 [len] "r" (self.mapped.len),
272 : .{ .memory = true }),
273 .s390x => asm volatile (
274 \\ lgr %%r2, %[ptr]
275 \\ lgr %%r3, %[len]
276 \\ svc 91 # SYS_munmap
277 \\ lghi %%r2, 0
278 \\ svc 1 # SYS_exit
279 :
280 : [ptr] "r" (@intFromPtr(self.mapped.ptr)),
281 [len] "r" (self.mapped.len),
282 : .{ .memory = true }),
283 .sh, .sheb => asm volatile (
284 \\ mov #91, r3 ! SYS_munmap
285 \\ mov %[ptr], r4
286 \\ mov %[len], r5
287 \\ trapa #31
288 \\ or r0, r0
289 \\ or r0, r0
290 \\ or r0, r0
291 \\ or r0, r0
292 \\ or r0, r0
293 \\ mov #1, r3 ! SYS_exit
294 \\ mov #0, r4
295 \\ trapa #31
296 \\ or r0, r0
297 \\ or r0, r0
298 \\ or r0, r0
299 \\ or r0, r0
300 \\ or r0, r0
301 :
302 : [ptr] "r" (@intFromPtr(self.mapped.ptr)),
303 [len] "r" (self.mapped.len),
304 : .{ .memory = true }),
305 .sparc => asm volatile (
306 \\ # See sparc64 comments below.
307 \\ 1:
308 \\ cmp %%fp, 0
309 \\ beq 2f
310 \\ nop
311 \\ ba 1b
312 \\ restore
313 \\ 2:
314 \\ mov 73, %%g1 // SYS_munmap
315 \\ mov %[ptr], %%o0
316 \\ mov %[len], %%o1
317 \\ t 0x3 # ST_FLUSH_WINDOWS
318 \\ t 0x10
319 \\ mov 1, %%g1 // SYS_exit
320 \\ mov 0, %%o0
321 \\ t 0x10
322 :
323 : [ptr] "r" (@intFromPtr(self.mapped.ptr)),
324 [len] "r" (self.mapped.len),
325 : .{ .memory = true }),
326 .sparc64 => asm volatile (
327 \\ # SPARCs really don't like it when active stack frames
328 \\ # is unmapped (it will result in a segfault), so we
329 \\ # force-deactivate it by running `restore` until
330 \\ # all frames are cleared.
331 \\ 1:
332 \\ cmp %%fp, 0
333 \\ beq 2f
334 \\ nop
335 \\ ba 1b
336 \\ restore
337 \\ 2:
338 \\ mov 73, %%g1 // SYS_munmap
339 \\ mov %[ptr], %%o0
340 \\ mov %[len], %%o1
341 \\ # Flush register window contents to prevent background
342 \\ # memory access before unmapping the stack.
343 \\ flushw
344 \\ t 0x6d
345 \\ mov 1, %%g1 // SYS_exit
346 \\ mov 0, %%o0
347 \\ t 0x6d
348 :
349 : [ptr] "r" (@intFromPtr(self.mapped.ptr)),
350 [len] "r" (self.mapped.len),
351 : .{ .memory = true }),
352 .loongarch32, .loongarch64 => asm volatile (
353 \\ or $a0, $zero, %[ptr]
354 \\ or $a1, $zero, %[len]
355 \\ ori $a7, $zero, 215 # SYS_munmap
356 \\ syscall 0 # call munmap
357 \\ ori $a0, $zero, 0
358 \\ ori $a7, $zero, 93 # SYS_exit
359 \\ syscall 0 # call exit
360 :
361 : [ptr] "r" (@intFromPtr(self.mapped.ptr)),
362 [len] "r" (self.mapped.len),
363 : .{ .memory = true }),
364 else => |cpu_arch| @compileError("Unsupported linux arch: " ++ @tagName(cpu_arch)),
365 }
366 unreachable;
367 }
368 },
369 else => void,
370 };
371
372 const AllocateError = error{OutOfMemory};
373
374 fn allocate(stack_size: usize) AllocateError!*Thread {
375 if (use_pthreads) {
376 @compileError("TODO");
377 } else if (is_windows) {
378 @compileError("TODO");
379 } else if (native_os == .linux) {
380 const linux = std.os.linux;
381 const page_size = std.heap.pageSize();
382
383 var guard_offset: usize = undefined;
384 var stack_offset: usize = undefined;
385 var tls_offset: usize = undefined;
386 var instance_offset: usize = undefined;
387
388 const map_bytes = blk: {
389 var bytes: usize = page_size;
390 guard_offset = bytes;
391
392 bytes += @max(page_size, stack_size);
393 bytes = std.mem.alignForward(usize, bytes, page_size);
394 stack_offset = bytes;
395
396 bytes = std.mem.alignForward(usize, bytes, linux.tls.area_desc.alignment);
397 tls_offset = bytes;
398 bytes += linux.tls.area_desc.size;
399
400 bytes = std.mem.alignForward(usize, bytes, @alignOf(Thread));
401 instance_offset = bytes;
402 bytes += @sizeOf(Thread);
403
404 bytes = std.mem.alignForward(usize, bytes, page_size);
405 break :blk bytes;
406 };
407
408 // Map all memory needed without read/write permissions to avoid
409 // committing the whole region right away. Anonymous mapping ensures
410 // file descriptor limits are not exceeded.
411 const mapped = posix.mmap(
412 null,
413 map_bytes,
414 posix.PROT.NONE,
415 .{ .TYPE = .PRIVATE, .ANONYMOUS = true },
416 -1,
417 0,
418 ) catch |err| switch (err) {
419 error.MemoryMappingNotSupported => unreachable,
420 error.AccessDenied => unreachable,
421 error.PermissionDenied => unreachable,
422 error.ProcessFdQuotaExceeded => unreachable,
423 error.SystemFdQuotaExceeded => unreachable,
424 error.MappingAlreadyExists => unreachable,
425 else => |e| return e,
426 };
427 assert(mapped.len >= map_bytes);
428 errdefer posix.munmap(mapped);
429
430 // map everything but the guard page as read/write
431 posix.mprotect(
432 @alignCast(mapped[guard_offset..]),
433 posix.PROT.READ | posix.PROT.WRITE,
434 ) catch |err| switch (err) {
435 error.AccessDenied => unreachable,
436 else => |e| return e,
437 };
438
439 // Prepare the TLS segment and prepare a user_desc struct when needed on x86
440 var tls_ptr = linux.tls.prepareArea(mapped[tls_offset..]);
441 var user_desc: if (builtin.target.cpu.arch == .x86) linux.user_desc else void = undefined;
442 if (builtin.target.cpu.arch == .x86) {
443 defer tls_ptr = @intFromPtr(&user_desc);
444 user_desc = .{
445 .entry_number = linux.tls.area_desc.gdt_entry_number,
446 .base_addr = tls_ptr,
447 .limit = 0xfffff,
448 .flags = .{
449 .seg_32bit = 1,
450 .contents = 0, // Data
451 .read_exec_only = 0,
452 .limit_in_pages = 1,
453 .seg_not_present = 0,
454 .useable = 1,
455 },
456 };
457 }
458
459 const instance: *Thread = @ptrCast(@alignCast(&mapped[instance_offset]));
460 instance.* = .{
461 .signal_id = undefined, // Initialized on spawn.
462 .completion = .{
463 .mapped = mapped,
464 .stack_offset = stack_offset,
465 },
466 };
467 return instance;
468 } else {
469 @compileError("unimplemented");
470 }
471 }
472
473 const SpawnError = error{
474 ThreadQuotaExceeded,
475 SystemResources,
476 Unexpected,
477 };
478
479 fn spawn(thread: *Thread) SpawnError!void {
480 if (use_pthreads) {
481 const c = std.c;
482 const stack_size = {}; // TODO
483
484 var attr: c.pthread_attr_t = undefined;
485 if (c.pthread_attr_init(&attr) != .SUCCESS) return error.SystemResources;
486 defer assert(c.pthread_attr_destroy(&attr) == .SUCCESS);
487
488 assert(c.pthread_attr_setstacksize(&attr, stack_size) == .SUCCESS);
489 assert(c.pthread_attr_setguardsize(&attr, std.heap.pageSize()) == .SUCCESS);
490
491 var handle: c.pthread_t = undefined;
492 switch (c.pthread_create(
493 &handle,
494 &attr,
495 posixStart,
496 @ptrCast(thread),
497 )) {
498 .SUCCESS => {
499 thread.signal_id = handle;
500 return;
501 },
502 .AGAIN => return error.SystemResources,
503 .PERM => unreachable,
504 .INVAL => unreachable,
505 else => |err| return posix.unexpectedErrno(err),
506 }
507 @compileError("TODO");
508 } else if (is_windows) {
509 @compileError("TODO");
510 } else if (native_os == .linux) {
511 const linux = std.os.linux;
512
513 const flags: u32 = linux.CLONE.THREAD | linux.CLONE.DETACHED |
514 linux.CLONE.VM | linux.CLONE.FS | linux.CLONE.FILES |
515 linux.CLONE.PARENT_SETTID | linux.CLONE.CHILD_CLEARTID |
516 linux.CLONE.SIGHAND | linux.CLONE.SYSVSEM | linux.CLONE.SETTLS;
517
518 switch (linux.errno(linux.clone(
519 linuxStart,
520 @intFromPtr(&thread.completion.mapped[thread.completion.stack_offset]),
521 flags,
522 @intFromPtr(thread),
523 &thread.parent_tid,
524 thread.completion.tls_ptr,
525 &thread.child_tid.raw,
526 ))) {
527 .SUCCESS => return,
528 .AGAIN => return error.ThreadQuotaExceeded,
529 .INVAL => unreachable,
530 .NOMEM => return error.SystemResources,
531 .NOSPC => unreachable,
532 .PERM => unreachable,
533 .USERS => unreachable,
534 else => |err| return posix.unexpectedErrno(err),
535 }
536 } else {
537 @compileError("unimplemented");
538 }
539 }
540
541 fn linuxStart(raw_arg: usize) callconv(.c) u8 {
542 const t: *Thread = @ptrFromInt(raw_arg);
543 worker(t);
544 switch (t.completion.swap(.completed, .seq_cst)) {
545 .running => return 0,
546 .completed => unreachable,
547 .detached => t.completion.freeAndExit(),
548 }
549 unreachable;
550 }
551
552 fn posixStart(raw_arg: ?*anyopaque) callconv(.c) ?*anyopaque {
553 const t: *Thread = @ptrCast(@alignCast(raw_arg));
554 worker(t);
555 return null;
556 }
557
558 fn worker(t: *Thread) void {
559 current = t;
560
561 t.mutex.lock();
562
563 while (true) {
564 while (t.run_queue.popFirst()) |closure_node| {
565 t.mutex.unlock();
566 const closure: *Closure = @fieldParentPtr("node", closure_node);
567 closure.start(closure);
568 t.mutex.lock();
569 }
570 if (t.join_requested) break;
571 t.cond.wait(&t.mutex);
572 }
573 }
574
575 fn checkCancel(thread: *Thread) error{Canceled}!void {
576 const closure = thread.current_closure orelse return;
577 switch (@cmpxchgStrong(
578 CancelStatus,
579 &closure.cancel_status,
580 .requested,
581 .acknowledged,
582 .acq_rel,
583 .acquire,
584 ) orelse return error.Canceled) {
585 .none => return,
586 .requested => unreachable,
587 .acknowledged => unreachable,
588 _ => return,
589 }
590 }
591
592 fn beginSyscall(thread: *Thread) error{Canceled}!void {
593 const closure = thread.current_closure orelse return;
594
595 switch (@cmpxchgStrong(
596 CancelStatus,
597 &closure.cancel_status,
598 .none,
599 thread.signal_id,
600 .acq_rel,
601 .acquire,
602 ) orelse return) {
603 .none => unreachable,
604 .requested => {
605 @atomicStore(CancelStatus, &closure.cancel_status, .acknowledged, .acquire);
606 return error.Canceled;
607 },
608 .acknowledged => unreachable,
609 _ => unreachable,
610 }
611 }
612
613 fn endSyscall(thread: *Thread) error{Canceled}!void {
614 const closure = thread.current_closure orelse return;
41615
42 pub fn init(n: usize) ThreadCapacity {
43 assert(n != 0);
616 switch (@cmpxchgStrong(
617 CancelStatus,
618 &closure.cancel_status,
619 thread.signal_id,
620 .none,
621 .acq_rel,
622 .release,
623 ) orelse return) {
624 .none => unreachable,
625 .requested => {
626 @atomicStore(CancelStatus, &closure.cancel_status, .acknowledged, .release);
627 return error.Canceled;
628 },
629 .acknowledged => return,
630 _ => unreachable,
631 }
632 }
633};
634
635pub const Capacity = enum(isize) {
636 unknown = -30000,
637 _,
638
639 pub fn init(n: isize) Capacity {
640 assert(n > 0);
44641 return @enumFromInt(n);
45642 }
46643
47 pub fn get(tc: ThreadCapacity) ?usize {
644 pub fn get(tc: Capacity) ?usize {
48645 if (tc == .unknown) return null;
49646 return @intFromEnum(tc);
50647 }
51648};
52649
53threadlocal var current_closure: ?*Closure = null;
650pub const default_stack_size = 16 * 1024 * 1024;
651pub const use_pthreads = !is_windows and native_os != .wasi and builtin.link_libc;
54652
55653const max_iovecs_len = 8;
56654const splat_buffer_size = 64;
......@@ -59,85 +657,108 @@ comptime {
59657 if (@TypeOf(posix.IOV_MAX) != void) assert(max_iovecs_len <= posix.IOV_MAX);
60658}
61659
62const CancelId = enum(usize) {
660const CancelStatus = enum(usize) {
661 /// Cancellation has neither been requested, nor checked. The async
662 /// operation will check status before entering a blocking syscall.
663 /// This is also the status used for uninteruptible tasks.
63664 none = 0,
64 canceling = std.math.maxInt(usize),
665 /// Cancellation has been requested and the status will be checked before
666 /// entering a blocking syscall.
667 requested = std.math.maxInt(usize) - 1,
668 /// Cancellation has been acknowledged and is in progress. Signals should
669 /// not be sent.
670 acknowledged = std.math.maxInt(usize),
671 /// Stores a `Thread.SignalId` and indicates that sending a signal to this thread
672 /// is needed in order to cancel. This state is set before going into
673 /// a blocking operation that needs to get unblocked via signal.
65674 _,
66675
67 const ThreadId = if (std.Thread.use_pthreads) std.c.pthread_t else std.Thread.Id;
68
69 fn currentThread() CancelId {
70 if (std.Thread.use_pthreads) {
71 return @enumFromInt(@intFromPtr(std.c.pthread_self()));
72 } else {
73 return @enumFromInt(std.Thread.getCurrentId());
74 }
75 }
676 const Unpacked = union(enum) {
677 none,
678 requested,
679 acknowledeged,
680 signal_id: Thread.SignalId,
681 };
76682
77 fn toThreadId(cancel_id: CancelId) ThreadId {
78 if (std.Thread.use_pthreads) {
79 return @ptrFromInt(@intFromEnum(cancel_id));
80 } else {
81 return @intCast(@intFromEnum(cancel_id));
82 }
683 fn unpack(cs: CancelStatus) Unpacked {
684 return switch (cs) {
685 .none => .none,
686 .requested => .requested,
687 .acknowledged => .acknowledged,
688 _ => |signal_id| .{ .signal_id = signal_id },
689 };
83690 }
84691};
85692
86693const Closure = struct {
87694 start: Start,
88695 node: std.SinglyLinkedList.Node = .{},
89 cancel_tid: CancelId,
696 cancel_status: CancelStatus,
90697 /// Whether this task bumps minimum number of threads in the pool.
91698 is_concurrent: bool,
92699
93700 const Start = *const fn (*Closure) void;
94701
95 fn requestCancel(closure: *Closure) void {
96 switch (@atomicRmw(CancelId, &closure.cancel_tid, .Xchg, .canceling, .acq_rel)) {
97 .none, .canceling => {},
98 else => |tid| {
99 if (std.Thread.use_pthreads) {
100 const rc = std.c.pthread_kill(tid.toThreadId(), .IO);
101 if (is_debug) assert(rc == 0);
102 } else if (native_os == .linux) {
103 _ = std.os.linux.tgkill(std.os.linux.getpid(), @bitCast(tid.toThreadId()), .IO);
104 }
105 },
702 fn requestCancel(closure: *Closure, t: *Threaded) void {
703 var signal_id = switch (@atomicRmw(CancelStatus, &closure.cancel_status, .Xchg, .requested, .monotonic).unpack()) {
704 .none, .acknowledged, .requested => return,
705 else => |signal_id| signal_id,
706 };
707 // The task will enter a blocking syscall before checking for cancellation again.
708 // We can send a signal to interrupt the syscall, but if it arrives before
709 // the syscall instruction, it will be missed. Therefore, this code tries
710 // again until the cancellation request is acknowledged.
711 const max_attempts = 3;
712 for (0..max_attempts) |_| {
713 if (use_pthreads) {
714 const rc = std.c.pthread_kill(signal_id.toThreadId(), .IO);
715 if (is_debug) assert(rc == 0);
716 } else if (native_os == .linux) {
717 const pid: posix.pid_t = p: {
718 const cached_pid = @atomicLoad(Pid, &t.pid, .monotonic);
719 if (cached_pid != .unknown) break :p @intFromEnum(cached_pid);
720 const pid = std.os.linux.getpid();
721 @atomicStore(Pid, &t.pid, @enumFromInt(pid), .monotonic);
722 break :p pid;
723 };
724 _ = std.os.linux.tgkill(pid, @bitCast(signal_id.toThreadId()), .IO);
725 } else {
726 return;
727 }
728
729 // TODO make this a nanosleep with 1 << attempt duration
730 std.Thread.yield() catch {};
731
732 switch (@atomicRmw(CancelStatus, &closure.cancel_status, .Xchg, .requested, .monotonic).unpack()) {
733 .requested => continue,
734 .none, .acknowledged => return,
735 else => |new_signal_id| signal_id = new_signal_id,
736 }
106737 }
107738 }
108739};
109740
110pub const InitError = std.Thread.CpuCountError || Allocator.Error;
741pub const CpuCountError = error{
742 PermissionDenied,
743 SystemResources,
744 Unsupported,
745} || Io.UnexpectedError;
111746
112747/// Related:
113748/// * `init_single_threaded`
114pub fn init(
115 /// Must be threadsafe. Only used for the following functions:
116 /// * `Io.VTable.async`
117 /// * `Io.VTable.concurrent`
118 /// * `Io.VTable.groupAsync`
119 /// If these functions are avoided, then `Allocator.failing` may be passed
120 /// here.
121 gpa: Allocator,
122) Threaded {
749pub fn init() Threaded {
123750 const cpu_count = std.Thread.getCpuCount();
124751
125752 var t: Threaded = .{
126 .allocator = gpa,
127753 .threads = .empty,
128 .stack_size = std.Thread.SpawnConfig.default_stack_size,
129 .thread_capacity = .init(if (cpu_count) |n| .init(n) else |_| .unknown),
130 .thread_capacity_error = if (cpu_count) |_| null else |e| e,
754 .capacity = .init(if (cpu_count) |n| .init(n) else |_| .unknown),
755 .capacity_error = if (cpu_count) |_| null else |e| e,
131756 .concurrent_count = 0,
132757 .old_sig_io = undefined,
133758 .old_sig_pipe = undefined,
134759 .have_signal_handler = false,
135760 };
136761
137 if (cpu_count) |n| {
138 t.threads.ensureTotalCapacityPrecise(gpa, n - 1) catch {};
139 } else |_| {}
140
141762 if (posix.Sigaction != void) {
142763 // This causes sending `posix.SIG.IO` to thread to interrupt blocking
143764 // syscalls, returning `posix.E.INTR`.
......@@ -161,11 +782,9 @@ pub fn init(
161782/// * cancel requests have no effect.
162783/// * `deinit` is safe, but unnecessary to call.
163784pub const init_single_threaded: Threaded = .{
164 .allocator = .failing,
165785 .threads = .empty,
166 .stack_size = std.Thread.SpawnConfig.default_stack_size,
167 .thread_capacity = .init(.init(1)),
168 .thread_capacity_error = null,
786 .capacity = .init(.init(1)),
787 .capacity_error = null,
169788 .concurrent_count = 0,
170789 .old_sig_io = undefined,
171790 .old_sig_pipe = undefined,
......@@ -173,9 +792,7 @@ pub const init_single_threaded: Threaded = .{
173792};
174793
175794pub fn deinit(t: *Threaded) void {
176 const gpa = t.allocator;
177 t.join();
178 t.threads.deinit(gpa);
795 join(t);
179796 if (is_windows and t.wsa.status == .initialized) {
180797 if (ws2_32.WSACleanup() != 0) recoverableOsBugDetected();
181798 }
......@@ -186,46 +803,29 @@ pub fn deinit(t: *Threaded) void {
186803 t.* = undefined;
187804}
188805
189pub fn setThreadCapacity(t: *Threaded, n: usize) void {
190 t.thread_capacity.store(.init(n), .monotonic);
806pub fn setCapacity(t: *Threaded, n: usize) void {
807 t.capacity.store(.init(n), .monotonic);
191808}
192809
193pub fn getThreadCapacity(t: *Threaded) ?usize {
194 return t.thread_capacity.load(.monotonic).get();
195}
196
197pub fn getCurrentThreadId() usize {
198 @panic("TODO");
810pub fn getCapacity(t: *Threaded) ?usize {
811 return t.capacity.load(.monotonic).get();
199812}
200813
201814fn join(t: *Threaded) void {
202815 if (builtin.single_threaded) return;
203 {
204 t.mutex.lock();
205 defer t.mutex.unlock();
206 t.join_requested = true;
207 }
208 t.cond.broadcast();
209 for (t.threads.items) |thread| thread.join();
210}
211816
212fn worker(t: *Threaded) void {
213 t.mutex.lock();
214 defer t.mutex.unlock();
215
216 while (true) {
217 while (t.run_queue.popFirst()) |closure_node| {
218 t.mutex.unlock();
219 const closure: *Closure = @fieldParentPtr("node", closure_node);
220 const is_concurrent = closure.is_concurrent;
221 closure.start(closure);
222 t.mutex.lock();
223 if (is_concurrent) {
224 t.concurrent_count -= 1;
817 {
818 var it: ?*const std.SinglyLinkedList.Node = &t.main_thread.list_node;
819 while (it) |n| : (it = n.next) {
820 const thread: *Thread = @fieldParentPtr("list_node", n);
821 {
822 thread.mutex.lock();
823 defer thread.mutex.unlock();
824 thread.join_requested = true;
825 thread.cond.signal();
225826 }
827 thread.join();
226828 }
227 if (t.join_requested) break;
228 t.cond.wait(&t.mutex);
229829 }
230830}
231831
......@@ -237,7 +837,6 @@ pub fn io(t: *Threaded) Io {
237837 .concurrent = concurrent,
238838 .await = await,
239839 .cancel = cancel,
240 .cancelRequested = cancelRequested,
241840 .select = select,
242841
243842 .groupAsync = groupAsync,
......@@ -333,7 +932,6 @@ pub fn ioBasic(t: *Threaded) Io {
333932 .concurrent = concurrent,
334933 .await = await,
335934 .cancel = cancel,
336 .cancelRequested = cancelRequested,
337935 .select = select,
338936
339937 .groupAsync = groupAsync,
......@@ -428,22 +1026,10 @@ const AsyncClosure = struct {
4281026
4291027 fn start(closure: *Closure) void {
4301028 const ac: *AsyncClosure = @alignCast(@fieldParentPtr("closure", closure));
431 const tid: CancelId = .currentThread();
432 if (@cmpxchgStrong(CancelId, &closure.cancel_tid, .none, tid, .acq_rel, .acquire)) |cancel_tid| {
433 assert(cancel_tid == .canceling);
434 // Even though we already know the task is canceled, we must still
435 // run the closure in order to make the return value valid and in
436 // case there are side effects.
437 }
438 current_closure = closure;
1029 const current_thread = Thread.current;
1030 current_thread.current_closure = closure;
4391031 ac.func(ac.contextPointer(), ac.resultPointer());
440 current_closure = null;
441
442 // In case a cancel happens after successful task completion, prevents
443 // signal from being delivered to the thread in `requestCancel`.
444 if (@cmpxchgStrong(CancelId, &closure.cancel_tid, tid, .none, .acq_rel, .acquire)) |cancel_tid| {
445 assert(cancel_tid == .canceling);
446 }
1032 current_thread.current_closure = null;
4471033
4481034 if (@atomicRmw(?*ResetEvent, &ac.select_condition, .Xchg, done_reset_event, .release)) |select_reset| {
4491035 assert(select_reset != done_reset_event);
......@@ -464,14 +1050,14 @@ const AsyncClosure = struct {
4641050 }
4651051
4661052 fn init(
467 gpa: Allocator,
1053 ac: *AsyncClosure,
4681054 mode: enum { async, concurrent },
4691055 result_len: usize,
4701056 result_alignment: Alignment,
4711057 context: []const u8,
4721058 context_alignment: Alignment,
4731059 func: *const fn (context: *const anyopaque, result: *anyopaque) void,
474 ) Allocator.Error!*AsyncClosure {
1060 ) void {
4751061 const max_context_misalignment = context_alignment.toByteUnits() -| @alignOf(AsyncClosure);
4761062 const worst_case_context_offset = context_alignment.forward(@sizeOf(AsyncClosure) + max_context_misalignment);
4771063 const worst_case_result_offset = result_alignment.forward(worst_case_context_offset + context.len);
......@@ -529,7 +1115,7 @@ fn async(
5291115 }
5301116
5311117 const t: *Threaded = @ptrCast(@alignCast(userdata));
532 const cpu_count = t.getThreadCapacity() orelse {
1118 const may_spawn = takeCapacity(t) catch {
5331119 return concurrent(userdata, result.len, result_alignment, context, context_alignment, start) catch {
5341120 start(context.ptr, result.ptr);
5351121 return null;
......@@ -538,42 +1124,25 @@ fn async(
5381124
5391125 const gpa = t.allocator;
5401126 const ac = AsyncClosure.init(gpa, .async, result.len, result_alignment, context, context_alignment, start) catch {
1127 returnCapacity(t);
5411128 start(context.ptr, result.ptr);
5421129 return null;
5431130 };
5441131
545 t.mutex.lock();
1132 @memcpy(ac.contextPointer()[0..context.len], context);
5461133
547 const thread_capacity = cpu_count - 1 + t.concurrent_count;
1134 if (may_spawn) {
1135 // TODO Allocate Thread
5481136
549 t.threads.ensureTotalCapacityPrecise(gpa, thread_capacity) catch {
550 t.mutex.unlock();
551 ac.deinit(gpa);
552 start(context.ptr, result.ptr);
553 return null;
554 };
1137 thread.run_queue.prepend(&ac.closure.node);
5551138
556 t.run_queue.prepend(&ac.closure.node);
1139 // TODO start thread
5571140
558 if (t.threads.items.len < thread_capacity) {
559 const thread = std.Thread.spawn(.{ .stack_size = t.stack_size }, worker, .{t}) catch {
560 if (t.threads.items.len == 0) {
561 assert(t.run_queue.popFirst() == &ac.closure.node);
562 t.mutex.unlock();
563 ac.deinit(gpa);
564 start(context.ptr, result.ptr);
565 return null;
566 }
567 // Rely on other workers to do it.
568 t.mutex.unlock();
569 t.cond.signal();
570 return @ptrCast(ac);
571 };
572 t.threads.appendAssumeCapacity(thread);
1141 return @ptrCast(ac);
5731142 }
5741143
575 t.mutex.unlock();
576 t.cond.signal();
1144 const thread = Thread.current;
1145 thread.run_queue.prepend(&ac.closure.node);
5771146 return @ptrCast(ac);
5781147}
5791148
......@@ -588,7 +1157,7 @@ fn concurrent(
5881157 if (builtin.single_threaded) return error.ConcurrencyUnavailable;
5891158
5901159 const t: *Threaded = @ptrCast(@alignCast(userdata));
591 const cpu_count = t.getThreadCapacity() orelse 1;
1160 const cpu_count = t.getCapacity() orelse 1;
5921161
5931162 const gpa = t.allocator;
5941163 const ac = AsyncClosure.init(gpa, .concurrent, result_len, result_alignment, context, context_alignment, start) catch {
......@@ -598,9 +1167,9 @@ fn concurrent(
5981167 t.mutex.lock();
5991168
6001169 t.concurrent_count += 1;
601 const thread_capacity = cpu_count - 1 + t.concurrent_count;
1170 const capacity = cpu_count - 1 + t.concurrent_count;
6021171
603 t.threads.ensureTotalCapacity(gpa, thread_capacity) catch {
1172 t.threads.ensureTotalCapacity(gpa, capacity) catch {
6041173 t.mutex.unlock();
6051174 ac.deinit(gpa);
6061175 return error.ConcurrencyUnavailable;
......@@ -608,7 +1177,7 @@ fn concurrent(
6081177
6091178 t.run_queue.prepend(&ac.closure.node);
6101179
611 if (t.threads.items.len < thread_capacity) {
1180 if (t.threads.items.len < capacity) {
6121181 const thread = std.Thread.spawn(.{ .stack_size = t.stack_size }, worker, .{t}) catch {
6131182 assert(t.run_queue.popFirst() == &ac.closure.node);
6141183 t.mutex.unlock();
......@@ -635,24 +1204,13 @@ const GroupClosure = struct {
6351204
6361205 fn start(closure: *Closure) void {
6371206 const gc: *GroupClosure = @alignCast(@fieldParentPtr("closure", closure));
638 const tid: CancelId = .currentThread();
1207 const current_thread = Thread.current;
6391208 const group = gc.group;
6401209 const group_state: *std.atomic.Value(usize) = @ptrCast(&group.state);
6411210 const reset_event: *ResetEvent = @ptrCast(&group.context);
642 if (@cmpxchgStrong(CancelId, &closure.cancel_tid, .none, tid, .acq_rel, .acquire)) |cancel_tid| {
643 assert(cancel_tid == .canceling);
644 // Even though we already know the task is canceled, we must still
645 // run the closure in case there are side effects.
646 }
647 current_closure = closure;
1211 current_thread.current_closure = closure;
6481212 gc.func(group, gc.contextPointer());
649 current_closure = null;
650
651 // In case a cancel happens after successful task completion, prevents
652 // signal from being delivered to the thread in `requestCancel`.
653 if (@cmpxchgStrong(CancelId, &closure.cancel_tid, tid, .none, .acq_rel, .acquire)) |cancel_tid| {
654 assert(cancel_tid == .canceling);
655 }
1213 current_thread.current_closure = null;
6561214
6571215 const prev_state = group_state.fetchSub(sync_one_pending, .acq_rel);
6581216 assert((prev_state / sync_one_pending) > 0);
......@@ -717,7 +1275,7 @@ fn groupAsync(
7171275 if (builtin.single_threaded) return start(group, context.ptr);
7181276
7191277 const t: *Threaded = @ptrCast(@alignCast(userdata));
720 const cpu_count = t.getThreadCapacity() orelse 1;
1278 const cpu_count = t.getCapacity() orelse 1;
7211279
7221280 const gpa = t.allocator;
7231281 const gc = GroupClosure.init(gpa, t, group, context, context_alignment, start) catch {
......@@ -730,9 +1288,9 @@ fn groupAsync(
7301288 gc.node = .{ .next = @ptrCast(@alignCast(group.token)) };
7311289 group.token = &gc.node;
7321290
733 const thread_capacity = cpu_count - 1 + t.concurrent_count;
1291 const capacity = cpu_count - 1 + t.concurrent_count;
7341292
735 t.threads.ensureTotalCapacityPrecise(gpa, thread_capacity) catch {
1293 t.threads.ensureTotalCapacityPrecise(gpa, capacity) catch {
7361294 t.mutex.unlock();
7371295 gc.deinit(gpa);
7381296 return start(group, context.ptr);
......@@ -740,7 +1298,7 @@ fn groupAsync(
7401298
7411299 t.run_queue.prepend(&gc.closure.node);
7421300
743 if (t.threads.items.len < thread_capacity) {
1301 if (t.threads.items.len < capacity) {
7441302 const thread = std.Thread.spawn(.{ .stack_size = t.stack_size }, worker, .{t}) catch {
7451303 assert(t.run_queue.popFirst() == &gc.closure.node);
7461304 t.mutex.unlock();
......@@ -775,7 +1333,7 @@ fn groupWait(userdata: ?*anyopaque, group: *Io.Group, token: *anyopaque) void {
7751333 var node: *std.SinglyLinkedList.Node = @ptrCast(@alignCast(token));
7761334 while (true) {
7771335 const gc: *GroupClosure = @fieldParentPtr("node", node);
778 gc.closure.requestCancel();
1336 gc.closure.requestCancel(t);
7791337 node = node.next orelse break;
7801338 }
7811339 reset_event.waitUncancelable();
......@@ -801,7 +1359,7 @@ fn groupCancel(userdata: ?*anyopaque, group: *Io.Group, token: *anyopaque) void
8011359 var node: *std.SinglyLinkedList.Node = @ptrCast(@alignCast(token));
8021360 while (true) {
8031361 const gc: *GroupClosure = @fieldParentPtr("node", node);
804 gc.closure.requestCancel();
1362 gc.closure.requestCancel(t);
8051363 node = node.next orelse break;
8061364 }
8071365 }
......@@ -844,21 +1402,10 @@ fn cancel(
8441402 _ = result_alignment;
8451403 const t: *Threaded = @ptrCast(@alignCast(userdata));
8461404 const ac: *AsyncClosure = @ptrCast(@alignCast(any_future));
847 ac.closure.requestCancel();
1405 ac.closure.requestCancel(t);
8481406 ac.waitAndDeinit(t.allocator, result);
8491407}
8501408
851fn cancelRequested(userdata: ?*anyopaque) bool {
852 const t: *Threaded = @ptrCast(@alignCast(userdata));
853 _ = t;
854 const closure = current_closure orelse return false;
855 return @atomicLoad(CancelId, &closure.cancel_tid, .acquire) == .canceling;
856}
857
858fn checkCancel(t: *Threaded) error{Canceled}!void {
859 if (cancelRequested(t)) return error.Canceled;
860}
861
8621409fn mutexLock(userdata: ?*anyopaque, prev_state: Io.Mutex.State, mutex: *Io.Mutex) Io.Cancelable!void {
8631410 if (builtin.single_threaded) unreachable; // Interface should have prevented this.
8641411 if (native_os == .netbsd) @panic("TODO");
......@@ -1043,35 +1590,47 @@ const dirMake = switch (native_os) {
10431590
10441591fn dirMakePosix(userdata: ?*anyopaque, dir: Io.Dir, sub_path: []const u8, mode: Io.Dir.Mode) Io.Dir.MakeError!void {
10451592 const t: *Threaded = @ptrCast(@alignCast(userdata));
1593 _ = t;
1594 const current_thread = Thread.current;
10461595
10471596 var path_buffer: [posix.PATH_MAX]u8 = undefined;
10481597 const sub_path_posix = try pathToPosix(sub_path, &path_buffer);
10491598
1599 try current_thread.beginSyscall();
10501600 while (true) {
1051 try t.checkCancel();
10521601 switch (posix.errno(posix.system.mkdirat(dir.handle, sub_path_posix, mode))) {
1053 .SUCCESS => return,
1054 .INTR => continue,
1055 .CANCELED => return error.Canceled,
1056
1057 .ACCES => return error.AccessDenied,
1058 .BADF => |err| return errnoBug(err), // File descriptor used after closed.
1059 .PERM => return error.PermissionDenied,
1060 .DQUOT => return error.DiskQuota,
1061 .EXIST => return error.PathAlreadyExists,
1062 .FAULT => |err| return errnoBug(err),
1063 .LOOP => return error.SymLinkLoop,
1064 .MLINK => return error.LinkQuotaExceeded,
1065 .NAMETOOLONG => return error.NameTooLong,
1066 .NOENT => return error.FileNotFound,
1067 .NOMEM => return error.SystemResources,
1068 .NOSPC => return error.NoSpaceLeft,
1069 .NOTDIR => return error.NotDir,
1070 .ROFS => return error.ReadOnlyFileSystem,
1071 // dragonfly: when dir_fd is unlinked from filesystem
1072 .NOTCONN => return error.FileNotFound,
1073 .ILSEQ => return error.BadPathName,
1074 else => |err| return posix.unexpectedErrno(err),
1602 .SUCCESS => {
1603 try current_thread.endSyscall();
1604 break;
1605 },
1606 .INTR => {
1607 try current_thread.checkCancel();
1608 continue;
1609 },
1610 else => |e| {
1611 try current_thread.endSyscall();
1612 switch (e) {
1613 .CANCELED => return error.Canceled,
1614 .ACCES => return error.AccessDenied,
1615 .BADF => |err| return errnoBug(err), // File descriptor used after closed.
1616 .PERM => return error.PermissionDenied,
1617 .DQUOT => return error.DiskQuota,
1618 .EXIST => return error.PathAlreadyExists,
1619 .FAULT => |err| return errnoBug(err),
1620 .LOOP => return error.SymLinkLoop,
1621 .MLINK => return error.LinkQuotaExceeded,
1622 .NAMETOOLONG => return error.NameTooLong,
1623 .NOENT => return error.FileNotFound,
1624 .NOMEM => return error.SystemResources,
1625 .NOSPC => return error.NoSpaceLeft,
1626 .NOTDIR => return error.NotDir,
1627 .ROFS => return error.ReadOnlyFileSystem,
1628 // dragonfly: when dir_fd is unlinked from filesystem
1629 .NOTCONN => return error.FileNotFound,
1630 .ILSEQ => return error.BadPathName,
1631 else => |err| return posix.unexpectedErrno(err),
1632 }
1633 },
10751634 }
10761635 }
10771636}
......@@ -1981,6 +2540,7 @@ fn dirOpenFilePosix(
19812540 flags: Io.File.OpenFlags,
19822541) Io.File.OpenError!Io.File {
19832542 const t: *Threaded = @ptrCast(@alignCast(userdata));
2543 const current_thread = Thread.current;
19842544
19852545 var path_buffer: [posix.PATH_MAX]u8 = undefined;
19862546 const sub_path_posix = try pathToPosix(sub_path, &path_buffer);
......@@ -2017,40 +2577,52 @@ fn dirOpenFilePosix(
20172577 },
20182578 };
20192579
2020 const fd: posix.fd_t = while (true) {
2021 try t.checkCancel();
2580 try current_thread.beginSyscall();
2581 const fd = while (true) {
20222582 const rc = openat_sym(dir.handle, sub_path_posix, os_flags, @as(posix.mode_t, 0));
20232583 switch (posix.errno(rc)) {
2024 .SUCCESS => break @intCast(rc),
2025 .INTR => continue,
2026 .CANCELED => return error.Canceled,
2027
2028 .FAULT => |err| return errnoBug(err),
2029 .INVAL => return error.BadPathName,
2030 .BADF => |err| return errnoBug(err), // File descriptor used after closed.
2031 .ACCES => return error.AccessDenied,
2032 .FBIG => return error.FileTooBig,
2033 .OVERFLOW => return error.FileTooBig,
2034 .ISDIR => return error.IsDir,
2035 .LOOP => return error.SymLinkLoop,
2036 .MFILE => return error.ProcessFdQuotaExceeded,
2037 .NAMETOOLONG => return error.NameTooLong,
2038 .NFILE => return error.SystemFdQuotaExceeded,
2039 .NODEV => return error.NoDevice,
2040 .NOENT => return error.FileNotFound,
2041 .SRCH => return error.ProcessNotFound,
2042 .NOMEM => return error.SystemResources,
2043 .NOSPC => return error.NoSpaceLeft,
2044 .NOTDIR => return error.NotDir,
2045 .PERM => return error.PermissionDenied,
2046 .EXIST => return error.PathAlreadyExists,
2047 .BUSY => return error.DeviceBusy,
2048 .OPNOTSUPP => return error.FileLocksNotSupported,
2049 .AGAIN => return error.WouldBlock,
2050 .TXTBSY => return error.FileBusy,
2051 .NXIO => return error.NoDevice,
2052 .ILSEQ => return error.BadPathName,
2053 else => |err| return posix.unexpectedErrno(err),
2584 .SUCCESS => {
2585 const fd: posix.fd_t = @intCast(rc);
2586 errdefer posix.close(fd);
2587 try current_thread.endSyscall();
2588 break fd;
2589 },
2590 .INTR => {
2591 try current_thread.checkCancel();
2592 continue;
2593 },
2594 else => |e| {
2595 try current_thread.endSyscall();
2596 switch (e) {
2597 .CANCELED => return error.Canceled,
2598 .FAULT => |err| return errnoBug(err),
2599 .INVAL => return error.BadPathName,
2600 .BADF => |err| return errnoBug(err), // File descriptor used after closed.
2601 .ACCES => return error.AccessDenied,
2602 .FBIG => return error.FileTooBig,
2603 .OVERFLOW => return error.FileTooBig,
2604 .ISDIR => return error.IsDir,
2605 .LOOP => return error.SymLinkLoop,
2606 .MFILE => return error.ProcessFdQuotaExceeded,
2607 .NAMETOOLONG => return error.NameTooLong,
2608 .NFILE => return error.SystemFdQuotaExceeded,
2609 .NODEV => return error.NoDevice,
2610 .NOENT => return error.FileNotFound,
2611 .SRCH => return error.ProcessNotFound,
2612 .NOMEM => return error.SystemResources,
2613 .NOSPC => return error.NoSpaceLeft,
2614 .NOTDIR => return error.NotDir,
2615 .PERM => return error.PermissionDenied,
2616 .EXIST => return error.PathAlreadyExists,
2617 .BUSY => return error.DeviceBusy,
2618 .OPNOTSUPP => return error.FileLocksNotSupported,
2619 .AGAIN => return error.WouldBlock,
2620 .TXTBSY => return error.FileBusy,
2621 .NXIO => return error.NoDevice,
2622 .ILSEQ => return error.BadPathName,
2623 else => |err| return posix.unexpectedErrno(err),
2624 }
2625 },
20542626 }
20552627 };
20562628 errdefer posix.close(fd);
......@@ -6208,6 +6780,7 @@ fn initializeWsa(t: *Threaded) error{NetworkDown}!void {
62086780
62096781fn doNothingSignalHandler(_: posix.SIG) callconv(.c) void {}
62106782
6783
62116784test {
62126785 _ = @import("Threaded/test.zig");
62136786}
lib/std/Thread.zig+6-313
......@@ -1,6 +1,6 @@
1//! This struct represents a kernel thread, and acts as a namespace for concurrency
2//! primitives that operate on kernel threads. For concurrency primitives that support
3//! both evented I/O and async I/O, see the respective names in the top level std namespace.
1//! This struct represents a kernel thread, and acts as a namespace for
2//! concurrency primitives that operate on kernel threads. For concurrency
3//! primitives that interact with the I/O interface, see `std.Io`.
44
55const std = @import("std.zig");
66const builtin = @import("builtin");
......@@ -20,7 +20,7 @@ pub const RwLock = @import("Thread/RwLock.zig");
2020pub const Pool = @import("Thread/Pool.zig");
2121pub const WaitGroup = @import("Thread/WaitGroup.zig");
2222
23pub const use_pthreads = native_os != .windows and native_os != .wasi and builtin.link_libc;
23pub const use_pthreads = std.Io.Threaded.use_pthreads;
2424
2525/// A thread-safe logical boolean value which can be `set` and `unset`.
2626///
......@@ -422,12 +422,7 @@ pub fn getCurrentId() Id {
422422 return Impl.getCurrentId();
423423}
424424
425pub const CpuCountError = error{
426 PermissionDenied,
427 SystemResources,
428 Unsupported,
429 Unexpected,
430};
425pub const CpuCountError = std.Io.Threaded.CpuCountError;
431426
432427/// Returns the platforms view on the number of logical CPU cores available.
433428///
......@@ -446,7 +441,7 @@ pub const SpawnConfig = struct {
446441 /// The allocator to be used to allocate memory for the to-be-spawned thread
447442 allocator: ?std.mem.Allocator = null,
448443
449 pub const default_stack_size = 16 * 1024 * 1024;
444 pub const default_stack_size = std.Io.Threaded.default_stack_size;
450445};
451446
452447pub const SpawnError = error{
......@@ -1215,308 +1210,6 @@ const LinuxThreadImpl = struct {
12151210
12161211 thread: *ThreadCompletion,
12171212
1218 const ThreadCompletion = struct {
1219 completion: Completion = Completion.init(.running),
1220 child_tid: std.atomic.Value(i32) = std.atomic.Value(i32).init(1),
1221 parent_tid: i32 = undefined,
1222 mapped: []align(std.heap.page_size_min) u8,
1223
1224 /// Calls `munmap(mapped.ptr, mapped.len)` then `exit(1)` without touching the stack (which lives in `mapped.ptr`).
1225 /// Ported over from musl libc's pthread detached implementation:
1226 /// https://github.com/ifduyue/musl/search?q=__unmapself
1227 fn freeAndExit(self: *ThreadCompletion) noreturn {
1228 switch (target.cpu.arch) {
1229 .x86 => asm volatile (
1230 \\ movl $91, %%eax # SYS_munmap
1231 \\ movl %[ptr], %%ebx
1232 \\ movl %[len], %%ecx
1233 \\ int $128
1234 \\ movl $1, %%eax # SYS_exit
1235 \\ movl $0, %%ebx
1236 \\ int $128
1237 :
1238 : [ptr] "r" (@intFromPtr(self.mapped.ptr)),
1239 [len] "r" (self.mapped.len),
1240 : .{ .memory = true }),
1241 .x86_64 => asm volatile (switch (target.abi) {
1242 .gnux32, .muslx32 =>
1243 \\ movl $0x4000000b, %%eax # SYS_munmap
1244 \\ syscall
1245 \\ movl $0x4000003c, %%eax # SYS_exit
1246 \\ xor %%rdi, %%rdi
1247 \\ syscall
1248 ,
1249 else =>
1250 \\ movl $11, %%eax # SYS_munmap
1251 \\ syscall
1252 \\ movl $60, %%eax # SYS_exit
1253 \\ xor %%rdi, %%rdi
1254 \\ syscall
1255 ,
1256 }
1257 :
1258 : [ptr] "{rdi}" (@intFromPtr(self.mapped.ptr)),
1259 [len] "{rsi}" (self.mapped.len),
1260 ),
1261 .arm, .armeb, .thumb, .thumbeb => asm volatile (
1262 \\ mov r7, #91 // SYS_munmap
1263 \\ mov r0, %[ptr]
1264 \\ mov r1, %[len]
1265 \\ svc 0
1266 \\ mov r7, #1 // SYS_exit
1267 \\ mov r0, #0
1268 \\ svc 0
1269 :
1270 : [ptr] "r" (@intFromPtr(self.mapped.ptr)),
1271 [len] "r" (self.mapped.len),
1272 : .{ .memory = true }),
1273 .aarch64, .aarch64_be => asm volatile (
1274 \\ mov x8, #215 // SYS_munmap
1275 \\ mov x0, %[ptr]
1276 \\ mov x1, %[len]
1277 \\ svc 0
1278 \\ mov x8, #93 // SYS_exit
1279 \\ mov x0, #0
1280 \\ svc 0
1281 :
1282 : [ptr] "r" (@intFromPtr(self.mapped.ptr)),
1283 [len] "r" (self.mapped.len),
1284 : .{ .memory = true }),
1285 .alpha => asm volatile (
1286 \\ ldi $0, 73 # SYS_munmap
1287 \\ mov %[ptr], $16
1288 \\ mov %[len], $17
1289 \\ callsys
1290 \\ ldi $0, 1 # SYS_exit
1291 \\ ldi $16, 0
1292 \\ callsys
1293 :
1294 : [ptr] "r" (@intFromPtr(self.mapped.ptr)),
1295 [len] "r" (self.mapped.len),
1296 : .{ .memory = true }),
1297 .hexagon => asm volatile (
1298 \\ r6 = #215 // SYS_munmap
1299 \\ r0 = %[ptr]
1300 \\ r1 = %[len]
1301 \\ trap0(#1)
1302 \\ r6 = #93 // SYS_exit
1303 \\ r0 = #0
1304 \\ trap0(#1)
1305 :
1306 : [ptr] "r" (@intFromPtr(self.mapped.ptr)),
1307 [len] "r" (self.mapped.len),
1308 : .{ .memory = true }),
1309 .hppa => asm volatile (
1310 \\ ldi 91, %%r20 /* SYS_munmap */
1311 \\ copy %[ptr], %%r26
1312 \\ copy %[len], %%r25
1313 \\ ble 0x100(%%sr2, %%r0)
1314 \\ ldi 1, %%r20 /* SYS_exit */
1315 \\ ldi 0, %%r26
1316 \\ ble 0x100(%%sr2, %%r0)
1317 :
1318 : [ptr] "r" (@intFromPtr(self.mapped.ptr)),
1319 [len] "r" (self.mapped.len),
1320 : .{ .memory = true }),
1321 .m68k => asm volatile (
1322 \\ move.l #91, %%d0 // SYS_munmap
1323 \\ move.l %[ptr], %%d1
1324 \\ move.l %[len], %%d2
1325 \\ trap #0
1326 \\ move.l #1, %%d0 // SYS_exit
1327 \\ move.l #0, %%d1
1328 \\ trap #0
1329 :
1330 : [ptr] "r" (@intFromPtr(self.mapped.ptr)),
1331 [len] "r" (self.mapped.len),
1332 : .{ .memory = true }),
1333 .microblaze, .microblazeel => asm volatile (
1334 \\ ori r12, r0, 91 # SYS_munmap
1335 \\ ori r5, %[ptr], 0
1336 \\ ori r6, %[len], 0
1337 \\ brki r14, 0x8
1338 \\ ori r12, r0, 1 # SYS_exit
1339 \\ or r5, r0, r0
1340 \\ brki r14, 0x8
1341 :
1342 : [ptr] "r" (@intFromPtr(self.mapped.ptr)),
1343 [len] "r" (self.mapped.len),
1344 : .{ .memory = true }),
1345 // We set `sp` to the address of the current function as a workaround for a Linux
1346 // kernel bug that caused syscalls to return EFAULT if the stack pointer is invalid.
1347 // The bug was introduced in 46e12c07b3b9603c60fc1d421ff18618241cb081 and fixed in
1348 // 7928eb0370d1133d0d8cd2f5ddfca19c309079d5.
1349 .mips, .mipsel => asm volatile (
1350 \\ move $sp, $t9
1351 \\ li $v0, 4091 # SYS_munmap
1352 \\ move $a0, %[ptr]
1353 \\ move $a1, %[len]
1354 \\ syscall
1355 \\ li $v0, 4001 # SYS_exit
1356 \\ li $a0, 0
1357 \\ syscall
1358 :
1359 : [ptr] "r" (@intFromPtr(self.mapped.ptr)),
1360 [len] "r" (self.mapped.len),
1361 : .{ .memory = true }),
1362 .mips64, .mips64el => asm volatile (switch (target.abi) {
1363 .gnuabin32, .muslabin32 =>
1364 \\ li $v0, 6011 # SYS_munmap
1365 \\ move $a0, %[ptr]
1366 \\ move $a1, %[len]
1367 \\ syscall
1368 \\ li $v0, 6058 # SYS_exit
1369 \\ li $a0, 0
1370 \\ syscall
1371 ,
1372 else =>
1373 \\ li $v0, 5011 # SYS_munmap
1374 \\ move $a0, %[ptr]
1375 \\ move $a1, %[len]
1376 \\ syscall
1377 \\ li $v0, 5058 # SYS_exit
1378 \\ li $a0, 0
1379 \\ syscall
1380 ,
1381 }
1382 :
1383 : [ptr] "r" (@intFromPtr(self.mapped.ptr)),
1384 [len] "r" (self.mapped.len),
1385 : .{ .memory = true }),
1386 .or1k => asm volatile (
1387 \\ l.ori r11, r0, 215 # SYS_munmap
1388 \\ l.ori r3, %[ptr]
1389 \\ l.ori r4, %[len]
1390 \\ l.sys 1
1391 \\ l.ori r11, r0, 93 # SYS_exit
1392 \\ l.ori r3, r0, r0
1393 \\ l.sys 1
1394 :
1395 : [ptr] "r" (@intFromPtr(self.mapped.ptr)),
1396 [len] "r" (self.mapped.len),
1397 : .{ .memory = true }),
1398 .powerpc, .powerpcle, .powerpc64, .powerpc64le => asm volatile (
1399 \\ li 0, 91 # SYS_munmap
1400 \\ mr 3, %[ptr]
1401 \\ mr 4, %[len]
1402 \\ sc
1403 \\ li 0, 1 # SYS_exit
1404 \\ li 3, 0
1405 \\ sc
1406 \\ blr
1407 :
1408 : [ptr] "r" (@intFromPtr(self.mapped.ptr)),
1409 [len] "r" (self.mapped.len),
1410 : .{ .memory = true }),
1411 .riscv32, .riscv64 => asm volatile (
1412 \\ li a7, 215 # SYS_munmap
1413 \\ mv a0, %[ptr]
1414 \\ mv a1, %[len]
1415 \\ ecall
1416 \\ li a7, 93 # SYS_exit
1417 \\ mv a0, zero
1418 \\ ecall
1419 :
1420 : [ptr] "r" (@intFromPtr(self.mapped.ptr)),
1421 [len] "r" (self.mapped.len),
1422 : .{ .memory = true }),
1423 .s390x => asm volatile (
1424 \\ lgr %%r2, %[ptr]
1425 \\ lgr %%r3, %[len]
1426 \\ svc 91 # SYS_munmap
1427 \\ lghi %%r2, 0
1428 \\ svc 1 # SYS_exit
1429 :
1430 : [ptr] "r" (@intFromPtr(self.mapped.ptr)),
1431 [len] "r" (self.mapped.len),
1432 : .{ .memory = true }),
1433 .sh, .sheb => asm volatile (
1434 \\ mov #91, r3 ! SYS_munmap
1435 \\ mov %[ptr], r4
1436 \\ mov %[len], r5
1437 \\ trapa #31
1438 \\ or r0, r0
1439 \\ or r0, r0
1440 \\ or r0, r0
1441 \\ or r0, r0
1442 \\ or r0, r0
1443 \\ mov #1, r3 ! SYS_exit
1444 \\ mov #0, r4
1445 \\ trapa #31
1446 \\ or r0, r0
1447 \\ or r0, r0
1448 \\ or r0, r0
1449 \\ or r0, r0
1450 \\ or r0, r0
1451 :
1452 : [ptr] "r" (@intFromPtr(self.mapped.ptr)),
1453 [len] "r" (self.mapped.len),
1454 : .{ .memory = true }),
1455 .sparc => asm volatile (
1456 \\ # See sparc64 comments below.
1457 \\ 1:
1458 \\ cmp %%fp, 0
1459 \\ beq 2f
1460 \\ nop
1461 \\ ba 1b
1462 \\ restore
1463 \\ 2:
1464 \\ mov 73, %%g1 // SYS_munmap
1465 \\ mov %[ptr], %%o0
1466 \\ mov %[len], %%o1
1467 \\ t 0x3 # ST_FLUSH_WINDOWS
1468 \\ t 0x10
1469 \\ mov 1, %%g1 // SYS_exit
1470 \\ mov 0, %%o0
1471 \\ t 0x10
1472 :
1473 : [ptr] "r" (@intFromPtr(self.mapped.ptr)),
1474 [len] "r" (self.mapped.len),
1475 : .{ .memory = true }),
1476 .sparc64 => asm volatile (
1477 \\ # SPARCs really don't like it when active stack frames
1478 \\ # is unmapped (it will result in a segfault), so we
1479 \\ # force-deactivate it by running `restore` until
1480 \\ # all frames are cleared.
1481 \\ 1:
1482 \\ cmp %%fp, 0
1483 \\ beq 2f
1484 \\ nop
1485 \\ ba 1b
1486 \\ restore
1487 \\ 2:
1488 \\ mov 73, %%g1 // SYS_munmap
1489 \\ mov %[ptr], %%o0
1490 \\ mov %[len], %%o1
1491 \\ # Flush register window contents to prevent background
1492 \\ # memory access before unmapping the stack.
1493 \\ flushw
1494 \\ t 0x6d
1495 \\ mov 1, %%g1 // SYS_exit
1496 \\ mov 0, %%o0
1497 \\ t 0x6d
1498 :
1499 : [ptr] "r" (@intFromPtr(self.mapped.ptr)),
1500 [len] "r" (self.mapped.len),
1501 : .{ .memory = true }),
1502 .loongarch32, .loongarch64 => asm volatile (
1503 \\ or $a0, $zero, %[ptr]
1504 \\ or $a1, $zero, %[len]
1505 \\ ori $a7, $zero, 215 # SYS_munmap
1506 \\ syscall 0 # call munmap
1507 \\ ori $a0, $zero, 0
1508 \\ ori $a7, $zero, 93 # SYS_exit
1509 \\ syscall 0 # call exit
1510 :
1511 : [ptr] "r" (@intFromPtr(self.mapped.ptr)),
1512 [len] "r" (self.mapped.len),
1513 : .{ .memory = true }),
1514 else => |cpu_arch| @compileError("Unsupported linux arch: " ++ @tagName(cpu_arch)),
1515 }
1516 unreachable;
1517 }
1518 };
1519
15201213 fn spawn(config: SpawnConfig, comptime f: anytype, args: anytype) !Impl {
15211214 const page_size = std.heap.pageSize();
15221215 const Args = @TypeOf(args);