io_driver.zig (30013B)
1 const std = @import("std"); 2 const Arena = @import("arena.zig").Arena; 3 const codecs = @import("codecs.zig"); 4 const reduce = @import("reduce.zig"); 5 const tree = @import("tree.zig"); 6 7 const c = @cImport({ 8 @cInclude("uv.h"); 9 }); 10 11 // --------------------------------------------------------------------------- 12 // Action tag constants (must match lib/io.tri and IODriver.hs) 13 // --------------------------------------------------------------------------- 14 15 pub const ActionTag = enum(u8) { 16 pure = 0, 17 bind = 1, 18 putStr = 10, 19 getLine = 11, 20 readFile = 20, 21 writeFile = 21, 22 ask = 30, 23 local = 31, 24 get = 40, 25 put = 41, 26 fork = 60, 27 await = 61, 28 yield = 62, 29 sleep = 63, 30 }; 31 32 pub const Action = union(ActionTag) { 33 pure: u32, 34 bind: struct { left: u32, k: u32 }, 35 putStr: u32, 36 getLine, 37 readFile: u32, 38 writeFile: struct { path: u32, contents: u32 }, 39 ask, 40 local: struct { f: u32, action: u32 }, 41 get, 42 put: u32, 43 fork: u32, 44 await: u32, 45 yield, 46 sleep: u32, 47 }; 48 49 // --------------------------------------------------------------------------- 50 // Error codes (must match IODriver.hs) 51 // --------------------------------------------------------------------------- 52 53 const ERR_DOES_NOT_EXIST: u64 = 1; 54 const ERR_PERMISSION: u64 = 2; 55 const ERR_ALREADY_EXISTS: u64 = 3; 56 const ERR_IO_OTHER: u64 = 4; 57 const ERR_POLICY_DENY: u64 = 20; 58 const ERR_INVALID_ACTION: u64 = 40; 59 const ERR_INVALID_STRING: u64 = 41; 60 61 // --------------------------------------------------------------------------- 62 // Permissions 63 // --------------------------------------------------------------------------- 64 65 pub const IOPerms = struct { 66 allow_read_all: bool = false, 67 allow_write_all: bool = false, 68 }; 69 70 // --------------------------------------------------------------------------- 71 // IO sentinel detection 72 // --------------------------------------------------------------------------- 73 74 pub fn isIOSentinel(arena: *Arena, root: u32) !?u32 { 75 const node = arena.get(root); 76 if (node.* != .fork) return null; 77 78 const sentinel = node.fork.left; 79 const rest = node.fork.right; 80 81 const sentinel_str = try codecs.toString(arena, sentinel); 82 defer { 83 if (sentinel_str) |s| { 84 arena.allocator.free(s); 85 } 86 } 87 if (sentinel_str == null) return null; 88 if (!std.mem.eql(u8, sentinel_str.?, "tricuIO")) return null; 89 90 const rest_node = arena.get(rest); 91 if (rest_node.* != .fork) return null; 92 93 const version_num = try codecs.toNumber(arena, rest_node.fork.left); 94 if (version_num == null or version_num.? != 1) return null; 95 96 return rest_node.fork.right; 97 } 98 99 // --------------------------------------------------------------------------- 100 // Action decoding 101 // --------------------------------------------------------------------------- 102 103 pub fn decodeAction(arena: *Arena, root: u32) !?Action { 104 const node = arena.get(root); 105 if (node.* != .fork) return null; 106 107 const tag_num = try codecs.toNumber(arena, node.fork.left); 108 if (tag_num == null) return null; 109 110 const tag: ActionTag = switch (tag_num.?) { 111 0 => .pure, 112 1 => .bind, 113 10 => .putStr, 114 11 => .getLine, 115 20 => .readFile, 116 21 => .writeFile, 117 30 => .ask, 118 31 => .local, 119 40 => .get, 120 41 => .put, 121 60 => .fork, 122 61 => .await, 123 62 => .yield, 124 63 => .sleep, 125 else => return null, 126 }; 127 128 const payload = node.fork.right; 129 130 return switch (tag) { 131 .pure => Action{ .pure = payload }, 132 .bind => blk: { 133 const payload_node = arena.get(payload); 134 if (payload_node.* != .fork) return null; 135 break :blk Action{ .bind = .{ .left = payload_node.fork.left, .k = payload_node.fork.right } }; 136 }, 137 .putStr => Action{ .putStr = payload }, 138 .getLine => Action.getLine, 139 .readFile => Action{ .readFile = payload }, 140 .writeFile => blk: { 141 const payload_node = arena.get(payload); 142 if (payload_node.* != .fork) return null; 143 break :blk Action{ .writeFile = .{ .path = payload_node.fork.left, .contents = payload_node.fork.right } }; 144 }, 145 .ask => Action.ask, 146 .local => blk: { 147 const payload_node = arena.get(payload); 148 if (payload_node.* != .fork) return null; 149 break :blk Action{ .local = .{ .f = payload_node.fork.left, .action = payload_node.fork.right } }; 150 }, 151 .get => Action.get, 152 .put => Action{ .put = payload }, 153 .fork => Action{ .fork = payload }, 154 .await => Action{ .await = payload }, 155 .yield => Action.yield, 156 .sleep => Action{ .sleep = payload }, 157 }; 158 } 159 160 // --------------------------------------------------------------------------- 161 // Response tree constructors 162 // --------------------------------------------------------------------------- 163 164 pub fn makePure(arena: *Arena, val: u32) !u32 { 165 const tag = try codecs.ofNumber(arena, 0); 166 return try arena.alloc(.{ .fork = .{ .left = tag, .right = val } }); 167 } 168 169 pub fn makeOkResult(arena: *Arena, val: u32) !u32 { 170 const ok_tag = try arena.alloc(.{ .stem = .{ .child = try arena.alloc(.leaf) } }); 171 const val_pair = try arena.alloc(.{ .fork = .{ .left = val, .right = try arena.alloc(.leaf) } }); 172 return try arena.alloc(.{ .fork = .{ .left = ok_tag, .right = val_pair } }); 173 } 174 175 pub fn makeErrResult(arena: *Arena, code: u64) !u32 { 176 const code_tree = try codecs.ofNumber(arena, code); 177 const code_pair = try arena.alloc(.{ .fork = .{ .left = code_tree, .right = try arena.alloc(.leaf) } }); 178 return try arena.alloc(.{ .fork = .{ .left = try arena.alloc(.leaf), .right = code_pair } }); 179 } 180 181 // --------------------------------------------------------------------------- 182 // Frame stack and runtime 183 // --------------------------------------------------------------------------- 184 185 const Frame = union(enum) { 186 bind: u32, // continuation k 187 local: u32, // old env 188 }; 189 190 const Runtime = struct { 191 env: u32, 192 state: u32, 193 }; 194 195 // --------------------------------------------------------------------------- 196 // Helper: reduce a term in a scratch arena and copy the result back 197 // --------------------------------------------------------------------------- 198 199 fn reduceInScratch(gpa: std.mem.Allocator, arena: *Arena, term: u32) !u32 { 200 var scratch = Arena.init(gpa); 201 defer scratch.deinit(); 202 const scratch_root = try tree.copyTree(arena.nodes.items, &scratch, term); 203 const scratch_result = try reduce.reduce(scratch_root, &scratch, std.math.maxInt(u64)); 204 return try tree.copyTree(scratch.nodes.items, arena, scratch_result); 205 } 206 207 // --------------------------------------------------------------------------- 208 // Task 209 // --------------------------------------------------------------------------- 210 211 const Task = struct { 212 id: u64, 213 parent: ?*Task, 214 frames: std.ArrayList(Frame), 215 runtime: Runtime, 216 current: u32, 217 status: enum { runnable, blocked, completed }, 218 result: ?u32, 219 waiting_for: ?u64, 220 221 fn init(gpa: std.mem.Allocator, id: u64, parent: ?*Task, env: u32, state: u32, current: u32) !*Task { 222 const task = try gpa.create(Task); 223 task.* = .{ 224 .id = id, 225 .parent = parent, 226 .frames = std.ArrayList(Frame).empty, 227 .runtime = .{ .env = env, .state = state }, 228 .current = current, 229 .status = .runnable, 230 .result = null, 231 .waiting_for = null, 232 }; 233 return task; 234 } 235 236 fn deinit(self: *Task, gpa: std.mem.Allocator) void { 237 self.frames.deinit(gpa); 238 gpa.destroy(self); 239 } 240 241 // finishValue processes a value through the frame stack. 242 // Returns true if the task has completed (no more frames). 243 fn finishValue(self: *Task, arena: *Arena, value: u32) !bool { 244 if (self.frames.pop()) |frame| { 245 switch (frame) { 246 .bind => |k| { 247 self.current = try arena.alloc(.{ .app = .{ .func = k, .arg = value } }); 248 return false; 249 }, 250 .local => |old_env| { 251 self.runtime.env = old_env; 252 self.current = try makePure(arena, value); 253 return false; 254 }, 255 } 256 } else { 257 self.current = value; 258 return true; 259 } 260 } 261 }; 262 263 // --------------------------------------------------------------------------- 264 // Scheduler 265 // --------------------------------------------------------------------------- 266 267 const Scheduler = struct { 268 gpa: std.mem.Allocator, 269 loop: *c.uv_loop_t, 270 arena: *Arena, 271 tasks: std.ArrayList(*Task), 272 runnable: std.ArrayList(*Task), 273 next_id: u64, 274 perms: IOPerms, 275 276 fn init(gpa: std.mem.Allocator, loop: *c.uv_loop_t, arena: *Arena, perms: IOPerms) !Scheduler { 277 const sched = Scheduler{ 278 .gpa = gpa, 279 .loop = loop, 280 .arena = arena, 281 .tasks = std.ArrayList(*Task).empty, 282 .runnable = std.ArrayList(*Task).empty, 283 .next_id = 1, 284 .perms = perms, 285 }; 286 return sched; 287 } 288 289 fn deinit(self: *Scheduler) void { 290 for (self.tasks.items) |task| { 291 task.deinit(self.gpa); 292 } 293 self.tasks.deinit(self.gpa); 294 self.runnable.deinit(self.gpa); 295 } 296 297 fn createTask(self: *Scheduler, parent: ?*Task, env: u32, state: u32, current: u32) !*Task { 298 const id = self.next_id; 299 self.next_id += 1; 300 const task = try Task.init(self.gpa, id, parent, env, state, current); 301 try self.tasks.append(self.gpa, task); 302 return task; 303 } 304 305 fn run(self: *Scheduler) !void { 306 while (true) { 307 if (self.runnable.items.len > 0) { 308 const task = self.runnable.orderedRemove(0); 309 try self.stepTask(task); 310 } else if (self.hasPendingHandles()) { 311 _ = c.uv_run(self.loop, c.UV_RUN_ONCE); 312 } else { 313 break; 314 } 315 } 316 } 317 318 fn hasPendingHandles(self: *Scheduler) bool { 319 return c.uv_loop_alive(self.loop) != 0; 320 } 321 322 fn completeTask(self: *Scheduler, task: *Task) !void { 323 task.status = .completed; 324 task.result = task.current; 325 // Unblock any tasks waiting for this one 326 for (self.tasks.items) |t| { 327 if (t.status == .blocked and t.waiting_for == task.id) { 328 t.status = .runnable; 329 t.waiting_for = null; 330 t.current = try makePure(self.arena, task.result.?); 331 try self.runnable.append(self.gpa, t); 332 } 333 } 334 } 335 336 fn stepTask(self: *Scheduler, task: *Task) !void { 337 const reduced = try reduceInScratch(self.gpa, self.arena, task.current); 338 339 const decoded = try decodeAction(self.arena, reduced); 340 if (decoded == null) { 341 // Not a recognized action — if no frames, it's the final result. 342 // Otherwise treat as invalid. 343 if (task.frames.items.len == 0) { 344 task.current = reduced; 345 try self.completeTask(task); 346 return; 347 } 348 const err = try makeErrResult(self.arena, ERR_INVALID_ACTION); 349 if (try task.finishValue(self.arena, err)) { 350 try self.completeTask(task); 351 } else { 352 try self.runnable.append(self.gpa, task); 353 } 354 return; 355 } 356 357 switch (decoded.?) { 358 .pure => |val| { 359 if (try task.finishValue(self.arena, val)) { 360 try self.completeTask(task); 361 } else { 362 try self.runnable.append(self.gpa, task); 363 } 364 }, 365 366 .bind => |b| { 367 try task.frames.append(self.gpa, .{ .bind = b.k }); 368 task.current = b.left; 369 try self.runnable.append(self.gpa, task); 370 }, 371 372 .putStr => |str_tree| { 373 const str = try codecs.toString(self.arena, str_tree) orelse { 374 const err = try makeErrResult(self.arena, ERR_INVALID_STRING); 375 if (try task.finishValue(self.arena, err)) { 376 try self.completeTask(task); 377 } else { 378 try self.runnable.append(self.gpa, task); 379 } 380 return; 381 }; 382 defer self.gpa.free(str); 383 _ = std.c.write(1, str.ptr, str.len); 384 const leaf = try self.arena.alloc(.leaf); 385 if (try task.finishValue(self.arena, leaf)) { 386 try self.completeTask(task); 387 } else { 388 try self.runnable.append(self.gpa, task); 389 } 390 }, 391 392 .getLine => { 393 var buf: [4096]u8 = undefined; 394 var len: usize = 0; 395 while (len < buf.len) { 396 const n = std.c.read(0, buf[len..].ptr, 1); 397 if (n <= 0) break; 398 if (buf[len] == '\n') break; 399 len += 1; 400 } 401 const line = buf[0..len]; 402 const str_tree = try codecs.ofString(self.arena, line); 403 if (try task.finishValue(self.arena, str_tree)) { 404 try self.completeTask(task); 405 } else { 406 try self.runnable.append(self.gpa, task); 407 } 408 }, 409 410 .readFile => |path_tree| { 411 const path = try codecs.toString(self.arena, path_tree) orelse { 412 const err = try makeErrResult(self.arena, ERR_INVALID_STRING); 413 if (try task.finishValue(self.arena, err)) { 414 try self.completeTask(task); 415 } else { 416 try self.runnable.append(self.gpa, task); 417 } 418 return; 419 }; 420 421 if (!self.perms.allow_read_all) { 422 self.arena.allocator.free(path); 423 const err = try makeErrResult(self.arena, ERR_POLICY_DENY); 424 if (try task.finishValue(self.arena, err)) { 425 try self.completeTask(task); 426 } else { 427 try self.runnable.append(self.gpa, task); 428 } 429 return; 430 } 431 432 const ctx = try self.gpa.create(FileReadCtx); 433 ctx.* = .{ 434 .scheduler = self, 435 .task = task, 436 .arena = self.arena, 437 .gpa = self.gpa, 438 .fd = -1, 439 .buf = std.ArrayList(u8).empty, 440 .path = path, 441 .req = undefined, 442 .read_buf = null, 443 }; 444 ctx.req.data = ctx; 445 _ = c.uv_fs_open(self.loop, &ctx.req, ctx.path.ptr, c.O_RDONLY, 0, file_open_cb); 446 }, 447 448 .writeFile => |wf| { 449 const path = try codecs.toString(self.arena, wf.path) orelse { 450 const err = try makeErrResult(self.arena, ERR_INVALID_STRING); 451 if (try task.finishValue(self.arena, err)) { 452 try self.completeTask(task); 453 } else { 454 try self.runnable.append(self.gpa, task); 455 } 456 return; 457 }; 458 459 const contents = try codecs.toString(self.arena, wf.contents) orelse { 460 self.arena.allocator.free(path); 461 const err = try makeErrResult(self.arena, ERR_INVALID_STRING); 462 if (try task.finishValue(self.arena, err)) { 463 try self.completeTask(task); 464 } else { 465 try self.runnable.append(self.gpa, task); 466 } 467 return; 468 }; 469 470 if (!self.perms.allow_write_all) { 471 self.arena.allocator.free(path); 472 self.arena.allocator.free(contents); 473 const err = try makeErrResult(self.arena, ERR_POLICY_DENY); 474 if (try task.finishValue(self.arena, err)) { 475 try self.completeTask(task); 476 } else { 477 try self.runnable.append(self.gpa, task); 478 } 479 return; 480 } 481 482 const ctx = try self.gpa.create(FileWriteCtx); 483 ctx.* = .{ 484 .scheduler = self, 485 .task = task, 486 .arena = self.arena, 487 .gpa = self.gpa, 488 .fd = -1, 489 .path = path, 490 .contents = contents, 491 .written = false, 492 .req = undefined, 493 }; 494 ctx.req.data = ctx; 495 const flags = c.O_WRONLY | c.O_CREAT | c.O_TRUNC; 496 _ = c.uv_fs_open(self.loop, &ctx.req, ctx.path.ptr, flags, 0o644, file_write_open_cb); 497 }, 498 499 .ask => { 500 if (try task.finishValue(self.arena, task.runtime.env)) { 501 try self.completeTask(task); 502 } else { 503 try self.runnable.append(self.gpa, task); 504 } 505 }, 506 507 .local => |loc| { 508 const new_env = try reduceInScratch(self.gpa, self.arena, try self.arena.alloc(.{ .app = .{ .func = loc.f, .arg = task.runtime.env } })); 509 try task.frames.append(self.gpa, .{ .local = task.runtime.env }); 510 task.runtime.env = new_env; 511 task.current = loc.action; 512 try self.runnable.append(self.gpa, task); 513 }, 514 515 .get => { 516 if (try task.finishValue(self.arena, task.runtime.state)) { 517 try self.completeTask(task); 518 } else { 519 try self.runnable.append(self.gpa, task); 520 } 521 }, 522 523 .put => |new_state| { 524 task.runtime.state = new_state; 525 const leaf = try self.arena.alloc(.leaf); 526 if (try task.finishValue(self.arena, leaf)) { 527 try self.completeTask(task); 528 } else { 529 try self.runnable.append(self.gpa, task); 530 } 531 }, 532 533 .fork => |action| { 534 const child = try self.createTask(task, task.runtime.env, task.runtime.state, action); 535 try self.runnable.append(self.gpa, child); 536 const handle = try codecs.ofNumber(self.arena, child.id); 537 if (try task.finishValue(self.arena, handle)) { 538 try self.completeTask(task); 539 } else { 540 try self.runnable.append(self.gpa, task); 541 } 542 }, 543 544 .await => |handle_tree| { 545 const handle = try codecs.toNumber(self.arena, handle_tree) orelse { 546 const err = try makeErrResult(self.arena, ERR_INVALID_ACTION); 547 if (try task.finishValue(self.arena, err)) { 548 try self.completeTask(task); 549 } else { 550 try self.runnable.append(self.gpa, task); 551 } 552 return; 553 }; 554 var found: ?*Task = null; 555 for (self.tasks.items) |t| { 556 if (t.id == handle) { 557 found = t; 558 break; 559 } 560 } 561 if (found == null) { 562 const err = try makeErrResult(self.arena, ERR_INVALID_ACTION); 563 if (try task.finishValue(self.arena, err)) { 564 try self.completeTask(task); 565 } else { 566 try self.runnable.append(self.gpa, task); 567 } 568 return; 569 } 570 if (found.?.status == .completed) { 571 const result = found.?.result.?; 572 if (try task.finishValue(self.arena, result)) { 573 try self.completeTask(task); 574 } else { 575 try self.runnable.append(self.gpa, task); 576 } 577 } else { 578 task.status = .blocked; 579 task.waiting_for = handle; 580 // Task remains out of runnable until child completes 581 } 582 }, 583 584 .yield => { 585 const leaf = try self.arena.alloc(.leaf); 586 if (try task.finishValue(self.arena, leaf)) { 587 try self.completeTask(task); 588 } else { 589 try self.runnable.append(self.gpa, task); 590 } 591 }, 592 593 .sleep => |ms_tree| { 594 const ms = try codecs.toNumber(self.arena, ms_tree) orelse 0; 595 const ctx = try self.gpa.create(SleepCtx); 596 ctx.* = .{ 597 .scheduler = self, 598 .task = task, 599 .arena = self.arena, 600 .timer = undefined, 601 }; 602 ctx.timer.data = ctx; 603 _ = c.uv_timer_init(self.loop, &ctx.timer); 604 _ = c.uv_timer_start(&ctx.timer, sleep_cb, @intCast(ms), 0); 605 }, 606 } 607 } 608 }; 609 610 // --------------------------------------------------------------------------- 611 // Async file read 612 // --------------------------------------------------------------------------- 613 614 const FileReadCtx = struct { 615 scheduler: *Scheduler, 616 task: *Task, 617 arena: *Arena, 618 gpa: std.mem.Allocator, 619 fd: c_int, 620 buf: std.ArrayList(u8), 621 path: []const u8, 622 req: c.uv_fs_t, 623 read_buf: ?[]u8, 624 }; 625 626 fn mapUvErr(uv_err: c_int) u64 { 627 return switch (uv_err) { 628 c.UV_ENOENT => ERR_DOES_NOT_EXIST, 629 c.UV_EACCES => ERR_PERMISSION, 630 c.UV_EEXIST => ERR_ALREADY_EXISTS, 631 else => ERR_IO_OTHER, 632 }; 633 } 634 635 fn file_open_cb(req: [*c]c.uv_fs_t) callconv(.c) void { 636 const ctx = @as(*FileReadCtx, @ptrCast(@alignCast(req.*.data))); 637 const result = req.*.result; 638 c.uv_fs_req_cleanup(req); 639 if (result < 0) { 640 const err = makeErrResult(ctx.arena, mapUvErr(@intCast(-result))) catch { 641 ctx.gpa.destroy(ctx); 642 return; 643 }; 644 if (ctx.task.finishValue(ctx.arena, err) catch false) { 645 ctx.scheduler.completeTask(ctx.task) catch {}; 646 } else { 647 ctx.scheduler.runnable.append(ctx.scheduler.gpa, ctx.task) catch {}; 648 } 649 ctx.buf.deinit(ctx.gpa); 650 ctx.gpa.free(ctx.path); 651 ctx.gpa.destroy(ctx); 652 return; 653 } 654 ctx.fd = @intCast(result); 655 const read_buf = ctx.gpa.alloc(u8, 4096) catch unreachable; 656 ctx.read_buf = read_buf; 657 var uv_buf = c.uv_buf_init(@ptrCast(read_buf.ptr), @intCast(read_buf.len)); 658 _ = c.uv_fs_read(ctx.scheduler.loop, req, ctx.fd, &uv_buf, 1, -1, file_read_cb); 659 } 660 661 fn file_read_cb(req: [*c]c.uv_fs_t) callconv(.c) void { 662 const ctx = @as(*FileReadCtx, @ptrCast(@alignCast(req.*.data))); 663 const nread = req.*.result; 664 c.uv_fs_req_cleanup(req); 665 if (nread < 0) { 666 _ = c.uv_fs_close(ctx.scheduler.loop, req, ctx.fd, null); 667 const err = makeErrResult(ctx.arena, mapUvErr(@intCast(-nread))) catch { 668 ctx.gpa.destroy(ctx); 669 return; 670 }; 671 if (ctx.task.finishValue(ctx.arena, err) catch false) { 672 ctx.scheduler.completeTask(ctx.task) catch {}; 673 } else { 674 ctx.scheduler.runnable.append(ctx.scheduler.gpa, ctx.task) catch {}; 675 } 676 if (ctx.read_buf) |b| ctx.gpa.free(b); 677 ctx.buf.deinit(ctx.gpa); 678 ctx.gpa.free(ctx.path); 679 ctx.gpa.destroy(ctx); 680 return; 681 } 682 if (nread == 0) { 683 // EOF 684 _ = c.uv_fs_close(ctx.scheduler.loop, req, ctx.fd, null); 685 const bytes_tree = codecs.ofBytes(ctx.arena, ctx.buf.items) catch { 686 ctx.gpa.destroy(ctx); 687 return; 688 }; 689 const ok = makeOkResult(ctx.arena, bytes_tree) catch { 690 ctx.gpa.destroy(ctx); 691 return; 692 }; 693 if (ctx.task.finishValue(ctx.arena, ok) catch false) { 694 ctx.scheduler.completeTask(ctx.task) catch {}; 695 } else { 696 ctx.scheduler.runnable.append(ctx.scheduler.gpa, ctx.task) catch {}; 697 } 698 if (ctx.read_buf) |b| ctx.gpa.free(b); 699 ctx.buf.deinit(ctx.gpa); 700 ctx.gpa.free(ctx.path); 701 ctx.gpa.destroy(ctx); 702 return; 703 } 704 const data = ctx.read_buf.?[0..@intCast(nread)]; 705 ctx.buf.appendSlice(ctx.gpa, data) catch unreachable; 706 const read_buf = ctx.gpa.alloc(u8, 4096) catch unreachable; 707 ctx.read_buf = read_buf; 708 var uv_buf = c.uv_buf_init(@ptrCast(read_buf.ptr), @intCast(read_buf.len)); 709 _ = c.uv_fs_read(ctx.scheduler.loop, req, ctx.fd, &uv_buf, 1, -1, file_read_cb); 710 } 711 712 // --------------------------------------------------------------------------- 713 // Async file write 714 // --------------------------------------------------------------------------- 715 716 const FileWriteCtx = struct { 717 scheduler: *Scheduler, 718 task: *Task, 719 arena: *Arena, 720 gpa: std.mem.Allocator, 721 fd: c_int, 722 path: []const u8, 723 contents: []const u8, 724 written: bool, 725 req: c.uv_fs_t, 726 }; 727 728 fn file_write_open_cb(req: [*c]c.uv_fs_t) callconv(.c) void { 729 const ctx = @as(*FileWriteCtx, @ptrCast(@alignCast(req.*.data))); 730 const result = req.*.result; 731 c.uv_fs_req_cleanup(req); 732 if (result < 0) { 733 const err = makeErrResult(ctx.arena, mapUvErr(@intCast(-result))) catch { 734 ctx.gpa.destroy(ctx); 735 return; 736 }; 737 if (ctx.task.finishValue(ctx.arena, err) catch false) { 738 ctx.scheduler.completeTask(ctx.task) catch {}; 739 } else { 740 ctx.scheduler.runnable.append(ctx.scheduler.gpa, ctx.task) catch {}; 741 } 742 ctx.gpa.free(ctx.path); 743 ctx.gpa.free(ctx.contents); 744 ctx.gpa.destroy(ctx); 745 return; 746 } 747 ctx.fd = @intCast(result); 748 var uv_buf = c.uv_buf_init(@ptrCast(@constCast(ctx.contents.ptr)), @intCast(ctx.contents.len)); 749 _ = c.uv_fs_write(ctx.scheduler.loop, req, ctx.fd, &uv_buf, 1, 0, file_write_cb); 750 } 751 752 fn file_write_cb(req: [*c]c.uv_fs_t) callconv(.c) void { 753 const ctx = @as(*FileWriteCtx, @ptrCast(@alignCast(req.*.data))); 754 const nwrite = req.*.result; 755 c.uv_fs_req_cleanup(req); 756 if (nwrite < 0) { 757 _ = c.uv_fs_close(ctx.scheduler.loop, req, ctx.fd, null); 758 const err = makeErrResult(ctx.arena, mapUvErr(@intCast(-nwrite))) catch { 759 ctx.gpa.destroy(ctx); 760 return; 761 }; 762 if (ctx.task.finishValue(ctx.arena, err) catch false) { 763 ctx.scheduler.completeTask(ctx.task) catch {}; 764 } else { 765 ctx.scheduler.runnable.append(ctx.scheduler.gpa, ctx.task) catch {}; 766 } 767 ctx.gpa.free(ctx.path); 768 ctx.gpa.free(ctx.contents); 769 ctx.gpa.destroy(ctx); 770 return; 771 } 772 _ = c.uv_fs_close(ctx.scheduler.loop, req, ctx.fd, file_write_close_cb); 773 } 774 775 fn file_write_close_cb(req: [*c]c.uv_fs_t) callconv(.c) void { 776 const ctx = @as(*FileWriteCtx, @ptrCast(@alignCast(req.*.data))); 777 c.uv_fs_req_cleanup(req); 778 const leaf = ctx.arena.alloc(.leaf) catch { 779 ctx.gpa.destroy(ctx); 780 return; 781 }; 782 const ok = makeOkResult(ctx.arena, leaf) catch { 783 ctx.gpa.destroy(ctx); 784 return; 785 }; 786 if (ctx.task.finishValue(ctx.arena, ok) catch false) { 787 ctx.scheduler.completeTask(ctx.task) catch {}; 788 } else { 789 ctx.scheduler.runnable.append(ctx.scheduler.gpa, ctx.task) catch {}; 790 } 791 ctx.gpa.free(ctx.path); 792 ctx.gpa.free(ctx.contents); 793 ctx.gpa.destroy(ctx); 794 } 795 796 // --------------------------------------------------------------------------- 797 // Async sleep 798 // --------------------------------------------------------------------------- 799 800 const SleepCtx = struct { 801 scheduler: *Scheduler, 802 task: *Task, 803 arena: *Arena, 804 timer: c.uv_timer_t, 805 }; 806 807 fn sleep_cb(handle: [*c]c.uv_timer_t) callconv(.c) void { 808 const ctx = @as(*SleepCtx, @ptrCast(@alignCast(handle.*.data))); 809 defer ctx.scheduler.gpa.destroy(ctx); 810 const leaf = ctx.arena.alloc(.leaf) catch { 811 ctx.scheduler.runnable.append(ctx.scheduler.gpa, ctx.task) catch {}; 812 return; 813 }; 814 if (ctx.task.finishValue(ctx.arena, leaf) catch false) { 815 ctx.scheduler.completeTask(ctx.task) catch {}; 816 } else { 817 ctx.scheduler.runnable.append(ctx.scheduler.gpa, ctx.task) catch {}; 818 } 819 } 820 821 // --------------------------------------------------------------------------- 822 // Public entry point 823 // --------------------------------------------------------------------------- 824 825 pub fn runIO(gpa: std.mem.Allocator, arena: *Arena, program: u32, perms: IOPerms) !u32 { 826 const action_tree = try isIOSentinel(arena, program) orelse { 827 return error.InvalidIOSentinel; 828 }; 829 830 var loop: c.uv_loop_t = undefined; 831 const rc = c.uv_loop_init(&loop); 832 if (rc != 0) return error.LoopInitFailed; 833 defer _ = c.uv_loop_close(&loop); 834 835 var scheduler = try Scheduler.init(gpa, &loop, arena, perms); 836 defer scheduler.deinit(); 837 838 const main_task = try scheduler.createTask(null, try arena.alloc(.leaf), try arena.alloc(.leaf), action_tree); 839 try scheduler.runnable.append(gpa, main_task); 840 841 try scheduler.run(); 842 843 // Return the main task's result 844 return main_task.result orelse program; 845 }