| author | |
| committer | |
| log | 02142a54d22c20b07009780d7c79d90950d3b2c8 |
| tree | 91d0f80adbe515e475d71d0350cb95e07e7246ad |
| parent | 99229ceb55be717aa1995f1a0105bc6b96a910f6 |
While the general guidance remains useful, it is not the case that
error.Canceled will always pass across the Group task function boundary.
Remove the too-aggressive assertions and add unit test coverage.
Closes #30096
Closes #31340
Closes #313586 files changed, 89 insertions(+), 139 deletions(-)
lib/std/Io.zig+16-22| ... | @@ -124,7 +124,7 @@ pub const VTable = struct { | ... | @@ -124,7 +124,7 @@ pub const VTable = struct { |
| 124 | /// Copied and then passed to `start`. | 124 | /// Copied and then passed to `start`. |
| 125 | context: []const u8, | 125 | context: []const u8, |
| 126 | context_alignment: std.mem.Alignment, | 126 | context_alignment: std.mem.Alignment, |
| 127 | start: *const fn (context: *const anyopaque) Cancelable!void, | 127 | start: *const fn (context: *const anyopaque) void, |
| 128 | ) void, | 128 | ) void, |
| 129 | /// Thread-safe. | 129 | /// Thread-safe. |
| 130 | groupConcurrent: *const fn ( | 130 | groupConcurrent: *const fn ( |
| ... | @@ -135,7 +135,7 @@ pub const VTable = struct { | ... | @@ -135,7 +135,7 @@ pub const VTable = struct { |
| 135 | /// Copied and then passed to `start`. | 135 | /// Copied and then passed to `start`. |
| 136 | context: []const u8, | 136 | context: []const u8, |
| 137 | context_alignment: std.mem.Alignment, | 137 | context_alignment: std.mem.Alignment, |
| 138 | start: *const fn (context: *const anyopaque) Cancelable!void, | 138 | start: *const fn (context: *const anyopaque) void, |
| 139 | ) ConcurrentError!void, | 139 | ) ConcurrentError!void, |
| 140 | groupAwait: *const fn (?*anyopaque, *Group, token: *anyopaque) Cancelable!void, | 140 | groupAwait: *const fn (?*anyopaque, *Group, token: *anyopaque) Cancelable!void, |
| 141 | groupCancel: *const fn (?*anyopaque, *Group, token: *anyopaque) void, | 141 | groupCancel: *const fn (?*anyopaque, *Group, token: *anyopaque) void, |
| ... | @@ -1169,19 +1169,18 @@ pub const Group = struct { | ... | @@ -1169,19 +1169,18 @@ pub const Group = struct { |
| 1169 | /// instead of becoming associated with a `Future`. | 1169 | /// instead of becoming associated with a `Future`. |
| 1170 | /// | 1170 | /// |
| 1171 | /// The return type of `function` must be coercible to `Cancelable!void`. | 1171 | /// The return type of `function` must be coercible to `Cancelable!void`. |
| 1172 | /// `function` returning `error.Canceled` does nothing because it is an | ||
| 1173 | /// cancelation propagation boundary. | ||
| 1172 | /// | 1174 | /// |
| 1173 | /// Once this function is called, there are resources associated with the | 1175 | /// Once this function is called, there are resources associated with the |
| 1174 | /// group. To release those resources, `Group.await` or `Group.cancel` must | 1176 | /// group. To release those resources, `Group.await` or `Group.cancel` must |
| 1175 | /// eventually be called. | 1177 | /// eventually be called. |
| 1176 | /// | ||
| 1177 | /// If `error.Canceled` is returned from any operation this task performs, | ||
| 1178 | /// it is asserted that `function` returns `error.Canceled`. | ||
| 1179 | pub fn async(g: *Group, io: Io, function: anytype, args: std.meta.ArgsTuple(@TypeOf(function))) void { | 1178 | pub fn async(g: *Group, io: Io, function: anytype, args: std.meta.ArgsTuple(@TypeOf(function))) void { |
| 1180 | const Args = @TypeOf(args); | 1179 | const Args = @TypeOf(args); |
| 1181 | const TypeErased = struct { | 1180 | const TypeErased = struct { |
| 1182 | fn start(context: *const anyopaque) Cancelable!void { | 1181 | fn start(context: *const anyopaque) void { |
| 1183 | const args_casted: *const Args = @ptrCast(@alignCast(context)); | 1182 | const args_casted: *const Args = @ptrCast(@alignCast(context)); |
| 1184 | return @call(.auto, function, args_casted.*); | 1183 | _ = @as(Cancelable!void, @call(.auto, function, args_casted.*)) catch {}; |
| 1185 | } | 1184 | } |
| 1186 | }; | 1185 | }; |
| 1187 | io.vtable.groupAsync(io.userdata, g, @ptrCast(&args), .of(Args), TypeErased.start); | 1186 | io.vtable.groupAsync(io.userdata, g, @ptrCast(&args), .of(Args), TypeErased.start); |
| ... | @@ -1191,19 +1190,18 @@ pub const Group = struct { | ... | @@ -1191,19 +1190,18 @@ pub const Group = struct { |
| 1191 | /// `Group` instead of becoming associated with a `Future`. | 1190 | /// `Group` instead of becoming associated with a `Future`. |
| 1192 | /// | 1191 | /// |
| 1193 | /// The return type of `function` must be coercible to `Cancelable!void`. | 1192 | /// The return type of `function` must be coercible to `Cancelable!void`. |
| 1193 | /// `function` returning `error.Canceled` does nothing because it is an | ||
| 1194 | /// cancelation propagation boundary. | ||
| 1194 | /// | 1195 | /// |
| 1195 | /// Once this function is called, there are resources associated with the | 1196 | /// Once this function is called, there are resources associated with the |
| 1196 | /// group. To release those resources, `Group.await` or `Group.cancel` must | 1197 | /// group. To release those resources, `Group.await` or `Group.cancel` must |
| 1197 | /// eventually be called. | 1198 | /// eventually be called. |
| 1198 | /// | ||
| 1199 | /// If `error.Canceled` is returned from any operation this task performs, | ||
| 1200 | /// it is asserted that `function` returns `error.Canceled`. | ||
| 1201 | pub fn concurrent(g: *Group, io: Io, function: anytype, args: std.meta.ArgsTuple(@TypeOf(function))) ConcurrentError!void { | 1199 | pub fn concurrent(g: *Group, io: Io, function: anytype, args: std.meta.ArgsTuple(@TypeOf(function))) ConcurrentError!void { |
| 1202 | const Args = @TypeOf(args); | 1200 | const Args = @TypeOf(args); |
| 1203 | const TypeErased = struct { | 1201 | const TypeErased = struct { |
| 1204 | fn start(context: *const anyopaque) Cancelable!void { | 1202 | fn start(context: *const anyopaque) void { |
| 1205 | const args_casted: *const Args = @ptrCast(@alignCast(context)); | 1203 | const args_casted: *const Args = @ptrCast(@alignCast(context)); |
| 1206 | return @call(.auto, function, args_casted.*); | 1204 | _ = @as(Cancelable!void, @call(.auto, function, args_casted.*)) catch {}; |
| 1207 | } | 1205 | } |
| 1208 | }; | 1206 | }; |
| 1209 | return io.vtable.groupConcurrent(io.userdata, g, @ptrCast(&args), .of(Args), TypeErased.start); | 1207 | return io.vtable.groupConcurrent(io.userdata, g, @ptrCast(&args), .of(Args), TypeErased.start); |
| ... | @@ -1352,15 +1350,13 @@ pub fn Select(comptime U: type) type { | ... | @@ -1352,15 +1350,13 @@ pub fn Select(comptime U: type) type { |
| 1352 | const Context = struct { | 1350 | const Context = struct { |
| 1353 | select: *S, | 1351 | select: *S, |
| 1354 | args: @TypeOf(args), | 1352 | args: @TypeOf(args), |
| 1355 | fn start(type_erased_context: *const anyopaque) Cancelable!void { | 1353 | fn start(type_erased_context: *const anyopaque) void { |
| 1356 | const context: *const @This() = @ptrCast(@alignCast(type_erased_context)); | 1354 | const context: *const @This() = @ptrCast(@alignCast(type_erased_context)); |
| 1357 | const raw_result = @call(.auto, function, context.args); | 1355 | const result = @call(.auto, function, context.args); |
| 1358 | const elem = @unionInit(U, @tagName(field), raw_result); | 1356 | const elem = @unionInit(U, @tagName(field), result); |
| 1359 | context.select.queue.putOneUncancelable(context.select.io, elem) catch |err| switch (err) { | 1357 | context.select.queue.putOneUncancelable(context.select.io, elem) catch |err| switch (err) { |
| 1360 | error.Closed => {}, | 1358 | error.Closed => {}, |
| 1361 | }; | 1359 | }; |
| 1362 | if (@typeInfo(@TypeOf(raw_result)) == .error_union) | ||
| 1363 | _ = raw_result catch |err| if (err == error.Canceled) return error.Canceled; | ||
| 1364 | } | 1360 | } |
| 1365 | }; | 1361 | }; |
| 1366 | const context: Context = .{ .select = s, .args = args }; | 1362 | const context: Context = .{ .select = s, .args = args }; |
| ... | @@ -1391,15 +1387,13 @@ pub fn Select(comptime U: type) type { | ... | @@ -1391,15 +1387,13 @@ pub fn Select(comptime U: type) type { |
| 1391 | const Context = struct { | 1387 | const Context = struct { |
| 1392 | select: *S, | 1388 | select: *S, |
| 1393 | args: @TypeOf(args), | 1389 | args: @TypeOf(args), |
| 1394 | fn start(type_erased_context: *const anyopaque) Cancelable!void { | 1390 | fn start(type_erased_context: *const anyopaque) void { |
| 1395 | const context: *const @This() = @ptrCast(@alignCast(type_erased_context)); | 1391 | const context: *const @This() = @ptrCast(@alignCast(type_erased_context)); |
| 1396 | const raw_result = @call(.auto, function, context.args); | 1392 | const result = @call(.auto, function, context.args); |
| 1397 | const elem = @unionInit(U, @tagName(field), raw_result); | 1393 | const elem = @unionInit(U, @tagName(field), result); |
| 1398 | context.select.queue.putOneUncancelable(context.select.io, elem) catch |err| switch (err) { | 1394 | context.select.queue.putOneUncancelable(context.select.io, elem) catch |err| switch (err) { |
| 1399 | error.Closed => {}, | 1395 | error.Closed => {}, |
| 1400 | }; | 1396 | }; |
| 1401 | if (@typeInfo(@TypeOf(raw_result)) == .error_union) | ||
| 1402 | _ = raw_result catch |err| if (err == error.Canceled) return error.Canceled; | ||
| 1403 | } | 1397 | } |
| 1404 | }; | 1398 | }; |
| 1405 | const context: Context = .{ .select = s, .args = args }; | 1399 | const context: Context = .{ .select = s, .args = args }; |
lib/std/Io/Dispatch.zig+5-26| ... | @@ -1331,7 +1331,7 @@ const Group = struct { | ... | @@ -1331,7 +1331,7 @@ const Group = struct { |
| 1331 | evented: *Evented, | 1331 | evented: *Evented, |
| 1332 | group: Group, | 1332 | group: Group, |
| 1333 | fiber: *Fiber, | 1333 | fiber: *Fiber, |
| 1334 | start: *const fn (context: *const anyopaque) Io.Cancelable!void, | 1334 | start: *const fn (context: *const anyopaque) void, |
| 1335 | 1335 | ||
| 1336 | fn fromFiber(fiber: *Fiber) *Group.AsyncClosure { | 1336 | fn fromFiber(fiber: *Fiber) *Group.AsyncClosure { |
| 1337 | return @ptrFromInt(Fiber.max_context_align.max(.of(Group.AsyncClosure)).backward( | 1337 | return @ptrFromInt(Fiber.max_context_align.max(.of(Group.AsyncClosure)).backward( |
| ... | @@ -1370,11 +1370,7 @@ const Group = struct { | ... | @@ -1370,11 +1370,7 @@ const Group = struct { |
| 1370 | const ev = closure.evented; | 1370 | const ev = closure.evented; |
| 1371 | const fiber = closure.fiber; | 1371 | const fiber = closure.fiber; |
| 1372 | message.handle(ev); | 1372 | message.handle(ev); |
| 1373 | if (closure.start(closure.contextPointer())) { | 1373 | closure.start(closure.contextPointer()); |
| 1374 | assert(!fiber.cancel_protection.acknowledged); // group task acknowledged cancelation but did not return `error.Canceled` | ||
| 1375 | } else |err| switch (err) { | ||
| 1376 | error.Canceled => assert(fiber.cancel_protection.acknowledged), // group task returned `error.Canceled` but was never canceled | ||
| 1377 | } | ||
| 1378 | if (closure.group.removeFiber(ev, fiber)) |awaiter| ev.queue.async(awaiter, &Fiber.@"resume"); | 1374 | if (closure.group.removeFiber(ev, fiber)) |awaiter| ev.queue.async(awaiter, &Fiber.@"resume"); |
| 1379 | ev.yield(.destroy); | 1375 | ev.yield(.destroy); |
| 1380 | unreachable; // switched to dead fiber | 1376 | unreachable; // switched to dead fiber |
| ... | @@ -1387,28 +1383,11 @@ fn groupAsync( | ... | @@ -1387,28 +1383,11 @@ fn groupAsync( |
| 1387 | type_erased: *Io.Group, | 1383 | type_erased: *Io.Group, |
| 1388 | context: []const u8, | 1384 | context: []const u8, |
| 1389 | context_alignment: Alignment, | 1385 | context_alignment: Alignment, |
| 1390 | start: *const fn (context: *const anyopaque) Io.Cancelable!void, | 1386 | start: *const fn (context: *const anyopaque) void, |
| 1391 | ) void { | 1387 | ) void { |
| 1392 | const ev: *Evented = @ptrCast(@alignCast(userdata)); | 1388 | const ev: *Evented = @ptrCast(@alignCast(userdata)); |
| 1393 | return groupConcurrent(ev, type_erased, context, context_alignment, start) catch { | 1389 | return groupConcurrent(ev, type_erased, context, context_alignment, start) catch { |
| 1394 | const fiber = Thread.current().currentFiber(); | 1390 | start(context.ptr); |
| 1395 | const pre_acknowledged = fiber.cancel_protection.acknowledged; | ||
| 1396 | const result = start(context.ptr); | ||
| 1397 | const post_acknowledged = fiber.cancel_protection.acknowledged; | ||
| 1398 | if (result) { | ||
| 1399 | if (pre_acknowledged) { | ||
| 1400 | assert(post_acknowledged); // group task called `recancel` but was not canceled | ||
| 1401 | } else { | ||
| 1402 | assert(!post_acknowledged); // group task acknowledged cancelation but did not return `error.Canceled` | ||
| 1403 | } | ||
| 1404 | } else |err| switch (err) { | ||
| 1405 | // Don't swallow the cancelation: make it visible to the `Group.async` caller. | ||
| 1406 | error.Canceled => { | ||
| 1407 | assert(!pre_acknowledged); // group task called `recancel` but was not canceled | ||
| 1408 | assert(post_acknowledged); // group task returned `error.Canceled` but was never canceled | ||
| 1409 | fiber.cancel_protection.recancel(); | ||
| 1410 | }, | ||
| 1411 | } | ||
| 1412 | }; | 1391 | }; |
| 1413 | } | 1392 | } |
| 1414 | 1393 | ||
| ... | @@ -1417,7 +1396,7 @@ fn groupConcurrent( | ... | @@ -1417,7 +1396,7 @@ fn groupConcurrent( |
| 1417 | type_erased: *Io.Group, | 1396 | type_erased: *Io.Group, |
| 1418 | context: []const u8, | 1397 | context: []const u8, |
| 1419 | context_alignment: Alignment, | 1398 | context_alignment: Alignment, |
| 1420 | start: *const fn (context: *const anyopaque) Io.Cancelable!void, | 1399 | start: *const fn (context: *const anyopaque) void, |
| 1421 | ) Io.ConcurrentError!void { | 1400 | ) Io.ConcurrentError!void { |
| 1422 | assert(context_alignment.compare(.lte, Fiber.max_context_align)); // TODO | 1401 | assert(context_alignment.compare(.lte, Fiber.max_context_align)); // TODO |
| 1423 | assert(context.len <= Fiber.max_context_size); // TODO | 1402 | assert(context.len <= Fiber.max_context_size); // TODO |
lib/std/Io/Kqueue.zig+2-2| ... | @@ -766,7 +766,7 @@ fn groupAsync( | ... | @@ -766,7 +766,7 @@ fn groupAsync( |
| 766 | type_erased: *Io.Group, | 766 | type_erased: *Io.Group, |
| 767 | context: []const u8, | 767 | context: []const u8, |
| 768 | context_alignment: Alignment, | 768 | context_alignment: Alignment, |
| 769 | start: *const fn (context: *const anyopaque) Io.Cancelable!void, | 769 | start: *const fn (context: *const anyopaque) void, |
| 770 | ) void { | 770 | ) void { |
| 771 | const k: *Kqueue = @ptrCast(@alignCast(userdata)); | 771 | const k: *Kqueue = @ptrCast(@alignCast(userdata)); |
| 772 | _ = k; | 772 | _ = k; |
| ... | @@ -782,7 +782,7 @@ fn groupConcurrent( | ... | @@ -782,7 +782,7 @@ fn groupConcurrent( |
| 782 | type_erased: *Io.Group, | 782 | type_erased: *Io.Group, |
| 783 | context: []const u8, | 783 | context: []const u8, |
| 784 | context_alignment: Alignment, | 784 | context_alignment: Alignment, |
| 785 | start: *const fn (context: *const anyopaque) Io.Cancelable!void, | 785 | start: *const fn (context: *const anyopaque) void, |
| 786 | ) Io.ConcurrentError!void { | 786 | ) Io.ConcurrentError!void { |
| 787 | const k: *Kqueue = @ptrCast(@alignCast(userdata)); | 787 | const k: *Kqueue = @ptrCast(@alignCast(userdata)); |
| 788 | _ = k; | 788 | _ = k; |
lib/std/Io/Threaded.zig+7-58| ... | @@ -412,7 +412,7 @@ const Group = struct { | ... | @@ -412,7 +412,7 @@ const Group = struct { |
| 412 | const Task = struct { | 412 | const Task = struct { |
| 413 | runnable: Runnable, | 413 | runnable: Runnable, |
| 414 | group: *Io.Group, | 414 | group: *Io.Group, |
| 415 | func: *const fn (context: *const anyopaque) Io.Cancelable!void, | 415 | func: *const fn (context: *const anyopaque) void, |
| 416 | context_alignment: Alignment, | 416 | context_alignment: Alignment, |
| 417 | alloc_len: usize, | 417 | alloc_len: usize, |
| 418 | 418 | ||
| ... | @@ -422,7 +422,7 @@ const Group = struct { | ... | @@ -422,7 +422,7 @@ const Group = struct { |
| 422 | group: Group, | 422 | group: Group, |
| 423 | context: []const u8, | 423 | context: []const u8, |
| 424 | context_alignment: Alignment, | 424 | context_alignment: Alignment, |
| 425 | func: *const fn (context: *const anyopaque) Io.Cancelable!void, | 425 | func: *const fn (context: *const anyopaque) void, |
| 426 | ) Allocator.Error!*Task { | 426 | ) Allocator.Error!*Task { |
| 427 | const max_context_misalignment = context_alignment.toByteUnits() -| @alignOf(Task); | 427 | const max_context_misalignment = context_alignment.toByteUnits() -| @alignOf(Task); |
| 428 | const worst_case_context_offset = context_alignment.forward(@sizeOf(Task) + max_context_misalignment); | 428 | const worst_case_context_offset = context_alignment.forward(@sizeOf(Task) + max_context_misalignment); |
| ... | @@ -477,21 +477,7 @@ const Group = struct { | ... | @@ -477,21 +477,7 @@ const Group = struct { |
| 477 | }, .monotonic); | 477 | }, .monotonic); |
| 478 | } | 478 | } |
| 479 | 479 | ||
| 480 | const result = task.func(task.contextPointer()); | 480 | task.func(task.contextPointer()); |
| 481 | const cancel_acknowledged = switch (thread.status.load(.monotonic).cancelation) { | ||
| 482 | .none, .canceling => false, | ||
| 483 | .canceled => true, | ||
| 484 | .parked => unreachable, | ||
| 485 | .blocked => unreachable, | ||
| 486 | .blocked_alertable => unreachable, | ||
| 487 | .blocked_alertable_canceling => unreachable, | ||
| 488 | .blocked_canceling => unreachable, | ||
| 489 | }; | ||
| 490 | if (result) { | ||
| 491 | assert(!cancel_acknowledged); // group task acknowledged cancelation but did not return `error.Canceled` | ||
| 492 | } else |err| switch (err) { | ||
| 493 | error.Canceled => assert(cancel_acknowledged), // group task returned `error.Canceled` but was never canceled | ||
| 494 | } | ||
| 495 | 481 | ||
| 496 | thread.status.store(.{ .cancelation = .none, .awaitable = .null }, .monotonic); | 482 | thread.status.store(.{ .cancelation = .none, .awaitable = .null }, .monotonic); |
| 497 | const old_status = group.status().fetchSub(.{ | 483 | const old_status = group.status().fetchSub(.{ |
| ... | @@ -2272,7 +2258,7 @@ fn groupAsync( | ... | @@ -2272,7 +2258,7 @@ fn groupAsync( |
| 2272 | type_erased: *Io.Group, | 2258 | type_erased: *Io.Group, |
| 2273 | context: []const u8, | 2259 | context: []const u8, |
| 2274 | context_alignment: Alignment, | 2260 | context_alignment: Alignment, |
| 2275 | start: *const fn (context: *const anyopaque) Io.Cancelable!void, | 2261 | start: *const fn (context: *const anyopaque) void, |
| 2276 | ) void { | 2262 | ) void { |
| 2277 | const t: *Threaded = @ptrCast(@alignCast(userdata)); | 2263 | const t: *Threaded = @ptrCast(@alignCast(userdata)); |
| 2278 | const g: Group = .{ .ptr = type_erased }; | 2264 | const g: Group = .{ .ptr = type_erased }; |
| ... | @@ -2323,47 +2309,10 @@ fn groupAsync( | ... | @@ -2323,47 +2309,10 @@ fn groupAsync( |
| 2323 | condSignal(&t.cond); | 2309 | condSignal(&t.cond); |
| 2324 | } | 2310 | } |
| 2325 | fn groupAsyncEager( | 2311 | fn groupAsyncEager( |
| 2326 | start: *const fn (context: *const anyopaque) Io.Cancelable!void, | 2312 | start: *const fn (context: *const anyopaque) void, |
| 2327 | context: *const anyopaque, | 2313 | context: *const anyopaque, |
| 2328 | ) void { | 2314 | ) void { |
| 2329 | const pre_acknowledged = if (Thread.current) |thread| ack: { | 2315 | start(context); |
| 2330 | break :ack switch (thread.status.load(.monotonic).cancelation) { | ||
| 2331 | .none, .canceling => false, | ||
| 2332 | .canceled => true, | ||
| 2333 | .parked => unreachable, | ||
| 2334 | .blocked => unreachable, | ||
| 2335 | .blocked_alertable => unreachable, | ||
| 2336 | .blocked_alertable_canceling => unreachable, | ||
| 2337 | .blocked_canceling => unreachable, | ||
| 2338 | }; | ||
| 2339 | } else false; | ||
| 2340 | const result = start(context); | ||
| 2341 | const post_acknowledged = if (Thread.current) |thread| ack: { | ||
| 2342 | break :ack switch (thread.status.load(.monotonic).cancelation) { | ||
| 2343 | .none, .canceling => false, | ||
| 2344 | .canceled => true, | ||
| 2345 | .parked => unreachable, | ||
| 2346 | .blocked => unreachable, | ||
| 2347 | .blocked_alertable => unreachable, | ||
| 2348 | .blocked_alertable_canceling => unreachable, | ||
| 2349 | .blocked_canceling => unreachable, | ||
| 2350 | }; | ||
| 2351 | } else false; | ||
| 2352 | |||
| 2353 | if (result) { | ||
| 2354 | if (pre_acknowledged) { | ||
| 2355 | assert(post_acknowledged); // group task called `recancel` but was not canceled | ||
| 2356 | } else { | ||
| 2357 | assert(!post_acknowledged); // group task acknowledged cancelation but did not return `error.Canceled` | ||
| 2358 | } | ||
| 2359 | } else |err| switch (err) { | ||
| 2360 | // Don't swallow the cancelation: make it visible to the `Group.async` caller. | ||
| 2361 | error.Canceled => { | ||
| 2362 | assert(!pre_acknowledged); // group task called `recancel` but was not canceled | ||
| 2363 | assert(post_acknowledged); // group task returned `error.Canceled` but was never canceled | ||
| 2364 | recancelInner(); | ||
| 2365 | }, | ||
| 2366 | } | ||
| 2367 | } | 2316 | } |
| 2368 | 2317 | ||
| 2369 | fn groupConcurrent( | 2318 | fn groupConcurrent( |
| ... | @@ -2371,7 +2320,7 @@ fn groupConcurrent( | ... | @@ -2371,7 +2320,7 @@ fn groupConcurrent( |
| 2371 | type_erased: *Io.Group, | 2320 | type_erased: *Io.Group, |
| 2372 | context: []const u8, | 2321 | context: []const u8, |
| 2373 | context_alignment: Alignment, | 2322 | context_alignment: Alignment, |
| 2374 | start: *const fn (context: *const anyopaque) Io.Cancelable!void, | 2323 | start: *const fn (context: *const anyopaque) void, |
| 2375 | ) Io.ConcurrentError!void { | 2324 | ) Io.ConcurrentError!void { |
| 2376 | if (builtin.single_threaded) return error.ConcurrencyUnavailable; | 2325 | if (builtin.single_threaded) return error.ConcurrencyUnavailable; |
| 2377 | 2326 |
lib/std/Io/Uring.zig+5-26| ... | @@ -1738,7 +1738,7 @@ const Group = struct { | ... | @@ -1738,7 +1738,7 @@ const Group = struct { |
| 1738 | evented: *Evented, | 1738 | evented: *Evented, |
| 1739 | group: Group, | 1739 | group: Group, |
| 1740 | fiber: *Fiber, | 1740 | fiber: *Fiber, |
| 1741 | start: *const fn (context: *const anyopaque) Io.Cancelable!void, | 1741 | start: *const fn (context: *const anyopaque) void, |
| 1742 | 1742 | ||
| 1743 | fn fromFiber(fiber: *Fiber) *Group.AsyncClosure { | 1743 | fn fromFiber(fiber: *Fiber) *Group.AsyncClosure { |
| 1744 | return @ptrFromInt(Fiber.max_context_align.max(.of(Group.AsyncClosure)).backward( | 1744 | return @ptrFromInt(Fiber.max_context_align.max(.of(Group.AsyncClosure)).backward( |
| ... | @@ -1784,11 +1784,7 @@ const Group = struct { | ... | @@ -1784,11 +1784,7 @@ const Group = struct { |
| 1784 | const fiber = closure.fiber; | 1784 | const fiber = closure.fiber; |
| 1785 | message.handle(ev); | 1785 | message.handle(ev); |
| 1786 | assert(fiber.status.queue_next == null); | 1786 | assert(fiber.status.queue_next == null); |
| 1787 | if (closure.start(closure.contextPointer())) { | 1787 | closure.start(closure.contextPointer()); |
| 1788 | assert(!fiber.cancel_protection.acknowledged); // group task acknowledged cancelation but did not return `error.Canceled` | ||
| 1789 | } else |err| switch (err) { | ||
| 1790 | error.Canceled => assert(fiber.cancel_protection.acknowledged), // group task returned `error.Canceled` but was never canceled | ||
| 1791 | } | ||
| 1792 | ev.yield(closure.group.removeFiber(ev, fiber), .destroy); | 1788 | ev.yield(closure.group.removeFiber(ev, fiber), .destroy); |
| 1793 | unreachable; // switched to dead fiber | 1789 | unreachable; // switched to dead fiber |
| 1794 | } | 1790 | } |
| ... | @@ -1800,28 +1796,11 @@ fn groupAsync( | ... | @@ -1800,28 +1796,11 @@ fn groupAsync( |
| 1800 | type_erased: *Io.Group, | 1796 | type_erased: *Io.Group, |
| 1801 | context: []const u8, | 1797 | context: []const u8, |
| 1802 | context_alignment: Alignment, | 1798 | context_alignment: Alignment, |
| 1803 | start: *const fn (context: *const anyopaque) Io.Cancelable!void, | 1799 | start: *const fn (context: *const anyopaque) void, |
| 1804 | ) void { | 1800 | ) void { |
| 1805 | const ev: *Evented = @ptrCast(@alignCast(userdata)); | 1801 | const ev: *Evented = @ptrCast(@alignCast(userdata)); |
| 1806 | return groupConcurrent(ev, type_erased, context, context_alignment, start) catch { | 1802 | return groupConcurrent(ev, type_erased, context, context_alignment, start) catch { |
| 1807 | const fiber = Thread.current().currentFiber(); | 1803 | start(context.ptr); |
| 1808 | const pre_acknowledged = fiber.cancel_protection.acknowledged; | ||
| 1809 | const result = start(context.ptr); | ||
| 1810 | const post_acknowledged = fiber.cancel_protection.acknowledged; | ||
| 1811 | if (result) { | ||
| 1812 | if (pre_acknowledged) { | ||
| 1813 | assert(post_acknowledged); // group task called `recancel` but was not canceled | ||
| 1814 | } else { | ||
| 1815 | assert(!post_acknowledged); // group task acknowledged cancelation but did not return `error.Canceled` | ||
| 1816 | } | ||
| 1817 | } else |err| switch (err) { | ||
| 1818 | // Don't swallow the cancelation: make it visible to the `Group.async` caller. | ||
| 1819 | error.Canceled => { | ||
| 1820 | assert(!pre_acknowledged); // group task called `recancel` but was not canceled | ||
| 1821 | assert(post_acknowledged); // group task returned `error.Canceled` but was never canceled | ||
| 1822 | fiber.cancel_protection.recancel(); | ||
| 1823 | }, | ||
| 1824 | } | ||
| 1825 | }; | 1804 | }; |
| 1826 | } | 1805 | } |
| 1827 | 1806 | ||
| ... | @@ -1830,7 +1809,7 @@ fn groupConcurrent( | ... | @@ -1830,7 +1809,7 @@ fn groupConcurrent( |
| 1830 | type_erased: *Io.Group, | 1809 | type_erased: *Io.Group, |
| 1831 | context: []const u8, | 1810 | context: []const u8, |
| 1832 | context_alignment: Alignment, | 1811 | context_alignment: Alignment, |
| 1833 | start: *const fn (context: *const anyopaque) Io.Cancelable!void, | 1812 | start: *const fn (context: *const anyopaque) void, |
| 1834 | ) Io.ConcurrentError!void { | 1813 | ) Io.ConcurrentError!void { |
| 1835 | assert(context_alignment.compare(.lte, Fiber.max_context_align)); // TODO | 1814 | assert(context_alignment.compare(.lte, Fiber.max_context_align)); // TODO |
| 1836 | assert(context.len <= Fiber.max_context_size); // TODO | 1815 | assert(context.len <= Fiber.max_context_size); // TODO |
lib/std/Io/test.zig+54-5| ... | @@ -255,8 +255,6 @@ test "Group.cancel" { | ... | @@ -255,8 +255,6 @@ test "Group.cancel" { |
| 255 | } | 255 | } |
| 256 | 256 | ||
| 257 | test "Group.concurrent" { | 257 | test "Group.concurrent" { |
| 258 | if (builtin.os.tag == .linux and !builtin.link_libc) return error.SkipZigTest; // https://codeberg.org/ziglang/zig/issues/30096 | ||
| 259 | |||
| 260 | const io = testing.io; | 258 | const io = testing.io; |
| 261 | 259 | ||
| 262 | var group: Io.Group = .init; | 260 | var group: Io.Group = .init; |
| ... | @@ -265,14 +263,14 @@ test "Group.concurrent" { | ... | @@ -265,14 +263,14 @@ test "Group.concurrent" { |
| 265 | 263 | ||
| 266 | group.concurrent(io, count, .{ 1, 10, &results[0] }) catch |err| switch (err) { | 264 | group.concurrent(io, count, .{ 1, 10, &results[0] }) catch |err| switch (err) { |
| 267 | error.ConcurrencyUnavailable => { | 265 | error.ConcurrencyUnavailable => { |
| 268 | try testing.expect(builtin.single_threaded); | 266 | try expect(builtin.single_threaded); |
| 269 | return; | 267 | return; |
| 270 | }, | 268 | }, |
| 271 | }; | 269 | }; |
| 272 | 270 | ||
| 273 | group.concurrent(io, count, .{ 20, 30, &results[1] }) catch |err| switch (err) { | 271 | group.concurrent(io, count, .{ 20, 30, &results[1] }) catch |err| switch (err) { |
| 274 | error.ConcurrencyUnavailable => { | 272 | error.ConcurrencyUnavailable => { |
| 275 | try testing.expect(builtin.single_threaded); | 273 | try expect(builtin.single_threaded); |
| 276 | return; | 274 | return; |
| 277 | }, | 275 | }, |
| 278 | }; | 276 | }; |
| ... | @@ -282,6 +280,57 @@ test "Group.concurrent" { | ... | @@ -282,6 +280,57 @@ test "Group.concurrent" { |
| 282 | try testing.expectEqualSlices(usize, &.{ 45, 245 }, &results); | 280 | try testing.expectEqualSlices(usize, &.{ 45, 245 }, &results); |
| 283 | } | 281 | } |
| 284 | 282 | ||
| 283 | test "Group materializes error.Cancel" { | ||
| 284 | const S = struct { | ||
| 285 | fn task() Io.Cancelable!void { | ||
| 286 | return error.Canceled; | ||
| 287 | } | ||
| 288 | }; | ||
| 289 | |||
| 290 | const io = testing.io; | ||
| 291 | |||
| 292 | var group: Io.Group = .init; | ||
| 293 | |||
| 294 | group.async(io, S.task, .{}); | ||
| 295 | group.concurrent(io, S.task, .{}) catch |err| switch (err) { | ||
| 296 | error.ConcurrencyUnavailable => { | ||
| 297 | try expect(builtin.single_threaded); | ||
| 298 | return; | ||
| 299 | }, | ||
| 300 | }; | ||
| 301 | |||
| 302 | try group.await(io); | ||
| 303 | } | ||
| 304 | |||
| 305 | test "Group task receives cancelation unknowingly" { | ||
| 306 | const S = struct { | ||
| 307 | io: Io, | ||
| 308 | err: ?Io.Cancelable!void, | ||
| 309 | |||
| 310 | fn task(s: *@This()) void { | ||
| 311 | foo(s); | ||
| 312 | } | ||
| 313 | |||
| 314 | fn foo(s: *@This()) void { | ||
| 315 | s.err = s.io.sleep(.fromSeconds(300), .awake); | ||
| 316 | } | ||
| 317 | }; | ||
| 318 | |||
| 319 | const io = testing.io; | ||
| 320 | |||
| 321 | var group: Io.Group = .init; | ||
| 322 | var result: S = .{ .io = io, .err = null }; | ||
| 323 | group.concurrent(io, S.task, .{&result}) catch |err| switch (err) { | ||
| 324 | error.ConcurrencyUnavailable => { | ||
| 325 | try expect(builtin.single_threaded); | ||
| 326 | return; | ||
| 327 | }, | ||
| 328 | }; | ||
| 329 | group.cancel(io); | ||
| 330 | |||
| 331 | try expectError(error.Canceled, result.err.?); | ||
| 332 | } | ||
| 333 | |||
| 285 | fn testQueue(comptime len: usize) !void { | 334 | fn testQueue(comptime len: usize) !void { |
| 286 | const io = testing.io; | 335 | const io = testing.io; |
| 287 | var buf: [len]usize = undefined; | 336 | var buf: [len]usize = undefined; |
| ... | @@ -541,7 +590,7 @@ test "random" { | ... | @@ -541,7 +590,7 @@ test "random" { |
| 541 | io.random(@ptrCast(&b)); | 590 | io.random(@ptrCast(&b)); |
| 542 | io.random(@ptrCast(&c)); | 591 | io.random(@ptrCast(&c)); |
| 543 | 592 | ||
| 544 | try std.testing.expect(a ^ b ^ c != 0); | 593 | try expect(a ^ b ^ c != 0); |
| 545 | } | 594 | } |
| 546 | 595 | ||
| 547 | test "randomSecure" { | 596 | test "randomSecure" { |