schedule

fn schedule(ev: *Evented, thread: *Thread, ready_queue: Fiber.Queue) bool

File

Code

fn schedule(ev: *Evented, thread: *Thread, ready_queue: Fiber.Queue) bool {
    // shared fields of previous `Thread` must be initialized before later ones are marked as active
    const new_thread_index = @atomicLoad(u32, &ev.threads.active, .acquire);
    for (0..@min(max_idle_search, new_thread_index)) |_| {
        defer thread.idle_search_index += 1;
        if (thread.idle_search_index == new_thread_index) thread.idle_search_index = 0;
        const idle_search_thread = &ev.threads.allocated[0..new_thread_index][thread.idle_search_index];
        if (idle_search_thread == thread) continue;
        if (@cmpxchgWeak(
            ?*Fiber,
            &idle_search_thread.ready_queue,
            null,
            ready_queue.head,
            .release,
            .monotonic,
        )) |_| continue;
        thread.enqueue().* = .{
            .opcode = .MSG_RING,
            .flags = linux.IOSQE_CQE_SKIP_SUCCESS,
            .ioprio = 0,
            .fd = idle_search_thread.io_uring.fd,
            .off = @backingInt(Completion.Userdata.wakeup),
            .addr = @backingInt(linux.IORING_MSG_RING_COMMAND.DATA),
            .len = 0,
            .rw_flags = 0,
            .user_data = @backingInt(Completion.Userdata.wakeup),
            .buf_index = 0,
            .personality = 0,
            .splice_fd_in = 0,
            .addr3 = 0,
            .resv = 0,
        };
        return true;
    }
    spawn_thread: {
        // previous failed reservations must have completed before retrying
        if (new_thread_index == ev.threads.allocated.len or @cmpxchgWeak(
            u32,
            &ev.threads.reserved,
            new_thread_index,
            new_thread_index + 1,
            .acquire,
            .monotonic,
        ) != null) break :spawn_thread;
        const new_thread = &ev.threads.allocated[new_thread_index];
        const next_thread_index = new_thread_index + 1;
        var params = std.mem.zeroInit(linux.io_uring_params, .{
            .flags = linux.IORING_SETUP_ATTACH_WQ |
                linux.IORING_SETUP_R_DISABLED |
                linux.IORING_SETUP_COOP_TASKRUN |
                linux.IORING_SETUP_SINGLE_ISSUER,
            .wq_fd = @as(u32, @intCast(ev.threads.allocated[0].io_uring.fd)),
        });
        new_thread.* = .{
            .required_align = {},
            .thread = undefined,
            .idle_context = undefined,
            .current_context = &new_thread.idle_context,
            .ready_queue = ready_queue.head,
            .free_queue = null,
            .io_uring = IoUring.init_params(@as(u16, 1) << ev.log2_ring_entries, &params) catch |err| {
                @atomicStore(u32, &ev.threads.reserved, new_thread_index, .release);
                // no more access to `thread` after giving up reservation
                log.warn("unable to create worker thread due to io_uring init failure: {s}", .{
                    @errorName(err),
                });
                break :spawn_thread;
            },
            .idle_search_index = 0,
            .steal_ready_search_index = 0,
            .steal_free_search_index = 0,
            .name_arena = .{},
            .csprng = .uninitialized,
        };
        new_thread.thread = std.Thread.spawn(.{
            .stack_size = idle_stack_size,
            .allocator = ev.allocator(),
        }, threadEntry, .{ ev, new_thread_index }) catch |err| {
            new_thread.io_uring.deinit();
            @atomicStore(u32, &ev.threads.reserved, new_thread_index, .release);
            // no more access to `thread` after giving up reservation
            log.warn("unable to create worker thread due spawn failure: {s}", .{@errorName(err)});
            break :spawn_thread;
        };
        // shared fields of `Thread` must be initialized before being marked active
        @atomicStore(u32, &ev.threads.active, next_thread_index, .release);
        return false;
    }
    // nobody wanted it, so just queue it on ourselves
    while (true) ready_queue.tail.status.queue_next = @cmpxchgWeak(
        ?*Fiber,
        &thread.ready_queue,
        ready_queue.tail.status.queue_next,
        ready_queue.head,
        .acq_rel,
        .acquire,
    ) orelse break;
    return false;
}