batchDrainSubmitted

fn batchDrainSubmitted(
    ev: *Evented,
    maybe_sync: *CancelRegion.Sync.Maybe,
    batch: *Io.Batch,
    concurrency: bool,
) (Io.ConcurrentError || Io.Cancelable)!void

File

Code

fn batchDrainSubmitted(
    ev: *Evented,
    maybe_sync: *CancelRegion.Sync.Maybe,
    batch: *Io.Batch,
    concurrency: bool,
) (Io.ConcurrentError || Io.Cancelable)!void {
    var index = batch.submitted.head;
    if (index == .none) return;
    const thread = try maybe_sync.cancelRegion().awaitIoUring();
    errdefer batch.submitted.head = index;
    while (index != .none) {
        const storage = &batch.storage[index.toIndex()];
        const next_index = storage.submission.node.next;
        if (@as(?Io.Operation.Result, result: switch (storage.submission.operation) {
            .file_read_streaming => |o| {
                const buffer = for (o.data) |buffer| {
                    if (buffer.len > 0) break buffer;
                } else break :result .{ .file_read_streaming = 0 };
                const fd = o.file.handle;
                storage.* = .{ .pending = .{
                    .node = .{ .prev = batch.pending.tail, .next = .none },
                    .tag = .file_read_streaming,
                    .userdata = undefined,
                } };
                thread.enqueue().* = .{
                    .opcode = .READ,
                    .flags = 0,
                    .ioprio = 0,
                    .fd = fd,
                    .off = std.math.maxInt(u64),
                    .addr = @intFromPtr(buffer.ptr),
                    .len = @min(buffer.len, 0xfffff000),
                    .rw_flags = 0,
                    .user_data = @intFromPtr(&storage.pending.userdata) | 0b10,
                    .buf_index = 0,
                    .personality = 0,
                    .splice_fd_in = 0,
                    .addr3 = 0,
                    .resv = 0,
                };
                break :result null;
            },
            .file_write_streaming => |o| {
                const buffer = buffer: {
                    if (o.header.len != 0) break :buffer o.header;
                    for (o.data[0 .. o.data.len - 1]) |buffer| {
                        if (buffer.len > 0) break :buffer buffer;
                    }
                    if (o.splat > 0) break :buffer o.data[o.data.len - 1];
                    break :result .{ .file_write_streaming = 0 };
                };
                const fd = o.file.handle;
                storage.* = .{ .pending = .{
                    .node = .{ .prev = batch.pending.tail, .next = .none },
                    .tag = .file_write_streaming,
                    .userdata = undefined,
                } };
                thread.enqueue().* = .{
                    .opcode = .WRITE,
                    .flags = 0,
                    .ioprio = 0,
                    .fd = fd,
                    .off = std.math.maxInt(u64),
                    .addr = @intFromPtr(buffer.ptr),
                    .len = @min(buffer.len, 0xfffff000),
                    .rw_flags = 0,
                    .user_data = @intFromPtr(&storage.pending.userdata) | 0b10,
                    .buf_index = 0,
                    .personality = 0,
                    .splice_fd_in = 0,
                    .addr3 = 0,
                    .resv = 0,
                };
                break :result null;
            },
            .device_io_control => |o| if (concurrency)
                return error.ConcurrencyUnavailable
            else
                .{ .device_io_control = try ev.deviceIoControl(try maybe_sync.enterSync(ev), o) },
            .net_receive => |o| {
                _ = o;
                @panic("TODO implement batchDrainSubmitted for net_receive");
            },
            .net_read => |o| {
                _ = o;
                @panic("TODO implement batchDrainSubmitted for net_read");
            },
        })) |result| {
            switch (batch.completed.tail) {
                .none => batch.completed.head = index,
                else => |tail_index| batch.storage[tail_index.toIndex()].completion.node.next = index,
            }
            batch.completed.tail = index;
            storage.* = .{ .completion = .{ .node = .{ .next = .none }, .result = result } };
        } else {
            switch (batch.pending.tail) {
                .none => batch.pending.head = index,
                else => |tail_index| batch.storage[tail_index.toIndex()].pending.node.next = index,
            }
            batch.pending.tail = index;
            storage.pending.userdata[0] = @intFromPtr(batch);
        }
        index = next_index;
    }
    batch.submitted = .{ .head = .none, .tail = .none };
}