io_v2_tasks.zig (29021B)
1 //! Structured task state machine, owned by one event loop. No user code, host IO, 2 //! allocation during stop/settlement, or v1 scheduler reuse. The driver owns body 3 //! continuations, prepares their storage before admission, and visits every live 4 //! Payload here when registering evaluator GC roots. 5 const std = @import("std"); 6 const R = @import("io_v2_resources.zig"); 7 pub const Phase = enum { open, joining, releasing }; 8 pub const Kind = enum { root, lexical, spawned }; 9 pub const Error = R.Fault || error{ Stopped, StaleTask, JoinCycle, InvalidTransition }; 10 11 pub fn Scheduler(comptime Backend: type, comptime Payload: type) type { 12 return struct { 13 const Self = @This(); 14 pub const Registry = R.Registry(Backend); 15 pub const Diagnostics = struct { 16 items: [16]Payload = undefined, 17 len: usize = 0, 18 omitted: u64 = 0, 19 pub fn append(self: *Diagnostics, other: Diagnostics) void { 20 for (other.items[0..other.len]) |item| { 21 if (self.len < 16) { 22 self.items[self.len] = item; 23 self.len += 1; 24 } else self.omitted += 1; 25 } 26 self.omitted += other.omitted; 27 } 28 pub fn one(value: Payload) Diagnostics { 29 var result: Diagnostics = .{}; 30 result.items[0] = value; 31 result.len = 1; 32 return result; 33 } 34 }; 35 pub const Outcome = struct { 36 kind: enum { completed, failed, cancelled }, 37 value: Payload, 38 cleanup: Diagnostics = .{}, 39 fn primary(self: Outcome) Outcome { 40 return .{ .kind = self.kind, .value = self.value }; 41 } 42 }; 43 pub const Group = struct { 44 parent: ?R.Scope, 45 task: u64, 46 kind: Kind, 47 propagate: bool, 48 phase: Phase = .open, 49 reason: ?Outcome = null, 50 body: ?Outcome = null, 51 cleanup: Diagnostics = .{}, 52 lexical: ?R.Scope = null, 53 wait: ?u64 = null, 54 host_calls: u32 = 0, 55 visited: bool = false, 56 walked: bool = false, 57 }; 58 const Task = struct { 59 owner: R.Scope, 60 supervised: bool, 61 ticket: bool, 62 waiter: ?R.Scope = null, 63 outcome: ?Outcome = null, 64 }; 65 allocator: std.mem.Allocator, 66 registry: *Registry, 67 owner_stopped: Payload, 68 cycle: Payload, 69 groups: std.AutoHashMapUnmanaged(R.Scope, Group) = .empty, 70 tasks: std.AutoHashMapUnmanaged(u64, Task) = .empty, 71 72 pub fn init(allocator: std.mem.Allocator, registry: *Registry, owner_stopped: Payload, cycle: Payload) !Self { 73 var self: Self = .{ .allocator = allocator, .registry = registry, .owner_stopped = owner_stopped, .cycle = cycle }; 74 try self.groups.put(allocator, 0, .{ .parent = null, .task = 0, .kind = .root, .propagate = false }); 75 return self; 76 } 77 pub fn deinit(self: *Self) void { 78 self.tasks.deinit(self.allocator); 79 self.groups.deinit(self.allocator); 80 } 81 pub fn checkpoint(self: *Self, scope: R.Scope) Error!void { 82 const group = self.groups.get(scope) orelse return error.ClosedScope; 83 if (group.reason != null) return error.Stopped; 84 if (group.phase != .open or group.lexical != null or group.wait != null) return error.InvalidTransition; 85 } 86 pub fn nested(self: *Self, parent: R.Scope) Error!R.Scope { 87 try self.checkpoint(parent); 88 const task = self.groups.get(parent).?.task; 89 try self.groups.ensureUnusedCapacity(self.allocator, 1); 90 const child = (try self.registry.reserveChild(parent, &.{})).scope; 91 self.groups.putAssumeCapacityNoClobber(child, .{ .parent = parent, .task = task, .kind = .lexical, .propagate = true }); 92 self.groups.getPtr(parent).?.lexical = child; 93 return child; 94 } 95 pub fn spawn(self: *Self, owner: R.Scope, supervised: bool, keep: bool, tokens: []const R.Token) Error!Registry.Transfer { 96 try self.checkpoint(owner); 97 try self.groups.ensureUnusedCapacity(self.allocator, 1); 98 try self.tasks.ensureUnusedCapacity(self.allocator, 1); 99 const child = try self.registry.reserveChild(owner, tokens); 100 // Everything following resource commitment is allocation-free. 101 self.groups.putAssumeCapacityNoClobber(child.scope, .{ .parent = owner, .task = child.scope, .kind = .spawned, .propagate = !supervised }); 102 self.tasks.putAssumeCapacityNoClobber(child.scope, .{ .owner = owner, .supervised = supervised, .ticket = keep }); 103 return child; 104 } 105 fn recordStop(self: *Self, scope: R.Scope, reason: Outcome) bool { 106 const group = self.groups.getPtr(scope) orelse return false; 107 if (group.reason != null) return false; 108 group.reason = reason.primary(); 109 self.registry.beginCloseScope(scope); 110 return true; 111 } 112 // No allocation or host interruption may prevent requests reaching the 113 // entire active subtree. First recorded reason wins, including while joining. 114 pub fn stop(self: *Self, scope: R.Scope, reason: Outcome) void { 115 std.debug.assert(reason.kind != .completed); 116 if (!self.recordStop(scope, reason)) return; 117 var changed = true; 118 while (changed) { 119 changed = false; 120 var it = self.groups.iterator(); 121 while (it.next()) |entry| { 122 const g = entry.value_ptr.*; 123 if (g.parent) |parent| { 124 if (g.reason) |why| { 125 if (why.kind == .failed and g.propagate) changed = self.recordStop(parent, why) or changed; 126 } 127 if (self.groups.get(parent).?.reason != null) { 128 changed = self.recordStop(entry.key_ptr.*, .{ .kind = .cancelled, .value = self.owner_stopped }) or changed; 129 } 130 } 131 } 132 } 133 } 134 fn checked(self: *Self, current: R.Scope, token: R.Token) Error!*Task { 135 if (token.run != self.registry.identity) return error.StaleTask; 136 const task = self.tasks.getPtr(token.key) orelse return error.StaleTask; 137 if (!self.registry.within(current, task.owner)) return error.OutOfScope; 138 return task; 139 } 140 fn markTask(self: *Self, task: u64) bool { 141 var changed = false; 142 var it = self.groups.valueIterator(); 143 while (it.next()) |g| if (g.task == task and !g.visited) { 144 g.visited = true; 145 changed = true; 146 }; 147 return changed; 148 } 149 // Allocation-free reachability over task identities, not lexical scope 150 // IDs. Active implicit joins contribute edges just like explicit tickets. 151 fn reaches(self: *Self, start: u64, wanted: u64) bool { 152 if (start == wanted) return true; 153 var reset = self.groups.valueIterator(); 154 while (reset.next()) |g| { 155 g.visited = false; 156 g.walked = false; 157 } 158 _ = self.markTask(start); 159 var changed = true; 160 while (changed) { 161 changed = false; 162 var it = self.groups.iterator(); 163 while (it.next()) |entry| { 164 if (!entry.value_ptr.visited or entry.value_ptr.walked) continue; 165 entry.value_ptr.walked = true; 166 const g = entry.value_ptr.*; 167 if (g.wait) |target| { 168 if (target == wanted) return true; 169 changed = self.markTask(target) or changed; 170 } 171 if (g.phase == .joining) { 172 var children = self.tasks.iterator(); 173 while (children.next()) |child| { 174 if (child.value_ptr.owner != entry.key_ptr.* or child.value_ptr.outcome != null) continue; 175 if (child.key_ptr.* == wanted) return true; 176 changed = self.markTask(child.key_ptr.*) or changed; 177 } 178 } 179 } 180 } 181 return false; 182 } 183 pub fn claimJoin(self: *Self, current: R.Scope, token: R.Token) Error!void { 184 try self.checkpoint(current); 185 const task = try self.checked(current, token); 186 if (!task.ticket) return error.StaleTask; 187 const identity = self.groups.get(current).?.task; 188 if (self.registry.within(current, token.key) or self.reaches(token.key, identity)) return error.JoinCycle; 189 // Consume before waiting. Cancellation must not restore this ticket. 190 task.ticket = false; 191 task.waiter = current; 192 self.groups.getPtr(current).?.wait = token.key; 193 } 194 pub fn pollJoin(self: *Self, current: R.Scope) Error!?Outcome { 195 const g = self.groups.get(current) orelse return error.ClosedScope; 196 const target = g.wait orelse return error.InvalidTransition; 197 const task = self.tasks.get(target) orelse return error.StaleTask; 198 const outcome = task.outcome orelse return null; 199 self.groups.getPtr(current).?.wait = null; 200 _ = self.tasks.remove(target); 201 return outcome; 202 } 203 pub fn abandonJoin(self: *Self, current: R.Scope) void { 204 const group = self.groups.getPtr(current) orelse return; 205 const target = group.wait orelse return; 206 group.wait = null; 207 const task = self.tasks.getPtr(target).?; 208 task.waiter = null; 209 if (task.outcome != null) _ = self.tasks.remove(target); 210 } 211 pub fn discard(self: *Self, current: R.Scope, token: R.Token) Error!void { 212 try self.checkpoint(current); 213 const task = try self.checked(current, token); 214 if (!task.ticket) return error.StaleTask; 215 task.ticket = false; 216 if (task.outcome != null) _ = self.tasks.remove(token.key); 217 } 218 pub fn cancel(self: *Self, current: R.Scope, token: R.Token, reason: Payload) Error!void { 219 try self.checkpoint(current); 220 const task = try self.checked(current, token); 221 if (task.outcome != null) return error.StaleTask; 222 self.stop(token.key, .{ .kind = .cancelled, .value = reason }); 223 } 224 pub fn bodyDone(self: *Self, scope: R.Scope, outcome: Outcome) Error!void { 225 const g = self.groups.get(scope) orelse return error.ClosedScope; 226 if (g.phase != .open or g.lexical != null) return error.InvalidTransition; 227 self.abandonJoin(scope); 228 if (outcome.kind != .completed) self.stop(scope, outcome); 229 self.groups.getPtr(scope).?.body = outcome; 230 self.registry.beginCloseScope(scope); 231 // Detect before adding implicit edges. A cycle is terminal; explicit 232 // join cycles instead leave the ticket untouched and return an error. 233 var it = self.tasks.iterator(); 234 while (it.next()) |entry| { 235 if (entry.value_ptr.owner == scope and entry.value_ptr.outcome == null and self.reaches(entry.key_ptr.*, g.task)) { 236 self.stop(scope, .{ .kind = .failed, .value = self.cycle }); 237 break; 238 } 239 } 240 self.groups.getPtr(scope).?.phase = .joining; 241 } 242 // The task owns host calls even when a newly acquired resource belongs to 243 // an ancestor. Cancellation never detaches an unfinished callback. 244 pub fn beginHost(self: *Self, scope: R.Scope) Error!void { 245 try self.checkpoint(scope); 246 const group = self.groups.getPtr(scope).?; 247 group.host_calls = std.math.add(u32, group.host_calls, 1) catch return error.OutOfMemory; 248 } 249 pub fn endHost(self: *Self, scope: R.Scope) Error!void { 250 const group = self.groups.getPtr(scope) orelse return error.ClosedScope; 251 if (group.host_calls == 0) return error.InvalidTransition; 252 group.host_calls -= 1; 253 } 254 pub fn beginRelease(self: *Self, scope: R.Scope) Error!bool { 255 const g = self.groups.get(scope) orelse return error.ClosedScope; 256 if (g.phase != .joining) return error.InvalidTransition; 257 var children = self.tasks.valueIterator(); 258 while (children.next()) |task| if (task.owner == scope and task.outcome == null) return false; 259 if (g.host_calls != 0 or !self.registry.readyForRelease(scope)) return false; 260 self.groups.getPtr(scope).?.phase = .releasing; 261 return true; 262 } 263 pub fn settle(self: *Self, scope: R.Scope, own_cleanup: Diagnostics) Error!Outcome { 264 const g = self.groups.get(scope) orelse return error.ClosedScope; 265 if (g.phase != .releasing) return error.InvalidTransition; 266 try self.registry.finishScope(scope); 267 var final = (g.reason orelse g.body.?).primary(); 268 var errors = g.body.?.cleanup; 269 errors.append(g.cleanup); 270 errors.append(own_cleanup); 271 if (final.kind == .completed and errors.len != 0) { 272 final = .{ .kind = .failed, .value = errors.items[0] }; 273 std.mem.copyForwards(Payload, errors.items[0 .. errors.len - 1], errors.items[1..errors.len]); 274 errors.len -= 1; 275 } 276 final.cleanup = errors; 277 if (final.kind == .failed) self.stop(scope, final); 278 // Drop all unclaimed tickets owned by this scope, including completed 279 // results. No iteration pointer survives a removal/restart. 280 while (true) { 281 var owned: ?u64 = null; 282 var it = self.tasks.iterator(); 283 while (it.next()) |entry| if (entry.value_ptr.owner == scope) { 284 owned = entry.key_ptr.*; 285 break; 286 }; 287 if (owned) |key| { 288 _ = self.tasks.remove(key); 289 } else break; 290 } 291 if (g.parent) |parent| { 292 if (g.kind == .lexical) { 293 self.groups.getPtr(parent).?.cleanup.append(final.cleanup); 294 self.groups.getPtr(parent).?.lexical = null; 295 } else { 296 const task = self.tasks.getPtr(scope).?; 297 if (!task.supervised) self.groups.getPtr(parent).?.cleanup.append(final.cleanup); 298 task.outcome = final; 299 if (!task.ticket and task.waiter == null) _ = self.tasks.remove(scope); 300 } 301 } 302 _ = self.groups.remove(scope); 303 return if (g.kind == .lexical) final.primary() else final; 304 } 305 // Embeddings must root these plus their body/continuation/host-reply roots. 306 pub fn visitPayloads(self: *const Self, context: anytype, visit: anytype) !void { 307 try visit(context, self.owner_stopped); 308 try visit(context, self.cycle); 309 var groups = self.groups.valueIterator(); 310 while (groups.next()) |g| { 311 if (g.reason) |reason| try visit(context, reason.value); 312 if (g.body) |body| { 313 try visit(context, body.value); 314 for (body.cleanup.items[0..body.cleanup.len]) |v| try visit(context, v); 315 } 316 for (g.cleanup.items[0..g.cleanup.len]) |v| try visit(context, v); 317 } 318 var tasks = self.tasks.valueIterator(); 319 while (tasks.next()) |task| if (task.outcome) |outcome| { 320 try visit(context, outcome.value); 321 for (outcome.cleanup.items[0..outcome.cleanup.len]) |v| try visit(context, v); 322 }; 323 } 324 }; 325 } 326 327 const testing = std.testing; 328 const S = Scheduler(u32, u32); 329 fn taskToken(s: *S, id: u64) R.Token { 330 return .{ .run = s.registry.identity, .key = id }; 331 } 332 fn finish(s: *S, scope: R.Scope, body: S.Outcome, cleanup: S.Diagnostics) !S.Outcome { 333 try s.bodyDone(scope, body); 334 try testing.expect(try s.beginRelease(scope)); 335 return release(s, scope, cleanup); 336 } 337 fn release(s: *S, scope: R.Scope, cleanup: S.Diagnostics) !S.Outcome { 338 while (s.registry.newestOwned(scope)) |resource| { 339 _ = (try s.registry.claimClose(scope, resource, true)).?; 340 try s.registry.finishClose(resource); 341 } 342 return s.settle(scope, cleanup); 343 } 344 fn ok(value: u32) S.Outcome { 345 return .{ .kind = .completed, .value = value }; 346 } 347 fn fail(value: u32) S.Outcome { 348 return .{ .kind = .failed, .value = value }; 349 } 350 fn cancelled(value: u32) S.Outcome { 351 return .{ .kind = .cancelled, .value = value }; 352 } 353 354 test "normal exit closes admission, joins, remains cancellable and retains no task history" { 355 var registry = try S.Registry.init(testing.allocator, 1, .{ 0, 0, 0 }); 356 defer registry.deinit(); 357 var s = try S.init(testing.allocator, ®istry, 90, 91); 358 defer s.deinit(); 359 const child = try s.spawn(0, false, false, &.{}); 360 try s.bodyDone(0, ok(4)); 361 try testing.expect(!try s.beginRelease(0)); 362 try testing.expectError(error.ClosedScope, registry.beginAcquire(0)); 363 s.stop(0, cancelled(77)); 364 try testing.expectEqual(@as(u32, 90), s.groups.get(child.scope).?.reason.?.value); 365 _ = try finish(&s, child.scope, ok(5), .{}); 366 try testing.expect(try s.beginRelease(0)); 367 const outcome = try release(&s, 0, .{}); 368 try testing.expectEqual(.cancelled, outcome.kind); 369 try testing.expectEqual(@as(u32, 77), outcome.value); 370 try testing.expectEqual(@as(u32, 0), s.tasks.count()); 371 try testing.expectEqual(@as(u32, 0), s.groups.count()); 372 } 373 test "unsupervised first failure stops siblings before release; supervision contains failure" { 374 var registry = try S.Registry.init(testing.allocator, 1, .{ 0, 0, 0 }); 375 defer registry.deinit(); 376 var s = try S.init(testing.allocator, ®istry, 90, 91); 377 defer s.deinit(); 378 const contained = (try s.spawn(0, true, true, &.{})).scope; 379 _ = try finish(&s, contained, fail(5), .{}); 380 try testing.expectEqual(@as(?S.Outcome, null), s.groups.get(0).?.reason); 381 const first = (try s.spawn(0, false, true, &.{})).scope; 382 const second = (try s.spawn(0, false, true, &.{})).scope; 383 s.stop(first, fail(7)); 384 s.stop(second, fail(8)); 385 try testing.expectEqual(@as(u32, 7), s.groups.get(0).?.reason.?.value); 386 try testing.expectEqual(.cancelled, s.groups.get(second).?.reason.?.kind); 387 try testing.expectError(error.ClosedScope, registry.beginAcquire(second)); 388 _ = try finish(&s, first, ok(0), .{}); 389 _ = try finish(&s, second, ok(0), .{}); 390 const outcome = try finish(&s, 0, ok(0), .{}); 391 try testing.expectEqual(@as(u32, 7), outcome.value); 392 try testing.expectEqual(@as(u32, 0), s.tasks.count()); 393 } 394 test "consumable join tickets, cancellation and abandoned waits" { 395 var registry = try S.Registry.init(testing.allocator, 1, .{ 0, 0, 0 }); 396 defer registry.deinit(); 397 var s = try S.init(testing.allocator, ®istry, 90, 91); 398 defer s.deinit(); 399 const child = (try s.spawn(0, true, true, &.{})).scope; 400 const other = (try s.spawn(0, true, true, &.{})).scope; 401 try s.claimJoin(other, taskToken(&s, child)); 402 try testing.expectEqual(@as(?S.Outcome, null), try s.pollJoin(other)); 403 try testing.expectError(error.StaleTask, s.discard(0, taskToken(&s, child))); 404 try s.cancel(0, taskToken(&s, other), 13); 405 _ = try finish(&s, other, ok(0), .{}); 406 try testing.expectEqual(@as(?R.Scope, null), s.tasks.get(child).?.waiter); 407 _ = try finish(&s, child, ok(123), .{}); 408 try testing.expect(!s.tasks.contains(child)); 409 try s.claimJoin(0, taskToken(&s, other)); 410 const result = (try s.pollJoin(0)).?; 411 try testing.expectEqual(.cancelled, result.kind); 412 try testing.expectEqual(@as(u32, 13), result.value); 413 try testing.expectError(error.StaleTask, s.claimJoin(0, taskToken(&s, other))); 414 _ = try finish(&s, 0, ok(0), .{}); 415 } 416 test "explicit and implicit join cycles include lexical task identities" { 417 var registry = try S.Registry.init(testing.allocator, 1, .{ 0, 0, 0 }); 418 defer registry.deinit(); 419 var s = try S.init(testing.allocator, ®istry, 90, 91); 420 defer s.deinit(); 421 const a = (try s.spawn(0, true, true, &.{})).scope; 422 var b = (try s.spawn(0, true, true, &.{})).scope; 423 const nested = try s.nested(a); 424 try testing.expectError(error.JoinCycle, s.claimJoin(nested, taskToken(&s, a))); 425 try s.claimJoin(nested, taskToken(&s, b)); 426 try testing.expectError(error.JoinCycle, s.claimJoin(b, taskToken(&s, a))); 427 try testing.expect(s.tasks.get(a).?.ticket); 428 s.abandonJoin(nested); 429 _ = try finish(&s, b, ok(0), .{}); 430 // The abandoned ticket was consumed, so use a fresh task for this cycle. 431 b = (try s.spawn(0, true, true, &.{})).scope; 432 // b waits for a; a then implicitly joins a nested-scope child waiting on b. 433 const c = (try s.spawn(nested, true, true, &.{})).scope; 434 try s.claimJoin(b, taskToken(&s, a)); 435 try s.claimJoin(c, taskToken(&s, b)); 436 try s.bodyDone(nested, ok(0)); 437 try testing.expectEqual(@as(u32, 91), s.groups.get(nested).?.reason.?.value); 438 try testing.expectEqual(@as(u32, 91), s.groups.get(a).?.reason.?.value); 439 _ = try finish(&s, c, ok(0), .{}); 440 try testing.expect(try s.beginRelease(nested)); 441 _ = try release(&s, nested, .{}); 442 _ = try finish(&s, a, ok(0), .{}); 443 const result = (try s.pollJoin(b)).?; 444 try testing.expectEqual(.failed, result.kind); 445 try testing.expectEqual(@as(u32, 91), result.value); 446 _ = try finish(&s, b, ok(0), .{}); 447 _ = try finish(&s, 0, ok(0), .{}); 448 } 449 test "cleanup fails success, preserves primary and propagates once with explicit omissions" { 450 var registry = try S.Registry.init(testing.allocator, 1, .{ 0, 0, 0 }); 451 defer registry.deinit(); 452 var s = try S.init(testing.allocator, ®istry, 90, 91); 453 defer s.deinit(); 454 const inner = try s.nested(0); 455 const child = (try s.spawn(inner, false, false, &.{})).scope; 456 const child_result = try finish(&s, child, ok(0), S.Diagnostics.one(17)); 457 try testing.expectEqual(.failed, child_result.kind); 458 try testing.expectEqual(@as(u32, 17), child_result.value); 459 var cleanup: S.Diagnostics = .{}; 460 for (0..25) |i| cleanup.append(S.Diagnostics.one(@intCast(100 + i))); 461 const inner_result = try finish(&s, inner, ok(0), cleanup); 462 try testing.expectEqual(@as(usize, 0), inner_result.cleanup.len); // lexical return omits cleanup 463 const result = try finish(&s, 0, fail(999), S.Diagnostics.one(200)); 464 try testing.expectEqual(@as(u32, 17), result.value); // first stop, not late body fault 465 try testing.expectEqual(@as(usize, 16), result.cleanup.len); 466 try testing.expectEqual(@as(u64, 10), result.cleanup.omitted); 467 try testing.expectEqual(@as(u32, 100), result.cleanup.items[0]); 468 } 469 test "host callbacks and ancestor-owned acquisition settle before cancelled task retirement" { 470 var registry = try S.Registry.init(testing.allocator, 1, .{ 0, 0, 0 }); 471 defer registry.deinit(); 472 var s = try S.init(testing.allocator, ®istry, 90, 91); 473 defer s.deinit(); 474 const child = (try s.spawn(0, true, false, &.{})).scope; 475 try s.beginHost(child); 476 const admission = try registry.beginAcquire(0); 477 try s.cancel(0, taskToken(&s, child), 77); 478 try s.bodyDone(child, ok(0)); 479 try testing.expect(!try s.beginRelease(child)); 480 const resource = try registry.publish(admission, 5, .{}); 481 try s.endHost(child); 482 try testing.expectError(error.InvalidTransition, s.endHost(child)); 483 try testing.expect(try s.beginRelease(child)); 484 _ = try release(&s, child, .{}); 485 try testing.expectEqual(@as(u32, 5), (try registry.lookup(0, resource)).backend); 486 _ = try finish(&s, 0, ok(0), .{}); 487 } 488 fn allocationProbe(allocator: std.mem.Allocator) !void { 489 var registry = try S.Registry.init(allocator, 1, .{ 0, 0, 0 }); 490 defer registry.deinit(); 491 var s = try S.init(allocator, ®istry, 90, 91); 492 defer s.deinit(); 493 for (0..40) |_| { 494 const resource = try registry.publish(try registry.beginAcquire(0), 22, .{}); 495 const before = registry.next; 496 const count = s.groups.count(); 497 const task_count = s.tasks.count(); 498 _ = s.spawn(0, false, false, &.{resource}) catch |err| { 499 try testing.expectEqual(before, registry.next); 500 try testing.expectEqual(count, s.groups.count()); 501 try testing.expectEqual(task_count, s.tasks.count()); 502 try testing.expectEqual(@as(u32, 22), (try registry.lookup(0, resource)).backend); 503 return err; 504 }; 505 } 506 s.stop(0, cancelled(8)); 507 while (s.tasks.count() != 0) { 508 var it = s.tasks.keyIterator(); 509 const child = it.next().?.*; 510 _ = try finish(&s, child, ok(0), .{}); 511 } 512 _ = try finish(&s, 0, ok(0), .{}); 513 } 514 test "allocation failures never expose partly published tasks or moved handles" { 515 try testing.checkAllAllocationFailures(testing.allocator, allocationProbe, .{}); 516 } 517 fn capture(list: *std.ArrayList(u32), value: u32) !void { 518 try list.append(testing.allocator, value); 519 } 520 test "GC root visitation drops consumed results and repeated tasks retain no history" { 521 var registry = try S.Registry.init(testing.allocator, 1, .{ 0, 0, 0 }); 522 defer registry.deinit(); 523 var s = try S.init(testing.allocator, ®istry, 90, 91); 524 defer s.deinit(); 525 var roots: std.ArrayList(u32) = .empty; 526 defer roots.deinit(testing.allocator); 527 for (0..2048) |_| { 528 const child = (try s.spawn(0, true, true, &.{})).scope; 529 _ = try finish(&s, child, ok(12345), .{}); 530 roots.clearRetainingCapacity(); 531 try s.visitPayloads(&roots, capture); 532 try testing.expect(std.mem.indexOfScalar(u32, roots.items, 12345) != null); 533 try s.claimJoin(0, taskToken(&s, child)); 534 try testing.expectEqual(@as(u32, 12345), (try s.pollJoin(0)).?.value); 535 roots.clearRetainingCapacity(); 536 try s.visitPayloads(&roots, capture); 537 try testing.expectEqualSlices(u32, &.{ 90, 91 }, roots.items); 538 try testing.expectEqual(@as(u32, 0), s.tasks.count()); 539 try testing.expectEqual(@as(u32, 1), s.groups.count()); 540 } 541 _ = try finish(&s, 0, ok(0), .{}); 542 } 543 544 test "cancelled primaries survive cleanup and supervised discarded cleanup stays contained" { 545 var registry = try S.Registry.init(testing.allocator, 1, .{ 0, 0, 0 }); 546 defer registry.deinit(); 547 var s = try S.init(testing.allocator, ®istry, 90, 91); 548 defer s.deinit(); 549 const discarded = (try s.spawn(0, true, false, &.{})).scope; 550 _ = try finish(&s, discarded, ok(3), S.Diagnostics.one(8)); 551 try testing.expect(s.groups.get(0).?.reason == null); 552 try testing.expectEqual(@as(usize, 0), s.groups.get(0).?.cleanup.len); 553 const outcome = try finish(&s, 0, cancelled(19), S.Diagnostics.one(20)); 554 try testing.expectEqual(.cancelled, outcome.kind); 555 try testing.expectEqual(@as(u32, 19), outcome.value); 556 try testing.expectEqualSlices(u32, &.{20}, outcome.cleanup.items[0..outcome.cleanup.len]); 557 } 558 test "stop, host settlement, and teardown need no allocation" { 559 var allocator = std.testing.FailingAllocator.init(testing.allocator, .{}); 560 var registry = try S.Registry.init(allocator.allocator(), 1, .{ 0, 0, 0 }); 561 defer registry.deinit(); 562 var s = try S.init(allocator.allocator(), ®istry, 90, 91); 563 defer s.deinit(); 564 const first = (try s.spawn(0, false, true, &.{})).scope; 565 const lexical = try s.nested(first); 566 const child = (try s.spawn(lexical, false, true, &.{})).scope; 567 try s.beginHost(child); 568 allocator.fail_index = allocator.alloc_index; 569 s.stop(0, cancelled(4)); 570 try s.bodyDone(child, ok(1)); 571 try testing.expect(!try s.beginRelease(child)); 572 try s.endHost(child); 573 try testing.expect(try s.beginRelease(child)); 574 _ = try release(&s, child, S.Diagnostics.one(7)); 575 _ = try finish(&s, lexical, ok(1), .{}); 576 _ = try finish(&s, first, ok(1), .{}); 577 const outcome = try finish(&s, 0, ok(1), .{}); 578 try testing.expectEqual(.cancelled, outcome.kind); 579 try testing.expectEqualSlices(u32, &.{7}, outcome.cleanup.items[0..outcome.cleanup.len]); 580 try testing.expectEqual(@as(u32, 0), registry.resources.count()); 581 try testing.expectEqual(@as(u32, 0), registry.scopes.count()); 582 }