wasmer_wasix/syscalls/wasix/
sock_recv_from.rs

1use std::{mem::MaybeUninit, task::Waker};
2
3use super::*;
4use crate::{net::socket::TimeType, syscalls::*};
5
6/// ### `sock_recv_from()`
7/// Receive a message and its peer address from a socket.
8/// Note: This is similar to `recvfrom` in POSIX, though it also supports reading
9/// the data into multiple buffers in the manner of `readv`.
10///
11/// ## Parameters
12///
13/// * `ri_data` - List of scatter/gather vectors to which to store data.
14/// * `ri_flags` - Message flags.
15///
16/// ## Return
17///
18/// Number of bytes stored in ri_data and message flags.
19#[instrument(level = "trace", skip_all, fields(%sock, nread = field::Empty, peer = field::Empty), ret)]
20pub fn sock_recv_from<M: MemorySize>(
21    mut ctx: FunctionEnvMut<'_, WasiEnv>,
22    sock: WasiFd,
23    ri_data: WasmPtr<__wasi_iovec_t<M>, M>,
24    ri_data_len: M::Offset,
25    ri_flags: RiFlags,
26    ro_data_len: WasmPtr<M::Offset, M>,
27    ro_flags: WasmPtr<RoFlags, M>,
28    ro_addr: WasmPtr<__wasi_addr_port_t, M>,
29) -> Result<Errno, WasiError> {
30    WasiEnv::do_pending_operations(&mut ctx)?;
31
32    sock_recv_from_internal(
33        ctx,
34        sock,
35        ri_data,
36        ri_data_len,
37        ri_flags,
38        ro_data_len,
39        ro_flags,
40        ro_addr,
41    )
42}
43
44pub(super) fn sock_recv_from_internal<M: MemorySize>(
45    mut ctx: FunctionEnvMut<'_, WasiEnv>,
46    sock: WasiFd,
47    ri_data: WasmPtr<__wasi_iovec_t<M>, M>,
48    ri_data_len: M::Offset,
49    ri_flags: RiFlags,
50    ro_data_len: WasmPtr<M::Offset, M>,
51    ro_flags: WasmPtr<RoFlags, M>,
52    ro_addr: WasmPtr<__wasi_addr_port_t, M>,
53) -> Result<Errno, WasiError> {
54    let peek = (ri_flags & __WASI_SOCK_RECV_INPUT_PEEK) != 0;
55    let nonblocking_flag = (ri_flags & __WASI_SOCK_RECV_INPUT_DONT_WAIT) != 0;
56
57    let mut env = ctx.data();
58    // Check rights first to preserve error precedence
59    wasi_try_ok!(__sock_check_rights(env, sock, Rights::SOCK_RECV_FROM));
60
61    let memory = unsafe { env.memory_view(&ctx) };
62    let iovs_arr = wasi_try_mem_ok!(ri_data.slice(&memory, ri_data_len));
63
64    let max_size = wasi_try_ok!(checked_sock_recv_size::<M>(
65        iovs_arr.iter(),
66        env.capabilities.max_sock_recv_size
67    ));
68
69    let (bytes_read, peer) = {
70        if max_size <= 10240 {
71            let mut buf: [MaybeUninit<u8>; 10240] = unsafe { MaybeUninit::uninit().assume_init() };
72            let writer = &mut buf[..max_size];
73            let (amt, peer) = wasi_try_ok!(__sock_asyncify(
74                env,
75                sock,
76                Rights::SOCK_RECV_FROM,
77                |socket, fd| async move {
78                    let nonblocking =
79                        nonblocking_flag || fd.inner.flags.contains(Fdflags::NONBLOCK);
80                    let timeout = socket
81                        .opt_time(TimeType::ReadTimeout)
82                        .ok()
83                        .flatten()
84                        .unwrap_or(Duration::from_secs(30));
85                    socket
86                        .recv_from(
87                            env.tasks().deref(),
88                            writer,
89                            Some(timeout),
90                            nonblocking,
91                            peek,
92                        )
93                        .await
94                },
95            ));
96
97            if amt > 0 {
98                let buf: &[MaybeUninit<u8>] = &buf[..amt];
99                let buf: &[u8] = unsafe { std::mem::transmute(buf) };
100                wasi_try_ok!(copy_from_slice(buf, &memory, iovs_arr).map(|_| (amt, peer)))
101            } else {
102                (amt, peer)
103            }
104        } else {
105            let (data, peer) = wasi_try_ok!(__sock_asyncify(
106                env,
107                sock,
108                Rights::SOCK_RECV_FROM,
109                |socket, fd| async move {
110                    let nonblocking = fd.inner.flags.contains(Fdflags::NONBLOCK);
111                    let timeout = socket
112                        .opt_time(TimeType::ReadTimeout)
113                        .ok()
114                        .flatten()
115                        .unwrap_or(Duration::from_secs(30));
116
117                    let mut buf = Vec::with_capacity(max_size);
118                    unsafe {
119                        buf.set_len(max_size);
120                    }
121                    socket
122                        .recv_from(
123                            env.tasks().deref(),
124                            &mut buf,
125                            Some(timeout),
126                            nonblocking,
127                            peek,
128                        )
129                        .await
130                        .map(|(amt, addr)| {
131                            unsafe {
132                                buf.set_len(amt);
133                            }
134                            let buf: Vec<u8> = unsafe { std::mem::transmute(buf) };
135                            (buf, addr)
136                        })
137                }
138            ));
139
140            let data_len = data.len();
141            if data_len > 0 {
142                let mut reader = &data[..];
143                wasi_try_ok!(read_bytes(reader, &memory, iovs_arr).map(|_| (data_len, peer)))
144            } else {
145                (0, peer)
146            }
147        }
148    };
149    Span::current()
150        .record("nread", bytes_read)
151        .record("peer", format!("{peer:?}"));
152
153    wasi_try_ok!(write_ip_port(&memory, ro_addr, peer.ip(), peer.port()));
154
155    let bytes_read: M::Offset = wasi_try_ok!(bytes_read.try_into().map_err(|_| Errno::Overflow));
156    wasi_try_mem_ok!(ro_flags.write(&memory, 0));
157    wasi_try_mem_ok!(ro_data_len.write(&memory, bytes_read));
158
159    Ok(Errno::Success)
160}
wasmer_wasix/syscalls/wasix/sock_recv_from.rs

wasmer_wasix/syscalls/wasix/
sock_recv_from.rs
