batchAwaitConcurrent

fn batchAwaitConcurrent(userdata: ?*anyopaque, b: *Io.Batch, timeout: Io.Timeout) Io.Batch.AwaitConcurrentError!void

File

Code

fn batchAwaitConcurrent(userdata: ?*anyopaque, b: *Io.Batch, timeout: Io.Timeout) Io.Batch.AwaitConcurrentError!void {
    const t: *Threaded = @ptrCast(@alignCast(userdata));
    if (is_windows) {
        const deadline: ?Io.Clock.Timestamp = timeout.toTimestamp(io(t));
        try batchDrainSubmittedWindows(t, b, true);
        while (b.pending.head != .none and b.completed.head == .none) {
            var delay_interval: windows.LARGE_INTEGER = interval: {
                const d = deadline orelse break :interval std.math.minInt(windows.LARGE_INTEGER);
                break :interval timeoutToWindowsInterval(.{ .deadline = d }).?;
            };
            const alertable_syscall = try AlertableSyscall.start();
            const delay_rc = windows.ntdll.NtDelayExecution(.TRUE, &delay_interval);
            alertable_syscall.finish();
            switch (delay_rc) {
                .SUCCESS, .TIMEOUT => {
                    // The thread woke due to the timeout. Although spurious
                    // timeouts are OK, when no deadline is passed we must not
                    // return `error.Timeout`.
                    if (timeout != .none and b.completed.head == .none) return error.Timeout;
                },
                else => {},
            }
        }
        return;
    }
    if (native_os == .wasi) {
        // TODO call poll_oneoff
        return error.ConcurrencyUnavailable;
    }
    if (!have_poll) return error.ConcurrencyUnavailable;
    var poll_buffer: [poll_buffer_len]posix.pollfd = undefined;
    var poll_storage: struct {
        gpa: Allocator,
        batch: *Io.Batch,
        slice: []posix.pollfd,
        len: u32,

        fn add(storage: *@This(), fd: File.Handle, events: @FieldType(posix.pollfd, "events")) Io.ConcurrentError!void {
            const len = storage.len;
            if (len == poll_buffer_len) {
                const slice: []posix.pollfd = if (storage.batch.userdata) |batch_userdata|
                    @as([*]posix.pollfd, @ptrCast(@alignCast(batch_userdata)))[0..storage.batch.storage.len]
                else allocation: {
                    const allocation = storage.gpa.alloc(posix.pollfd, storage.batch.storage.len) catch
                        return error.ConcurrencyUnavailable;
                    storage.batch.userdata = allocation.ptr;
                    break :allocation allocation;
                };
                @memcpy(slice[0..poll_buffer_len], storage.slice);
                storage.slice = slice;
            }
            storage.slice[len] = .{
                .fd = fd,
                .events = events,
                .revents = 0,
            };
            storage.len = len + 1;
        }
    } = .{ .gpa = t.allocator, .batch = b, .slice = &poll_buffer, .len = 0 };
    {
        var index = b.submitted.head;
        while (index != .none) {
            const storage = &b.storage[index.toIndex()];
            const submission = storage.submission;
            switch (submission.operation) {
                .file_read_streaming => |o| try poll_storage.add(o.file.handle, posix.POLL.IN | posix.POLL.ERR),
                .file_write_streaming => |o| try poll_storage.add(o.file.handle, posix.POLL.OUT | posix.POLL.ERR),
                .device_io_control => |o| try poll_storage.add(o.file.handle, posix.POLL.IN | posix.POLL.OUT | posix.POLL.ERR),
                .net_receive => |*o| nb: {
                    var data_i: usize = 0;
                    const result: Io.Operation.Result = .{ .net_receive = for (o.message_buffer, 0..) |*msg, msg_i| {
                        const remaining_data_buffer = o.data_buffer[data_i..];
                        netReceivePosix(o.socket_handle, msg, remaining_data_buffer, o.flags, true) catch |err| switch (err) {
                            error.Canceled => |e| return e,
                            error.WouldBlock => {
                                if (msg_i != 0) break .{ null, msg_i };
                                try poll_storage.add(o.socket_handle, posix.POLL.IN | posix.POLL.ERR);
                                break :nb;
                            },
                            else => |e| break .{ e, 0 },
                        };
                        data_i += msg.data.len;
                    } else .{ null, o.message_buffer.len } };
                    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;
                },
                .net_read => |o| try poll_storage.add(o.socket_handle, posix.POLL.IN | posix.POLL.ERR),
            }
            index = submission.node.next;
        }
    }
    switch (poll_storage.len) {
        0 => return,
        1 => if (timeout == .none and b.completed.head == .none) {
            const index = b.submitted.head;
            const storage = &b.storage[index.toIndex()];
            const result = try operate(t, storage.submission.operation);

            b.submitted = .{ .head = .none, .tail = .none };

            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;
            return;
        },
        else => {},
    }
    const t_io = io(t);
    const deadline = timeout.toTimestamp(t_io);
    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;
            }
            const d = deadline orelse break :t -1;
            const duration = d.durationFromNow(t_io);
            break :t @min(@max(0, duration.raw.toMilliseconds()), std.math.maxInt(i32));
        };
        const syscall = try Syscall.start();
        const rc = posix.system.poll(poll_storage.slice.ptr, poll_storage.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;
                    }
                    // Although spurious timeouts are OK, when no deadline is
                    // passed we must not return `error.Timeout`.
                    if (deadline == null) continue;
                    return error.Timeout;
                }
                var prev_index: Io.Operation.OptionalIndex = .none;
                var index = b.submitted.head;
                for (poll_storage.slice[0..poll_storage.len]) |poll_entry| {
                    const submission = &b.storage[index.toIndex()].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,
                        }
                        b.completed.tail = index;
                        b.storage[index.toIndex()] = .{ .completion = .{
                            .node = .{ .next = .none },
                            .result = result,
                        } };
                    } else prev_index = index;
                    index = next_index;
                }
                assert(index == .none);
                return;
            },
            .INTR => continue,
            else => return error.ConcurrencyUnavailable,
        }
    }
}