authorgravatar for mlugg@mlugg.co.ukMatthew Lugg <mlugg@mlugg.co.uk> 2025-12-21 14:51:06+00:00
committergravatar for mlugg@mlugg.co.ukMatthew Lugg <mlugg@mlugg.co.uk> 2025-12-22 11:54:57+00:00
log4772f1a9f418d91e3e157b4f45a8edb7ed7975f0
tree4315cf812b9148738c3e67501b4ab2d14e896d7e
parent49f17237b0332680cebf3e4d83dedc3db46356e9
signaturelock-open Commit is signed but in an unrecognized format.

std.Io: make it safe to call `Group.async` within a group task

2 files changed, 35 insertions(+), 12 deletions(-)

lib/std/Io.zig+19-6
...@@ -1016,9 +1016,14 @@ pub fn Future(Result: type) type {...@@ -1016,9 +1016,14 @@ pub fn Future(Result: type) type {
1016pub const Group = struct {1016pub const Group = struct {
1017 state: usize,1017 state: usize,
1018 context: ?*anyopaque,1018 context: ?*anyopaque,
1019 token: ?*anyopaque,1019 /// This value indicates whether or not a group has pending tasks. `null`
1020 /// means there are no pending tasks, and no resources associated with the
1021 /// group, so `await` and `cancel` return immediately without calling the
1022 /// implementation. This means that `token` must be accessed atomically to
1023 /// avoid racing with the check in `await` and `cancel`.
1024 token: std.atomic.Value(?*anyopaque),
10201025
1021 pub const init: Group = .{ .state = 0, .context = null, .token = null };1026 pub const init: Group = .{ .state = 0, .context = null, .token = .init(null) };
10221027
1023 /// Calls `function` with `args` asynchronously. The resource spawned is1028 /// Calls `function` with `args` asynchronously. The resource spawned is
1024 /// owned by the group.1029 /// owned by the group.
...@@ -1081,10 +1086,14 @@ pub const Group = struct {...@@ -1081,10 +1086,14 @@ pub const Group = struct {
1081 /// cancellation requests propagate to all members of the group.1086 /// cancellation requests propagate to all members of the group.
1082 ///1087 ///
1083 /// Idempotent. Not threadsafe.1088 /// Idempotent. Not threadsafe.
1089 ///
1090 /// It is safe to call this function concurrently with `Group.async` or
1091 /// `Group.concurrent`, provided that the group does not complete until
1092 /// the call to `Group.async` or `Group.concurrent` returns.
1084 pub fn wait(g: *Group, io: Io) void {1093 pub fn wait(g: *Group, io: Io) void {
1085 const token = g.token orelse return;1094 const token = g.token.load(.acquire) orelse return;
1086 g.token = null;
1087 io.vtable.groupWait(io.userdata, g, token);1095 io.vtable.groupWait(io.userdata, g, token);
1096 assert(g.token.raw == null);
1088 }1097 }
10891098
1090 /// Equivalent to `wait` but immediately requests cancellation on all1099 /// Equivalent to `wait` but immediately requests cancellation on all
...@@ -1093,10 +1102,14 @@ pub const Group = struct {...@@ -1093,10 +1102,14 @@ pub const Group = struct {
1093 /// For a description of cancelation and cancelation points, see `Future.cancel`.1102 /// For a description of cancelation and cancelation points, see `Future.cancel`.
1094 ///1103 ///
1095 /// Idempotent. Not threadsafe.1104 /// Idempotent. Not threadsafe.
1105 ///
1106 /// It is safe to call this function concurrently with `Group.async` or
1107 /// `Group.concurrent`, provided that the group does not complete until
1108 /// the call to `Group.async` or `Group.concurrent` returns.
1096 pub fn cancel(g: *Group, io: Io) void {1109 pub fn cancel(g: *Group, io: Io) void {
1097 const token = g.token orelse return;1110 const token = g.token.load(.acquire) orelse return;
1098 g.token = null;
1099 io.vtable.groupCancel(io.userdata, g, token);1111 io.vtable.groupCancel(io.userdata, g, token);
1112 assert(g.token.raw == null);
1100 }1113 }
1101};1114};
11021115
lib/std/Io/Threaded.zig+16-6
...@@ -1117,8 +1117,8 @@ fn groupAsync(...@@ -1117,8 +1117,8 @@ fn groupAsync(
1117 }1117 }
11181118
1119 // Append to the group linked list inside the mutex to make `Io.Group.async` thread-safe.1119 // Append to the group linked list inside the mutex to make `Io.Group.async` thread-safe.
1120 gc.node = .{ .next = @ptrCast(@alignCast(group.token)) };1120 gc.node = .{ .next = @ptrCast(@alignCast(group.token.load(.monotonic))) };
1121 group.token = &gc.node;1121 group.token.store(&gc.node, .monotonic);
11221122
1123 t.run_queue.prepend(&gc.closure.node);1123 t.run_queue.prepend(&gc.closure.node);
11241124
...@@ -1169,8 +1169,8 @@ fn groupConcurrent(...@@ -1169,8 +1169,8 @@ fn groupConcurrent(
1169 }1169 }
11701170
1171 // Append to the group linked list inside the mutex to make `Io.Group.concurrent` thread-safe.1171 // Append to the group linked list inside the mutex to make `Io.Group.concurrent` thread-safe.
1172 gc.node = .{ .next = @ptrCast(@alignCast(group.token)) };1172 gc.node = .{ .next = @ptrCast(@alignCast(group.token.load(.monotonic))) };
1173 group.token = &gc.node;1173 group.token.store(&gc.node, .monotonic);
11741174
1175 t.run_queue.prepend(&gc.closure.node);1175 t.run_queue.prepend(&gc.closure.node);
11761176
...@@ -1187,7 +1187,7 @@ fn groupWait(userdata: ?*anyopaque, group: *Io.Group, token: *anyopaque) void {...@@ -1187,7 +1187,7 @@ fn groupWait(userdata: ?*anyopaque, group: *Io.Group, token: *anyopaque) void {
1187 const t: *Threaded = @ptrCast(@alignCast(userdata));1187 const t: *Threaded = @ptrCast(@alignCast(userdata));
1188 const gpa = t.allocator;1188 const gpa = t.allocator;
11891189
1190 if (builtin.single_threaded) return;1190 if (builtin.single_threaded) unreachable; // we never set `group.token` to non-`null`
11911191
1192 const group_state: *std.atomic.Value(usize) = @ptrCast(&group.state);1192 const group_state: *std.atomic.Value(usize) = @ptrCast(&group.state);
1193 const event: *Io.Event = @ptrCast(&group.context);1193 const event: *Io.Event = @ptrCast(&group.context);
...@@ -1212,13 +1212,18 @@ fn groupWait(userdata: ?*anyopaque, group: *Io.Group, token: *anyopaque) void {...@@ -1212,13 +1212,18 @@ fn groupWait(userdata: ?*anyopaque, group: *Io.Group, token: *anyopaque) void {
1212 gc.deinit(gpa);1212 gc.deinit(gpa);
1213 node = node_next orelse break;1213 node = node_next orelse break;
1214 }1214 }
1215
1216 // Since the group has now finished, it's illegal to add more tasks to it until we return. It's
1217 // also illegal for us to race with another `await` or `cancel`. Therefore, we must be the only
1218 // thread who can access `group` right now.
1219 group.token.raw = null;
1215}1220}
12161221
1217fn groupCancel(userdata: ?*anyopaque, group: *Io.Group, token: *anyopaque) void {1222fn groupCancel(userdata: ?*anyopaque, group: *Io.Group, token: *anyopaque) void {
1218 const t: *Threaded = @ptrCast(@alignCast(userdata));1223 const t: *Threaded = @ptrCast(@alignCast(userdata));
1219 const gpa = t.allocator;1224 const gpa = t.allocator;
12201225
1221 if (builtin.single_threaded) return;1226 if (builtin.single_threaded) unreachable; // we never set `group.token` to non-`null`
12221227
1223 {1228 {
1224 var node: *std.SinglyLinkedList.Node = @ptrCast(@alignCast(token));1229 var node: *std.SinglyLinkedList.Node = @ptrCast(@alignCast(token));
...@@ -1244,6 +1249,11 @@ fn groupCancel(userdata: ?*anyopaque, group: *Io.Group, token: *anyopaque) void...@@ -1244,6 +1249,11 @@ fn groupCancel(userdata: ?*anyopaque, group: *Io.Group, token: *anyopaque) void
1244 node = node_next orelse break;1249 node = node_next orelse break;
1245 }1250 }
1246 }1251 }
1252
1253 // Since the group has now finished, it's illegal to add more tasks to it until we return. It's
1254 // also illegal for us to race with another `await` or `cancel`. Therefore, we must be the only
1255 // thread who can access `group` right now.
1256 group.token.raw = null;
1247}1257}
12481258
1249fn recancel(userdata: ?*anyopaque) void {1259fn recancel(userdata: ?*anyopaque) void {