batchSourceEvent

fn batchSourceEvent(context: ?*anyopaque) callconv(.c) void {
    const storage: *Io.Operation.Storage = @ptrCast(@alignCast(context));
    const pending = &storage.pending;
    const operation_userdata: *BatchOperationUserdata = .fromErased(&pending.userdata);
    const batch = operation_userdata.batch;
    const source = operation_userdata.source;
    const index: Io.Operation.OptionalIndex = .fromIndex(storage - batch.storage.ptr);
    const result: Io.Operation.Result = result: switch (pending.tag) {
        .file_read_streaming => {
            const operation = &operation_userdata.operation.file_read_streaming;
            break :result .{ .file_read_streaming = fileReadStreamingLimit(
                @intCast(source.get_handle()),
                operation.data_ptr[0..operation.data_len],
                .limited(source.get_data()),
            ) catch |err| switch (err) {
                error.Canceled => return Thread.current().currentFiber().cancel_protection.recancel(),
                error.WouldBlock => return,
                else => |e| e,
            } };
        },
        .file_write_streaming => {
            const operation = &operation_userdata.operation.file_write_streaming;
            break :result .{ .file_write_streaming = fileWriteStreamingLimit(
                @intCast(source.get_handle()),
                operation.header_ptr[0..operation.header_len],
                operation.data_ptr[0..operation.data_len],
                operation.splat,
                .limited(source.get_data()),
            ) catch |err| switch (err) {
                error.Canceled => return Thread.current().currentFiber().cancel_protection.recancel(),
                error.WouldBlock => return,
                else => |e| e,
            } };
        },
        .device_io_control => unreachable,
        .net_receive => @panic("TODO implement batched net_receive"),
        .net_read => @panic("TODO implement batched net_read"),
        .net_write => @panic("TODO implement batched net_write"),
    };

    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 => |next_index| batch.storage[next_index.toIndex()].pending.node.prev = pending.node.prev,
    }

    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;

    source.as_object().release();
    const queue: c.dispatch.queue_t = @ptrCast(batch.userdata);
    const waiter: *BatchWaiter = @ptrCast(@alignCast(queue.as_object().get_context()));
    BatchWaiter.signal(waiter);
}