authorgravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2025-10-24 13:05:13-07:00
committergravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2025-10-29 06:20:51-07:00
loga8f95e5176ebd734ccd2fd4d92cced6ab4cc2c07
tree1760bdd8a879a46cccc87ab71002e78b318b7aec
parent85e159e652afe976a6f720e041e88b196f062a9f

std.Io.Threaded: implement cancellation for pthreads

not to be confused with pthread_cancel, which is a useless API.

2 files changed, 47 insertions(+), 28 deletions(-)

lib/std/Io/Threaded.zig+45-28
......@@ -36,30 +36,47 @@ comptime {
3636 if (@TypeOf(posix.IOV_MAX) != void) assert(max_iovecs_len <= posix.IOV_MAX);
3737}
3838
39const CancelId = enum(usize) {
40 none = 0,
41 canceling = std.math.maxInt(usize),
42 _,
43
44 const ThreadId = if (std.Thread.use_pthreads) std.c.pthread_t else std.Thread.Id;
45
46 fn currentThread() CancelId {
47 if (std.Thread.use_pthreads) {
48 return @enumFromInt(@intFromPtr(std.c.pthread_self()));
49 } else {
50 return @enumFromInt(std.Thread.getCurrentId());
51 }
52 }
53
54 fn toThreadId(cancel_id: CancelId) ThreadId {
55 if (std.Thread.use_pthreads) {
56 return @ptrFromInt(@intFromEnum(cancel_id));
57 } else {
58 return @intCast(@intFromEnum(cancel_id));
59 }
60 }
61};
62
3963const Closure = struct {
4064 start: Start,
4165 node: std.SinglyLinkedList.Node = .{},
42 cancel_tid: std.Thread.Id,
66 cancel_tid: CancelId,
4367 /// Whether this task bumps minimum number of threads in the pool.
4468 is_concurrent: bool,
4569
4670 const Start = *const fn (*Closure) void;
4771
48 const canceling_tid: std.Thread.Id = switch (@typeInfo(std.Thread.Id)) {
49 .int => |int_info| switch (int_info.signedness) {
50 .signed => -1,
51 .unsigned => std.math.maxInt(std.Thread.Id),
52 },
53 .pointer => @ptrFromInt(std.math.maxInt(usize)),
54 else => @compileError("unsupported std.Thread.Id: " ++ @typeName(std.Thread.Id)),
55 };
56
5772 fn requestCancel(closure: *Closure) void {
58 switch (@atomicRmw(std.Thread.Id, &closure.cancel_tid, .Xchg, canceling_tid, .acq_rel)) {
59 0, canceling_tid => {},
73 switch (@atomicRmw(CancelId, &closure.cancel_tid, .Xchg, .canceling, .acq_rel)) {
74 .none, .canceling => {},
6075 else => |tid| switch (native_os) {
61 .linux => _ = std.os.linux.tgkill(std.os.linux.getpid(), @bitCast(tid), posix.SIG.IO),
62 else => {},
76 .linux => _ = std.os.linux.tgkill(std.os.linux.getpid(), @bitCast(tid.toThreadId()), posix.SIG.IO),
77 else => if (std.Thread.use_pthreads) {
78 assert(std.c.pthread_kill(tid.toThreadId(), posix.SIG.IO) == 0);
79 },
6380 },
6481 }
6582 }
......@@ -342,9 +359,9 @@ const AsyncClosure = struct {
342359
343360 fn start(closure: *Closure) void {
344361 const ac: *AsyncClosure = @alignCast(@fieldParentPtr("closure", closure));
345 const tid = std.Thread.getCurrentId();
346 if (@cmpxchgStrong(std.Thread.Id, &closure.cancel_tid, 0, tid, .acq_rel, .acquire)) |cancel_tid| {
347 assert(cancel_tid == Closure.canceling_tid);
362 const tid: CancelId = .currentThread();
363 if (@cmpxchgStrong(CancelId, &closure.cancel_tid, .none, tid, .acq_rel, .acquire)) |cancel_tid| {
364 assert(cancel_tid == .canceling);
348365 // Even though we already know the task is canceled, we must still
349366 // run the closure in order to make the return value valid and in
350367 // case there are side effects.
......@@ -355,8 +372,8 @@ const AsyncClosure = struct {
355372
356373 // In case a cancel happens after successful task completion, prevents
357374 // signal from being delivered to the thread in `requestCancel`.
358 if (@cmpxchgStrong(std.Thread.Id, &closure.cancel_tid, tid, 0, .acq_rel, .acquire)) |cancel_tid| {
359 assert(cancel_tid == Closure.canceling_tid);
375 if (@cmpxchgStrong(CancelId, &closure.cancel_tid, tid, .none, .acq_rel, .acquire)) |cancel_tid| {
376 assert(cancel_tid == .canceling);
360377 }
361378
362379 if (@atomicRmw(?*ResetEvent, &ac.select_condition, .Xchg, done_reset_event, .release)) |select_reset| {
......@@ -418,7 +435,7 @@ fn async(
418435
419436 ac.* = .{
420437 .closure = .{
421 .cancel_tid = 0,
438 .cancel_tid = .none,
422439 .start = AsyncClosure.start,
423440 .is_concurrent = false,
424441 },
......@@ -488,7 +505,7 @@ fn concurrent(
488505
489506 ac.* = .{
490507 .closure = .{
491 .cancel_tid = 0,
508 .cancel_tid = .none,
492509 .start = AsyncClosure.start,
493510 .is_concurrent = true,
494511 },
......@@ -540,12 +557,12 @@ const GroupClosure = struct {
540557
541558 fn start(closure: *Closure) void {
542559 const gc: *GroupClosure = @alignCast(@fieldParentPtr("closure", closure));
543 const tid = std.Thread.getCurrentId();
560 const tid: CancelId = .currentThread();
544561 const group = gc.group;
545562 const group_state: *std.atomic.Value(usize) = @ptrCast(&group.state);
546563 const reset_event: *ResetEvent = @ptrCast(&group.context);
547 if (@cmpxchgStrong(std.Thread.Id, &closure.cancel_tid, 0, tid, .acq_rel, .acquire)) |cancel_tid| {
548 assert(cancel_tid == Closure.canceling_tid);
564 if (@cmpxchgStrong(CancelId, &closure.cancel_tid, .none, tid, .acq_rel, .acquire)) |cancel_tid| {
565 assert(cancel_tid == .canceling);
549566 // We already know the task is canceled before running the callback. Since all closures
550567 // in a Group have void return type, we can return early.
551568 syncFinish(group_state, reset_event);
......@@ -557,8 +574,8 @@ const GroupClosure = struct {
557574
558575 // In case a cancel happens after successful task completion, prevents
559576 // signal from being delivered to the thread in `requestCancel`.
560 if (@cmpxchgStrong(std.Thread.Id, &closure.cancel_tid, tid, 0, .acq_rel, .acquire)) |cancel_tid| {
561 assert(cancel_tid == Closure.canceling_tid);
577 if (@cmpxchgStrong(CancelId, &closure.cancel_tid, tid, .none, .acq_rel, .acquire)) |cancel_tid| {
578 assert(cancel_tid == .canceling);
562579 }
563580
564581 syncFinish(group_state, reset_event);
......@@ -626,7 +643,7 @@ fn groupAsync(
626643 }));
627644 gc.* = .{
628645 .closure = .{
629 .cancel_tid = 0,
646 .cancel_tid = .none,
630647 .start = GroupClosure.start,
631648 .is_concurrent = false,
632649 },
......@@ -771,7 +788,7 @@ fn cancelRequested(userdata: ?*anyopaque) bool {
771788 const t: *Threaded = @ptrCast(@alignCast(userdata));
772789 _ = t;
773790 const closure = current_closure orelse return false;
774 return @atomicLoad(std.Thread.Id, &closure.cancel_tid, .acquire) == Closure.canceling_tid;
791 return @atomicLoad(CancelId, &closure.cancel_tid, .acquire) == .canceling;
775792}
776793
777794fn checkCancel(t: *Threaded) error{Canceled}!void {
lib/std/c.zig+2
......@@ -10763,6 +10763,8 @@ pub const pthread_setname_np = switch (native_os) {
1076310763};
1076410764
1076510765pub extern "c" fn pthread_getname_np(thread: pthread_t, name: [*:0]u8, len: usize) c_int;
10766pub extern "c" fn pthread_kill(pthread_t, signal: c_int) c_int;
10767
1076610768pub const pthread_threadid_np = switch (native_os) {
1076710769 .macos, .ios, .tvos, .watchos, .visionos => private.pthread_threadid_np,
1076810770 else => {},