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-07-20 10:38:38-07:00
log4f214b97ec73bb9122a6d50aedbfdb0c4677fb4c
treeb6f6e75c7ab6723d83efccaf5fa1f2dc5eaedc12
parent50724cf1c37b95fcc71544596f978e917eb1aa4f

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;...@@ -9,26 +9,25 @@ const IoUring = std.os.linux.IoUring;
99
10gpa: Allocator,10gpa: Allocator,
11mutex: std.Thread.Mutex,11mutex: std.Thread.Mutex,
12cond: std.Thread.Condition,
13queue: std.DoublyLinkedList(void),12queue: std.DoublyLinkedList(void),
13/// Atomic copy of queue.len
14queue_len: usize,
14free: std.DoublyLinkedList(void),15free: std.DoublyLinkedList(void),
15main_context: Context,16main_fiber: Fiber,
16exit_awaiter: ?*Fiber,17idle_count: usize,
17threads: std.ArrayListUnmanaged(Thread),18threads: std.ArrayListUnmanaged(Thread),
18/// 1 bit per thread, same order as `thread_index`.19exiting: bool,
19idle_iourings: []usize,
2020
21threadlocal var thread_index: u32 = undefined;21threadlocal var thread_index: u32 = undefined;
2222
23/// Empirically saw 10KB being used by the self-hosted backend for logging.23/// 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
26const io_uring_entries = 64;26const io_uring_entries = 64;
2727
28const Thread = struct {28const Thread = struct {
29 thread: std.Thread,29 thread: std.Thread,
30 idle_context: Context,30 idle_context: Context,
31 current_idle_context: *Context,
32 current_context: *Context,31 current_context: *Context,
33 io_uring: IoUring,32 io_uring: IoUring,
3433
...@@ -103,98 +102,92 @@ pub fn io(el: *EventLoop) Io {...@@ -103,98 +102,92 @@ pub fn io(el: *EventLoop) Io {
103}102}
104103
105pub fn init(el: *EventLoop, gpa: Allocator) !void {104pub fn init(el: *EventLoop, gpa: Allocator) !void {
106 const n_threads: usize = @max((std.Thread.getCpuCount() catch 1), 1);105 const threads_size = @max(std.Thread.getCpuCount() catch 1, 1) * @sizeOf(Thread);
107 const threads_bytes = n_threads * @sizeOf(Thread);106 const idle_stack_end_offset = std.mem.alignForward(usize, threads_size + idle_stack_size, std.heap.page_size_max);
108 const idle_context_offset = std.mem.alignForward(usize, threads_bytes, @alignOf(Context));107 const allocated_slice = try gpa.alignedAlloc(u8, @alignOf(Thread), idle_stack_end_offset);
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);
111 errdefer gpa.free(allocated_slice);108 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);
115 el.* = .{109 el.* = .{
116 .gpa = gpa,110 .gpa = gpa,
117 .mutex = .{},111 .mutex = .{},
118 .cond = .{},
119 .queue = .{},112 .queue = .{},
113 .queue_len = 0,
120 .free = .{},114 .free = .{},
121 .main_context = undefined,115 .main_fiber = undefined,
122 .exit_awaiter = null,116 .idle_count = 0,
123 .threads = .initBuffer(@ptrCast(allocated_slice[0..threads_bytes])),117 .threads = .initBuffer(@ptrCast(allocated_slice[0..threads_size])),
124 .idle_iourings = idle_iourings,118 .exiting = false,
125 };119 };
120 thread_index = 0;
126 const main_thread = el.threads.addOneAssumeCapacity();121 const main_thread = el.threads.addOneAssumeCapacity();
127 main_thread.io_uring = try IoUring.init(io_uring_entries, 0);122 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)]));123 const idle_stack_end: [*]usize = @alignCast(@ptrCast(allocated_slice[idle_stack_end_offset..].ptr));
129 const idle_stack_end: [*]align(@max(@alignOf(Thread), @alignOf(Context))) usize = @alignCast(@ptrCast(allocated_slice[idle_stack_end_offset..].ptr));
130 (idle_stack_end - 1)[0..1].* = .{@intFromPtr(el)};124 (idle_stack_end - 1)[0..1].* = .{@intFromPtr(el)};
131 main_idle_context.* = .{125 main_thread.idle_context = .{
132 .rsp = @intFromPtr(idle_stack_end - 1),126 .rsp = @intFromPtr(idle_stack_end - 1),
133 .rbp = 0,127 .rbp = 0,
134 .rip = @intFromPtr(&mainIdleEntry),128 .rip = @intFromPtr(&mainIdleEntry),
135 };129 };
136 std.log.debug("created main idle {*}", .{main_idle_context});130 std.log.debug("created main idle {*}", .{&main_thread.idle_context});
137 main_thread.current_idle_context = main_idle_context;131 std.log.debug("created main {*}", .{&el.main_fiber});
138 std.log.debug("created main {*}", .{&el.main_context});132 main_thread.current_context = &el.main_fiber.context;
139 main_thread.current_context = &el.main_context;
140}133}
141134
142pub fn deinit(el: *EventLoop) void {135pub fn deinit(el: *EventLoop) void {
143 assert(el.queue.len == 0); // pending async136 assert(el.queue.len == 0); // pending async
144 el.yield(null, &el.exit_awaiter);137 el.yield(null, .exit);
145 while (el.free.pop()) |free_node| {138 while (el.free.pop()) |free_node| {
146 const free_fiber: *Fiber = @alignCast(@fieldParentPtr("queue_node", free_node));139 const free_fiber: *Fiber = @alignCast(@fieldParentPtr("queue_node", free_node));
147 el.gpa.free(free_fiber.allocatedSlice());140 el.gpa.free(free_fiber.allocatedSlice());
148 }141 }
149 const idle_context_offset = std.mem.alignForward(usize, el.threads.capacity * @sizeOf(Thread), @alignOf(Context));142 const idle_stack_end_offset = std.mem.alignForward(usize, el.threads.capacity * @sizeOf(Thread) + idle_stack_size, std.heap.page_size_max);
150 const idle_stack_end = std.mem.alignForward(usize, idle_context_offset + idle_stack_size, std.heap.page_size_max);143 const allocated_ptr: [*]align(@alignOf(Thread)) u8 = @alignCast(@ptrCast(el.threads.items.ptr));
151 const allocated_ptr: [*]align(@max(@alignOf(Thread), @alignOf(Context))) u8 = @alignCast(@ptrCast(el.threads.items.ptr));
152 for (el.threads.items[1..]) |*thread| thread.thread.join();144 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]);
154}146}
155147
156const PendingTask = union(enum) {148fn yield(el: *EventLoop, optional_fiber: ?*Fiber, pending_task: SwitchMessage.PendingTask) void {
157 none,
158 register_awaiter: *?*Fiber,
159 io_uring_submit: *IoUring,
160};
161
162fn yield(el: *EventLoop, optional_fiber: ?*Fiber, pending_task: PendingTask) void {
163 const thread: *Thread = &el.threads.items[thread_index];149 const thread: *Thread = &el.threads.items[thread_index];
164 const ready_context: *Context = ready_context: {150 const ready_context: *Context = ready_context: {
165 const ready_fiber: *Fiber = optional_fiber orelse if (ready_node: {151 const ready_fiber: *Fiber = optional_fiber orelse if (ready_node: {
166 el.mutex.lock();152 el.mutex.lock();
167 defer el.mutex.unlock();153 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;
169 }) |ready_node|157 }) |ready_node|
170 @alignCast(@fieldParentPtr("queue_node", ready_node))158 @alignCast(@fieldParentPtr("queue_node", ready_node))
171 else159 else
172 break :ready_context thread.current_idle_context;160 break :ready_context &thread.idle_context;
173 break :ready_context &ready_fiber.context;161 break :ready_context &ready_fiber.context;
174 };162 };
175 const message: SwitchMessage = .{163 const message: SwitchMessage = .{
176 .prev_context = thread.current_context,164 .contexts = .{
177 .ready_context = ready_context,165 .prev = thread.current_context,
166 .ready = ready_context,
167 },
178 .pending_task = pending_task,168 .pending_task = pending_task,
179 };169 };
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 });
181 contextSwitch(&message).handle(el);171 contextSwitch(&message).handle(el);
182}172}
183173
184fn schedule(el: *EventLoop, fiber: *Fiber) void {174fn schedule(el: *EventLoop, fiber: *Fiber) void {
185 el.mutex.lock();175 if (idle_count: {
186 el.queue.append(&fiber.queue_node);176 el.mutex.lock();
187 //for (el.idle_iourings) |*int| {177 defer el.mutex.unlock();
188 // const idler_subset = @atomicLoad(usize, int, .unordered);178 el.queue.append(&fiber.queue_node);
189 // if (idler_subset == 0) continue;179 @atomicStore(usize, &el.queue_len, el.queue.len, .unordered);
190 //180 break :idle_count el.idle_count;
191 //}181 } > 0) {
192 if (el.idle_count > 0) {182 _ = std.os.linux.futex2_wake(&el.queue_len, std.math.maxInt(usize), 1, switch (@bitSizeOf(usize)) {
193 el.mutex.unlock();183 8 => std.os.linux.FUTEX2.SIZE_U8,
194 el.cond.signal();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
195 return;189 return;
196 }190 }
197 defer el.mutex.unlock();
198 if (el.threads.items.len == el.threads.capacity) return;191 if (el.threads.items.len == el.threads.capacity) return;
199 const thread = el.threads.addOneAssumeCapacity();192 const thread = el.threads.addOneAssumeCapacity();
200 thread.thread = std.Thread.spawn(.{193 thread.thread = std.Thread.spawn(.{
...@@ -216,64 +209,101 @@ fn recycle(el: *EventLoop, fiber: *Fiber) void {...@@ -216,64 +209,101 @@ fn recycle(el: *EventLoop, fiber: *Fiber) void {
216209
217fn mainIdle(el: *EventLoop, message: *const SwitchMessage) callconv(.withStackAlign(.c, @max(@alignOf(Thread), @alignOf(Context)))) noreturn {210fn mainIdle(el: *EventLoop, message: *const SwitchMessage) callconv(.withStackAlign(.c, @max(@alignOf(Thread), @alignOf(Context)))) noreturn {
218 message.handle(el);211 message.handle(el);
219 el.yield(el.idle(), null);212 el.idle();
213 el.yield(&el.main_fiber, .nothing);
220 unreachable; // switched to dead fiber214 unreachable; // switched to dead fiber
221}215}
222216
223fn threadEntry(el: *EventLoop, index: usize) void {217fn threadEntry(el: *EventLoop, index: usize) void {
224 thread_index = index;218 thread_index = @intCast(index);
225 const thread: *Thread = &el.threads.items[index];219 const thread: *Thread = &el.threads.items[index];
226 std.log.debug("created thread idle {*}", .{&thread.idle_context});220 std.log.debug("created thread idle {*}", .{&thread.idle_context});
227 thread.io_uring = IoUring.init(io_uring_entries, 0) catch |err| {221 thread.io_uring = IoUring.init(io_uring_entries, 0) catch |err| {
228 std.log.warn("exiting worker thread during init due to io_uring init failure: {s}", .{@errorName(err)});222 std.log.warn("exiting worker thread during init due to io_uring init failure: {s}", .{@errorName(err)});
229 return;223 return;
230 };224 };
231 thread.current_idle_context = &thread.idle_context;
232 thread.current_context = &thread.idle_context;225 thread.current_context = &thread.idle_context;
233 _ = el.idle();226 el.idle();
234}227}
235228
236fn idle(el: *EventLoop) *Fiber {229const UserData = enum(u64) {
230 queue_len_futex_wait,
231 _,
232};
233
234fn idle(el: *EventLoop) void {
237 const thread: *Thread = &el.threads.items[thread_index];235 const thread: *Thread = &el.threads.items[thread_index];
238 // The idle fiber only runs on one thread.
239 const iou = &thread.io_uring;236 const iou = &thread.io_uring;
240 var cqes_buffer: [io_uring_entries]std.os.linux.io_uring_cqe = undefined;237 var cqes_buffer: [io_uring_entries]std.os.linux.io_uring_cqe = undefined;
238 var futex_is_scheduled: bool = false;
241239
242 while (true) {240 while (true) {
243 el.yield(null, null);241 el.yield(null, .nothing);
244 if (@atomicLoad(?*Fiber, &el.exit_awaiter, .acquire)) |exit_awaiter| {242 if (@atomicLoad(bool, &el.exiting, .acquire)) return;
245 el.cond.broadcast();243 if (!futex_is_scheduled) {
246 return exit_awaiter;244 const sqe = getSqe(&thread.io_uring);
247 }245 sqe.prep_rw(.FUTEX_WAIT, switch (@bitSizeOf(usize)) {
248 // TODO add uring to bit set246 8 => std.os.linux.FUTEX2.SIZE_U8,
249 const n = iou.copy_cqes(&cqes_buffer, 1) catch @panic("TODO handle copy_cqes error");247 16 => std.os.linux.FUTEX2.SIZE_U16,
250 const cqes = cqes_buffer[0..n];248 32 => std.os.linux.FUTEX2.SIZE_U32,
251 for (cqes) |cqe| {249 64 => std.os.linux.FUTEX2.SIZE_U64,
252 const fiber: *Fiber = @ptrFromInt(cqe.user_data);250 else => @compileError("unsupported @sizeOf(usize)"),
253 const res: *i32 = @ptrCast(@alignCast(fiber.resultPointer()));251 } | std.os.linux.FUTEX2.PRIVATE, @intFromPtr(&el.queue_len), 0, 0);
254 res.* = cqe.res;252 sqe.addr3 = std.math.maxInt(u64);
255 el.schedule(fiber);253 sqe.user_data = @intFromEnum(UserData.queue_len_futex_wait);
254 futex_is_scheduled = true;
256 }255 }
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 };
257 }272 }
258}273}
259274
260const SwitchMessage = extern struct {275const SwitchMessage = struct {
261 prev_context: *Context,276 contexts: extern struct {
262 ready_context: *Context,277 prev: *Context,
278 ready: *Context,
279 },
263 pending_task: PendingTask,280 pending_task: PendingTask,
264281
282 const PendingTask = union(enum) {
283 nothing,
284 register_awaiter: *?*Fiber,
285 exit,
286 };
287
265 fn handle(message: *const SwitchMessage, el: *EventLoop) void {288 fn handle(message: *const SwitchMessage, el: *EventLoop) void {
266 const thread: *Thread = &el.threads.items[thread_index];289 const thread: *Thread = &el.threads.items[thread_index];
267 thread.current_context = message.ready_context;290 thread.current_context = message.contexts.ready;
268 switch (message.pending_task) {291 switch (message.pending_task) {
269 .none => {},292 .nothing => {},
270 .register_awaiter => |awaiter| {293 .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));
272 if (@atomicRmw(?*Fiber, awaiter, .Xchg, prev_fiber, .acq_rel) == Fiber.finished) el.schedule(prev_fiber);295 if (@atomicRmw(?*Fiber, awaiter, .Xchg, prev_fiber, .acq_rel) == Fiber.finished) el.schedule(prev_fiber);
273 },296 },
274 .io_uring_submit => |iou| {297 .exit => {
275 _ = iou.flush_sq();298 @atomicStore(bool, &el.exiting, true, .unordered);
276 // TODO: determine whether this return value should be used299 @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
277 },307 },
278 }308 }
279 }309 }
...@@ -289,7 +319,7 @@ const Context = switch (builtin.cpu.arch) {...@@ -289,7 +319,7 @@ const Context = switch (builtin.cpu.arch) {
289};319};
290320
291inline fn contextSwitch(message: *const SwitchMessage) *const SwitchMessage {321inline fn contextSwitch(message: *const SwitchMessage) *const SwitchMessage {
292 return switch (builtin.cpu.arch) {322 return @fieldParentPtr("contexts", switch (builtin.cpu.arch) {
293 .x86_64 => asm volatile (323 .x86_64 => asm volatile (
294 \\ movq 0(%%rsi), %%rax324 \\ movq 0(%%rsi), %%rax
295 \\ movq 8(%%rsi), %%rcx325 \\ movq 8(%%rsi), %%rcx
...@@ -301,8 +331,8 @@ inline fn contextSwitch(message: *const SwitchMessage) *const SwitchMessage {...@@ -301,8 +331,8 @@ inline fn contextSwitch(message: *const SwitchMessage) *const SwitchMessage {
301 \\ movq 8(%%rcx), %%rbp331 \\ movq 8(%%rcx), %%rbp
302 \\ jmpq *16(%%rcx)332 \\ jmpq *16(%%rcx)
303 \\0:333 \\0:
304 : [received_message] "={rsi}" (-> *const SwitchMessage),334 : [received_message] "={rsi}" (-> *const @FieldType(SwitchMessage, "contexts")),
305 : [message_to_send] "{rsi}" (message),335 : [message_to_send] "{rsi}" (&message.contexts),
306 : "rax", "rcx", "rdx", "rbx", "rdi", //336 : "rax", "rcx", "rdx", "rbx", "rdi", //
307 "r8", "r9", "r10", "r11", "r12", "r13", "r14", "r15", //337 "r8", "r9", "r10", "r11", "r12", "r13", "r14", "r15", //
308 "mm0", "mm1", "mm2", "mm3", "mm4", "mm5", "mm6", "mm7", //338 "mm0", "mm1", "mm2", "mm3", "mm4", "mm5", "mm6", "mm7", //
...@@ -313,7 +343,7 @@ inline fn contextSwitch(message: *const SwitchMessage) *const SwitchMessage {...@@ -313,7 +343,7 @@ inline fn contextSwitch(message: *const SwitchMessage) *const SwitchMessage {
313 "fpsr", "fpcr", "mxcsr", "rflags", "dirflag", "memory"343 "fpsr", "fpcr", "mxcsr", "rflags", "dirflag", "memory"
314 ),344 ),
315 else => |arch| @compileError("unimplemented architecture: " ++ @tagName(arch)),345 else => |arch| @compileError("unimplemented architecture: " ++ @tagName(arch)),
316 };346 });
317}347}
318348
319fn mainIdleEntry() callconv(.naked) void {349fn mainIdleEntry() callconv(.naked) void {
...@@ -401,7 +431,7 @@ const AsyncClosure = struct {...@@ -401,7 +431,7 @@ const AsyncClosure = struct {
401 std.log.debug("{*} performing async", .{closure.fiber});431 std.log.debug("{*} performing async", .{closure.fiber});
402 closure.start(closure.contextPointer(), closure.fiber.resultPointer());432 closure.start(closure.contextPointer(), closure.fiber.resultPointer());
403 const awaiter = @atomicRmw(?*Fiber, &closure.fiber.awaiter, .Xchg, Fiber.finished, .acq_rel);433 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);
405 unreachable; // switched to dead fiber435 unreachable; // switched to dead fiber
406 }436 }
407};437};
...@@ -409,17 +439,93 @@ const AsyncClosure = struct {...@@ -409,17 +439,93 @@ const AsyncClosure = struct {
409pub fn @"await"(userdata: ?*anyopaque, any_future: *std.Io.AnyFuture, result: []u8) void {439pub fn @"await"(userdata: ?*anyopaque, any_future: *std.Io.AnyFuture, result: []u8) void {
410 const event_loop: *EventLoop = @alignCast(@ptrCast(userdata));440 const event_loop: *EventLoop = @alignCast(@ptrCast(userdata));
411 const future_fiber: *Fiber = @alignCast(@ptrCast(any_future));441 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 });
413 @memcpy(result, future_fiber.resultPointer());443 @memcpy(result, future_fiber.resultPointer());
414 event_loop.recycle(future_fiber);444 event_loop.recycle(future_fiber);
415}445}
416446
417pub fn createFile(userdata: ?*anyopaque, dir: std.fs.Dir, sub_path: []const u8, flags: std.fs.File.CreateFlags) std.fs.File.OpenError!std.fs.File {447pub 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;448 const el: *EventLoop = @ptrCast(@alignCast(userdata));
419 _ = dir;449
420 _ = sub_path;450 const posix = std.posix;
421 _ = flags;451 const sub_path_c = try posix.toPosixPath(sub_path);
422 @panic("TODO");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 }
423}529}
424530
425pub fn openFile(userdata: ?*anyopaque, dir: std.fs.Dir, sub_path: []const u8, flags: std.fs.File.OpenFlags) std.fs.File.OpenError!std.fs.File {531pub 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...@@ -476,7 +582,7 @@ pub fn openFile(userdata: ?*anyopaque, dir: std.fs.Dir, sub_path: []const u8, fl
476 sqe.prep_openat(dir.fd, &sub_path_c, os_flags, 0);582 sqe.prep_openat(dir.fd, &sub_path_c, os_flags, 0);
477 sqe.user_data = @intFromPtr(fiber);583 sqe.user_data = @intFromPtr(fiber);
478584
479 el.yield(null, .{ .io_uring_submit = iou });585 el.yield(null, .nothing);
480586
481 const result: *i32 = @alignCast(@ptrCast(fiber.resultPointer()[0..@sizeOf(posix.fd_t)]));587 const result: *i32 = @alignCast(@ptrCast(fiber.resultPointer()[0..@sizeOf(posix.fd_t)]));
482 const rc = result.*;588 const rc = result.*;
...@@ -510,8 +616,6 @@ pub fn openFile(userdata: ?*anyopaque, dir: std.fs.Dir, sub_path: []const u8, fl...@@ -510,8 +616,6 @@ pub fn openFile(userdata: ?*anyopaque, dir: std.fs.Dir, sub_path: []const u8, fl
510 .NXIO => return error.NoDevice,616 .NXIO => return error.NoDevice,
511 else => |err| return posix.unexpectedErrno(err),617 else => |err| return posix.unexpectedErrno(err),
512 }618 }
513
514 return .{ .handle = result.* };
515}619}
516620
517fn errno(signed: i32) std.posix.E {621fn errno(signed: i32) std.posix.E {
...@@ -524,21 +628,109 @@ fn getSqe(iou: *IoUring) *std.os.linux.io_uring_sqe {...@@ -524,21 +628,109 @@ fn getSqe(iou: *IoUring) *std.os.linux.io_uring_sqe {
524}628}
525629
526pub fn closeFile(userdata: ?*anyopaque, file: std.fs.File) void {630pub fn closeFile(userdata: ?*anyopaque, file: std.fs.File) void {
527 _ = userdata;631 const el: *EventLoop = @ptrCast(@alignCast(userdata));
528 _ = file;632
529 @panic("TODO");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 }
530}655}
531656
532pub fn read(userdata: ?*anyopaque, file: std.fs.File, buffer: []u8) std.fs.File.ReadError!usize {657pub fn read(userdata: ?*anyopaque, file: std.fs.File, buffer: []u8) std.fs.File.ReadError!usize {
533 _ = userdata;658 const el: *EventLoop = @ptrCast(@alignCast(userdata));
534 _ = file;659
535 _ = buffer;660 const posix = std.posix;
536 @panic("TODO");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 }
537}693}
538694
539pub fn write(userdata: ?*anyopaque, file: std.fs.File, buffer: []const u8) std.fs.File.WriteError!usize {695pub fn write(userdata: ?*anyopaque, file: std.fs.File, buffer: []const u8) std.fs.File.WriteError!usize {
540 _ = userdata;696 const el: *EventLoop = @ptrCast(@alignCast(userdata));
541 _ = file;697
542 _ = buffer;698 const posix = std.posix;
543 @panic("TODO");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 }
544}736}