io_v2_core.zig (25491B)
1 //! Private structured core with a synchronous host interface. The default host 2 //! rejects all named requests. Optional test hosts provide real resources without 3 //! enabling any public v2 entrypoint. Host calls must settle before returning. 4 const std = @import("std"); 5 const Arena = @import("arena.zig").Arena; 6 const P = @import("io_v2_protocol.zig"); 7 const R = @import("io_v2_resources.zig"); 8 const T = @import("io_v2_tasks.zig"); 9 const C = @import("codecs.zig"); 10 pub const Outcome = T.Scheduler(u8, u32).Outcome; 11 const Body = struct { 12 action: u32, 13 frames: std.ArrayList(u32) = .empty, 14 mode: enum { runnable, lexical, joining, done } = .runnable, 15 }; 16 fn EngineType(comptime Host: type) type { 17 const S = T.Scheduler(Host.Backend, u32); 18 return struct { 19 const Engine = @This(); 20 host: *Host, 21 cleanup_failure: u32, 22 arena: *Arena, 23 scheduler: *S, 24 bodies: std.AutoHashMapUnmanaged(u64, Body) = .empty, 25 leaf: u32, 26 fuel: u64, 27 cursor: u64 = std.math.maxInt(u64), 28 fn deinit(self: *Engine) void { 29 var it = self.bodies.valueIterator(); 30 while (it.next()) |b| b.frames.deinit(self.arena.allocator); 31 self.bodies.deinit(self.arena.allocator); 32 } 33 pub fn reader(self: *Engine) P.Reader { 34 return .{ .arena = self.arena, .fuel = self.fuel }; 35 } 36 fn pair(self: *Engine, a: u32, b: u32) !u32 { 37 return self.arena.alloc(.{ .fork = .{ .left = a, .right = b } }); 38 } 39 pub fn text(self: *Engine, value: []const u8) !u32 { 40 return C.ofString(self.arena, value); 41 } 42 pub fn list(self: *Engine, values: []const u32) !u32 { 43 return C.ofList(self.arena, values); 44 } 45 pub fn problem(self: *Engine, category: []const u8, op: []const u8, detail: []const u8) !u32 { 46 return self.list(&.{ try self.text(category), try self.text(op), try self.text(detail), try self.text("none") }); 47 } 48 pub fn result(self: *Engine, success: bool, value: u32) !u32 { 49 return self.pair(if (success) try self.arena.alloc(.{ .stem = .{ .child = self.leaf } }) else self.leaf, try self.pair(value, self.leaf)); 50 } 51 pub fn marker(self: *Engine, name: []const u8, key: u64) !u32 { 52 return self.list(&.{ try self.text(name), try C.ofNumber(self.arena, self.scheduler.registry.identity), try C.ofNumber(self.arena, key) }); 53 } 54 fn literal(self: *Engine, id: u32, expected: []const u8) !bool { 55 var rest = id; 56 for (expected) |c| { 57 const cell = try self.reader().pair(rest) orelse return false; 58 if ((try self.reader().nat(cell.first, 7)) != @as(?u64, c)) return false; 59 rest = cell.second; 60 } 61 return try self.reader().node(rest) == .leaf; 62 } 63 pub fn token(self: *Engine, id: u32, name: []const u8) !?R.Token { 64 const tag = try self.reader().pair(id) orelse return null; 65 const run_field = try self.reader().pair(tag.second) orelse return null; 66 const key = try self.reader().pair(run_field.second) orelse return null; 67 if (try self.reader().node(key.second) != .leaf or !try self.literal(tag.first, name)) return null; 68 return .{ .run = try self.reader().nat(run_field.first, 64) orelse return null, .key = try self.reader().nat(key.first, 64) orelse return null }; 69 } 70 fn outcomeTree(self: *Engine, outcome: S.Outcome) !u32 { 71 if (outcome.kind == .completed) return self.list(&.{ try self.text("completed"), outcome.value }); 72 var diagnostics: [17]u32 = undefined; 73 @memcpy(diagnostics[0..outcome.cleanup.len], outcome.cleanup.items[0..outcome.cleanup.len]); 74 var len = outcome.cleanup.len; 75 if (outcome.cleanup.omitted != 0) { 76 var buffer: [20]u8 = undefined; 77 diagnostics[len] = try self.problem("cleanup-omitted", "close", try std.fmt.bufPrint(&buffer, "{d}", .{outcome.cleanup.omitted})); 78 len += 1; 79 } 80 return self.list(&.{ try self.text(if (outcome.kind == .failed) "failed" else "cancelled"), outcome.value, try self.list(diagnostics[0..len]) }); 81 } 82 fn done(self: *Engine, scope: u64, outcome: S.Outcome) !void { 83 try self.scheduler.bodyDone(scope, outcome); 84 const body = self.bodies.getPtr(scope).?; 85 body.mode = .done; 86 body.action = self.leaf; 87 body.frames.clearRetainingCapacity(); 88 } 89 fn resumeValue(self: *Engine, scope: u64, value: u32) !void { 90 const body = self.bodies.getPtr(scope).?; 91 if (body.frames.pop()) |continuation| { 92 body.action = try self.arena.allocApp(continuation, value); 93 body.mode = .runnable; 94 } else try self.done(scope, .{ .kind = .completed, .value = value }); 95 } 96 fn fault(self: *Engine, scope: u64, detail: []const u8) !void { 97 try self.done(scope, .{ .kind = .failed, .value = try self.problem("protocol-error", "core/decode", detail) }); 98 } 99 fn operationError(self: *Engine, scope: u64, op: []const u8, err: anyerror) !void { 100 const category = switch (err) { 101 error.StaleTask => "stale-task", 102 error.JoinCycle => "join-cycle", 103 error.OutOfScope => "out-of-scope", 104 error.StaleHandle => "stale-handle", 105 error.NotOwner => "not-owner", 106 error.BorrowedResource => "borrowed-resource", 107 error.InternalResource => "internal-resource", 108 error.ClosedHandle => "closed-handle", 109 error.Busy => "busy", 110 error.DuplicateHandle => "duplicate-handle", 111 error.ClosedScope => "closed-scope", 112 error.InvalidArgument => "invalid-argument", 113 else => return err, 114 }; 115 try self.resumeValue(scope, try self.result(false, try self.problem(category, op, ""))); 116 } 117 fn step(self: *Engine, scope: u64) !void { 118 const body = self.bodies.get(scope).?; 119 const group = self.scheduler.groups.get(scope).?; 120 if (body.mode == .done or body.mode == .lexical) return; 121 if (group.reason) |reason| return self.done(scope, reason); 122 if (body.mode == .joining) { 123 if (try self.scheduler.pollJoin(scope)) |outcome| try self.resumeValue(scope, try self.result(true, try self.outcomeTree(outcome))); 124 return; 125 } 126 const action = P.decode(self.arena, body.action, self.fuel) catch |err| return switch (err) { 127 error.InvalidComputationShape => self.fault(scope, "invalid computation shape"), 128 error.UnknownComputationConstructor => self.fault(scope, "unknown computation constructor"), 129 error.InvalidFacilityHeader => self.fault(scope, "invalid facility header"), 130 else => err, 131 }; 132 switch (action) { 133 .request => |request| { 134 return self.resumeValue(scope, try self.host.request(self, scope, request)); 135 }, 136 .core => |core| switch (core.code) { 137 .pure => try self.resumeValue(scope, core.payload), 138 .fail => try self.done(scope, .{ .kind = .failed, .value = core.payload }), 139 .cancelled => try self.done(scope, .{ .kind = .cancelled, .value = core.payload }), 140 .bind => { 141 const parts = try self.reader().pair(core.payload) orelse return self.fault(scope, "invalid v2 computation"); 142 try self.bodies.getPtr(scope).?.frames.append(self.arena.allocator, parts.second); 143 self.bodies.getPtr(scope).?.action = parts.first; 144 }, 145 .current_scope => try self.resumeValue(scope, try self.marker("io.v2.scope", scope)), 146 .scope => { 147 try self.bodies.ensureUnusedCapacity(self.arena.allocator, 1); 148 const child = try self.scheduler.nested(scope); 149 const child_action = try self.arena.allocApp(core.payload, try self.marker("io.v2.scope", child)); 150 self.bodies.putAssumeCapacityNoClobber(child, .{ .action = child_action }); 151 self.bodies.getPtr(scope).?.action = self.leaf; 152 self.bodies.getPtr(scope).?.mode = .lexical; 153 }, 154 .spawn => { 155 const supervised = try self.reader().pair(core.payload) orelse return self.operationError(scope, "core/spawn", error.InvalidArgument); 156 const keep = try self.reader().pair(supervised.second) orelse return self.operationError(scope, "core/spawn", error.InvalidArgument); 157 const handles = try self.reader().pair(keep.second) orelse return self.operationError(scope, "core/spawn", error.InvalidArgument); 158 const contained = try self.reader().nat(supervised.first, 1) orelse return self.operationError(scope, "core/spawn", error.InvalidArgument); 159 const retained = try self.reader().nat(keep.first, 1) orelse return self.operationError(scope, "core/spawn", error.InvalidArgument); 160 var tokens: [R.transfer_limit]R.Token = undefined; 161 var count: usize = 0; 162 var rest = handles.first; 163 while (try self.reader().node(rest) != .leaf) { 164 if (count == tokens.len) return self.operationError(scope, "core/spawn", error.InvalidArgument); 165 const cell = try self.reader().pair(rest) orelse return self.operationError(scope, "core/spawn", error.InvalidArgument); 166 tokens[count] = try self.token(cell.first, "io.v2.resource") orelse return self.operationError(scope, "core/spawn", error.InvalidArgument); 167 count += 1; 168 rest = cell.second; 169 } 170 try self.bodies.ensureUnusedCapacity(self.arena.allocator, 1); 171 const child = self.scheduler.spawn(scope, contained == 1, retained == 1, tokens[0..count]) catch |err| return self.operationError(scope, "core/spawn", err); 172 var fresh: [R.transfer_limit]u32 = undefined; 173 for (child.tokens[0..count], 0..) |t, i| fresh[i] = try self.marker("io.v2.resource", t.key); 174 const child_action = try self.arena.allocApp(handles.second, try self.list(fresh[0..count])); 175 self.bodies.putAssumeCapacityNoClobber(child.scope, .{ .action = child_action }); 176 try self.resumeValue(scope, try self.result(true, try self.list(&.{ try self.marker("io.v2.task", child.scope), if (retained == 1) try self.marker("io.v2.join", child.scope) else self.leaf }))); 177 }, 178 .join, .discard => { 179 const op: []const u8 = if (core.code == .join) "core/join" else "core/discard"; 180 const ref = try self.token(core.payload, "io.v2.join") orelse return self.operationError(scope, op, error.InvalidArgument); 181 if (core.code == .discard) { 182 self.scheduler.discard(scope, ref) catch |err| return self.operationError(scope, op, err); 183 try self.resumeValue(scope, try self.result(true, self.leaf)); 184 } else { 185 self.scheduler.claimJoin(scope, ref) catch |err| return self.operationError(scope, op, err); 186 self.bodies.getPtr(scope).?.action = self.leaf; 187 self.bodies.getPtr(scope).?.mode = .joining; 188 } 189 }, 190 .cancel => { 191 const parts = try self.reader().pair(core.payload) orelse return self.operationError(scope, "core/cancel", error.InvalidArgument); 192 const ref = try self.token(parts.first, "io.v2.task") orelse return self.operationError(scope, "core/cancel", error.InvalidArgument); 193 self.scheduler.cancel(scope, ref, parts.second) catch |err| return self.operationError(scope, "core/cancel", err); 194 try self.resumeValue(scope, try self.result(true, self.leaf)); 195 }, 196 }, 197 } 198 } 199 fn settleOne(self: *Engine) !?S.Outcome { 200 var it = self.scheduler.groups.iterator(); 201 while (it.next()) |entry| { 202 if (entry.value_ptr.phase != .joining) continue; 203 const scope = entry.key_ptr.*; 204 if (!try self.scheduler.beginRelease(scope)) continue; 205 const group = self.scheduler.groups.get(scope).?; 206 const cleanup = self.releaseScope(scope); 207 const outcome = try self.scheduler.settle(scope, cleanup); 208 var removed = self.bodies.fetchRemove(scope).?.value; 209 removed.frames.deinit(self.arena.allocator); 210 if (scope == 0) return outcome; 211 if (group.kind == .lexical) { 212 const parent = group.parent.?; 213 self.bodies.getPtr(parent).?.mode = .runnable; 214 if (outcome.kind == .completed) try self.resumeValue(parent, outcome.value) else try self.done(parent, outcome); 215 } 216 break; // iterator invalidated by settlement 217 } 218 return null; 219 } 220 // Synchronous hosts have no outstanding OS work. Abort settles every logical 221 // scope/task without allocating, including a child admitted just before its 222 // body/continuation allocation failed. Physical release is allocation-free. 223 // Asynchronous hosts require a separate cancellation/draining pump. 224 fn abort(self: *Engine) void { 225 self.scheduler.stop(0, .{ .kind = .cancelled, .value = self.leaf }); 226 while (self.scheduler.groups.count() != 0) { 227 var progressed = false; 228 var bodies = self.scheduler.groups.iterator(); 229 while (bodies.next()) |entry| { 230 if (entry.value_ptr.phase != .open or entry.value_ptr.lexical != null) continue; 231 self.scheduler.bodyDone(entry.key_ptr.*, entry.value_ptr.reason.?) catch unreachable; 232 if (self.bodies.getPtr(entry.key_ptr.*)) |body| { 233 body.mode = .done; 234 body.action = self.leaf; 235 body.frames.clearRetainingCapacity(); 236 } 237 progressed = true; 238 } 239 var groups = self.scheduler.groups.iterator(); 240 while (groups.next()) |entry| { 241 const scope = entry.key_ptr.*; 242 if (entry.value_ptr.phase != .joining) continue; 243 if (!(self.scheduler.beginRelease(scope) catch unreachable)) continue; 244 const cleanup = self.releaseScope(scope); 245 _ = self.scheduler.settle(scope, cleanup) catch unreachable; 246 if (self.bodies.fetchRemove(scope)) |removed| { 247 var body = removed.value; 248 body.frames.deinit(self.arena.allocator); 249 } 250 progressed = true; 251 break; 252 } 253 std.debug.assert(progressed); // no host callbacks exist in this harness 254 } 255 std.debug.assert(self.scheduler.tasks.count() == 0); 256 std.debug.assert(self.scheduler.registry.scopes.count() == 0); 257 std.debug.assert(self.scheduler.registry.resources.count() == 0); 258 } 259 fn releaseScope(self: *Engine, scope: u64) S.Diagnostics { 260 var cleanup: S.Diagnostics = .{}; 261 while (self.scheduler.registry.newestOwned(scope)) |ref| { 262 const resource = (self.scheduler.registry.claimClose(scope, ref, true) catch unreachable).?; 263 if (!resource.flags.borrowed) self.host.close(self.scheduler.registry, resource.backend) catch { 264 cleanup.append(S.Diagnostics.one(self.cleanup_failure)); 265 }; 266 self.scheduler.registry.finishClose(ref) catch unreachable; 267 } 268 return cleanup; 269 } 270 fn appendRoot(arena: *Arena, id: u32) !void { 271 try arena.roots.append(arena.allocator, id); 272 } 273 fn roots(self: *Engine, base: usize) !void { 274 self.arena.roots.shrinkRetainingCapacity(base); 275 try appendRoot(self.arena, self.leaf); 276 try appendRoot(self.arena, self.cleanup_failure); 277 try self.scheduler.visitPayloads(self.arena, appendRoot); 278 var it = self.bodies.valueIterator(); 279 while (it.next()) |body| { 280 try appendRoot(self.arena, body.action); 281 try self.arena.roots.appendSlice(self.arena.allocator, body.frames.items); 282 } 283 } 284 fn nextBody(self: *Engine) ?u64 { 285 var first: ?u64 = null; 286 var after: ?u64 = null; 287 var it = self.bodies.iterator(); 288 while (it.next()) |entry| { 289 if (entry.value_ptr.mode == .done or entry.value_ptr.mode == .lexical) continue; 290 const key = entry.key_ptr.*; 291 if (first == null or key < first.?) first = key; 292 if (key > self.cursor and (after == null or key < after.?)) after = key; 293 } 294 self.cursor = after orelse first orelse return null; 295 return self.cursor; 296 } 297 }; 298 } 299 300 // Caller retains input/output roots. Fatal reducer/allocator errors drain the 301 // synchronous host before propagating; host-side cleanup records must not allocate. 302 // Host is deliberately SYNCHRONOUS: request and close must return only after 303 // their work settles. No pending callbacks, detached work or roots outside the 304 // scheduler are supported by this interface. close must not allocate and must 305 // retire its physical resource even on error. Async hosts need a different pump. 306 pub const UnsupportedHost = struct { 307 pub const Backend = u8; 308 pub const standard = [3]u8{ 0, 0, 0 }; 309 pub fn request(_: *@This(), engine: anytype, _: u64, req: anytype) !u32 { 310 return unsupported(engine, req); 311 } 312 pub fn close(_: *@This(), _: anytype, _: u8) error{}!void { 313 unreachable; 314 } 315 }; 316 pub fn unsupported(engine: anytype, request: anytype) !u32 { 317 var operation: [129]u8 = undefined; 318 var detail: [40]u8 = undefined; 319 const op = try std.fmt.bufPrint(&operation, "{s}/{s}", .{ request.group.slice(), request.operation.slice() }); 320 const message = try std.fmt.bufPrint(&detail, "group revision {d}", .{request.revision}); 321 return engine.result(false, try engine.problem("unsupported", op, message)); 322 } 323 pub fn run(arena: *Arena, action: u32, identity: u64, fuel: u64) !Outcome { 324 var host: UnsupportedHost = .{}; 325 return runWithHost(UnsupportedHost, &host, arena, action, identity, fuel); 326 } 327 pub fn runWithHost(comptime Host: type, host: *Host, arena: *Arena, action: u32, identity: u64, fuel: u64) !T.Scheduler(Host.Backend, u32).Outcome { 328 const S = T.Scheduler(Host.Backend, u32); 329 const Engine = EngineType(Host); 330 const base = arena.roots.items.len; 331 defer arena.roots.shrinkRetainingCapacity(base); 332 var registry = try S.Registry.init(arena.allocator, identity, Host.standard); 333 defer registry.deinit(); 334 var scheduler = try S.init(arena.allocator, ®istry, try C.ofString(arena, "owning group stopped"), try C.ofString(arena, "task join cycle")); 335 defer scheduler.deinit(); 336 const cleanup_failure = try C.ofList(arena, &.{ try C.ofString(arena, "cleanup-failed"), try C.ofString(arena, "resource/close"), try C.ofString(arena, "host close failed"), try C.ofString(arena, "unknown") }); 337 var engine: Engine = .{ .host = host, .cleanup_failure = cleanup_failure, .arena = arena, .scheduler = &scheduler, .leaf = try arena.alloc(.leaf), .fuel = fuel }; 338 defer engine.deinit(); 339 errdefer engine.abort(); 340 try engine.bodies.put(arena.allocator, 0, .{ .action = action }); 341 while (true) { 342 try engine.roots(base); 343 if (try engine.settleOne()) |outcome| return outcome; 344 // Settlement may construct a resumed continuation. Refresh before demand. 345 try engine.roots(base); 346 if (engine.nextBody()) |scope| try engine.step(scope); 347 } 348 } 349 350 const testing = std.testing; 351 fn pair(a: *Arena, x: u32, y: u32) !u32 { 352 return a.alloc(.{ .fork = .{ .left = x, .right = y } }); 353 } 354 fn makeAction(a: *Arena, tag: u64, payload: u32) !u32 { 355 return pair(a, try C.ofNumber(a, tag), payload); 356 } 357 fn constant(a: *Arena, value: u32) !u32 { 358 return pair(a, try a.alloc(.leaf), value); 359 } 360 fn poison(a: *Arena) !u32 { 361 const p = try a.alloc(.leaf); 362 a.get(p).* = .{ .app = .{ .func = p, .arg = p } }; 363 return p; 364 } 365 test "real bind and scope continuations with demanded GC and opaque discarded arguments" { 366 var a = Arena.init(testing.allocator); 367 defer a.deinit(); 368 const p = try poison(&a); 369 var program = try makeAction(&a, 0, try C.ofNumber(&a, 42)); 370 for (0..100) |_| { 371 program = try makeAction(&a, 1, try pair(&a, try makeAction(&a, 0, p), try constant(&a, program))); 372 program = try makeAction(&a, 3, try constant(&a, program)); 373 } 374 a.gc_enabled = true; 375 try a.setGcPolicy(1, 1); 376 const outcome = try run(&a, program, 1, 1000); 377 try testing.expectEqual(.completed, outcome.kind); 378 // No reduction until the result has an externally retained root. 379 try a.roots.append(a.allocator, outcome.value); 380 try testing.expectEqual(@as(?u64, 42), try C.toNumber(&a, outcome.value)); 381 try testing.expect(a.collections > 0); 382 a.roots.clearRetainingCapacity(); 383 try a.collect(&.{}); 384 try testing.expectEqual(@as(usize, 0), a.last_live); 385 } 386 test "terminal diagnostics and cancelled reasons are not demanded" { 387 for ([_]u64{ 2, 9 }) |tag| { 388 var a = Arena.init(testing.allocator); 389 defer a.deinit(); 390 const p = try poison(&a); 391 const outcome = try run(&a, try makeAction(&a, tag, p), 1, 0); 392 try testing.expectEqual(p, outcome.value); 393 try testing.expectEqual(if (tag == 2) @TypeOf(outcome.kind).failed else .cancelled, outcome.kind); 394 try testing.expect(a.get(p).* == .app); 395 } 396 } 397 test "unknown valid facility is an ordinary Result without demanding arguments" { 398 var a = Arena.init(testing.allocator); 399 defer a.deinit(); 400 const p = try poison(&a); 401 const request = try makeAction(&a, 10, try pair(&a, try C.ofString(&a, "future"), try pair(&a, try C.ofNumber(&a, 9), try pair(&a, try C.ofString(&a, "call"), p)))); 402 const outcome = try run(&a, request, 1, 0); 403 try testing.expectEqual(.completed, outcome.kind); 404 const result = a.get(outcome.value).fork; 405 try testing.expect(a.get(result.left).* == .leaf); 406 const problem = a.get(result.right).fork.left; 407 const category = a.get(problem).fork.left; 408 const name = (try C.toString(&a, category)).?; 409 defer testing.allocator.free(name); 410 try testing.expectEqualStrings("unsupported", name); 411 try testing.expect(a.get(p).* == .app); 412 } 413 fn allocationProbe(allocator: std.mem.Allocator) !void { 414 var a = Arena.init(allocator); 415 defer a.deinit(); 416 const value = try C.ofNumber(&a, 3); 417 var program = try makeAction(&a, 0, value); 418 for (0..8) |_| program = try makeAction(&a, 3, try constant(&a, program)); 419 _ = try run(&a, program, 1, 100); 420 try testing.expectEqual(@as(usize, 0), a.roots.items.len); 421 } 422 test "private core allocator failures unwind all host memory and temporary roots" { 423 try testing.checkAllAllocationFailures(testing.allocator, allocationProbe, .{}); 424 } 425 426 fn spawnAllocationProbe(allocator: std.mem.Allocator) !void { 427 var a = Arena.init(allocator); 428 defer a.deinit(); 429 const leaf = try a.alloc(.leaf); 430 const one = try C.ofNumber(&a, 1); 431 const child = try constant(&a, try makeAction(&a, 0, try C.ofNumber(&a, 3))); 432 const spawned = try makeAction(&a, 5, try pair(&a, one, try pair(&a, one, try pair(&a, leaf, child)))); 433 // First child in this independent run owns ID4. Constructing the reference 434 // independently also checks the exact source token representation. 435 const ticket = try C.ofList(&a, &.{ try C.ofString(&a, "io.v2.join"), one, try C.ofNumber(&a, 4) }); 436 const program = try makeAction(&a, 1, try pair(&a, spawned, try constant(&a, try makeAction(&a, 6, ticket)))); 437 a.gc_enabled = true; 438 try a.setGcPolicy(1, 1); 439 const outcome = try run(&a, program, 1, 100); 440 try testing.expectEqual(.completed, outcome.kind); 441 try testing.expectEqual(@as(usize, 0), a.roots.items.len); 442 } 443 test "spawn publication and join reply allocation failures settle every admitted child" { 444 try testing.checkAllAllocationFailures(testing.allocator, spawnAllocationProbe, .{}); 445 }