io_v2_resources.zig (20889B)
1 //! Single-owner-event-loop registry. This module performs no OS operations and 2 //! never invokes user callbacks. Scheduler publication must reserve its own memory 3 //! before reserveChild; no fallible work may separate that commit from admission. 4 //! Backend values are owned by the embedding. deinit frees registry memory only; 5 //! the embedding must drain admitted opens/leases/releases and settle scopes first. 6 const std = @import("std"); 7 pub const Scope = u64; 8 pub const Token = struct { run: u64, key: u64 }; 9 pub const Access = enum { exclusive, read, write }; 10 pub const Flags = packed struct { borrowed: bool = false, internal: bool = false, duplex: bool = false }; 11 pub const Fault = error{ ClosedScope, StaleHandle, OutOfScope, DuplicateHandle, NotOwner, BorrowedResource, InternalResource, ClosedHandle, Busy, InvalidAdmission, InvalidLease, InvalidClose, IdExhausted, TransferLimit, OutOfMemory }; 12 pub const transfer_limit = 256; 13 14 pub fn Registry(comptime Backend: type) type { 15 return struct { 16 const Self = @This(); 17 const ScopeState = struct { parent: ?Scope, admitting: bool = true, pending: usize = 0, operations: usize = 0, children: usize = 0 }; 18 pub const Resource = struct { owner: Scope, backend: Backend, flags: Flags, busy: u2 = 0, closing: bool = false, releasing: bool = false }; 19 const LeaseState = struct { scope: Scope, resource: u64, mask: u2 }; 20 pub const Lease = struct { token: Token, backend: Backend }; 21 pub const Transfer = struct { scope: Scope, tokens: [transfer_limit]Token = undefined, len: usize }; 22 allocator: std.mem.Allocator, 23 identity: u64, 24 next: u64 = 4, // root=0; standard references are reserved at 1..3 25 scopes: std.AutoHashMapUnmanaged(Scope, ScopeState) = .empty, 26 resources: std.AutoHashMapUnmanaged(u64, Resource) = .empty, 27 admissions: std.AutoHashMapUnmanaged(u64, Scope) = .empty, 28 leases: std.AutoHashMapUnmanaged(u64, LeaseState) = .empty, 29 30 pub fn init(allocator: std.mem.Allocator, identity: u64, standard: [3]Backend) !Self { 31 var self: Self = .{ .allocator = allocator, .identity = identity }; 32 errdefer self.deinit(); 33 try self.scopes.put(allocator, 0, .{ .parent = null }); 34 for (standard, 1..) |backend, standard_key| try self.resources.put(allocator, standard_key, .{ .owner = 0, .backend = backend, .flags = .{ .borrowed = true } }); 35 return self; 36 } 37 pub fn deinit(self: *Self) void { 38 self.leases.deinit(self.allocator); 39 self.admissions.deinit(self.allocator); 40 self.resources.deinit(self.allocator); 41 self.scopes.deinit(self.allocator); 42 } 43 fn room(self: *const Self, count: u64) Fault!void { 44 _ = std.math.add(u64, self.next, count) catch return error.IdExhausted; 45 } 46 pub fn within(self: *const Self, current: Scope, owner: Scope) bool { 47 var cursor = current; 48 while (self.scopes.get(cursor)) |scope| { 49 if (cursor == owner) return true; 50 cursor = scope.parent orelse return false; 51 } 52 return false; 53 } 54 fn openScope(self: *Self, scope: Scope) Fault!*ScopeState { 55 const state = self.scopes.getPtr(scope) orelse return error.ClosedScope; 56 if (!state.admitting) return error.ClosedScope; 57 return state; 58 } 59 fn key(self: *const Self, token: Token) Fault!u64 { 60 if (token.run != self.identity) return error.StaleHandle; 61 return token.key; 62 } 63 pub fn lookup(self: *const Self, current: Scope, token: Token) Fault!Resource { 64 const resource = self.resources.get(try self.key(token)) orelse return error.StaleHandle; 65 if (!self.within(current, resource.owner)) return error.OutOfScope; 66 return resource; 67 } 68 pub fn beginCloseScope(self: *Self, scope: Scope) void { 69 if (self.scopes.getPtr(scope)) |state| state.admitting = false; 70 } 71 // Admission is counted through host acquisition and publication/abort. 72 // Allocate all map capacity BEFORE the host can create a resource. 73 pub fn beginAcquire(self: *Self, owner: Scope) Fault!Token { 74 _ = try self.openScope(owner); 75 try self.room(1); 76 try self.admissions.ensureUnusedCapacity(self.allocator, 1); 77 try self.resources.ensureTotalCapacity(self.allocator, std.math.add(u32, self.resources.count(), std.math.add(u32, self.admissions.count(), 1) catch return error.OutOfMemory) catch return error.OutOfMemory); 78 const id = self.next; 79 self.next += 1; 80 self.admissions.putAssumeCapacityNoClobber(id, owner); 81 self.scopes.getPtr(owner).?.pending += 1; 82 return .{ .run = self.identity, .key = id }; 83 } 84 pub fn abortAcquire(self: *Self, admission: Token) Fault!void { 85 const removed = self.admissions.fetchRemove(try self.key(admission)) orelse return error.InvalidAdmission; 86 self.scopes.getPtr(removed.value).?.pending -= 1; 87 } 88 // No allocation. A counter-exhaustion failure leaves admission live; 89 // the embedding must release its raw backend and abort that admission. 90 // IDs reflect publication order, not asynchronous open-start order. 91 pub fn publish(self: *Self, admission: Token, backend: Backend, flags: Flags) Fault!Token { 92 const id = try self.key(admission); 93 const owner = self.admissions.get(id) orelse return error.InvalidAdmission; 94 try self.room(1); 95 const key_value = self.next; 96 self.next += 1; 97 self.resources.putAssumeCapacityNoClobber(key_value, .{ .owner = owner, .backend = backend, .flags = flags }); 98 _ = self.admissions.remove(id); 99 self.scopes.getPtr(owner).?.pending -= 1; 100 return .{ .run = self.identity, .key = key_value }; 101 } 102 pub fn reserveChild(self: *Self, parent: Scope, tokens: []const Token) Fault!Transfer { 103 _ = try self.openScope(parent); 104 if (tokens.len > transfer_limit) return error.TransferLimit; 105 var sorted: [transfer_limit]u64 = undefined; 106 // Validate every token before any mutation. No backend copies escape 107 // on failure and no old token is invalidated on partial validation. 108 for (tokens, 0..) |token, i| { 109 for (tokens[0..i]) |previous| if (std.meta.eql(previous, token)) return error.DuplicateHandle; 110 const resource = try self.lookup(parent, token); 111 if (resource.owner != parent) return error.NotOwner; 112 if (resource.flags.borrowed) return error.BorrowedResource; 113 if (resource.flags.internal) return error.InternalResource; 114 if (resource.closing) return error.ClosedHandle; 115 if (resource.busy != 0) return error.Busy; 116 sorted[i] = token.key; 117 } 118 std.mem.sort(u64, sorted[0..tokens.len], {}, std.sort.asc(u64)); 119 try self.room(1 + tokens.len); 120 try self.scopes.ensureUnusedCapacity(self.allocator, 1); 121 var result: Transfer = .{ .scope = self.next, .len = tokens.len }; 122 self.next += 1 + tokens.len; 123 self.scopes.putAssumeCapacityNoClobber(result.scope, .{ .parent = parent }); 124 self.scopes.getPtr(parent).?.children += 1; 125 // Remove/insert each entry without allocation; capacity does not grow. 126 // Sorted original IDs preserve acquisition order, not argument order. 127 for (sorted[0..tokens.len], 0..) |old, index| { 128 var resource = self.resources.fetchRemove(old).?.value; 129 resource.owner = result.scope; 130 const fresh = result.scope + 1 + index; 131 self.resources.putAssumeCapacityNoClobber(fresh, resource); 132 for (tokens, 0..) |token, i| if (token.key == old) { 133 result.tokens[i] = .{ .run = self.identity, .key = fresh }; 134 }; 135 } 136 return result; 137 } 138 pub fn beginLease(self: *Self, current: Scope, token: Token, access: Access, internal: bool) Fault!Lease { 139 const resource = try self.lookup(current, token); 140 if (resource.closing) return error.ClosedHandle; 141 if (resource.flags.internal and !internal) return error.InternalResource; 142 const mask: u2 = if (!resource.flags.duplex or access == .exclusive) 3 else if (access == .read) 1 else 2; 143 if (resource.busy & mask != 0) return error.Busy; 144 try self.room(1); 145 try self.leases.ensureUnusedCapacity(self.allocator, 1); 146 const id = self.next; 147 self.next += 1; 148 self.leases.putAssumeCapacityNoClobber(id, .{ .scope = current, .resource = token.key, .mask = mask }); 149 self.resources.getPtr(token.key).?.busy |= mask; 150 self.scopes.getPtr(current).?.operations += 1; 151 return .{ .token = .{ .run = self.identity, .key = id }, .backend = resource.backend }; 152 } 153 pub fn endLease(self: *Self, lease: Token) Fault!void { 154 const entry = self.leases.fetchRemove(try self.key(lease)) orelse return error.InvalidLease; 155 self.resources.getPtr(entry.value.resource).?.busy &= ~entry.value.mask; 156 self.scopes.getPtr(entry.value.scope).?.operations -= 1; 157 } 158 // Mark closing before waiting, blocking handoff and new leases. null 159 // means a lease or another close is in flight, not permission to retry IO. 160 pub fn claimClose(self: *Self, current: Scope, token: Token, settling: bool) Fault!?Resource { 161 const resource = try self.lookup(current, token); 162 if (!settling) { 163 if (resource.flags.internal) return error.InternalResource; 164 if (resource.flags.borrowed) return error.BorrowedResource; 165 } 166 const entry = self.resources.getPtr(token.key).?; 167 entry.closing = true; 168 if (entry.busy != 0 or entry.releasing) return null; 169 entry.releasing = true; 170 return entry.*; 171 } 172 // Exactly once, including after a host release error. Keep the entry until 173 // the callback settles so finishScope cannot outrun asynchronous close. 174 pub fn finishClose(self: *Self, token: Token) Fault!void { 175 const id = try self.key(token); 176 const entry = self.resources.get(id) orelse return error.StaleHandle; 177 if (!entry.releasing) return error.InvalidClose; 178 _ = self.resources.remove(id); 179 } 180 pub fn newestOwned(self: *const Self, scope: Scope) ?Token { 181 var newest: ?u64 = null; 182 var iterator = self.resources.iterator(); 183 while (iterator.next()) |entry| { 184 if (entry.value_ptr.owner == scope and (newest == null or entry.key_ptr.* > newest.?)) newest = entry.key_ptr.*; 185 } 186 return if (newest) |id| .{ .run = self.identity, .key = id } else null; 187 } 188 // Scheduler must first settle descendants. No reachability-based lifetime 189 // extension, recursive implicit cancellation, or silent orphan removal here. 190 pub fn readyForRelease(self: *const Self, scope: Scope) bool { 191 const state = self.scopes.get(scope) orelse return false; 192 return !state.admitting and state.children == 0 and state.pending == 0 and state.operations == 0; 193 } 194 pub fn finishScope(self: *Self, scope: Scope) Fault!void { 195 const state = self.scopes.get(scope) orelse return error.ClosedScope; 196 if (state.admitting or state.children != 0 or state.pending != 0 or state.operations != 0 or self.newestOwned(scope) != null) return error.Busy; 197 _ = self.scopes.remove(scope); 198 if (state.parent) |parent| self.scopes.getPtr(parent).?.children -= 1; 199 } 200 }; 201 } 202 203 const testing = std.testing; 204 const TestRegistry = Registry(u32); 205 fn create(r: *TestRegistry, scope: Scope, value: u32, flags: Flags) !Token { 206 return r.publish(try r.beginAcquire(scope), value, flags); 207 } 208 fn closeScope(r: *TestRegistry, scope: Scope) !void { 209 r.beginCloseScope(scope); 210 while (r.newestOwned(scope)) |token| { 211 _ = (try r.claimClose(scope, token, true)) orelse return error.Busy; 212 try r.finishClose(token); 213 } 214 try r.finishScope(scope); 215 } 216 test "closing admission preserves existing descendant borrowing and admitted opens" { 217 var r = try TestRegistry.init(testing.allocator, 7, .{ 10, 11, 12 }); 218 defer r.deinit(); 219 const child = (try r.reserveChild(0, &.{})).scope; 220 const admission = try r.beginAcquire(0); 221 r.beginCloseScope(0); 222 try testing.expectError(error.ClosedScope, r.beginAcquire(0)); 223 try testing.expectError(error.ClosedScope, r.reserveChild(0, &.{})); 224 try testing.expectError(error.Busy, r.finishScope(0)); 225 const token = try r.publish(admission, 99, .{}); 226 const lease = try r.beginLease(child, token, .read, false); 227 try testing.expectEqual(@as(u32, 99), lease.backend); 228 r.beginCloseScope(child); 229 try testing.expectError(error.Busy, r.finishScope(child)); 230 try r.endLease(lease.token); 231 try closeScope(&r, child); 232 try closeScope(&r, 0); 233 try testing.expectEqual(@as(u32, 0), r.scopes.count()); 234 try testing.expectEqual(@as(u32, 0), r.resources.count()); 235 } 236 test "handoff validates atomically, rekeys aliases, and preserves acquisition order" { 237 var r = try TestRegistry.init(testing.allocator, 8, .{ 0, 0, 0 }); 238 defer r.deinit(); 239 const a = try create(&r, 0, 101, .{}); 240 const b = try create(&r, 0, 102, .{}); 241 const next = r.next; 242 try testing.expectError(error.DuplicateHandle, r.reserveChild(0, &.{ a, a })); 243 try testing.expectError(error.StaleHandle, r.reserveChild(0, &.{ a, .{ .run = 9, .key = b.key } })); 244 try testing.expectError(error.BorrowedResource, r.reserveChild(0, &.{ a, .{ .run = 8, .key = 1 } })); 245 try testing.expectEqual(next, r.next); 246 const transferred = try r.reserveChild(0, &.{ b, a }); 247 try testing.expectError(error.StaleHandle, r.lookup(0, a)); 248 try testing.expectError(error.StaleHandle, r.lookup(transferred.scope, b)); 249 try testing.expectError(error.OutOfScope, r.lookup(0, transferred.tokens[0])); 250 try testing.expectEqual(@as(u32, 102), (try r.lookup(transferred.scope, transferred.tokens[0])).backend); 251 try testing.expectEqual(transferred.tokens[0], r.newestOwned(transferred.scope).?); 252 const inner = (try r.reserveChild(transferred.scope, &.{})).scope; 253 try testing.expectError(error.NotOwner, r.reserveChild(inner, transferred.tokens[1..2])); 254 try closeScope(&r, inner); 255 try closeScope(&r, transferred.scope); 256 try closeScope(&r, 0); 257 } 258 test "direction leases, close claims and stale completions cannot cross ownership" { 259 var r = try TestRegistry.init(testing.allocator, 1, .{ 0, 0, 0 }); 260 defer r.deinit(); 261 const token = try create(&r, 0, 5, .{ .duplex = true }); 262 const read = try r.beginLease(0, token, .read, false); 263 const write = try r.beginLease(0, token, .write, false); 264 try testing.expectError(error.Busy, r.beginLease(0, token, .read, false)); 265 try testing.expectError(error.Busy, r.reserveChild(0, &.{token})); 266 try r.endLease(read.token); 267 const read2 = try r.beginLease(0, token, .read, false); 268 try testing.expectError(error.InvalidLease, r.endLease(read.token)); 269 try testing.expectEqual(@as(?TestRegistry.Resource, null), try r.claimClose(0, token, false)); 270 try testing.expectError(error.ClosedHandle, r.beginLease(0, token, .read, false)); 271 try testing.expectError(error.ClosedHandle, r.reserveChild(0, &.{token})); 272 try r.endLease(write.token); 273 try r.endLease(read2.token); 274 _ = (try r.claimClose(0, token, false)).?; 275 try testing.expectEqual(@as(?TestRegistry.Resource, null), try r.claimClose(0, token, false)); 276 r.beginCloseScope(0); 277 try testing.expectError(error.Busy, r.finishScope(0)); 278 try r.finishClose(token); // also required after failed host close 279 try testing.expectError(error.StaleHandle, r.finishClose(token)); 280 try closeScope(&r, 0); 281 } 282 test "single file cursor, internal resources and borrowed standards stay distinct" { 283 var r = try TestRegistry.init(testing.allocator, 1, .{ 0, 0, 0 }); 284 defer r.deinit(); 285 const file = try create(&r, 0, 0, .{}); 286 const reader = try r.beginLease(0, file, .read, false); 287 try testing.expectError(error.Busy, r.beginLease(0, file, .write, false)); 288 try r.endLease(reader.token); 289 const hidden = try create(&r, 0, 0, .{ .internal = true }); 290 try testing.expectError(error.InternalResource, r.beginLease(0, hidden, .read, false)); 291 try testing.expectError(error.InternalResource, r.claimClose(0, hidden, false)); 292 try testing.expectError(error.InternalResource, r.reserveChild(0, &.{hidden})); 293 const internal = try r.beginLease(0, hidden, .exclusive, true); 294 try r.endLease(internal.token); 295 try testing.expectError(error.BorrowedResource, r.claimClose(0, .{ .run = 1, .key = 2 }, false)); 296 try closeScope(&r, 0); 297 } 298 test "publication order, multiple admitted opens, and explicit abort" { 299 var r = try TestRegistry.init(testing.allocator, 1, .{ 0, 0, 0 }); 300 defer r.deinit(); 301 var admissions: [64]Token = undefined; 302 for (&admissions) |*a| a.* = try r.beginAcquire(0); 303 r.beginCloseScope(0); 304 var index: usize = admissions.len; 305 while (index > 1) { 306 index -= 1; 307 const token = try r.publish(admissions[index], @intCast(index), .{}); 308 try testing.expectEqual(token, r.newestOwned(0).?); 309 } 310 try testing.expectError(error.Busy, r.finishScope(0)); 311 try r.abortAcquire(admissions[0]); 312 try testing.expectError(error.InvalidAdmission, r.abortAcquire(admissions[0])); 313 try closeScope(&r, 0); 314 } 315 test "ID exhaustion is fail closed without mutation or handle reuse" { 316 var r = try TestRegistry.init(testing.allocator, 1, .{ 0, 0, 0 }); 317 defer r.deinit(); 318 const a = try r.beginAcquire(0); 319 r.next = std.math.maxInt(u64); 320 try testing.expectError(error.IdExhausted, r.publish(a, 0, .{})); 321 try testing.expectError(error.IdExhausted, r.reserveChild(0, &.{})); 322 try testing.expectError(error.IdExhausted, r.beginAcquire(0)); 323 try testing.expectError(error.IdExhausted, r.beginLease(0, .{ .run = 1, .key = 1 }, .read, false)); 324 try testing.expectEqual(@as(u32, 1), r.admissions.count()); 325 try r.abortAcquire(a); 326 try closeScope(&r, 0); 327 } 328 fn allocationProbe(allocator: std.mem.Allocator) !void { 329 var r = try TestRegistry.init(allocator, 1, .{ 0, 0, 0 }); 330 defer r.deinit(); 331 const token = try create(&r, 0, 5, .{}); 332 // Fill the scope map to its load limit so the tested transfer must grow it. 333 while (r.scopes.count() < r.scopes.capacity() * 80 / 100) _ = try r.reserveChild(0, &.{}); 334 const scope_count = r.scopes.count(); 335 const count = r.resources.count(); 336 const next = r.next; 337 const child = r.reserveChild(0, &.{token}) catch |err| { 338 try testing.expectEqual(next, r.next); 339 try testing.expectEqual(scope_count, r.scopes.count()); 340 try testing.expectEqual(count, r.resources.count()); 341 try testing.expectEqual(@as(u32, 5), (try r.lookup(0, token)).backend); 342 return err; 343 }; 344 const lease = try r.beginLease(child.scope, child.tokens[0], .read, false); 345 try r.endLease(lease.token); 346 try closeScope(&r, child.scope); 347 while (r.scopes.count() > 1) { 348 var it = r.scopes.keyIterator(); 349 while (it.next()) |scope| { 350 if (scope.* == 0) continue; 351 try closeScope(&r, scope.*); 352 break; 353 } 354 } 355 try closeScope(&r, 0); 356 } 357 test "allocator failure never partially transfers resources" { 358 try testing.checkAllAllocationFailures(testing.allocator, allocationProbe, .{}); 359 } 360 test "repeated scopes, acquisitions and leases retain no historical entries" { 361 var r = try TestRegistry.init(testing.allocator, 1, .{ 0, 0, 0 }); 362 defer r.deinit(); 363 for (0..4096) |_| { 364 const child = (try r.reserveChild(0, &.{})).scope; 365 const token = try create(&r, child, 0, .{}); 366 const lease = try r.beginLease(child, token, .exclusive, false); 367 try r.endLease(lease.token); 368 try closeScope(&r, child); 369 try testing.expectEqual(@as(u32, 1), r.scopes.count()); 370 try testing.expectEqual(@as(u32, 3), r.resources.count()); 371 try testing.expectEqual(@as(u32, 0), r.admissions.count()); 372 try testing.expectEqual(@as(u32, 0), r.leases.count()); 373 } 374 try closeScope(&r, 0); 375 }