authorgravatar for mlugg@mlugg.co.ukMatthew Lugg <mlugg@mlugg.co.uk> 2025-12-31 01:10:07+00:00
committergravatar for mlugg@mlugg.co.ukMatthew Lugg <mlugg@mlugg.co.uk> 2026-01-03 15:45:11+00:00
logf306a9f84a006a6429f485a6c99ac26723f1e1e4
treea8d87cdaa3e38b4a6f2549a0a42d3d4268c804a8
parentb8a09bcbd955c2d97216ffef23246054aa3fe756
signaturelock-open Commit is signed but in an unrecognized format.

std: rebase fixups and cancelation changes

This commit includes some API changes which I agreed with Andrew as a follow-up to the recent `Io.Group` changes: * `Io.Group.await` *does* propagate cancelation to group tasks; it then waits for them to complete, and *also* returns `error.Canceled`. The assertion that group tasks handle `error.Canceled` "correctly" means this behavior is loosely analagous to how awaiting a future works. The important thing is that the semantics of `Group.await` and `Future.await` are similar, and `error.Canceled` will always be visible to the caller (assuming correct API usage). * `Io.Group.awaitUncancelable` is removed. * `Future.await` calls `recancel` only if the "child" task (the future being awaited) did not acknowledge cancelation. If it did, then it is assumed that the future will propagate `error.Canceled` through `await` as needed.

6 files changed, 86 insertions(+), 39 deletions(-)

