authorgravatar for jacobly@ziglang.orgJacob Young <jacobly@ziglang.org> 2025-03-29 10:56:45-04:00
committergravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2025-10-02 16:30:59-07:00
log708bac1a57e6cf8f9e3709ab25fac1ccbaa31520
treecab8ea806afce87a7def6a0553d68078006c9773
parent507f973b5eb3e42d0e8dd151539c929d3b91fe08

EventLoop: fix futex usage

How silly of me to forget that the kernel doesn't implement its own API. The scheduling is not great, but at least doesn't deadlock or hammer.

1 files changed, 41 insertions(+), 38 deletions(-)

lib/std/Io/EventLoop.zig+41-38
...@@ -11,7 +11,7 @@ gpa: Allocator,...@@ -11,7 +11,7 @@ gpa: Allocator,
11mutex: std.Thread.Mutex,11mutex: std.Thread.Mutex,
12queue: std.DoublyLinkedList(void),12queue: std.DoublyLinkedList(void),
13/// Atomic copy of queue.len13/// Atomic copy of queue.len
14queue_len: usize,14queue_len: u32,
15free: std.DoublyLinkedList(void),15free: std.DoublyLinkedList(void),
16main_fiber: Fiber,16main_fiber: Fiber,
17idle_count: usize,17idle_count: usize,
...@@ -20,8 +20,8 @@ exiting: bool,...@@ -20,8 +20,8 @@ exiting: bool,
2020
21threadlocal var thread_index: u32 = undefined;21threadlocal var thread_index: u32 = undefined;
2222
23/// Empirically saw 10KB being used by the self-hosted backend for logging.23/// Empirically saw >128KB being used by the self-hosted backend to panic.
24const idle_stack_size = 64 * 1024;24const idle_stack_size = 256 * 1024;
2525
26const io_uring_entries = 64;26const io_uring_entries = 64;
2727
...@@ -143,6 +143,7 @@ pub fn deinit(el: *EventLoop) void {...@@ -143,6 +143,7 @@ pub fn deinit(el: *EventLoop) void {
143 const allocated_ptr: [*]align(@alignOf(Thread)) u8 = @alignCast(@ptrCast(el.threads.items.ptr));143 const allocated_ptr: [*]align(@alignOf(Thread)) u8 = @alignCast(@ptrCast(el.threads.items.ptr));
144 for (el.threads.items[1..]) |*thread| thread.thread.join();144 for (el.threads.items[1..]) |*thread| thread.thread.join();
145 el.gpa.free(allocated_ptr[0..idle_stack_end_offset]);145 el.gpa.free(allocated_ptr[0..idle_stack_end_offset]);
146 el.* = undefined;
146}147}
147148
148fn yield(el: *EventLoop, optional_fiber: ?*Fiber, pending_task: SwitchMessage.PendingTask) void {149fn yield(el: *EventLoop, optional_fiber: ?*Fiber, pending_task: SwitchMessage.PendingTask) void {
...@@ -151,8 +152,9 @@ fn yield(el: *EventLoop, optional_fiber: ?*Fiber, pending_task: SwitchMessage.Pe...@@ -151,8 +152,9 @@ fn yield(el: *EventLoop, optional_fiber: ?*Fiber, pending_task: SwitchMessage.Pe
151 const ready_fiber: *Fiber = optional_fiber orelse if (ready_node: {152 const ready_fiber: *Fiber = optional_fiber orelse if (ready_node: {
152 el.mutex.lock();153 el.mutex.lock();
153 defer el.mutex.unlock();154 defer el.mutex.unlock();
155 const expected_queue_len = std.math.lossyCast(u32, el.queue.len);
154 const ready_node = el.queue.pop();156 const ready_node = el.queue.pop();
155 @atomicStore(usize, &el.queue_len, el.queue.len, .unordered);157 _ = @cmpxchgStrong(u32, &el.queue_len, expected_queue_len, std.math.lossyCast(u32, el.queue.len), .monotonic, .monotonic);
156 break :ready_node ready_node;158 break :ready_node ready_node;
157 }) |ready_node|159 }) |ready_node|
158 @alignCast(@fieldParentPtr("queue_node", ready_node))160 @alignCast(@fieldParentPtr("queue_node", ready_node))
...@@ -172,20 +174,16 @@ fn yield(el: *EventLoop, optional_fiber: ?*Fiber, pending_task: SwitchMessage.Pe...@@ -172,20 +174,16 @@ fn yield(el: *EventLoop, optional_fiber: ?*Fiber, pending_task: SwitchMessage.Pe
172}174}
173175
174fn schedule(el: *EventLoop, fiber: *Fiber) void {176fn schedule(el: *EventLoop, fiber: *Fiber) void {
177 std.log.debug("scheduling {*}", .{fiber});
175 if (idle_count: {178 if (idle_count: {
176 el.mutex.lock();179 el.mutex.lock();
177 defer el.mutex.unlock();180 defer el.mutex.unlock();
181 const expected_queue_len = std.math.lossyCast(u32, el.queue.len);
178 el.queue.append(&fiber.queue_node);182 el.queue.append(&fiber.queue_node);
179 @atomicStore(usize, &el.queue_len, el.queue.len, .unordered);183 _ = @cmpxchgStrong(u32, &el.queue_len, expected_queue_len, std.math.lossyCast(u32, el.queue.len), .monotonic, .monotonic);
180 break :idle_count el.idle_count;184 break :idle_count el.idle_count;
181 } > 0) {185 } > 0) {
182 _ = std.os.linux.futex2_wake(&el.queue_len, std.math.maxInt(usize), 1, switch (@bitSizeOf(usize)) {186 _ = std.os.linux.futex2_wake(&el.queue_len, std.math.maxInt(u32), 1, std.os.linux.FUTEX2.SIZE_U32 | std.os.linux.FUTEX2.PRIVATE); // TODO: io_uring
183 8 => std.os.linux.FUTEX2.SIZE_U8,
184 16 => std.os.linux.FUTEX2.SIZE_U16,
185 32 => std.os.linux.FUTEX2.SIZE_U32,
186 64 => std.os.linux.FUTEX2.SIZE_U64,
187 else => @compileError("unsupported @sizeOf(usize)"),
188 } | std.os.linux.FUTEX2.PRIVATE); // TODO: io_uring
189 return;187 return;
190 }188 }
191 if (el.threads.items.len == el.threads.capacity) return;189 if (el.threads.items.len == el.threads.capacity) return;
...@@ -226,8 +224,8 @@ fn threadEntry(el: *EventLoop, index: usize) void {...@@ -226,8 +224,8 @@ fn threadEntry(el: *EventLoop, index: usize) void {
226 el.idle();224 el.idle();
227}225}
228226
229const UserData = enum(u64) {227const CompletionKey = enum(u64) {
230 queue_len_futex_wait,228 queue_len_futex_wait = 1,
231 _,229 _,
232};230};
233231
...@@ -235,33 +233,44 @@ fn idle(el: *EventLoop) void {...@@ -235,33 +233,44 @@ fn idle(el: *EventLoop) void {
235 const thread: *Thread = &el.threads.items[thread_index];233 const thread: *Thread = &el.threads.items[thread_index];
236 const iou = &thread.io_uring;234 const iou = &thread.io_uring;
237 var cqes_buffer: [io_uring_entries]std.os.linux.io_uring_cqe = undefined;235 var cqes_buffer: [io_uring_entries]std.os.linux.io_uring_cqe = undefined;
238 var futex_is_scheduled: bool = false;236 var queue_len_futex_is_scheduled: bool = false;
239237
240 while (true) {238 while (true) {
241 el.yield(null, .nothing);239 el.yield(null, .nothing);
242 if (@atomicLoad(bool, &el.exiting, .acquire)) return;240 if (@atomicLoad(bool, &el.exiting, .acquire)) return;
243 if (!futex_is_scheduled) {241 if (!queue_len_futex_is_scheduled) {
244 const sqe = getSqe(&thread.io_uring);242 const sqe = getSqe(&thread.io_uring);
245 sqe.prep_rw(.FUTEX_WAIT, switch (@bitSizeOf(usize)) {243 sqe.prep_rw(.FUTEX_WAIT, std.os.linux.FUTEX2.SIZE_U32 | std.os.linux.FUTEX2.PRIVATE, @intFromPtr(&el.queue_len), 0, 0);
246 8 => std.os.linux.FUTEX2.SIZE_U8,244 sqe.addr3 = std.math.maxInt(u32);
247 16 => std.os.linux.FUTEX2.SIZE_U16,245 sqe.user_data = @intFromEnum(CompletionKey.queue_len_futex_wait);
248 32 => std.os.linux.FUTEX2.SIZE_U32,246 queue_len_futex_is_scheduled = true;
249 64 => std.os.linux.FUTEX2.SIZE_U64,
250 else => @compileError("unsupported @sizeOf(usize)"),
251 } | std.os.linux.FUTEX2.PRIVATE, @intFromPtr(&el.queue_len), 0, 0);
252 sqe.addr3 = std.math.maxInt(u64);
253 sqe.user_data = @intFromEnum(UserData.queue_len_futex_wait);
254 futex_is_scheduled = true;
255 }247 }
256 _ = iou.submit_and_wait(1) catch |err| switch (err) {248 _ = iou.submit_and_wait(1) catch |err| switch (err) {
257 error.SignalInterrupt => 0,249 error.SignalInterrupt => std.log.debug("submit_and_wait: SignalInterrupt", .{}),
258 else => @panic(@errorName(err)),250 else => @panic(@errorName(err)),
259 };251 };
260 for (cqes_buffer[0 .. iou.copy_cqes(&cqes_buffer, 1) catch |err| switch (err) {252 for (cqes_buffer[0 .. iou.copy_cqes(&cqes_buffer, 1) catch |err| switch (err) {
261 error.SignalInterrupt => 0,253 error.SignalInterrupt => cqes_len: {
254 std.log.debug("copy_cqes: SignalInterrupt", .{});
255 break :cqes_len 0;
256 },
262 else => @panic(@errorName(err)),257 else => @panic(@errorName(err)),
263 }]) |cqe| switch (@as(UserData, @enumFromInt(cqe.user_data))) {258 }]) |cqe| switch (@as(CompletionKey, @enumFromInt(cqe.user_data))) {
264 .queue_len_futex_wait => futex_is_scheduled = false,259 .queue_len_futex_wait => {
260 switch (errno(cqe.res)) {
261 .SUCCESS, .AGAIN => {},
262 .INVAL => unreachable,
263 else => |err| {
264 std.posix.unexpectedErrno(err) catch {};
265 @panic("unexpected");
266 },
267 }
268 std.log.debug("{*} woken up with queue size of {d}", .{
269 &thread.idle_context,
270 @atomicLoad(u32, &el.queue_len, .unordered),
271 });
272 queue_len_futex_is_scheduled = false;
273 },
265 _ => {274 _ => {
266 const fiber: *Fiber = @ptrFromInt(cqe.user_data);275 const fiber: *Fiber = @ptrFromInt(cqe.user_data);
267 const res: *i32 = @ptrCast(@alignCast(fiber.resultPointer()));276 const res: *i32 = @ptrCast(@alignCast(fiber.resultPointer()));
...@@ -296,14 +305,8 @@ const SwitchMessage = struct {...@@ -296,14 +305,8 @@ const SwitchMessage = struct {
296 },305 },
297 .exit => {306 .exit => {
298 @atomicStore(bool, &el.exiting, true, .unordered);307 @atomicStore(bool, &el.exiting, true, .unordered);
299 @atomicStore(usize, &el.queue_len, std.math.maxInt(usize), .release);308 @atomicStore(u32, &el.queue_len, std.math.maxInt(u32), .release);
300 _ = std.os.linux.futex2_wake(&el.queue_len, std.math.maxInt(usize), std.math.maxInt(i32), switch (@bitSizeOf(usize)) {309 _ = std.os.linux.futex2_wake(&el.queue_len, std.math.maxInt(u32), std.math.maxInt(i32), std.os.linux.FUTEX2.SIZE_U32 | std.os.linux.FUTEX2.PRIVATE); // TODO: use io_uring
301 8 => std.os.linux.FUTEX2.SIZE_U8,
302 16 => std.os.linux.FUTEX2.SIZE_U16,
303 32 => std.os.linux.FUTEX2.SIZE_U32,
304 64 => std.os.linux.FUTEX2.SIZE_U64,
305 else => @compileError("unsupported @sizeOf(usize)"),
306 } | std.os.linux.FUTEX2.PRIVATE); // TODO: use io_uring
307 },310 },
308 }311 }
309 }312 }