authorgravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2025-03-27 23:59:35-07:00
committergravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2025-10-02 16:30:59-07:00
log3ee239963094ee0268210f8afc95ba6518b61ad4
treedc38ab34e5f9bd8e8e41a5e2f1db44ae442801a6
parentacd02e978b2772f7909af035a205edff28aae757

update threaded fibers impl to actually storing args

sorry, something still not working correctly

1 files changed, 42 insertions(+), 18 deletions(-)

lib/std/Io/EventLoop.zig+42-18
...@@ -18,7 +18,11 @@ threads: std.ArrayListUnmanaged(Thread),...@@ -18,7 +18,11 @@ threads: std.ArrayListUnmanaged(Thread),
18threadlocal var current_idle_context: *Context = undefined;18threadlocal var current_idle_context: *Context = undefined;
19threadlocal var current_fiber_context: *Context = undefined;19threadlocal var current_fiber_context: *Context = undefined;
2020
21/// Also used for context.
21const max_result_len = 64;22const max_result_len = 64;
23/// Also used for context.
24const max_result_align: std.mem.Alignment = .@"16";
25
22const min_stack_size = 4 * 1024 * 1024;26const min_stack_size = 4 * 1024 * 1024;
23const idle_stack_size = 32 * 1024;27const idle_stack_size = 32 * 1024;
24const stack_align = 16;28const stack_align = 16;
...@@ -29,6 +33,8 @@ const Thread = struct {...@@ -29,6 +33,8 @@ const Thread = struct {
29};33};
3034
31const Fiber = struct {35const Fiber = struct {
36 _: void align(max_result_align.toByteUnits()) = {},
37
32 context: Context,38 context: Context,
33 awaiter: ?*Fiber,39 awaiter: ?*Fiber,
34 queue_node: std.DoublyLinkedList(void).Node,40 queue_node: std.DoublyLinkedList(void).Node,
...@@ -39,14 +45,27 @@ const Fiber = struct {...@@ -39,14 +45,27 @@ const Fiber = struct {
39 const base: [*]align(@alignOf(Fiber)) u8 = @ptrCast(f);45 const base: [*]align(@alignOf(Fiber)) u8 = @ptrCast(f);
40 return base[0..std.mem.alignForward(46 return base[0..std.mem.alignForward(
41 usize,47 usize,
42 @sizeOf(Fiber) + max_result_len + min_stack_size,48 resultOffset() + max_result_len + min_stack_size,
43 std.heap.page_size_max,49 std.heap.page_size_max,
44 )];50 )];
45 }51 }
4652
53 fn argsOffset() usize {
54 return max_result_align.forward(@sizeOf(Fiber));
55 }
56
57 fn resultOffset() usize {
58 return max_result_align.forward(argsOffset() + max_result_len);
59 }
60
61 fn argsSlice(f: *Fiber) []u8 {
62 const base: [*]align(@alignOf(Fiber)) u8 = @ptrCast(f);
63 return base[argsOffset()..][0..max_result_len];
64 }
65
47 fn resultSlice(f: *Fiber) []u8 {66 fn resultSlice(f: *Fiber) []u8 {
48 const base: [*]align(@alignOf(Fiber)) u8 = @ptrCast(f);67 const base: [*]align(@alignOf(Fiber)) u8 = @ptrCast(f);
49 return base[@sizeOf(Fiber)..][0..max_result_len];68 return base[resultOffset()..][0..max_result_len];
50 }69 }
5170
52 fn stackEndPointer(f: *Fiber) [*]u8 {71 fn stackEndPointer(f: *Fiber) [*]u8 {
...@@ -102,7 +121,7 @@ pub fn deinit(el: *EventLoop) void {...@@ -102,7 +121,7 @@ pub fn deinit(el: *EventLoop) void {
102 assert(el.queue.len == 0); // pending async121 assert(el.queue.len == 0); // pending async
103 el.yield(null, &el.exit_awaiter);122 el.yield(null, &el.exit_awaiter);
104 while (el.free.pop()) |free_node| {123 while (el.free.pop()) |free_node| {
105 const free_fiber: *Fiber = @fieldParentPtr("queue_node", free_node);124 const free_fiber: *Fiber = @alignCast(@fieldParentPtr("queue_node", free_node));
106 el.gpa.free(free_fiber.allocatedSlice());125 el.gpa.free(free_fiber.allocatedSlice());
107 }126 }
108 const idle_context_offset = std.mem.alignForward(usize, el.threads.items.len * @sizeOf(Thread), @alignOf(Context));127 const idle_context_offset = std.mem.alignForward(usize, el.threads.items.len * @sizeOf(Thread), @alignOf(Context));
...@@ -112,8 +131,7 @@ pub fn deinit(el: *EventLoop) void {...@@ -112,8 +131,7 @@ pub fn deinit(el: *EventLoop) void {
112 el.gpa.free(allocated_ptr[0..idle_stack_end]);131 el.gpa.free(allocated_ptr[0..idle_stack_end]);
113}132}
114133
115fn allocateFiber(el: *EventLoop, result_len: usize) error{OutOfMemory}!*Fiber {134fn allocateFiber(el: *EventLoop) error{OutOfMemory}!*Fiber {
116 assert(result_len <= max_result_len);
117 const free_node = free_node: {135 const free_node = free_node: {
118 el.mutex.lock();136 el.mutex.lock();
119 defer el.mutex.unlock();137 defer el.mutex.unlock();
...@@ -121,12 +139,12 @@ fn allocateFiber(el: *EventLoop, result_len: usize) error{OutOfMemory}!*Fiber {...@@ -121,12 +139,12 @@ fn allocateFiber(el: *EventLoop, result_len: usize) error{OutOfMemory}!*Fiber {
121 } orelse {139 } orelse {
122 const n = std.mem.alignForward(140 const n = std.mem.alignForward(
123 usize,141 usize,
124 @sizeOf(Fiber) + max_result_len + min_stack_size,142 Fiber.resultOffset() + max_result_len + min_stack_size,
125 std.heap.page_size_max,143 std.heap.page_size_max,
126 );144 );
127 return @alignCast(@ptrCast(try el.gpa.alignedAlloc(u8, @alignOf(Fiber), n)));145 return @alignCast(@ptrCast(try el.gpa.alignedAlloc(u8, @alignOf(Fiber), n)));
128 };146 };
129 return @fieldParentPtr("queue_node", free_node);147 return @alignCast(@fieldParentPtr("queue_node", free_node));
130}148}
131149
132fn yield(el: *EventLoop, optional_fiber: ?*Fiber, register_awaiter: ?*?*Fiber) void {150fn yield(el: *EventLoop, optional_fiber: ?*Fiber, register_awaiter: ?*?*Fiber) void {
...@@ -136,7 +154,7 @@ fn yield(el: *EventLoop, optional_fiber: ?*Fiber, register_awaiter: ?*?*Fiber) v...@@ -136,7 +154,7 @@ fn yield(el: *EventLoop, optional_fiber: ?*Fiber, register_awaiter: ?*?*Fiber) v
136 defer el.mutex.unlock();154 defer el.mutex.unlock();
137 break :ready_node el.queue.pop();155 break :ready_node el.queue.pop();
138 }) |ready_node|156 }) |ready_node|
139 @fieldParentPtr("queue_node", ready_node)157 @alignCast(@fieldParentPtr("queue_node", ready_node))
140 else158 else
141 break :ready_context current_idle_context;159 break :ready_context current_idle_context;
142 break :ready_context &ready_fiber.context;160 break :ready_context &ready_fiber.context;
...@@ -213,7 +231,7 @@ const SwitchMessage = extern struct {...@@ -213,7 +231,7 @@ const SwitchMessage = extern struct {
213 register_awaiter: ?*?*Fiber,231 register_awaiter: ?*?*Fiber,
214232
215 fn handle(message: *const SwitchMessage, el: *EventLoop) void {233 fn handle(message: *const SwitchMessage, el: *EventLoop) void {
216 const prev_fiber: *Fiber = @fieldParentPtr("context", message.prev_context);234 const prev_fiber: *Fiber = @alignCast(@fieldParentPtr("context", message.prev_context));
217 current_fiber_context = message.ready_context;235 current_fiber_context = message.ready_context;
218 if (message.register_awaiter) |awaiter| if (@atomicRmw(?*Fiber, awaiter, .Xchg, prev_fiber, .acq_rel) == Fiber.finished) el.schedule(prev_fiber);236 if (message.register_awaiter) |awaiter| if (@atomicRmw(?*Fiber, awaiter, .Xchg, prev_fiber, .acq_rel) == Fiber.finished) el.schedule(prev_fiber);
219 }237 }
...@@ -279,17 +297,25 @@ fn fiberEntry() callconv(.naked) void {...@@ -279,17 +297,25 @@ fn fiberEntry() callconv(.naked) void {
279297
280pub fn @"async"(298pub fn @"async"(
281 userdata: ?*anyopaque,299 userdata: ?*anyopaque,
282 eager_result: []u8,300 result: []u8,
283 context: ?*anyopaque,301 result_alignment: std.mem.Alignment,
284 start: *const fn (context: ?*anyopaque, result: *anyopaque) void,302 context: []const u8,
303 context_alignment: std.mem.Alignment,
304 start: *const fn (context: *const anyopaque, result: *anyopaque) void,
285) ?*std.Io.AnyFuture {305) ?*std.Io.AnyFuture {
306 assert(result_alignment.compare(.lte, max_result_align)); // TODO
307 assert(context_alignment.compare(.lte, max_result_align)); // TODO
308 assert(result.len <= max_result_len); // TODO
309 assert(context.len <= max_result_len); // TODO
310
286 const event_loop: *EventLoop = @alignCast(@ptrCast(userdata));311 const event_loop: *EventLoop = @alignCast(@ptrCast(userdata));
287 const fiber = event_loop.allocateFiber(eager_result.len) catch {312 const fiber = event_loop.allocateFiber() catch {
288 start(context, eager_result.ptr);313 start(context.ptr, result.ptr);
289 return null;314 return null;
290 };315 };
291 fiber.awaiter = null;316 fiber.awaiter = null;
292 fiber.queue_node = .{ .data = {} };317 fiber.queue_node = .{ .data = {} };
318 @memcpy(fiber.argsSlice()[0..context.len], context);
293 std.log.debug("allocated {*}", .{fiber});319 std.log.debug("allocated {*}", .{fiber});
294320
295 const closure: *AsyncClosure = @ptrFromInt(std.mem.alignBackward(321 const closure: *AsyncClosure = @ptrFromInt(std.mem.alignBackward(
...@@ -299,7 +325,6 @@ pub fn @"async"(...@@ -299,7 +325,6 @@ pub fn @"async"(
299 ));325 ));
300 closure.* = .{326 closure.* = .{
301 .event_loop = event_loop,327 .event_loop = event_loop,
302 .context = context,
303 .fiber = fiber,328 .fiber = fiber,
304 .start = start,329 .start = start,
305 };330 };
...@@ -316,14 +341,13 @@ pub fn @"async"(...@@ -316,14 +341,13 @@ pub fn @"async"(
316341
317const AsyncClosure = struct {342const AsyncClosure = struct {
318 event_loop: *EventLoop,343 event_loop: *EventLoop,
319 context: ?*anyopaque,
320 fiber: *Fiber,344 fiber: *Fiber,
321 start: *const fn (context: ?*anyopaque, result: *anyopaque) void,345 start: *const fn (context: *const anyopaque, result: *anyopaque) void,
322346
323 fn call(closure: *AsyncClosure, message: *const SwitchMessage) callconv(.c) noreturn {347 fn call(closure: *AsyncClosure, message: *const SwitchMessage) callconv(.c) noreturn {
324 message.handle(closure.event_loop);348 message.handle(closure.event_loop);
325 std.log.debug("{*} performing async", .{closure.fiber});349 std.log.debug("{*} performing async", .{closure.fiber});
326 closure.start(closure.context, closure.fiber.resultSlice().ptr);350 closure.start(closure.fiber.argsSlice().ptr, closure.fiber.resultSlice().ptr);
327 const awaiter = @atomicRmw(?*Fiber, &closure.fiber.awaiter, .Xchg, Fiber.finished, .acq_rel);351 const awaiter = @atomicRmw(?*Fiber, &closure.fiber.awaiter, .Xchg, Fiber.finished, .acq_rel);
328 closure.event_loop.yield(awaiter, null);352 closure.event_loop.yield(awaiter, null);
329 unreachable; // switched to dead fiber353 unreachable; // switched to dead fiber