netReceive

fn netReceive(
    ev: *Evented,
    cancel_region: *CancelRegion,
    handle: net.Socket.Handle,
    message_buffer: []net.IncomingMessage,
    data_buffer: []u8,
    flags: net.ReceiveFlags,
) struct

File

Code

fn netReceive(
    ev: *Evented,
    cancel_region: *CancelRegion,
    handle: net.Socket.Handle,
    message_buffer: []net.IncomingMessage,
    data_buffer: []u8,
    flags: net.ReceiveFlags,
) struct { ?net.Socket.ReceiveError, usize } {
    var message_i: usize = 0;
    var data_i: usize = 0;

    while (true) {
        if (message_buffer.len - message_i == 0) return .{ null, message_i };
        const message = &message_buffer[message_i];
        const remaining_data_buffer = data_buffer[data_i..];
        var storage: PosixAddress = undefined;
        var iov: iovec = .{ .base = remaining_data_buffer.ptr, .len = remaining_data_buffer.len };
        var msg: linux.msghdr = .{
            .name = &storage.any,
            .namelen = @sizeOf(PosixAddress),
            .iov = (&iov)[0..1],
            .iovlen = 1,
            .control = message.control.ptr,
            .controllen = @intCast(message.control.len),
            .flags = undefined,
        };

        const thread = cancel_region.awaitIoUring() catch |err| return .{ err, message_i };
        thread.enqueue().* = .{
            .opcode = .RECVMSG,
            .flags = 0,
            .ioprio = 0,
            .fd = handle,
            .off = 0,
            .addr = @intFromPtr(&msg),
            .len = 0,
            .rw_flags = linux.MSG.NOSIGNAL |
                @as(u32, if (flags.oob) linux.MSG.OOB else 0) |
                @as(u32, if (flags.peek) linux.MSG.PEEK else 0) |
                @as(u32, if (flags.trunc) linux.MSG.TRUNC else 0),
            .user_data = @intFromPtr(cancel_region.fiber),
            .buf_index = 0,
            .personality = 0,
            .splice_fd_in = 0,
            .addr3 = 0,
            .resv = 0,
        };
        ev.yield(null, .nothing);
        const completion = cancel_region.completion();
        switch (completion.errno()) {
            .SUCCESS => {
                const data = remaining_data_buffer[0..@intCast(completion.result)];
                data_i += data.len;
                message.* = .{
                    .from = addressFromPosix(&storage),
                    .data = data,
                    .control = if (msg.control) |ptr| @as([*]u8, @ptrCast(ptr))[0..msg.controllen] else message.control,
                    .flags = .{
                        .eor = msg.flags & linux.MSG.EOR != 0,
                        .trunc = msg.flags & linux.MSG.TRUNC != 0,
                        .ctrunc = msg.flags & linux.MSG.CTRUNC != 0,
                        .oob = msg.flags & linux.MSG.OOB != 0,
                        .errqueue = msg.flags & linux.MSG.ERRQUEUE != 0,
                    },
                };
                message_i += 1;
                continue;
            },
            .AGAIN => unreachable,
            .INTR, .CANCELED => {},
            .BADF => |err| return .{ errnoBug(err), message_i },
            .NFILE => return .{ error.SystemFdQuotaExceeded, message_i },
            .MFILE => return .{ error.ProcessFdQuotaExceeded, message_i },
            .FAULT => |err| return .{ errnoBug(err), message_i },
            .INVAL => |err| return .{ errnoBug(err), message_i },
            .NOBUFS => return .{ error.SystemResources, message_i },
            .NOMEM => return .{ error.SystemResources, message_i },
            .NOTCONN => return .{ error.SocketUnconnected, message_i },
            .NOTSOCK => |err| return .{ errnoBug(err), message_i },
            .MSGSIZE => return .{ error.MessageOversize, message_i },
            .PIPE => return .{ error.SocketUnconnected, message_i },
            .OPNOTSUPP => |err| return .{ errnoBug(err), message_i },
            .CONNRESET => return .{ error.ConnectionResetByPeer, message_i },
            .NETDOWN => return .{ error.NetworkDown, message_i },
            else => |err| return .{ unexpectedErrno(err), message_i },
        }
    }
}