wasmer_wasix/syscalls/wasix/
sock_send_to.rs

1use std::task::Waker;
2
3use super::*;
4use crate::{net::MAX_SOCKET_PAYLOAD, net::socket::TimeType, syscalls::*};
5
6/// ### `sock_send_to()`
7/// Send a message on a socket to a specific address.
8/// Note: This is similar to `sendto` in POSIX, though it also supports writing
9/// the data from multiple buffers in the manner of `writev`.
10///
11/// ## Parameters
12///
13/// * `si_data` - List of scatter/gather vectors to which to retrieve data
14/// * `si_flags` - Message flags.
15/// * `addr` - Address of the socket to send message to
16///
17/// ## Return
18///
19/// Number of bytes transmitted.
20#[instrument(level = "trace", skip_all, fields(%sock, ?addr, nsent = field::Empty), ret)]
21pub fn sock_send_to<M: MemorySize>(
22    mut ctx: FunctionEnvMut<'_, WasiEnv>,
23    sock: WasiFd,
24    si_data: WasmPtr<__wasi_ciovec_t<M>, M>,
25    si_data_len: M::Offset,
26    si_flags: SiFlags,
27    addr: WasmPtr<__wasi_addr_port_t, M>,
28    ret_data_len: WasmPtr<M::Offset, M>,
29) -> Result<Errno, WasiError> {
30    WasiEnv::do_pending_operations(&mut ctx)?;
31
32    let env = ctx.data();
33    let memory = unsafe { env.memory_view(&ctx) };
34    let iovs_arr = wasi_try_mem_ok!(si_data.slice(&memory, si_data_len));
35
36    let (addr_ip, addr_port) = {
37        let memory = unsafe { env.memory_view(&ctx) };
38        wasi_try_ok!(read_ip_port(&memory, addr))
39    };
40    let addr = SocketAddr::new(addr_ip, addr_port);
41    Span::current().record("addr", format!("{addr:?}"));
42
43    let bytes_written = wasi_try_ok!(sock_send_to_internal(
44        &mut ctx,
45        sock,
46        FdWriteSource::Iovs {
47            iovs: si_data,
48            iovs_len: si_data_len
49        },
50        si_flags,
51        addr,
52    )?);
53
54    #[cfg(feature = "journal")]
55    if ctx.data().enable_journal {
56        JournalEffector::save_sock_send_to::<M>(
57            &ctx,
58            sock,
59            bytes_written,
60            si_data,
61            si_data_len,
62            addr,
63            si_flags,
64        )
65        .map_err(|err| {
66            tracing::error!("failed to save sock_send_to event - {}", err);
67            WasiError::Exit(ExitCode::from(Errno::Fault))
68        })?;
69    }
70
71    Span::current().record("nsent", bytes_written);
72
73    let env = ctx.data();
74    let memory = unsafe { env.memory_view(&ctx) };
75    let bytes_written: M::Offset =
76        wasi_try_ok!(bytes_written.try_into().map_err(|_| Errno::Overflow));
77    wasi_try_mem_ok!(ret_data_len.write(&memory, bytes_written));
78
79    Ok(Errno::Success)
80}
81
82pub(crate) fn sock_send_to_internal<M: MemorySize>(
83    ctx: &mut FunctionEnvMut<'_, WasiEnv>,
84    sock: WasiFd,
85    si_data: FdWriteSource<'_, M>,
86    si_flags: SiFlags,
87    addr: SocketAddr,
88) -> Result<Result<usize, Errno>, WasiError> {
89    let env = ctx.data();
90    let net = env.net().clone();
91    let tasks = ctx.data().tasks().clone();
92
93    // Auto-bind UDP
94    wasi_try_ok_ok!(__sock_upgrade(
95        ctx,
96        sock,
97        Rights::SOCK_SEND_TO,
98        move |mut socket, flags| async move { socket.auto_bind_udp(tasks.deref(), net.deref()).await }
99    ));
100
101    let env = ctx.data();
102    let memory = unsafe { env.memory_view(&ctx) };
103
104    let nonblocking_flag = (si_flags & __WASI_SOCK_SEND_INPUT_DONT_WAIT) != 0;
105
106    let bytes_written = {
107        wasi_try_ok_ok!(__sock_asyncify(
108            env,
109            sock,
110            Rights::SOCK_SEND_TO,
111            |socket, fd| async move {
112                let nonblocking = nonblocking_flag || fd.inner.flags.contains(Fdflags::NONBLOCK);
113                let timeout = socket
114                    .opt_time(TimeType::WriteTimeout)
115                    .ok()
116                    .flatten()
117                    .unwrap_or(Duration::from_secs(30));
118
119                if socket.is_dgram() {
120                    let data = si_data.coalesce(&memory, MAX_SOCKET_PAYLOAD)?;
121                    return socket
122                        .send_to::<M>(
123                            env.tasks().deref(),
124                            data.as_ref(),
125                            addr,
126                            Some(timeout),
127                            nonblocking,
128                        )
129                        .await;
130                }
131
132                match si_data {
133                    FdWriteSource::Iovs { iovs, iovs_len } => {
134                        let iovs_arr = iovs.slice(&memory, iovs_len).map_err(mem_error_to_wasi)?;
135                        let iovs_arr = iovs_arr.access().map_err(mem_error_to_wasi)?;
136
137                        let mut sent = 0usize;
138                        for iovs in iovs_arr.iter() {
139                            let buf = WasmPtr::<u8, M>::new(iovs.buf)
140                                .slice(&memory, iovs.buf_len)
141                                .map_err(mem_error_to_wasi)?
142                                .access()
143                                .map_err(mem_error_to_wasi)?;
144                            let local_sent = match socket
145                                .send_to::<M>(
146                                    env.tasks().deref(),
147                                    buf.as_ref(),
148                                    addr,
149                                    Some(timeout),
150                                    nonblocking,
151                                )
152                                .await
153                            {
154                                Ok(s) => s,
155                                Err(_) if sent > 0 => break,
156                                Err(err) => return Err(err),
157                            };
158                            sent += local_sent;
159                            if local_sent != buf.len() {
160                                break;
161                            }
162                        }
163                        Ok(sent)
164                    }
165                    FdWriteSource::Buffer(data) => {
166                        socket
167                            .send_to::<M>(
168                                env.tasks().deref(),
169                                data.as_ref(),
170                                addr,
171                                Some(timeout),
172                                nonblocking,
173                            )
174                            .await
175                    }
176                }
177            },
178        ))
179    };
180    trace!(
181        %bytes_written,
182    );
183
184    Ok(Ok(bytes_written))
185}
wasmer_wasix/syscalls/wasix/sock_send_to.rs

wasmer_wasix/syscalls/wasix/
sock_send_to.rs
