fn idle(ev: *Evented, thread: *Thread) void {
var maybe_ready_fiber: ?*Fiber = null;
while (true) {
while (maybe_ready_fiber orelse ev.findReadyFiber(thread)) |ready_fiber| {
ev.yield(ready_fiber, .nothing);
maybe_ready_fiber = null;
}
_ = thread.io_uring.submit_and_wait(1) catch |err| switch (err) {
error.SignalInterrupt => {},
else => |e| @panic(@errorName(e)),
};
var maybe_ready_queue: ?Fiber.Queue = null;
while (true) {
var cqes_buffer: [1 << 8]linux.io_uring_cqe = undefined;
const cqes = cqes_buffer[0 .. thread.io_uring.copy_cqes(&cqes_buffer, 0) catch |err| switch (err) {
error.SignalInterrupt => 0,
else => |e| @panic(@errorName(e)),
}];
if (cqes.len == 0) break;
for (cqes) |cqe| if (cqe.flags & linux.IORING_CQE_F_SKIP == 0) switch (@as(
Completion.Userdata,
@fromBackingInt(@intCast(cqe.user_data)),
)) {
.unused => unreachable, // bad submission queued?
.wakeup => {},
.futex_wake => switch (Completion.errno(.{ .result = cqe.res, .flags = cqe.flags })) {
.SUCCESS => recoverableOsBugDetected(), // success is skipped
.INVAL => {}, // invalid futex_wait() on ptr done elsewhere
.INTR, .CANCELED => recoverableOsBugDetected(), // `Completion.Userdata.futex_wake` is not cancelable
.FAULT => {}, // pointer became invalid while doing the wake
else => recoverableOsBugDetected(), // deadlock due to operating system bug
},
.close => switch (Completion.errno(.{ .result = cqe.res, .flags = cqe.flags })) {
.BADF => recoverableOsBugDetected(), // Always a race condition.
.INTR => {}, // This is still a success. See https://github.com/ziglang/zig/issues/2425
else => {},
},
.cleanup => @panic("failed to notify other threads that we are exiting"),
.exit => {
assert(maybe_ready_fiber == null and maybe_ready_queue == null); // pending async
return;
},
_ => if (@as(?*Fiber, ready_fiber: switch (@as(u2, @truncate(cqe.user_data))) {
0b00 => {
const ready_fiber: *Fiber = @ptrFromInt(cqe.user_data & ~@as(usize, 0b11));
ready_fiber.resultPointer(Completion).* = .{
.result = cqe.res,
.flags = cqe.flags,
};
break :ready_fiber ready_fiber;
},
0b01 => {
thread.enqueue().* = .{
.opcode = .ASYNC_CANCEL,
.flags = linux.IOSQE_CQE_SKIP_SUCCESS,
.ioprio = 0,
.fd = 0,
.off = 0,
.addr = cqe.user_data & ~@as(usize, 0b11),
.len = 0,
.rw_flags = 0,
.user_data = @backingInt(Completion.Userdata.wakeup),
.buf_index = 0,
.personality = 0,
.splice_fd_in = 0,
.addr3 = 0,
.resv = 0,
};
break :ready_fiber null;
},
0b10 => {
const batch_userdata: *Io.Operation.Storage.Pending.Userdata =
@ptrFromInt(cqe.user_data & ~@as(usize, 0b11));
const batch: *Io.Batch = @ptrFromInt(batch_userdata[0]);
var next: usize = 0b00;
batch_userdata[0..3].* = .{ next, @as(u32, @bitCast(cqe.res)), cqe.flags };
while (true) {
next = @cmpxchgWeak(
usize,
@as(*usize, @ptrCast(&batch.userdata)),
next,
cqe.user_data,
.release,
.acquire,
) orelse break;
batch_userdata[0] = next;
}
break :ready_fiber switch (@as(u2, @truncate(next))) {
0b00, 0b01 => @ptrFromInt(next & ~@as(usize, 0b11)),
0b10, 0b11 => null,
};
},
0b11 => switch (Completion.errno(.{ .result = cqe.res, .flags = cqe.flags })) {
.SUCCESS => unreachable, // no event count specified
.TIME => {
const context: *usize = @ptrFromInt(cqe.user_data & ~@as(usize, 0b11));
const fiber = @atomicRmw(usize, context, .Add, 0b01, .acquire);
break :ready_fiber switch (@as(u2, @truncate(fiber))) {
else => unreachable, // timeout completed multiple times
0b00 => @ptrFromInt(fiber & ~@as(usize, 0b11)),
0b10 => null,
};
},
.CANCELED => null, // user data may have been invalidated
else => |err| unexpectedErrno(err) catch null,
},
})) |ready_fiber| {
assert(ready_fiber.status.queue_next == null);
if (maybe_ready_fiber == null) {
maybe_ready_fiber = ready_fiber;
} else if (maybe_ready_queue) |*ready_queue| {
ready_queue.tail.status.queue_next = ready_fiber;
ready_queue.tail = ready_fiber;
} else maybe_ready_queue = .{ .head = ready_fiber, .tail = ready_fiber };
},
};
}
if (maybe_ready_queue) |ready_queue| _ = ev.schedule(thread, ready_queue);
}
}