lib/compiler/build_runner.zig+1-1
...@@ -849,7 +849,7 @@ fn runStepNames(...@@ -849,7 +849,7 @@ fn runStepNames(
849 defer f.deinit();849 defer f.deinit();
850850
851 f.start();851 f.start();
852 f.waitAndPrintReport();852 try f.waitAndPrintReport();
853 }853 }
854854
855 // Every test has a state855 // Every test has a state
lib/std/Build/Fuzz.zig+2-2
...@@ -513,11 +513,11 @@ fn addEntryPoint(fuzz: *Fuzz, coverage_id: u64, addr: u64) error{ AlreadyReporte...@@ -513,11 +513,11 @@ fn addEntryPoint(fuzz: *Fuzz, coverage_id: u64, addr: u64) error{ AlreadyReporte
513 try coverage_map.entry_points.append(fuzz.gpa, @intCast(index));513 try coverage_map.entry_points.append(fuzz.gpa, @intCast(index));
514}514}
515515
516pub fn waitAndPrintReport(fuzz: *Fuzz) void {516pub fn waitAndPrintReport(fuzz: *Fuzz) Io.Cancelable!void {
517 assert(fuzz.mode == .limit);517 assert(fuzz.mode == .limit);
518 const io = fuzz.io;518 const io = fuzz.io;
519519
520 fuzz.group.awaitUncancelable(io);520 try fuzz.group.await(io);
521 fuzz.group = .init;521 fuzz.group = .init;
522522
523 std.debug.print("======= FUZZING REPORT =======\n", .{});523 std.debug.print("======= FUZZING REPORT =======\n", .{});
lib/std/Io.zig+3-12
...@@ -1111,7 +1111,9 @@ pub const Group = struct {...@@ -1111,7 +1111,9 @@ pub const Group = struct {
1111 }1111 }
11121112
1113 /// Blocks until all tasks of the group finish. During this time,1113 /// Blocks until all tasks of the group finish. During this time,
1114 /// cancelation requests propagate to all members of the group.1114 /// cancelation requests propagate to all members of the group, and
1115 /// will also cause `error.Canceled` to be returned when the group
1116 /// does ultimately finish.
1115 ///1117 ///
1116 /// Idempotent. Not threadsafe.1118 /// Idempotent. Not threadsafe.
1117 ///1119 ///
...@@ -1124,17 +1126,6 @@ pub const Group = struct {...@@ -1124,17 +1126,6 @@ pub const Group = struct {
1124 assert(g.token.raw == null);1126 assert(g.token.raw == null);
1125 }1127 }
11261128
1127 /// Equivalent to `await` but temporarily blocks cancelation while waiting.
1128 pub fn awaitUncancelable(g: *Group, io: Io) void {
1129 const token = g.token.load(.acquire) orelse return;
1130 const prev = swapCancelProtection(io, .blocked);
1131 defer _ = swapCancelProtection(io, prev);
1132 io.vtable.groupAwait(io.userdata, g, token) catch |err| switch (err) {
1133 error.Canceled => unreachable,
1134 };
1135 assert(g.token.raw == null);
1136 }
1137
1138 /// Equivalent to `await` but immediately requests cancelation on all1129 /// Equivalent to `await` but immediately requests cancelation on all
1139 /// members of the group.1130 /// members of the group.
1140 ///1131 ///
lib/std/Io/Threaded.zig+76-20
...@@ -182,7 +182,7 @@ const Group = struct {...@@ -182,7 +182,7 @@ const Group = struct {
182 const Task = struct {182 const Task = struct {
183 runnable: Runnable,183 runnable: Runnable,
184 group: *Io.Group,184 group: *Io.Group,
185 func: *const fn (context: *const anyopaque) void,185 func: *const fn (context: *const anyopaque) Io.Cancelable!void,
186 context_alignment: Alignment,186 context_alignment: Alignment,
187 alloc_len: usize,187 alloc_len: usize,
188188
...@@ -192,7 +192,7 @@ const Group = struct {...@@ -192,7 +192,7 @@ const Group = struct {
192 group: Group,192 group: Group,
193 context: []const u8,193 context: []const u8,
194 context_alignment: Alignment,194 context_alignment: Alignment,
195 func: *const fn (context: *const anyopaque) void,195 func: *const fn (context: *const anyopaque) Io.Cancelable!void,
196 ) Allocator.Error!*Task {196 ) Allocator.Error!*Task {
197 const max_context_misalignment = context_alignment.toByteUnits() -| @alignOf(Task);197 const max_context_misalignment = context_alignment.toByteUnits() -| @alignOf(Task);
198 const worst_case_context_offset = context_alignment.forward(@sizeOf(Task) + max_context_misalignment);198 const worst_case_context_offset = context_alignment.forward(@sizeOf(Task) + max_context_misalignment);
...@@ -247,7 +247,20 @@ const Group = struct {...@@ -247,7 +247,20 @@ const Group = struct {
247 }, .monotonic);247 }, .monotonic);
248 }248 }
249249
250 assertGroupResult(task.func(task.contextPointer()));250 const result = task.func(task.contextPointer());
251 const cancel_acknowledged = switch (thread.status.load(.monotonic).cancelation) {
252 .none, .canceling => false,
253 .canceled => true,
254 .parked => unreachable,
255 .blocked => unreachable,
256 .blocked_windows_dns => unreachable,
257 .blocked_canceling => unreachable,
258 };
259 if (result) {
260 assert(!cancel_acknowledged); // group task acknowledged cancelation but did not return `error.Canceled`
261 } else |err| switch (err) {
262 error.Canceled => assert(cancel_acknowledged), // group task returned `error.Canceled` but was never canceled
263 }
251264
252 thread.status.store(.{ .cancelation = .none, .awaitable = .null }, .monotonic);265 thread.status.store(.{ .cancelation = .none, .awaitable = .null }, .monotonic);
253 const old_status = group.status().fetchSub(.{266 const old_status = group.status().fetchSub(.{
...@@ -348,7 +361,8 @@ const Future = struct {...@@ -348,7 +361,8 @@ const Future = struct {
348 pending_awaited = 0b01,361 pending_awaited = 0b01,
349 /// Like `pending`, but the future is being canceled. `Future.awaiter` is populated.362 /// Like `pending`, but the future is being canceled. `Future.awaiter` is populated.
350 pending_canceled = 0b11,363 pending_canceled = 0b11,
351 /// The future has already completed. `thread` is `null`.364 /// The future has already completed. `thread` is `.null`, unless the future terminated
365 /// with an acknowledged cancel request, in which case `thread` is `.all_ones`.
352 done = 0b10,366 done = 0b10,
353 },367 },
354 /// When the future begins execution, this is atomically updated from `null` to the thread running the368 /// When the future begins execution, this is atomically updated from `null` to the thread running the
...@@ -437,10 +451,18 @@ const Future = struct {...@@ -437,10 +451,18 @@ const Future = struct {
437451
438 future.func(future.contextPointer(), future.resultPointer());452 future.func(future.contextPointer(), future.resultPointer());
439453
454 const had_acknowledged_cancel = switch (thread.status.load(.monotonic).cancelation) {
455 .none, .canceling => false,
456 .canceled => true,
457 .parked => unreachable,
458 .blocked => unreachable,
459 .blocked_windows_dns => unreachable,
460 .blocked_canceling => unreachable,
461 };
440 thread.status.store(.{ .cancelation = .none, .awaitable = .null }, .monotonic);462 thread.status.store(.{ .cancelation = .none, .awaitable = .null }, .monotonic);
441 const old_status = future.status.swap(.{463 const old_status = future.status.swap(.{
442 .tag = .done,464 .tag = .done,
443 .thread = .null,465 .thread = if (had_acknowledged_cancel) .all_ones else .null,
444 }, .acq_rel); // acquire `future.awaiter`, release results466 }, .acq_rel); // acquire `future.awaiter`, release results
445 switch (old_status.tag) {467 switch (old_status.tag) {
446 .pending => {},468 .pending => {},
...@@ -1712,11 +1734,11 @@ fn groupAsync(...@@ -1712,11 +1734,11 @@ fn groupAsync(
1712 const t: *Threaded = @ptrCast(@alignCast(userdata));1734 const t: *Threaded = @ptrCast(@alignCast(userdata));
1713 const g: Group = .{ .ptr = type_erased };1735 const g: Group = .{ .ptr = type_erased };
17141736
1715 if (builtin.single_threaded) return start(context.ptr) catch unreachable;1737 if (builtin.single_threaded) return groupAsyncEager(start, context.ptr);
17161738
1717 const gpa = t.allocator;1739 const gpa = t.allocator;
1718 const task = Group.Task.create(gpa, g, context, context_alignment, start) catch |err| switch (err) {1740 const task = Group.Task.create(gpa, g, context, context_alignment, start) catch |err| switch (err) {
1719 error.OutOfMemory => return t.assertGroupResult(start(context.ptr)),1741 error.OutOfMemory => return groupAsyncEager(start, context.ptr),
1720 };1742 };
17211743
1722 t.mutex.lock();1744 t.mutex.lock();
...@@ -1726,7 +1748,7 @@ fn groupAsync(...@@ -1726,7 +1748,7 @@ fn groupAsync(
1726 if (busy_count >= @intFromEnum(t.async_limit)) {1748 if (busy_count >= @intFromEnum(t.async_limit)) {
1727 t.mutex.unlock();1749 t.mutex.unlock();
1728 task.destroy(gpa);1750 task.destroy(gpa);
1729 return t.assertGroupResult(start(context.ptr));1751 return groupAsyncEager(start, context.ptr);
1730 }1752 }
17311753
1732 t.busy_count = busy_count + 1;1754 t.busy_count = busy_count + 1;
...@@ -1739,7 +1761,7 @@ fn groupAsync(...@@ -1739,7 +1761,7 @@ fn groupAsync(
1739 t.busy_count = busy_count;1761 t.busy_count = busy_count;
1740 t.mutex.unlock();1762 t.mutex.unlock();
1741 task.destroy(gpa);1763 task.destroy(gpa);
1742 return t.assertGroupResult(start(context.ptr));1764 return groupAsyncEager(start, context.ptr);
1743 };1765 };
1744 thread.detach();1766 thread.detach();
1745 }1767 }
...@@ -1757,23 +1779,45 @@ fn groupAsync(...@@ -1757,23 +1779,45 @@ fn groupAsync(
1757 t.mutex.unlock();1779 t.mutex.unlock();
1758 t.cond.signal();1780 t.cond.signal();
1759}1781}
17601782fn groupAsyncEager(
1761fn assertGroupResult(result: Io.Cancelable!void) void {1783 start: *const fn (context: *const anyopaque) Io.Cancelable!void,
1762 const cancel_acknowledged = if (Thread.current) |thread|1784 context: *const anyopaque,
1763 switch (thread.status.load(.monotonic).cancelation) {1785) void {
1786 const pre_acknowledged = if (Thread.current) |thread| ack: {
1787 break :ack switch (thread.status.load(.monotonic).cancelation) {
1764 .none, .canceling => false,1788 .none, .canceling => false,
1765 .canceled => true,1789 .canceled => true,
1766 .parked => unreachable,1790 .parked => unreachable,
1767 .blocked => unreachable,1791 .blocked => unreachable,
1768 .blocked_windows_dns => unreachable,1792 .blocked_windows_dns => unreachable,
1769 .blocked_canceling => unreachable,1793 .blocked_canceling => unreachable,
1770 }1794 };
1771 else1795 } else false;
1772 false;1796 const result = start(context);
1797 const post_acknowledged = if (Thread.current) |thread| ack: {
1798 break :ack switch (thread.status.load(.monotonic).cancelation) {
1799 .none, .canceling => false,
1800 .canceled => true,
1801 .parked => unreachable,
1802 .blocked => unreachable,
1803 .blocked_windows_dns => unreachable,
1804 .blocked_canceling => unreachable,
1805 };
1806 } else false;
1807
1773 if (result) {1808 if (result) {
1774 assert(!cancel_acknowledged); // group task acknowledged cancelation but did not return `error.Canceled`1809 if (pre_acknowledged) {
1810 assert(post_acknowledged); // group task called `recancel` but was not canceled
1811 } else {
1812 assert(!post_acknowledged); // group task acknowledged cancelation but did not return `error.Canceled`
1813 }
1775 } else |err| switch (err) {1814 } else |err| switch (err) {
1776 error.Canceled => assert(cancel_acknowledged), // group task returned `error.Canceled` but was never canceled1815 // Don't swallow the cancelation: make it visible to the `Group.async` caller.
1816 error.Canceled => {
1817 assert(!pre_acknowledged); // group task called `recancel` but was not canceled
1818 assert(post_acknowledged); // group task returned `error.Canceled` but was never canceled
1819 recancelInner();
1820 },
1777 }1821 }
1778}1822}
17791823
...@@ -1920,6 +1964,9 @@ fn groupCancel(userdata: ?*anyopaque, type_erased: *Io.Group, initial_token: *an...@@ -1920,6 +1964,9 @@ fn groupCancel(userdata: ?*anyopaque, type_erased: *Io.Group, initial_token: *an
1920fn recancel(userdata: ?*anyopaque) void {1964fn recancel(userdata: ?*anyopaque) void {
1921 const t: *Threaded = @ptrCast(@alignCast(userdata));1965 const t: *Threaded = @ptrCast(@alignCast(userdata));
1922 _ = t;1966 _ = t;
1967 recancelInner();
1968}
1969fn recancelInner() void {
1923 const thread = Thread.current.?; // called `recancel` but was not canceled1970 const thread = Thread.current.?; // called `recancel` but was not canceled
1924 switch (thread.status.fetchXor(.{1971 switch (thread.status.fetchXor(.{
1925 .cancelation = @enumFromInt(0b001),1972 .cancelation = @enumFromInt(0b001),
...@@ -1993,7 +2040,16 @@ fn await(...@@ -1993,7 +2040,16 @@ fn await(
1993 future.waitForCancelWithSignaling(t, &num_completed, null);2040 future.waitForCancelWithSignaling(t, &num_completed, null);
1994 },2041 },
1995 }2042 }
1996 recancel(t);2043 // If the future did not acknowledge the cancelation, we need to mark it outstanding
2044 // for us. Because `future.status.tag == .done`, the information about whether there
2045 // was an acknowledged cancelation is encoded in `future.status.thread`.
2046 const final_status = future.status.load(.monotonic);
2047 assert(final_status.tag == .done);
2048 switch (final_status.thread) {
2049 .null => recancelInner(), // cancelation was not acknowledged, so it's ours
2050 .all_ones => {}, // cancelation was acknowledged, so it was this task's job to propagate it
2051 _ => unreachable,
2052 }
1997 },2053 },
1998 },2054 },
1999 .pending_awaited => unreachable, // `await` raced with `await`2055 .pending_awaited => unreachable, // `await` raced with `await`
...@@ -11669,7 +11725,7 @@ fn unlockStderr(userdata: ?*anyopaque) void {...@@ -11669,7 +11725,7 @@ fn unlockStderr(userdata: ?*anyopaque) void {
11669 const t: *Threaded = @ptrCast(@alignCast(userdata));11725 const t: *Threaded = @ptrCast(@alignCast(userdata));
11670 t.stderr_writer.interface.flush() catch |err| switch (err) {11726 t.stderr_writer.interface.flush() catch |err| switch (err) {
11671 error.WriteFailed => switch (t.stderr_writer.err.?) {11727 error.WriteFailed => switch (t.stderr_writer.err.?) {
11672 error.Canceled => recancel(t),11728 error.Canceled => recancelInner(),
11673 else => {},11729 else => {},
11674 },11730 },
11675 };11731 };
lib/std/Io/Threaded/test.zig+1-1
...@@ -124,7 +124,7 @@ test "Group.async context alignment" {...@@ -124,7 +124,7 @@ test "Group.async context alignment" {
124 var group: std.Io.Group = .init;124 var group: std.Io.Group = .init;
125 var result: ByteArray512 = undefined;125 var result: ByteArray512 = undefined;
126 group.async(io, concatByteArraysResultPtr, .{ a, b, &result });126 group.async(io, concatByteArraysResultPtr, .{ a, b, &result });
127 group.awaitUncancelable(io);127 try group.await(io);
128 try std.testing.expectEqualSlices(u8, &expected.x, &result.x);128 try std.testing.expectEqualSlices(u8, &expected.x, &result.x);
129}129}
130130
lib/std/Io/test.zig+3-3
...@@ -194,7 +194,7 @@ test "Group" {...@@ -194,7 +194,7 @@ test "Group" {
194 group.async(io, count, .{ 1, 10, &results[0] });194 group.async(io, count, .{ 1, 10, &results[0] });
195 group.async(io, count, .{ 20, 30, &results[1] });195 group.async(io, count, .{ 20, 30, &results[1] });
196196
197 group.awaitUncancelable(io);197 try group.await(io);
198198
199 try testing.expectEqualSlices(usize, &.{ 45, 245 }, &results);199 try testing.expectEqualSlices(usize, &.{ 45, 245 }, &results);
200}200}
...@@ -544,9 +544,9 @@ test "tasks spawned in group after Group.cancel are canceled" {...@@ -544,9 +544,9 @@ test "tasks spawned in group after Group.cancel are canceled" {
544 group.concurrent(io, blockUntilCanceled, .{io}) catch {};544 group.concurrent(io, blockUntilCanceled, .{io}) catch {};
545 group.async(io, blockUntilCanceled, .{io});545 group.async(io, blockUntilCanceled, .{io});
546 }546 }
547 fn blockUntilCanceled(io: Io) void {547 fn blockUntilCanceled(io: Io) Io.Cancelable!void {
548 while (true) io.sleep(.fromSeconds(100_000), .awake) catch |err| switch (err) {548 while (true) io.sleep(.fromSeconds(100_000), .awake) catch |err| switch (err) {
549 error.Canceled => return,549 error.Canceled => |e| return e,
550 error.UnsupportedClock => @panic("unsupported clock"),550 error.UnsupportedClock => @panic("unsupported clock"),
551 error.Unexpected => @panic("unexpected"),551 error.Unexpected => @panic("unexpected"),
552 };552 };