batchAwaitAsync

fn batchAwaitAsync(userdata: ?*anyopaque, b: *Io.Batch) Io.Cancelable!void

File

Code

fn batchAwaitAsync(userdata: ?*anyopaque, b: *Io.Batch) Io.Cancelable!void {
    const t: *Threaded = @ptrCast(@alignCast(userdata));
    if (is_windows) {
        batchDrainSubmittedWindows(t, b, false) catch |err| switch (err) {
            error.ConcurrencyUnavailable => unreachable, // passed concurrency=false
            else => |e| return e,
        };
        const alertable_syscall = try AlertableSyscall.start();
        while (b.pending.head != .none and b.completed.head == .none) waitForApcOrAlert();
        alertable_syscall.finish();
        return;
    }
    if (have_poll) {
        var poll_buffer: [poll_buffer_len]posix.pollfd = undefined;
        var poll_len: u32 = 0;
        {
            var index = b.submitted.head;
            while (index != .none and poll_len < poll_buffer_len) {
                const submission = &b.storage[index.toIndex()].submission;
                switch (submission.operation) {
                    .file_read_streaming => |o| {
                        poll_buffer[poll_len] = .{
                            .fd = o.file.handle,
                            .events = posix.POLL.IN | posix.POLL.ERR,
                            .revents = 0,
                        };
                        poll_len += 1;
                    },
                    .file_write_streaming => |o| {
                        poll_buffer[poll_len] = .{
                            .fd = o.file.handle,
                            .events = posix.POLL.OUT | posix.POLL.ERR,
                            .revents = 0,
                        };
                        poll_len += 1;
                    },
                    .device_io_control => |o| {
                        poll_buffer[poll_len] = .{
                            .fd = o.file.handle,
                            .events = posix.POLL.OUT | posix.POLL.IN | posix.POLL.ERR,
                            .revents = 0,
                        };
                        poll_len += 1;
                    },
                    .net_receive => |*o| {
                        poll_buffer[poll_len] = .{
                            .fd = o.socket_handle,
                            .events = posix.POLL.IN | posix.POLL.ERR,
                            .revents = 0,
                        };
                        poll_len += 1;
                    },
                    .net_read => |o| {
                        poll_buffer[poll_len] = .{
                            .fd = o.socket_handle,
                            .events = posix.POLL.IN | posix.POLL.ERR,
                            .revents = 0,
                        };
                        poll_len += 1;
                    },
                }
                index = submission.node.next;
            }
        }
        switch (poll_len) {
            0 => return,
            1 => {},
            else => while (true) {
                const timeout_ms: i32 = t: {
                    if (b.completed.head != .none) {
                        // It is legal to call batchWait with already completed
                        // operations in the ring. In such case, we need to avoid
                        // blocking in the poll syscall, but we can still take this
                        // opportunity to find additional ready operations.
                        break :t 0;
                    }
                    break :t std.math.maxInt(i32);
                };
                const syscall = try Syscall.start();
                const rc = posix.system.poll(&poll_buffer, poll_len, timeout_ms);
                syscall.finish();
                switch (posix.errno(rc)) {
                    .SUCCESS => {
                        if (rc == 0) {
                            if (b.completed.head != .none) {
                                // Since there are already completions available in the
                                // queue, this is neither a timeout nor a case for
                                // retrying.
                                return;
                            }
                            continue;
                        }
                        var prev_index: Io.Operation.OptionalIndex = .none;
                        var index = b.submitted.head;
                        for (poll_buffer[0..poll_len]) |poll_entry| {
                            const storage = &b.storage[index.toIndex()];
                            const submission = &storage.submission;
                            const next_index = submission.node.next;
                            if (poll_entry.revents != 0) {
                                const result = try operate(t, submission.operation);

                                switch (prev_index) {
                                    .none => b.submitted.head = next_index,
                                    else => b.storage[prev_index.toIndex()].submission.node.next = next_index,
                                }
                                if (next_index == .none) b.submitted.tail = prev_index;

                                switch (b.completed.tail) {
                                    .none => b.completed.head = index,
                                    else => |tail_index| b.storage[tail_index.toIndex()].completion.node.next = index,
                                }
                                storage.* = .{ .completion = .{ .node = .{ .next = .none }, .result = result } };
                                b.completed.tail = index;
                            } else prev_index = index;
                            index = next_index;
                        }
                        assert(index == .none);
                        return;
                    },
                    .INTR => continue,
                    else => break,
                }
            },
        }
    }

    var tail_index = b.completed.tail;
    defer b.completed.tail = tail_index;
    var index = b.submitted.head;
    errdefer b.submitted.head = index;
    while (index != .none) {
        const storage = &b.storage[index.toIndex()];
        const submission = &storage.submission;
        const next_index = submission.node.next;
        const result = try operate(t, submission.operation);

        switch (tail_index) {
            .none => b.completed.head = index,
            else => b.storage[tail_index.toIndex()].completion.node.next = index,
        }
        storage.* = .{ .completion = .{ .node = .{ .next = .none }, .result = result } };
        tail_index = index;
        index = next_index;
    }
    b.submitted = .{ .head = .none, .tail = .none };
}