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-07-20 10:38:38-07:00
log90cc7f4adf62e0c95af9b19cde880f05893fcc67
tree279d3227d78926a5533ea8c4c42d1d8ada7dc7d5
parent4f214b97ec73bb9122a6d50aedbfdb0c4677fb4c

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,
1111mutex: std.Thread.Mutex,
1212queue: std.DoublyLinkedList(void),
1313/// Atomic copy of queue.len
14queue_len: usize,
14queue_len: u32,
1515free: std.DoublyLinkedList(void),
1616main_fiber: Fiber,
1717idle_count: usize,
......@@ -20,8 +20,8 @@ exiting: bool,
2020
2121threadlocal var thread_index: u32 = undefined;
2222
23/// Empirically saw 10KB being used by the self-hosted backend for logging.
24const idle_stack_size = 64 * 1024;
23/// Empirically saw >128KB being used by the self-hosted backend to panic.
24const idle_stack_size = 256 * 1024;
2525
2626const io_uring_entries = 64;
2727
......@@ -143,6 +143,7 @@ pub fn deinit(el: *EventLoop) void {
143143 const allocated_ptr: [*]align(@alignOf(Thread)) u8 = @alignCast(@ptrCast(el.threads.items.ptr));
144144 for (el.threads.items[1..]) |*thread| thread.thread.join();
145145 el.gpa.free(allocated_ptr[0..idle_stack_end_offset]);
146 el.* = undefined;
146147}
147148
148149fn 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
151152 const ready_fiber: *Fiber = optional_fiber orelse if (ready_node: {
152153 el.mutex.lock();
153154 defer el.mutex.unlock();
155 const expected_queue_len = std.math.lossyCast(u32, el.queue.len);
154156 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);
156158 break :ready_node ready_node;
157159 }) |ready_node|
158160 @alignCast(@fieldParentPtr("queue_node", ready_node))
......@@ -172,20 +174,16 @@ fn yield(el: *EventLoop, optional_fiber: ?*Fiber, pending_task: SwitchMessage.Pe
172174}
173175
174176fn schedule(el: *EventLoop, fiber: *Fiber) void {
177 std.log.debug("scheduling {*}", .{fiber});
175178 if (idle_count: {
176179 el.mutex.lock();
177180 defer el.mutex.unlock();
181 const expected_queue_len = std.math.lossyCast(u32, el.queue.len);
178182 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);
180184 break :idle_count el.idle_count;
181185 } > 0) {
182 _ = std.os.linux.futex2_wake(&el.queue_len, std.math.maxInt(usize), 1, switch (@bitSizeOf(usize)) {
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
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
189187 return;
190188 }
191189 if (el.threads.items.len == el.threads.capacity) return;
......@@ -226,8 +224,8 @@ fn threadEntry(el: *EventLoop, index: usize) void {
226224 el.idle();
227225}
228226
229const UserData = enum(u64) {
230 queue_len_futex_wait,
227const CompletionKey = enum(u64) {
228 queue_len_futex_wait = 1,
231229 _,
232230};
233231
......@@ -235,33 +233,44 @@ fn idle(el: *EventLoop) void {
235233 const thread: *Thread = &el.threads.items[thread_index];
236234 const iou = &thread.io_uring;
237235 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
240238 while (true) {
241239 el.yield(null, .nothing);
242240 if (@atomicLoad(bool, &el.exiting, .acquire)) return;
243 if (!futex_is_scheduled) {
241 if (!queue_len_futex_is_scheduled) {
244242 const sqe = getSqe(&thread.io_uring);
245 sqe.prep_rw(.FUTEX_WAIT, switch (@bitSizeOf(usize)) {
246 8 => std.os.linux.FUTEX2.SIZE_U8,
247 16 => std.os.linux.FUTEX2.SIZE_U16,
248 32 => std.os.linux.FUTEX2.SIZE_U32,
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;
243 sqe.prep_rw(.FUTEX_WAIT, std.os.linux.FUTEX2.SIZE_U32 | std.os.linux.FUTEX2.PRIVATE, @intFromPtr(&el.queue_len), 0, 0);
244 sqe.addr3 = std.math.maxInt(u32);
245 sqe.user_data = @intFromEnum(CompletionKey.queue_len_futex_wait);
246 queue_len_futex_is_scheduled = true;
255247 }
256248 _ = iou.submit_and_wait(1) catch |err| switch (err) {
257 error.SignalInterrupt => 0,
249 error.SignalInterrupt => std.log.debug("submit_and_wait: SignalInterrupt", .{}),
258250 else => @panic(@errorName(err)),
259251 };
260252 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 },
262257 else => @panic(@errorName(err)),
263 }]) |cqe| switch (@as(UserData, @enumFromInt(cqe.user_data))) {
264 .queue_len_futex_wait => futex_is_scheduled = false,
258 }]) |cqe| switch (@as(CompletionKey, @enumFromInt(cqe.user_data))) {
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 },
265274 _ => {
266275 const fiber: *Fiber = @ptrFromInt(cqe.user_data);
267276 const res: *i32 = @ptrCast(@alignCast(fiber.resultPointer()));
......@@ -296,14 +305,8 @@ const SwitchMessage = struct {
296305 },
297306 .exit => {
298307 @atomicStore(bool, &el.exiting, true, .unordered);
299 @atomicStore(usize, &el.queue_len, std.math.maxInt(usize), .release);
300 _ = std.os.linux.futex2_wake(&el.queue_len, std.math.maxInt(usize), std.math.maxInt(i32), switch (@bitSizeOf(usize)) {
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
308 @atomicStore(u32, &el.queue_len, std.math.maxInt(u32), .release);
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
307310 },
308311 }
309312 }