authorgravatar for jacobly@ziglang.orgJacob Young <jacobly@ziglang.org> 2025-03-29 02:31:27-04:00
committergravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2025-10-02 16:30:59-07:00
log507f973b5eb3e42d0e8dd151539c929d3b91fe08
treed904e27ed5f8763d9fe59a16c62e0630eb48295f
parentb215ddc9fbb6a396a7ec19e1ee988aa400b887a0

EventLoop: get file operations working

Something is horribly wrong with scheduling, as can be seen in the debug output, but at least it somehow manages to exit cleanly...

1 files changed, 300 insertions(+), 108 deletions(-)

lib/std/Io/EventLoop.zig+300-108
......@@ -9,26 +9,25 @@ const IoUring = std.os.linux.IoUring;
99
1010gpa: Allocator,
1111mutex: std.Thread.Mutex,
12cond: std.Thread.Condition,
1312queue: std.DoublyLinkedList(void),
13/// Atomic copy of queue.len
14queue_len: usize,
1415free: std.DoublyLinkedList(void),
15main_context: Context,
16exit_awaiter: ?*Fiber,
16main_fiber: Fiber,
17idle_count: usize,
1718threads: std.ArrayListUnmanaged(Thread),
18/// 1 bit per thread, same order as `thread_index`.
19idle_iourings: []usize,
19exiting: bool,
2020
2121threadlocal var thread_index: u32 = undefined;
2222
2323/// Empirically saw 10KB being used by the self-hosted backend for logging.
24const idle_stack_size = 32 * 1024;
24const idle_stack_size = 64 * 1024;
2525
2626const io_uring_entries = 64;
2727
2828const Thread = struct {
2929 thread: std.Thread,
3030 idle_context: Context,
31 current_idle_context: *Context,
3231 current_context: *Context,
3332 io_uring: IoUring,
3433
......@@ -103,98 +102,92 @@ pub fn io(el: *EventLoop) Io {
103102}
104103
105104pub fn init(el: *EventLoop, gpa: Allocator) !void {
106 const n_threads: usize = @max((std.Thread.getCpuCount() catch 1), 1);
107 const threads_bytes = n_threads * @sizeOf(Thread);
108 const idle_context_offset = std.mem.alignForward(usize, threads_bytes, @alignOf(Context));
109 const idle_stack_end_offset = std.mem.alignForward(usize, idle_context_offset + idle_stack_size, std.heap.page_size_max);
110 const allocated_slice = try gpa.alignedAlloc(u8, @max(@alignOf(Thread), @alignOf(Context)), idle_stack_end_offset);
105 const threads_size = @max(std.Thread.getCpuCount() catch 1, 1) * @sizeOf(Thread);
106 const idle_stack_end_offset = std.mem.alignForward(usize, threads_size + idle_stack_size, std.heap.page_size_max);
107 const allocated_slice = try gpa.alignedAlloc(u8, @alignOf(Thread), idle_stack_end_offset);
111108 errdefer gpa.free(allocated_slice);
112 const idle_iourings = try gpa.alloc(usize, (n_threads + @bitSizeOf(usize) - 1) / @bitSizeOf(usize));
113 errdefer gpa.free(idle_iourings);
114 @memset(idle_iourings, 0);
115109 el.* = .{
116110 .gpa = gpa,
117111 .mutex = .{},
118 .cond = .{},
119112 .queue = .{},
113 .queue_len = 0,
120114 .free = .{},
121 .main_context = undefined,
122 .exit_awaiter = null,
123 .threads = .initBuffer(@ptrCast(allocated_slice[0..threads_bytes])),
124 .idle_iourings = idle_iourings,
115 .main_fiber = undefined,
116 .idle_count = 0,
117 .threads = .initBuffer(@ptrCast(allocated_slice[0..threads_size])),
118 .exiting = false,
125119 };
120 thread_index = 0;
126121 const main_thread = el.threads.addOneAssumeCapacity();
127122 main_thread.io_uring = try IoUring.init(io_uring_entries, 0);
128 const main_idle_context: *Context = @alignCast(std.mem.bytesAsValue(Context, allocated_slice[idle_context_offset..][0..@sizeOf(Context)]));
129 const idle_stack_end: [*]align(@max(@alignOf(Thread), @alignOf(Context))) usize = @alignCast(@ptrCast(allocated_slice[idle_stack_end_offset..].ptr));
123 const idle_stack_end: [*]usize = @alignCast(@ptrCast(allocated_slice[idle_stack_end_offset..].ptr));
130124 (idle_stack_end - 1)[0..1].* = .{@intFromPtr(el)};
131 main_idle_context.* = .{
125 main_thread.idle_context = .{
132126 .rsp = @intFromPtr(idle_stack_end - 1),
133127 .rbp = 0,
134128 .rip = @intFromPtr(&mainIdleEntry),
135129 };
136 std.log.debug("created main idle {*}", .{main_idle_context});
137 main_thread.current_idle_context = main_idle_context;
138 std.log.debug("created main {*}", .{&el.main_context});
139 main_thread.current_context = &el.main_context;
130 std.log.debug("created main idle {*}", .{&main_thread.idle_context});
131 std.log.debug("created main {*}", .{&el.main_fiber});
132 main_thread.current_context = &el.main_fiber.context;
140133}
141134
142135pub fn deinit(el: *EventLoop) void {
143136 assert(el.queue.len == 0); // pending async
144 el.yield(null, &el.exit_awaiter);
137 el.yield(null, .exit);
145138 while (el.free.pop()) |free_node| {
146139 const free_fiber: *Fiber = @alignCast(@fieldParentPtr("queue_node", free_node));
147140 el.gpa.free(free_fiber.allocatedSlice());
148141 }
149 const idle_context_offset = std.mem.alignForward(usize, el.threads.capacity * @sizeOf(Thread), @alignOf(Context));
150 const idle_stack_end = std.mem.alignForward(usize, idle_context_offset + idle_stack_size, std.heap.page_size_max);
151 const allocated_ptr: [*]align(@max(@alignOf(Thread), @alignOf(Context))) u8 = @alignCast(@ptrCast(el.threads.items.ptr));
142 const idle_stack_end_offset = std.mem.alignForward(usize, el.threads.capacity * @sizeOf(Thread) + idle_stack_size, std.heap.page_size_max);
143 const allocated_ptr: [*]align(@alignOf(Thread)) u8 = @alignCast(@ptrCast(el.threads.items.ptr));
152144 for (el.threads.items[1..]) |*thread| thread.thread.join();
153 el.gpa.free(allocated_ptr[0..idle_stack_end]);
145 el.gpa.free(allocated_ptr[0..idle_stack_end_offset]);
154146}
155147
156const PendingTask = union(enum) {
157 none,
158 register_awaiter: *?*Fiber,
159 io_uring_submit: *IoUring,
160};
161
162fn yield(el: *EventLoop, optional_fiber: ?*Fiber, pending_task: PendingTask) void {
148fn yield(el: *EventLoop, optional_fiber: ?*Fiber, pending_task: SwitchMessage.PendingTask) void {
163149 const thread: *Thread = &el.threads.items[thread_index];
164150 const ready_context: *Context = ready_context: {
165151 const ready_fiber: *Fiber = optional_fiber orelse if (ready_node: {
166152 el.mutex.lock();
167153 defer el.mutex.unlock();
168 break :ready_node el.queue.pop();
154 const ready_node = el.queue.pop();
155 @atomicStore(usize, &el.queue_len, el.queue.len, .unordered);
156 break :ready_node ready_node;
169157 }) |ready_node|
170158 @alignCast(@fieldParentPtr("queue_node", ready_node))
171159 else
172 break :ready_context thread.current_idle_context;
160 break :ready_context &thread.idle_context;
173161 break :ready_context &ready_fiber.context;
174162 };
175163 const message: SwitchMessage = .{
176 .prev_context = thread.current_context,
177 .ready_context = ready_context,
164 .contexts = .{
165 .prev = thread.current_context,
166 .ready = ready_context,
167 },
178168 .pending_task = pending_task,
179169 };
180 std.log.debug("switching from {*} to {*}", .{ message.prev_context, message.ready_context });
170 std.log.debug("switching from {*} to {*}", .{ message.contexts.prev, message.contexts.ready });
181171 contextSwitch(&message).handle(el);
182172}
183173
184174fn schedule(el: *EventLoop, fiber: *Fiber) void {
185 el.mutex.lock();
186 el.queue.append(&fiber.queue_node);
187 //for (el.idle_iourings) |*int| {
188 // const idler_subset = @atomicLoad(usize, int, .unordered);
189 // if (idler_subset == 0) continue;
190 //
191 //}
192 if (el.idle_count > 0) {
193 el.mutex.unlock();
194 el.cond.signal();
175 if (idle_count: {
176 el.mutex.lock();
177 defer el.mutex.unlock();
178 el.queue.append(&fiber.queue_node);
179 @atomicStore(usize, &el.queue_len, el.queue.len, .unordered);
180 break :idle_count el.idle_count;
181 } > 0) {
182 _ = std.os.linux.futex2_wake(&el.queue_len, std.math.maxInt(usize), 1, switch (@bitSizeOf(usize)) {
183 8 => std.os.linux.FUTEX2.SIZE_U8,
184 16 => std.os.linux.FUTEX2.SIZE_U16,
185 32 => std.os.linux.FUTEX2.SIZE_U32,
186 64 => std.os.linux.FUTEX2.SIZE_U64,
187 else => @compileError("unsupported @sizeOf(usize)"),
188 } | std.os.linux.FUTEX2.PRIVATE); // TODO: io_uring
195189 return;
196190 }
197 defer el.mutex.unlock();
198191 if (el.threads.items.len == el.threads.capacity) return;
199192 const thread = el.threads.addOneAssumeCapacity();
200193 thread.thread = std.Thread.spawn(.{
......@@ -216,64 +209,101 @@ fn recycle(el: *EventLoop, fiber: *Fiber) void {
216209
217210fn mainIdle(el: *EventLoop, message: *const SwitchMessage) callconv(.withStackAlign(.c, @max(@alignOf(Thread), @alignOf(Context)))) noreturn {
218211 message.handle(el);
219 el.yield(el.idle(), null);
212 el.idle();
213 el.yield(&el.main_fiber, .nothing);
220214 unreachable; // switched to dead fiber
221215}
222216
223217fn threadEntry(el: *EventLoop, index: usize) void {
224 thread_index = index;
218 thread_index = @intCast(index);
225219 const thread: *Thread = &el.threads.items[index];
226220 std.log.debug("created thread idle {*}", .{&thread.idle_context});
227221 thread.io_uring = IoUring.init(io_uring_entries, 0) catch |err| {
228222 std.log.warn("exiting worker thread during init due to io_uring init failure: {s}", .{@errorName(err)});
229223 return;
230224 };
231 thread.current_idle_context = &thread.idle_context;
232225 thread.current_context = &thread.idle_context;
233 _ = el.idle();
226 el.idle();
234227}
235228
236fn idle(el: *EventLoop) *Fiber {
229const UserData = enum(u64) {
230 queue_len_futex_wait,
231 _,
232};
233
234fn idle(el: *EventLoop) void {
237235 const thread: *Thread = &el.threads.items[thread_index];
238 // The idle fiber only runs on one thread.
239236 const iou = &thread.io_uring;
240237 var cqes_buffer: [io_uring_entries]std.os.linux.io_uring_cqe = undefined;
238 var futex_is_scheduled: bool = false;
241239
242240 while (true) {
243 el.yield(null, null);
244 if (@atomicLoad(?*Fiber, &el.exit_awaiter, .acquire)) |exit_awaiter| {
245 el.cond.broadcast();
246 return exit_awaiter;
247 }
248 // TODO add uring to bit set
249 const n = iou.copy_cqes(&cqes_buffer, 1) catch @panic("TODO handle copy_cqes error");
250 const cqes = cqes_buffer[0..n];
251 for (cqes) |cqe| {
252 const fiber: *Fiber = @ptrFromInt(cqe.user_data);
253 const res: *i32 = @ptrCast(@alignCast(fiber.resultPointer()));
254 res.* = cqe.res;
255 el.schedule(fiber);
241 el.yield(null, .nothing);
242 if (@atomicLoad(bool, &el.exiting, .acquire)) return;
243 if (!futex_is_scheduled) {
244 const sqe = getSqe(&thread.io_uring);
245 sqe.prep_rw(.FUTEX_WAIT, switch (@bitSizeOf(usize)) {
246 8 => std.os.linux.FUTEX2.SIZE_U8,
247 16 => std.os.linux.FUTEX2.SIZE_U16,
248 32 => std.os.linux.FUTEX2.SIZE_U32,
249 64 => std.os.linux.FUTEX2.SIZE_U64,
250 else => @compileError("unsupported @sizeOf(usize)"),
251 } | std.os.linux.FUTEX2.PRIVATE, @intFromPtr(&el.queue_len), 0, 0);
252 sqe.addr3 = std.math.maxInt(u64);
253 sqe.user_data = @intFromEnum(UserData.queue_len_futex_wait);
254 futex_is_scheduled = true;
256255 }
256 _ = iou.submit_and_wait(1) catch |err| switch (err) {
257 error.SignalInterrupt => 0,
258 else => @panic(@errorName(err)),
259 };
260 for (cqes_buffer[0 .. iou.copy_cqes(&cqes_buffer, 1) catch |err| switch (err) {
261 error.SignalInterrupt => 0,
262 else => @panic(@errorName(err)),
263 }]) |cqe| switch (@as(UserData, @enumFromInt(cqe.user_data))) {
264 .queue_len_futex_wait => futex_is_scheduled = false,
265 _ => {
266 const fiber: *Fiber = @ptrFromInt(cqe.user_data);
267 const res: *i32 = @ptrCast(@alignCast(fiber.resultPointer()));
268 res.* = cqe.res;
269 el.schedule(fiber);
270 },
271 };
257272 }
258273}
259274
260const SwitchMessage = extern struct {
261 prev_context: *Context,
262 ready_context: *Context,
275const SwitchMessage = struct {
276 contexts: extern struct {
277 prev: *Context,
278 ready: *Context,
279 },
263280 pending_task: PendingTask,
264281
282 const PendingTask = union(enum) {
283 nothing,
284 register_awaiter: *?*Fiber,
285 exit,
286 };
287
265288 fn handle(message: *const SwitchMessage, el: *EventLoop) void {
266289 const thread: *Thread = &el.threads.items[thread_index];
267 thread.current_context = message.ready_context;
290 thread.current_context = message.contexts.ready;
268291 switch (message.pending_task) {
269 .none => {},
292 .nothing => {},
270293 .register_awaiter => |awaiter| {
271 const prev_fiber: *Fiber = @alignCast(@fieldParentPtr("context", message.prev_context));
294 const prev_fiber: *Fiber = @alignCast(@fieldParentPtr("context", message.contexts.prev));
272295 if (@atomicRmw(?*Fiber, awaiter, .Xchg, prev_fiber, .acq_rel) == Fiber.finished) el.schedule(prev_fiber);
273296 },
274 .io_uring_submit => |iou| {
275 _ = iou.flush_sq();
276 // TODO: determine whether this return value should be used
297 .exit => {
298 @atomicStore(bool, &el.exiting, true, .unordered);
299 @atomicStore(usize, &el.queue_len, std.math.maxInt(usize), .release);
300 _ = std.os.linux.futex2_wake(&el.queue_len, std.math.maxInt(usize), std.math.maxInt(i32), switch (@bitSizeOf(usize)) {
301 8 => std.os.linux.FUTEX2.SIZE_U8,
302 16 => std.os.linux.FUTEX2.SIZE_U16,
303 32 => std.os.linux.FUTEX2.SIZE_U32,
304 64 => std.os.linux.FUTEX2.SIZE_U64,
305 else => @compileError("unsupported @sizeOf(usize)"),
306 } | std.os.linux.FUTEX2.PRIVATE); // TODO: use io_uring
277307 },
278308 }
279309 }
......@@ -289,7 +319,7 @@ const Context = switch (builtin.cpu.arch) {
289319};
290320
291321inline fn contextSwitch(message: *const SwitchMessage) *const SwitchMessage {
292 return switch (builtin.cpu.arch) {
322 return @fieldParentPtr("contexts", switch (builtin.cpu.arch) {
293323 .x86_64 => asm volatile (
294324 \\ movq 0(%%rsi), %%rax
295325 \\ movq 8(%%rsi), %%rcx
......@@ -301,8 +331,8 @@ inline fn contextSwitch(message: *const SwitchMessage) *const SwitchMessage {
301331 \\ movq 8(%%rcx), %%rbp
302332 \\ jmpq *16(%%rcx)
303333 \\0:
304 : [received_message] "={rsi}" (-> *const SwitchMessage),
305 : [message_to_send] "{rsi}" (message),
334 : [received_message] "={rsi}" (-> *const @FieldType(SwitchMessage, "contexts")),
335 : [message_to_send] "{rsi}" (&message.contexts),
306336 : "rax", "rcx", "rdx", "rbx", "rdi", //
307337 "r8", "r9", "r10", "r11", "r12", "r13", "r14", "r15", //
308338 "mm0", "mm1", "mm2", "mm3", "mm4", "mm5", "mm6", "mm7", //
......@@ -313,7 +343,7 @@ inline fn contextSwitch(message: *const SwitchMessage) *const SwitchMessage {
313343 "fpsr", "fpcr", "mxcsr", "rflags", "dirflag", "memory"
314344 ),
315345 else => |arch| @compileError("unimplemented architecture: " ++ @tagName(arch)),
316 };
346 });
317347}
318348
319349fn mainIdleEntry() callconv(.naked) void {
......@@ -401,7 +431,7 @@ const AsyncClosure = struct {
401431 std.log.debug("{*} performing async", .{closure.fiber});
402432 closure.start(closure.contextPointer(), closure.fiber.resultPointer());
403433 const awaiter = @atomicRmw(?*Fiber, &closure.fiber.awaiter, .Xchg, Fiber.finished, .acq_rel);
404 closure.event_loop.yield(awaiter, null);
434 closure.event_loop.yield(awaiter, .nothing);
405435 unreachable; // switched to dead fiber
406436 }
407437};
......@@ -409,17 +439,93 @@ const AsyncClosure = struct {
409439pub fn @"await"(userdata: ?*anyopaque, any_future: *std.Io.AnyFuture, result: []u8) void {
410440 const event_loop: *EventLoop = @alignCast(@ptrCast(userdata));
411441 const future_fiber: *Fiber = @alignCast(@ptrCast(any_future));
412 if (@atomicLoad(?*Fiber, &future_fiber.awaiter, .acquire) != Fiber.finished) event_loop.yield(null, &future_fiber.awaiter);
442 if (@atomicLoad(?*Fiber, &future_fiber.awaiter, .acquire) != Fiber.finished) event_loop.yield(null, .{ .register_awaiter = &future_fiber.awaiter });
413443 @memcpy(result, future_fiber.resultPointer());
414444 event_loop.recycle(future_fiber);
415445}
416446
417447pub fn createFile(userdata: ?*anyopaque, dir: std.fs.Dir, sub_path: []const u8, flags: std.fs.File.CreateFlags) std.fs.File.OpenError!std.fs.File {
418 _ = userdata;
419 _ = dir;
420 _ = sub_path;
421 _ = flags;
422 @panic("TODO");
448 const el: *EventLoop = @ptrCast(@alignCast(userdata));
449
450 const posix = std.posix;
451 const sub_path_c = try posix.toPosixPath(sub_path);
452
453 var os_flags: posix.O = .{
454 .ACCMODE = if (flags.read) .RDWR else .WRONLY,
455 .CREAT = true,
456 .TRUNC = flags.truncate,
457 .EXCL = flags.exclusive,
458 };
459 if (@hasField(posix.O, "LARGEFILE")) os_flags.LARGEFILE = true;
460 if (@hasField(posix.O, "CLOEXEC")) os_flags.CLOEXEC = true;
461
462 // Use the O locking flags if the os supports them to acquire the lock
463 // atomically. Note that the NONBLOCK flag is removed after the openat()
464 // call is successful.
465 const has_flock_open_flags = @hasField(posix.O, "EXLOCK");
466 if (has_flock_open_flags) switch (flags.lock) {
467 .none => {},
468 .shared => {
469 os_flags.SHLOCK = true;
470 os_flags.NONBLOCK = flags.lock_nonblocking;
471 },
472 .exclusive => {
473 os_flags.EXLOCK = true;
474 os_flags.NONBLOCK = flags.lock_nonblocking;
475 },
476 };
477 const have_flock = @TypeOf(posix.system.flock) != void;
478
479 if (have_flock and !has_flock_open_flags and flags.lock != .none) {
480 @panic("TODO");
481 }
482
483 if (has_flock_open_flags and flags.lock_nonblocking) {
484 @panic("TODO");
485 }
486
487 const thread: *Thread = &el.threads.items[thread_index];
488 const iou = &thread.io_uring;
489 const sqe = getSqe(iou);
490 const fiber = thread.currentFiber();
491
492 sqe.prep_openat(dir.fd, &sub_path_c, os_flags, flags.mode);
493 sqe.user_data = @intFromPtr(fiber);
494
495 el.yield(null, .nothing);
496
497 const result: *i32 = @alignCast(@ptrCast(fiber.resultPointer()[0..@sizeOf(posix.fd_t)]));
498 const rc = result.*;
499 switch (errno(rc)) {
500 .SUCCESS => return .{ .handle = rc },
501 .INTR => @panic("TODO is this reachable?"),
502 .CANCELED => @panic("TODO figure out how this error code fits into things"),
503
504 .FAULT => unreachable,
505 .INVAL => return error.BadPathName,
506 .BADF => unreachable,
507 .ACCES => return error.AccessDenied,
508 .FBIG => return error.FileTooBig,
509 .OVERFLOW => return error.FileTooBig,
510 .ISDIR => return error.IsDir,
511 .LOOP => return error.SymLinkLoop,
512 .MFILE => return error.ProcessFdQuotaExceeded,
513 .NAMETOOLONG => return error.NameTooLong,
514 .NFILE => return error.SystemFdQuotaExceeded,
515 .NODEV => return error.NoDevice,
516 .NOENT => return error.FileNotFound,
517 .NOMEM => return error.SystemResources,
518 .NOSPC => return error.NoSpaceLeft,
519 .NOTDIR => return error.NotDir,
520 .PERM => return error.PermissionDenied,
521 .EXIST => return error.PathAlreadyExists,
522 .BUSY => return error.DeviceBusy,
523 .OPNOTSUPP => return error.FileLocksNotSupported,
524 .AGAIN => return error.WouldBlock,
525 .TXTBSY => return error.FileBusy,
526 .NXIO => return error.NoDevice,
527 else => |err| return posix.unexpectedErrno(err),
528 }
423529}
424530
425531pub fn openFile(userdata: ?*anyopaque, dir: std.fs.Dir, sub_path: []const u8, flags: std.fs.File.OpenFlags) std.fs.File.OpenError!std.fs.File {
......@@ -476,7 +582,7 @@ pub fn openFile(userdata: ?*anyopaque, dir: std.fs.Dir, sub_path: []const u8, fl
476582 sqe.prep_openat(dir.fd, &sub_path_c, os_flags, 0);
477583 sqe.user_data = @intFromPtr(fiber);
478584
479 el.yield(null, .{ .io_uring_submit = iou });
585 el.yield(null, .nothing);
480586
481587 const result: *i32 = @alignCast(@ptrCast(fiber.resultPointer()[0..@sizeOf(posix.fd_t)]));
482588 const rc = result.*;
......@@ -510,8 +616,6 @@ pub fn openFile(userdata: ?*anyopaque, dir: std.fs.Dir, sub_path: []const u8, fl
510616 .NXIO => return error.NoDevice,
511617 else => |err| return posix.unexpectedErrno(err),
512618 }
513
514 return .{ .handle = result.* };
515619}
516620
517621fn errno(signed: i32) std.posix.E {
......@@ -524,21 +628,109 @@ fn getSqe(iou: *IoUring) *std.os.linux.io_uring_sqe {
524628}
525629
526630pub fn closeFile(userdata: ?*anyopaque, file: std.fs.File) void {
527 _ = userdata;
528 _ = file;
529 @panic("TODO");
631 const el: *EventLoop = @ptrCast(@alignCast(userdata));
632
633 const posix = std.posix;
634
635 const thread: *Thread = &el.threads.items[thread_index];
636 const iou = &thread.io_uring;
637 const sqe = getSqe(iou);
638 const fiber = thread.currentFiber();
639
640 sqe.prep_close(file.handle);
641 sqe.user_data = @intFromPtr(fiber);
642
643 el.yield(null, .nothing);
644
645 const result: *i32 = @alignCast(@ptrCast(fiber.resultPointer()[0..@sizeOf(posix.fd_t)]));
646 const rc = result.*;
647 switch (errno(rc)) {
648 .SUCCESS => return,
649 .INTR => @panic("TODO is this reachable?"),
650 .CANCELED => @panic("TODO figure out how this error code fits into things"),
651
652 .BADF => unreachable, // Always a race condition.
653 else => return,
654 }
530655}
531656
532657pub fn read(userdata: ?*anyopaque, file: std.fs.File, buffer: []u8) std.fs.File.ReadError!usize {
533 _ = userdata;
534 _ = file;
535 _ = buffer;
536 @panic("TODO");
658 const el: *EventLoop = @ptrCast(@alignCast(userdata));
659
660 const posix = std.posix;
661
662 const thread: *Thread = &el.threads.items[thread_index];
663 const iou = &thread.io_uring;
664 const sqe = getSqe(iou);
665 const fiber = thread.currentFiber();
666
667 sqe.prep_read(file.handle, buffer, std.math.maxInt(u64));
668 sqe.user_data = @intFromPtr(fiber);
669
670 el.yield(null, .nothing);
671
672 const result: *i32 = @alignCast(@ptrCast(fiber.resultPointer()[0..@sizeOf(posix.fd_t)]));
673 const rc = result.*;
674 switch (errno(rc)) {
675 .SUCCESS => return @as(u32, @bitCast(rc)),
676 .INTR => @panic("TODO is this reachable?"),
677 .CANCELED => @panic("TODO figure out how this error code fits into things"),
678
679 .INVAL => unreachable,
680 .FAULT => unreachable,
681 .NOENT => return error.ProcessNotFound,
682 .AGAIN => return error.WouldBlock,
683 .BADF => return error.NotOpenForReading, // Can be a race condition.
684 .IO => return error.InputOutput,
685 .ISDIR => return error.IsDir,
686 .NOBUFS => return error.SystemResources,
687 .NOMEM => return error.SystemResources,
688 .NOTCONN => return error.SocketNotConnected,
689 .CONNRESET => return error.ConnectionResetByPeer,
690 .TIMEDOUT => return error.ConnectionTimedOut,
691 else => |err| return posix.unexpectedErrno(err),
692 }
537693}
538694
539695pub fn write(userdata: ?*anyopaque, file: std.fs.File, buffer: []const u8) std.fs.File.WriteError!usize {
540 _ = userdata;
541 _ = file;
542 _ = buffer;
543 @panic("TODO");
696 const el: *EventLoop = @ptrCast(@alignCast(userdata));
697
698 const posix = std.posix;
699
700 const thread: *Thread = &el.threads.items[thread_index];
701 const iou = &thread.io_uring;
702 const sqe = getSqe(iou);
703 const fiber = thread.currentFiber();
704
705 sqe.prep_write(file.handle, buffer, std.math.maxInt(u64));
706 sqe.user_data = @intFromPtr(fiber);
707
708 el.yield(null, .nothing);
709
710 const result: *i32 = @alignCast(@ptrCast(fiber.resultPointer()[0..@sizeOf(posix.fd_t)]));
711 const rc = result.*;
712 switch (errno(rc)) {
713 .SUCCESS => return @as(u32, @bitCast(rc)),
714 .INTR => @panic("TODO is this reachable?"),
715 .CANCELED => @panic("TODO figure out how this error code fits into things"),
716
717 .INVAL => return error.InvalidArgument,
718 .FAULT => unreachable,
719 .NOENT => return error.ProcessNotFound,
720 .AGAIN => return error.WouldBlock,
721 .BADF => return error.NotOpenForWriting, // can be a race condition.
722 .DESTADDRREQ => unreachable, // `connect` was never called.
723 .DQUOT => return error.DiskQuota,
724 .FBIG => return error.FileTooBig,
725 .IO => return error.InputOutput,
726 .NOSPC => return error.NoSpaceLeft,
727 .ACCES => return error.AccessDenied,
728 .PERM => return error.PermissionDenied,
729 .PIPE => return error.BrokenPipe,
730 .CONNRESET => return error.ConnectionResetByPeer,
731 .BUSY => return error.DeviceBusy,
732 .NXIO => return error.NoDevice,
733 .MSGSIZE => return error.MessageTooBig,
734 else => |err| return posix.unexpectedErrno(err),
735 }
544736}