batchDrainSubmitted

fn batchDrainSubmitted(
    ev: *Evented,
    batch: *Io.Batch,
    concurrency: bool,
) (Io.ConcurrentError || Io.Cancelable)!?c.dispatch.queue_t

File

Code

fn batchDrainSubmitted(
    ev: *Evented,
    batch: *Io.Batch,
    concurrency: bool,
) (Io.ConcurrentError || Io.Cancelable)!?c.dispatch.queue_t {
    var index = batch.submitted.head;
    if (index == .none) return @ptrCast(batch.userdata);
    errdefer batch.submitted.head = index;
    const maybe_queue: ?c.dispatch.queue_t = if (batch.userdata) |batch_userdata|
        @ptrCast(batch_userdata)
    else maybe_queue: {
        const queue = c.dispatch.queue_create_with_target(
            "org.ziglang.std.Io.Dispatch.Batch",
            .SERIAL(),
            ev.queue,
        ) orelse if (concurrency) return error.ConcurrencyUnavailable else break :maybe_queue null;
        queue.as_object().@"suspend"();
        batch.userdata = queue;
        break :maybe_queue queue;
    };
    while (index != .none) {
        const storage = &batch.storage[index.toIndex()];
        const next_index = storage.submission.node.next;
        if (@as(?Io.Operation.Result, result: {
            if (maybe_queue) |queue| switch (storage.submission.operation) {
                .file_read_streaming => |operation| {
                    const data = for (operation.data, 0..) |buffer, data_index| {
                        if (buffer.len > 0) break operation.data[data_index..];
                    } else break :result .{ .file_read_streaming = 0 };
                    const source = c.dispatch.source_create(
                        .READ,
                        @bitCast(@as(isize, operation.file.handle)),
                        .none,
                        queue,
                    ) orelse break :result .{ .file_read_streaming = error.SystemResources };
                    storage.* = .{ .pending = .{
                        .node = .{ .prev = batch.pending.tail, .next = .none },
                        .tag = .file_read_streaming,
                        .userdata = undefined,
                    } };
                    const operation_userdata: *BatchOperationUserdata =
                        .fromErased(&storage.pending.userdata);
                    operation_userdata.* = .{
                        .batch = batch,
                        .source = source,
                        .operation = .{ .file_read_streaming = .{
                            .data_ptr = data.ptr,
                            .data_len = data.len,
                        } },
                    };
                    source.as_object().set_context(storage);
                    source.set_event_handler(&batchSourceEvent);
                    source.set_cancel_handler(&batchSourceCancel);
                    source.as_object().activate();
                    break :result null;
                },
                .file_write_streaming => |operation| {
                    const data = for (operation.data, 0..) |buffer, data_index| {
                        if (buffer.len > 0) break operation.data[data_index..];
                    } else if (operation.header.len > 0)
                        operation.data[0..1]
                    else
                        break :result .{ .file_write_streaming = 0 };
                    const source = c.dispatch.source_create(
                        .WRITE,
                        @bitCast(@as(isize, operation.file.handle)),
                        .none,
                        queue,
                    ) orelse break :result .{ .file_write_streaming = error.SystemResources };
                    storage.* = .{ .pending = .{
                        .node = .{ .prev = batch.pending.tail, .next = .none },
                        .tag = .file_write_streaming,
                        .userdata = undefined,
                    } };
                    const operation_userdata: *BatchOperationUserdata =
                        .fromErased(&storage.pending.userdata);
                    operation_userdata.* = .{
                        .batch = batch,
                        .source = source,
                        .operation = .{ .file_write_streaming = .{
                            .header_ptr = operation.header.ptr,
                            .header_len = operation.header.len,
                            .data_ptr = data.ptr,
                            .data_len = data.len,
                            .splat = operation.splat,
                        } },
                    };
                    source.as_object().set_context(storage);
                    source.set_event_handler(&batchSourceEvent);
                    source.set_cancel_handler(&batchSourceCancel);
                    source.as_object().activate();
                    break :result null;
                },
                .device_io_control => {},
                .net_receive => @panic("TODO implement batched net_receive"),
                .net_read => @panic("TODO implement batched net_read"),
            };
            if (concurrency) return error.ConcurrencyUnavailable;
            break :result try operate(ev, storage.submission.operation);
        })) |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;
        }
        index = next_index;
    }
    batch.submitted = .{ .head = .none, .tail = .none };
    return maybe_queue;
}