batchAwaitConcurrent

fn batchAwaitConcurrent(
    userdata: ?*anyopaque,
    batch: *Io.Batch,
    timeout: Io.Timeout,
) Io.Batch.AwaitConcurrentError!void {
    const ev: *Evented = @ptrCast(@alignCast(userdata));
    const queue = try ev.batchDrainSubmitted(batch, true) orelse return;
    if (batch.pending.head == .none) return;
    var waiter: BatchWaiter = .{
        .sleeper = .init(ev.queue, Thread.current().currentFiber()),
        .queue = queue,
    };
    if (batch.completed.head == .none) switch (timeout) {
        .none => {},
        else => {
            const timer = c.dispatch.source_create(.TIMER, 0, .none, queue) orelse
                return error.ConcurrencyUnavailable;
            assert(timer != BatchWaiter.already_signaled);
            timer.as_object().set_context(&waiter);
            timer.set_event_handler(&BatchWaiter.signal);
            timer.set_cancel_handler(&BatchWaiter.@"suspend");
            timer.set_timer(ev.timeFromTimeout(timeout), c.dispatch.TIME_FOREVER, ev.leeway);
            timer.as_object().activate();
            waiter.timer = timer;
        },
    } else BatchWaiter.signal(&waiter);
    queue.as_object().set_context(&waiter);
    ev.yield(.{ .@"resume" = queue.as_object() });
}