authorgravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2025-10-28 15:06:07-07:00
committergravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2025-10-29 06:20:52-07:00
log03fd132b1ce82f767ffdb6cc886d1934b40c6071
treef243046bf40b9595edba59b49ab59686b8b03f67
parent6c794ce7bceeefceeee43a5514e633162ee946a9

std.Io: fix Group.wait unsoundness

Previously if a Group.wait was canceled, then a subsequent call to wait() or cancel() would trip an assertion in the synchronization code.

4 files changed, 33 insertions(+), 82 deletions(-)

lib/std/Io.zig+7-19
...@@ -653,8 +653,7 @@ pub const VTable = struct {...@@ -653,8 +653,7 @@ pub const VTable = struct {
653 context_alignment: std.mem.Alignment,653 context_alignment: std.mem.Alignment,
654 start: *const fn (*Group, context: *const anyopaque) void,654 start: *const fn (*Group, context: *const anyopaque) void,
655 ) void,655 ) void,
656 groupWait: *const fn (?*anyopaque, *Group, token: *anyopaque) Cancelable!void,656 groupWait: *const fn (?*anyopaque, *Group, token: *anyopaque) void,
657 groupWaitUncancelable: *const fn (?*anyopaque, *Group, token: *anyopaque) void,
658 groupCancel: *const fn (?*anyopaque, *Group, token: *anyopaque) void,657 groupCancel: *const fn (?*anyopaque, *Group, token: *anyopaque) void,
659658
660 /// Blocks until one of the futures from the list has a result ready, such659 /// Blocks until one of the futures from the list has a result ready, such
...@@ -1038,29 +1037,18 @@ pub const Group = struct {...@@ -1038,29 +1037,18 @@ pub const Group = struct {
1038 io.vtable.groupAsync(io.userdata, g, @ptrCast((&args)[0..1]), .of(Args), TypeErased.start);1037 io.vtable.groupAsync(io.userdata, g, @ptrCast((&args)[0..1]), .of(Args), TypeErased.start);
1039 }1038 }
10401039
1041 /// Blocks until all tasks of the group finish.1040 /// Blocks until all tasks of the group finish. During this time,
1042 ///1041 /// cancellation requests propagate to all members of the group.
1043 /// On success, further calls to `wait`, `waitUncancelable`, and `cancel`
1044 /// do nothing.
1045 ///
1046 /// Not threadsafe.
1047 pub fn wait(g: *Group, io: Io) Cancelable!void {
1048 const token = g.token orelse return;
1049 try io.vtable.groupWait(io.userdata, g, token);
1050 g.token = null;
1051 }
1052
1053 /// Equivalent to `wait` except uninterruptible.
1054 ///1042 ///
1055 /// Idempotent. Not threadsafe.1043 /// Idempotent. Not threadsafe.
1056 pub fn waitUncancelable(g: *Group, io: Io) void {1044 pub fn wait(g: *Group, io: Io) void {
1057 const token = g.token orelse return;1045 const token = g.token orelse return;
1058 g.token = null;1046 g.token = null;
1059 io.vtable.groupWaitUncancelable(io.userdata, g, token);1047 io.vtable.groupWait(io.userdata, g, token);
1060 }1048 }
10611049
1062 /// Equivalent to `wait` but requests cancellation on all tasks owned by1050 /// Equivalent to `wait` but immediately requests cancellation on all
1063 /// the group.1051 /// members of the group.
1064 ///1052 ///
1065 /// Idempotent. Not threadsafe.1053 /// Idempotent. Not threadsafe.
1066 pub fn cancel(g: *Group, io: Io) void {1054 pub fn cancel(g: *Group, io: Io) void {
lib/std/Io/Kqueue.zig+1-10
...@@ -859,7 +859,6 @@ pub fn io(k: *Kqueue) Io {...@@ -859,7 +859,6 @@ pub fn io(k: *Kqueue) Io {
859859
860 .groupAsync = groupAsync,860 .groupAsync = groupAsync,
861 .groupWait = groupWait,861 .groupWait = groupWait,
862 .groupWaitUncancelable = groupWaitUncancelable,
863 .groupCancel = groupCancel,862 .groupCancel = groupCancel,
864863
865 .mutexLock = mutexLock,864 .mutexLock = mutexLock,
...@@ -1027,15 +1026,7 @@ fn groupAsync(...@@ -1027,15 +1026,7 @@ fn groupAsync(
1027 @panic("TODO");1026 @panic("TODO");
1028}1027}
10291028
1030fn groupWait(userdata: ?*anyopaque, group: *Io.Group, token: *anyopaque) Io.Cancelable!void {1029fn groupWait(userdata: ?*anyopaque, group: *Io.Group, token: *anyopaque) void {
1031 const k: *Kqueue = @ptrCast(@alignCast(userdata));
1032 _ = k;
1033 _ = group;
1034 _ = token;
1035 @panic("TODO");
1036}
1037
1038fn groupWaitUncancelable(userdata: ?*anyopaque, group: *Io.Group, token: *anyopaque) void {
1039 const k: *Kqueue = @ptrCast(@alignCast(userdata));1030 const k: *Kqueue = @ptrCast(@alignCast(userdata));
1040 _ = k;1031 _ = k;
1041 _ = group;1032 _ = group;
lib/std/Io/Threaded.zig+23-50
...@@ -177,7 +177,6 @@ pub fn io(t: *Threaded) Io {...@@ -177,7 +177,6 @@ pub fn io(t: *Threaded) Io {
177177
178 .groupAsync = groupAsync,178 .groupAsync = groupAsync,
179 .groupWait = groupWait,179 .groupWait = groupWait,
180 .groupWaitUncancelable = groupWaitUncancelable,
181 .groupCancel = groupCancel,180 .groupCancel = groupCancel,
182181
183 .mutexLock = mutexLock,182 .mutexLock = mutexLock,
...@@ -274,7 +273,6 @@ pub fn ioBasic(t: *Threaded) Io {...@@ -274,7 +273,6 @@ pub fn ioBasic(t: *Threaded) Io {
274273
275 .groupAsync = groupAsync,274 .groupAsync = groupAsync,
276 .groupWait = groupWait,275 .groupWait = groupWait,
277 .groupWaitUncancelable = groupWaitUncancelable,
278 .groupCancel = groupCancel,276 .groupCancel = groupCancel,
279277
280 .mutexLock = mutexLock,278 .mutexLock = mutexLock,
...@@ -579,7 +577,9 @@ const GroupClosure = struct {...@@ -579,7 +577,9 @@ const GroupClosure = struct {
579 assert(cancel_tid == .canceling);577 assert(cancel_tid == .canceling);
580 }578 }
581579
582 syncFinish(group_state, reset_event);580 const prev_state = group_state.fetchSub(sync_one_pending, .acq_rel);
581 assert((prev_state / sync_one_pending) > 0);
582 if (prev_state == (sync_one_pending | sync_is_waiting)) reset_event.set();
583 }583 }
584584
585 fn free(gc: *GroupClosure, gpa: Allocator) void {585 fn free(gc: *GroupClosure, gpa: Allocator) void {
...@@ -602,29 +602,6 @@ const GroupClosure = struct {...@@ -602,29 +602,6 @@ const GroupClosure = struct {
602602
603 const sync_is_waiting: usize = 1 << 0;603 const sync_is_waiting: usize = 1 << 0;
604 const sync_one_pending: usize = 1 << 1;604 const sync_one_pending: usize = 1 << 1;
605
606 fn syncStart(state: *std.atomic.Value(usize)) void {
607 const prev_state = state.fetchAdd(sync_one_pending, .monotonic);
608 assert((prev_state / sync_one_pending) < (std.math.maxInt(usize) / sync_one_pending));
609 }
610
611 fn syncFinish(state: *std.atomic.Value(usize), event: *ResetEvent) void {
612 const prev_state = state.fetchSub(sync_one_pending, .acq_rel);
613 assert((prev_state / sync_one_pending) > 0);
614 if (prev_state == (sync_one_pending | sync_is_waiting)) event.set();
615 }
616
617 fn syncWait(t: *Threaded, state: *std.atomic.Value(usize), event: *ResetEvent) Io.Cancelable!void {
618 const prev_state = state.fetchAdd(sync_is_waiting, .acquire);
619 assert(prev_state & sync_is_waiting == 0);
620 if ((prev_state / sync_one_pending) > 0) try event.wait(t);
621 }
622
623 fn syncWaitUncancelable(state: *std.atomic.Value(usize), event: *ResetEvent) void {
624 const prev_state = state.fetchAdd(sync_is_waiting, .acquire);
625 assert(prev_state & sync_is_waiting == 0);
626 if ((prev_state / sync_one_pending) > 0) event.waitUncancelable();
627 }
628};605};
629606
630fn groupAsync(607fn groupAsync(
...@@ -686,32 +663,14 @@ fn groupAsync(...@@ -686,32 +663,14 @@ fn groupAsync(
686 // This needs to be done before unlocking the mutex to avoid a race with663 // This needs to be done before unlocking the mutex to avoid a race with
687 // the associated task finishing.664 // the associated task finishing.
688 const group_state: *std.atomic.Value(usize) = @ptrCast(&group.state);665 const group_state: *std.atomic.Value(usize) = @ptrCast(&group.state);
689 GroupClosure.syncStart(group_state);666 const prev_state = group_state.fetchAdd(GroupClosure.sync_one_pending, .monotonic);
667 assert((prev_state / GroupClosure.sync_one_pending) < (std.math.maxInt(usize) / GroupClosure.sync_one_pending));
690668
691 t.mutex.unlock();669 t.mutex.unlock();
692 t.cond.signal();670 t.cond.signal();
693}671}
694672
695fn groupWait(userdata: ?*anyopaque, group: *Io.Group, token: *anyopaque) Io.Cancelable!void {673fn groupWait(userdata: ?*anyopaque, group: *Io.Group, token: *anyopaque) void {
696 const t: *Threaded = @ptrCast(@alignCast(userdata));
697 const gpa = t.allocator;
698
699 if (builtin.single_threaded) return;
700
701 const group_state: *std.atomic.Value(usize) = @ptrCast(&group.state);
702 const reset_event: *ResetEvent = @ptrCast(&group.context);
703 try GroupClosure.syncWait(t, group_state, reset_event);
704
705 var node: *std.SinglyLinkedList.Node = @ptrCast(@alignCast(token));
706 while (true) {
707 const gc: *GroupClosure = @fieldParentPtr("node", node);
708 const node_next = node.next;
709 gc.free(gpa);
710 node = node_next orelse break;
711 }
712}
713
714fn groupWaitUncancelable(userdata: ?*anyopaque, group: *Io.Group, token: *anyopaque) void {
715 const t: *Threaded = @ptrCast(@alignCast(userdata));674 const t: *Threaded = @ptrCast(@alignCast(userdata));
716 const gpa = t.allocator;675 const gpa = t.allocator;
717676
...@@ -719,7 +678,19 @@ fn groupWaitUncancelable(userdata: ?*anyopaque, group: *Io.Group, token: *anyopa...@@ -719,7 +678,19 @@ fn groupWaitUncancelable(userdata: ?*anyopaque, group: *Io.Group, token: *anyopa
719678
720 const group_state: *std.atomic.Value(usize) = @ptrCast(&group.state);679 const group_state: *std.atomic.Value(usize) = @ptrCast(&group.state);
721 const reset_event: *ResetEvent = @ptrCast(&group.context);680 const reset_event: *ResetEvent = @ptrCast(&group.context);
722 GroupClosure.syncWaitUncancelable(group_state, reset_event);681 const prev_state = group_state.fetchAdd(GroupClosure.sync_is_waiting, .acquire);
682 assert(prev_state & GroupClosure.sync_is_waiting == 0);
683 if ((prev_state / GroupClosure.sync_one_pending) > 0) reset_event.wait(t) catch |err| switch (err) {
684 error.Canceled => {
685 var node: *std.SinglyLinkedList.Node = @ptrCast(@alignCast(token));
686 while (true) {
687 const gc: *GroupClosure = @fieldParentPtr("node", node);
688 gc.closure.requestCancel();
689 node = node.next orelse break;
690 }
691 reset_event.waitUncancelable();
692 },
693 };
723694
724 var node: *std.SinglyLinkedList.Node = @ptrCast(@alignCast(token));695 var node: *std.SinglyLinkedList.Node = @ptrCast(@alignCast(token));
725 while (true) {696 while (true) {
...@@ -747,7 +718,9 @@ fn groupCancel(userdata: ?*anyopaque, group: *Io.Group, token: *anyopaque) void...@@ -747,7 +718,9 @@ fn groupCancel(userdata: ?*anyopaque, group: *Io.Group, token: *anyopaque) void
747718
748 const group_state: *std.atomic.Value(usize) = @ptrCast(&group.state);719 const group_state: *std.atomic.Value(usize) = @ptrCast(&group.state);
749 const reset_event: *ResetEvent = @ptrCast(&group.context);720 const reset_event: *ResetEvent = @ptrCast(&group.context);
750 GroupClosure.syncWaitUncancelable(group_state, reset_event);721 const prev_state = group_state.fetchAdd(GroupClosure.sync_is_waiting, .acquire);
722 assert(prev_state & GroupClosure.sync_is_waiting == 0);
723 if ((prev_state / GroupClosure.sync_one_pending) > 0) reset_event.waitUncancelable();
751724
752 {725 {
753 var node: *std.SinglyLinkedList.Node = @ptrCast(@alignCast(token));726 var node: *std.SinglyLinkedList.Node = @ptrCast(@alignCast(token));
...@@ -1549,7 +1522,7 @@ fn dirAccessPosix(...@@ -1549,7 +1522,7 @@ fn dirAccessPosix(
1549 .FAULT => |err| return errnoBug(err),1522 .FAULT => |err| return errnoBug(err),
1550 .IO => return error.InputOutput,1523 .IO => return error.InputOutput,
1551 .NOMEM => return error.SystemResources,1524 .NOMEM => return error.SystemResources,
1552 .ILSEQ => return error.BadPathName, // TODO move to wasi1525 .ILSEQ => return error.BadPathName,
1553 else => |err| return posix.unexpectedErrno(err),1526 else => |err| return posix.unexpectedErrno(err),
1554 }1527 }
1555 }1528 }
lib/std/Io/net/HostName.zig+2-3
...@@ -280,9 +280,8 @@ pub fn connectMany(...@@ -280,9 +280,8 @@ pub fn connectMany(
280 .address => |address| group.async(io, enqueueConnection, .{ address, io, results, options }),280 .address => |address| group.async(io, enqueueConnection, .{ address, io, results, options }),
281 .canonical_name => continue,281 .canonical_name => continue,
282 .end => |lookup_result| {282 .end => |lookup_result| {
283 results.putOneUncancelable(io, .{283 group.wait(io);
284 .end = if (group.wait(io)) lookup_result else |err| err,284 results.putOneUncancelable(io, .{ .end = lookup_result });
285 });
286 return;285 return;
287 },286 },
288 } else |err| switch (err) {287 } else |err| switch (err) {