idle

fn idle(ev: *Evented, thread: *Thread) void

File

Code

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);
    }
}