authorgravatar for jacobly@ziglang.orgJacob Young <jacobly@ziglang.org> 2025-03-27 01:49:01-04:00
committergravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2025-07-20 10:38:38-07:00
log93054125fe0d608dd6813213b453133cb28f915c
treef408ab6649f7b2006024306174adfc25afbf8696
parent07ee4977dab9696a511fab69fe136280c9bfa0fc

EventLoop: prepare for threading

1 files changed, 65 insertions(+), 48 deletions(-)

lib/std/Io/EventLoop.zig+65-48
......@@ -5,6 +5,7 @@ const Io = std.Io;
55const EventLoop = @This();
66
77gpa: Allocator,
8mutex: std.Thread.Mutex,
89queue: std.DoublyLinkedList(void),
910free: std.DoublyLinkedList(void),
1011main_fiber_buffer: [@sizeOf(Fiber) + max_result_len]u8 align(@alignOf(Fiber)),
......@@ -39,6 +40,7 @@ const Fiber = struct {
3940pub fn init(el: *EventLoop, gpa: Allocator) void {
4041 el.* = .{
4142 .gpa = gpa,
43 .mutex = .{},
4244 .queue = .{},
4345 .free = .{},
4446 .main_fiber_buffer = undefined,
......@@ -48,7 +50,11 @@ pub fn init(el: *EventLoop, gpa: Allocator) void {
4850
4951fn allocateFiber(el: *EventLoop, result_len: usize) error{OutOfMemory}!*Fiber {
5052 assert(result_len <= max_result_len);
51 const free_node = el.free.pop() orelse {
53 const free_node = free_node: {
54 el.mutex.lock();
55 defer el.mutex.unlock();
56 break :free_node el.free.pop();
57 } orelse {
5258 const n = std.mem.alignForward(
5359 usize,
5460 @sizeOf(Fiber) + max_result_len + min_stack_size,
......@@ -59,36 +65,48 @@ fn allocateFiber(el: *EventLoop, result_len: usize) error{OutOfMemory}!*Fiber {
5965 return @fieldParentPtr("queue_node", free_node);
6066}
6167
62fn yield(el: *EventLoop, optional_fiber: ?*Fiber) void {
63 if (optional_fiber) |fiber| {
64 const old = &current_fiber.regs;
65 current_fiber = fiber;
66 contextSwitch(old, &fiber.regs);
67 return;
68 }
69 if (el.queue.pop()) |node| {
70 const fiber: *Fiber = @fieldParentPtr("queue_node", node);
71 const old = &current_fiber.regs;
72 current_fiber = fiber;
73 contextSwitch(old, &fiber.regs);
74 return;
75 }
76 @panic("everything is done");
68fn yield(el: *EventLoop, optional_fiber: ?*Fiber, register_awaiter: ?*?*Fiber) void {
69 const message: SwitchMessage = .{
70 .ready_fiber = optional_fiber orelse if (ready_node: {
71 el.mutex.lock();
72 defer el.mutex.unlock();
73 break :ready_node el.queue.pop();
74 }) |ready_node|
75 @fieldParentPtr("queue_node", ready_node)
76 else if (register_awaiter) |_|
77 @panic("no other fiber to switch to in order to be able to register this fiber as an awaiter") // time to switch to an idle fiber?
78 else
79 return, // nothing to do
80 .register_awaiter = register_awaiter,
81 };
82 std.log.debug("switching from {*} to {*}", .{ current_fiber, message.ready_fiber });
83 SwitchMessage.handle(@ptrFromInt(contextSwitch(&current_fiber.regs, &message.ready_fiber.regs, @intFromPtr(&message))), el);
7784}
7885
79/// Equivalent to calling `yield` and then giving the fiber back to the event loop.
80fn exit(el: *EventLoop, optional_fiber: ?*Fiber) noreturn {
81 yield(el, optional_fiber);
82 @panic("TODO recycle the fiber");
83}
86const SwitchMessage = struct {
87 ready_fiber: *Fiber,
88 register_awaiter: ?*?*Fiber,
89
90 fn handle(message: *const SwitchMessage, el: *EventLoop) void {
91 const prev_fiber = current_fiber;
92 current_fiber = message.ready_fiber;
93 if (message.register_awaiter) |awaiter| if (@atomicRmw(?*Fiber, awaiter, .Xchg, prev_fiber, .acq_rel) == Fiber.finished) el.schedule(prev_fiber);
94 }
95};
8496
8597fn schedule(el: *EventLoop, fiber: *Fiber) void {
98 el.mutex.lock();
99 defer el.mutex.unlock();
86100 el.queue.append(&fiber.queue_node);
87101}
88102
89fn myFiber(el: *EventLoop) *Fiber {
90 _ = el;
91 return current_fiber;
103fn recycle(el: *EventLoop, fiber: *Fiber) void {
104 std.log.debug("recyling {*}", .{fiber});
105 fiber.awaiter = undefined;
106 @memset(fiber.resultPointer()[0..max_result_len], undefined);
107 el.mutex.lock();
108 defer el.mutex.unlock();
109 el.free.append(&fiber.queue_node);
92110}
93111
94112const Regs = extern struct {
......@@ -101,7 +119,7 @@ const Regs = extern struct {
101119 rbp: usize,
102120};
103121
104const contextSwitch: *const fn (old: *Regs, new: *Regs) callconv(.c) void = @ptrCast(&contextSwitch_naked);
122const contextSwitch: *const fn (old: *Regs, new: *Regs, message: usize) callconv(.c) usize = @ptrCast(&contextSwitch_naked);
105123
106124noinline fn contextSwitch_naked() callconv(.naked) void {
107125 asm volatile (
......@@ -121,6 +139,7 @@ noinline fn contextSwitch_naked() callconv(.naked) void {
121139 \\movq 0x28(%%rsi), %%rbx
122140 \\movq 0x30(%%rsi), %%rbp
123141 \\
142 \\movq %%rdx, %%rax
124143 \\ret
125144 );
126145}
......@@ -128,6 +147,7 @@ noinline fn contextSwitch_naked() callconv(.naked) void {
128147fn popRet() callconv(.naked) void {
129148 asm volatile (
130149 \\pop %%rdi
150 \\movq %%rax, %%rsi
131151 \\ret
132152 );
133153}
......@@ -145,6 +165,7 @@ pub fn @"async"(
145165 };
146166 fiber.awaiter = null;
147167 fiber.queue_node = .{ .data = {} };
168 std.log.debug("allocated {*}", .{fiber});
148169
149170 const closure: *AsyncClosure = @ptrFromInt(std.mem.alignBackward(
150171 usize,
......@@ -157,14 +178,16 @@ pub fn @"async"(
157178 .fiber = fiber,
158179 .start = start,
159180 };
160 const stack_end_ptr: [*]align(16) usize = @alignCast(@ptrCast(closure));
161 (stack_end_ptr - 1)[0] = 0;
162 (stack_end_ptr - 2)[0] = @intFromPtr(&AsyncClosure.call);
163 (stack_end_ptr - 3)[0] = @intFromPtr(closure);
164 (stack_end_ptr - 4)[0] = @intFromPtr(&popRet);
165
181 const stack_end: [*]align(16) usize = @alignCast(@ptrCast(closure));
182 const stack_top = (stack_end - 4)[0..4];
183 stack_top.* = .{
184 @intFromPtr(&popRet),
185 @intFromPtr(closure),
186 @intFromPtr(&AsyncClosure.call),
187 0,
188 };
166189 fiber.regs = .{
167 .rsp = @intFromPtr(stack_end_ptr - 4),
190 .rsp = @intFromPtr(stack_top),
168191 .r15 = 0,
169192 .r14 = 0,
170193 .r13 = 0,
......@@ -181,30 +204,24 @@ const AsyncClosure = struct {
181204 _: void align(16) = {},
182205 event_loop: *EventLoop,
183206 context: ?*anyopaque,
184 fiber: *EventLoop.Fiber,
207 fiber: *Fiber,
185208 start: *const fn (context: ?*anyopaque, result: *anyopaque) void,
186209
187 fn call(closure: *AsyncClosure) callconv(.c) void {
188 std.log.debug("wrap called in async", .{});
210 fn call(closure: *AsyncClosure, message: *const SwitchMessage) callconv(.c) noreturn {
211 message.handle(closure.event_loop);
212 std.log.debug("{*} performing async", .{closure.fiber});
189213 closure.start(closure.context, closure.fiber.resultPointer());
190 const awaiter = @atomicRmw(?*EventLoop.Fiber, &closure.fiber.awaiter, .Xchg, EventLoop.Fiber.finished, .seq_cst);
191 closure.event_loop.exit(awaiter);
214 const awaiter = @atomicRmw(?*Fiber, &closure.fiber.awaiter, .Xchg, Fiber.finished, .acq_rel);
215 closure.event_loop.yield(awaiter, null);
216 unreachable; // switched to dead fiber
192217 }
193218};
194219
195220pub fn @"await"(userdata: ?*anyopaque, any_future: *std.Io.AnyFuture, result: []u8) void {
196221 const event_loop: *EventLoop = @alignCast(@ptrCast(userdata));
197 const future_fiber: *EventLoop.Fiber = @alignCast(@ptrCast(any_future));
222 const future_fiber: *Fiber = @alignCast(@ptrCast(any_future));
198223 const result_src = future_fiber.resultPointer()[0..result.len];
199 const my_fiber = event_loop.myFiber();
200
201 const prev = @atomicRmw(?*EventLoop.Fiber, &future_fiber.awaiter, .Xchg, my_fiber, .seq_cst);
202 if (prev == EventLoop.Fiber.finished) {
203 @memcpy(result, result_src);
204 return;
205 }
206 event_loop.yield(prev);
207 // Resumed when the value is available.
208 std.log.debug("yield returned in await", .{});
224 if (@atomicLoad(?*Fiber, &future_fiber.awaiter, .acquire) != Fiber.finished) event_loop.yield(null, &future_fiber.awaiter);
209225 @memcpy(result, result_src);
226 event_loop.recycle(future_fiber);
210227}