authorgravatar for jacobly@ziglang.orgJacob Young <jacobly@ziglang.org> 2025-03-27 17:19:53-04:00
committergravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2025-07-20 10:38:38-07:00
logc7b406f2adbe06710b8d1aca9599fb607ff26f96
treefb8e78dc7690f65535a6cfe81262df96b97a9a42
parent03bb08d337ca6b70164b4207624f5b52b4900fa3

EventLoop: add threads

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

lib/std/Io/EventLoop.zig+78-20
......@@ -7,15 +7,25 @@ const EventLoop = @This();
77
88gpa: Allocator,
99mutex: std.Thread.Mutex,
10cond: std.Thread.Condition,
1011queue: std.DoublyLinkedList(void),
1112free: std.DoublyLinkedList(void),
1213main_fiber_buffer: [@sizeOf(Fiber) + max_result_len]u8 align(@alignOf(Fiber)),
14exiting: bool,
15idle_count: usize,
16threads: std.ArrayListUnmanaged(Thread),
1317
18threadlocal var current_thread: *Thread = undefined;
1419threadlocal var current_fiber: *Fiber = undefined;
1520
1621const max_result_len = 64;
1722const min_stack_size = 4 * 1024 * 1024;
1823
24const Thread = struct {
25 thread: std.Thread,
26 idle_fiber: Fiber,
27};
28
1929const Fiber = struct {
2030 context: Context,
2131 awaiter: ?*Fiber,
......@@ -23,32 +33,58 @@ const Fiber = struct {
2333
2434 const finished: ?*Fiber = @ptrFromInt(std.mem.alignBackward(usize, std.math.maxInt(usize), @alignOf(Fiber)));
2535
26 fn resultPointer(f: *Fiber) [*]u8 {
27 const base: [*]u8 = @ptrCast(f);
28 return base + @sizeOf(Fiber);
29 }
30
31 fn stackEndPointer(f: *Fiber) [*]u8 {
32 const base: [*]u8 = @ptrCast(f);
33 return base + std.mem.alignForward(
36 fn allocatedSlice(f: *Fiber) []align(@alignOf(Fiber)) u8 {
37 const base: [*]align(@alignOf(Fiber)) u8 = @ptrCast(f);
38 return base[0..std.mem.alignForward(
3439 usize,
3540 @sizeOf(Fiber) + max_result_len + min_stack_size,
3641 std.heap.page_size_max,
37 );
42 )];
43 }
44
45 fn resultSlice(f: *Fiber) []u8 {
46 const base: [*]align(@alignOf(Fiber)) u8 = @ptrCast(f);
47 return base[@sizeOf(Fiber)..][0..max_result_len];
48 }
49
50 fn stackEndPointer(f: *Fiber) [*]u8 {
51 const allocated_slice = f.allocatedSlice();
52 return allocated_slice[allocated_slice.len..].ptr;
3853 }
3954};
4055
41pub fn init(el: *EventLoop, gpa: Allocator) void {
56pub fn init(el: *EventLoop, gpa: Allocator) error{OutOfMemory}!void {
4257 el.* = .{
4358 .gpa = gpa,
4459 .mutex = .{},
60 .cond = .{},
4561 .queue = .{},
4662 .free = .{},
4763 .main_fiber_buffer = undefined,
64 .exiting = false,
65 .idle_count = 0,
66 .threads = try .initCapacity(gpa, @max(std.Thread.getCpuCount() catch 1, 1)),
4867 };
68 current_thread = el.threads.addOneAssumeCapacity();
4969 current_fiber = @ptrCast(&el.main_fiber_buffer);
5070}
5171
72pub fn deinit(el: *EventLoop) void {
73 {
74 el.mutex.lock();
75 defer el.mutex.unlock();
76 assert(el.queue.len == 0); // pending async
77 el.exiting = true;
78 }
79 el.cond.broadcast();
80 while (el.free.pop()) |free_node| {
81 const free_fiber: *Fiber = @fieldParentPtr("queue_node", free_node);
82 el.gpa.free(free_fiber.allocatedSlice());
83 }
84 for (el.threads.items[1..]) |*thread| thread.thread.join();
85 el.threads.deinit(el.gpa);
86}
87
5288fn allocateFiber(el: *EventLoop, result_len: usize) error{OutOfMemory}!*Fiber {
5389 assert(result_len <= max_result_len);
5490 const free_node = free_node: {
......@@ -73,10 +109,8 @@ fn yield(el: *EventLoop, optional_fiber: ?*Fiber, register_awaiter: ?*?*Fiber) v
73109 break :ready_node el.queue.pop();
74110 }) |ready_node|
75111 @fieldParentPtr("queue_node", ready_node)
76 else if (register_awaiter) |_| // time to switch to an idle fiber?
77 @panic("no other fiber to switch to in order to be able to register this fiber as an awaiter")
78 else // nothing to do
79 return;
112 else
113 &current_thread.idle_fiber;
80114 const message: SwitchMessage = .{
81115 .prev_context = &current_fiber.context,
82116 .ready_context = &ready_fiber.context,
......@@ -90,20 +124,44 @@ fn yield(el: *EventLoop, optional_fiber: ?*Fiber, register_awaiter: ?*?*Fiber) v
90124}
91125
92126fn schedule(el: *EventLoop, fiber: *Fiber) void {
93 el.mutex.lock();
94 defer el.mutex.unlock();
95 el.queue.append(&fiber.queue_node);
127 signal: {
128 el.mutex.lock();
129 defer el.mutex.unlock();
130 el.queue.append(&fiber.queue_node);
131 if (el.idle_count > 0) break :signal;
132 if (el.threads.items.len == el.threads.capacity) return;
133 const thread = el.threads.addOneAssumeCapacity();
134 thread.thread = std.Thread.spawn(.{
135 .stack_size = min_stack_size,
136 .allocator = el.gpa,
137 }, threadEntry, .{ el, thread }) catch return;
138 }
139 el.cond.signal();
96140}
97141
98142fn recycle(el: *EventLoop, fiber: *Fiber) void {
99143 std.log.debug("recyling {*}", .{fiber});
100144 fiber.awaiter = undefined;
101 @memset(fiber.resultPointer()[0..max_result_len], undefined);
145 @memset(fiber.resultSlice(), undefined);
102146 el.mutex.lock();
103147 defer el.mutex.unlock();
104148 el.free.append(&fiber.queue_node);
105149}
106150
151fn threadEntry(el: *EventLoop, thread: *Thread) void {
152 current_thread = thread;
153 current_fiber = &thread.idle_fiber;
154 while (true) {
155 el.yield(null, null);
156 el.mutex.lock();
157 defer el.mutex.unlock();
158 if (el.exiting) return;
159 el.idle_count += 1;
160 defer el.idle_count -= 1;
161 el.cond.wait(&el.mutex);
162 }
163}
164
107165const SwitchMessage = extern struct {
108166 prev_context: *Context,
109167 ready_context: *Context,
......@@ -209,7 +267,7 @@ const AsyncClosure = struct {
209267 fn call(closure: *AsyncClosure, message: *const SwitchMessage) callconv(.c) noreturn {
210268 message.handle(closure.event_loop);
211269 std.log.debug("{*} performing async", .{closure.fiber});
212 closure.start(closure.context, closure.fiber.resultPointer());
270 closure.start(closure.context, closure.fiber.resultSlice().ptr);
213271 const awaiter = @atomicRmw(?*Fiber, &closure.fiber.awaiter, .Xchg, Fiber.finished, .acq_rel);
214272 closure.event_loop.yield(awaiter, null);
215273 unreachable; // switched to dead fiber
......@@ -219,7 +277,7 @@ const AsyncClosure = struct {
219277pub fn @"await"(userdata: ?*anyopaque, any_future: *std.Io.AnyFuture, result: []u8) void {
220278 const event_loop: *EventLoop = @alignCast(@ptrCast(userdata));
221279 const future_fiber: *Fiber = @alignCast(@ptrCast(any_future));
222 const result_src = future_fiber.resultPointer()[0..result.len];
280 const result_src = future_fiber.resultSlice()[0..result.len];
223281 if (@atomicLoad(?*Fiber, &future_fiber.awaiter, .acquire) != Fiber.finished) event_loop.yield(null, &future_fiber.awaiter);
224282 @memcpy(result, result_src);
225283 event_loop.recycle(future_fiber);