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-10-02 16:30:59-07:00
log2c1ceb4c9ce1cbead655264879dd0cee16fa3fa4
tree9b903567f88d01146a6966a4ed107800f6cf5099
parent19e7613a2d9c487c7f6cb9bfe0815c7d33ac68df

EventLoop: add threads

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

lib/std/Io/EventLoop.zig+78-20
...@@ -7,15 +7,25 @@ const EventLoop = @This();...@@ -7,15 +7,25 @@ const EventLoop = @This();
77
8gpa: Allocator,8gpa: Allocator,
9mutex: std.Thread.Mutex,9mutex: std.Thread.Mutex,
10cond: std.Thread.Condition,
10queue: std.DoublyLinkedList(void),11queue: std.DoublyLinkedList(void),
11free: std.DoublyLinkedList(void),12free: std.DoublyLinkedList(void),
12main_fiber_buffer: [@sizeOf(Fiber) + max_result_len]u8 align(@alignOf(Fiber)),13main_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;
14threadlocal var current_fiber: *Fiber = undefined;19threadlocal var current_fiber: *Fiber = undefined;
1520
16const max_result_len = 64;21const max_result_len = 64;
17const min_stack_size = 4 * 1024 * 1024;22const min_stack_size = 4 * 1024 * 1024;
1823
24const Thread = struct {
25 thread: std.Thread,
26 idle_fiber: Fiber,
27};
28
19const Fiber = struct {29const Fiber = struct {
20 context: Context,30 context: Context,
21 awaiter: ?*Fiber,31 awaiter: ?*Fiber,
...@@ -23,32 +33,58 @@ const Fiber = struct {...@@ -23,32 +33,58 @@ const Fiber = struct {
2333
24 const finished: ?*Fiber = @ptrFromInt(std.mem.alignBackward(usize, std.math.maxInt(usize), @alignOf(Fiber)));34 const finished: ?*Fiber = @ptrFromInt(std.mem.alignBackward(usize, std.math.maxInt(usize), @alignOf(Fiber)));
2535
26 fn resultPointer(f: *Fiber) [*]u8 {36 fn allocatedSlice(f: *Fiber) []align(@alignOf(Fiber)) u8 {
27 const base: [*]u8 = @ptrCast(f);37 const base: [*]align(@alignOf(Fiber)) u8 = @ptrCast(f);
28 return base + @sizeOf(Fiber);38 return base[0..std.mem.alignForward(
29 }
30
31 fn stackEndPointer(f: *Fiber) [*]u8 {
32 const base: [*]u8 = @ptrCast(f);
33 return base + std.mem.alignForward(
34 usize,39 usize,
35 @sizeOf(Fiber) + max_result_len + min_stack_size,40 @sizeOf(Fiber) + max_result_len + min_stack_size,
36 std.heap.page_size_max,41 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;
38 }53 }
39};54};
4055
41pub fn init(el: *EventLoop, gpa: Allocator) void {56pub fn init(el: *EventLoop, gpa: Allocator) error{OutOfMemory}!void {
42 el.* = .{57 el.* = .{
43 .gpa = gpa,58 .gpa = gpa,
44 .mutex = .{},59 .mutex = .{},
60 .cond = .{},
45 .queue = .{},61 .queue = .{},
46 .free = .{},62 .free = .{},
47 .main_fiber_buffer = undefined,63 .main_fiber_buffer = undefined,
64 .exiting = false,
65 .idle_count = 0,
66 .threads = try .initCapacity(gpa, @max(std.Thread.getCpuCount() catch 1, 1)),
48 };67 };
68 current_thread = el.threads.addOneAssumeCapacity();
49 current_fiber = @ptrCast(&el.main_fiber_buffer);69 current_fiber = @ptrCast(&el.main_fiber_buffer);
50}70}
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
52fn allocateFiber(el: *EventLoop, result_len: usize) error{OutOfMemory}!*Fiber {88fn allocateFiber(el: *EventLoop, result_len: usize) error{OutOfMemory}!*Fiber {
53 assert(result_len <= max_result_len);89 assert(result_len <= max_result_len);
54 const free_node = free_node: {90 const free_node = free_node: {
...@@ -73,10 +109,8 @@ fn yield(el: *EventLoop, optional_fiber: ?*Fiber, register_awaiter: ?*?*Fiber) v...@@ -73,10 +109,8 @@ fn yield(el: *EventLoop, optional_fiber: ?*Fiber, register_awaiter: ?*?*Fiber) v
73 break :ready_node el.queue.pop();109 break :ready_node el.queue.pop();
74 }) |ready_node|110 }) |ready_node|
75 @fieldParentPtr("queue_node", ready_node)111 @fieldParentPtr("queue_node", ready_node)
76 else if (register_awaiter) |_| // time to switch to an idle fiber?112 else
77 @panic("no other fiber to switch to in order to be able to register this fiber as an awaiter")113 &current_thread.idle_fiber;
78 else // nothing to do
79 return;
80 const message: SwitchMessage = .{114 const message: SwitchMessage = .{
81 .prev_context = &current_fiber.context,115 .prev_context = &current_fiber.context,
82 .ready_context = &ready_fiber.context,116 .ready_context = &ready_fiber.context,
...@@ -90,20 +124,44 @@ fn yield(el: *EventLoop, optional_fiber: ?*Fiber, register_awaiter: ?*?*Fiber) v...@@ -90,20 +124,44 @@ fn yield(el: *EventLoop, optional_fiber: ?*Fiber, register_awaiter: ?*?*Fiber) v
90}124}
91125
92fn schedule(el: *EventLoop, fiber: *Fiber) void {126fn schedule(el: *EventLoop, fiber: *Fiber) void {
93 el.mutex.lock();127 signal: {
94 defer el.mutex.unlock();128 el.mutex.lock();
95 el.queue.append(&fiber.queue_node);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();
96}140}
97141
98fn recycle(el: *EventLoop, fiber: *Fiber) void {142fn recycle(el: *EventLoop, fiber: *Fiber) void {
99 std.log.debug("recyling {*}", .{fiber});143 std.log.debug("recyling {*}", .{fiber});
100 fiber.awaiter = undefined;144 fiber.awaiter = undefined;
101 @memset(fiber.resultPointer()[0..max_result_len], undefined);145 @memset(fiber.resultSlice(), undefined);
102 el.mutex.lock();146 el.mutex.lock();
103 defer el.mutex.unlock();147 defer el.mutex.unlock();
104 el.free.append(&fiber.queue_node);148 el.free.append(&fiber.queue_node);
105}149}
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
107const SwitchMessage = extern struct {165const SwitchMessage = extern struct {
108 prev_context: *Context,166 prev_context: *Context,
109 ready_context: *Context,167 ready_context: *Context,
...@@ -209,7 +267,7 @@ const AsyncClosure = struct {...@@ -209,7 +267,7 @@ const AsyncClosure = struct {
209 fn call(closure: *AsyncClosure, message: *const SwitchMessage) callconv(.c) noreturn {267 fn call(closure: *AsyncClosure, message: *const SwitchMessage) callconv(.c) noreturn {
210 message.handle(closure.event_loop);268 message.handle(closure.event_loop);
211 std.log.debug("{*} performing async", .{closure.fiber});269 std.log.debug("{*} performing async", .{closure.fiber});
212 closure.start(closure.context, closure.fiber.resultPointer());270 closure.start(closure.context, closure.fiber.resultSlice().ptr);
213 const awaiter = @atomicRmw(?*Fiber, &closure.fiber.awaiter, .Xchg, Fiber.finished, .acq_rel);271 const awaiter = @atomicRmw(?*Fiber, &closure.fiber.awaiter, .Xchg, Fiber.finished, .acq_rel);
214 closure.event_loop.yield(awaiter, null);272 closure.event_loop.yield(awaiter, null);
215 unreachable; // switched to dead fiber273 unreachable; // switched to dead fiber
...@@ -219,7 +277,7 @@ const AsyncClosure = struct {...@@ -219,7 +277,7 @@ const AsyncClosure = struct {
219pub fn @"await"(userdata: ?*anyopaque, any_future: *std.Io.AnyFuture, result: []u8) void {277pub fn @"await"(userdata: ?*anyopaque, any_future: *std.Io.AnyFuture, result: []u8) void {
220 const event_loop: *EventLoop = @alignCast(@ptrCast(userdata));278 const event_loop: *EventLoop = @alignCast(@ptrCast(userdata));
221 const future_fiber: *Fiber = @alignCast(@ptrCast(any_future));279 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];
223 if (@atomicLoad(?*Fiber, &future_fiber.awaiter, .acquire) != Fiber.finished) event_loop.yield(null, &future_fiber.awaiter);281 if (@atomicLoad(?*Fiber, &future_fiber.awaiter, .acquire) != Fiber.finished) event_loop.yield(null, &future_fiber.awaiter);
224 @memcpy(result, result_src);282 @memcpy(result, result_src);
225 event_loop.recycle(future_fiber);283 event_loop.recycle(future_fiber);