wasmer_wasix/syscalls/wasix/
sock_send.rs

1use std::{mem::MaybeUninit, task::Waker};
2
3use super::*;
4use crate::{net::MAX_SOCKET_PAYLOAD, net::socket::TimeType, syscalls::*};
5
6/// ### `sock_send()`
7/// Send a message on a socket.
8/// Note: This is similar to `send` 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///
16/// ## Return
17///
18/// Number of bytes transmitted.
19#[instrument(level = "trace", skip_all, fields(%fd, nsent = field::Empty), ret)]
20pub fn sock_send<M: MemorySize>(
21    mut ctx: FunctionEnvMut<'_, WasiEnv>,
22    fd: WasiFd,
23    si_data: WasmPtr<__wasi_ciovec_t<M>, M>,
24    si_data_len: M::Offset,
25    si_flags: SiFlags,
26    ret_data_len: WasmPtr<M::Offset, M>,
27) -> Result<Errno, WasiError> {
28    WasiEnv::do_pending_operations(&mut ctx)?;
29
30    let env = ctx.data();
31    let fd_entry = wasi_try_ok!(env.state.fs.get_fd(fd));
32    let enable_journal = env.enable_journal;
33    let guard = fd_entry.inode.read();
34    // Some guests route socket-like wakeups through pipe-backed fds.
35    let use_write = matches!(guard.deref(), Kind::DuplexPipe { .. } | Kind::PipeTx { .. });
36    drop(guard);
37
38    let bytes_written = if use_write {
39        let offset = { fd_entry.inner.offset.load(Ordering::Acquire) as usize };
40
41        wasi_try_ok!(fd_write_internal::<M>(
42            &mut ctx,
43            fd,
44            fd_entry,
45            FdWriteSource::Iovs {
46                iovs: si_data,
47                iovs_len: si_data_len
48            },
49            offset as u64,
50            true,
51            enable_journal
52        )?)
53    } else {
54        wasi_try_ok!(sock_send_internal::<M>(
55            &ctx,
56            fd,
57            FdWriteSource::Iovs {
58                iovs: si_data,
59                iovs_len: si_data_len
60            },
61            si_flags,
62        )?)
63    };
64
65    #[cfg(feature = "journal")]
66    if ctx.data().enable_journal {
67        JournalEffector::save_sock_send(&ctx, fd, bytes_written, si_data, si_data_len, si_flags)
68            .map_err(|err| {
69                tracing::error!("failed to save sock_send event - {}", err);
70                WasiError::Exit(ExitCode::from(Errno::Fault))
71            })?;
72    }
73
74    Span::current().record("nsent", bytes_written);
75
76    let env = ctx.data();
77    let memory = unsafe { env.memory_view(&ctx) };
78    let bytes_written: M::Offset =
79        wasi_try_ok!(bytes_written.try_into().map_err(|_| Errno::Overflow));
80    wasi_try_mem_ok!(ret_data_len.write(&memory, bytes_written));
81
82    Ok(Errno::Success)
83}
84
85pub(crate) fn sock_send_internal<M: MemorySize>(
86    ctx: &FunctionEnvMut<'_, WasiEnv>,
87    sock: WasiFd,
88    si_data: FdWriteSource<'_, M>,
89    si_flags: SiFlags,
90) -> Result<Result<usize, Errno>, WasiError> {
91    let env = ctx.data();
92    let memory = unsafe { env.memory_view(&ctx) };
93    let runtime = env.runtime.clone();
94
95    let nonblocking_flag = (si_flags & __WASI_SOCK_SEND_INPUT_DONT_WAIT) != 0;
96
97    let bytes_written = wasi_try_ok_ok!(__sock_asyncify(
98        env,
99        sock,
100        Rights::SOCK_SEND,
101        |socket, fd| async move {
102            let nonblocking = nonblocking_flag || fd.inner.flags.contains(Fdflags::NONBLOCK);
103            let timeout = socket
104                .opt_time(TimeType::WriteTimeout)
105                .ok()
106                .flatten()
107                .unwrap_or(Duration::from_secs(30));
108
109            if socket.is_dgram() {
110                let data = si_data.coalesce(&memory, MAX_SOCKET_PAYLOAD)?;
111                return socket
112                    .send(
113                        env.tasks().deref(),
114                        data.as_ref(),
115                        Some(timeout),
116                        nonblocking,
117                    )
118                    .await;
119            }
120
121            match si_data {
122                FdWriteSource::Iovs { iovs, iovs_len } => {
123                    let iovs_arr = iovs.slice(&memory, iovs_len).map_err(mem_error_to_wasi)?;
124                    let iovs_arr = iovs_arr.access().map_err(mem_error_to_wasi)?;
125
126                    let mut sent = 0usize;
127                    for iovs in iovs_arr.iter() {
128                        let buf = WasmPtr::<u8, M>::new(iovs.buf)
129                            .slice(&memory, iovs.buf_len)
130                            .map_err(mem_error_to_wasi)?
131                            .access()
132                            .map_err(mem_error_to_wasi)?;
133                        let local_sent = match socket
134                            .send(
135                                env.tasks().deref(),
136                                buf.as_ref(),
137                                Some(timeout),
138                                nonblocking,
139                            )
140                            .await
141                        {
142                            Ok(s) => s,
143                            Err(_) if sent > 0 => break,
144                            Err(err) => return Err(err),
145                        };
146                        sent += local_sent;
147                        if local_sent != buf.len() {
148                            break;
149                        }
150                    }
151                    Ok(sent)
152                }
153                FdWriteSource::Buffer(data) => {
154                    socket
155                        .send(
156                            env.tasks().deref(),
157                            data.as_ref(),
158                            Some(timeout),
159                            nonblocking,
160                        )
161                        .await
162                }
163            }
164        }
165    ));
166    trace!(
167        %bytes_written,
168    );
169
170    Ok(Ok(bytes_written))
171}
wasmer_wasix/syscalls/wasix/sock_send.rs

wasmer_wasix/syscalls/wasix/
sock_send.rs
