authorgravatar for jacobly@ziglang.orgJacob Young <jacobly@ziglang.org> 2025-04-01 03:45:31-04:00
committergravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2025-07-20 10:38:39-07:00
logfdf15fae9779c1b0fc05b12abf0865245dd95d4f
tree5e2f1448325a56bc7b3615f3aec05499e1645ea2
parentf2b24479eed677677cc7615fa120d0d0cb87b389

EventLoop: implement detached fibers

1 files changed, 78 insertions(+), 46 deletions(-)

lib/std/Io/EventLoop.zig+78-46
......@@ -9,9 +9,12 @@ const IoUring = std.os.linux.IoUring;
99
1010/// Must be a thread-safe allocator.
1111gpa: Allocator,
12mutex: std.Thread.Mutex,
1312main_fiber_buffer: [@sizeOf(Fiber) + Fiber.max_result_size]u8 align(@alignOf(Fiber)),
1413threads: Thread.List,
14detached: struct {
15 mutex: std.Io.Mutex,
16 list: std.DoublyLinkedList(void),
17},
1518
1619/// Empirically saw >128KB being used by the self-hosted backend to panic.
1720const idle_stack_size = 256 * 1024;
......@@ -167,13 +170,16 @@ pub fn init(el: *EventLoop, gpa: Allocator) !void {
167170 errdefer gpa.free(allocated_slice);
168171 el.* = .{
169172 .gpa = gpa,
170 .mutex = .{},
171173 .main_fiber_buffer = undefined,
172174 .threads = .{
173175 .allocated = @ptrCast(allocated_slice[0..threads_size]),
174176 .reserved = 1,
175177 .active = 1,
176178 },
179 .detached = .{
180 .mutex = .init,
181 .list = .{},
182 },
177183 };
178184 const main_fiber: *Fiber = @ptrCast(&el.main_fiber_buffer);
179185 main_fiber.* = .{
......@@ -207,6 +213,23 @@ pub fn init(el: *EventLoop, gpa: Allocator) !void {
207213}
208214
209215pub fn deinit(el: *EventLoop) void {
216 while (true) cancel(el, detached_future: {
217 el.detached.mutex.lock(el.io()) catch |err| switch (err) {
218 error.Canceled => unreachable, // main fiber cannot be canceled
219 };
220 defer el.detached.mutex.unlock(el.io());
221 const detached: *DetachedClosure = @fieldParentPtr(
222 "detached_queue_node",
223 el.detached.list.pop() orelse break,
224 );
225 // notify the detached fiber that it is no longer allowed to recycle itself
226 detached.detached_queue_node = .{
227 .prev = &detached.detached_queue_node,
228 .next = &detached.detached_queue_node,
229 .data = {},
230 };
231 break :detached_future @ptrCast(detached.fiber);
232 }, &.{}, .@"1");
210233 const active_threads = @atomicLoad(u32, &el.threads.active, .acquire);
211234 for (el.threads.allocated[0..active_threads]) |*thread| {
212235 const ready_fiber = @atomicLoad(?*Fiber, &thread.ready_queue, .monotonic);
......@@ -460,7 +483,7 @@ const SwitchMessage = struct {
460483 const PendingTask = union(enum) {
461484 nothing,
462485 reschedule,
463 recycle: *Fiber,
486 recycle,
464487 register_awaiter: *?*Fiber,
465488 mutex_lock: struct {
466489 prev_state: Io.Mutex.State,
......@@ -483,8 +506,10 @@ const SwitchMessage = struct {
483506 assert(prev_fiber.queue_next == null);
484507 el.schedule(thread, .{ .head = prev_fiber, .tail = prev_fiber });
485508 },
486 .recycle => |fiber| {
487 el.recycle(fiber);
509 .recycle => {
510 const prev_fiber: *Fiber = @alignCast(@fieldParentPtr("context", message.contexts.prev));
511 assert(prev_fiber.queue_next == null);
512 el.recycle(prev_fiber);
488513 },
489514 .register_awaiter => |awaiter| {
490515 const prev_fiber: *Fiber = @alignCast(@fieldParentPtr("context", message.contexts.prev));
......@@ -609,21 +634,7 @@ fn fiberEntry() callconv(.naked) void {
609634 switch (builtin.cpu.arch) {
610635 .x86_64 => asm volatile (
611636 \\ leaq 8(%%rsp), %%rdi
612 \\ jmp %[AsyncClosure_call:P]
613 :
614 : [AsyncClosure_call] "X" (&AsyncClosure.call),
615 ),
616 else => |arch| @compileError("unimplemented architecture: " ++ @tagName(arch)),
617 }
618}
619
620fn fiberEntryDetached() callconv(.naked) void {
621 switch (builtin.cpu.arch) {
622 .x86_64 => asm volatile (
623 \\ leaq 8(%%rsp), %%rdi
624 \\ jmp %[DetachedClosure_call:P]
625 :
626 : [DetachedClosure_call] "X" (&DetachedClosure.call),
637 \\ jmpq *(%%rsp)
627638 ),
628639 else => |arch| @compileError("unimplemented architecture: " ++ @tagName(arch)),
629640 }
......@@ -649,29 +660,6 @@ const AsyncClosure = struct {
649660 }
650661};
651662
652const DetachedClosure = struct {
653 event_loop: *EventLoop,
654 fiber: *Fiber,
655 start: *const fn (context: *const anyopaque) void,
656
657 fn contextPointer(closure: *DetachedClosure) [*]align(Fiber.max_context_align.toByteUnits()) u8 {
658 return @alignCast(@as([*]u8, @ptrCast(closure)) + @sizeOf(DetachedClosure));
659 }
660
661 fn call(closure: *DetachedClosure, message: *const SwitchMessage) callconv(.withStackAlign(.c, @alignOf(DetachedClosure))) noreturn {
662 message.handle(closure.event_loop);
663 std.log.debug("{*} performing async detached", .{closure.fiber});
664 closure.start(closure.contextPointer());
665 const awaiter = @atomicRmw(?*Fiber, &closure.fiber.awaiter, .Xchg, Fiber.finished, .acq_rel);
666 if (awaiter) |a| {
667 closure.event_loop.yield(a, .nothing);
668 } else {
669 closure.event_loop.yield(null, .{ .recycle = closure.fiber });
670 }
671 unreachable; // switched to dead fiber
672 }
673};
674
675663fn @"async"(
676664 userdata: ?*anyopaque,
677665 result: []u8,
......@@ -695,11 +683,13 @@ fn @"async"(
695683 const closure: *AsyncClosure = @ptrFromInt(Fiber.max_context_align.max(.of(AsyncClosure)).backward(
696684 @intFromPtr(fiber.allocatedEnd()) - Fiber.max_context_size,
697685 ) - @sizeOf(AsyncClosure));
686 const stack_end: [*]usize = @alignCast(@ptrCast(closure));
687 (stack_end - 1)[0..1].* = .{@intFromPtr(&AsyncClosure.call)};
698688 fiber.* = .{
699689 .required_align = {},
700690 .context = switch (builtin.cpu.arch) {
701691 .x86_64 => .{
702 .rsp = @intFromPtr(closure) - @sizeOf(usize),
692 .rsp = @intFromPtr(stack_end - 1),
703693 .rbp = 0,
704694 .rip = @intFromPtr(&fiberEntry),
705695 },
......@@ -722,6 +712,34 @@ fn @"async"(
722712 return @ptrCast(fiber);
723713}
724714
715const DetachedClosure = struct {
716 event_loop: *EventLoop,
717 fiber: *Fiber,
718 start: *const fn (context: *const anyopaque) void,
719 detached_queue_node: std.DoublyLinkedList(void).Node,
720
721 fn contextPointer(closure: *DetachedClosure) [*]align(Fiber.max_context_align.toByteUnits()) u8 {
722 return @alignCast(@as([*]u8, @ptrCast(closure)) + @sizeOf(DetachedClosure));
723 }
724
725 fn call(closure: *DetachedClosure, message: *const SwitchMessage) callconv(.withStackAlign(.c, @alignOf(DetachedClosure))) noreturn {
726 message.handle(closure.event_loop);
727 std.log.debug("{*} performing async detached", .{closure.fiber});
728 closure.start(closure.contextPointer());
729 const awaiter = @atomicRmw(?*Fiber, &closure.fiber.awaiter, .Xchg, Fiber.finished, .acq_rel);
730 closure.event_loop.yield(awaiter, pending_task: {
731 closure.event_loop.detached.mutex.lock(closure.event_loop.io()) catch |err| switch (err) {
732 error.Canceled => break :pending_task .nothing,
733 };
734 defer closure.event_loop.detached.mutex.unlock(closure.event_loop.io());
735 if (closure.detached_queue_node.next == &closure.detached_queue_node) break :pending_task .nothing;
736 closure.event_loop.detached.list.remove(&closure.detached_queue_node);
737 break :pending_task .recycle;
738 });
739 unreachable; // switched to dead fiber
740 }
741};
742
725743fn go(
726744 userdata: ?*anyopaque,
727745 context: []const u8,
......@@ -742,13 +760,15 @@ fn go(
742760 const closure: *DetachedClosure = @ptrFromInt(Fiber.max_context_align.max(.of(DetachedClosure)).backward(
743761 @intFromPtr(fiber.allocatedEnd()) - Fiber.max_context_size,
744762 ) - @sizeOf(DetachedClosure));
763 const stack_end: [*]usize = @alignCast(@ptrCast(closure));
764 (stack_end - 1)[0..1].* = .{@intFromPtr(&DetachedClosure.call)};
745765 fiber.* = .{
746766 .required_align = {},
747767 .context = switch (builtin.cpu.arch) {
748768 .x86_64 => .{
749 .rsp = @intFromPtr(closure) - @sizeOf(usize),
769 .rsp = @intFromPtr(stack_end - 1),
750770 .rbp = 0,
751 .rip = @intFromPtr(&fiberEntryDetached),
771 .rip = @intFromPtr(&fiberEntry),
752772 },
753773 else => |arch| @compileError("unimplemented architecture: " ++ @tagName(arch)),
754774 },
......@@ -761,7 +781,19 @@ fn go(
761781 .event_loop = event_loop,
762782 .fiber = fiber,
763783 .start = start,
784 .detached_queue_node = .{ .data = {} },
764785 };
786 {
787 event_loop.detached.mutex.lock(event_loop.io()) catch |err| switch (err) {
788 error.Canceled => {
789 event_loop.recycle(fiber);
790 start(context.ptr);
791 return;
792 },
793 };
794 defer event_loop.detached.mutex.unlock(event_loop.io());
795 event_loop.detached.list.append(&closure.detached_queue_node);
796 }
765797 @memcpy(closure.contextPointer(), context);
766798
767799 event_loop.schedule(current_thread, .{ .head = fiber, .tail = fiber });