Select direct queue push

Description

Select direct queue push.

• A separate thread pushes a value directly into the Select queue via putOneUncancelable.
• This bypasses the usual sel.concurrent path — it's a raw queue push.
• sel.await() receives the .direct value directly, then cancels the blocked inbox source.

Diagram

 wild thread ──sel.queue.putOneUncancelable──► Select queue
              (bypasses concurrent fn mechanism)
 │
 sel.await() ──► .direct u32 value
 │
 sel.cancelDiscard() ──► cancels blocking .inbox source

Source

pub fn select_direct_queue_push(allocator: std.mem.Allocator, io: std.Io) !void {
    const mbh: MailboxHandle = try mailbox.new(io, allocator);
    defer {
        var rem: std.DoublyLinkedList = mailbox.close(mbh);
        items.freeList(&rem, allocator);
        mailbox.destroy(mbh, allocator);
    }

    var buf: [4]MasterEvent = undefined;
    var sel: std.Io.Select(MasterEvent) = std.Io.Select(MasterEvent).init(io, &buf);
    var pusher_fut = try setupSourcesAndPusher(mbh, io, &sel);
    try awaitDirectPushAndShutdown(&sel, &pusher_fut, io);
    std.log.info("done: direct push bypassed concurrent fn path", .{});
}

const MasterEvent = union(enum) {
    inbox: mailbox.ReceiveResult,
    direct: u32,
};

fn pusherFn(sel_ptr: *std.Io.Select(MasterEvent)) void {
    sel_ptr.queue.putOneUncancelable(sel_ptr.io, .{ .direct = 99 }) catch {};
}

fn setupSourcesAndPusher(mbh: MailboxHandle, io: std.Io, sel: *std.Io.Select(MasterEvent)) !std.Io.Future(void) {
    try sel.concurrent(.inbox, mailbox.receiveResult, .{ mbh, null });
    return io.concurrent(pusherFn, .{sel});
}

fn awaitDirectPushAndShutdown(sel: *std.Io.Select(MasterEvent), pusher_fut: *std.Io.Future(void), io: std.Io) !void {
    const event: MasterEvent = try sel.await();
    switch (event) {
        .direct => |v| {
            try helpers.expect(error.SelectDirectPushFailed, v == 99, "wrong direct push value");
            std.log.info("direct push: received {d}", .{v});
        },
        .inbox => return error.SelectDirectPushFailed,
    }
    pusher_fut.await(io);
    sel.cancelDiscard();
}

const items = @import("../items/items.zig");
const helpers = @import("../helpers/helpers.zig");
const matryoshka = @import("matryoshka");
const std = @import("std");
const mailbox = matryoshka.mailbox;
const polynode = matryoshka.polynode;
const Slot = polynode.Slot;
const MailboxHandle = mailbox.MailboxHandle;