netRead

fn netRead(userdata: ?*anyopaque, fd: net.Socket.Handle, data: [][]u8) net.Stream.Reader.Error!usize

File

Code

fn netRead(userdata: ?*anyopaque, fd: net.Socket.Handle, data: [][]u8) net.Stream.Reader.Error!usize {
    const k: *Kqueue = @ptrCast(@alignCast(userdata));

    var iovecs_buffer: [max_iovecs_len]posix.iovec = undefined;
    var i: usize = 0;
    for (data) |buf| {
        if (iovecs_buffer.len - i == 0) break;
        if (buf.len != 0) {
            iovecs_buffer[i] = .{ .base = buf.ptr, .len = buf.len };
            i += 1;
        }
    }
    const dest = iovecs_buffer[0..i];
    assert(dest[0].len > 0);

    while (true) {
        try k.checkCancel();
        const rc = posix.system.readv(fd, dest.ptr, @intCast(dest.len));
        switch (posix.errno(rc)) {
            .SUCCESS => return @intCast(rc),
            .INTR => continue,
            .CANCELED => return error.Canceled,
            .AGAIN => {
                const thread: *Thread = .current();
                const fiber = thread.currentFiber();
                const ident: u32 = @bitCast(fd);
                const filter = std.c.EVFILT.READ;
                const gop = thread.wait_queues.getOrPut(k.gpa, .{
                    .ident = ident,
                    .filter = filter,
                }) catch return error.SystemResources;
                if (gop.found_existing) {
                    const tail_fiber = gop.value_ptr.*;
                    assert(tail_fiber.queue_next == null);
                    tail_fiber.queue_next = fiber;
                    gop.value_ptr.* = fiber;
                } else {
                    gop.value_ptr.* = fiber;
                    const changes = [_]posix.Kevent{
                        .{
                            .ident = ident,
                            .filter = filter,
                            .flags = std.c.EV.ADD | std.c.EV.ONESHOT,
                            .fflags = 0,
                            .data = 0,
                            .udata = @intFromPtr(fiber),
                        },
                    };
                    assert(0 == (kevent(thread.kq_fd, &changes, &.{}, null) catch |err| {
                        // TODO handle EINTR for cancellation purposes
                        @panic(@errorName(err)); // TODO
                    }));
                }
                yield(k, null, .nothing);
                continue;
            },

            .INVAL => |err| return errnoBug(err),
            .FAULT => |err| return errnoBug(err),
            .BADF => |err| return errnoBug(err), // File descriptor used after closed.
            .NOBUFS => return error.SystemResources,
            .NOMEM => return error.SystemResources,
            .NOTCONN => return error.SocketUnconnected,
            .CONNRESET => return error.ConnectionResetByPeer,
            .TIMEDOUT => return error.Timeout,
            .PIPE => return error.SocketUnconnected,
            .NETDOWN => return error.NetworkDown,
            else => |err| return posix.unexpectedErrno(err),
        }
    }
}