authorgravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2025-12-29 17:15:58-08:00
committergravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2025-12-29 22:47:34-08:00
log2adfd4d107f071f91608bef22c7e91b1a9a93470
treeef9da0bd175a2b4ddd12842e44df243606f353bf
parentf862762f091637ad0d17c4735226c0c477c2cb3e

std.Io: fix and improve Group API

Rename `wait` to `await` to be consistent with Future API. The convention here is that this set of functionality goes together: * async/concurrent * await/cancel Also rename Select `wait` to `await` for the same reason. `Group.await` now can return `error.Canceled`. Furthermore, `Group.await` does not auto-propagate cancelation. Instead, users should follow the pattern of `defer group.cancel(io);` after initialization, and doing `try group.await(io);` at the end of the success path. Advanced logic can choose to do something other than this pattern in the event of cancelation. Additionally, fixes a bug in `std.Io.Threaded` future await, in which it swallowed an `error.Canceled`. Now if a task is canceled while awaiting a future, after propagating the cancel request, it also recancels, meaning that the awaiting task will properly detect its own cancelation at the next cancelation point. Furthermore, fixes a bug in the compiler where `error.Canceled` was being swallowed in `dispatchPrelinkWork`. Finally, fixes std.crypto code that inappropriately used `catch unreachable` in response to cancelation without even so much as a comment explaining why it was believed to be unreachable. Now, those functions have `error.Canceled` in the error set and propagate cancelation properly. With this way of doing things, `Group.await` has a nice property: even if all tasks in the group are CPU bound and without cancelation points, the `Group.await` can still be canceled. In such case, the task that was waiting for `await` wakes up with a chance to do some more resource cleanup tasks, such as canceling more things, before entering the deferred `Group.cancel` call at which point it has to suspend until the canceled but uninterruptible CPU bound tasks complete. closes #30601

16 files changed, 108 insertions(+), 78 deletions(-)

