wasmer_wasix/net/
socket.rs

1use std::{
2    future::Future,
3    io,
4    mem::MaybeUninit,
5    net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr},
6    pin::Pin,
7    sync::{Arc, RwLock, RwLockWriteGuard},
8    task::{Context, Poll},
9    time::Duration,
10};
11
12#[cfg(feature = "enable-serde")]
13use serde_derive::{Deserialize, Serialize};
14use virtual_mio::InterestHandler;
15use virtual_net::{
16    NetworkError, VirtualIcmpSocket, VirtualNetworking, VirtualRawSocket, VirtualTcpListener,
17    VirtualTcpSocket, VirtualUdpSocket, net_error_into_io_err,
18};
19use wasmer_types::MemorySize;
20use wasmer_wasix_types::wasi::{Addressfamily, Errno, Rights, SockProto, Sockoption, Socktype};
21
22use crate::{VirtualTaskManager, net::net_error_into_wasi_err};
23
24#[derive(Debug)]
25#[cfg_attr(feature = "enable-serde", derive(Serialize, Deserialize))]
26pub enum InodeHttpSocketType {
27    /// Used to feed the bytes into the request itself
28    Request,
29    /// Used to receive the bytes from the HTTP server
30    Response,
31    /// Used to read the headers from the HTTP server
32    Headers,
33}
34
35#[derive(Debug)]
36pub struct SocketProperties {
37    pub family: Addressfamily,
38    pub ty: Socktype,
39    pub pt: SockProto,
40    pub only_v6: bool,
41    pub reuse_port: bool,
42    pub reuse_addr: bool,
43    pub no_delay: Option<bool>,
44    pub keep_alive: Option<bool>,
45    pub dont_route: Option<bool>,
46    pub send_buf_size: Option<usize>,
47    pub recv_buf_size: Option<usize>,
48    pub write_timeout: Option<Duration>,
49    pub read_timeout: Option<Duration>,
50    pub accept_timeout: Option<Duration>,
51    pub connect_timeout: Option<Duration>,
52    pub handler: Option<Box<dyn InterestHandler + Send + Sync>>,
53}
54
55#[derive(Debug)]
56//#[cfg_attr(feature = "enable-serde", derive(Serialize, Deserialize))]
57pub enum InodeSocketKind {
58    PreSocket {
59        props: SocketProperties,
60        addr: Option<SocketAddr>,
61    },
62    Icmp(Box<dyn VirtualIcmpSocket + Sync>),
63    Raw(Box<dyn VirtualRawSocket + Sync>),
64    TcpListener {
65        socket: Box<dyn VirtualTcpListener + Sync>,
66        accept_timeout: Option<Duration>,
67    },
68    TcpStream {
69        socket: Box<dyn VirtualTcpSocket + Sync>,
70        write_timeout: Option<Duration>,
71        read_timeout: Option<Duration>,
72    },
73    UdpSocket {
74        socket: Box<dyn VirtualUdpSocket + Sync>,
75        peer: Option<SocketAddr>,
76    },
77    RemoteSocket {
78        props: SocketProperties,
79        local_addr: SocketAddr,
80        peer_addr: SocketAddr,
81        ttl: u32,
82        multicast_ttl: u32,
83        is_dead: bool,
84    },
85}
86
87pub enum WasiSocketOption {
88    Noop,
89    ReusePort,
90    ReuseAddr,
91    NoDelay,
92    DontRoute,
93    OnlyV6,
94    Broadcast,
95    MulticastLoopV4,
96    MulticastLoopV6,
97    Promiscuous,
98    Listening,
99    LastError,
100    KeepAlive,
101    Linger,
102    OobInline,
103    RecvBufSize,
104    SendBufSize,
105    RecvLowat,
106    SendLowat,
107    RecvTimeout,
108    SendTimeout,
109    ConnectTimeout,
110    AcceptTimeout,
111    Ttl,
112    MulticastTtlV4,
113    Type,
114    Proto,
115}
116
117impl TryFrom<Sockoption> for WasiSocketOption {
118    type Error = Errno;
119
120    fn try_from(opt: Sockoption) -> Result<Self, Self::Error> {
121        use WasiSocketOption::*;
122        match opt {
123            Sockoption::Noop => Ok(Noop),
124            Sockoption::ReusePort => Ok(ReusePort),
125            Sockoption::ReuseAddr => Ok(ReuseAddr),
126            Sockoption::NoDelay => Ok(NoDelay),
127            Sockoption::DontRoute => Ok(DontRoute),
128            Sockoption::OnlyV6 => Ok(OnlyV6),
129            Sockoption::Broadcast => Ok(Broadcast),
130            Sockoption::MulticastLoopV4 => Ok(MulticastLoopV4),
131            Sockoption::MulticastLoopV6 => Ok(MulticastLoopV6),
132            Sockoption::Promiscuous => Ok(Promiscuous),
133            Sockoption::Listening => Ok(Listening),
134            Sockoption::LastError => Ok(LastError),
135            Sockoption::KeepAlive => Ok(KeepAlive),
136            Sockoption::Linger => Ok(Linger),
137            Sockoption::OobInline => Ok(OobInline),
138            Sockoption::RecvBufSize => Ok(RecvBufSize),
139            Sockoption::SendBufSize => Ok(SendBufSize),
140            Sockoption::RecvLowat => Ok(RecvLowat),
141            Sockoption::SendLowat => Ok(SendLowat),
142            Sockoption::RecvTimeout => Ok(RecvTimeout),
143            Sockoption::SendTimeout => Ok(SendTimeout),
144            Sockoption::ConnectTimeout => Ok(ConnectTimeout),
145            Sockoption::AcceptTimeout => Ok(AcceptTimeout),
146            Sockoption::Ttl => Ok(Ttl),
147            Sockoption::MulticastTtlV4 => Ok(MulticastTtlV4),
148            Sockoption::Type => Ok(Type),
149            Sockoption::Proto => Ok(Proto),
150            _ => Err(Errno::Inval),
151        }
152    }
153}
154
155#[derive(Debug)]
156pub enum WasiSocketStatus {
157    Opening,
158    Opened,
159    Closed,
160    Failed,
161}
162
163#[derive(Debug, Copy, Clone, PartialEq, Eq)]
164#[cfg_attr(feature = "enable-serde", derive(Serialize, Deserialize))]
165pub enum TimeType {
166    ReadTimeout,
167    WriteTimeout,
168    AcceptTimeout,
169    ConnectTimeout,
170    BindTimeout,
171    Linger,
172}
173
174impl From<TimeType> for wasmer_journal::SocketOptTimeType {
175    fn from(value: TimeType) -> Self {
176        match value {
177            TimeType::ReadTimeout => Self::ReadTimeout,
178            TimeType::WriteTimeout => Self::WriteTimeout,
179            TimeType::AcceptTimeout => Self::AcceptTimeout,
180            TimeType::ConnectTimeout => Self::ConnectTimeout,
181            TimeType::BindTimeout => Self::BindTimeout,
182            TimeType::Linger => Self::Linger,
183        }
184    }
185}
186
187impl From<wasmer_journal::SocketOptTimeType> for TimeType {
188    fn from(value: wasmer_journal::SocketOptTimeType) -> Self {
189        use wasmer_journal::SocketOptTimeType;
190        match value {
191            SocketOptTimeType::ReadTimeout => TimeType::ReadTimeout,
192            SocketOptTimeType::WriteTimeout => TimeType::WriteTimeout,
193            SocketOptTimeType::AcceptTimeout => TimeType::AcceptTimeout,
194            SocketOptTimeType::ConnectTimeout => TimeType::ConnectTimeout,
195            SocketOptTimeType::BindTimeout => TimeType::BindTimeout,
196            SocketOptTimeType::Linger => TimeType::Linger,
197        }
198    }
199}
200
201#[derive(Debug)]
202//#[cfg_attr(feature = "enable-serde", derive(Serialize, Deserialize))]
203pub(crate) struct InodeSocketProtected {
204    pub kind: InodeSocketKind,
205}
206
207#[derive(Debug)]
208//#[cfg_attr(feature = "enable-serde", derive(Serialize, Deserialize))]
209pub(crate) struct InodeSocketInner {
210    pub protected: RwLock<InodeSocketProtected>,
211}
212
213#[derive(Debug, Clone)]
214//#[cfg_attr(feature = "enable-serde", derive(Serialize, Deserialize))]
215pub struct InodeSocket {
216    pub(crate) inner: Arc<InodeSocketInner>,
217}
218
219impl InodeSocket {
220    pub fn new(kind: InodeSocketKind) -> Self {
221        let protected = InodeSocketProtected { kind };
222        Self {
223            inner: Arc::new(InodeSocketInner {
224                protected: RwLock::new(protected),
225            }),
226        }
227    }
228
229    pub fn poll_read_ready(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<usize>> {
230        let mut inner = self.inner.protected.write().unwrap();
231        inner.poll_read_ready(cx)
232    }
233
234    pub fn poll_write_ready(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<usize>> {
235        let mut inner = self.inner.protected.write().unwrap();
236        inner.poll_read_ready(cx)
237    }
238
239    // When a sendto or connect call comes in for a UDP "pre-socket", it must be bound to
240    // an ephemeral port automatically.
241    // Apparently, clippy fails to recognize the write-locked guard being passed into
242    // the other function, hence the `allow` attribute.
243    #[allow(clippy::await_holding_lock, clippy::readonly_write_lock)]
244    pub async fn auto_bind_udp(
245        &self,
246        tasks: &dyn VirtualTaskManager,
247        net: &dyn VirtualNetworking,
248    ) -> Result<Option<InodeSocket>, Errno> {
249        let timeout = self
250            .opt_time(TimeType::BindTimeout)
251            .ok()
252            .flatten()
253            .unwrap_or(Duration::from_secs(30));
254        let inner = self.inner.protected.write().unwrap();
255        match &inner.kind {
256            InodeSocketKind::PreSocket { props, .. } if props.ty == Socktype::Dgram => {
257                let addr = match props.family {
258                    Addressfamily::Inet4 => SocketAddr::new(IpAddr::V4(Ipv4Addr::UNSPECIFIED), 0),
259                    Addressfamily::Inet6 => SocketAddr::new(IpAddr::V6(Ipv6Addr::UNSPECIFIED), 0),
260                    _ => return Err(Errno::Notsup),
261                };
262                Self::bind_internal(tasks, net, addr, timeout, inner).await
263            }
264            _ => Ok(None),
265        }
266    }
267
268    pub async fn bind(
269        &self,
270        tasks: &dyn VirtualTaskManager,
271        net: &dyn VirtualNetworking,
272        set_addr: SocketAddr,
273    ) -> Result<Option<InodeSocket>, Errno> {
274        let timeout = self
275            .opt_time(TimeType::BindTimeout)
276            .ok()
277            .flatten()
278            .unwrap_or(Duration::from_secs(30));
279        let inner = self.inner.protected.write().unwrap();
280        Self::bind_internal(tasks, net, set_addr, timeout, inner).await
281    }
282
283    // The lock is dropped before awaiting, but clippy doesn't realize it
284    #[allow(clippy::await_holding_lock)]
285    async fn bind_internal(
286        tasks: &dyn VirtualTaskManager,
287        net: &dyn VirtualNetworking,
288        set_addr: SocketAddr,
289        timeout: Duration,
290        mut inner: RwLockWriteGuard<'_, InodeSocketProtected>,
291    ) -> Result<Option<InodeSocket>, Errno> {
292        let socket = {
293            match &mut inner.kind {
294                InodeSocketKind::PreSocket { props, addr, .. } => {
295                    match props.family {
296                        Addressfamily::Inet4 => {
297                            if !set_addr.is_ipv4() {
298                                tracing::debug!(
299                                    "IP address is the wrong type IPv4 ({set_addr}) vs IPv6 family"
300                                );
301                                return Err(Errno::Inval);
302                            }
303                        }
304                        Addressfamily::Inet6 => {
305                            if !set_addr.is_ipv6() {
306                                tracing::debug!(
307                                    "IP address is the wrong type IPv6 ({set_addr}) vs IPv4 family"
308                                );
309                                return Err(Errno::Inval);
310                            }
311                        }
312                        _ => {
313                            return Err(Errno::Notsup);
314                        }
315                    }
316
317                    addr.replace(set_addr);
318                    let addr = (*addr).unwrap();
319
320                    match props.ty {
321                        Socktype::Stream => {
322                            // we already set the socket address - next we need a listen or connect so nothing
323                            // more to do at this time
324                            return Ok(None);
325                        }
326                        Socktype::Dgram => {
327                            let reuse_port = props.reuse_port;
328                            let reuse_addr = props.reuse_addr;
329
330                            net.bind_udp(addr, reuse_port, reuse_addr)
331                        }
332                        _ => return Err(Errno::Inval),
333                    }
334                }
335                InodeSocketKind::RemoteSocket {
336                    props,
337                    local_addr: addr,
338                    ..
339                } => {
340                    match props.family {
341                        Addressfamily::Inet4 => {
342                            if !set_addr.is_ipv4() {
343                                tracing::debug!(
344                                    "IP address is the wrong type IPv4 ({set_addr}) vs IPv6 family"
345                                );
346                                return Err(Errno::Inval);
347                            }
348                        }
349                        Addressfamily::Inet6 => {
350                            if !set_addr.is_ipv6() {
351                                tracing::debug!(
352                                    "IP address is the wrong type IPv6 ({set_addr}) vs IPv4 family"
353                                );
354                                return Err(Errno::Inval);
355                            }
356                        }
357                        _ => {
358                            return Err(Errno::Notsup);
359                        }
360                    }
361
362                    *addr = set_addr;
363                    let addr = *addr;
364
365                    match props.ty {
366                        Socktype::Stream => {
367                            // we already set the socket address - next we need a listen or connect so nothing
368                            // more to do at this time
369                            return Ok(None);
370                        }
371                        Socktype::Dgram => {
372                            let reuse_port = props.reuse_port;
373                            let reuse_addr = props.reuse_addr;
374
375                            net.bind_udp(addr, reuse_port, reuse_addr)
376                        }
377                        _ => return Err(Errno::Inval),
378                    }
379                }
380                _ => return Err(Errno::Notsup),
381            }
382        };
383
384        drop(inner);
385
386        tokio::select! {
387            socket = socket => {
388                let socket = socket.map_err(net_error_into_wasi_err)?;
389                Ok(Some(InodeSocket::new(InodeSocketKind::UdpSocket { socket, peer: None })))
390            },
391            _ = tasks.sleep_now(timeout) => Err(Errno::Timedout)
392        }
393    }
394
395    pub async fn listen(
396        &self,
397        tasks: &dyn VirtualTaskManager,
398        net: &dyn VirtualNetworking,
399        _backlog: usize,
400    ) -> Result<Option<InodeSocket>, Errno> {
401        let timeout = self
402            .opt_time(TimeType::AcceptTimeout)
403            .ok()
404            .flatten()
405            .unwrap_or(Duration::from_secs(30));
406
407        let socket = {
408            let inner = self.inner.protected.read().unwrap();
409            match &inner.kind {
410                InodeSocketKind::PreSocket { props, addr, .. } => match props.ty {
411                    Socktype::Stream => {
412                        if addr.is_none() {
413                            tracing::warn!("wasi[?]::sock_listen - failed - address not set");
414                            return Err(Errno::Inval);
415                        }
416                        let addr = *addr.as_ref().unwrap();
417                        let only_v6 = props.only_v6;
418                        let reuse_port = props.reuse_port;
419                        let reuse_addr = props.reuse_addr;
420                        drop(inner);
421
422                        net.listen_tcp(addr, only_v6, reuse_port, reuse_addr)
423                    }
424                    ty => {
425                        tracing::warn!(
426                            "wasi[?]::sock_listen - failed - not supported(pre-socket:{:?})",
427                            ty
428                        );
429                        return Err(Errno::Notsup);
430                    }
431                },
432                InodeSocketKind::RemoteSocket {
433                    props,
434                    local_addr: addr,
435                    ..
436                } => match props.ty {
437                    Socktype::Stream => {
438                        let addr = *addr;
439                        let only_v6 = props.only_v6;
440                        let reuse_port = props.reuse_port;
441                        let reuse_addr = props.reuse_addr;
442                        drop(inner);
443
444                        net.listen_tcp(addr, only_v6, reuse_port, reuse_addr)
445                    }
446                    ty => {
447                        tracing::warn!(
448                            "wasi[?]::sock_listen - failed - not supported(remote-socket:{:?})",
449                            ty
450                        );
451                        return Err(Errno::Notsup);
452                    }
453                },
454                InodeSocketKind::Icmp(_) => {
455                    tracing::warn!("wasi[?]::sock_listen - failed - not supported(icmp)");
456                    return Err(Errno::Notsup);
457                }
458                InodeSocketKind::Raw(_) => {
459                    tracing::warn!("wasi[?]::sock_listen - failed - not supported(raw)");
460                    return Err(Errno::Notsup);
461                }
462                InodeSocketKind::TcpListener { .. } => {
463                    tracing::warn!(
464                        "wasi[?]::sock_listen - failed - already listening (tcp-listener)"
465                    );
466                    return Err(Errno::Notsup);
467                }
468                InodeSocketKind::TcpStream { .. } => {
469                    tracing::warn!("wasi[?]::sock_listen - failed - not supported(tcp-stream)");
470                    return Err(Errno::Notsup);
471                }
472                InodeSocketKind::UdpSocket { .. } => {
473                    tracing::warn!("wasi[?]::sock_listen - failed - not supported(udp-socket)");
474                    return Err(Errno::Notsup);
475                }
476            }
477        };
478
479        tokio::select! {
480            socket = socket => {
481                let socket = socket.map_err(net_error_into_wasi_err)?;
482                Ok(Some(InodeSocket::new(InodeSocketKind::TcpListener {
483                    socket,
484                    accept_timeout: Some(timeout),
485                })))
486            },
487            _ = tasks.sleep_now(timeout) => Err(Errno::Timedout)
488        }
489    }
490
491    pub async fn accept(
492        &self,
493        tasks: &dyn VirtualTaskManager,
494        nonblocking: bool,
495        timeout: Option<Duration>,
496    ) -> Result<(Box<dyn VirtualTcpSocket + Sync>, SocketAddr), Errno> {
497        struct SocketAccepter<'a> {
498            sock: &'a InodeSocket,
499            nonblocking: bool,
500            handler_registered: bool,
501        }
502        impl Drop for SocketAccepter<'_> {
503            fn drop(&mut self) {
504                if self.handler_registered {
505                    let mut inner = self.sock.inner.protected.write().unwrap();
506                    inner.remove_handler();
507                }
508            }
509        }
510        impl Future for SocketAccepter<'_> {
511            type Output = Result<(Box<dyn VirtualTcpSocket + Sync>, SocketAddr), Errno>;
512            fn poll(
513                mut self: Pin<&mut Self>,
514                cx: &mut std::task::Context<'_>,
515            ) -> std::task::Poll<Self::Output> {
516                loop {
517                    let mut inner = self.sock.inner.protected.write().unwrap();
518                    return match &mut inner.kind {
519                        InodeSocketKind::TcpListener { socket, .. } => match socket.try_accept() {
520                            Ok((child, addr)) => Poll::Ready(Ok((child, addr))),
521                            Err(NetworkError::WouldBlock) if self.nonblocking => {
522                                Poll::Ready(Err(Errno::Again))
523                            }
524                            Err(NetworkError::WouldBlock) if !self.handler_registered => {
525                                let res = socket.set_handler(cx.waker().into());
526                                if let Err(err) = res {
527                                    return Poll::Ready(Err(net_error_into_wasi_err(err)));
528                                }
529                                drop(inner);
530                                self.handler_registered = true;
531                                continue;
532                            }
533                            Err(NetworkError::WouldBlock) => Poll::Pending,
534                            Err(err) => Poll::Ready(Err(net_error_into_wasi_err(err))),
535                        },
536                        InodeSocketKind::PreSocket { .. } => Poll::Ready(Err(Errno::Notconn)),
537                        _ => Poll::Ready(Err(Errno::Notsup)),
538                    };
539                }
540            }
541        }
542
543        let acceptor = SocketAccepter {
544            sock: self,
545            nonblocking,
546            handler_registered: false,
547        };
548        if let Some(timeout) = timeout {
549            tokio::select! {
550                res = acceptor => res,
551                _ = tasks.sleep_now(timeout) => Err(Errno::Timedout)
552            }
553        } else {
554            acceptor.await
555        }
556    }
557
558    pub fn close(&self) -> Result<(), Errno> {
559        let mut inner = self.inner.protected.write().unwrap();
560        match &mut inner.kind {
561            InodeSocketKind::TcpListener { .. } => {}
562            InodeSocketKind::TcpStream { socket, .. } => {
563                socket.close().map_err(net_error_into_wasi_err)?;
564            }
565            InodeSocketKind::Icmp(_) => {}
566            InodeSocketKind::UdpSocket { .. } => {}
567            InodeSocketKind::Raw(_) => {}
568            InodeSocketKind::PreSocket { .. } => return Err(Errno::Notconn),
569            InodeSocketKind::RemoteSocket { .. } => {}
570        };
571        Ok(())
572    }
573
574    pub async fn connect(
575        &mut self,
576        tasks: &dyn VirtualTaskManager,
577        net: &dyn VirtualNetworking,
578        peer: SocketAddr,
579        timeout: Option<std::time::Duration>,
580        nonblocking: bool,
581    ) -> Result<Option<InodeSocket>, Errno> {
582        let new_write_timeout;
583        let new_read_timeout;
584
585        let timeout = timeout.unwrap_or(Duration::from_secs(30));
586
587        let handler;
588        let connect = {
589            let mut inner = self.inner.protected.write().unwrap();
590            match &mut inner.kind {
591                InodeSocketKind::PreSocket { props, addr, .. } => {
592                    handler = props.handler.take();
593                    new_write_timeout = props.write_timeout;
594                    new_read_timeout = props.read_timeout;
595                    match props.ty {
596                        Socktype::Stream => {
597                            let no_delay = props.no_delay;
598                            let keep_alive = props.keep_alive;
599                            let dont_route = props.dont_route;
600                            let addr = match addr {
601                                Some(a) => *a,
602                                None => {
603                                    let ip = match peer.is_ipv4() {
604                                        true => IpAddr::V4(Ipv4Addr::UNSPECIFIED),
605                                        false => IpAddr::V6(Ipv6Addr::UNSPECIFIED),
606                                    };
607                                    SocketAddr::new(ip, 0)
608                                }
609                            };
610                            Box::pin(async move {
611                                let mut ret = net.connect_tcp(addr, peer).await?;
612                                if let Some(no_delay) = no_delay {
613                                    ret.set_nodelay(no_delay).ok();
614                                }
615                                if let Some(keep_alive) = keep_alive {
616                                    ret.set_keepalive(keep_alive).ok();
617                                }
618                                if let Some(dont_route) = dont_route {
619                                    ret.set_dontroute(dont_route).ok();
620                                }
621                                if !nonblocking {
622                                    futures::future::poll_fn(|cx| ret.poll_write_ready(cx)).await?;
623                                }
624                                Ok(ret)
625                            })
626                        }
627                        Socktype::Dgram => return Err(Errno::Inval),
628                        _ => return Err(Errno::Notsup),
629                    }
630                }
631                InodeSocketKind::UdpSocket {
632                    peer: target_peer, ..
633                } => {
634                    target_peer.replace(peer);
635                    return Ok(None);
636                }
637                InodeSocketKind::RemoteSocket { peer_addr, .. } => {
638                    *peer_addr = peer;
639                    return Ok(None);
640                }
641                _ => return Err(Errno::Notsup),
642            }
643        };
644
645        let mut socket = tokio::select! {
646            res = connect => res.map_err(net_error_into_wasi_err)?,
647            _ = tasks.sleep_now(timeout) => return Err(Errno::Timedout)
648        };
649
650        if let Some(handler) = handler {
651            socket
652                .set_handler(handler)
653                .map_err(net_error_into_wasi_err)?;
654        }
655
656        let socket = InodeSocket::new(InodeSocketKind::TcpStream {
657            socket,
658            write_timeout: new_write_timeout,
659            read_timeout: new_read_timeout,
660        });
661
662        Ok(Some(socket))
663    }
664
665    pub fn status(&self) -> Result<WasiSocketStatus, Errno> {
666        let inner = self.inner.protected.read().unwrap();
667        Ok(match &inner.kind {
668            InodeSocketKind::PreSocket { .. } => WasiSocketStatus::Opening,
669            InodeSocketKind::TcpListener { .. } => WasiSocketStatus::Opened,
670            InodeSocketKind::TcpStream { .. } => WasiSocketStatus::Opened,
671            InodeSocketKind::UdpSocket { .. } => WasiSocketStatus::Opened,
672            InodeSocketKind::RemoteSocket { is_dead, .. } => match is_dead {
673                true => WasiSocketStatus::Closed,
674                false => WasiSocketStatus::Opened,
675            },
676            _ => WasiSocketStatus::Failed,
677        })
678    }
679
680    pub fn addr_local(&self) -> Result<SocketAddr, Errno> {
681        let inner = self.inner.protected.read().unwrap();
682        Ok(match &inner.kind {
683            InodeSocketKind::PreSocket { props, addr, .. } => {
684                if let Some(addr) = addr {
685                    *addr
686                } else {
687                    SocketAddr::new(
688                        match props.family {
689                            Addressfamily::Inet4 => IpAddr::V4(Ipv4Addr::UNSPECIFIED),
690                            Addressfamily::Inet6 => IpAddr::V6(Ipv6Addr::UNSPECIFIED),
691                            _ => return Err(Errno::Inval),
692                        },
693                        0,
694                    )
695                }
696            }
697            InodeSocketKind::Icmp(sock) => sock.addr_local().map_err(net_error_into_wasi_err)?,
698            InodeSocketKind::TcpListener { socket, .. } => {
699                socket.addr_local().map_err(net_error_into_wasi_err)?
700            }
701            InodeSocketKind::TcpStream { socket, .. } => {
702                socket.addr_local().map_err(net_error_into_wasi_err)?
703            }
704            InodeSocketKind::UdpSocket { socket, .. } => {
705                socket.addr_local().map_err(net_error_into_wasi_err)?
706            }
707            InodeSocketKind::RemoteSocket {
708                local_addr: addr, ..
709            } => *addr,
710            _ => return Err(Errno::Notsup),
711        })
712    }
713
714    pub fn addr_peer(&self) -> Result<SocketAddr, Errno> {
715        let inner = self.inner.protected.read().unwrap();
716        Ok(match &inner.kind {
717            InodeSocketKind::PreSocket { props, .. } => SocketAddr::new(
718                match props.family {
719                    Addressfamily::Inet4 => IpAddr::V4(Ipv4Addr::UNSPECIFIED),
720                    Addressfamily::Inet6 => IpAddr::V6(Ipv6Addr::UNSPECIFIED),
721                    _ => return Err(Errno::Inval),
722                },
723                0,
724            ),
725            InodeSocketKind::TcpStream { socket, .. } => {
726                socket.addr_peer().map_err(net_error_into_wasi_err)?
727            }
728            InodeSocketKind::UdpSocket { socket, .. } => socket
729                .addr_peer()
730                .map_err(net_error_into_wasi_err)?
731                .map(Ok)
732                .unwrap_or_else(|| {
733                    socket
734                        .addr_local()
735                        .map_err(net_error_into_wasi_err)
736                        .map(|addr| {
737                            SocketAddr::new(
738                                match addr {
739                                    SocketAddr::V4(_) => IpAddr::V4(Ipv4Addr::UNSPECIFIED),
740                                    SocketAddr::V6(_) => IpAddr::V6(Ipv6Addr::UNSPECIFIED),
741                                },
742                                0,
743                            )
744                        })
745                })?,
746            InodeSocketKind::RemoteSocket { peer_addr, .. } => *peer_addr,
747            _ => return Err(Errno::Notsup),
748        })
749    }
750
751    pub fn set_opt_flag(&mut self, option: WasiSocketOption, val: bool) -> Result<(), Errno> {
752        let mut inner = self.inner.protected.write().unwrap();
753        match &mut inner.kind {
754            InodeSocketKind::PreSocket { props, .. }
755            | InodeSocketKind::RemoteSocket { props, .. } => {
756                match option {
757                    WasiSocketOption::OnlyV6 => props.only_v6 = val,
758                    WasiSocketOption::ReusePort => props.reuse_port = val,
759                    WasiSocketOption::ReuseAddr => props.reuse_addr = val,
760                    WasiSocketOption::NoDelay => props.no_delay = Some(val),
761                    WasiSocketOption::KeepAlive => props.keep_alive = Some(val),
762                    WasiSocketOption::DontRoute => props.dont_route = Some(val),
763                    _ => return Err(Errno::Inval),
764                };
765            }
766            InodeSocketKind::Raw(sock) => match option {
767                WasiSocketOption::Promiscuous => {
768                    sock.set_promiscuous(val).map_err(net_error_into_wasi_err)?
769                }
770                _ => return Err(Errno::Inval),
771            },
772            InodeSocketKind::TcpStream { socket, .. } => match option {
773                WasiSocketOption::NoDelay => {
774                    socket.set_nodelay(val).map_err(net_error_into_wasi_err)?
775                }
776                WasiSocketOption::KeepAlive => {
777                    socket.set_keepalive(val).map_err(net_error_into_wasi_err)?
778                }
779                WasiSocketOption::DontRoute => {
780                    socket.set_dontroute(val).map_err(net_error_into_wasi_err)?
781                }
782                _ => return Err(Errno::Inval),
783            },
784            InodeSocketKind::TcpListener { .. } => return Err(Errno::Inval),
785            InodeSocketKind::UdpSocket { socket, .. } => match option {
786                WasiSocketOption::Broadcast => {
787                    socket.set_broadcast(val).map_err(net_error_into_wasi_err)?
788                }
789                WasiSocketOption::MulticastLoopV4 => socket
790                    .set_multicast_loop_v4(val)
791                    .map_err(net_error_into_wasi_err)?,
792                WasiSocketOption::MulticastLoopV6 => socket
793                    .set_multicast_loop_v6(val)
794                    .map_err(net_error_into_wasi_err)?,
795                _ => return Err(Errno::Inval),
796            },
797            _ => return Err(Errno::Notsup),
798        }
799        Ok(())
800    }
801
802    pub fn get_opt_flag(&self, option: WasiSocketOption) -> Result<bool, Errno> {
803        let mut inner = self.inner.protected.write().unwrap();
804        Ok(match &mut inner.kind {
805            InodeSocketKind::PreSocket { props, .. }
806            | InodeSocketKind::RemoteSocket { props, .. } => match option {
807                WasiSocketOption::OnlyV6 => props.only_v6,
808                WasiSocketOption::ReusePort => props.reuse_port,
809                WasiSocketOption::ReuseAddr => props.reuse_addr,
810                WasiSocketOption::NoDelay => props.no_delay.unwrap_or_default(),
811                WasiSocketOption::KeepAlive => props.keep_alive.unwrap_or_default(),
812                _ => return Err(Errno::Inval),
813            },
814            InodeSocketKind::Raw(sock) => match option {
815                WasiSocketOption::Promiscuous => {
816                    sock.promiscuous().map_err(net_error_into_wasi_err)?
817                }
818                _ => return Err(Errno::Inval),
819            },
820            InodeSocketKind::TcpStream { socket, .. } => match option {
821                WasiSocketOption::NoDelay => socket.nodelay().map_err(net_error_into_wasi_err)?,
822                WasiSocketOption::KeepAlive => {
823                    socket.keepalive().map_err(net_error_into_wasi_err)?
824                }
825                WasiSocketOption::DontRoute => {
826                    socket.dontroute().map_err(net_error_into_wasi_err)?
827                }
828                _ => return Err(Errno::Inval),
829            },
830            InodeSocketKind::UdpSocket { socket, .. } => match option {
831                WasiSocketOption::Broadcast => {
832                    socket.broadcast().map_err(net_error_into_wasi_err)?
833                }
834                WasiSocketOption::MulticastLoopV4 => socket
835                    .multicast_loop_v4()
836                    .map_err(net_error_into_wasi_err)?,
837                WasiSocketOption::MulticastLoopV6 => socket
838                    .multicast_loop_v6()
839                    .map_err(net_error_into_wasi_err)?,
840                _ => return Err(Errno::Inval),
841            },
842            _ => return Err(Errno::Notsup),
843        })
844    }
845
846    pub fn set_send_buf_size(&mut self, size: usize) -> Result<(), Errno> {
847        let mut inner = self.inner.protected.write().unwrap();
848        match &mut inner.kind {
849            InodeSocketKind::PreSocket { props, .. }
850            | InodeSocketKind::RemoteSocket { props, .. } => {
851                props.send_buf_size = Some(size);
852            }
853            InodeSocketKind::TcpStream { socket, .. } => {
854                socket
855                    .set_send_buf_size(size)
856                    .map_err(net_error_into_wasi_err)?;
857            }
858            _ => return Err(Errno::Notsup),
859        }
860        Ok(())
861    }
862
863    pub fn send_buf_size(&self) -> Result<usize, Errno> {
864        let inner = self.inner.protected.read().unwrap();
865        match &inner.kind {
866            InodeSocketKind::PreSocket { props, .. }
867            | InodeSocketKind::RemoteSocket { props, .. } => {
868                Ok(props.send_buf_size.unwrap_or_default())
869            }
870            InodeSocketKind::TcpStream { socket, .. } => {
871                socket.send_buf_size().map_err(net_error_into_wasi_err)
872            }
873            _ => Err(Errno::Notsup),
874        }
875    }
876
877    pub fn set_recv_buf_size(&mut self, size: usize) -> Result<(), Errno> {
878        let mut inner = self.inner.protected.write().unwrap();
879        match &mut inner.kind {
880            InodeSocketKind::PreSocket { props, .. }
881            | InodeSocketKind::RemoteSocket { props, .. } => {
882                props.recv_buf_size = Some(size);
883            }
884            InodeSocketKind::TcpStream { socket, .. } => {
885                socket
886                    .set_recv_buf_size(size)
887                    .map_err(net_error_into_wasi_err)?;
888            }
889            _ => return Err(Errno::Notsup),
890        }
891        Ok(())
892    }
893
894    pub fn recv_buf_size(&self) -> Result<usize, Errno> {
895        let inner = self.inner.protected.read().unwrap();
896        match &inner.kind {
897            InodeSocketKind::PreSocket { props, .. }
898            | InodeSocketKind::RemoteSocket { props, .. } => {
899                Ok(props.recv_buf_size.unwrap_or_default())
900            }
901            InodeSocketKind::TcpStream { socket, .. } => {
902                socket.recv_buf_size().map_err(net_error_into_wasi_err)
903            }
904            _ => Err(Errno::Notsup),
905        }
906    }
907
908    pub fn set_linger(&mut self, linger: Option<std::time::Duration>) -> Result<(), Errno> {
909        let mut inner = self.inner.protected.write().unwrap();
910        match &mut inner.kind {
911            InodeSocketKind::TcpStream { socket, .. } => {
912                socket.set_linger(linger).map_err(net_error_into_wasi_err)
913            }
914            InodeSocketKind::RemoteSocket { .. } => Ok(()),
915            InodeSocketKind::PreSocket { .. } => Err(Errno::Io),
916            _ => Err(Errno::Notsup),
917        }
918    }
919
920    pub fn linger(&self) -> Result<Option<std::time::Duration>, Errno> {
921        let inner = self.inner.protected.read().unwrap();
922        match &inner.kind {
923            InodeSocketKind::TcpStream { socket, .. } => {
924                socket.linger().map_err(net_error_into_wasi_err)
925            }
926            InodeSocketKind::PreSocket { .. } => Err(Errno::Io),
927            _ => Err(Errno::Notsup),
928        }
929    }
930
931    pub fn set_opt_time(
932        &self,
933        ty: TimeType,
934        timeout: Option<std::time::Duration>,
935    ) -> Result<(), Errno> {
936        let mut inner = self.inner.protected.write().unwrap();
937        match &mut inner.kind {
938            InodeSocketKind::TcpStream {
939                write_timeout,
940                read_timeout,
941                ..
942            } => {
943                match ty {
944                    TimeType::WriteTimeout => *write_timeout = timeout,
945                    TimeType::ReadTimeout => *read_timeout = timeout,
946                    _ => return Err(Errno::Inval),
947                }
948                Ok(())
949            }
950            InodeSocketKind::TcpListener { accept_timeout, .. } => {
951                match ty {
952                    TimeType::AcceptTimeout => *accept_timeout = timeout,
953                    _ => return Err(Errno::Inval),
954                }
955                Ok(())
956            }
957            InodeSocketKind::PreSocket { props, .. }
958            | InodeSocketKind::RemoteSocket { props, .. } => {
959                match ty {
960                    TimeType::ConnectTimeout => props.connect_timeout = timeout,
961                    TimeType::AcceptTimeout => props.accept_timeout = timeout,
962                    TimeType::ReadTimeout => props.read_timeout = timeout,
963                    TimeType::WriteTimeout => props.write_timeout = timeout,
964                    _ => return Err(Errno::Io),
965                }
966                Ok(())
967            }
968            _ => Err(Errno::Notsup),
969        }
970    }
971
972    pub fn opt_time(&self, ty: TimeType) -> Result<Option<std::time::Duration>, Errno> {
973        let inner = self.inner.protected.read().unwrap();
974        match &inner.kind {
975            InodeSocketKind::TcpStream {
976                read_timeout,
977                write_timeout,
978                ..
979            } => Ok(match ty {
980                TimeType::ReadTimeout => *read_timeout,
981                TimeType::WriteTimeout => *write_timeout,
982                _ => return Err(Errno::Inval),
983            }),
984            InodeSocketKind::TcpListener { accept_timeout, .. } => Ok(match ty {
985                TimeType::AcceptTimeout => *accept_timeout,
986                _ => return Err(Errno::Inval),
987            }),
988            InodeSocketKind::PreSocket { props, .. }
989            | InodeSocketKind::RemoteSocket { props, .. } => match ty {
990                TimeType::ConnectTimeout => Ok(props.connect_timeout),
991                TimeType::AcceptTimeout => Ok(props.accept_timeout),
992                TimeType::ReadTimeout => Ok(props.read_timeout),
993                TimeType::WriteTimeout => Ok(props.write_timeout),
994                _ => Err(Errno::Inval),
995            },
996            _ => Err(Errno::Notsup),
997        }
998    }
999
1000    pub fn set_ttl(&self, ttl: u32) -> Result<(), Errno> {
1001        let mut inner = self.inner.protected.write().unwrap();
1002        match &mut inner.kind {
1003            InodeSocketKind::TcpStream { socket, .. } => {
1004                socket.set_ttl(ttl).map_err(net_error_into_wasi_err)
1005            }
1006            InodeSocketKind::UdpSocket { socket, .. } => {
1007                socket.set_ttl(ttl).map_err(net_error_into_wasi_err)
1008            }
1009            InodeSocketKind::RemoteSocket { ttl: set_ttl, .. } => {
1010                *set_ttl = ttl;
1011                Ok(())
1012            }
1013            InodeSocketKind::PreSocket { .. } => Err(Errno::Io),
1014            _ => Err(Errno::Notsup),
1015        }
1016    }
1017
1018    pub fn ttl(&self) -> Result<u32, Errno> {
1019        let inner = self.inner.protected.read().unwrap();
1020        match &inner.kind {
1021            InodeSocketKind::TcpStream { socket, .. } => {
1022                socket.ttl().map_err(net_error_into_wasi_err)
1023            }
1024            InodeSocketKind::UdpSocket { socket, .. } => {
1025                socket.ttl().map_err(net_error_into_wasi_err)
1026            }
1027            InodeSocketKind::RemoteSocket { ttl, .. } => Ok(*ttl),
1028            InodeSocketKind::PreSocket { .. } => Err(Errno::Io),
1029            _ => Err(Errno::Notsup),
1030        }
1031    }
1032
1033    pub fn set_multicast_ttl_v4(&self, ttl: u32) -> Result<(), Errno> {
1034        let mut inner = self.inner.protected.write().unwrap();
1035        match &mut inner.kind {
1036            InodeSocketKind::UdpSocket { socket, .. } => socket
1037                .set_multicast_ttl_v4(ttl)
1038                .map_err(net_error_into_wasi_err),
1039            InodeSocketKind::RemoteSocket {
1040                multicast_ttl: set_ttl,
1041                ..
1042            } => {
1043                *set_ttl = ttl;
1044                Ok(())
1045            }
1046            InodeSocketKind::PreSocket { .. } => Err(Errno::Io),
1047            _ => Err(Errno::Notsup),
1048        }
1049    }
1050
1051    pub fn multicast_ttl_v4(&self) -> Result<u32, Errno> {
1052        let inner = self.inner.protected.read().unwrap();
1053        match &inner.kind {
1054            InodeSocketKind::UdpSocket { socket, .. } => {
1055                socket.multicast_ttl_v4().map_err(net_error_into_wasi_err)
1056            }
1057            InodeSocketKind::RemoteSocket { multicast_ttl, .. } => Ok(*multicast_ttl),
1058            InodeSocketKind::PreSocket { .. } => Err(Errno::Io),
1059            _ => Err(Errno::Notsup),
1060        }
1061    }
1062
1063    pub fn join_multicast_v4(&self, multiaddr: Ipv4Addr, iface: Ipv4Addr) -> Result<(), Errno> {
1064        let mut inner = self.inner.protected.write().unwrap();
1065        match &mut inner.kind {
1066            InodeSocketKind::UdpSocket { socket, .. } => socket
1067                .join_multicast_v4(multiaddr, iface)
1068                .map_err(net_error_into_wasi_err),
1069            InodeSocketKind::RemoteSocket { .. } => Ok(()),
1070            InodeSocketKind::PreSocket { .. } => Err(Errno::Io),
1071            _ => Err(Errno::Notsup),
1072        }
1073    }
1074
1075    pub fn leave_multicast_v4(&self, multiaddr: Ipv4Addr, iface: Ipv4Addr) -> Result<(), Errno> {
1076        let mut inner = self.inner.protected.write().unwrap();
1077        match &mut inner.kind {
1078            InodeSocketKind::UdpSocket { socket, .. } => socket
1079                .leave_multicast_v4(multiaddr, iface)
1080                .map_err(net_error_into_wasi_err),
1081            InodeSocketKind::RemoteSocket { .. } => Ok(()),
1082            InodeSocketKind::PreSocket { .. } => Err(Errno::Io),
1083            _ => Err(Errno::Notsup),
1084        }
1085    }
1086
1087    pub fn join_multicast_v6(&self, multiaddr: Ipv6Addr, iface: u32) -> Result<(), Errno> {
1088        let mut inner = self.inner.protected.write().unwrap();
1089        match &mut inner.kind {
1090            InodeSocketKind::UdpSocket { socket, .. } => socket
1091                .join_multicast_v6(multiaddr, iface)
1092                .map_err(net_error_into_wasi_err),
1093            InodeSocketKind::RemoteSocket { .. } => Ok(()),
1094            InodeSocketKind::PreSocket { .. } => Err(Errno::Io),
1095            _ => Err(Errno::Notsup),
1096        }
1097    }
1098
1099    pub fn leave_multicast_v6(&mut self, multiaddr: Ipv6Addr, iface: u32) -> Result<(), Errno> {
1100        let mut inner = self.inner.protected.write().unwrap();
1101        match &mut inner.kind {
1102            InodeSocketKind::UdpSocket { socket, .. } => socket
1103                .leave_multicast_v6(multiaddr, iface)
1104                .map_err(net_error_into_wasi_err),
1105            InodeSocketKind::RemoteSocket { .. } => Ok(()),
1106            InodeSocketKind::PreSocket { .. } => Err(Errno::Io),
1107            _ => Err(Errno::Notsup),
1108        }
1109    }
1110
1111    pub async fn send(
1112        &self,
1113        tasks: &dyn VirtualTaskManager,
1114        buf: &[u8],
1115        timeout: Option<Duration>,
1116        nonblocking: bool,
1117    ) -> Result<usize, Errno> {
1118        struct SocketSender<'a, 'b> {
1119            inner: &'a InodeSocketInner,
1120            data: &'b [u8],
1121            nonblocking: bool,
1122            handler_registered: bool,
1123        }
1124        impl Drop for SocketSender<'_, '_> {
1125            fn drop(&mut self) {
1126                if self.handler_registered {
1127                    let mut inner = self.inner.protected.write().unwrap();
1128                    inner.remove_handler();
1129                }
1130            }
1131        }
1132        impl Future for SocketSender<'_, '_> {
1133            type Output = Result<usize, Errno>;
1134            fn poll(
1135                mut self: Pin<&mut Self>,
1136                cx: &mut std::task::Context<'_>,
1137            ) -> Poll<Self::Output> {
1138                loop {
1139                    let mut inner = self.inner.protected.write().unwrap();
1140                    let res = match &mut inner.kind {
1141                        InodeSocketKind::Raw(socket) => socket.try_send(self.data),
1142                        InodeSocketKind::TcpStream { socket, .. } => socket.try_send(self.data),
1143                        InodeSocketKind::UdpSocket { socket, peer } => {
1144                            if let Some(peer) = peer {
1145                                socket.try_send_to(self.data, *peer)
1146                            } else {
1147                                Err(NetworkError::NotConnected)
1148                            }
1149                        }
1150                        InodeSocketKind::PreSocket { .. } => {
1151                            return Poll::Ready(Err(Errno::Notconn));
1152                        }
1153                        InodeSocketKind::RemoteSocket { is_dead, .. } => {
1154                            return match is_dead {
1155                                true => Poll::Ready(Err(Errno::Connreset)),
1156                                false => Poll::Ready(Ok(self.data.len())),
1157                            };
1158                        }
1159                        _ => return Poll::Ready(Err(Errno::Notsup)),
1160                    };
1161                    return match res {
1162                        Ok(amt) => Poll::Ready(Ok(amt)),
1163                        Err(NetworkError::WouldBlock) if self.nonblocking => {
1164                            Poll::Ready(Err(Errno::Again))
1165                        }
1166                        Err(NetworkError::WouldBlock) if !self.handler_registered => {
1167                            inner
1168                                .set_handler(cx.waker().into())
1169                                .map_err(net_error_into_wasi_err)?;
1170                            drop(inner);
1171                            self.handler_registered = true;
1172                            continue;
1173                        }
1174                        Err(NetworkError::WouldBlock) => Poll::Pending,
1175                        Err(err) => Poll::Ready(Err(net_error_into_wasi_err(err))),
1176                    };
1177                }
1178            }
1179        }
1180
1181        let poller = SocketSender {
1182            inner: &self.inner,
1183            data: buf,
1184            nonblocking,
1185            handler_registered: false,
1186        };
1187        if let Some(timeout) = timeout {
1188            tokio::select! {
1189                res = poller => res,
1190                _ = tasks.sleep_now(timeout) => Err(Errno::Timedout)
1191            }
1192        } else {
1193            poller.await
1194        }
1195    }
1196
1197    pub async fn send_to<M: MemorySize>(
1198        &self,
1199        tasks: &dyn VirtualTaskManager,
1200        buf: &[u8],
1201        addr: SocketAddr,
1202        timeout: Option<Duration>,
1203        nonblocking: bool,
1204    ) -> Result<usize, Errno> {
1205        struct SocketSender<'a, 'b> {
1206            inner: &'a InodeSocketInner,
1207            data: &'b [u8],
1208            addr: SocketAddr,
1209            nonblocking: bool,
1210            handler_registered: bool,
1211        }
1212        impl Drop for SocketSender<'_, '_> {
1213            fn drop(&mut self) {
1214                if self.handler_registered {
1215                    let mut inner = self.inner.protected.write().unwrap();
1216                    inner.remove_handler();
1217                }
1218            }
1219        }
1220        impl Future for SocketSender<'_, '_> {
1221            type Output = Result<usize, Errno>;
1222            fn poll(
1223                mut self: Pin<&mut Self>,
1224                cx: &mut std::task::Context<'_>,
1225            ) -> Poll<Self::Output> {
1226                loop {
1227                    let mut inner = self.inner.protected.write().unwrap();
1228                    let res = match &mut inner.kind {
1229                        InodeSocketKind::Icmp(socket) => socket.try_send_to(self.data, self.addr),
1230                        InodeSocketKind::UdpSocket { socket, .. } => {
1231                            socket.try_send_to(self.data, self.addr)
1232                        }
1233                        InodeSocketKind::PreSocket { .. } => {
1234                            return Poll::Ready(Err(Errno::Notconn));
1235                        }
1236                        InodeSocketKind::RemoteSocket { is_dead, .. } => {
1237                            return match is_dead {
1238                                true => Poll::Ready(Err(Errno::Connreset)),
1239                                false => Poll::Ready(Ok(self.data.len())),
1240                            };
1241                        }
1242                        _ => return Poll::Ready(Err(Errno::Notsup)),
1243                    };
1244                    return match res {
1245                        Ok(amt) => Poll::Ready(Ok(amt)),
1246                        Err(NetworkError::WouldBlock) if self.nonblocking => {
1247                            Poll::Ready(Err(Errno::Again))
1248                        }
1249                        Err(NetworkError::WouldBlock) if !self.handler_registered => {
1250                            inner
1251                                .set_handler(cx.waker().into())
1252                                .map_err(net_error_into_wasi_err)?;
1253                            self.handler_registered = true;
1254                            drop(inner);
1255                            continue;
1256                        }
1257                        Err(NetworkError::WouldBlock) => Poll::Pending,
1258                        Err(err) => Poll::Ready(Err(net_error_into_wasi_err(err))),
1259                    };
1260                }
1261            }
1262        }
1263
1264        let poller = SocketSender {
1265            inner: &self.inner,
1266            data: buf,
1267            addr,
1268            nonblocking,
1269            handler_registered: false,
1270        };
1271        if let Some(timeout) = timeout {
1272            tokio::select! {
1273                res = poller => res,
1274                _ = tasks.sleep_now(timeout) => Err(Errno::Timedout)
1275            }
1276        } else {
1277            poller.await
1278        }
1279    }
1280
1281    pub async fn recv(
1282        &self,
1283        tasks: &dyn VirtualTaskManager,
1284        buf: &mut [MaybeUninit<u8>],
1285        timeout: Option<Duration>,
1286        nonblocking: bool,
1287        peek: bool,
1288    ) -> Result<usize, Errno> {
1289        struct SocketReceiver<'a, 'b> {
1290            inner: &'a InodeSocketInner,
1291            data: &'b mut [MaybeUninit<u8>],
1292            nonblocking: bool,
1293            peek: bool,
1294            handler_registered: bool,
1295        }
1296        impl Drop for SocketReceiver<'_, '_> {
1297            fn drop(&mut self) {
1298                if self.handler_registered {
1299                    let mut inner = self.inner.protected.write().unwrap();
1300                    inner.remove_handler();
1301                }
1302            }
1303        }
1304        impl Future for SocketReceiver<'_, '_> {
1305            type Output = Result<usize, Errno>;
1306            fn poll(
1307                mut self: Pin<&mut Self>,
1308                cx: &mut std::task::Context<'_>,
1309            ) -> Poll<Self::Output> {
1310                loop {
1311                    let peek = self.peek;
1312                    let mut inner = self.inner.protected.write().unwrap();
1313                    let res = match &mut inner.kind {
1314                        InodeSocketKind::Raw(socket) => socket.try_recv(self.data, peek),
1315                        InodeSocketKind::TcpStream { socket, .. } => {
1316                            socket.try_recv(self.data, peek)
1317                        }
1318                        InodeSocketKind::UdpSocket { socket, peer } => {
1319                            if let Some(peer) = peer {
1320                                match socket.try_recv_from(self.data, peek) {
1321                                    Ok((amt, addr)) if addr == *peer => Ok(amt),
1322                                    Ok(_) => Err(NetworkError::WouldBlock),
1323                                    Err(err) => Err(err),
1324                                }
1325                            } else {
1326                                match socket.try_recv_from(self.data, peek) {
1327                                    Ok((amt, _)) => Ok(amt),
1328                                    Err(err) => Err(err),
1329                                }
1330                            }
1331                        }
1332                        InodeSocketKind::RemoteSocket { is_dead, .. } => {
1333                            return match is_dead {
1334                                true => Poll::Ready(Ok(0)),
1335                                false => Poll::Pending,
1336                            };
1337                        }
1338                        InodeSocketKind::PreSocket { .. } => {
1339                            return Poll::Ready(Err(Errno::Notconn));
1340                        }
1341                        _ => return Poll::Ready(Err(Errno::Notsup)),
1342                    };
1343                    return match res {
1344                        Ok(amt) => Poll::Ready(Ok(amt)),
1345                        Err(NetworkError::WouldBlock) if self.nonblocking => {
1346                            Poll::Ready(Err(Errno::Again))
1347                        }
1348                        Err(NetworkError::WouldBlock) if !self.handler_registered => {
1349                            inner
1350                                .set_handler(cx.waker().into())
1351                                .map_err(net_error_into_wasi_err)?;
1352                            self.handler_registered = true;
1353                            drop(inner);
1354                            continue;
1355                        }
1356
1357                        Err(NetworkError::WouldBlock) => Poll::Pending,
1358                        Err(err) => Poll::Ready(Err(net_error_into_wasi_err(err))),
1359                    };
1360                }
1361            }
1362        }
1363
1364        let poller = SocketReceiver {
1365            inner: &self.inner,
1366            data: buf,
1367            nonblocking,
1368            peek,
1369            handler_registered: false,
1370        };
1371        if let Some(timeout) = timeout {
1372            tokio::select! {
1373                res = poller => res,
1374                _ = tasks.sleep_now(timeout) => Err(Errno::Timedout)
1375            }
1376        } else {
1377            poller.await
1378        }
1379    }
1380
1381    pub async fn recv_from(
1382        &self,
1383        tasks: &dyn VirtualTaskManager,
1384        buf: &mut [MaybeUninit<u8>],
1385        timeout: Option<Duration>,
1386        nonblocking: bool,
1387        peek: bool,
1388    ) -> Result<(usize, SocketAddr), Errno> {
1389        struct SocketReceiver<'a, 'b> {
1390            inner: &'a InodeSocketInner,
1391            data: &'b mut [MaybeUninit<u8>],
1392            nonblocking: bool,
1393            peek: bool,
1394            handler_registered: bool,
1395        }
1396        impl Drop for SocketReceiver<'_, '_> {
1397            fn drop(&mut self) {
1398                if self.handler_registered {
1399                    let mut inner = self.inner.protected.write().unwrap();
1400                    inner.remove_handler();
1401                }
1402            }
1403        }
1404        impl Future for SocketReceiver<'_, '_> {
1405            type Output = Result<(usize, SocketAddr), Errno>;
1406            fn poll(
1407                mut self: Pin<&mut Self>,
1408                cx: &mut std::task::Context<'_>,
1409            ) -> Poll<Self::Output> {
1410                let peek = self.peek;
1411                let mut inner = self.inner.protected.write().unwrap();
1412                loop {
1413                    let res = match &mut inner.kind {
1414                        InodeSocketKind::Icmp(socket) => socket.try_recv_from(self.data, peek),
1415                        InodeSocketKind::UdpSocket { socket, .. } => {
1416                            socket.try_recv_from(self.data, peek)
1417                        }
1418                        InodeSocketKind::RemoteSocket {
1419                            is_dead, peer_addr, ..
1420                        } => {
1421                            return match is_dead {
1422                                true => Poll::Ready(Ok((0, *peer_addr))),
1423                                false => Poll::Pending,
1424                            };
1425                        }
1426                        InodeSocketKind::PreSocket { .. } => {
1427                            return Poll::Ready(Err(Errno::Notconn));
1428                        }
1429                        _ => return Poll::Ready(Err(Errno::Notsup)),
1430                    };
1431                    return match res {
1432                        Ok((amt, addr)) => Poll::Ready(Ok((amt, addr))),
1433                        Err(NetworkError::WouldBlock) if self.nonblocking => {
1434                            Poll::Ready(Err(Errno::Again))
1435                        }
1436                        Err(NetworkError::WouldBlock) if !self.handler_registered => {
1437                            inner
1438                                .set_handler(cx.waker().into())
1439                                .map_err(net_error_into_wasi_err)?;
1440                            self.handler_registered = true;
1441                            continue;
1442                        }
1443                        Err(NetworkError::WouldBlock) => Poll::Pending,
1444                        Err(err) => Poll::Ready(Err(net_error_into_wasi_err(err))),
1445                    };
1446                }
1447            }
1448        }
1449
1450        let poller = SocketReceiver {
1451            inner: &self.inner,
1452            data: buf,
1453            nonblocking,
1454            peek,
1455            handler_registered: false,
1456        };
1457        if let Some(timeout) = timeout {
1458            tokio::select! {
1459                res = poller => res,
1460                _ = tasks.sleep_now(timeout) => Err(Errno::Timedout)
1461            }
1462        } else {
1463            poller.await
1464        }
1465    }
1466
1467    pub fn shutdown(&mut self, how: std::net::Shutdown) -> Result<(), Errno> {
1468        let mut inner = self.inner.protected.write().unwrap();
1469        match &mut inner.kind {
1470            InodeSocketKind::TcpStream { socket, .. } => {
1471                socket.shutdown(how).map_err(net_error_into_wasi_err)?;
1472            }
1473            InodeSocketKind::RemoteSocket { .. } => return Ok(()),
1474            InodeSocketKind::PreSocket { .. } => return Err(Errno::Notconn),
1475            _ => return Err(Errno::Notsup),
1476        }
1477        Ok(())
1478    }
1479
1480    pub async fn can_write(&self) -> bool {
1481        if let Ok(mut guard) = self.inner.protected.try_write() {
1482            #[allow(clippy::match_like_matches_macro)]
1483            match &mut guard.kind {
1484                InodeSocketKind::TcpStream { .. }
1485                | InodeSocketKind::UdpSocket { .. }
1486                | InodeSocketKind::Raw(..) => true,
1487                InodeSocketKind::RemoteSocket { is_dead, .. } => !(*is_dead),
1488                _ => false,
1489            }
1490        } else {
1491            false
1492        }
1493    }
1494}
1495
1496impl InodeSocketProtected {
1497    pub fn remove_handler(&mut self) {
1498        match &mut self.kind {
1499            InodeSocketKind::TcpListener { socket, .. } => socket.remove_handler(),
1500            InodeSocketKind::TcpStream { socket, .. } => socket.remove_handler(),
1501            InodeSocketKind::UdpSocket { socket, .. } => socket.remove_handler(),
1502            InodeSocketKind::Raw(socket) => socket.remove_handler(),
1503            InodeSocketKind::Icmp(socket) => socket.remove_handler(),
1504            InodeSocketKind::PreSocket { props, .. } => {
1505                props.handler.take();
1506            }
1507            InodeSocketKind::RemoteSocket { props, .. } => {
1508                props.handler.take();
1509            }
1510        }
1511    }
1512
1513    pub fn poll_read_ready(&mut self, cx: &mut Context<'_>) -> Poll<io::Result<usize>> {
1514        match &mut self.kind {
1515            InodeSocketKind::TcpListener { socket, .. } => socket.poll_read_ready(cx),
1516            InodeSocketKind::TcpStream { socket, .. } => socket.poll_read_ready(cx),
1517            InodeSocketKind::UdpSocket { socket, .. } => socket.poll_read_ready(cx),
1518            InodeSocketKind::Raw(socket) => socket.poll_read_ready(cx),
1519            InodeSocketKind::Icmp(socket) => socket.poll_read_ready(cx),
1520            InodeSocketKind::PreSocket { .. } => Poll::Pending,
1521            InodeSocketKind::RemoteSocket { is_dead, .. } => match is_dead {
1522                true => Poll::Ready(Ok(0)),
1523                false => Poll::Pending,
1524            },
1525        }
1526        .map_err(net_error_into_io_err)
1527    }
1528
1529    pub fn poll_write_ready(&mut self, cx: &mut Context<'_>) -> Poll<io::Result<usize>> {
1530        match &mut self.kind {
1531            InodeSocketKind::TcpListener { socket, .. } => socket.poll_write_ready(cx),
1532            InodeSocketKind::TcpStream { socket, .. } => socket.poll_write_ready(cx),
1533            InodeSocketKind::UdpSocket { socket, .. } => socket.poll_write_ready(cx),
1534            InodeSocketKind::Raw(socket) => socket.poll_write_ready(cx),
1535            InodeSocketKind::Icmp(socket) => socket.poll_write_ready(cx),
1536            InodeSocketKind::PreSocket { .. } => Poll::Pending,
1537            InodeSocketKind::RemoteSocket { is_dead, .. } => match is_dead {
1538                true => Poll::Ready(Ok(0)),
1539                false => Poll::Pending,
1540            },
1541        }
1542        .map_err(net_error_into_io_err)
1543    }
1544
1545    pub fn set_handler(
1546        &mut self,
1547        handler: Box<dyn InterestHandler + Send + Sync>,
1548    ) -> virtual_net::Result<()> {
1549        match &mut self.kind {
1550            InodeSocketKind::TcpListener { socket, .. } => socket.set_handler(handler),
1551            InodeSocketKind::TcpStream { socket, .. } => socket.set_handler(handler),
1552            InodeSocketKind::UdpSocket { socket, .. } => socket.set_handler(handler),
1553            InodeSocketKind::Raw(socket) => socket.set_handler(handler),
1554            InodeSocketKind::Icmp(socket) => socket.set_handler(handler),
1555            InodeSocketKind::PreSocket { props, .. }
1556            | InodeSocketKind::RemoteSocket { props, .. } => {
1557                props.handler.replace(handler);
1558                Ok(())
1559            }
1560        }
1561    }
1562}
1563
1564// TODO: review allow...
1565#[allow(dead_code)]
1566pub(crate) fn all_socket_rights() -> Rights {
1567    Rights::FD_FDSTAT_SET_FLAGS
1568        .union(Rights::FD_FILESTAT_GET)
1569        .union(Rights::FD_READ)
1570        .union(Rights::FD_WRITE)
1571        .union(Rights::POLL_FD_READWRITE)
1572        .union(Rights::SOCK_SHUTDOWN)
1573        .union(Rights::SOCK_CONNECT)
1574        .union(Rights::SOCK_LISTEN)
1575        .union(Rights::SOCK_BIND)
1576        .union(Rights::SOCK_ACCEPT)
1577        .union(Rights::SOCK_RECV)
1578        .union(Rights::SOCK_SEND)
1579        .union(Rights::SOCK_ADDR_LOCAL)
1580        .union(Rights::SOCK_ADDR_REMOTE)
1581        .union(Rights::SOCK_RECV_FROM)
1582        .union(Rights::SOCK_SEND_TO)
1583}
wasmer_wasix/net/socket.rs

wasmer_wasix/net/
socket.rs
