authorgravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2025-10-15 13:48:51-07:00
committergravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2025-10-29 06:20:49-07:00
log060fd975d95d4472f98bb2c7760afb111d162580
tree5729e2d2ea7134f2ad11b3dd6fb19ba1b5a8eecb
parent10bfbd7d60b721af9bed3f74fa7ab5171471ee46

std.Io.Group: add cancellation support to "wait"

4 files changed, 216 insertions(+), 32 deletions(-)

BRANCH_TODO+1-1
......@@ -3,7 +3,7 @@
33* Threaded: finish windows impl
44* Threaded: glibc impl of netLookup
55
6* fix Group.wait not handling cancelation (need to move impl of ResetEvent to Threaded)
6* eliminate dependency on std.Thread (Mutex, Condition, maybe more)
77* implement cancelRequest for non-linux posix
88* finish converting all Threaded into directly calling system functions and handling EINTR
99* audit the TODOs
lib/std/Io.zig+15-5
......@@ -641,12 +641,13 @@ pub const VTable = struct {
641641 context_alignment: std.mem.Alignment,
642642 start: *const fn (*Group, context: *const anyopaque) void,
643643 ) void,
644 groupWait: *const fn (?*anyopaque, *Group, token: *anyopaque) void,
644 groupWait: *const fn (?*anyopaque, *Group, token: *anyopaque) Cancelable!void,
645 groupWaitUncancelable: *const fn (?*anyopaque, *Group, token: *anyopaque) void,
645646 groupCancel: *const fn (?*anyopaque, *Group, token: *anyopaque) void,
646647
647648 /// Blocks until one of the futures from the list has a result ready, such
648649 /// that awaiting it will not block. Returns that index.
649 select: *const fn (?*anyopaque, futures: []const *AnyFuture) usize,
650 select: *const fn (?*anyopaque, futures: []const *AnyFuture) Cancelable!usize,
650651
651652 mutexLock: *const fn (?*anyopaque, prev_state: Mutex.State, mutex: *Mutex) Cancelable!void,
652653 mutexLockUncancelable: *const fn (?*anyopaque, prev_state: Mutex.State, mutex: *Mutex) void,
......@@ -1017,10 +1018,19 @@ pub const Group = struct {
10171018 /// Blocks until all tasks of the group finish.
10181019 ///
10191020 /// Idempotent. Not threadsafe.
1020 pub fn wait(g: *Group, io: Io) void {
1021 pub fn wait(g: *Group, io: Io) Cancelable!void {
10211022 const token = g.token orelse return;
10221023 g.token = null;
1023 io.vtable.groupWait(io.userdata, g, token);
1024 return io.vtable.groupWait(io.userdata, g, token);
1025 }
1026
1027 /// Equivalent to `wait` except uninterruptible.
1028 ///
1029 /// Idempotent. Not threadsafe.
1030 pub fn waitUncancelable(g: *Group, io: Io) void {
1031 const token = g.token orelse return;
1032 g.token = null;
1033 io.vtable.groupWaitUncancelable(io.userdata, g, token);
10241034 }
10251035
10261036 /// Equivalent to `wait` but requests cancellation on all tasks owned by
......@@ -1095,7 +1105,7 @@ pub fn Select(comptime U: type) type {
10951105 /// Asserts there is at least one more `outstanding` task.
10961106 ///
10971107 /// Not threadsafe.
1098 pub fn wait(s: *S) Io.Cancelable!U {
1108 pub fn wait(s: *S) Cancelable!U {
10991109 s.outstanding -= 1;
11001110 return s.queue.getOne(s.io);
11011111 }
lib/std/Io/Threaded.zig+199-25
......@@ -13,7 +13,6 @@ const IpAddress = std.Io.net.IpAddress;
1313const Allocator = std.mem.Allocator;
1414const assert = std.debug.assert;
1515const posix = std.posix;
16const ResetEvent = std.Thread.ResetEvent;
1716
1817/// Thread-safe.
1918allocator: Allocator,
......@@ -153,8 +152,10 @@ pub fn io(t: *Threaded) Io {
153152 .cancel = cancel,
154153 .cancelRequested = cancelRequested,
155154 .select = select,
155
156156 .groupAsync = groupAsync,
157157 .groupWait = groupWait,
158 .groupWaitUncancelable = groupWaitUncancelable,
158159 .groupCancel = groupCancel,
159160
160161 .mutexLock = mutexLock,
......@@ -300,7 +301,7 @@ const AsyncClosure = struct {
300301 }
301302
302303 fn waitAndFree(ac: *AsyncClosure, gpa: Allocator, result: []u8) void {
303 ac.reset_event.wait();
304 ac.reset_event.waitUncancelable();
304305 @memcpy(result, ac.resultPointer()[0..result.len]);
305306 free(ac, gpa, result.len);
306307 }
......@@ -472,7 +473,7 @@ const GroupClosure = struct {
472473 assert(cancel_tid == Closure.canceling_tid);
473474 // We already know the task is canceled before running the callback. Since all closures
474475 // in a Group have void return type, we can return early.
475 std.Thread.WaitGroup.finishStateless(group_state, reset_event);
476 syncFinish(group_state, reset_event);
476477 return;
477478 }
478479 current_closure = closure;
......@@ -485,7 +486,7 @@ const GroupClosure = struct {
485486 assert(cancel_tid == Closure.canceling_tid);
486487 }
487488
488 std.Thread.WaitGroup.finishStateless(group_state, reset_event);
489 syncFinish(group_state, reset_event);
489490 }
490491
491492 fn free(gc: *GroupClosure, gpa: Allocator) void {
......@@ -505,6 +506,32 @@ const GroupClosure = struct {
505506 const base: [*]u8 = @ptrCast(gc);
506507 return base + contextOffset(gc.context_alignment);
507508 }
509
510 const sync_is_waiting: usize = 1 << 0;
511 const sync_one_pending: usize = 1 << 1;
512
513 fn syncStart(state: *std.atomic.Value(usize)) void {
514 const prev_state = state.fetchAdd(sync_one_pending, .monotonic);
515 assert((prev_state / sync_one_pending) < (std.math.maxInt(usize) / sync_one_pending));
516 }
517
518 fn syncFinish(state: *std.atomic.Value(usize), event: *ResetEvent) void {
519 const prev_state = state.fetchSub(sync_one_pending, .acq_rel);
520 assert((prev_state / sync_one_pending) > 0);
521 if (prev_state == (sync_one_pending | sync_is_waiting)) event.set();
522 }
523
524 fn syncWait(t: *Threaded, state: *std.atomic.Value(usize), event: *ResetEvent) Io.Cancelable!void {
525 const prev_state = state.fetchAdd(sync_is_waiting, .acquire);
526 assert(prev_state & sync_is_waiting == 0);
527 if ((prev_state / sync_one_pending) > 0) try event.wait(t);
528 }
529
530 fn syncWaitUncancelable(state: *std.atomic.Value(usize), event: *ResetEvent) void {
531 const prev_state = state.fetchAdd(sync_is_waiting, .acquire);
532 assert(prev_state & sync_is_waiting == 0);
533 if ((prev_state / sync_one_pending) > 0) event.waitUncancelable();
534 }
508535};
509536
510537fn groupAsync(
......@@ -566,22 +593,40 @@ fn groupAsync(
566593 // This needs to be done before unlocking the mutex to avoid a race with
567594 // the associated task finishing.
568595 const group_state: *std.atomic.Value(usize) = @ptrCast(&group.state);
569 std.Thread.WaitGroup.startStateless(group_state);
596 GroupClosure.syncStart(group_state);
570597
571598 t.mutex.unlock();
572599 t.cond.signal();
573600}
574601
575fn groupWait(userdata: ?*anyopaque, group: *Io.Group, token: *anyopaque) void {
602fn groupWait(userdata: ?*anyopaque, group: *Io.Group, token: *anyopaque) Io.Cancelable!void {
576603 const t: *Threaded = @ptrCast(@alignCast(userdata));
577604 const gpa = t.allocator;
578605
579606 if (builtin.single_threaded) return;
580607
581 // TODO these primitives are too high level, need to check cancel on EINTR
582608 const group_state: *std.atomic.Value(usize) = @ptrCast(&group.state);
583609 const reset_event: *ResetEvent = @ptrCast(&group.context);
584 std.Thread.WaitGroup.waitStateless(group_state, reset_event);
610 try GroupClosure.syncWait(t, group_state, reset_event);
611
612 var node: *std.SinglyLinkedList.Node = @ptrCast(@alignCast(token));
613 while (true) {
614 const gc: *GroupClosure = @fieldParentPtr("node", node);
615 const node_next = node.next;
616 gc.free(gpa);
617 node = node_next orelse break;
618 }
619}
620
621fn groupWaitUncancelable(userdata: ?*anyopaque, group: *Io.Group, token: *anyopaque) void {
622 const t: *Threaded = @ptrCast(@alignCast(userdata));
623 const gpa = t.allocator;
624
625 if (builtin.single_threaded) return;
626
627 const group_state: *std.atomic.Value(usize) = @ptrCast(&group.state);
628 const reset_event: *ResetEvent = @ptrCast(&group.context);
629 GroupClosure.syncWaitUncancelable(group_state, reset_event);
585630
586631 var node: *std.SinglyLinkedList.Node = @ptrCast(@alignCast(token));
587632 while (true) {
......@@ -609,7 +654,7 @@ fn groupCancel(userdata: ?*anyopaque, group: *Io.Group, token: *anyopaque) void
609654
610655 const group_state: *std.atomic.Value(usize) = @ptrCast(&group.state);
611656 const reset_event: *ResetEvent = @ptrCast(&group.context);
612 std.Thread.WaitGroup.waitStateless(group_state, reset_event);
657 GroupClosure.syncWaitUncancelable(group_state, reset_event);
613658
614659 {
615660 var node: *std.SinglyLinkedList.Node = @ptrCast(@alignCast(token));
......@@ -661,22 +706,20 @@ fn checkCancel(t: *Threaded) error{Canceled}!void {
661706fn mutexLock(userdata: ?*anyopaque, prev_state: Io.Mutex.State, mutex: *Io.Mutex) Io.Cancelable!void {
662707 const t: *Threaded = @ptrCast(@alignCast(userdata));
663708 if (prev_state == .contended) {
664 try t.checkCancel();
665 futexWait(@ptrCast(&mutex.state), @intFromEnum(Io.Mutex.State.contended));
709 try futexWait(t, @ptrCast(&mutex.state), @intFromEnum(Io.Mutex.State.contended));
666710 }
667711 while (@atomicRmw(Io.Mutex.State, &mutex.state, .Xchg, .contended, .acquire) != .unlocked) {
668 try t.checkCancel();
669 futexWait(@ptrCast(&mutex.state), @intFromEnum(Io.Mutex.State.contended));
712 try futexWait(t, @ptrCast(&mutex.state), @intFromEnum(Io.Mutex.State.contended));
670713 }
671714}
672715
673716fn mutexLockUncancelable(userdata: ?*anyopaque, prev_state: Io.Mutex.State, mutex: *Io.Mutex) void {
674717 _ = userdata;
675718 if (prev_state == .contended) {
676 futexWait(@ptrCast(&mutex.state), @intFromEnum(Io.Mutex.State.contended));
719 futexWaitUncancelable(@ptrCast(&mutex.state), @intFromEnum(Io.Mutex.State.contended));
677720 }
678721 while (@atomicRmw(Io.Mutex.State, &mutex.state, .Xchg, .contended, .acquire) != .unlocked) {
679 futexWait(@ptrCast(&mutex.state), @intFromEnum(Io.Mutex.State.contended));
722 futexWaitUncancelable(@ptrCast(&mutex.state), @intFromEnum(Io.Mutex.State.contended));
680723 }
681724}
682725
......@@ -708,7 +751,7 @@ fn conditionWaitUncancelable(userdata: ?*anyopaque, cond: *Io.Condition, mutex:
708751 defer mutex.lockUncancelable(t_io);
709752
710753 while (true) {
711 futexWait(cond_epoch, epoch);
754 futexWaitUncancelable(cond_epoch, epoch);
712755 epoch = cond_epoch.load(.acquire);
713756 state = cond_state.load(.monotonic);
714757 while (state & signal_mask != 0) {
......@@ -747,8 +790,7 @@ fn conditionWait(userdata: ?*anyopaque, cond: *Io.Condition, mutex: *Io.Mutex) I
747790 defer mutex.lockUncancelable(t.io());
748791
749792 while (true) {
750 try t.checkCancel();
751 futexWait(cond_epoch, epoch);
793 try futexWait(t, cond_epoch, epoch);
752794
753795 epoch = cond_epoch.load(.acquire);
754796 state = cond_state.load(.monotonic);
......@@ -1708,9 +1750,8 @@ fn sleepPosix(userdata: ?*anyopaque, timeout: Io.Timeout) Io.SleepError!void {
17081750 }
17091751}
17101752
1711fn select(userdata: ?*anyopaque, futures: []const *Io.AnyFuture) usize {
1753fn select(userdata: ?*anyopaque, futures: []const *Io.AnyFuture) Io.Cancelable!usize {
17121754 const t: *Threaded = @ptrCast(@alignCast(userdata));
1713 _ = t;
17141755
17151756 var reset_event: ResetEvent = .unset;
17161757
......@@ -1720,20 +1761,20 @@ fn select(userdata: ?*anyopaque, futures: []const *Io.AnyFuture) usize {
17201761 for (futures[0..i]) |cleanup_future| {
17211762 const cleanup_closure: *AsyncClosure = @ptrCast(@alignCast(cleanup_future));
17221763 if (@atomicRmw(?*ResetEvent, &cleanup_closure.select_condition, .Xchg, null, .seq_cst) == AsyncClosure.done_reset_event) {
1723 cleanup_closure.reset_event.wait(); // Ensure no reference to our stack-allocated reset_event.
1764 cleanup_closure.reset_event.waitUncancelable(); // Ensure no reference to our stack-allocated reset_event.
17241765 }
17251766 }
17261767 return i;
17271768 }
17281769 }
17291770
1730 reset_event.wait();
1771 try reset_event.wait(t);
17311772
17321773 var result: ?usize = null;
17331774 for (futures, 0..) |future, i| {
17341775 const closure: *AsyncClosure = @ptrCast(@alignCast(future));
17351776 if (@atomicRmw(?*ResetEvent, &closure.select_condition, .Xchg, null, .seq_cst) == AsyncClosure.done_reset_event) {
1736 closure.reset_event.wait(); // Ensure no reference to our stack-allocated reset_event.
1777 closure.reset_event.waitUncancelable(); // Ensure no reference to our stack-allocated reset_event.
17371778 if (result == null) result = i; // In case multiple are ready, return first.
17381779 }
17391780 }
......@@ -3320,11 +3361,12 @@ fn copyCanon(canonical_name_buffer: *[HostName.max_len]u8, name: []const u8) Hos
33203361 return .{ .bytes = dest };
33213362}
33223363
3323pub fn futexWait(ptr: *const std.atomic.Value(u32), expect: u32) void {
3364fn futexWait(t: *Threaded, ptr: *const std.atomic.Value(u32), expect: u32) Io.Cancelable!void {
33243365 @branchHint(.cold);
33253366
33263367 if (native_os == .linux) {
33273368 const linux = std.os.linux;
3369 try t.checkCancel();
33283370 const rc = linux.futex_4arg(ptr, .{ .cmd = .WAIT, .private = true }, expect, null);
33293371 if (builtin.mode == .Debug) switch (linux.E.init(rc)) {
33303372 .SUCCESS => {}, // notified by `wake()`
......@@ -3341,7 +3383,28 @@ pub fn futexWait(ptr: *const std.atomic.Value(u32), expect: u32) void {
33413383 @compileError("TODO");
33423384}
33433385
3344pub fn futexWaitDuration(ptr: *const std.atomic.Value(u32), expect: u32, timeout: Io.Duration) void {
3386pub fn futexWaitUncancelable(ptr: *const std.atomic.Value(u32), expect: u32) void {
3387 @branchHint(.cold);
3388
3389 if (native_os == .linux) {
3390 const linux = std.os.linux;
3391 const rc = linux.futex_4arg(ptr, .{ .cmd = .WAIT, .private = true }, expect, null);
3392 if (builtin.mode == .Debug) switch (linux.E.init(rc)) {
3393 .SUCCESS => {}, // notified by `wake()`
3394 .INTR => {}, // gives caller a chance to check cancellation
3395 .AGAIN => {}, // ptr.* != expect
3396 .INVAL => {}, // possibly timeout overflow
3397 .TIMEDOUT => unreachable,
3398 .FAULT => unreachable, // ptr was invalid
3399 else => unreachable,
3400 };
3401 return;
3402 }
3403
3404 @compileError("TODO");
3405}
3406
3407pub fn futexWaitDurationUncancelable(ptr: *const std.atomic.Value(u32), expect: u32, timeout: Io.Duration) void {
33453408 @branchHint(.cold);
33463409
33473410 if (native_os == .linux) {
......@@ -3384,3 +3447,114 @@ pub fn futexWake(ptr: *const std.atomic.Value(u32), max_waiters: u32) void {
33843447
33853448 @compileError("TODO");
33863449}
3450
3451/// A thread-safe logical boolean value which can be `set` and `unset`.
3452///
3453/// It can also block threads until the value is set with cancelation via timed
3454/// waits. Statically initializable; four bytes on all targets.
3455pub const ResetEvent = enum(u32) {
3456 unset = 0,
3457 waiting = 1,
3458 is_set = 2,
3459
3460 /// Returns whether the logical boolean is `set`.
3461 ///
3462 /// Once `reset` is called, this returns false until the next `set`.
3463 ///
3464 /// The memory accesses before the `set` can be said to happen before
3465 /// `isSet` returns true.
3466 pub fn isSet(re: *const ResetEvent) bool {
3467 if (builtin.single_threaded) return switch (re.*) {
3468 .unset => false,
3469 .waiting => unreachable,
3470 .is_set => true,
3471 };
3472 // Acquire barrier ensures memory accesses before `set` happen before
3473 // returning true.
3474 return @atomicLoad(ResetEvent, re, .acquire) == .is_set;
3475 }
3476
3477 /// Blocks the calling thread until `set` is called.
3478 ///
3479 /// This is effectively a more efficient version of `while (!isSet()) {}`.
3480 ///
3481 /// The memory accesses before the `set` can be said to happen before `wait` returns.
3482 pub fn wait(re: *ResetEvent, t: *Threaded) Io.Cancelable!void {
3483 if (builtin.single_threaded) switch (re.*) {
3484 .unset => unreachable, // Deadlock, no other threads to wake us up.
3485 .waiting => unreachable, // Invalid state.
3486 .is_set => return,
3487 };
3488 if (re.isSet()) {
3489 @branchHint(.likely);
3490 return;
3491 }
3492 // Try to set the state from `unset` to `waiting` to indicate to the
3493 // `set` thread that others are blocked on the ResetEvent. Avoid using
3494 // any strict barriers until we know the ResetEvent is set.
3495 var state = @atomicLoad(ResetEvent, re, .acquire);
3496 if (state == .unset) {
3497 state = @cmpxchgStrong(ResetEvent, re, state, .waiting, .acquire, .acquire) orelse .waiting;
3498 }
3499 while (state == .waiting) {
3500 try futexWait(t, @ptrCast(re), @intFromEnum(ResetEvent.waiting));
3501 state = @atomicLoad(ResetEvent, re, .acquire);
3502 }
3503 assert(state == .is_set);
3504 }
3505
3506 /// Same as `wait` except uninterruptible.
3507 pub fn waitUncancelable(re: *ResetEvent) void {
3508 if (builtin.single_threaded) switch (re.*) {
3509 .unset => unreachable, // Deadlock, no other threads to wake us up.
3510 .waiting => unreachable, // Invalid state.
3511 .is_set => return,
3512 };
3513 if (re.isSet()) {
3514 @branchHint(.likely);
3515 return;
3516 }
3517 // Try to set the state from `unset` to `waiting` to indicate to the
3518 // `set` thread that others are blocked on the ResetEvent. Avoid using
3519 // any strict barriers until we know the ResetEvent is set.
3520 var state = @atomicLoad(ResetEvent, re, .acquire);
3521 if (state == .unset) {
3522 state = @cmpxchgStrong(ResetEvent, re, state, .waiting, .acquire, .acquire) orelse .waiting;
3523 }
3524 while (state == .waiting) {
3525 futexWaitUncancelable(@ptrCast(re), @intFromEnum(ResetEvent.waiting));
3526 state = @atomicLoad(ResetEvent, re, .acquire);
3527 }
3528 assert(state == .is_set);
3529 }
3530
3531 /// Marks the logical boolean as `set` and unblocks any threads in `wait`
3532 /// or `timedWait` to observe the new state.
3533 ///
3534 /// The logical boolean stays `set` until `reset` is called, making future
3535 /// `set` calls do nothing semantically.
3536 ///
3537 /// The memory accesses before `set` can be said to happen before `isSet`
3538 /// returns true or `wait`/`timedWait` return successfully.
3539 pub fn set(re: *ResetEvent) void {
3540 if (builtin.single_threaded) {
3541 re.* = .is_set;
3542 return;
3543 }
3544 if (@atomicRmw(ResetEvent, re, .Xchg, .is_set, .release) == .waiting) {
3545 futexWake(@ptrCast(re), std.math.maxInt(u32));
3546 }
3547 }
3548
3549 /// Unmarks the ResetEvent as if `set` was never called.
3550 ///
3551 /// Assumes no threads are blocked in `wait` or `timedWait`. Concurrent
3552 /// calls to `set`, `isSet` and `reset` are allowed.
3553 pub fn reset(re: *ResetEvent) void {
3554 if (builtin.single_threaded) {
3555 re.* = .unset;
3556 return;
3557 }
3558 @atomicStore(ResetEvent, re, .unset, .monotonic);
3559 }
3560};
lib/std/Io/net/HostName.zig+1-1
......@@ -273,7 +273,7 @@ pub fn connectMany(
273273 .address => |address| group.async(io, enqueueConnection, .{ address, io, results, options }),
274274 .canonical_name => continue,
275275 .end => |lookup_result| {
276 group.wait(io);
276 group.waitUncancelable(io);
277277 results.putOneUncancelable(io, .{ .end = lookup_result });
278278 return;
279279 },