SwitchMessage

const SwitchMessage = struct

File

Code

const SwitchMessage = struct {
    contexts: Io.fiber.Switch,
    pending_task: PendingTask,

    const PendingTask = union(enum) {
        nothing,
        reschedule,
        await: *Fiber,
        group_await: Group,
        group_cancel: Group,
        batch_await: *Io.Batch,
        destroy,
        exit,
    };

    fn handle(message: *const SwitchMessage, ev: *Evented) void {
        const thread: *Thread = .current();
        thread.current_context = message.contexts.new;
        if (tracy.enable) {
            if (message.contexts.new != &thread.idle_context) {
                const fiber: *Fiber = @alignCast(@fieldParentPtr("context", message.contexts.new));
                tracy.fiberEnter(fiber.name);
            } else tracy.fiberLeave();
        }
        switch (message.pending_task) {
            .nothing => {},
            .reschedule => if (message.contexts.old != &thread.idle_context) {
                const fiber: *Fiber = @alignCast(@fieldParentPtr("context", message.contexts.old));
                assert(fiber.status.queue_next == null);
                _ = ev.schedule(thread, .{ .head = fiber, .tail = fiber });
            },
            .await => |awaiting| {
                const awaiter: *Fiber = @alignCast(@fieldParentPtr("context", message.contexts.old));
                assert(awaiter.status.queue_next == null);
                if (@atomicRmw(?*Fiber, &awaiting.link.awaiter, .Xchg, awaiter, .acq_rel) ==
                    Fiber.finished) _ = ev.schedule(thread, .{ .head = awaiter, .tail = awaiter });
            },
            .group_await => |group| {
                const fiber: *Fiber = @alignCast(@fieldParentPtr("context", message.contexts.old));
                if (group.await(ev, fiber))
                    _ = ev.schedule(thread, .{ .head = fiber, .tail = fiber });
            },
            .group_cancel => |group| {
                const fiber: *Fiber = @alignCast(@fieldParentPtr("context", message.contexts.old));
                if (group.cancel(ev, fiber))
                    _ = ev.schedule(thread, .{ .head = fiber, .tail = fiber });
            },
            .batch_await => |batch| {
                const fiber: *Fiber = @alignCast(@fieldParentPtr("context", message.contexts.old));
                if (@cmpxchgStrong(
                    ?*anyopaque,
                    &batch.userdata,
                    null,
                    fiber,
                    .release,
                    .monotonic,
                )) |head| {
                    assert(@as(u2, @truncate(@intFromPtr(head))) != 0b00);
                    _ = ev.schedule(thread, .{ .head = fiber, .tail = fiber });
                }
            },
            .destroy => {
                const fiber: *Fiber = @alignCast(@fieldParentPtr("context", message.contexts.old));
                fiber.destroy();
            },
            .exit => for (
                ev.threads.allocated[0..@atomicLoad(u32, &ev.threads.active, .acquire)],
            ) |*each_thread| {
                thread.enqueue().* = .{
                    .opcode = .MSG_RING,
                    .flags = linux.IOSQE_CQE_SKIP_SUCCESS,
                    .ioprio = 0,
                    .fd = each_thread.io_uring.fd,
                    .off = @backingInt(Completion.Userdata.exit),
                    .addr = @backingInt(linux.IORING_MSG_RING_COMMAND.DATA),
                    .len = 0,
                    .rw_flags = 0,
                    .user_data = @backingInt(Completion.Userdata.cleanup),
                    .buf_index = 0,
                    .personality = 0,
                    .splice_fd_in = 0,
                    .addr3 = 0,
                    .resv = 0,
                };
            },
        }
    }
}