authorgravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2025-12-29 18:24:40-08:00
committergravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2025-12-29 22:51:06-08:00
log2a02c7a0d59e25bac07a6ed2948a29438fb05527
treee68a1855272d8d64d1934802501eb2d45f0992f3
parent2adfd4d107f071f91608bef22c7e91b1a9a93470

std.Io.Group: async and concurrent support Cancelable results

Now, the return type of functions spawned with `Group.async` and `Group.concurrent` may be anything that coerces to `Io.Cancelable!void`. Before this commit, group tasks were the only exception to the rule "error.Canceled should never be swallowed". Now, there is no exception, and it is enforced with an assertion upon closure completion. Finally, fixes a case of swallowing error.Canceled in the compiler, solving a TODO. There are three ways to handle `error.Canceled`. In order of most common: 1. Propagate it 2. After receiving it, io.recancel() and then don't propagate it 3. Make it unreachable with io.swapCancelProtection()

4 files changed, 61 insertions(+), 24 deletions(-)

lib/std/Io.zig+7-7
......@@ -631,7 +631,7 @@ pub const VTable = struct {
631631 /// Copied and then passed to `start`.
632632 context: []const u8,
633633 context_alignment: std.mem.Alignment,
634 start: *const fn (*Group, context: *const anyopaque) void,
634 start: *const fn (*Group, context: *const anyopaque) Cancelable!void,
635635 ) void,
636636 /// Thread-safe.
637637 groupConcurrent: *const fn (
......@@ -642,7 +642,7 @@ pub const VTable = struct {
642642 /// Copied and then passed to `start`.
643643 context: []const u8,
644644 context_alignment: std.mem.Alignment,
645 start: *const fn (*Group, context: *const anyopaque) void,
645 start: *const fn (*Group, context: *const anyopaque) Cancelable!void,
646646 ) ConcurrentError!void,
647647 groupAwait: *const fn (?*anyopaque, *Group, token: *anyopaque) Cancelable!void,
648648 groupCancel: *const fn (?*anyopaque, *Group, token: *anyopaque) void,
......@@ -1082,10 +1082,10 @@ pub const Group = struct {
10821082 pub fn async(g: *Group, io: Io, function: anytype, args: std.meta.ArgsTuple(@TypeOf(function))) void {
10831083 const Args = @TypeOf(args);
10841084 const TypeErased = struct {
1085 fn start(group: *Group, context: *const anyopaque) void {
1085 fn start(group: *Group, context: *const anyopaque) Cancelable!void {
10861086 _ = group;
10871087 const args_casted: *const Args = @ptrCast(@alignCast(context));
1088 @call(.auto, function, args_casted.*);
1088 return @call(.auto, function, args_casted.*);
10891089 }
10901090 };
10911091 io.vtable.groupAsync(io.userdata, g, @ptrCast(&args), .of(Args), TypeErased.start);
......@@ -1110,10 +1110,10 @@ pub const Group = struct {
11101110 pub fn concurrent(g: *Group, io: Io, function: anytype, args: std.meta.ArgsTuple(@TypeOf(function))) ConcurrentError!void {
11111111 const Args = @TypeOf(args);
11121112 const TypeErased = struct {
1113 fn start(group: *Group, context: *const anyopaque) void {
1113 fn start(group: *Group, context: *const anyopaque) Cancelable!void {
11141114 _ = group;
11151115 const args_casted: *const Args = @ptrCast(@alignCast(context));
1116 @call(.auto, function, args_casted.*);
1116 return @call(.auto, function, args_casted.*);
11171117 }
11181118 };
11191119 return io.vtable.groupConcurrent(io.userdata, g, @ptrCast(&args), .of(Args), TypeErased.start);
......@@ -1265,7 +1265,7 @@ pub fn Select(comptime U: type) type {
12651265 ) void {
12661266 const Args = @TypeOf(args);
12671267 const TypeErased = struct {
1268 fn start(group: *Group, context: *const anyopaque) void {
1268 fn start(group: *Group, context: *const anyopaque) Cancelable!void {
12691269 const args_casted: *const Args = @ptrCast(@alignCast(context));
12701270 const unerased_select: *S = @fieldParentPtr("group", group);
12711271 const elem = @unionInit(U, @tagName(field), @call(.auto, function, args_casted.*));
lib/std/Io/Threaded.zig+25-9
......@@ -1279,7 +1279,7 @@ const GroupClosure = struct {
12791279 group: *Io.Group,
12801280 /// Points to sibling `GroupClosure`. Used for walking the group to cancel all.
12811281 node: std.SinglyLinkedList.Node,
1282 func: *const fn (*Io.Group, context: *anyopaque) void,
1282 func: *const fn (*Io.Group, context: *anyopaque) Io.Cancelable!void,
12831283 context_alignment: Alignment,
12841284 alloc_len: usize,
12851285
......@@ -1292,7 +1292,7 @@ const GroupClosure = struct {
12921292 current_thread.current_closure = closure;
12931293 current_thread.cancel_protection = .unblocked;
12941294
1295 gc.func(group, gc.contextPointer());
1295 assertResult(closure, gc.func(group, gc.contextPointer()));
12961296
12971297 current_thread.current_closure = null;
12981298 current_thread.cancel_protection = undefined;
......@@ -1302,6 +1302,16 @@ const GroupClosure = struct {
13021302 if (prev_state == (sync_one_pending | sync_is_waiting)) event.set(ioBasic(t));
13031303 }
13041304
1305 fn assertResult(closure: *Closure, result: Io.Cancelable!void) void {
1306 if (result) |_| switch (closure.cancel_status.unpack()) {
1307 .none, .requested => {},
1308 .acknowledged => unreachable, // task illegally swallowed error.Canceled
1309 .signal_id => unreachable,
1310 } else |err| switch (err) {
1311 error.Canceled => assert(closure.cancel_status == .acknowledged),
1312 }
1313 }
1314
13051315 fn contextPointer(gc: *GroupClosure) [*]u8 {
13061316 const base: [*]u8 = @ptrCast(gc);
13071317 const context_offset = gc.context_alignment.forward(@intFromPtr(gc) + @sizeOf(GroupClosure)) - @intFromPtr(gc);
......@@ -1314,7 +1324,7 @@ const GroupClosure = struct {
13141324 group: *Io.Group,
13151325 context: []const u8,
13161326 context_alignment: Alignment,
1317 func: *const fn (*Io.Group, context: *const anyopaque) void,
1327 func: *const fn (*Io.Group, context: *const anyopaque) Io.Cancelable!void,
13181328 ) Allocator.Error!*GroupClosure {
13191329 const max_context_misalignment = context_alignment.toByteUnits() -| @alignOf(GroupClosure);
13201330 const worst_case_context_offset = context_alignment.forward(@sizeOf(GroupClosure) + max_context_misalignment);
......@@ -1352,14 +1362,14 @@ fn groupAsync(
13521362 group: *Io.Group,
13531363 context: []const u8,
13541364 context_alignment: Alignment,
1355 start: *const fn (*Io.Group, context: *const anyopaque) void,
1365 start: *const fn (*Io.Group, context: *const anyopaque) Io.Cancelable!void,
13561366) void {
13571367 const t: *Threaded = @ptrCast(@alignCast(userdata));
1358 if (builtin.single_threaded) return start(group, context.ptr);
1368 if (builtin.single_threaded) return start(group, context.ptr) catch unreachable;
13591369
13601370 const gpa = t.allocator;
13611371 const gc = GroupClosure.init(gpa, group, context, context_alignment, start) catch
1362 return start(group, context.ptr);
1372 return t.assertGroupResult(start(group, context.ptr));
13631373
13641374 t.mutex.lock();
13651375
......@@ -1368,7 +1378,7 @@ fn groupAsync(
13681378 if (busy_count >= @intFromEnum(t.async_limit)) {
13691379 t.mutex.unlock();
13701380 gc.deinit(gpa);
1371 return start(group, context.ptr);
1381 return t.assertGroupResult(start(group, context.ptr));
13721382 }
13731383
13741384 t.busy_count = busy_count + 1;
......@@ -1381,7 +1391,7 @@ fn groupAsync(
13811391 t.busy_count = busy_count;
13821392 t.mutex.unlock();
13831393 gc.deinit(gpa);
1384 return start(group, context.ptr);
1394 return t.assertGroupResult(start(group, context.ptr));
13851395 };
13861396 thread.detach();
13871397 }
......@@ -1402,12 +1412,18 @@ fn groupAsync(
14021412 t.cond.signal();
14031413}
14041414
1415fn assertGroupResult(t: *Threaded, result: Io.Cancelable!void) void {
1416 const current_thread: *Thread = .getCurrent(t);
1417 const current_closure = current_thread.current_closure orelse return;
1418 GroupClosure.assertResult(current_closure, result);
1419}
1420
14051421fn groupConcurrent(
14061422 userdata: ?*anyopaque,
14071423 group: *Io.Group,
14081424 context: []const u8,
14091425 context_alignment: Alignment,
1410 start: *const fn (*Io.Group, context: *const anyopaque) void,
1426 start: *const fn (*Io.Group, context: *const anyopaque) Io.Cancelable!void,
14111427) Io.ConcurrentError!void {
14121428 if (builtin.single_threaded) return error.ConcurrencyUnavailable;
14131429
lib/std/Io/test.zig+27-6
......@@ -207,27 +207,48 @@ fn count(a: usize, b: usize, result: *usize) void {
207207 result.* = sum;
208208}
209209
210test "Group cancellation" {
210test "Group cancelation" {
211211 const io = testing.io;
212212
213213 var group: Io.Group = .init;
214 var results: [2]usize = undefined;
214 var results: [4]usize = .{ 0, 0, 0, 0 };
215215
216 // TODO when robust cancelation is available, make the sleep timeouts much
217 // longer so that it causes the unit test to be failed if not canceled.
218 // https://codeberg.org/ziglang/zig/issues/30049
216219 group.async(io, sleep, .{ io, &results[0] });
217220 group.async(io, sleep, .{ io, &results[1] });
221 group.async(io, sleepUncancelable, .{ io, &results[2] });
222 group.async(io, sleepRecancel, .{ io, &results[3] });
218223
219224 group.cancel(io);
220225
221 try testing.expectEqualSlices(usize, &.{ 1, 1 }, &results);
226 try testing.expectEqualSlices(usize, &.{ 1, 1, 1, 1 }, &results);
227}
228
229fn sleep(io: Io, result: *usize) error{Canceled}!void {
230 defer result.* = 1;
231 io.sleep(.fromMilliseconds(1), .awake) catch |err| switch (err) {
232 error.Canceled => |e| return e,
233 else => {},
234 };
222235}
223236
224fn sleep(io: Io, result: *usize) void {
225 // TODO when cancellation race bug is fixed, make this timeout much longer so that
226 // it causes the unit test to be failed if not canceled.
237fn sleepUncancelable(io: Io, result: *usize) void {
238 const old_prot = io.swapCancelProtection(.blocked);
239 defer _ = io.swapCancelProtection(old_prot);
227240 io.sleep(.fromMilliseconds(1), .awake) catch {};
228241 result.* = 1;
229242}
230243
244fn sleepRecancel(io: Io, result: *usize) void {
245 io.sleep(.fromMilliseconds(1), .awake) catch |err| switch (err) {
246 error.Canceled => io.recancel(),
247 else => {},
248 };
249 result.* = 1;
250}
251
231252test "Group concurrent" {
232253 const io = testing.io;
233254
src/Package/Fetch.zig+2-2
......@@ -843,13 +843,13 @@ pub fn relativePathDigest(pkg_root: Cache.Path, cache_root: Cache.Directory) Pac
843843 return .initPath(pkg_root.sub_path, pkg_root.root_dir.eql(cache_root));
844844}
845845
846pub fn workerRun(f: *Fetch, prog_name: []const u8) void {
846pub fn workerRun(f: *Fetch, prog_name: []const u8) Io.Cancelable!void {
847847 const prog_node = f.prog_node.start(prog_name, 0);
848848 defer prog_node.end();
849849
850850 run(f) catch |err| switch (err) {
851851 error.OutOfMemory => f.oom_flag = true,
852 error.Canceled => {}, // TODO make groupAsync functions be cancelable and assert proper value was returned
852 error.Canceled => |e| return e,
853853 error.FetchFailed => {
854854 // Nothing to do because the errors are already reported in `error_bundle`,
855855 // and a reference is kept to the `Fetch` task inside `all_fetches`.