batchDrainReady

fn batchDrainReady(batch: *Io.Batch) Io.Timeout.Error!void

File

Code

fn batchDrainReady(batch: *Io.Batch) Io.Timeout.Error!void {
    while (@atomicRmw(?*anyopaque, &batch.userdata, .Xchg, null, .acquire)) |head| {
        var next: usize = @intFromPtr(head);
        var timeout = false;
        while (cond: switch (@as(u2, @truncate(next))) {
            0b00 => if (timeout) return error.Timeout else false,
            0b01 => {
                assert(!timeout);
                return error.Timeout;
            },
            0b10 => true,
            0b11 => {
                assert(!timeout);
                timeout = true;
                break :cond true;
            },
        }) {
            const operation_userdata: *Io.Operation.Storage.Pending.Userdata =
                @ptrFromInt(next & ~@as(usize, 0b11));
            next = operation_userdata[0];
            const completion: Completion = .{
                .result = @bitCast(@as(u32, @intCast(operation_userdata[1]))),
                .flags = @intCast(operation_userdata[2]),
            };
            const pending: *Io.Operation.Storage.Pending =
                @fieldParentPtr("userdata", operation_userdata);
            const storage: *Io.Operation.Storage = @fieldParentPtr("pending", pending);
            const index: Io.Operation.OptionalIndex = .fromIndex(storage - batch.storage.ptr);
            assert(completion.flags & linux.IORING_CQE_F_SKIP == 0);
            switch (pending.node.prev) {
                .none => batch.pending.head = pending.node.next,
                else => |prev_index| batch.storage[prev_index.toIndex()].pending.node.next =
                    pending.node.next,
            }
            switch (pending.node.next) {
                .none => batch.pending.tail = pending.node.prev,
                else => |prev_index| batch.storage[prev_index.toIndex()].pending.node.prev =
                    pending.node.prev,
            }
            if (@as(?Io.Operation.Result, result: switch (pending.tag) {
                .file_read_streaming => .{
                    .file_read_streaming = switch (completion.errno()) {
                        .SUCCESS => @as(u32, @bitCast(completion.result)),
                        .INTR => 0,
                        .CANCELED => break :result null,
                        .INVAL => |err| errnoBug(err),
                        .FAULT => |err| errnoBug(err),
                        .AGAIN => error.WouldBlock,
                        .BADF => |err| errnoBug(err), // File descriptor used after closed
                        .IO => error.InputOutput,
                        .ISDIR => error.IsDir,
                        .NOBUFS => error.SystemResources,
                        .NOMEM => error.SystemResources,
                        .NOTCONN => error.SocketUnconnected,
                        .CONNRESET => error.ConnectionResetByPeer,
                        else => |err| unexpectedErrno(err),
                    },
                },
                .file_write_streaming => .{
                    .file_write_streaming = switch (completion.errno()) {
                        .SUCCESS => @as(u32, @bitCast(completion.result)),
                        .INTR => 0,
                        .CANCELED => break :result null,
                        .INVAL => |err| errnoBug(err),
                        .FAULT => |err| errnoBug(err),
                        .AGAIN => error.WouldBlock,
                        .BADF => error.NotOpenForWriting, // Can be a race condition.
                        .DESTADDRREQ => |err| errnoBug(err), // `connect` was never called.
                        .DQUOT => error.DiskQuota,
                        .FBIG => error.FileTooBig,
                        .IO => error.InputOutput,
                        .NOSPC => error.NoSpaceLeft,
                        .PERM => error.PermissionDenied,
                        .PIPE => error.BrokenPipe,
                        .CONNRESET => |err| errnoBug(err), // Not a socket handle.
                        .BUSY => error.DeviceBusy,
                        else => |err| unexpectedErrno(err),
                    },
                },
                .device_io_control => unreachable,
                .net_receive => @panic("TODO"),
                .net_read => @panic("TODO"),
            })) |result| {
                switch (batch.completed.tail) {
                    .none => batch.completed.head = index,
                    else => |tail_index| batch.storage[tail_index.toIndex()].completion.node.next =
                        index,
                }
                storage.* = .{ .completion = .{ .node = .{ .next = .none }, .result = result } };
                batch.completed.tail = index;
            } else {
                switch (batch.unused.tail) {
                    .none => batch.unused.head = index,
                    else => |tail_index| batch.storage[tail_index.toIndex()].unused.next = index,
                }
                storage.* = .{ .unused = .{ .prev = batch.unused.tail, .next = .none } };
                batch.unused.tail = index;
            }
        }
    }
}