lib/compiler/build_runner.zig+5-3
......@@ -748,7 +748,7 @@ fn runStepNames(
748748 defer step_prog.end();
749749
750750 var group: Io.Group = .init;
751 defer group.wait(io);
751 defer group.cancel(io);
752752
753753 // Here we spawn the initial set of tasks with a nice heuristic -
754754 // dependency order. Each worker when it finishes a step will then
......@@ -760,6 +760,8 @@ fn runStepNames(
760760
761761 group.async(io, workerMakeOneStep, .{ &group, b, step, step_prog, run });
762762 }
763
764 try group.await(io);
763765 }
764766
765767 assert(run.memory_blocked_steps.items.len == 0);
......@@ -820,7 +822,7 @@ fn runStepNames(
820822 // * Memory-mapping to share data between the fuzzer and build runner.
821823 // * COFF/PE support added to `std.debug.Info` (it needs a batching API for resolving
822824 // many addresses to source locations).
823 .windows => fatal("--fuzz not yet implemented for {s}", .{@tagName(builtin.os.tag)}),
825 .windows => fatal("--fuzz not yet implemented for {t}", .{builtin.os.tag}),
824826 else => {},
825827 }
826828 if (@bitSizeOf(usize) != 64) {
......@@ -843,7 +845,7 @@ fn runStepNames(
843845 step_stack.keys(),
844846 parent_prog_node,
845847 mode,
846 ) catch |err| fatal("failed to start fuzzer: {s}", .{@errorName(err)});
848 ) catch |err| fatal("failed to start fuzzer: {t}", .{err});
847849 defer f.deinit();
848850
849851 f.start();
lib/std/Build/Fuzz.zig+3-3
......@@ -78,7 +78,7 @@ pub fn init(
7878 all_steps: []const *Build.Step,
7979 root_prog_node: std.Progress.Node,
8080 mode: Mode,
81) Allocator.Error!Fuzz {
81) error{ OutOfMemory, Canceled }!Fuzz {
8282 const run_steps: []const *Step.Run = steps: {
8383 var steps: std.ArrayList(*Step.Run) = .empty;
8484 defer steps.deinit(gpa);
......@@ -98,7 +98,7 @@ pub fn init(
9898 if (steps.items.len == 0) fatal("no fuzz tests found", .{});
9999 rebuild_node.setEstimatedTotalItems(steps.items.len);
100100 const run_steps = try gpa.dupe(*Step.Run, steps.items);
101 rebuild_group.wait(io);
101 try rebuild_group.await(io);
102102 break :steps run_steps;
103103 };
104104 errdefer gpa.free(run_steps);
......@@ -517,7 +517,7 @@ pub fn waitAndPrintReport(fuzz: *Fuzz) void {
517517 assert(fuzz.mode == .limit);
518518 const io = fuzz.io;
519519
520 fuzz.group.wait(io);
520 fuzz.group.awaitUncancelable(io);
521521 fuzz.group = .init;
522522
523523 std.debug.print("======= FUZZING REPORT =======\n", .{});
lib/std/Io.zig+23-12
......@@ -436,7 +436,7 @@ pub fn Poller(comptime StreamEnum: type) type {
436436 // Cancel the pending read into the FIFO.
437437 _ = windows.kernel32.CancelIo(handle);
438438
439 // We have to wait for the handle to be signalled, i.e. for the cancellation to complete.
439 // We have to wait for the handle to be signalled, i.e. for the cancelation to complete.
440440 switch (windows.kernel32.WaitForSingleObject(handle, windows.INFINITE)) {
441441 windows.WAIT_OBJECT_0 => {},
442442 windows.WAIT_FAILED => return windows.unexpectedError(windows.GetLastError()),
......@@ -644,7 +644,7 @@ pub const VTable = struct {
644644 context_alignment: std.mem.Alignment,
645645 start: *const fn (*Group, context: *const anyopaque) void,
646646 ) ConcurrentError!void,
647 groupWait: *const fn (?*anyopaque, *Group, token: *anyopaque) void,
647 groupAwait: *const fn (?*anyopaque, *Group, token: *anyopaque) Cancelable!void,
648648 groupCancel: *const fn (?*anyopaque, *Group, token: *anyopaque) void,
649649
650650 recancel: *const fn (?*anyopaque) void,
......@@ -1023,7 +1023,7 @@ pub fn Future(Result: type) type {
10231023 any_future: ?*AnyFuture,
10241024 result: Result,
10251025
1026 /// Equivalent to `await` but places a cancellation request. This causes the task to receive
1026 /// Equivalent to `await` but places a cancelation request. This causes the task to receive
10271027 /// `error.Canceled` from its next "cancelation point" (if any). A cancelation point is a
10281028 /// call to a function in `Io` which can return `error.Canceled`.
10291029 ///
......@@ -1071,7 +1071,7 @@ pub const Group = struct {
10711071 /// already been called and completed, or it has successfully been assigned
10721072 /// a unit of concurrency.
10731073 ///
1074 /// After this is called, `wait` or `cancel` must be called before the
1074 /// After this is called, `await` or `cancel` must be called before the
10751075 /// group is deinitialized.
10761076 ///
10771077 /// Threadsafe.
......@@ -1092,11 +1092,11 @@ pub const Group = struct {
10921092 }
10931093
10941094 /// Calls `function` with `args`, such that the function is not guaranteed
1095 /// to have returned until `wait` is called, allowing the caller to
1095 /// to have returned until `await` is called, allowing the caller to
10961096 /// progress while waiting for any `Io` operations.
10971097 ///
10981098 /// The resource spawned is owned by the group; after this is called,
1099 /// `wait` or `cancel` must be called before the group is deinitialized.
1099 /// `await` or `cancel` must be called before the group is deinitialized.
11001100 ///
11011101 /// This has stronger guarantee than `async`, placing restrictions on what kind
11021102 /// of `Io` implementations are supported. By calling `async` instead, one
......@@ -1120,20 +1120,31 @@ pub const Group = struct {
11201120 }
11211121
11221122 /// Blocks until all tasks of the group finish. During this time,
1123 /// cancellation requests propagate to all members of the group.
1123 /// cancelation requests propagate to all members of the group.
11241124 ///
11251125 /// Idempotent. Not threadsafe.
11261126 ///
11271127 /// It is safe to call this function concurrently with `Group.async` or
11281128 /// `Group.concurrent`, provided that the group does not complete until
11291129 /// the call to `Group.async` or `Group.concurrent` returns.
1130 pub fn wait(g: *Group, io: Io) void {
1130 pub fn await(g: *Group, io: Io) Cancelable!void {
11311131 const token = g.token.load(.acquire) orelse return;
1132 io.vtable.groupWait(io.userdata, g, token);
1132 try io.vtable.groupAwait(io.userdata, g, token);
11331133 assert(g.token.raw == null);
11341134 }
11351135
1136 /// Equivalent to `wait` but immediately requests cancellation on all
1136 /// Equivalent to `await` but temporarily blocks cancelation while waiting.
1137 pub fn awaitUncancelable(g: *Group, io: Io) void {
1138 const token = g.token.load(.acquire) orelse return;
1139 const prev = swapCancelProtection(io, .blocked);
1140 defer _ = swapCancelProtection(io, prev);
1141 io.vtable.groupAwait(io.userdata, g, token) catch |err| switch (err) {
1142 error.Canceled => unreachable,
1143 };
1144 assert(g.token.raw == null);
1145 }
1146
1147 /// Equivalent to `await` but immediately requests cancelation on all
11371148 /// members of the group.
11381149 ///
11391150 /// For a description of cancelation and cancelation points, see `Future.cancel`.
......@@ -1272,7 +1283,7 @@ pub fn Select(comptime U: type) type {
12721283 /// Asserts there is at least one more `outstanding` task.
12731284 ///
12741285 /// Not threadsafe.
1275 pub fn wait(s: *S) Cancelable!U {
1286 pub fn await(s: *S) Cancelable!U {
12761287 s.outstanding -= 1;
12771288 return s.queue.getOne(s.io) catch |err| switch (err) {
12781289 error.Canceled => |e| return e,
......@@ -1280,7 +1291,7 @@ pub fn Select(comptime U: type) type {
12801291 };
12811292 }
12821293
1283 /// Equivalent to `wait` but requests cancellation on all remaining
1294 /// Equivalent to `wait` but requests cancelation on all remaining
12841295 /// tasks owned by the select.
12851296 ///
12861297 /// For a description of cancelation and cancelation points, see `Future.cancel`.
lib/std/Io/Threaded.zig+12-13
......@@ -795,7 +795,7 @@ pub fn io(t: *Threaded) Io {
795795
796796 .groupAsync = groupAsync,
797797 .groupConcurrent = groupConcurrent,
798 .groupWait = groupWait,
798 .groupAwait = groupAwait,
799799 .groupCancel = groupCancel,
800800
801801 .recancel = recancel,
......@@ -933,7 +933,7 @@ pub fn ioBasic(t: *Threaded) Io {
933933
934934 .groupAsync = groupAsync,
935935 .groupConcurrent = groupConcurrent,
936 .groupWait = groupWait,
936 .groupAwait = groupAwait,
937937 .groupCancel = groupCancel,
938938
939939 .recancel = recancel,
......@@ -1166,6 +1166,7 @@ const AsyncClosure = struct {
11661166 error.Canceled => {
11671167 ac.closure.requestCancel(t);
11681168 ac.event.waitUncancelable(ioBasic(t));
1169 recancel(t);
11691170 },
11701171 };
11711172 @memcpy(result, ac.resultPointer()[0..result.len]);
......@@ -1452,7 +1453,7 @@ fn groupConcurrent(
14521453 t.cond.signal();
14531454}
14541455
1455fn groupWait(userdata: ?*anyopaque, group: *Io.Group, initial_token: *anyopaque) void {
1456fn groupAwait(userdata: ?*anyopaque, group: *Io.Group, initial_token: *anyopaque) Io.Cancelable!void {
14561457 const t: *Threaded = @ptrCast(@alignCast(userdata));
14571458 const gpa = t.allocator;
14581459
......@@ -1464,16 +1465,14 @@ fn groupWait(userdata: ?*anyopaque, group: *Io.Group, initial_token: *anyopaque)
14641465 const event: *Io.Event = @ptrCast(&group.context);
14651466 const prev_state = group_state.fetchAdd(GroupClosure.sync_is_waiting, .acquire);
14661467 assert(prev_state & GroupClosure.sync_is_waiting == 0);
1467 if ((prev_state / GroupClosure.sync_one_pending) > 0) event.wait(ioBasic(t)) catch |err| switch (err) {
1468 error.Canceled => {
1469 var it: ?*std.SinglyLinkedList.Node = @ptrCast(@alignCast(group.token.load(.monotonic)));
1470 while (it) |node| : (it = node.next) {
1471 const gc: *GroupClosure = @fieldParentPtr("node", node);
1472 gc.closure.requestCancel(t);
1473 }
1474 event.waitUncancelable(ioBasic(t));
1475 },
1476 };
1468 {
1469 errdefer _ = group_state.fetchSub(GroupClosure.sync_is_waiting, .monotonic);
1470 // This event.wait can return error.Canceled, in which case this logic does
1471 // *not* propagate cancel requests to each group member. Instead, the user
1472 // code will likely do this with a defered call to groupCancel, or,
1473 // intentionally not do this.
1474 if ((prev_state / GroupClosure.sync_one_pending) > 0) try event.wait(ioBasic(t));
1475 }
14771476
14781477 // Since the group has now finished, it's illegal to add more tasks to it until we return. It's
14791478 // also illegal for us to race with another `await` or `cancel`. Therefore, we must be the only
lib/std/Io/Threaded/test.zig+1-1
......@@ -124,7 +124,7 @@ test "Group.async context alignment" {
124124 var group: std.Io.Group = .init;
125125 var result: ByteArray512 = undefined;
126126 group.async(io, concatByteArraysResultPtr, .{ a, b, &result });
127 group.wait(io);
127 group.awaitUncancelable(io);
128128 try std.testing.expectEqualSlices(u8, &expected.x, &result.x);
129129}
130130
lib/std/Io/net/HostName.zig+1-1
......@@ -289,7 +289,7 @@ pub fn connectMany(
289289 } else |err| switch (err) {
290290 error.Canceled => |e| return e,
291291 error.Closed => {
292 group.wait(io);
292 try group.await(io);
293293 return lookup_future.await(io);
294294 },
295295 }
lib/std/Io/test.zig+2-2
......@@ -194,7 +194,7 @@ test "Group" {
194194 group.async(io, count, .{ 1, 10, &results[0] });
195195 group.async(io, count, .{ 20, 30, &results[1] });
196196
197 group.wait(io);
197 group.awaitUncancelable(io);
198198
199199 try testing.expectEqualSlices(usize, &.{ 45, 245 }, &results);
200200}
......@@ -249,7 +249,7 @@ test "Group concurrent" {
249249 },
250250 };
251251
252 group.wait(io);
252 try group.await(io);
253253
254254 try testing.expectEqualSlices(usize, &.{ 45, 245 }, &results);
255255}
lib/std/crypto.zig+1-1
......@@ -184,7 +184,7 @@ pub const pwhash = struct {
184184
185185 pub const Error = HasherError || error{AllocatorRequired};
186186 pub const HasherError = KdfError || phc_format.Error;
187 pub const KdfError = errors.Error || std.mem.Allocator.Error || std.Thread.SpawnError;
187 pub const KdfError = errors.Error || std.mem.Allocator.Error || std.Thread.SpawnError || std.Io.Cancelable;
188188
189189 pub const argon2 = @import("crypto/argon2.zig");
190190 pub const bcrypt = @import("crypto/bcrypt.zig");
lib/std/crypto/argon2.zig+16-14
......@@ -2,9 +2,9 @@
22// https://github.com/golang/crypto/tree/master/argon2
33// https://github.com/P-H-C/phc-winner-argon2
44
5const std = @import("std");
65const builtin = @import("builtin");
76
7const std = @import("std");
88const blake2 = crypto.hash.blake2;
99const crypto = std.crypto;
1010const Io = std.Io;
......@@ -53,23 +53,24 @@ pub const Mode = enum {
5353pub const Params = struct {
5454 const Self = @This();
5555
56 /// A [t]ime cost, which defines the amount of computation realized and therefore the execution
56 /// Time cost, which defines the amount of computation realized and therefore the execution
5757 /// time, given in number of iterations.
5858 t: u32,
5959
60 /// A [m]emory cost, which defines the memory usage, given in kibibytes.
60 /// Memory cost, which defines the memory usage, given in kibibytes.
6161 m: u32,
6262
63 /// A [p]arallelism degree, which defines the number of parallel threads.
63 /// Parallelism degree, which defines the number of independent tasks,
64 /// to be multiplexed onto threads when possible.
6465 p: u24,
6566
66 /// The [secret] parameter, which is used for keyed hashing. This allows a secret key to be input
67 /// The secret parameter, which is used for keyed hashing. This allows a secret key to be input
6768 /// at hashing time (from some external location) and be folded into the value of the hash. This
6869 /// means that even if your salts and hashes are compromised, an attacker cannot brute-force to
6970 /// find the password without the key.
7071 secret: ?[]const u8 = null,
7172
72 /// The [ad] parameter, which is used to fold any additional data into the hash value. Functionally,
73 /// The ad parameter, which is used to fold any additional data into the hash value. Functionally,
7374 /// this behaves almost exactly like the secret or salt parameters; the ad parameter is folding
7475 /// into the value of the hash. However, this parameter is used for different data. The salt
7576 /// should be a random string stored alongside your password. The secret should be a random key
......@@ -209,18 +210,18 @@ fn processBlocks(
209210 threads: u24,
210211 mode: Mode,
211212 io: Io,
212) void {
213) Io.Cancelable!void {
213214 const lanes = memory / threads;
214215 const segments = lanes / sync_points;
215216
216217 if (builtin.single_threaded or threads == 1) {
217 processBlocksSt(blocks, time, memory, threads, mode, lanes, segments);
218 processBlocksSync(blocks, time, memory, threads, mode, lanes, segments);
218219 } else {
219 processBlocksMt(blocks, time, memory, threads, mode, lanes, segments, io);
220 try processBlocksAsync(blocks, time, memory, threads, mode, lanes, segments, io);
220221 }
221222}
222223
223fn processBlocksSt(
224fn processBlocksSync(
224225 blocks: *Blocks,
225226 time: u32,
226227 memory: u32,
......@@ -241,7 +242,7 @@ fn processBlocksSt(
241242 }
242243}
243244
244fn processBlocksMt(
245fn processBlocksAsync(
245246 blocks: *Blocks,
246247 time: u32,
247248 memory: u32,
......@@ -250,19 +251,20 @@ fn processBlocksMt(
250251 lanes: u32,
251252 segments: u32,
252253 io: Io,
253) void {
254) Io.Cancelable!void {
254255 var n: u32 = 0;
255256 while (n < time) : (n += 1) {
256257 var slice: u32 = 0;
257258 while (slice < sync_points) : (slice += 1) {
258259 var group: Io.Group = .init;
260 defer group.cancel(io);
259261 var lane: u24 = 0;
260262 while (lane < threads) : (lane += 1) {
261263 group.async(io, processSegment, .{
262264 blocks, time, memory, threads, mode, lanes, segments, n, slice, lane,
263265 });
264266 }
265 group.wait(io);
267 try group.await(io);
266268 }
267269 }
268270}
......@@ -503,7 +505,7 @@ pub fn kdf(
503505 blocks.appendNTimesAssumeCapacity(@splat(0), memory);
504506
505507 initBlocks(&blocks, &h0, memory, params.p);
506 processBlocks(&blocks, params.t, memory, params.p, mode, io);
508 try processBlocks(&blocks, params.t, memory, params.p, mode, io);
507509 finalize(&blocks, memory, params.p, derived_key);
508510}
509511
lib/std/crypto/blake3.zig+10-6
......@@ -1,9 +1,11 @@
1const std = @import("std");
21const builtin = @import("builtin");
2
3const std = @import("std");
34const fmt = std.fmt;
45const mem = std.mem;
56const Io = std.Io;
67const Thread = std.Thread;
8const Allocator = std.mem.Allocator;
79
810const Vec4 = @Vector(4, u32);
911const Vec8 = @Vector(8, u32);
......@@ -767,7 +769,7 @@ fn buildMerkleTreeLayerParallel(
767769 key: [8]u32,
768770 flags: Flags,
769771 io: Io,
770) void {
772) Io.Cancelable!void {
771773 const num_parents = input_cvs.len / 2;
772774
773775 // Process sequentially with SIMD for smaller tree layers to avoid thread overhead
......@@ -787,6 +789,7 @@ fn buildMerkleTreeLayerParallel(
787789 const num_workers = Thread.getCpuCount() catch 1;
788790 const parents_per_worker = (num_parents + num_workers - 1) / num_workers;
789791 var group: Io.Group = .init;
792 defer group.cancel(io);
790793
791794 for (0..num_workers) |worker_id| {
792795 const start_idx = worker_id * parents_per_worker;
......@@ -801,7 +804,7 @@ fn buildMerkleTreeLayerParallel(
801804 .flags = flags,
802805 }});
803806 }
804 group.wait(io);
807 try group.await(io);
805808}
806809
807810fn parentOutput(parent_block: []const u8, key: [8]u32, flags: Flags) Output {
......@@ -987,7 +990,7 @@ pub const Blake3 = struct {
987990 d.final(out);
988991 }
989992
990 pub fn hashParallel(b: []const u8, out: []u8, options: Options, allocator: std.mem.Allocator, io: Io) !void {
993 pub fn hashParallel(b: []const u8, out: []u8, options: Options, allocator: Allocator, io: Io) error{ OutOfMemory, Canceled }!void {
991994 if (b.len < parallel_threshold) {
992995 return hash(b, out, options);
993996 }
......@@ -1008,6 +1011,7 @@ pub const Blake3 = struct {
10081011 const num_workers = thread_count;
10091012 const chunks_per_worker = (num_full_chunks + num_workers - 1) / num_workers;
10101013 var group: Io.Group = .init;
1014 defer group.cancel(io);
10111015
10121016 for (0..num_workers) |worker_id| {
10131017 const start_chunk = worker_id * chunks_per_worker;
......@@ -1022,7 +1026,7 @@ pub const Blake3 = struct {
10221026 .flags = flags,
10231027 }});
10241028 }
1025 group.wait(io);
1029 try group.await(io);
10261030
10271031 // Build Merkle tree in parallel layers using ping-pong buffers
10281032 const max_intermediate_size = (num_full_chunks + 1) / 2;
......@@ -1040,7 +1044,7 @@ pub const Blake3 = struct {
10401044 const has_odd = current_level.len % 2 == 1;
10411045 const next_level_size = num_parents + @intFromBool(has_odd);
10421046
1043 buildMerkleTreeLayerParallel(
1047 try buildMerkleTreeLayerParallel(
10441048 current_level[0 .. num_parents * 2],
10451049 next_level_buf[0..num_parents],
10461050 key_words,
lib/std/crypto/kangarootwelve.zig+9-7
......@@ -1,9 +1,10 @@
1const std = @import("std");
21const builtin = @import("builtin");
2
3const std = @import("std");
34const crypto = std.crypto;
45const Allocator = std.mem.Allocator;
56const Io = std.Io;
6const Thread = std.Thread;
7const assert = std.debug.assert;
78
89const TurboSHAKE128State = crypto.hash.sha3.TurboShake128(0x06);
910const TurboSHAKE256State = crypto.hash.sha3.TurboShake256(0x06);
......@@ -598,7 +599,7 @@ inline fn processNLeaves(
598599 output: []align(@alignOf(u64)) u8,
599600) void {
600601 const cv_size = Variant.cv_size;
601 comptime std.debug.assert(cv_size % @sizeOf(u64) == 0);
602 comptime assert(cv_size % @sizeOf(u64) == 0);
602603
603604 if (view.tryGetSlice(j, j + N * chunk_size)) |leaf_data| {
604605 var leaf_cvs: [N * cv_size]u8 = undefined;
......@@ -645,7 +646,7 @@ fn processLeafBatch(comptime Variant: type, ctx: LeafBatchContext) void {
645646 j += chunk_len;
646647 }
647648
648 std.debug.assert(cvs_offset == ctx.output_cvs.len);
649 assert(cvs_offset == ctx.output_cvs.len);
649650}
650651
651652/// Helper to process N leaves in SIMD and absorb CVs into state
......@@ -841,7 +842,7 @@ fn ktMultiThreaded(
841842 total_len: usize,
842843 output: []u8,
843844) !void {
844 comptime std.debug.assert(bytes_per_batch % (optimal_vector_len * chunk_size) == 0);
845 comptime assert(bytes_per_batch % (optimal_vector_len * chunk_size) == 0);
845846
846847 const cv_size = Variant.cv_size;
847848 const StateType = Variant.StateType;
......@@ -883,6 +884,7 @@ fn ktMultiThreaded(
883884 var pending_cv_lens: [256]usize = .{0} ** 256;
884885
885886 var select: Select = .init(io, select_buf);
887 defer select.cancel();
886888 var batches_spawned: usize = 0;
887889 var next_to_process: usize = 0;
888890
......@@ -901,7 +903,7 @@ fn ktMultiThreaded(
901903 batches_spawned += 1;
902904 }
903905
904 const result = select.wait() catch unreachable;
906 const result = try select.await();
905907 const batch = result.batch;
906908 const slot = batch.batch_idx % max_concurrent;
907909
......@@ -925,7 +927,7 @@ fn ktMultiThreaded(
925927 }
926928 }
927929
928 select.group.wait(io);
930 assert(select.outstanding == 0);
929931 }
930932
931933 if (has_partial_leaf) {
src/Compilation.zig+12-6
......@@ -4698,7 +4698,7 @@ fn performAllTheWork(
46984698 });
46994699 }
47004700
4701 astgen_group.wait(io);
4701 try astgen_group.await(io);
47024702 }
47034703
47044704 if (comp.zcu) |zcu| {
......@@ -4761,7 +4761,7 @@ fn performAllTheWork(
47614761 // Since we're skipping analysis, there are no ZCU link tasks.
47624762 comp.link_queue.finishZcuQueue(comp);
47634763 // Let other compilation work finish to collect as many errors as possible.
4764 misc_group.wait(io);
4764 try misc_group.await(io);
47654765 comp.link_queue.wait(io);
47664766 return;
47674767 }
......@@ -4850,18 +4850,22 @@ fn performAllTheWork(
48504850 comp.link_queue.finishZcuQueue(comp);
48514851
48524852 // Main thread work is all done, now just wait for all async work.
4853 misc_group.wait(io);
4853 try misc_group.await(io);
48544854 comp.link_queue.wait(io);
48554855}
48564856
48574857fn dispatchPrelinkWork(comp: *Compilation, main_progress_node: std.Progress.Node) void {
48584858 const io = comp.io;
48594859
4860 // TODO should this function be cancelable?
4861 const prev_cancel_prot = io.swapCancelProtection(.blocked);
4862 defer _ = io.swapCancelProtection(prev_cancel_prot);
4863
48604864 var prelink_group: Io.Group = .init;
48614865 defer prelink_group.cancel(io);
48624866
48634867 comp.queuePrelinkTasks(comp.oneshot_prelink_tasks.items) catch |err| switch (err) {
4864 error.Canceled => return,
4868 error.Canceled => unreachable, // see swapCancelProtection above
48654869 };
48664870 comp.oneshot_prelink_tasks.clearRetainingCapacity();
48674871
......@@ -5055,9 +5059,11 @@ fn dispatchPrelinkWork(comp: *Compilation, main_progress_node: std.Progress.Node
50555059 });
50565060 }
50575061
5058 prelink_group.wait(io);
5062 prelink_group.await(io) catch |err| switch (err) {
5063 error.Canceled => unreachable, // see swapCancelProtection above
5064 };
50595065 comp.link_queue.finishPrelinkQueue(comp) catch |err| switch (err) {
5060 error.Canceled => return,
5066 error.Canceled => unreachable, // see swapCancelProtection above
50615067 };
50625068}
50635069
src/Package/Fetch.zig+10-6
......@@ -146,6 +146,8 @@ pub const JobQueue = struct {
146146 pub const UnlazySet = std.AutoArrayHashMapUnmanaged(Package.Hash, void);
147147
148148 pub fn deinit(jq: *JobQueue) void {
149 const io = jq.io;
150 jq.group.cancel(io);
149151 if (jq.all_fetches.items.len == 0) return;
150152 const gpa = jq.all_fetches.items[0].arena.child_allocator;
151153 jq.table.deinit(gpa);
......@@ -847,7 +849,7 @@ pub fn workerRun(f: *Fetch, prog_name: []const u8) void {
847849
848850 run(f) catch |err| switch (err) {
849851 error.OutOfMemory => f.oom_flag = true,
850 error.Canceled => {},
852 error.Canceled => {}, // TODO make groupAsync functions be cancelable and assert proper value was returned
851853 error.FetchFailed => {
852854 // Nothing to do because the errors are already reported in `error_bundle`,
853855 // and a reference is kept to the `Fetch` task inside `all_fetches`.
......@@ -1517,12 +1519,12 @@ fn computeHash(f: *Fetch, pkg_path: Cache.Path, filter: Filter) RunError!Compute
15171519 // The final hash will be a hash of each file hashed independently. This
15181520 // allows hashing in parallel.
15191521 var group: Io.Group = .init;
1520 defer group.wait(io);
1522 defer group.cancel(io);
15211523
15221524 while (walker.next(io) catch |err| {
15231525 try eb.addRootErrorMessage(.{ .msg = try eb.printString(
1524 "unable to walk temporary directory '{f}': {s}",
1525 .{ pkg_path, @errorName(err) },
1526 "unable to walk temporary directory '{f}': {t}",
1527 .{ pkg_path, err },
15261528 ) });
15271529 return error.FetchFailed;
15281530 }) |entry| {
......@@ -1552,8 +1554,8 @@ fn computeHash(f: *Fetch, pkg_path: Cache.Path, filter: Filter) RunError!Compute
15521554 .file => .file,
15531555 .sym_link => .link,
15541556 else => return f.fail(f.location_tok, try eb.printString(
1555 "package contains '{s}' which has illegal file type '{s}'",
1556 .{ entry.path, @tagName(entry.kind) },
1557 "package contains '{s}' which has illegal file type '{t}'",
1558 .{ entry.path, entry.kind },
15571559 )),
15581560 };
15591561
......@@ -1573,6 +1575,8 @@ fn computeHash(f: *Fetch, pkg_path: Cache.Path, filter: Filter) RunError!Compute
15731575 group.async(io, workerHashFile, .{ io, root_dir, hashed_file });
15741576 try all_files.append(hashed_file);
15751577 }
1578
1579 try group.await(io);
15761580 }
15771581
15781582 {
src/link/MachO/hasher.zig+1-1
......@@ -48,7 +48,7 @@ pub fn ParallelHasher(comptime Hasher: type) type {
4848 });
4949 }
5050
51 group.wait(io);
51 try group.await(io);
5252 }
5353 for (results) |result| _ = try result;
5454 }
src/main.zig+1-1
......@@ -5284,7 +5284,7 @@ fn cmdBuild(gpa: Allocator, arena: Allocator, io: Io, args: []const []const u8)
52845284 );
52855285
52865286 job_queue.group.async(io, Package.Fetch.workerRun, .{ &fetch, "root" });
5287 job_queue.group.wait(io);
5287 try job_queue.group.await(io);
52885288
52895289 try job_queue.consolidateErrors();
52905290
tools/update_cpu_features.zig+1-1
......@@ -1951,7 +1951,7 @@ pub fn main() anyerror!void {
19511951 } });
19521952 }
19531953
1954 group.wait(io);
1954 try group.await(io);
19551955}
19561956
19571957const Job = struct {