batchAwaitConcurrent

fn batchAwaitConcurrent(
    userdata: ?*anyopaque,
    batch: *Io.Batch,
    timeout: Io.Timeout,
) Io.Batch.AwaitConcurrentError!void

File

Code

fn batchAwaitConcurrent(
    userdata: ?*anyopaque,
    batch: *Io.Batch,
    timeout: Io.Timeout,
) Io.Batch.AwaitConcurrentError!void {
    const ev: *Evented = @ptrCast(@alignCast(userdata));
    var maybe_sync: CancelRegion.Sync.Maybe = .{ .cancel_region = .init() };
    defer maybe_sync.deinit(ev);
    try ev.batchDrainSubmitted(&maybe_sync, batch, true);
    maybe_sync.leaveSync(ev);
    const timespec: linux.kernel_timespec, const clock: Io.Clock, const timeout_flags: u32 = while (true) {
        batchDrainReady(batch) catch |err| switch (err) {
            error.Timeout => unreachable, // no timeout
        };
        if (batch.completed.head != .none or batch.pending.head == .none) return;
        switch (timeout) {
            .none => ev.yield(null, .{ .batch_await = batch }),
            .duration => |duration| {
                const ns = duration.raw.toNanoseconds();
                break .{
                    .{
                        .sec = @intCast(@divFloor(ns, std.time.ns_per_s)),
                        .nsec = @intCast(@mod(ns, std.time.ns_per_s)),
                    },
                    duration.clock,
                    0,
                };
            },
            .deadline => |deadline| {
                const ns = deadline.raw.toNanoseconds();
                break .{
                    .{
                        .sec = @intCast(@divFloor(ns, std.time.ns_per_s)),
                        .nsec = @intCast(@mod(ns, std.time.ns_per_s)),
                    },
                    deadline.clock,
                    linux.IORING_TIMEOUT_ABS,
                };
            },
        }
    };
    {
        const thread = try maybe_sync.cancel_region.awaitIoUring();
        thread.enqueue().* = .{
            .opcode = .TIMEOUT,
            .flags = 0,
            .ioprio = 0,
            .fd = 0,
            .off = 0,
            .addr = @intFromPtr(&timespec),
            .len = 1,
            .rw_flags = timeout_flags | @as(u32, switch (clock) {
                .real => linux.IORING_TIMEOUT_REALTIME,
                else => 0,
                .boot => linux.IORING_TIMEOUT_BOOTTIME,
            }),
            .user_data = @intFromPtr(&batch.userdata) | 0b11,
            .buf_index = 0,
            .personality = 0,
            .splice_fd_in = 0,
            .addr3 = 0,
            .resv = 0,
        };
    }
    while (batch.completed.head == .none and batch.pending.head != .none) {
        ev.yield(null, .{ .batch_await = batch });
        batchDrainReady(batch) catch |err| switch (err) {
            error.Timeout => |e| return if (batch.completed.head == .none and
                batch.pending.head != .none) e,
        };
    }
    const thread = try maybe_sync.cancel_region.awaitIoUring();
    thread.enqueue().* = .{
        .opcode = .TIMEOUT_REMOVE,
        .flags = 0,
        .ioprio = 0,
        .fd = 0,
        .off = 0,
        .addr = @intFromPtr(&batch.userdata) | 0b11,
        .len = 0,
        .rw_flags = 0,
        .user_data = @intFromPtr(maybe_sync.cancel_region.fiber),
        .buf_index = 0,
        .personality = 0,
        .splice_fd_in = 0,
        .addr3 = 0,
        .resv = 0,
    };
    ev.yield(null, .nothing);
    switch (maybe_sync.cancel_region.errno()) {
        .SUCCESS => return,
        .BUSY, .NOENT => {},
        else => |err| unexpectedErrno(err) catch {},
    }
    while (true) {
        batchDrainReady(batch) catch |err| switch (err) {
            error.Timeout => return,
        };
        ev.yield(null, .{ .batch_await = batch });
    }
}