authorgravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2025-10-15 14:47:41-07:00
committergravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2025-10-29 06:20:49-07:00
log1e81c3a9259873fac197ed5dc6e2133f0a097243
treeaff7722e47948044694d90a752caa1c3654e689c
parent81b1bfbfbbf4f71bc79191ca36ac62c00c8ac92c

std.Io: rename EventLoop to IoUring

`std.Io.Evented` is introduced to select an appropriate Io implementation depending on OS

3 files changed, 1622 insertions(+), 1619 deletions(-)

lib/std/Io.zig+4-1
......@@ -557,7 +557,10 @@ test {
557557
558558const Io = @This();
559559
560pub const EventLoop = @import("Io/EventLoop.zig");
560pub const Evented = switch (builtin.os.tag) {
561 .linux => @import("Io/IoUring.zig"),
562 else => void,
563};
561564pub const Threaded = @import("Io/Threaded.zig");
562565pub const net = @import("Io/net.zig");
563566
lib/std/Io/EventLoop.zig deleted-1618
......@@ -1,1618 +0,0 @@
1const std = @import("../std.zig");
2const builtin = @import("builtin");
3const assert = std.debug.assert;
4const Allocator = std.mem.Allocator;
5const Io = std.Io;
6const EventLoop = @This();
7const Alignment = std.mem.Alignment;
8const IoUring = std.os.linux.IoUring;
9
10/// Must be a thread-safe allocator.
11gpa: Allocator,
12main_fiber_buffer: [@sizeOf(Fiber) + Fiber.max_result_size]u8 align(@alignOf(Fiber)),
13threads: Thread.List,
14detached: struct {
15 mutex: std.Io.Mutex,
16 list: std.DoublyLinkedList,
17},
18
19/// Empirically saw >128KB being used by the self-hosted backend to panic.
20const idle_stack_size = 256 * 1024;
21
22const max_idle_search = 4;
23const max_steal_ready_search = 4;
24
25const io_uring_entries = 64;
26
27const Thread = struct {
28 thread: std.Thread,
29 idle_context: Context,
30 current_context: *Context,
31 ready_queue: ?*Fiber,
32 io_uring: IoUring,
33 idle_search_index: u32,
34 steal_ready_search_index: u32,
35
36 const canceling: ?*Thread = @ptrFromInt(@alignOf(Thread));
37
38 threadlocal var self: *Thread = undefined;
39
40 fn current() *Thread {
41 return self;
42 }
43
44 fn currentFiber(thread: *Thread) *Fiber {
45 return @fieldParentPtr("context", thread.current_context);
46 }
47
48 const List = struct {
49 allocated: []Thread,
50 reserved: u32,
51 active: u32,
52 };
53};
54
55const Fiber = struct {
56 required_align: void align(4),
57 context: Context,
58 awaiter: ?*Fiber,
59 queue_next: ?*Fiber,
60 cancel_thread: ?*Thread,
61 awaiting_completions: std.StaticBitSet(3),
62
63 const finished: ?*Fiber = @ptrFromInt(@alignOf(Thread));
64
65 const max_result_align: Alignment = .@"16";
66 const max_result_size = max_result_align.forward(64);
67 /// This includes any stack realignments that need to happen, and also the
68 /// initial frame return address slot and argument frame, depending on target.
69 const min_stack_size = 4 * 1024 * 1024;
70 const max_context_align: Alignment = .@"16";
71 const max_context_size = max_context_align.forward(1024);
72 const max_closure_size: usize = @max(@sizeOf(AsyncClosure), @sizeOf(DetachedClosure));
73 const max_closure_align: Alignment = .max(.of(AsyncClosure), .of(DetachedClosure));
74 const allocation_size = std.mem.alignForward(
75 usize,
76 max_closure_align.max(max_context_align).forward(
77 max_result_align.forward(@sizeOf(Fiber)) + max_result_size + min_stack_size,
78 ) + max_closure_size + max_context_size,
79 std.heap.page_size_max,
80 );
81
82 fn allocate(el: *EventLoop) error{OutOfMemory}!*Fiber {
83 return @ptrCast(try el.gpa.alignedAlloc(u8, .of(Fiber), allocation_size));
84 }
85
86 fn allocatedSlice(f: *Fiber) []align(@alignOf(Fiber)) u8 {
87 return @as([*]align(@alignOf(Fiber)) u8, @ptrCast(f))[0..allocation_size];
88 }
89
90 fn allocatedEnd(f: *Fiber) [*]u8 {
91 const allocated_slice = f.allocatedSlice();
92 return allocated_slice[allocated_slice.len..].ptr;
93 }
94
95 fn resultPointer(f: *Fiber, comptime Result: type) *Result {
96 return @ptrCast(@alignCast(f.resultBytes(.of(Result))));
97 }
98
99 fn resultBytes(f: *Fiber, alignment: Alignment) [*]u8 {
100 return @ptrFromInt(alignment.forward(@intFromPtr(f) + @sizeOf(Fiber)));
101 }
102
103 fn enterCancelRegion(fiber: *Fiber, thread: *Thread) error{Canceled}!void {
104 if (@cmpxchgStrong(
105 ?*Thread,
106 &fiber.cancel_thread,
107 null,
108 thread,
109 .acq_rel,
110 .acquire,
111 )) |cancel_thread| {
112 assert(cancel_thread == Thread.canceling);
113 return error.Canceled;
114 }
115 }
116
117 fn exitCancelRegion(fiber: *Fiber, thread: *Thread) void {
118 if (@cmpxchgStrong(
119 ?*Thread,
120 &fiber.cancel_thread,
121 thread,
122 null,
123 .acq_rel,
124 .acquire,
125 )) |cancel_thread| assert(cancel_thread == Thread.canceling);
126 }
127
128 const Queue = struct { head: *Fiber, tail: *Fiber };
129};
130
131fn recycle(el: *EventLoop, fiber: *Fiber) void {
132 std.log.debug("recyling {*}", .{fiber});
133 assert(fiber.queue_next == null);
134 el.gpa.free(fiber.allocatedSlice());
135}
136
137pub fn io(el: *EventLoop) Io {
138 return .{
139 .userdata = el,
140 .vtable = &.{
141 .async = async,
142 .concurrent = concurrent,
143 .await = await,
144 .asyncDetached = asyncDetached,
145 .select = select,
146 .cancel = cancel,
147 .cancelRequested = cancelRequested,
148
149 .mutexLock = mutexLock,
150 .mutexUnlock = mutexUnlock,
151
152 .conditionWait = conditionWait,
153 .conditionWake = conditionWake,
154
155 .createFile = createFile,
156 .fileOpen = fileOpen,
157 .fileClose = fileClose,
158 .pread = pread,
159 .pwrite = pwrite,
160
161 .now = now,
162 .sleep = sleep,
163 },
164 };
165}
166
167pub fn init(el: *EventLoop, gpa: Allocator) !void {
168 const threads_size = @max(std.Thread.getCpuCount() catch 1, 1) * @sizeOf(Thread);
169 const idle_stack_end_offset = std.mem.alignForward(usize, threads_size + idle_stack_size, std.heap.page_size_max);
170 const allocated_slice = try gpa.alignedAlloc(u8, .of(Thread), idle_stack_end_offset);
171 errdefer gpa.free(allocated_slice);
172 el.* = .{
173 .gpa = gpa,
174 .main_fiber_buffer = undefined,
175 .threads = .{
176 .allocated = @ptrCast(allocated_slice[0..threads_size]),
177 .reserved = 1,
178 .active = 1,
179 },
180 .detached = .{
181 .mutex = .init,
182 .list = .{},
183 },
184 };
185 const main_fiber: *Fiber = @ptrCast(&el.main_fiber_buffer);
186 main_fiber.* = .{
187 .required_align = {},
188 .context = undefined,
189 .awaiter = null,
190 .queue_next = null,
191 .cancel_thread = null,
192 .awaiting_completions = .initEmpty(),
193 };
194 const main_thread = &el.threads.allocated[0];
195 Thread.self = main_thread;
196 const idle_stack_end: [*]align(16) usize = @ptrCast(@alignCast(allocated_slice[idle_stack_end_offset..].ptr));
197 (idle_stack_end - 1)[0..1].* = .{@intFromPtr(el)};
198 main_thread.* = .{
199 .thread = undefined,
200 .idle_context = switch (builtin.cpu.arch) {
201 .aarch64 => .{
202 .sp = @intFromPtr(idle_stack_end),
203 .fp = 0,
204 .pc = @intFromPtr(&mainIdleEntry),
205 },
206 .x86_64 => .{
207 .rsp = @intFromPtr(idle_stack_end - 1),
208 .rbp = 0,
209 .rip = @intFromPtr(&mainIdleEntry),
210 },
211 else => @compileError("unimplemented architecture"),
212 },
213 .current_context = &main_fiber.context,
214 .ready_queue = null,
215 .io_uring = try IoUring.init(io_uring_entries, 0),
216 .idle_search_index = 1,
217 .steal_ready_search_index = 1,
218 };
219 errdefer main_thread.io_uring.deinit();
220 std.log.debug("created main idle {*}", .{&main_thread.idle_context});
221 std.log.debug("created main {*}", .{main_fiber});
222}
223
224pub fn deinit(el: *EventLoop) void {
225 while (true) cancel(el, detached_future: {
226 el.detached.mutex.lock(el.io()) catch |err| switch (err) {
227 error.Canceled => unreachable, // main fiber cannot be canceled
228 };
229 defer el.detached.mutex.unlock(el.io());
230 const detached: *DetachedClosure = @fieldParentPtr(
231 "detached_queue_node",
232 el.detached.list.pop() orelse break,
233 );
234 // notify the detached fiber that it is no longer allowed to recycle itself
235 detached.detached_queue_node = .{
236 .prev = &detached.detached_queue_node,
237 .next = &detached.detached_queue_node,
238 };
239 break :detached_future @ptrCast(detached.fiber);
240 }, &.{}, .@"1");
241 const active_threads = @atomicLoad(u32, &el.threads.active, .acquire);
242 for (el.threads.allocated[0..active_threads]) |*thread| {
243 const ready_fiber = @atomicLoad(?*Fiber, &thread.ready_queue, .monotonic);
244 assert(ready_fiber == null or ready_fiber == Fiber.finished); // pending async
245 }
246 el.yield(null, .exit);
247 const allocated_ptr: [*]align(@alignOf(Thread)) u8 = @ptrCast(@alignCast(el.threads.allocated.ptr));
248 const idle_stack_end_offset = std.mem.alignForward(usize, el.threads.allocated.len * @sizeOf(Thread) + idle_stack_size, std.heap.page_size_max);
249 for (el.threads.allocated[1..active_threads]) |*thread| thread.thread.join();
250 el.gpa.free(allocated_ptr[0..idle_stack_end_offset]);
251 el.* = undefined;
252}
253
254fn findReadyFiber(el: *EventLoop, thread: *Thread) ?*Fiber {
255 if (@atomicRmw(?*Fiber, &thread.ready_queue, .Xchg, Fiber.finished, .acquire)) |ready_fiber| {
256 @atomicStore(?*Fiber, &thread.ready_queue, ready_fiber.queue_next, .release);
257 ready_fiber.queue_next = null;
258 return ready_fiber;
259 }
260 const active_threads = @atomicLoad(u32, &el.threads.active, .acquire);
261 for (0..@min(max_steal_ready_search, active_threads)) |_| {
262 defer thread.steal_ready_search_index += 1;
263 if (thread.steal_ready_search_index == active_threads) thread.steal_ready_search_index = 0;
264 const steal_ready_search_thread = &el.threads.allocated[0..active_threads][thread.steal_ready_search_index];
265 if (steal_ready_search_thread == thread) continue;
266 const ready_fiber = @atomicLoad(?*Fiber, &steal_ready_search_thread.ready_queue, .acquire) orelse continue;
267 if (ready_fiber == Fiber.finished) continue;
268 if (@cmpxchgWeak(
269 ?*Fiber,
270 &steal_ready_search_thread.ready_queue,
271 ready_fiber,
272 null,
273 .acquire,
274 .monotonic,
275 )) |_| continue;
276 @atomicStore(?*Fiber, &thread.ready_queue, ready_fiber.queue_next, .release);
277 ready_fiber.queue_next = null;
278 return ready_fiber;
279 }
280 // couldn't find anything to do, so we are now open for business
281 @atomicStore(?*Fiber, &thread.ready_queue, null, .monotonic);
282 return null;
283}
284
285fn yield(el: *EventLoop, maybe_ready_fiber: ?*Fiber, pending_task: SwitchMessage.PendingTask) void {
286 const thread: *Thread = .current();
287 const ready_context = if (maybe_ready_fiber orelse el.findReadyFiber(thread)) |ready_fiber|
288 &ready_fiber.context
289 else
290 &thread.idle_context;
291 const message: SwitchMessage = .{
292 .contexts = .{
293 .prev = thread.current_context,
294 .ready = ready_context,
295 },
296 .pending_task = pending_task,
297 };
298 std.log.debug("switching from {*} to {*}", .{ message.contexts.prev, message.contexts.ready });
299 contextSwitch(&message).handle(el);
300}
301
302fn schedule(el: *EventLoop, thread: *Thread, ready_queue: Fiber.Queue) void {
303 {
304 var fiber = ready_queue.head;
305 while (true) {
306 std.log.debug("scheduling {*}", .{fiber});
307 fiber = fiber.queue_next orelse break;
308 }
309 assert(fiber == ready_queue.tail);
310 }
311 // shared fields of previous `Thread` must be initialized before later ones are marked as active
312 const new_thread_index = @atomicLoad(u32, &el.threads.active, .acquire);
313 for (0..@min(max_idle_search, new_thread_index)) |_| {
314 defer thread.idle_search_index += 1;
315 if (thread.idle_search_index == new_thread_index) thread.idle_search_index = 0;
316 const idle_search_thread = &el.threads.allocated[0..new_thread_index][thread.idle_search_index];
317 if (idle_search_thread == thread) continue;
318 if (@cmpxchgWeak(
319 ?*Fiber,
320 &idle_search_thread.ready_queue,
321 null,
322 ready_queue.head,
323 .release,
324 .monotonic,
325 )) |_| continue;
326 getSqe(&thread.io_uring).* = .{
327 .opcode = .MSG_RING,
328 .flags = std.os.linux.IOSQE_CQE_SKIP_SUCCESS,
329 .ioprio = 0,
330 .fd = idle_search_thread.io_uring.fd,
331 .off = @intFromEnum(Completion.UserData.wakeup),
332 .addr = 0,
333 .len = 0,
334 .rw_flags = 0,
335 .user_data = @intFromEnum(Completion.UserData.wakeup),
336 .buf_index = 0,
337 .personality = 0,
338 .splice_fd_in = 0,
339 .addr3 = 0,
340 .resv = 0,
341 };
342 return;
343 }
344 spawn_thread: {
345 // previous failed reservations must have completed before retrying
346 if (new_thread_index == el.threads.allocated.len or @cmpxchgWeak(
347 u32,
348 &el.threads.reserved,
349 new_thread_index,
350 new_thread_index + 1,
351 .acquire,
352 .monotonic,
353 ) != null) break :spawn_thread;
354 const new_thread = &el.threads.allocated[new_thread_index];
355 const next_thread_index = new_thread_index + 1;
356 new_thread.* = .{
357 .thread = undefined,
358 .idle_context = undefined,
359 .current_context = &new_thread.idle_context,
360 .ready_queue = ready_queue.head,
361 .io_uring = IoUring.init(io_uring_entries, 0) catch |err| {
362 @atomicStore(u32, &el.threads.reserved, new_thread_index, .release);
363 // no more access to `thread` after giving up reservation
364 std.log.warn("unable to create worker thread due to io_uring init failure: {s}", .{@errorName(err)});
365 break :spawn_thread;
366 },
367 .idle_search_index = 0,
368 .steal_ready_search_index = 0,
369 };
370 new_thread.thread = std.Thread.spawn(.{
371 .stack_size = idle_stack_size,
372 .allocator = el.gpa,
373 }, threadEntry, .{ el, new_thread_index }) catch |err| {
374 new_thread.io_uring.deinit();
375 @atomicStore(u32, &el.threads.reserved, new_thread_index, .release);
376 // no more access to `thread` after giving up reservation
377 std.log.warn("unable to create worker thread due spawn failure: {s}", .{@errorName(err)});
378 break :spawn_thread;
379 };
380 // shared fields of `Thread` must be initialized before being marked active
381 @atomicStore(u32, &el.threads.active, next_thread_index, .release);
382 return;
383 }
384 // nobody wanted it, so just queue it on ourselves
385 while (@cmpxchgWeak(
386 ?*Fiber,
387 &thread.ready_queue,
388 ready_queue.tail.queue_next,
389 ready_queue.head,
390 .acq_rel,
391 .acquire,
392 )) |old_head| ready_queue.tail.queue_next = old_head;
393}
394
395fn mainIdle(el: *EventLoop, message: *const SwitchMessage) callconv(.withStackAlign(.c, @max(@alignOf(Thread), @alignOf(Context)))) noreturn {
396 message.handle(el);
397 el.idle(&el.threads.allocated[0]);
398 el.yield(@ptrCast(&el.main_fiber_buffer), .nothing);
399 unreachable; // switched to dead fiber
400}
401
402fn threadEntry(el: *EventLoop, index: u32) void {
403 const thread: *Thread = &el.threads.allocated[index];
404 Thread.self = thread;
405 std.log.debug("created thread idle {*}", .{&thread.idle_context});
406 el.idle(thread);
407}
408
409const Completion = struct {
410 const UserData = enum(usize) {
411 unused,
412 wakeup,
413 cleanup,
414 exit,
415 /// *Fiber
416 _,
417 };
418 result: i32,
419 flags: u32,
420};
421
422fn idle(el: *EventLoop, thread: *Thread) void {
423 var maybe_ready_fiber: ?*Fiber = null;
424 while (true) {
425 while (maybe_ready_fiber orelse el.findReadyFiber(thread)) |ready_fiber| {
426 el.yield(ready_fiber, .nothing);
427 maybe_ready_fiber = null;
428 }
429 _ = thread.io_uring.submit_and_wait(1) catch |err| switch (err) {
430 error.SignalInterrupt => std.log.warn("submit_and_wait failed with SignalInterrupt", .{}),
431 else => |e| @panic(@errorName(e)),
432 };
433 var cqes_buffer: [io_uring_entries]std.os.linux.io_uring_cqe = undefined;
434 var maybe_ready_queue: ?Fiber.Queue = null;
435 for (cqes_buffer[0 .. thread.io_uring.copy_cqes(&cqes_buffer, 0) catch |err| switch (err) {
436 error.SignalInterrupt => cqes_len: {
437 std.log.warn("copy_cqes failed with SignalInterrupt", .{});
438 break :cqes_len 0;
439 },
440 else => |e| @panic(@errorName(e)),
441 }]) |cqe| switch (@as(Completion.UserData, @enumFromInt(cqe.user_data))) {
442 .unused => unreachable, // bad submission queued?
443 .wakeup => {},
444 .cleanup => @panic("failed to notify other threads that we are exiting"),
445 .exit => {
446 assert(maybe_ready_fiber == null and maybe_ready_queue == null); // pending async
447 return;
448 },
449 _ => switch (errno(cqe.res)) {
450 .INTR => getSqe(&thread.io_uring).* = .{
451 .opcode = .ASYNC_CANCEL,
452 .flags = std.os.linux.IOSQE_CQE_SKIP_SUCCESS,
453 .ioprio = 0,
454 .fd = 0,
455 .off = 0,
456 .addr = cqe.user_data,
457 .len = 0,
458 .rw_flags = 0,
459 .user_data = @intFromEnum(Completion.UserData.wakeup),
460 .buf_index = 0,
461 .personality = 0,
462 .splice_fd_in = 0,
463 .addr3 = 0,
464 .resv = 0,
465 },
466 else => {
467 const fiber: *Fiber = @ptrFromInt(cqe.user_data);
468 assert(fiber.queue_next == null);
469 fiber.resultPointer(Completion).* = .{
470 .result = cqe.res,
471 .flags = cqe.flags,
472 };
473 if (maybe_ready_fiber == null) maybe_ready_fiber = fiber else if (maybe_ready_queue) |*ready_queue| {
474 ready_queue.tail.queue_next = fiber;
475 ready_queue.tail = fiber;
476 } else maybe_ready_queue = .{ .head = fiber, .tail = fiber };
477 },
478 },
479 };
480 if (maybe_ready_queue) |ready_queue| el.schedule(thread, ready_queue);
481 }
482}
483
484const SwitchMessage = struct {
485 contexts: extern struct {
486 prev: *Context,
487 ready: *Context,
488 },
489 pending_task: PendingTask,
490
491 const PendingTask = union(enum) {
492 nothing,
493 reschedule,
494 recycle,
495 register_awaiter: *?*Fiber,
496 register_select: []const *Io.AnyFuture,
497 mutex_lock: struct {
498 prev_state: Io.Mutex.State,
499 mutex: *Io.Mutex,
500 },
501 condition_wait: struct {
502 cond: *Io.Condition,
503 mutex: *Io.Mutex,
504 },
505 exit,
506 };
507
508 fn handle(message: *const SwitchMessage, el: *EventLoop) void {
509 const thread: *Thread = .current();
510 thread.current_context = message.contexts.ready;
511 switch (message.pending_task) {
512 .nothing => {},
513 .reschedule => if (message.contexts.prev != &thread.idle_context) {
514 const prev_fiber: *Fiber = @alignCast(@fieldParentPtr("context", message.contexts.prev));
515 assert(prev_fiber.queue_next == null);
516 el.schedule(thread, .{ .head = prev_fiber, .tail = prev_fiber });
517 },
518 .recycle => {
519 const prev_fiber: *Fiber = @alignCast(@fieldParentPtr("context", message.contexts.prev));
520 assert(prev_fiber.queue_next == null);
521 el.recycle(prev_fiber);
522 },
523 .register_awaiter => |awaiter| {
524 const prev_fiber: *Fiber = @alignCast(@fieldParentPtr("context", message.contexts.prev));
525 assert(prev_fiber.queue_next == null);
526 if (@atomicRmw(?*Fiber, awaiter, .Xchg, prev_fiber, .acq_rel) == Fiber.finished)
527 el.schedule(thread, .{ .head = prev_fiber, .tail = prev_fiber });
528 },
529 .register_select => |futures| {
530 const prev_fiber: *Fiber = @alignCast(@fieldParentPtr("context", message.contexts.prev));
531 assert(prev_fiber.queue_next == null);
532 for (futures) |any_future| {
533 const future_fiber: *Fiber = @ptrCast(@alignCast(any_future));
534 if (@atomicRmw(?*Fiber, &future_fiber.awaiter, .Xchg, prev_fiber, .acq_rel) == Fiber.finished) {
535 const closure: *AsyncClosure = .fromFiber(future_fiber);
536 if (!@atomicRmw(bool, &closure.already_awaited, .Xchg, true, .seq_cst)) {
537 el.schedule(thread, .{ .head = prev_fiber, .tail = prev_fiber });
538 }
539 }
540 }
541 },
542 .mutex_lock => |mutex_lock| {
543 const prev_fiber: *Fiber = @alignCast(@fieldParentPtr("context", message.contexts.prev));
544 assert(prev_fiber.queue_next == null);
545 var prev_state = mutex_lock.prev_state;
546 while (switch (prev_state) {
547 else => next_state: {
548 prev_fiber.queue_next = @ptrFromInt(@intFromEnum(prev_state));
549 break :next_state @cmpxchgWeak(
550 Io.Mutex.State,
551 &mutex_lock.mutex.state,
552 prev_state,
553 @enumFromInt(@intFromPtr(prev_fiber)),
554 .release,
555 .acquire,
556 );
557 },
558 .unlocked => @cmpxchgWeak(
559 Io.Mutex.State,
560 &mutex_lock.mutex.state,
561 .unlocked,
562 .locked_once,
563 .acquire,
564 .acquire,
565 ) orelse {
566 prev_fiber.queue_next = null;
567 el.schedule(thread, .{ .head = prev_fiber, .tail = prev_fiber });
568 return;
569 },
570 }) |next_state| prev_state = next_state;
571 },
572 .condition_wait => |condition_wait| {
573 const prev_fiber: *Fiber = @alignCast(@fieldParentPtr("context", message.contexts.prev));
574 assert(prev_fiber.queue_next == null);
575 const cond_impl = prev_fiber.resultPointer(ConditionImpl);
576 cond_impl.* = .{
577 .tail = prev_fiber,
578 .event = .queued,
579 };
580 if (@cmpxchgStrong(
581 ?*Fiber,
582 @as(*?*Fiber, @ptrCast(&condition_wait.cond.state)),
583 null,
584 prev_fiber,
585 .release,
586 .acquire,
587 )) |waiting_fiber| {
588 const waiting_cond_impl = waiting_fiber.?.resultPointer(ConditionImpl);
589 assert(waiting_cond_impl.tail.queue_next == null);
590 waiting_cond_impl.tail.queue_next = prev_fiber;
591 waiting_cond_impl.tail = prev_fiber;
592 }
593 condition_wait.mutex.unlock(el.io());
594 },
595 .exit => for (el.threads.allocated[0..@atomicLoad(u32, &el.threads.active, .acquire)]) |*each_thread| {
596 getSqe(&thread.io_uring).* = .{
597 .opcode = .MSG_RING,
598 .flags = std.os.linux.IOSQE_CQE_SKIP_SUCCESS,
599 .ioprio = 0,
600 .fd = each_thread.io_uring.fd,
601 .off = @intFromEnum(Completion.UserData.exit),
602 .addr = 0,
603 .len = 0,
604 .rw_flags = 0,
605 .user_data = @intFromEnum(Completion.UserData.cleanup),
606 .buf_index = 0,
607 .personality = 0,
608 .splice_fd_in = 0,
609 .addr3 = 0,
610 .resv = 0,
611 };
612 },
613 }
614 }
615};
616
617const Context = switch (builtin.cpu.arch) {
618 .aarch64 => extern struct {
619 sp: u64,
620 fp: u64,
621 pc: u64,
622 },
623 .x86_64 => extern struct {
624 rsp: u64,
625 rbp: u64,
626 rip: u64,
627 },
628 else => |arch| @compileError("unimplemented architecture: " ++ @tagName(arch)),
629};
630
631inline fn contextSwitch(message: *const SwitchMessage) *const SwitchMessage {
632 return @fieldParentPtr("contexts", switch (builtin.cpu.arch) {
633 .aarch64 => asm volatile (
634 \\ ldp x0, x2, [x1]
635 \\ ldr x3, [x2, #16]
636 \\ mov x4, sp
637 \\ stp x4, fp, [x0]
638 \\ adr x5, 0f
639 \\ ldp x4, fp, [x2]
640 \\ str x5, [x0, #16]
641 \\ mov sp, x4
642 \\ br x3
643 \\0:
644 : [received_message] "={x1}" (-> *const @FieldType(SwitchMessage, "contexts")),
645 : [message_to_send] "{x1}" (&message.contexts),
646 : .{
647 .x0 = true,
648 .x1 = true,
649 .x2 = true,
650 .x3 = true,
651 .x4 = true,
652 .x5 = true,
653 .x6 = true,
654 .x7 = true,
655 .x8 = true,
656 .x9 = true,
657 .x10 = true,
658 .x11 = true,
659 .x12 = true,
660 .x13 = true,
661 .x14 = true,
662 .x15 = true,
663 .x16 = true,
664 .x17 = true,
665 .x18 = true,
666 .x19 = true,
667 .x20 = true,
668 .x21 = true,
669 .x22 = true,
670 .x23 = true,
671 .x24 = true,
672 .x25 = true,
673 .x26 = true,
674 .x27 = true,
675 .x28 = true,
676 .x30 = true,
677 .z0 = true,
678 .z1 = true,
679 .z2 = true,
680 .z3 = true,
681 .z4 = true,
682 .z5 = true,
683 .z6 = true,
684 .z7 = true,
685 .z8 = true,
686 .z9 = true,
687 .z10 = true,
688 .z11 = true,
689 .z12 = true,
690 .z13 = true,
691 .z14 = true,
692 .z15 = true,
693 .z16 = true,
694 .z17 = true,
695 .z18 = true,
696 .z19 = true,
697 .z20 = true,
698 .z21 = true,
699 .z22 = true,
700 .z23 = true,
701 .z24 = true,
702 .z25 = true,
703 .z26 = true,
704 .z27 = true,
705 .z28 = true,
706 .z29 = true,
707 .z30 = true,
708 .z31 = true,
709 .p0 = true,
710 .p1 = true,
711 .p2 = true,
712 .p3 = true,
713 .p4 = true,
714 .p5 = true,
715 .p6 = true,
716 .p7 = true,
717 .p8 = true,
718 .p9 = true,
719 .p10 = true,
720 .p11 = true,
721 .p12 = true,
722 .p13 = true,
723 .p14 = true,
724 .p15 = true,
725 .fpcr = true,
726 .fpsr = true,
727 .ffr = true,
728 .memory = true,
729 }),
730 .x86_64 => asm volatile (
731 \\ movq 0(%%rsi), %%rax
732 \\ movq 8(%%rsi), %%rcx
733 \\ leaq 0f(%%rip), %%rdx
734 \\ movq %%rsp, 0(%%rax)
735 \\ movq %%rbp, 8(%%rax)
736 \\ movq %%rdx, 16(%%rax)
737 \\ movq 0(%%rcx), %%rsp
738 \\ movq 8(%%rcx), %%rbp
739 \\ jmpq *16(%%rcx)
740 \\0:
741 : [received_message] "={rsi}" (-> *const @FieldType(SwitchMessage, "contexts")),
742 : [message_to_send] "{rsi}" (&message.contexts),
743 : .{
744 .rax = true,
745 .rcx = true,
746 .rdx = true,
747 .rbx = true,
748 .rsi = true,
749 .rdi = true,
750 .r8 = true,
751 .r9 = true,
752 .r10 = true,
753 .r11 = true,
754 .r12 = true,
755 .r13 = true,
756 .r14 = true,
757 .r15 = true,
758 .mm0 = true,
759 .mm1 = true,
760 .mm2 = true,
761 .mm3 = true,
762 .mm4 = true,
763 .mm5 = true,
764 .mm6 = true,
765 .mm7 = true,
766 .zmm0 = true,
767 .zmm1 = true,
768 .zmm2 = true,
769 .zmm3 = true,
770 .zmm4 = true,
771 .zmm5 = true,
772 .zmm6 = true,
773 .zmm7 = true,
774 .zmm8 = true,
775 .zmm9 = true,
776 .zmm10 = true,
777 .zmm11 = true,
778 .zmm12 = true,
779 .zmm13 = true,
780 .zmm14 = true,
781 .zmm15 = true,
782 .zmm16 = true,
783 .zmm17 = true,
784 .zmm18 = true,
785 .zmm19 = true,
786 .zmm20 = true,
787 .zmm21 = true,
788 .zmm22 = true,
789 .zmm23 = true,
790 .zmm24 = true,
791 .zmm25 = true,
792 .zmm26 = true,
793 .zmm27 = true,
794 .zmm28 = true,
795 .zmm29 = true,
796 .zmm30 = true,
797 .zmm31 = true,
798 .fpsr = true,
799 .fpcr = true,
800 .mxcsr = true,
801 .rflags = true,
802 .dirflag = true,
803 .memory = true,
804 }),
805 else => |arch| @compileError("unimplemented architecture: " ++ @tagName(arch)),
806 });
807}
808
809fn mainIdleEntry() callconv(.naked) void {
810 switch (builtin.cpu.arch) {
811 .x86_64 => asm volatile (
812 \\ movq (%%rsp), %%rdi
813 \\ jmp %[mainIdle:P]
814 :
815 : [mainIdle] "X" (&mainIdle),
816 ),
817 .aarch64 => asm volatile (
818 \\ ldr x0, [sp, #-8]
819 \\ b %[mainIdle]
820 :
821 : [mainIdle] "X" (&mainIdle),
822 ),
823 else => |arch| @compileError("unimplemented architecture: " ++ @tagName(arch)),
824 }
825}
826
827fn fiberEntry() callconv(.naked) void {
828 switch (builtin.cpu.arch) {
829 .x86_64 => asm volatile (
830 \\ leaq 8(%%rsp), %%rdi
831 \\ jmpq *(%%rsp)
832 ),
833 .aarch64 => asm volatile (
834 \\ mov x0, sp
835 \\ ldr x2, [sp, #-8]
836 \\ br x2
837 ),
838 else => |arch| @compileError("unimplemented architecture: " ++ @tagName(arch)),
839 }
840}
841
842const AsyncClosure = struct {
843 event_loop: *EventLoop,
844 fiber: *Fiber,
845 start: *const fn (context: *const anyopaque, result: *anyopaque) void,
846 result_align: Alignment,
847 already_awaited: bool,
848
849 fn contextPointer(closure: *AsyncClosure) [*]align(Fiber.max_context_align.toByteUnits()) u8 {
850 return @alignCast(@as([*]u8, @ptrCast(closure)) + @sizeOf(AsyncClosure));
851 }
852
853 fn call(closure: *AsyncClosure, message: *const SwitchMessage) callconv(.withStackAlign(.c, @alignOf(AsyncClosure))) noreturn {
854 message.handle(closure.event_loop);
855 const fiber = closure.fiber;
856 std.log.debug("{*} performing async", .{fiber});
857 closure.start(closure.contextPointer(), fiber.resultBytes(closure.result_align));
858 const awaiter = @atomicRmw(?*Fiber, &fiber.awaiter, .Xchg, Fiber.finished, .acq_rel);
859 const ready_awaiter = r: {
860 const a = awaiter orelse break :r null;
861 if (@atomicRmw(bool, &closure.already_awaited, .Xchg, true, .acq_rel)) break :r null;
862 break :r a;
863 };
864 closure.event_loop.yield(ready_awaiter, .nothing);
865 unreachable; // switched to dead fiber
866 }
867
868 fn fromFiber(fiber: *Fiber) *AsyncClosure {
869 return @ptrFromInt(Fiber.max_context_align.max(.of(AsyncClosure)).backward(
870 @intFromPtr(fiber.allocatedEnd()) - Fiber.max_context_size,
871 ) - @sizeOf(AsyncClosure));
872 }
873};
874
875fn async(
876 userdata: ?*anyopaque,
877 result: []u8,
878 result_alignment: Alignment,
879 context: []const u8,
880 context_alignment: Alignment,
881 start: *const fn (context: *const anyopaque, result: *anyopaque) void,
882) ?*std.Io.AnyFuture {
883 return concurrent(userdata, result.len, result_alignment, context, context_alignment, start) catch {
884 start(context.ptr, result.ptr);
885 return null;
886 };
887}
888
889fn concurrent(
890 userdata: ?*anyopaque,
891 result_len: usize,
892 result_alignment: Alignment,
893 context: []const u8,
894 context_alignment: Alignment,
895 start: *const fn (context: *const anyopaque, result: *anyopaque) void,
896) error{OutOfMemory}!*std.Io.AnyFuture {
897 assert(result_alignment.compare(.lte, Fiber.max_result_align)); // TODO
898 assert(context_alignment.compare(.lte, Fiber.max_context_align)); // TODO
899 assert(result_len <= Fiber.max_result_size); // TODO
900 assert(context.len <= Fiber.max_context_size); // TODO
901
902 const event_loop: *EventLoop = @ptrCast(@alignCast(userdata));
903 const fiber = try Fiber.allocate(event_loop);
904 std.log.debug("allocated {*}", .{fiber});
905
906 const closure: *AsyncClosure = .fromFiber(fiber);
907 const stack_end: [*]align(16) usize = @ptrCast(@alignCast(closure));
908 (stack_end - 1)[0..1].* = .{@intFromPtr(&AsyncClosure.call)};
909 fiber.* = .{
910 .required_align = {},
911 .context = switch (builtin.cpu.arch) {
912 .x86_64 => .{
913 .rsp = @intFromPtr(stack_end - 1),
914 .rbp = 0,
915 .rip = @intFromPtr(&fiberEntry),
916 },
917 .aarch64 => .{
918 .sp = @intFromPtr(stack_end),
919 .fp = 0,
920 .pc = @intFromPtr(&fiberEntry),
921 },
922 else => |arch| @compileError("unimplemented architecture: " ++ @tagName(arch)),
923 },
924 .awaiter = null,
925 .queue_next = null,
926 .cancel_thread = null,
927 .awaiting_completions = .initEmpty(),
928 };
929 closure.* = .{
930 .event_loop = event_loop,
931 .fiber = fiber,
932 .start = start,
933 .result_align = result_alignment,
934 .already_awaited = false,
935 };
936 @memcpy(closure.contextPointer(), context);
937
938 event_loop.schedule(.current(), .{ .head = fiber, .tail = fiber });
939 return @ptrCast(fiber);
940}
941
942const DetachedClosure = struct {
943 event_loop: *EventLoop,
944 fiber: *Fiber,
945 start: *const fn (context: *const anyopaque) void,
946 detached_queue_node: std.DoublyLinkedList.Node,
947
948 fn contextPointer(closure: *DetachedClosure) [*]align(Fiber.max_context_align.toByteUnits()) u8 {
949 return @alignCast(@as([*]u8, @ptrCast(closure)) + @sizeOf(DetachedClosure));
950 }
951
952 fn call(closure: *DetachedClosure, message: *const SwitchMessage) callconv(.withStackAlign(.c, @alignOf(DetachedClosure))) noreturn {
953 message.handle(closure.event_loop);
954 std.log.debug("{*} performing async detached", .{closure.fiber});
955 closure.start(closure.contextPointer());
956 const awaiter = @atomicRmw(?*Fiber, &closure.fiber.awaiter, .Xchg, Fiber.finished, .acq_rel);
957 closure.event_loop.yield(awaiter, pending_task: {
958 closure.event_loop.detached.mutex.lock(closure.event_loop.io()) catch |err| switch (err) {
959 error.Canceled => break :pending_task .nothing,
960 };
961 defer closure.event_loop.detached.mutex.unlock(closure.event_loop.io());
962 if (closure.detached_queue_node.next == &closure.detached_queue_node) break :pending_task .nothing;
963 closure.event_loop.detached.list.remove(&closure.detached_queue_node);
964 break :pending_task .recycle;
965 });
966 unreachable; // switched to dead fiber
967 }
968};
969
970fn asyncDetached(
971 userdata: ?*anyopaque,
972 context: []const u8,
973 context_alignment: std.mem.Alignment,
974 start: *const fn (context: *const anyopaque) void,
975) void {
976 assert(context_alignment.compare(.lte, Fiber.max_context_align)); // TODO
977 assert(context.len <= Fiber.max_context_size); // TODO
978
979 const event_loop: *EventLoop = @ptrCast(@alignCast(userdata));
980 const fiber = Fiber.allocate(event_loop) catch {
981 start(context.ptr);
982 return;
983 };
984 std.log.debug("allocated {*}", .{fiber});
985
986 const current_thread: *Thread = .current();
987 const closure: *DetachedClosure = @ptrFromInt(Fiber.max_context_align.max(.of(DetachedClosure)).backward(
988 @intFromPtr(fiber.allocatedEnd()) - Fiber.max_context_size,
989 ) - @sizeOf(DetachedClosure));
990 const stack_end: [*]align(16) usize = @ptrCast(@alignCast(closure));
991 (stack_end - 1)[0..1].* = .{@intFromPtr(&DetachedClosure.call)};
992 fiber.* = .{
993 .required_align = {},
994 .context = switch (builtin.cpu.arch) {
995 .x86_64 => .{
996 .rsp = @intFromPtr(stack_end - 1),
997 .rbp = 0,
998 .rip = @intFromPtr(&fiberEntry),
999 },
1000 .aarch64 => .{
1001 .sp = @intFromPtr(stack_end),
1002 .fp = 0,
1003 .pc = @intFromPtr(&fiberEntry),
1004 },
1005 else => |arch| @compileError("unimplemented architecture: " ++ @tagName(arch)),
1006 },
1007 .awaiter = null,
1008 .queue_next = null,
1009 .cancel_thread = null,
1010 .awaiting_completions = .initEmpty(),
1011 };
1012 closure.* = .{
1013 .event_loop = event_loop,
1014 .fiber = fiber,
1015 .start = start,
1016 .detached_queue_node = .{},
1017 };
1018 {
1019 event_loop.detached.mutex.lock(event_loop.io()) catch |err| switch (err) {
1020 error.Canceled => {
1021 event_loop.recycle(fiber);
1022 start(context.ptr);
1023 return;
1024 },
1025 };
1026 defer event_loop.detached.mutex.unlock(event_loop.io());
1027 event_loop.detached.list.append(&closure.detached_queue_node);
1028 }
1029 @memcpy(closure.contextPointer(), context);
1030
1031 event_loop.schedule(current_thread, .{ .head = fiber, .tail = fiber });
1032}
1033
1034fn await(
1035 userdata: ?*anyopaque,
1036 any_future: *std.Io.AnyFuture,
1037 result: []u8,
1038 result_alignment: Alignment,
1039) void {
1040 const event_loop: *EventLoop = @ptrCast(@alignCast(userdata));
1041 const future_fiber: *Fiber = @ptrCast(@alignCast(any_future));
1042 if (@atomicLoad(?*Fiber, &future_fiber.awaiter, .acquire) != Fiber.finished)
1043 event_loop.yield(null, .{ .register_awaiter = &future_fiber.awaiter });
1044 @memcpy(result, future_fiber.resultBytes(result_alignment));
1045 event_loop.recycle(future_fiber);
1046}
1047
1048fn select(userdata: ?*anyopaque, futures: []const *Io.AnyFuture) usize {
1049 const el: *EventLoop = @ptrCast(@alignCast(userdata));
1050
1051 // Optimization to avoid the yield below.
1052 for (futures, 0..) |any_future, i| {
1053 const future_fiber: *Fiber = @ptrCast(@alignCast(any_future));
1054 if (@atomicLoad(?*Fiber, &future_fiber.awaiter, .acquire) == Fiber.finished)
1055 return i;
1056 }
1057
1058 el.yield(null, .{ .register_select = futures });
1059
1060 std.log.debug("back from select yield", .{});
1061
1062 const my_thread: *Thread = .current();
1063 const my_fiber = my_thread.currentFiber();
1064 var result: ?usize = null;
1065
1066 for (futures, 0..) |any_future, i| {
1067 const future_fiber: *Fiber = @ptrCast(@alignCast(any_future));
1068 if (@cmpxchgStrong(?*Fiber, &future_fiber.awaiter, my_fiber, null, .seq_cst, .seq_cst)) |awaiter| {
1069 if (awaiter == Fiber.finished) {
1070 if (result == null) result = i;
1071 } else if (awaiter) |a| {
1072 const closure: *AsyncClosure = .fromFiber(a);
1073 closure.already_awaited = false;
1074 }
1075 } else {
1076 const closure: *AsyncClosure = .fromFiber(my_fiber);
1077 closure.already_awaited = false;
1078 }
1079 }
1080
1081 return result.?;
1082}
1083
1084fn cancel(
1085 userdata: ?*anyopaque,
1086 any_future: *std.Io.AnyFuture,
1087 result: []u8,
1088 result_alignment: Alignment,
1089) void {
1090 const future_fiber: *Fiber = @ptrCast(@alignCast(any_future));
1091 if (@atomicRmw(
1092 ?*Thread,
1093 &future_fiber.cancel_thread,
1094 .Xchg,
1095 Thread.canceling,
1096 .acq_rel,
1097 )) |cancel_thread| if (cancel_thread != Thread.canceling) {
1098 getSqe(&Thread.current().io_uring).* = .{
1099 .opcode = .MSG_RING,
1100 .flags = std.os.linux.IOSQE_CQE_SKIP_SUCCESS,
1101 .ioprio = 0,
1102 .fd = cancel_thread.io_uring.fd,
1103 .off = @intFromPtr(future_fiber),
1104 .addr = 0,
1105 .len = @bitCast(-@as(i32, @intFromEnum(std.os.linux.E.INTR))),
1106 .rw_flags = 0,
1107 .user_data = @intFromEnum(Completion.UserData.cleanup),
1108 .buf_index = 0,
1109 .personality = 0,
1110 .splice_fd_in = 0,
1111 .addr3 = 0,
1112 .resv = 0,
1113 };
1114 };
1115 await(userdata, any_future, result, result_alignment);
1116}
1117
1118fn cancelRequested(userdata: ?*anyopaque) bool {
1119 _ = userdata;
1120 return @atomicLoad(?*Thread, &Thread.current().currentFiber().cancel_thread, .acquire) == Thread.canceling;
1121}
1122
1123fn createFile(
1124 userdata: ?*anyopaque,
1125 dir: Io.Dir,
1126 sub_path: []const u8,
1127 flags: Io.File.CreateFlags,
1128) Io.File.OpenError!Io.File {
1129 const el: *EventLoop = @ptrCast(@alignCast(userdata));
1130 const thread: *Thread = .current();
1131 const iou = &thread.io_uring;
1132 const fiber = thread.currentFiber();
1133 try fiber.enterCancelRegion(thread);
1134
1135 const posix = std.posix;
1136 const sub_path_c = try posix.toPosixPath(sub_path);
1137
1138 var os_flags: posix.O = .{
1139 .ACCMODE = if (flags.read) .RDWR else .WRONLY,
1140 .CREAT = true,
1141 .TRUNC = flags.truncate,
1142 .EXCL = flags.exclusive,
1143 };
1144 if (@hasField(posix.O, "LARGEFILE")) os_flags.LARGEFILE = true;
1145 if (@hasField(posix.O, "CLOEXEC")) os_flags.CLOEXEC = true;
1146
1147 // Use the O locking flags if the os supports them to acquire the lock
1148 // atomically. Note that the NONBLOCK flag is removed after the openat()
1149 // call is successful.
1150 const has_flock_open_flags = @hasField(posix.O, "EXLOCK");
1151 if (has_flock_open_flags) switch (flags.lock) {
1152 .none => {},
1153 .shared => {
1154 os_flags.SHLOCK = true;
1155 os_flags.NONBLOCK = flags.lock_nonblocking;
1156 },
1157 .exclusive => {
1158 os_flags.EXLOCK = true;
1159 os_flags.NONBLOCK = flags.lock_nonblocking;
1160 },
1161 };
1162 const have_flock = @TypeOf(posix.system.flock) != void;
1163
1164 if (have_flock and !has_flock_open_flags and flags.lock != .none) {
1165 @panic("TODO");
1166 }
1167
1168 if (has_flock_open_flags and flags.lock_nonblocking) {
1169 @panic("TODO");
1170 }
1171
1172 getSqe(iou).* = .{
1173 .opcode = .OPENAT,
1174 .flags = 0,
1175 .ioprio = 0,
1176 .fd = dir.handle,
1177 .off = 0,
1178 .addr = @intFromPtr(&sub_path_c),
1179 .len = @intCast(flags.mode),
1180 .rw_flags = @bitCast(os_flags),
1181 .user_data = @intFromPtr(fiber),
1182 .buf_index = 0,
1183 .personality = 0,
1184 .splice_fd_in = 0,
1185 .addr3 = 0,
1186 .resv = 0,
1187 };
1188
1189 el.yield(null, .nothing);
1190 fiber.exitCancelRegion(thread);
1191
1192 const completion = fiber.resultPointer(Completion);
1193 switch (errno(completion.result)) {
1194 .SUCCESS => return .{ .handle = completion.result },
1195 .INTR => unreachable,
1196 .CANCELED => return error.Canceled,
1197
1198 .FAULT => unreachable,
1199 .INVAL => return error.BadPathName,
1200 .BADF => unreachable,
1201 .ACCES => return error.AccessDenied,
1202 .FBIG => return error.FileTooBig,
1203 .OVERFLOW => return error.FileTooBig,
1204 .ISDIR => return error.IsDir,
1205 .LOOP => return error.SymLinkLoop,
1206 .MFILE => return error.ProcessFdQuotaExceeded,
1207 .NAMETOOLONG => return error.NameTooLong,
1208 .NFILE => return error.SystemFdQuotaExceeded,
1209 .NODEV => return error.NoDevice,
1210 .NOENT => return error.FileNotFound,
1211 .NOMEM => return error.SystemResources,
1212 .NOSPC => return error.NoSpaceLeft,
1213 .NOTDIR => return error.NotDir,
1214 .PERM => return error.PermissionDenied,
1215 .EXIST => return error.PathAlreadyExists,
1216 .BUSY => return error.DeviceBusy,
1217 .OPNOTSUPP => return error.FileLocksNotSupported,
1218 .AGAIN => return error.WouldBlock,
1219 .TXTBSY => return error.FileBusy,
1220 .NXIO => return error.NoDevice,
1221 else => |err| return posix.unexpectedErrno(err),
1222 }
1223}
1224
1225fn fileOpen(
1226 userdata: ?*anyopaque,
1227 dir: Io.Dir,
1228 sub_path: []const u8,
1229 flags: Io.File.OpenFlags,
1230) Io.File.OpenError!Io.File {
1231 const el: *EventLoop = @ptrCast(@alignCast(userdata));
1232 const thread: *Thread = .current();
1233 const iou = &thread.io_uring;
1234 const fiber = thread.currentFiber();
1235 try fiber.enterCancelRegion(thread);
1236
1237 const posix = std.posix;
1238 const sub_path_c = try posix.toPosixPath(sub_path);
1239
1240 var os_flags: posix.O = .{
1241 .ACCMODE = switch (flags.mode) {
1242 .read_only => .RDONLY,
1243 .write_only => .WRONLY,
1244 .read_write => .RDWR,
1245 },
1246 };
1247
1248 if (@hasField(posix.O, "CLOEXEC")) os_flags.CLOEXEC = true;
1249 if (@hasField(posix.O, "LARGEFILE")) os_flags.LARGEFILE = true;
1250 if (@hasField(posix.O, "NOCTTY")) os_flags.NOCTTY = !flags.allow_ctty;
1251
1252 // Use the O locking flags if the os supports them to acquire the lock
1253 // atomically.
1254 const has_flock_open_flags = @hasField(posix.O, "EXLOCK");
1255 if (has_flock_open_flags) {
1256 // Note that the NONBLOCK flag is removed after the openat() call
1257 // is successful.
1258 switch (flags.lock) {
1259 .none => {},
1260 .shared => {
1261 os_flags.SHLOCK = true;
1262 os_flags.NONBLOCK = flags.lock_nonblocking;
1263 },
1264 .exclusive => {
1265 os_flags.EXLOCK = true;
1266 os_flags.NONBLOCK = flags.lock_nonblocking;
1267 },
1268 }
1269 }
1270 const have_flock = @TypeOf(posix.system.flock) != void;
1271
1272 if (have_flock and !has_flock_open_flags and flags.lock != .none) {
1273 @panic("TODO");
1274 }
1275
1276 if (has_flock_open_flags and flags.lock_nonblocking) {
1277 @panic("TODO");
1278 }
1279
1280 getSqe(iou).* = .{
1281 .opcode = .OPENAT,
1282 .flags = 0,
1283 .ioprio = 0,
1284 .fd = dir.handle,
1285 .off = 0,
1286 .addr = @intFromPtr(&sub_path_c),
1287 .len = 0,
1288 .rw_flags = @bitCast(os_flags),
1289 .user_data = @intFromPtr(fiber),
1290 .buf_index = 0,
1291 .personality = 0,
1292 .splice_fd_in = 0,
1293 .addr3 = 0,
1294 .resv = 0,
1295 };
1296
1297 el.yield(null, .nothing);
1298 fiber.exitCancelRegion(thread);
1299
1300 const completion = fiber.resultPointer(Completion);
1301 switch (errno(completion.result)) {
1302 .SUCCESS => return .{ .handle = completion.result },
1303 .INTR => unreachable,
1304 .CANCELED => return error.Canceled,
1305
1306 .FAULT => unreachable,
1307 .INVAL => return error.BadPathName,
1308 .BADF => unreachable,
1309 .ACCES => return error.AccessDenied,
1310 .FBIG => return error.FileTooBig,
1311 .OVERFLOW => return error.FileTooBig,
1312 .ISDIR => return error.IsDir,
1313 .LOOP => return error.SymLinkLoop,
1314 .MFILE => return error.ProcessFdQuotaExceeded,
1315 .NAMETOOLONG => return error.NameTooLong,
1316 .NFILE => return error.SystemFdQuotaExceeded,
1317 .NODEV => return error.NoDevice,
1318 .NOENT => return error.FileNotFound,
1319 .NOMEM => return error.SystemResources,
1320 .NOSPC => return error.NoSpaceLeft,
1321 .NOTDIR => return error.NotDir,
1322 .PERM => return error.PermissionDenied,
1323 .EXIST => return error.PathAlreadyExists,
1324 .BUSY => return error.DeviceBusy,
1325 .OPNOTSUPP => return error.FileLocksNotSupported,
1326 .AGAIN => return error.WouldBlock,
1327 .TXTBSY => return error.FileBusy,
1328 .NXIO => return error.NoDevice,
1329 else => |err| return posix.unexpectedErrno(err),
1330 }
1331}
1332
1333fn fileClose(userdata: ?*anyopaque, file: Io.File) void {
1334 const el: *EventLoop = @ptrCast(@alignCast(userdata));
1335 const thread: *Thread = .current();
1336 const iou = &thread.io_uring;
1337 const fiber = thread.currentFiber();
1338
1339 getSqe(iou).* = .{
1340 .opcode = .CLOSE,
1341 .flags = 0,
1342 .ioprio = 0,
1343 .fd = file.handle,
1344 .off = 0,
1345 .addr = 0,
1346 .len = 0,
1347 .rw_flags = 0,
1348 .user_data = @intFromPtr(fiber),
1349 .buf_index = 0,
1350 .personality = 0,
1351 .splice_fd_in = 0,
1352 .addr3 = 0,
1353 .resv = 0,
1354 };
1355
1356 el.yield(null, .nothing);
1357
1358 const completion = fiber.resultPointer(Completion);
1359 switch (errno(completion.result)) {
1360 .SUCCESS => return,
1361 .INTR => unreachable,
1362 .CANCELED => return,
1363
1364 .BADF => unreachable, // Always a race condition.
1365 else => return,
1366 }
1367}
1368
1369fn pread(userdata: ?*anyopaque, file: Io.File, buffer: []u8, offset: std.posix.off_t) Io.File.PReadError!usize {
1370 const el: *EventLoop = @ptrCast(@alignCast(userdata));
1371 const thread: *Thread = .current();
1372 const iou = &thread.io_uring;
1373 const fiber = thread.currentFiber();
1374 try fiber.enterCancelRegion(thread);
1375
1376 getSqe(iou).* = .{
1377 .opcode = .READ,
1378 .flags = 0,
1379 .ioprio = 0,
1380 .fd = file.handle,
1381 .off = @bitCast(offset),
1382 .addr = @intFromPtr(buffer.ptr),
1383 .len = @min(buffer.len, 0x7ffff000),
1384 .rw_flags = 0,
1385 .user_data = @intFromPtr(fiber),
1386 .buf_index = 0,
1387 .personality = 0,
1388 .splice_fd_in = 0,
1389 .addr3 = 0,
1390 .resv = 0,
1391 };
1392
1393 el.yield(null, .nothing);
1394 fiber.exitCancelRegion(thread);
1395
1396 const completion = fiber.resultPointer(Completion);
1397 switch (errno(completion.result)) {
1398 .SUCCESS => return @as(u32, @bitCast(completion.result)),
1399 .INTR => unreachable,
1400 .CANCELED => return error.Canceled,
1401
1402 .INVAL => unreachable,
1403 .FAULT => unreachable,
1404 .NOENT => return error.ProcessNotFound,
1405 .AGAIN => return error.WouldBlock,
1406 .BADF => return error.NotOpenForReading, // Can be a race condition.
1407 .IO => return error.InputOutput,
1408 .ISDIR => return error.IsDir,
1409 .NOBUFS => return error.SystemResources,
1410 .NOMEM => return error.SystemResources,
1411 .NOTCONN => return error.SocketUnconnected,
1412 .CONNRESET => return error.ConnectionResetByPeer,
1413 .TIMEDOUT => return error.Timeout,
1414 .NXIO => return error.Unseekable,
1415 .SPIPE => return error.Unseekable,
1416 .OVERFLOW => return error.Unseekable,
1417 else => |err| return std.posix.unexpectedErrno(err),
1418 }
1419}
1420
1421fn pwrite(userdata: ?*anyopaque, file: Io.File, buffer: []const u8, offset: std.posix.off_t) Io.File.PWriteError!usize {
1422 const el: *EventLoop = @ptrCast(@alignCast(userdata));
1423 const thread: *Thread = .current();
1424 const iou = &thread.io_uring;
1425 const fiber = thread.currentFiber();
1426 try fiber.enterCancelRegion(thread);
1427
1428 getSqe(iou).* = .{
1429 .opcode = .WRITE,
1430 .flags = 0,
1431 .ioprio = 0,
1432 .fd = file.handle,
1433 .off = @bitCast(offset),
1434 .addr = @intFromPtr(buffer.ptr),
1435 .len = @min(buffer.len, 0x7ffff000),
1436 .rw_flags = 0,
1437 .user_data = @intFromPtr(fiber),
1438 .buf_index = 0,
1439 .personality = 0,
1440 .splice_fd_in = 0,
1441 .addr3 = 0,
1442 .resv = 0,
1443 };
1444
1445 el.yield(null, .nothing);
1446 fiber.exitCancelRegion(thread);
1447
1448 const completion = fiber.resultPointer(Completion);
1449 switch (errno(completion.result)) {
1450 .SUCCESS => return @as(u32, @bitCast(completion.result)),
1451 .INTR => unreachable,
1452 .CANCELED => return error.Canceled,
1453
1454 .INVAL => return error.InvalidArgument,
1455 .FAULT => unreachable,
1456 .NOENT => return error.ProcessNotFound,
1457 .AGAIN => return error.WouldBlock,
1458 .BADF => return error.NotOpenForWriting, // can be a race condition.
1459 .DESTADDRREQ => unreachable, // `connect` was never called.
1460 .DQUOT => return error.DiskQuota,
1461 .FBIG => return error.FileTooBig,
1462 .IO => return error.InputOutput,
1463 .NOSPC => return error.NoSpaceLeft,
1464 .ACCES => return error.AccessDenied,
1465 .PERM => return error.PermissionDenied,
1466 .PIPE => return error.BrokenPipe,
1467 .NXIO => return error.Unseekable,
1468 .SPIPE => return error.Unseekable,
1469 .OVERFLOW => return error.Unseekable,
1470 .BUSY => return error.DeviceBusy,
1471 .CONNRESET => return error.ConnectionResetByPeer,
1472 .MSGSIZE => return error.MessageTooBig,
1473 else => |err| return std.posix.unexpectedErrno(err),
1474 }
1475}
1476
1477fn now(userdata: ?*anyopaque, clockid: std.posix.clockid_t) Io.ClockGetTimeError!Io.Timestamp {
1478 _ = userdata;
1479 const timespec = try std.posix.clock_gettime(clockid);
1480 return @enumFromInt(@as(i128, timespec.sec) * std.time.ns_per_s + timespec.nsec);
1481}
1482
1483fn sleep(userdata: ?*anyopaque, clockid: std.posix.clockid_t, deadline: Io.Deadline) Io.SleepError!void {
1484 const el: *EventLoop = @ptrCast(@alignCast(userdata));
1485 const thread: *Thread = .current();
1486 const iou = &thread.io_uring;
1487 const fiber = thread.currentFiber();
1488 try fiber.enterCancelRegion(thread);
1489
1490 const deadline_nanoseconds: i96 = switch (deadline) {
1491 .duration => |duration| duration.nanoseconds,
1492 .timestamp => |timestamp| @intFromEnum(timestamp),
1493 };
1494 const timespec: std.os.linux.kernel_timespec = .{
1495 .sec = @intCast(@divFloor(deadline_nanoseconds, std.time.ns_per_s)),
1496 .nsec = @intCast(@mod(deadline_nanoseconds, std.time.ns_per_s)),
1497 };
1498 getSqe(iou).* = .{
1499 .opcode = .TIMEOUT,
1500 .flags = 0,
1501 .ioprio = 0,
1502 .fd = 0,
1503 .off = 0,
1504 .addr = @intFromPtr(&timespec),
1505 .len = 1,
1506 .rw_flags = @as(u32, switch (deadline) {
1507 .duration => 0,
1508 .timestamp => std.os.linux.IORING_TIMEOUT_ABS,
1509 }) | @as(u32, switch (clockid) {
1510 .REALTIME => std.os.linux.IORING_TIMEOUT_REALTIME,
1511 .MONOTONIC => 0,
1512 .BOOTTIME => std.os.linux.IORING_TIMEOUT_BOOTTIME,
1513 else => return error.UnsupportedClock,
1514 }),
1515 .user_data = @intFromPtr(fiber),
1516 .buf_index = 0,
1517 .personality = 0,
1518 .splice_fd_in = 0,
1519 .addr3 = 0,
1520 .resv = 0,
1521 };
1522
1523 el.yield(null, .nothing);
1524 fiber.exitCancelRegion(thread);
1525
1526 const completion = fiber.resultPointer(Completion);
1527 switch (errno(completion.result)) {
1528 .SUCCESS, .TIME => return,
1529 .INTR => unreachable,
1530 .CANCELED => return error.Canceled,
1531
1532 else => |err| return std.posix.unexpectedErrno(err),
1533 }
1534}
1535
1536fn mutexLock(userdata: ?*anyopaque, prev_state: Io.Mutex.State, mutex: *Io.Mutex) error{Canceled}!void {
1537 const el: *EventLoop = @ptrCast(@alignCast(userdata));
1538 el.yield(null, .{ .mutex_lock = .{ .prev_state = prev_state, .mutex = mutex } });
1539}
1540fn mutexUnlock(userdata: ?*anyopaque, prev_state: Io.Mutex.State, mutex: *Io.Mutex) void {
1541 var maybe_waiting_fiber: ?*Fiber = @ptrFromInt(@intFromEnum(prev_state));
1542 while (if (maybe_waiting_fiber) |waiting_fiber| @cmpxchgWeak(
1543 Io.Mutex.State,
1544 &mutex.state,
1545 @enumFromInt(@intFromPtr(waiting_fiber)),
1546 @enumFromInt(@intFromPtr(waiting_fiber.queue_next)),
1547 .release,
1548 .acquire,
1549 ) else @cmpxchgWeak(
1550 Io.Mutex.State,
1551 &mutex.state,
1552 .locked_once,
1553 .unlocked,
1554 .release,
1555 .acquire,
1556 ) orelse return) |next_state| maybe_waiting_fiber = @ptrFromInt(@intFromEnum(next_state));
1557 maybe_waiting_fiber.?.queue_next = null;
1558 const el: *EventLoop = @ptrCast(@alignCast(userdata));
1559 el.yield(maybe_waiting_fiber.?, .reschedule);
1560}
1561
1562const ConditionImpl = struct {
1563 tail: *Fiber,
1564 event: union(enum) {
1565 queued,
1566 wake: Io.Condition.Wake,
1567 },
1568};
1569
1570fn conditionWait(userdata: ?*anyopaque, cond: *Io.Condition, mutex: *Io.Mutex) Io.Cancelable!void {
1571 const el: *EventLoop = @ptrCast(@alignCast(userdata));
1572 el.yield(null, .{ .condition_wait = .{ .cond = cond, .mutex = mutex } });
1573 const thread = Thread.current();
1574 const fiber = thread.currentFiber();
1575 const cond_impl = fiber.resultPointer(ConditionImpl);
1576 try mutex.lock(el.io());
1577 switch (cond_impl.event) {
1578 .queued => {},
1579 .wake => |wake| if (fiber.queue_next) |next_fiber| switch (wake) {
1580 .one => if (@cmpxchgStrong(
1581 ?*Fiber,
1582 @as(*?*Fiber, @ptrCast(&cond.state)),
1583 null,
1584 next_fiber,
1585 .release,
1586 .acquire,
1587 )) |old_fiber| {
1588 const old_cond_impl = old_fiber.?.resultPointer(ConditionImpl);
1589 assert(old_cond_impl.tail.queue_next == null);
1590 old_cond_impl.tail.queue_next = next_fiber;
1591 old_cond_impl.tail = cond_impl.tail;
1592 },
1593 .all => el.schedule(thread, .{ .head = next_fiber, .tail = cond_impl.tail }),
1594 },
1595 }
1596 fiber.queue_next = null;
1597}
1598
1599fn conditionWake(userdata: ?*anyopaque, cond: *Io.Condition, wake: Io.Condition.Wake) void {
1600 const el: *EventLoop = @ptrCast(@alignCast(userdata));
1601 const waiting_fiber = @atomicRmw(?*Fiber, @as(*?*Fiber, @ptrCast(&cond.state)), .Xchg, null, .acquire) orelse return;
1602 waiting_fiber.resultPointer(ConditionImpl).event = .{ .wake = wake };
1603 el.yield(waiting_fiber, .reschedule);
1604}
1605
1606fn errno(signed: i32) std.os.linux.E {
1607 return .init(@bitCast(@as(isize, signed)));
1608}
1609
1610fn getSqe(iou: *IoUring) *std.os.linux.io_uring_sqe {
1611 while (true) return iou.get_sqe() catch {
1612 _ = iou.submit_and_wait(0) catch |err| switch (err) {
1613 error.SignalInterrupt => std.log.warn("submit_and_wait failed with SignalInterrupt", .{}),
1614 else => |e| @panic(@errorName(e)),
1615 };
1616 continue;
1617 };
1618}
lib/std/Io/IoUring.zig created+1618
......@@ -0,0 +1,1618 @@
1const std = @import("../std.zig");
2const builtin = @import("builtin");
3const assert = std.debug.assert;
4const Allocator = std.mem.Allocator;
5const Io = std.Io;
6const EventLoop = @This();
7const Alignment = std.mem.Alignment;
8const IoUring = std.os.linux.IoUring;
9
10/// Must be a thread-safe allocator.
11gpa: Allocator,
12main_fiber_buffer: [@sizeOf(Fiber) + Fiber.max_result_size]u8 align(@alignOf(Fiber)),
13threads: Thread.List,
14detached: struct {
15 mutex: std.Io.Mutex,
16 list: std.DoublyLinkedList,
17},
18
19/// Empirically saw >128KB being used by the self-hosted backend to panic.
20const idle_stack_size = 256 * 1024;
21
22const max_idle_search = 4;
23const max_steal_ready_search = 4;
24
25const io_uring_entries = 64;
26
27const Thread = struct {
28 thread: std.Thread,
29 idle_context: Context,
30 current_context: *Context,
31 ready_queue: ?*Fiber,
32 io_uring: IoUring,
33 idle_search_index: u32,
34 steal_ready_search_index: u32,
35
36 const canceling: ?*Thread = @ptrFromInt(@alignOf(Thread));
37
38 threadlocal var self: *Thread = undefined;
39
40 fn current() *Thread {
41 return self;
42 }
43
44 fn currentFiber(thread: *Thread) *Fiber {
45 return @fieldParentPtr("context", thread.current_context);
46 }
47
48 const List = struct {
49 allocated: []Thread,
50 reserved: u32,
51 active: u32,
52 };
53};
54
55const Fiber = struct {
56 required_align: void align(4),
57 context: Context,
58 awaiter: ?*Fiber,
59 queue_next: ?*Fiber,
60 cancel_thread: ?*Thread,
61 awaiting_completions: std.StaticBitSet(3),
62
63 const finished: ?*Fiber = @ptrFromInt(@alignOf(Thread));
64
65 const max_result_align: Alignment = .@"16";
66 const max_result_size = max_result_align.forward(64);
67 /// This includes any stack realignments that need to happen, and also the
68 /// initial frame return address slot and argument frame, depending on target.
69 const min_stack_size = 4 * 1024 * 1024;
70 const max_context_align: Alignment = .@"16";
71 const max_context_size = max_context_align.forward(1024);
72 const max_closure_size: usize = @max(@sizeOf(AsyncClosure), @sizeOf(DetachedClosure));
73 const max_closure_align: Alignment = .max(.of(AsyncClosure), .of(DetachedClosure));
74 const allocation_size = std.mem.alignForward(
75 usize,
76 max_closure_align.max(max_context_align).forward(
77 max_result_align.forward(@sizeOf(Fiber)) + max_result_size + min_stack_size,
78 ) + max_closure_size + max_context_size,
79 std.heap.page_size_max,
80 );
81
82 fn allocate(el: *EventLoop) error{OutOfMemory}!*Fiber {
83 return @ptrCast(try el.gpa.alignedAlloc(u8, .of(Fiber), allocation_size));
84 }
85
86 fn allocatedSlice(f: *Fiber) []align(@alignOf(Fiber)) u8 {
87 return @as([*]align(@alignOf(Fiber)) u8, @ptrCast(f))[0..allocation_size];
88 }
89
90 fn allocatedEnd(f: *Fiber) [*]u8 {
91 const allocated_slice = f.allocatedSlice();
92 return allocated_slice[allocated_slice.len..].ptr;
93 }
94
95 fn resultPointer(f: *Fiber, comptime Result: type) *Result {
96 return @ptrCast(@alignCast(f.resultBytes(.of(Result))));
97 }
98
99 fn resultBytes(f: *Fiber, alignment: Alignment) [*]u8 {
100 return @ptrFromInt(alignment.forward(@intFromPtr(f) + @sizeOf(Fiber)));
101 }
102
103 fn enterCancelRegion(fiber: *Fiber, thread: *Thread) error{Canceled}!void {
104 if (@cmpxchgStrong(
105 ?*Thread,
106 &fiber.cancel_thread,
107 null,
108 thread,
109 .acq_rel,
110 .acquire,
111 )) |cancel_thread| {
112 assert(cancel_thread == Thread.canceling);
113 return error.Canceled;
114 }
115 }
116
117 fn exitCancelRegion(fiber: *Fiber, thread: *Thread) void {
118 if (@cmpxchgStrong(
119 ?*Thread,
120 &fiber.cancel_thread,
121 thread,
122 null,
123 .acq_rel,
124 .acquire,
125 )) |cancel_thread| assert(cancel_thread == Thread.canceling);
126 }
127
128 const Queue = struct { head: *Fiber, tail: *Fiber };
129};
130
131fn recycle(el: *EventLoop, fiber: *Fiber) void {
132 std.log.debug("recyling {*}", .{fiber});
133 assert(fiber.queue_next == null);
134 el.gpa.free(fiber.allocatedSlice());
135}
136
137pub fn io(el: *EventLoop) Io {
138 return .{
139 .userdata = el,
140 .vtable = &.{
141 .async = async,
142 .concurrent = concurrent,
143 .await = await,
144 .asyncDetached = asyncDetached,
145 .select = select,
146 .cancel = cancel,
147 .cancelRequested = cancelRequested,
148
149 .mutexLock = mutexLock,
150 .mutexUnlock = mutexUnlock,
151
152 .conditionWait = conditionWait,
153 .conditionWake = conditionWake,
154
155 .createFile = createFile,
156 .fileOpen = fileOpen,
157 .fileClose = fileClose,
158 .pread = pread,
159 .pwrite = pwrite,
160
161 .now = now,
162 .sleep = sleep,
163 },
164 };
165}
166
167pub fn init(el: *EventLoop, gpa: Allocator) !void {
168 const threads_size = @max(std.Thread.getCpuCount() catch 1, 1) * @sizeOf(Thread);
169 const idle_stack_end_offset = std.mem.alignForward(usize, threads_size + idle_stack_size, std.heap.page_size_max);
170 const allocated_slice = try gpa.alignedAlloc(u8, .of(Thread), idle_stack_end_offset);
171 errdefer gpa.free(allocated_slice);
172 el.* = .{
173 .gpa = gpa,
174 .main_fiber_buffer = undefined,
175 .threads = .{
176 .allocated = @ptrCast(allocated_slice[0..threads_size]),
177 .reserved = 1,
178 .active = 1,
179 },
180 .detached = .{
181 .mutex = .init,
182 .list = .{},
183 },
184 };
185 const main_fiber: *Fiber = @ptrCast(&el.main_fiber_buffer);
186 main_fiber.* = .{
187 .required_align = {},
188 .context = undefined,
189 .awaiter = null,
190 .queue_next = null,
191 .cancel_thread = null,
192 .awaiting_completions = .initEmpty(),
193 };
194 const main_thread = &el.threads.allocated[0];
195 Thread.self = main_thread;
196 const idle_stack_end: [*]align(16) usize = @ptrCast(@alignCast(allocated_slice[idle_stack_end_offset..].ptr));
197 (idle_stack_end - 1)[0..1].* = .{@intFromPtr(el)};
198 main_thread.* = .{
199 .thread = undefined,
200 .idle_context = switch (builtin.cpu.arch) {
201 .aarch64 => .{
202 .sp = @intFromPtr(idle_stack_end),
203 .fp = 0,
204 .pc = @intFromPtr(&mainIdleEntry),
205 },
206 .x86_64 => .{
207 .rsp = @intFromPtr(idle_stack_end - 1),
208 .rbp = 0,
209 .rip = @intFromPtr(&mainIdleEntry),
210 },
211 else => @compileError("unimplemented architecture"),
212 },
213 .current_context = &main_fiber.context,
214 .ready_queue = null,
215 .io_uring = try IoUring.init(io_uring_entries, 0),
216 .idle_search_index = 1,
217 .steal_ready_search_index = 1,
218 };
219 errdefer main_thread.io_uring.deinit();
220 std.log.debug("created main idle {*}", .{&main_thread.idle_context});
221 std.log.debug("created main {*}", .{main_fiber});
222}
223
224pub fn deinit(el: *EventLoop) void {
225 while (true) cancel(el, detached_future: {
226 el.detached.mutex.lock(el.io()) catch |err| switch (err) {
227 error.Canceled => unreachable, // main fiber cannot be canceled
228 };
229 defer el.detached.mutex.unlock(el.io());
230 const detached: *DetachedClosure = @fieldParentPtr(
231 "detached_queue_node",
232 el.detached.list.pop() orelse break,
233 );
234 // notify the detached fiber that it is no longer allowed to recycle itself
235 detached.detached_queue_node = .{
236 .prev = &detached.detached_queue_node,
237 .next = &detached.detached_queue_node,
238 };
239 break :detached_future @ptrCast(detached.fiber);
240 }, &.{}, .@"1");
241 const active_threads = @atomicLoad(u32, &el.threads.active, .acquire);
242 for (el.threads.allocated[0..active_threads]) |*thread| {
243 const ready_fiber = @atomicLoad(?*Fiber, &thread.ready_queue, .monotonic);
244 assert(ready_fiber == null or ready_fiber == Fiber.finished); // pending async
245 }
246 el.yield(null, .exit);
247 const allocated_ptr: [*]align(@alignOf(Thread)) u8 = @ptrCast(@alignCast(el.threads.allocated.ptr));
248 const idle_stack_end_offset = std.mem.alignForward(usize, el.threads.allocated.len * @sizeOf(Thread) + idle_stack_size, std.heap.page_size_max);
249 for (el.threads.allocated[1..active_threads]) |*thread| thread.thread.join();
250 el.gpa.free(allocated_ptr[0..idle_stack_end_offset]);
251 el.* = undefined;
252}
253
254fn findReadyFiber(el: *EventLoop, thread: *Thread) ?*Fiber {
255 if (@atomicRmw(?*Fiber, &thread.ready_queue, .Xchg, Fiber.finished, .acquire)) |ready_fiber| {
256 @atomicStore(?*Fiber, &thread.ready_queue, ready_fiber.queue_next, .release);
257 ready_fiber.queue_next = null;
258 return ready_fiber;
259 }
260 const active_threads = @atomicLoad(u32, &el.threads.active, .acquire);
261 for (0..@min(max_steal_ready_search, active_threads)) |_| {
262 defer thread.steal_ready_search_index += 1;
263 if (thread.steal_ready_search_index == active_threads) thread.steal_ready_search_index = 0;
264 const steal_ready_search_thread = &el.threads.allocated[0..active_threads][thread.steal_ready_search_index];
265 if (steal_ready_search_thread == thread) continue;
266 const ready_fiber = @atomicLoad(?*Fiber, &steal_ready_search_thread.ready_queue, .acquire) orelse continue;
267 if (ready_fiber == Fiber.finished) continue;
268 if (@cmpxchgWeak(
269 ?*Fiber,
270 &steal_ready_search_thread.ready_queue,
271 ready_fiber,
272 null,
273 .acquire,
274 .monotonic,
275 )) |_| continue;
276 @atomicStore(?*Fiber, &thread.ready_queue, ready_fiber.queue_next, .release);
277 ready_fiber.queue_next = null;
278 return ready_fiber;
279 }
280 // couldn't find anything to do, so we are now open for business
281 @atomicStore(?*Fiber, &thread.ready_queue, null, .monotonic);
282 return null;
283}
284
285fn yield(el: *EventLoop, maybe_ready_fiber: ?*Fiber, pending_task: SwitchMessage.PendingTask) void {
286 const thread: *Thread = .current();
287 const ready_context = if (maybe_ready_fiber orelse el.findReadyFiber(thread)) |ready_fiber|
288 &ready_fiber.context
289 else
290 &thread.idle_context;
291 const message: SwitchMessage = .{
292 .contexts = .{
293 .prev = thread.current_context,
294 .ready = ready_context,
295 },
296 .pending_task = pending_task,
297 };
298 std.log.debug("switching from {*} to {*}", .{ message.contexts.prev, message.contexts.ready });
299 contextSwitch(&message).handle(el);
300}
301
302fn schedule(el: *EventLoop, thread: *Thread, ready_queue: Fiber.Queue) void {
303 {
304 var fiber = ready_queue.head;
305 while (true) {
306 std.log.debug("scheduling {*}", .{fiber});
307 fiber = fiber.queue_next orelse break;
308 }
309 assert(fiber == ready_queue.tail);
310 }
311 // shared fields of previous `Thread` must be initialized before later ones are marked as active
312 const new_thread_index = @atomicLoad(u32, &el.threads.active, .acquire);
313 for (0..@min(max_idle_search, new_thread_index)) |_| {
314 defer thread.idle_search_index += 1;
315 if (thread.idle_search_index == new_thread_index) thread.idle_search_index = 0;
316 const idle_search_thread = &el.threads.allocated[0..new_thread_index][thread.idle_search_index];
317 if (idle_search_thread == thread) continue;
318 if (@cmpxchgWeak(
319 ?*Fiber,
320 &idle_search_thread.ready_queue,
321 null,
322 ready_queue.head,
323 .release,
324 .monotonic,
325 )) |_| continue;
326 getSqe(&thread.io_uring).* = .{
327 .opcode = .MSG_RING,
328 .flags = std.os.linux.IOSQE_CQE_SKIP_SUCCESS,
329 .ioprio = 0,
330 .fd = idle_search_thread.io_uring.fd,
331 .off = @intFromEnum(Completion.UserData.wakeup),
332 .addr = 0,
333 .len = 0,
334 .rw_flags = 0,
335 .user_data = @intFromEnum(Completion.UserData.wakeup),
336 .buf_index = 0,
337 .personality = 0,
338 .splice_fd_in = 0,
339 .addr3 = 0,
340 .resv = 0,
341 };
342 return;
343 }
344 spawn_thread: {
345 // previous failed reservations must have completed before retrying
346 if (new_thread_index == el.threads.allocated.len or @cmpxchgWeak(
347 u32,
348 &el.threads.reserved,
349 new_thread_index,
350 new_thread_index + 1,
351 .acquire,
352 .monotonic,
353 ) != null) break :spawn_thread;
354 const new_thread = &el.threads.allocated[new_thread_index];
355 const next_thread_index = new_thread_index + 1;
356 new_thread.* = .{
357 .thread = undefined,
358 .idle_context = undefined,
359 .current_context = &new_thread.idle_context,
360 .ready_queue = ready_queue.head,
361 .io_uring = IoUring.init(io_uring_entries, 0) catch |err| {
362 @atomicStore(u32, &el.threads.reserved, new_thread_index, .release);
363 // no more access to `thread` after giving up reservation
364 std.log.warn("unable to create worker thread due to io_uring init failure: {s}", .{@errorName(err)});
365 break :spawn_thread;
366 },
367 .idle_search_index = 0,
368 .steal_ready_search_index = 0,
369 };
370 new_thread.thread = std.Thread.spawn(.{
371 .stack_size = idle_stack_size,
372 .allocator = el.gpa,
373 }, threadEntry, .{ el, new_thread_index }) catch |err| {
374 new_thread.io_uring.deinit();
375 @atomicStore(u32, &el.threads.reserved, new_thread_index, .release);
376 // no more access to `thread` after giving up reservation
377 std.log.warn("unable to create worker thread due spawn failure: {s}", .{@errorName(err)});
378 break :spawn_thread;
379 };
380 // shared fields of `Thread` must be initialized before being marked active
381 @atomicStore(u32, &el.threads.active, next_thread_index, .release);
382 return;
383 }
384 // nobody wanted it, so just queue it on ourselves
385 while (@cmpxchgWeak(
386 ?*Fiber,
387 &thread.ready_queue,
388 ready_queue.tail.queue_next,
389 ready_queue.head,
390 .acq_rel,
391 .acquire,
392 )) |old_head| ready_queue.tail.queue_next = old_head;
393}
394
395fn mainIdle(el: *EventLoop, message: *const SwitchMessage) callconv(.withStackAlign(.c, @max(@alignOf(Thread), @alignOf(Context)))) noreturn {
396 message.handle(el);
397 el.idle(&el.threads.allocated[0]);
398 el.yield(@ptrCast(&el.main_fiber_buffer), .nothing);
399 unreachable; // switched to dead fiber
400}
401
402fn threadEntry(el: *EventLoop, index: u32) void {
403 const thread: *Thread = &el.threads.allocated[index];
404 Thread.self = thread;
405 std.log.debug("created thread idle {*}", .{&thread.idle_context});
406 el.idle(thread);
407}
408
409const Completion = struct {
410 const UserData = enum(usize) {
411 unused,
412 wakeup,
413 cleanup,
414 exit,
415 /// *Fiber
416 _,
417 };
418 result: i32,
419 flags: u32,
420};
421
422fn idle(el: *EventLoop, thread: *Thread) void {
423 var maybe_ready_fiber: ?*Fiber = null;
424 while (true) {
425 while (maybe_ready_fiber orelse el.findReadyFiber(thread)) |ready_fiber| {
426 el.yield(ready_fiber, .nothing);
427 maybe_ready_fiber = null;
428 }
429 _ = thread.io_uring.submit_and_wait(1) catch |err| switch (err) {
430 error.SignalInterrupt => std.log.warn("submit_and_wait failed with SignalInterrupt", .{}),
431 else => |e| @panic(@errorName(e)),
432 };
433 var cqes_buffer: [io_uring_entries]std.os.linux.io_uring_cqe = undefined;
434 var maybe_ready_queue: ?Fiber.Queue = null;
435 for (cqes_buffer[0 .. thread.io_uring.copy_cqes(&cqes_buffer, 0) catch |err| switch (err) {
436 error.SignalInterrupt => cqes_len: {
437 std.log.warn("copy_cqes failed with SignalInterrupt", .{});
438 break :cqes_len 0;
439 },
440 else => |e| @panic(@errorName(e)),
441 }]) |cqe| switch (@as(Completion.UserData, @enumFromInt(cqe.user_data))) {
442 .unused => unreachable, // bad submission queued?
443 .wakeup => {},
444 .cleanup => @panic("failed to notify other threads that we are exiting"),
445 .exit => {
446 assert(maybe_ready_fiber == null and maybe_ready_queue == null); // pending async
447 return;
448 },
449 _ => switch (errno(cqe.res)) {
450 .INTR => getSqe(&thread.io_uring).* = .{
451 .opcode = .ASYNC_CANCEL,
452 .flags = std.os.linux.IOSQE_CQE_SKIP_SUCCESS,
453 .ioprio = 0,
454 .fd = 0,
455 .off = 0,
456 .addr = cqe.user_data,
457 .len = 0,
458 .rw_flags = 0,
459 .user_data = @intFromEnum(Completion.UserData.wakeup),
460 .buf_index = 0,
461 .personality = 0,
462 .splice_fd_in = 0,
463 .addr3 = 0,
464 .resv = 0,
465 },
466 else => {
467 const fiber: *Fiber = @ptrFromInt(cqe.user_data);
468 assert(fiber.queue_next == null);
469 fiber.resultPointer(Completion).* = .{
470 .result = cqe.res,
471 .flags = cqe.flags,
472 };
473 if (maybe_ready_fiber == null) maybe_ready_fiber = fiber else if (maybe_ready_queue) |*ready_queue| {
474 ready_queue.tail.queue_next = fiber;
475 ready_queue.tail = fiber;
476 } else maybe_ready_queue = .{ .head = fiber, .tail = fiber };
477 },
478 },
479 };
480 if (maybe_ready_queue) |ready_queue| el.schedule(thread, ready_queue);
481 }
482}
483
484const SwitchMessage = struct {
485 contexts: extern struct {
486 prev: *Context,
487 ready: *Context,
488 },
489 pending_task: PendingTask,
490
491 const PendingTask = union(enum) {
492 nothing,
493 reschedule,
494 recycle,
495 register_awaiter: *?*Fiber,
496 register_select: []const *Io.AnyFuture,
497 mutex_lock: struct {
498 prev_state: Io.Mutex.State,
499 mutex: *Io.Mutex,
500 },
501 condition_wait: struct {
502 cond: *Io.Condition,
503 mutex: *Io.Mutex,
504 },
505 exit,
506 };
507
508 fn handle(message: *const SwitchMessage, el: *EventLoop) void {
509 const thread: *Thread = .current();
510 thread.current_context = message.contexts.ready;
511 switch (message.pending_task) {
512 .nothing => {},
513 .reschedule => if (message.contexts.prev != &thread.idle_context) {
514 const prev_fiber: *Fiber = @alignCast(@fieldParentPtr("context", message.contexts.prev));
515 assert(prev_fiber.queue_next == null);
516 el.schedule(thread, .{ .head = prev_fiber, .tail = prev_fiber });
517 },
518 .recycle => {
519 const prev_fiber: *Fiber = @alignCast(@fieldParentPtr("context", message.contexts.prev));
520 assert(prev_fiber.queue_next == null);
521 el.recycle(prev_fiber);
522 },
523 .register_awaiter => |awaiter| {
524 const prev_fiber: *Fiber = @alignCast(@fieldParentPtr("context", message.contexts.prev));
525 assert(prev_fiber.queue_next == null);
526 if (@atomicRmw(?*Fiber, awaiter, .Xchg, prev_fiber, .acq_rel) == Fiber.finished)
527 el.schedule(thread, .{ .head = prev_fiber, .tail = prev_fiber });
528 },
529 .register_select => |futures| {
530 const prev_fiber: *Fiber = @alignCast(@fieldParentPtr("context", message.contexts.prev));
531 assert(prev_fiber.queue_next == null);
532 for (futures) |any_future| {
533 const future_fiber: *Fiber = @ptrCast(@alignCast(any_future));
534 if (@atomicRmw(?*Fiber, &future_fiber.awaiter, .Xchg, prev_fiber, .acq_rel) == Fiber.finished) {
535 const closure: *AsyncClosure = .fromFiber(future_fiber);
536 if (!@atomicRmw(bool, &closure.already_awaited, .Xchg, true, .seq_cst)) {
537 el.schedule(thread, .{ .head = prev_fiber, .tail = prev_fiber });
538 }
539 }
540 }
541 },
542 .mutex_lock => |mutex_lock| {
543 const prev_fiber: *Fiber = @alignCast(@fieldParentPtr("context", message.contexts.prev));
544 assert(prev_fiber.queue_next == null);
545 var prev_state = mutex_lock.prev_state;
546 while (switch (prev_state) {
547 else => next_state: {
548 prev_fiber.queue_next = @ptrFromInt(@intFromEnum(prev_state));
549 break :next_state @cmpxchgWeak(
550 Io.Mutex.State,
551 &mutex_lock.mutex.state,
552 prev_state,
553 @enumFromInt(@intFromPtr(prev_fiber)),
554 .release,
555 .acquire,
556 );
557 },
558 .unlocked => @cmpxchgWeak(
559 Io.Mutex.State,
560 &mutex_lock.mutex.state,
561 .unlocked,
562 .locked_once,
563 .acquire,
564 .acquire,
565 ) orelse {
566 prev_fiber.queue_next = null;
567 el.schedule(thread, .{ .head = prev_fiber, .tail = prev_fiber });
568 return;
569 },
570 }) |next_state| prev_state = next_state;
571 },
572 .condition_wait => |condition_wait| {
573 const prev_fiber: *Fiber = @alignCast(@fieldParentPtr("context", message.contexts.prev));
574 assert(prev_fiber.queue_next == null);
575 const cond_impl = prev_fiber.resultPointer(ConditionImpl);
576 cond_impl.* = .{
577 .tail = prev_fiber,
578 .event = .queued,
579 };
580 if (@cmpxchgStrong(
581 ?*Fiber,
582 @as(*?*Fiber, @ptrCast(&condition_wait.cond.state)),
583 null,
584 prev_fiber,
585 .release,
586 .acquire,
587 )) |waiting_fiber| {
588 const waiting_cond_impl = waiting_fiber.?.resultPointer(ConditionImpl);
589 assert(waiting_cond_impl.tail.queue_next == null);
590 waiting_cond_impl.tail.queue_next = prev_fiber;
591 waiting_cond_impl.tail = prev_fiber;
592 }
593 condition_wait.mutex.unlock(el.io());
594 },
595 .exit => for (el.threads.allocated[0..@atomicLoad(u32, &el.threads.active, .acquire)]) |*each_thread| {
596 getSqe(&thread.io_uring).* = .{
597 .opcode = .MSG_RING,
598 .flags = std.os.linux.IOSQE_CQE_SKIP_SUCCESS,
599 .ioprio = 0,
600 .fd = each_thread.io_uring.fd,
601 .off = @intFromEnum(Completion.UserData.exit),
602 .addr = 0,
603 .len = 0,
604 .rw_flags = 0,
605 .user_data = @intFromEnum(Completion.UserData.cleanup),
606 .buf_index = 0,
607 .personality = 0,
608 .splice_fd_in = 0,
609 .addr3 = 0,
610 .resv = 0,
611 };
612 },
613 }
614 }
615};
616
617const Context = switch (builtin.cpu.arch) {
618 .aarch64 => extern struct {
619 sp: u64,
620 fp: u64,
621 pc: u64,
622 },
623 .x86_64 => extern struct {
624 rsp: u64,
625 rbp: u64,
626 rip: u64,
627 },
628 else => |arch| @compileError("unimplemented architecture: " ++ @tagName(arch)),
629};
630
631inline fn contextSwitch(message: *const SwitchMessage) *const SwitchMessage {
632 return @fieldParentPtr("contexts", switch (builtin.cpu.arch) {
633 .aarch64 => asm volatile (
634 \\ ldp x0, x2, [x1]
635 \\ ldr x3, [x2, #16]
636 \\ mov x4, sp
637 \\ stp x4, fp, [x0]
638 \\ adr x5, 0f
639 \\ ldp x4, fp, [x2]
640 \\ str x5, [x0, #16]
641 \\ mov sp, x4
642 \\ br x3
643 \\0:
644 : [received_message] "={x1}" (-> *const @FieldType(SwitchMessage, "contexts")),
645 : [message_to_send] "{x1}" (&message.contexts),
646 : .{
647 .x0 = true,
648 .x1 = true,
649 .x2 = true,
650 .x3 = true,
651 .x4 = true,
652 .x5 = true,
653 .x6 = true,
654 .x7 = true,
655 .x8 = true,
656 .x9 = true,
657 .x10 = true,
658 .x11 = true,
659 .x12 = true,
660 .x13 = true,
661 .x14 = true,
662 .x15 = true,
663 .x16 = true,
664 .x17 = true,
665 .x18 = true,
666 .x19 = true,
667 .x20 = true,
668 .x21 = true,
669 .x22 = true,
670 .x23 = true,
671 .x24 = true,
672 .x25 = true,
673 .x26 = true,
674 .x27 = true,
675 .x28 = true,
676 .x30 = true,
677 .z0 = true,
678 .z1 = true,
679 .z2 = true,
680 .z3 = true,
681 .z4 = true,
682 .z5 = true,
683 .z6 = true,
684 .z7 = true,
685 .z8 = true,
686 .z9 = true,
687 .z10 = true,
688 .z11 = true,
689 .z12 = true,
690 .z13 = true,
691 .z14 = true,
692 .z15 = true,
693 .z16 = true,
694 .z17 = true,
695 .z18 = true,
696 .z19 = true,
697 .z20 = true,
698 .z21 = true,
699 .z22 = true,
700 .z23 = true,
701 .z24 = true,
702 .z25 = true,
703 .z26 = true,
704 .z27 = true,
705 .z28 = true,
706 .z29 = true,
707 .z30 = true,
708 .z31 = true,
709 .p0 = true,
710 .p1 = true,
711 .p2 = true,
712 .p3 = true,
713 .p4 = true,
714 .p5 = true,
715 .p6 = true,
716 .p7 = true,
717 .p8 = true,
718 .p9 = true,
719 .p10 = true,
720 .p11 = true,
721 .p12 = true,
722 .p13 = true,
723 .p14 = true,
724 .p15 = true,
725 .fpcr = true,
726 .fpsr = true,
727 .ffr = true,
728 .memory = true,
729 }),
730 .x86_64 => asm volatile (
731 \\ movq 0(%%rsi), %%rax
732 \\ movq 8(%%rsi), %%rcx
733 \\ leaq 0f(%%rip), %%rdx
734 \\ movq %%rsp, 0(%%rax)
735 \\ movq %%rbp, 8(%%rax)
736 \\ movq %%rdx, 16(%%rax)
737 \\ movq 0(%%rcx), %%rsp
738 \\ movq 8(%%rcx), %%rbp
739 \\ jmpq *16(%%rcx)
740 \\0:
741 : [received_message] "={rsi}" (-> *const @FieldType(SwitchMessage, "contexts")),
742 : [message_to_send] "{rsi}" (&message.contexts),
743 : .{
744 .rax = true,
745 .rcx = true,
746 .rdx = true,
747 .rbx = true,
748 .rsi = true,
749 .rdi = true,
750 .r8 = true,
751 .r9 = true,
752 .r10 = true,
753 .r11 = true,
754 .r12 = true,
755 .r13 = true,
756 .r14 = true,
757 .r15 = true,
758 .mm0 = true,
759 .mm1 = true,
760 .mm2 = true,
761 .mm3 = true,
762 .mm4 = true,
763 .mm5 = true,
764 .mm6 = true,
765 .mm7 = true,
766 .zmm0 = true,
767 .zmm1 = true,
768 .zmm2 = true,
769 .zmm3 = true,
770 .zmm4 = true,
771 .zmm5 = true,
772 .zmm6 = true,
773 .zmm7 = true,
774 .zmm8 = true,
775 .zmm9 = true,
776 .zmm10 = true,
777 .zmm11 = true,
778 .zmm12 = true,
779 .zmm13 = true,
780 .zmm14 = true,
781 .zmm15 = true,
782 .zmm16 = true,
783 .zmm17 = true,
784 .zmm18 = true,
785 .zmm19 = true,
786 .zmm20 = true,
787 .zmm21 = true,
788 .zmm22 = true,
789 .zmm23 = true,
790 .zmm24 = true,
791 .zmm25 = true,
792 .zmm26 = true,
793 .zmm27 = true,
794 .zmm28 = true,
795 .zmm29 = true,
796 .zmm30 = true,
797 .zmm31 = true,
798 .fpsr = true,
799 .fpcr = true,
800 .mxcsr = true,
801 .rflags = true,
802 .dirflag = true,
803 .memory = true,
804 }),
805 else => |arch| @compileError("unimplemented architecture: " ++ @tagName(arch)),
806 });
807}
808
809fn mainIdleEntry() callconv(.naked) void {
810 switch (builtin.cpu.arch) {
811 .x86_64 => asm volatile (
812 \\ movq (%%rsp), %%rdi
813 \\ jmp %[mainIdle:P]
814 :
815 : [mainIdle] "X" (&mainIdle),
816 ),
817 .aarch64 => asm volatile (
818 \\ ldr x0, [sp, #-8]
819 \\ b %[mainIdle]
820 :
821 : [mainIdle] "X" (&mainIdle),
822 ),
823 else => |arch| @compileError("unimplemented architecture: " ++ @tagName(arch)),
824 }
825}
826
827fn fiberEntry() callconv(.naked) void {
828 switch (builtin.cpu.arch) {
829 .x86_64 => asm volatile (
830 \\ leaq 8(%%rsp), %%rdi
831 \\ jmpq *(%%rsp)
832 ),
833 .aarch64 => asm volatile (
834 \\ mov x0, sp
835 \\ ldr x2, [sp, #-8]
836 \\ br x2
837 ),
838 else => |arch| @compileError("unimplemented architecture: " ++ @tagName(arch)),
839 }
840}
841
842const AsyncClosure = struct {
843 event_loop: *EventLoop,
844 fiber: *Fiber,
845 start: *const fn (context: *const anyopaque, result: *anyopaque) void,
846 result_align: Alignment,
847 already_awaited: bool,
848
849 fn contextPointer(closure: *AsyncClosure) [*]align(Fiber.max_context_align.toByteUnits()) u8 {
850 return @alignCast(@as([*]u8, @ptrCast(closure)) + @sizeOf(AsyncClosure));
851 }
852
853 fn call(closure: *AsyncClosure, message: *const SwitchMessage) callconv(.withStackAlign(.c, @alignOf(AsyncClosure))) noreturn {
854 message.handle(closure.event_loop);
855 const fiber = closure.fiber;
856 std.log.debug("{*} performing async", .{fiber});
857 closure.start(closure.contextPointer(), fiber.resultBytes(closure.result_align));
858 const awaiter = @atomicRmw(?*Fiber, &fiber.awaiter, .Xchg, Fiber.finished, .acq_rel);
859 const ready_awaiter = r: {
860 const a = awaiter orelse break :r null;
861 if (@atomicRmw(bool, &closure.already_awaited, .Xchg, true, .acq_rel)) break :r null;
862 break :r a;
863 };
864 closure.event_loop.yield(ready_awaiter, .nothing);
865 unreachable; // switched to dead fiber
866 }
867
868 fn fromFiber(fiber: *Fiber) *AsyncClosure {
869 return @ptrFromInt(Fiber.max_context_align.max(.of(AsyncClosure)).backward(
870 @intFromPtr(fiber.allocatedEnd()) - Fiber.max_context_size,
871 ) - @sizeOf(AsyncClosure));
872 }
873};
874
875fn async(
876 userdata: ?*anyopaque,
877 result: []u8,
878 result_alignment: Alignment,
879 context: []const u8,
880 context_alignment: Alignment,
881 start: *const fn (context: *const anyopaque, result: *anyopaque) void,
882) ?*std.Io.AnyFuture {
883 return concurrent(userdata, result.len, result_alignment, context, context_alignment, start) catch {
884 start(context.ptr, result.ptr);
885 return null;
886 };
887}
888
889fn concurrent(
890 userdata: ?*anyopaque,
891 result_len: usize,
892 result_alignment: Alignment,
893 context: []const u8,
894 context_alignment: Alignment,
895 start: *const fn (context: *const anyopaque, result: *anyopaque) void,
896) error{OutOfMemory}!*std.Io.AnyFuture {
897 assert(result_alignment.compare(.lte, Fiber.max_result_align)); // TODO
898 assert(context_alignment.compare(.lte, Fiber.max_context_align)); // TODO
899 assert(result_len <= Fiber.max_result_size); // TODO
900 assert(context.len <= Fiber.max_context_size); // TODO
901
902 const event_loop: *EventLoop = @ptrCast(@alignCast(userdata));
903 const fiber = try Fiber.allocate(event_loop);
904 std.log.debug("allocated {*}", .{fiber});
905
906 const closure: *AsyncClosure = .fromFiber(fiber);
907 const stack_end: [*]align(16) usize = @ptrCast(@alignCast(closure));
908 (stack_end - 1)[0..1].* = .{@intFromPtr(&AsyncClosure.call)};
909 fiber.* = .{
910 .required_align = {},
911 .context = switch (builtin.cpu.arch) {
912 .x86_64 => .{
913 .rsp = @intFromPtr(stack_end - 1),
914 .rbp = 0,
915 .rip = @intFromPtr(&fiberEntry),
916 },
917 .aarch64 => .{
918 .sp = @intFromPtr(stack_end),
919 .fp = 0,
920 .pc = @intFromPtr(&fiberEntry),
921 },
922 else => |arch| @compileError("unimplemented architecture: " ++ @tagName(arch)),
923 },
924 .awaiter = null,
925 .queue_next = null,
926 .cancel_thread = null,
927 .awaiting_completions = .initEmpty(),
928 };
929 closure.* = .{
930 .event_loop = event_loop,
931 .fiber = fiber,
932 .start = start,
933 .result_align = result_alignment,
934 .already_awaited = false,
935 };
936 @memcpy(closure.contextPointer(), context);
937
938 event_loop.schedule(.current(), .{ .head = fiber, .tail = fiber });
939 return @ptrCast(fiber);
940}
941
942const DetachedClosure = struct {
943 event_loop: *EventLoop,
944 fiber: *Fiber,
945 start: *const fn (context: *const anyopaque) void,
946 detached_queue_node: std.DoublyLinkedList.Node,
947
948 fn contextPointer(closure: *DetachedClosure) [*]align(Fiber.max_context_align.toByteUnits()) u8 {
949 return @alignCast(@as([*]u8, @ptrCast(closure)) + @sizeOf(DetachedClosure));
950 }
951
952 fn call(closure: *DetachedClosure, message: *const SwitchMessage) callconv(.withStackAlign(.c, @alignOf(DetachedClosure))) noreturn {
953 message.handle(closure.event_loop);
954 std.log.debug("{*} performing async detached", .{closure.fiber});
955 closure.start(closure.contextPointer());
956 const awaiter = @atomicRmw(?*Fiber, &closure.fiber.awaiter, .Xchg, Fiber.finished, .acq_rel);
957 closure.event_loop.yield(awaiter, pending_task: {
958 closure.event_loop.detached.mutex.lock(closure.event_loop.io()) catch |err| switch (err) {
959 error.Canceled => break :pending_task .nothing,
960 };
961 defer closure.event_loop.detached.mutex.unlock(closure.event_loop.io());
962 if (closure.detached_queue_node.next == &closure.detached_queue_node) break :pending_task .nothing;
963 closure.event_loop.detached.list.remove(&closure.detached_queue_node);
964 break :pending_task .recycle;
965 });
966 unreachable; // switched to dead fiber
967 }
968};
969
970fn asyncDetached(
971 userdata: ?*anyopaque,
972 context: []const u8,
973 context_alignment: std.mem.Alignment,
974 start: *const fn (context: *const anyopaque) void,
975) void {
976 assert(context_alignment.compare(.lte, Fiber.max_context_align)); // TODO
977 assert(context.len <= Fiber.max_context_size); // TODO
978
979 const event_loop: *EventLoop = @ptrCast(@alignCast(userdata));
980 const fiber = Fiber.allocate(event_loop) catch {
981 start(context.ptr);
982 return;
983 };
984 std.log.debug("allocated {*}", .{fiber});
985
986 const current_thread: *Thread = .current();
987 const closure: *DetachedClosure = @ptrFromInt(Fiber.max_context_align.max(.of(DetachedClosure)).backward(
988 @intFromPtr(fiber.allocatedEnd()) - Fiber.max_context_size,
989 ) - @sizeOf(DetachedClosure));
990 const stack_end: [*]align(16) usize = @ptrCast(@alignCast(closure));
991 (stack_end - 1)[0..1].* = .{@intFromPtr(&DetachedClosure.call)};
992 fiber.* = .{
993 .required_align = {},
994 .context = switch (builtin.cpu.arch) {
995 .x86_64 => .{
996 .rsp = @intFromPtr(stack_end - 1),
997 .rbp = 0,
998 .rip = @intFromPtr(&fiberEntry),
999 },
1000 .aarch64 => .{
1001 .sp = @intFromPtr(stack_end),
1002 .fp = 0,
1003 .pc = @intFromPtr(&fiberEntry),
1004 },
1005 else => |arch| @compileError("unimplemented architecture: " ++ @tagName(arch)),
1006 },
1007 .awaiter = null,
1008 .queue_next = null,
1009 .cancel_thread = null,
1010 .awaiting_completions = .initEmpty(),
1011 };
1012 closure.* = .{
1013 .event_loop = event_loop,
1014 .fiber = fiber,
1015 .start = start,
1016 .detached_queue_node = .{},
1017 };
1018 {
1019 event_loop.detached.mutex.lock(event_loop.io()) catch |err| switch (err) {
1020 error.Canceled => {
1021 event_loop.recycle(fiber);
1022 start(context.ptr);
1023 return;
1024 },
1025 };
1026 defer event_loop.detached.mutex.unlock(event_loop.io());
1027 event_loop.detached.list.append(&closure.detached_queue_node);
1028 }
1029 @memcpy(closure.contextPointer(), context);
1030
1031 event_loop.schedule(current_thread, .{ .head = fiber, .tail = fiber });
1032}
1033
1034fn await(
1035 userdata: ?*anyopaque,
1036 any_future: *std.Io.AnyFuture,
1037 result: []u8,
1038 result_alignment: Alignment,
1039) void {
1040 const event_loop: *EventLoop = @ptrCast(@alignCast(userdata));
1041 const future_fiber: *Fiber = @ptrCast(@alignCast(any_future));
1042 if (@atomicLoad(?*Fiber, &future_fiber.awaiter, .acquire) != Fiber.finished)
1043 event_loop.yield(null, .{ .register_awaiter = &future_fiber.awaiter });
1044 @memcpy(result, future_fiber.resultBytes(result_alignment));
1045 event_loop.recycle(future_fiber);
1046}
1047
1048fn select(userdata: ?*anyopaque, futures: []const *Io.AnyFuture) usize {
1049 const el: *EventLoop = @ptrCast(@alignCast(userdata));
1050
1051 // Optimization to avoid the yield below.
1052 for (futures, 0..) |any_future, i| {
1053 const future_fiber: *Fiber = @ptrCast(@alignCast(any_future));
1054 if (@atomicLoad(?*Fiber, &future_fiber.awaiter, .acquire) == Fiber.finished)
1055 return i;
1056 }
1057
1058 el.yield(null, .{ .register_select = futures });
1059
1060 std.log.debug("back from select yield", .{});
1061
1062 const my_thread: *Thread = .current();
1063 const my_fiber = my_thread.currentFiber();
1064 var result: ?usize = null;
1065
1066 for (futures, 0..) |any_future, i| {
1067 const future_fiber: *Fiber = @ptrCast(@alignCast(any_future));
1068 if (@cmpxchgStrong(?*Fiber, &future_fiber.awaiter, my_fiber, null, .seq_cst, .seq_cst)) |awaiter| {
1069 if (awaiter == Fiber.finished) {
1070 if (result == null) result = i;
1071 } else if (awaiter) |a| {
1072 const closure: *AsyncClosure = .fromFiber(a);
1073 closure.already_awaited = false;
1074 }
1075 } else {
1076 const closure: *AsyncClosure = .fromFiber(my_fiber);
1077 closure.already_awaited = false;
1078 }
1079 }
1080
1081 return result.?;
1082}
1083
1084fn cancel(
1085 userdata: ?*anyopaque,
1086 any_future: *std.Io.AnyFuture,
1087 result: []u8,
1088 result_alignment: Alignment,
1089) void {
1090 const future_fiber: *Fiber = @ptrCast(@alignCast(any_future));
1091 if (@atomicRmw(
1092 ?*Thread,
1093 &future_fiber.cancel_thread,
1094 .Xchg,
1095 Thread.canceling,
1096 .acq_rel,
1097 )) |cancel_thread| if (cancel_thread != Thread.canceling) {
1098 getSqe(&Thread.current().io_uring).* = .{
1099 .opcode = .MSG_RING,
1100 .flags = std.os.linux.IOSQE_CQE_SKIP_SUCCESS,
1101 .ioprio = 0,
1102 .fd = cancel_thread.io_uring.fd,
1103 .off = @intFromPtr(future_fiber),
1104 .addr = 0,
1105 .len = @bitCast(-@as(i32, @intFromEnum(std.os.linux.E.INTR))),
1106 .rw_flags = 0,
1107 .user_data = @intFromEnum(Completion.UserData.cleanup),
1108 .buf_index = 0,
1109 .personality = 0,
1110 .splice_fd_in = 0,
1111 .addr3 = 0,
1112 .resv = 0,
1113 };
1114 };
1115 await(userdata, any_future, result, result_alignment);
1116}
1117
1118fn cancelRequested(userdata: ?*anyopaque) bool {
1119 _ = userdata;
1120 return @atomicLoad(?*Thread, &Thread.current().currentFiber().cancel_thread, .acquire) == Thread.canceling;
1121}
1122
1123fn createFile(
1124 userdata: ?*anyopaque,
1125 dir: Io.Dir,
1126 sub_path: []const u8,
1127 flags: Io.File.CreateFlags,
1128) Io.File.OpenError!Io.File {
1129 const el: *EventLoop = @ptrCast(@alignCast(userdata));
1130 const thread: *Thread = .current();
1131 const iou = &thread.io_uring;
1132 const fiber = thread.currentFiber();
1133 try fiber.enterCancelRegion(thread);
1134
1135 const posix = std.posix;
1136 const sub_path_c = try posix.toPosixPath(sub_path);
1137
1138 var os_flags: posix.O = .{
1139 .ACCMODE = if (flags.read) .RDWR else .WRONLY,
1140 .CREAT = true,
1141 .TRUNC = flags.truncate,
1142 .EXCL = flags.exclusive,
1143 };
1144 if (@hasField(posix.O, "LARGEFILE")) os_flags.LARGEFILE = true;
1145 if (@hasField(posix.O, "CLOEXEC")) os_flags.CLOEXEC = true;
1146
1147 // Use the O locking flags if the os supports them to acquire the lock
1148 // atomically. Note that the NONBLOCK flag is removed after the openat()
1149 // call is successful.
1150 const has_flock_open_flags = @hasField(posix.O, "EXLOCK");
1151 if (has_flock_open_flags) switch (flags.lock) {
1152 .none => {},
1153 .shared => {
1154 os_flags.SHLOCK = true;
1155 os_flags.NONBLOCK = flags.lock_nonblocking;
1156 },
1157 .exclusive => {
1158 os_flags.EXLOCK = true;
1159 os_flags.NONBLOCK = flags.lock_nonblocking;
1160 },
1161 };
1162 const have_flock = @TypeOf(posix.system.flock) != void;
1163
1164 if (have_flock and !has_flock_open_flags and flags.lock != .none) {
1165 @panic("TODO");
1166 }
1167
1168 if (has_flock_open_flags and flags.lock_nonblocking) {
1169 @panic("TODO");
1170 }
1171
1172 getSqe(iou).* = .{
1173 .opcode = .OPENAT,
1174 .flags = 0,
1175 .ioprio = 0,
1176 .fd = dir.handle,
1177 .off = 0,
1178 .addr = @intFromPtr(&sub_path_c),
1179 .len = @intCast(flags.mode),
1180 .rw_flags = @bitCast(os_flags),
1181 .user_data = @intFromPtr(fiber),
1182 .buf_index = 0,
1183 .personality = 0,
1184 .splice_fd_in = 0,
1185 .addr3 = 0,
1186 .resv = 0,
1187 };
1188
1189 el.yield(null, .nothing);
1190 fiber.exitCancelRegion(thread);
1191
1192 const completion = fiber.resultPointer(Completion);
1193 switch (errno(completion.result)) {
1194 .SUCCESS => return .{ .handle = completion.result },
1195 .INTR => unreachable,
1196 .CANCELED => return error.Canceled,
1197
1198 .FAULT => unreachable,
1199 .INVAL => return error.BadPathName,
1200 .BADF => unreachable,
1201 .ACCES => return error.AccessDenied,
1202 .FBIG => return error.FileTooBig,
1203 .OVERFLOW => return error.FileTooBig,
1204 .ISDIR => return error.IsDir,
1205 .LOOP => return error.SymLinkLoop,
1206 .MFILE => return error.ProcessFdQuotaExceeded,
1207 .NAMETOOLONG => return error.NameTooLong,
1208 .NFILE => return error.SystemFdQuotaExceeded,
1209 .NODEV => return error.NoDevice,
1210 .NOENT => return error.FileNotFound,
1211 .NOMEM => return error.SystemResources,
1212 .NOSPC => return error.NoSpaceLeft,
1213 .NOTDIR => return error.NotDir,
1214 .PERM => return error.PermissionDenied,
1215 .EXIST => return error.PathAlreadyExists,
1216 .BUSY => return error.DeviceBusy,
1217 .OPNOTSUPP => return error.FileLocksNotSupported,
1218 .AGAIN => return error.WouldBlock,
1219 .TXTBSY => return error.FileBusy,
1220 .NXIO => return error.NoDevice,
1221 else => |err| return posix.unexpectedErrno(err),
1222 }
1223}
1224
1225fn fileOpen(
1226 userdata: ?*anyopaque,
1227 dir: Io.Dir,
1228 sub_path: []const u8,
1229 flags: Io.File.OpenFlags,
1230) Io.File.OpenError!Io.File {
1231 const el: *EventLoop = @ptrCast(@alignCast(userdata));
1232 const thread: *Thread = .current();
1233 const iou = &thread.io_uring;
1234 const fiber = thread.currentFiber();
1235 try fiber.enterCancelRegion(thread);
1236
1237 const posix = std.posix;
1238 const sub_path_c = try posix.toPosixPath(sub_path);
1239
1240 var os_flags: posix.O = .{
1241 .ACCMODE = switch (flags.mode) {
1242 .read_only => .RDONLY,
1243 .write_only => .WRONLY,
1244 .read_write => .RDWR,
1245 },
1246 };
1247
1248 if (@hasField(posix.O, "CLOEXEC")) os_flags.CLOEXEC = true;
1249 if (@hasField(posix.O, "LARGEFILE")) os_flags.LARGEFILE = true;
1250 if (@hasField(posix.O, "NOCTTY")) os_flags.NOCTTY = !flags.allow_ctty;
1251
1252 // Use the O locking flags if the os supports them to acquire the lock
1253 // atomically.
1254 const has_flock_open_flags = @hasField(posix.O, "EXLOCK");
1255 if (has_flock_open_flags) {
1256 // Note that the NONBLOCK flag is removed after the openat() call
1257 // is successful.
1258 switch (flags.lock) {
1259 .none => {},
1260 .shared => {
1261 os_flags.SHLOCK = true;
1262 os_flags.NONBLOCK = flags.lock_nonblocking;
1263 },
1264 .exclusive => {
1265 os_flags.EXLOCK = true;
1266 os_flags.NONBLOCK = flags.lock_nonblocking;
1267 },
1268 }
1269 }
1270 const have_flock = @TypeOf(posix.system.flock) != void;
1271
1272 if (have_flock and !has_flock_open_flags and flags.lock != .none) {
1273 @panic("TODO");
1274 }
1275
1276 if (has_flock_open_flags and flags.lock_nonblocking) {
1277 @panic("TODO");
1278 }
1279
1280 getSqe(iou).* = .{
1281 .opcode = .OPENAT,
1282 .flags = 0,
1283 .ioprio = 0,
1284 .fd = dir.handle,
1285 .off = 0,
1286 .addr = @intFromPtr(&sub_path_c),
1287 .len = 0,
1288 .rw_flags = @bitCast(os_flags),
1289 .user_data = @intFromPtr(fiber),
1290 .buf_index = 0,
1291 .personality = 0,
1292 .splice_fd_in = 0,
1293 .addr3 = 0,
1294 .resv = 0,
1295 };
1296
1297 el.yield(null, .nothing);
1298 fiber.exitCancelRegion(thread);
1299
1300 const completion = fiber.resultPointer(Completion);
1301 switch (errno(completion.result)) {
1302 .SUCCESS => return .{ .handle = completion.result },
1303 .INTR => unreachable,
1304 .CANCELED => return error.Canceled,
1305
1306 .FAULT => unreachable,
1307 .INVAL => return error.BadPathName,
1308 .BADF => unreachable,
1309 .ACCES => return error.AccessDenied,
1310 .FBIG => return error.FileTooBig,
1311 .OVERFLOW => return error.FileTooBig,
1312 .ISDIR => return error.IsDir,
1313 .LOOP => return error.SymLinkLoop,
1314 .MFILE => return error.ProcessFdQuotaExceeded,
1315 .NAMETOOLONG => return error.NameTooLong,
1316 .NFILE => return error.SystemFdQuotaExceeded,
1317 .NODEV => return error.NoDevice,
1318 .NOENT => return error.FileNotFound,
1319 .NOMEM => return error.SystemResources,
1320 .NOSPC => return error.NoSpaceLeft,
1321 .NOTDIR => return error.NotDir,
1322 .PERM => return error.PermissionDenied,
1323 .EXIST => return error.PathAlreadyExists,
1324 .BUSY => return error.DeviceBusy,
1325 .OPNOTSUPP => return error.FileLocksNotSupported,
1326 .AGAIN => return error.WouldBlock,
1327 .TXTBSY => return error.FileBusy,
1328 .NXIO => return error.NoDevice,
1329 else => |err| return posix.unexpectedErrno(err),
1330 }
1331}
1332
1333fn fileClose(userdata: ?*anyopaque, file: Io.File) void {
1334 const el: *EventLoop = @ptrCast(@alignCast(userdata));
1335 const thread: *Thread = .current();
1336 const iou = &thread.io_uring;
1337 const fiber = thread.currentFiber();
1338
1339 getSqe(iou).* = .{
1340 .opcode = .CLOSE,
1341 .flags = 0,
1342 .ioprio = 0,
1343 .fd = file.handle,
1344 .off = 0,
1345 .addr = 0,
1346 .len = 0,
1347 .rw_flags = 0,
1348 .user_data = @intFromPtr(fiber),
1349 .buf_index = 0,
1350 .personality = 0,
1351 .splice_fd_in = 0,
1352 .addr3 = 0,
1353 .resv = 0,
1354 };
1355
1356 el.yield(null, .nothing);
1357
1358 const completion = fiber.resultPointer(Completion);
1359 switch (errno(completion.result)) {
1360 .SUCCESS => return,
1361 .INTR => unreachable,
1362 .CANCELED => return,
1363
1364 .BADF => unreachable, // Always a race condition.
1365 else => return,
1366 }
1367}
1368
1369fn pread(userdata: ?*anyopaque, file: Io.File, buffer: []u8, offset: std.posix.off_t) Io.File.PReadError!usize {
1370 const el: *EventLoop = @ptrCast(@alignCast(userdata));
1371 const thread: *Thread = .current();
1372 const iou = &thread.io_uring;
1373 const fiber = thread.currentFiber();
1374 try fiber.enterCancelRegion(thread);
1375
1376 getSqe(iou).* = .{
1377 .opcode = .READ,
1378 .flags = 0,
1379 .ioprio = 0,
1380 .fd = file.handle,
1381 .off = @bitCast(offset),
1382 .addr = @intFromPtr(buffer.ptr),
1383 .len = @min(buffer.len, 0x7ffff000),
1384 .rw_flags = 0,
1385 .user_data = @intFromPtr(fiber),
1386 .buf_index = 0,
1387 .personality = 0,
1388 .splice_fd_in = 0,
1389 .addr3 = 0,
1390 .resv = 0,
1391 };
1392
1393 el.yield(null, .nothing);
1394 fiber.exitCancelRegion(thread);
1395
1396 const completion = fiber.resultPointer(Completion);
1397 switch (errno(completion.result)) {
1398 .SUCCESS => return @as(u32, @bitCast(completion.result)),
1399 .INTR => unreachable,
1400 .CANCELED => return error.Canceled,
1401
1402 .INVAL => unreachable,
1403 .FAULT => unreachable,
1404 .NOENT => return error.ProcessNotFound,
1405 .AGAIN => return error.WouldBlock,
1406 .BADF => return error.NotOpenForReading, // Can be a race condition.
1407 .IO => return error.InputOutput,
1408 .ISDIR => return error.IsDir,
1409 .NOBUFS => return error.SystemResources,
1410 .NOMEM => return error.SystemResources,
1411 .NOTCONN => return error.SocketUnconnected,
1412 .CONNRESET => return error.ConnectionResetByPeer,
1413 .TIMEDOUT => return error.Timeout,
1414 .NXIO => return error.Unseekable,
1415 .SPIPE => return error.Unseekable,
1416 .OVERFLOW => return error.Unseekable,
1417 else => |err| return std.posix.unexpectedErrno(err),
1418 }
1419}
1420
1421fn pwrite(userdata: ?*anyopaque, file: Io.File, buffer: []const u8, offset: std.posix.off_t) Io.File.PWriteError!usize {
1422 const el: *EventLoop = @ptrCast(@alignCast(userdata));
1423 const thread: *Thread = .current();
1424 const iou = &thread.io_uring;
1425 const fiber = thread.currentFiber();
1426 try fiber.enterCancelRegion(thread);
1427
1428 getSqe(iou).* = .{
1429 .opcode = .WRITE,
1430 .flags = 0,
1431 .ioprio = 0,
1432 .fd = file.handle,
1433 .off = @bitCast(offset),
1434 .addr = @intFromPtr(buffer.ptr),
1435 .len = @min(buffer.len, 0x7ffff000),
1436 .rw_flags = 0,
1437 .user_data = @intFromPtr(fiber),
1438 .buf_index = 0,
1439 .personality = 0,
1440 .splice_fd_in = 0,
1441 .addr3 = 0,
1442 .resv = 0,
1443 };
1444
1445 el.yield(null, .nothing);
1446 fiber.exitCancelRegion(thread);
1447
1448 const completion = fiber.resultPointer(Completion);
1449 switch (errno(completion.result)) {
1450 .SUCCESS => return @as(u32, @bitCast(completion.result)),
1451 .INTR => unreachable,
1452 .CANCELED => return error.Canceled,
1453
1454 .INVAL => return error.InvalidArgument,
1455 .FAULT => unreachable,
1456 .NOENT => return error.ProcessNotFound,
1457 .AGAIN => return error.WouldBlock,
1458 .BADF => return error.NotOpenForWriting, // can be a race condition.
1459 .DESTADDRREQ => unreachable, // `connect` was never called.
1460 .DQUOT => return error.DiskQuota,
1461 .FBIG => return error.FileTooBig,
1462 .IO => return error.InputOutput,
1463 .NOSPC => return error.NoSpaceLeft,
1464 .ACCES => return error.AccessDenied,
1465 .PERM => return error.PermissionDenied,
1466 .PIPE => return error.BrokenPipe,
1467 .NXIO => return error.Unseekable,
1468 .SPIPE => return error.Unseekable,
1469 .OVERFLOW => return error.Unseekable,
1470 .BUSY => return error.DeviceBusy,
1471 .CONNRESET => return error.ConnectionResetByPeer,
1472 .MSGSIZE => return error.MessageTooBig,
1473 else => |err| return std.posix.unexpectedErrno(err),
1474 }
1475}
1476
1477fn now(userdata: ?*anyopaque, clockid: std.posix.clockid_t) Io.ClockGetTimeError!Io.Timestamp {
1478 _ = userdata;
1479 const timespec = try std.posix.clock_gettime(clockid);
1480 return @enumFromInt(@as(i128, timespec.sec) * std.time.ns_per_s + timespec.nsec);
1481}
1482
1483fn sleep(userdata: ?*anyopaque, clockid: std.posix.clockid_t, deadline: Io.Deadline) Io.SleepError!void {
1484 const el: *EventLoop = @ptrCast(@alignCast(userdata));
1485 const thread: *Thread = .current();
1486 const iou = &thread.io_uring;
1487 const fiber = thread.currentFiber();
1488 try fiber.enterCancelRegion(thread);
1489
1490 const deadline_nanoseconds: i96 = switch (deadline) {
1491 .duration => |duration| duration.nanoseconds,
1492 .timestamp => |timestamp| @intFromEnum(timestamp),
1493 };
1494 const timespec: std.os.linux.kernel_timespec = .{
1495 .sec = @intCast(@divFloor(deadline_nanoseconds, std.time.ns_per_s)),
1496 .nsec = @intCast(@mod(deadline_nanoseconds, std.time.ns_per_s)),
1497 };
1498 getSqe(iou).* = .{
1499 .opcode = .TIMEOUT,
1500 .flags = 0,
1501 .ioprio = 0,
1502 .fd = 0,
1503 .off = 0,
1504 .addr = @intFromPtr(&timespec),
1505 .len = 1,
1506 .rw_flags = @as(u32, switch (deadline) {
1507 .duration => 0,
1508 .timestamp => std.os.linux.IORING_TIMEOUT_ABS,
1509 }) | @as(u32, switch (clockid) {
1510 .REALTIME => std.os.linux.IORING_TIMEOUT_REALTIME,
1511 .MONOTONIC => 0,
1512 .BOOTTIME => std.os.linux.IORING_TIMEOUT_BOOTTIME,
1513 else => return error.UnsupportedClock,
1514 }),
1515 .user_data = @intFromPtr(fiber),
1516 .buf_index = 0,
1517 .personality = 0,
1518 .splice_fd_in = 0,
1519 .addr3 = 0,
1520 .resv = 0,
1521 };
1522
1523 el.yield(null, .nothing);
1524 fiber.exitCancelRegion(thread);
1525
1526 const completion = fiber.resultPointer(Completion);
1527 switch (errno(completion.result)) {
1528 .SUCCESS, .TIME => return,
1529 .INTR => unreachable,
1530 .CANCELED => return error.Canceled,
1531
1532 else => |err| return std.posix.unexpectedErrno(err),
1533 }
1534}
1535
1536fn mutexLock(userdata: ?*anyopaque, prev_state: Io.Mutex.State, mutex: *Io.Mutex) error{Canceled}!void {
1537 const el: *EventLoop = @ptrCast(@alignCast(userdata));
1538 el.yield(null, .{ .mutex_lock = .{ .prev_state = prev_state, .mutex = mutex } });
1539}
1540fn mutexUnlock(userdata: ?*anyopaque, prev_state: Io.Mutex.State, mutex: *Io.Mutex) void {
1541 var maybe_waiting_fiber: ?*Fiber = @ptrFromInt(@intFromEnum(prev_state));
1542 while (if (maybe_waiting_fiber) |waiting_fiber| @cmpxchgWeak(
1543 Io.Mutex.State,
1544 &mutex.state,
1545 @enumFromInt(@intFromPtr(waiting_fiber)),
1546 @enumFromInt(@intFromPtr(waiting_fiber.queue_next)),
1547 .release,
1548 .acquire,
1549 ) else @cmpxchgWeak(
1550 Io.Mutex.State,
1551 &mutex.state,
1552 .locked_once,
1553 .unlocked,
1554 .release,
1555 .acquire,
1556 ) orelse return) |next_state| maybe_waiting_fiber = @ptrFromInt(@intFromEnum(next_state));
1557 maybe_waiting_fiber.?.queue_next = null;
1558 const el: *EventLoop = @ptrCast(@alignCast(userdata));
1559 el.yield(maybe_waiting_fiber.?, .reschedule);
1560}
1561
1562const ConditionImpl = struct {
1563 tail: *Fiber,
1564 event: union(enum) {
1565 queued,
1566 wake: Io.Condition.Wake,
1567 },
1568};
1569
1570fn conditionWait(userdata: ?*anyopaque, cond: *Io.Condition, mutex: *Io.Mutex) Io.Cancelable!void {
1571 const el: *EventLoop = @ptrCast(@alignCast(userdata));
1572 el.yield(null, .{ .condition_wait = .{ .cond = cond, .mutex = mutex } });
1573 const thread = Thread.current();
1574 const fiber = thread.currentFiber();
1575 const cond_impl = fiber.resultPointer(ConditionImpl);
1576 try mutex.lock(el.io());
1577 switch (cond_impl.event) {
1578 .queued => {},
1579 .wake => |wake| if (fiber.queue_next) |next_fiber| switch (wake) {
1580 .one => if (@cmpxchgStrong(
1581 ?*Fiber,
1582 @as(*?*Fiber, @ptrCast(&cond.state)),
1583 null,
1584 next_fiber,
1585 .release,
1586 .acquire,
1587 )) |old_fiber| {
1588 const old_cond_impl = old_fiber.?.resultPointer(ConditionImpl);
1589 assert(old_cond_impl.tail.queue_next == null);
1590 old_cond_impl.tail.queue_next = next_fiber;
1591 old_cond_impl.tail = cond_impl.tail;
1592 },
1593 .all => el.schedule(thread, .{ .head = next_fiber, .tail = cond_impl.tail }),
1594 },
1595 }
1596 fiber.queue_next = null;
1597}
1598
1599fn conditionWake(userdata: ?*anyopaque, cond: *Io.Condition, wake: Io.Condition.Wake) void {
1600 const el: *EventLoop = @ptrCast(@alignCast(userdata));
1601 const waiting_fiber = @atomicRmw(?*Fiber, @as(*?*Fiber, @ptrCast(&cond.state)), .Xchg, null, .acquire) orelse return;
1602 waiting_fiber.resultPointer(ConditionImpl).event = .{ .wake = wake };
1603 el.yield(waiting_fiber, .reschedule);
1604}
1605
1606fn errno(signed: i32) std.os.linux.E {
1607 return .init(@bitCast(@as(isize, signed)));
1608}
1609
1610fn getSqe(iou: *IoUring) *std.os.linux.io_uring_sqe {
1611 while (true) return iou.get_sqe() catch {
1612 _ = iou.submit_and_wait(0) catch |err| switch (err) {
1613 error.SignalInterrupt => std.log.warn("submit_and_wait failed with SignalInterrupt", .{}),
1614 else => |e| @panic(@errorName(e)),
1615 };
1616 continue;
1617 };
1618}