authorgravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2025-03-28 21:07:12-07:00
committergravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2025-10-02 16:30:59-07:00
logb215ddc9fbb6a396a7ec19e1ee988aa400b887a0
tree5043de8276ec51bb4301eac5d65b29bb8ca5451c
parentbe68b28b9b9114611e72a86f0ceee3d8c277ee20

WIP

1 files changed, 213 insertions(+), 28 deletions(-)

lib/std/Io/EventLoop.zig+213-28
......@@ -5,6 +5,7 @@ const Allocator = std.mem.Allocator;
55const Io = std.Io;
66const EventLoop = @This();
77const Alignment = std.mem.Alignment;
8const IoUring = std.os.linux.IoUring;
89
910gpa: Allocator,
1011mutex: std.Thread.Mutex,
......@@ -13,18 +14,27 @@ queue: std.DoublyLinkedList(void),
1314free: std.DoublyLinkedList(void),
1415main_context: Context,
1516exit_awaiter: ?*Fiber,
16idle_count: usize,
1717threads: std.ArrayListUnmanaged(Thread),
18/// 1 bit per thread, same order as `thread_index`.
19idle_iourings: []usize,
1820
19threadlocal var current_idle_context: *Context = undefined;
20threadlocal var current_context: *Context = undefined;
21threadlocal var thread_index: u32 = undefined;
2122
2223/// Empirically saw 10KB being used by the self-hosted backend for logging.
2324const idle_stack_size = 32 * 1024;
2425
26const io_uring_entries = 64;
27
2528const Thread = struct {
2629 thread: std.Thread,
2730 idle_context: Context,
31 current_idle_context: *Context,
32 current_context: *Context,
33 io_uring: IoUring,
34
35 fn currentFiber(thread: *Thread) *Fiber {
36 return @fieldParentPtr("context", thread.current_context);
37 }
2838};
2939
3040const Fiber = struct {
......@@ -83,16 +93,25 @@ pub fn io(el: *EventLoop) Io {
8393 .vtable = &.{
8494 .@"async" = @"async",
8595 .@"await" = @"await",
96 .createFile = createFile,
97 .openFile = openFile,
98 .closeFile = closeFile,
99 .read = read,
100 .write = write,
86101 },
87102 };
88103}
89104
90pub fn init(el: *EventLoop, gpa: Allocator) error{OutOfMemory}!void {
91 const threads_bytes = ((std.Thread.getCpuCount() catch 1) -| 1) * @sizeOf(Thread);
105pub 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);
92108 const idle_context_offset = std.mem.alignForward(usize, threads_bytes, @alignOf(Context));
93109 const idle_stack_end_offset = std.mem.alignForward(usize, idle_context_offset + idle_stack_size, std.heap.page_size_max);
94110 const allocated_slice = try gpa.alignedAlloc(u8, @max(@alignOf(Thread), @alignOf(Context)), idle_stack_end_offset);
95111 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);
96115 el.* = .{
97116 .gpa = gpa,
98117 .mutex = .{},
......@@ -101,9 +120,11 @@ pub fn init(el: *EventLoop, gpa: Allocator) error{OutOfMemory}!void {
101120 .free = .{},
102121 .main_context = undefined,
103122 .exit_awaiter = null,
104 .idle_count = 0,
105123 .threads = .initBuffer(@ptrCast(allocated_slice[0..threads_bytes])),
124 .idle_iourings = idle_iourings,
106125 };
126 const main_thread = el.threads.addOneAssumeCapacity();
127 main_thread.io_uring = try IoUring.init(io_uring_entries, 0);
107128 const main_idle_context: *Context = @alignCast(std.mem.bytesAsValue(Context, allocated_slice[idle_context_offset..][0..@sizeOf(Context)]));
108129 const idle_stack_end: [*]align(@max(@alignOf(Thread), @alignOf(Context))) usize = @alignCast(@ptrCast(allocated_slice[idle_stack_end_offset..].ptr));
109130 (idle_stack_end - 1)[0..1].* = .{@intFromPtr(el)};
......@@ -113,9 +134,9 @@ pub fn init(el: *EventLoop, gpa: Allocator) error{OutOfMemory}!void {
113134 .rip = @intFromPtr(&mainIdleEntry),
114135 };
115136 std.log.debug("created main idle {*}", .{main_idle_context});
116 current_idle_context = main_idle_context;
137 main_thread.current_idle_context = main_idle_context;
117138 std.log.debug("created main {*}", .{&el.main_context});
118 current_context = &el.main_context;
139 main_thread.current_context = &el.main_context;
119140}
120141
121142pub fn deinit(el: *EventLoop) void {
......@@ -125,14 +146,21 @@ pub fn deinit(el: *EventLoop) void {
125146 const free_fiber: *Fiber = @alignCast(@fieldParentPtr("queue_node", free_node));
126147 el.gpa.free(free_fiber.allocatedSlice());
127148 }
128 const idle_context_offset = std.mem.alignForward(usize, el.threads.items.len * @sizeOf(Thread), @alignOf(Context));
149 const idle_context_offset = std.mem.alignForward(usize, el.threads.capacity * @sizeOf(Thread), @alignOf(Context));
129150 const idle_stack_end = std.mem.alignForward(usize, idle_context_offset + idle_stack_size, std.heap.page_size_max);
130151 const allocated_ptr: [*]align(@max(@alignOf(Thread), @alignOf(Context))) u8 = @alignCast(@ptrCast(el.threads.items.ptr));
131 for (el.threads.items) |*thread| thread.thread.join();
152 for (el.threads.items[1..]) |*thread| thread.thread.join();
132153 el.gpa.free(allocated_ptr[0..idle_stack_end]);
133154}
134155
135fn yield(el: *EventLoop, optional_fiber: ?*Fiber, register_awaiter: ?*?*Fiber) void {
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 {
163 const thread: *Thread = &el.threads.items[thread_index];
136164 const ready_context: *Context = ready_context: {
137165 const ready_fiber: *Fiber = optional_fiber orelse if (ready_node: {
138166 el.mutex.lock();
......@@ -141,13 +169,13 @@ fn yield(el: *EventLoop, optional_fiber: ?*Fiber, register_awaiter: ?*?*Fiber) v
141169 }) |ready_node|
142170 @alignCast(@fieldParentPtr("queue_node", ready_node))
143171 else
144 break :ready_context current_idle_context;
172 break :ready_context thread.current_idle_context;
145173 break :ready_context &ready_fiber.context;
146174 };
147175 const message: SwitchMessage = .{
148 .prev_context = current_context,
176 .prev_context = thread.current_context,
149177 .ready_context = ready_context,
150 .register_awaiter = register_awaiter,
178 .pending_task = pending_task,
151179 };
152180 std.log.debug("switching from {*} to {*}", .{ message.prev_context, message.ready_context });
153181 contextSwitch(&message).handle(el);
......@@ -156,6 +184,11 @@ fn yield(el: *EventLoop, optional_fiber: ?*Fiber, register_awaiter: ?*?*Fiber) v
156184fn schedule(el: *EventLoop, fiber: *Fiber) void {
157185 el.mutex.lock();
158186 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 //}
159192 if (el.idle_count > 0) {
160193 el.mutex.unlock();
161194 el.cond.signal();
......@@ -167,7 +200,7 @@ fn schedule(el: *EventLoop, fiber: *Fiber) void {
167200 thread.thread = std.Thread.spawn(.{
168201 .stack_size = idle_stack_size,
169202 .allocator = el.gpa,
170 }, threadEntry, .{ el, thread }) catch {
203 }, threadEntry, .{ el, el.threads.items.len - 1 }) catch {
171204 el.threads.items.len -= 1;
172205 return;
173206 };
......@@ -187,38 +220,61 @@ fn mainIdle(el: *EventLoop, message: *const SwitchMessage) callconv(.withStackAl
187220 unreachable; // switched to dead fiber
188221}
189222
190fn threadEntry(el: *EventLoop, thread: *Thread) void {
223fn threadEntry(el: *EventLoop, index: usize) void {
224 thread_index = index;
225 const thread: *Thread = &el.threads.items[index];
191226 std.log.debug("created thread idle {*}", .{&thread.idle_context});
192 current_idle_context = &thread.idle_context;
193 current_context = &thread.idle_context;
227 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)});
229 return;
230 };
231 thread.current_idle_context = &thread.idle_context;
232 thread.current_context = &thread.idle_context;
194233 _ = el.idle();
195234}
196235
197236fn idle(el: *EventLoop) *Fiber {
237 const thread: *Thread = &el.threads.items[thread_index];
238 // The idle fiber only runs on one thread.
239 const iou = &thread.io_uring;
240 var cqes_buffer: [io_uring_entries]std.os.linux.io_uring_cqe = undefined;
241
198242 while (true) {
199243 el.yield(null, null);
200244 if (@atomicLoad(?*Fiber, &el.exit_awaiter, .acquire)) |exit_awaiter| {
201245 el.cond.broadcast();
202246 return exit_awaiter;
203247 }
204 el.mutex.lock();
205 defer el.mutex.unlock();
206 el.idle_count += 1;
207 defer el.idle_count -= 1;
208 el.cond.wait(&el.mutex);
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);
256 }
209257 }
210258}
211259
212260const SwitchMessage = extern struct {
213261 prev_context: *Context,
214262 ready_context: *Context,
215 register_awaiter: ?*?*Fiber,
263 pending_task: PendingTask,
216264
217265 fn handle(message: *const SwitchMessage, el: *EventLoop) void {
218 current_context = message.ready_context;
219 if (message.register_awaiter) |awaiter| {
220 const prev_fiber: *Fiber = @alignCast(@fieldParentPtr("context", message.prev_context));
221 if (@atomicRmw(?*Fiber, awaiter, .Xchg, prev_fiber, .acq_rel) == Fiber.finished) el.schedule(prev_fiber);
266 const thread: *Thread = &el.threads.items[thread_index];
267 thread.current_context = message.ready_context;
268 switch (message.pending_task) {
269 .none => {},
270 .register_awaiter => |awaiter| {
271 const prev_fiber: *Fiber = @alignCast(@fieldParentPtr("context", message.prev_context));
272 if (@atomicRmw(?*Fiber, awaiter, .Xchg, prev_fiber, .acq_rel) == Fiber.finished) el.schedule(prev_fiber);
273 },
274 .io_uring_submit => |iou| {
275 _ = iou.flush_sq();
276 // TODO: determine whether this return value should be used
277 },
222278 }
223279 }
224280};
......@@ -357,3 +413,132 @@ pub fn @"await"(userdata: ?*anyopaque, any_future: *std.Io.AnyFuture, result: []
357413 @memcpy(result, future_fiber.resultPointer());
358414 event_loop.recycle(future_fiber);
359415}
416
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 {
418 _ = userdata;
419 _ = dir;
420 _ = sub_path;
421 _ = flags;
422 @panic("TODO");
423}
424
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 {
426 const el: *EventLoop = @ptrCast(@alignCast(userdata));
427
428 const posix = std.posix;
429 const sub_path_c = try posix.toPosixPath(sub_path);
430
431 var os_flags: posix.O = .{
432 .ACCMODE = switch (flags.mode) {
433 .read_only => .RDONLY,
434 .write_only => .WRONLY,
435 .read_write => .RDWR,
436 },
437 };
438
439 if (@hasField(posix.O, "CLOEXEC")) os_flags.CLOEXEC = true;
440 if (@hasField(posix.O, "LARGEFILE")) os_flags.LARGEFILE = true;
441 if (@hasField(posix.O, "NOCTTY")) os_flags.NOCTTY = !flags.allow_ctty;
442
443 // Use the O locking flags if the os supports them to acquire the lock
444 // atomically.
445 const has_flock_open_flags = @hasField(posix.O, "EXLOCK");
446 if (has_flock_open_flags) {
447 // Note that the NONBLOCK flag is removed after the openat() call
448 // is successful.
449 switch (flags.lock) {
450 .none => {},
451 .shared => {
452 os_flags.SHLOCK = true;
453 os_flags.NONBLOCK = flags.lock_nonblocking;
454 },
455 .exclusive => {
456 os_flags.EXLOCK = true;
457 os_flags.NONBLOCK = flags.lock_nonblocking;
458 },
459 }
460 }
461 const have_flock = @TypeOf(posix.system.flock) != void;
462
463 if (have_flock and !has_flock_open_flags and flags.lock != .none) {
464 @panic("TODO");
465 }
466
467 if (has_flock_open_flags and flags.lock_nonblocking) {
468 @panic("TODO");
469 }
470
471 const thread: *Thread = &el.threads.items[thread_index];
472 const iou = &thread.io_uring;
473 const sqe = getSqe(iou);
474 const fiber = thread.currentFiber();
475
476 sqe.prep_openat(dir.fd, &sub_path_c, os_flags, 0);
477 sqe.user_data = @intFromPtr(fiber);
478
479 el.yield(null, .{ .io_uring_submit = iou });
480
481 const result: *i32 = @alignCast(@ptrCast(fiber.resultPointer()[0..@sizeOf(posix.fd_t)]));
482 const rc = result.*;
483 switch (errno(rc)) {
484 .SUCCESS => return .{ .handle = rc },
485 .INTR => @panic("TODO is this reachable?"),
486 .CANCELED => @panic("TODO figure out how this error code fits into things"),
487
488 .FAULT => unreachable,
489 .INVAL => return error.BadPathName,
490 .BADF => unreachable,
491 .ACCES => return error.AccessDenied,
492 .FBIG => return error.FileTooBig,
493 .OVERFLOW => return error.FileTooBig,
494 .ISDIR => return error.IsDir,
495 .LOOP => return error.SymLinkLoop,
496 .MFILE => return error.ProcessFdQuotaExceeded,
497 .NAMETOOLONG => return error.NameTooLong,
498 .NFILE => return error.SystemFdQuotaExceeded,
499 .NODEV => return error.NoDevice,
500 .NOENT => return error.FileNotFound,
501 .NOMEM => return error.SystemResources,
502 .NOSPC => return error.NoSpaceLeft,
503 .NOTDIR => return error.NotDir,
504 .PERM => return error.PermissionDenied,
505 .EXIST => return error.PathAlreadyExists,
506 .BUSY => return error.DeviceBusy,
507 .OPNOTSUPP => return error.FileLocksNotSupported,
508 .AGAIN => return error.WouldBlock,
509 .TXTBSY => return error.FileBusy,
510 .NXIO => return error.NoDevice,
511 else => |err| return posix.unexpectedErrno(err),
512 }
513
514 return .{ .handle = result.* };
515}
516
517fn errno(signed: i32) std.posix.E {
518 const int = if (signed > -4096 and signed < 0) -signed else 0;
519 return @enumFromInt(int);
520}
521
522fn getSqe(iou: *IoUring) *std.os.linux.io_uring_sqe {
523 return iou.get_sqe() catch @panic("TODO: handle submission queue full");
524}
525
526pub fn closeFile(userdata: ?*anyopaque, file: std.fs.File) void {
527 _ = userdata;
528 _ = file;
529 @panic("TODO");
530}
531
532pub fn read(userdata: ?*anyopaque, file: std.fs.File, buffer: []u8) std.fs.File.ReadError!usize {
533 _ = userdata;
534 _ = file;
535 _ = buffer;
536 @panic("TODO");
537}
538
539pub 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");
544}