WASIX POSIX compatibility tests

lib/cli/src/commands/run/wasi.rs

@@ -377,10 +377,12 @@ impl Wasi {
                 .unwrap();
 
             if have_current_dir {
-                builder.map_dir(".", MAPPED_CURRENT_DIR_DEFAULT_PATH)?
+                builder = builder.map_dir(".", MAPPED_CURRENT_DIR_DEFAULT_PATH)?;
             } else {
-                builder.map_dir(".", "/")?
+                builder = builder.map_dir(".", "/")?;
             }
+
+            builder
         };
 
         *builder.capabilities_mut() = self.capabilities();

lib/virtual-fs/src/host_fs.rs

@@ -114,7 +114,7 @@ impl crate::FileSystem for FileSystem {
                     .to_owned();
                 let path = Path::new("/").join(path);
 
-                let metadata = entry.metadata()?;
+                let metadata = fs::symlink_metadata(entry.path())?;
 
                 Ok(DirEntry {
                     path,
@@ -519,8 +519,21 @@ impl VirtualFile for File {
     }
 
     fn set_times(&mut self, atime: Option<u64>, mtime: Option<u64>) -> crate::Result<()> {
-        let atime = atime.map(|t| filetime::FileTime::from_unix_time(t as i64, 0));
-        let mtime = mtime.map(|t| filetime::FileTime::from_unix_time(t as i64, 0));
+        let to_filetime = |t: u64| -> crate::Result<filetime::FileTime> {
+            let secs = t / 1_000_000_000;
+            let nanos = (t % 1_000_000_000) as u32;
+            let secs = i64::try_from(secs).map_err(|_| crate::FsError::IOError)?;
+            Ok(filetime::FileTime::from_unix_time(secs, nanos))
+        };
+
+        let atime = match atime {
+            Some(t) => Some(to_filetime(t)?),
+            None => None,
+        };
+        let mtime = match mtime {
+            Some(t) => Some(to_filetime(t)?),
+            None => None,
+        };
 
         filetime::set_file_handle_times(&self.inner_std, atime, mtime)
             .map_err(|_| crate::FsError::IOError)

lib/virtual-fs/src/pipe.rs

@@ -18,6 +18,8 @@ use virtual_mio::{InterestHandler, InterestType};
 
 use crate::{ArcFile, FsError, VirtualFile};
 
+const PIPE_MAX_BYTES: usize = 64 * 1024;
+
 // Each pipe end is separately cloneable. The overall pipe
 // remains open as long as at least one tx end and one rx
 // end are still alive.
@@ -53,19 +55,19 @@ impl PipeRx {
         // Should return how much actual data was read from the provided slice
         write: impl FnOnce(&[u8]) -> Option<usize>,
     ) -> Option<usize> {
-        rx.buffer.as_mut().and_then(|read_buffer| {
+        if let Some(read_buffer) = rx.buffer.as_mut() {
             let buf_len = read_buffer.len();
             if buf_len > 0 {
                 let mut read = buf_len.min(max_len);
                 let inner_buf = &read_buffer[..read];
                 // read = ?;
                 read = write(inner_buf)?;
                 read_buffer.advance(read);
-                Some(read)
-            } else {
-                None
+                rx.pending_bytes = rx.pending_bytes.saturating_sub(read);
+                return Some(read);
             }
-        })
+        }
+        None
     }
 
     pub fn close(&mut self) {
@@ -151,6 +153,7 @@ struct PipeReceiver {
     // actual receiver, we can't make use of an mpmc channel
     chan: mpsc::UnboundedReceiver<Vec<u8>>,
     buffer: Option<Bytes>,
+    pending_bytes: usize,
     interest_handler: Option<Box<dyn InterestHandler>>,
 }
 
@@ -161,6 +164,7 @@ impl Pipe {
         let recv = Arc::new(Mutex::new(PipeReceiver {
             chan: rx,
             buffer: None,
+            pending_bytes: 0,
             interest_handler: None,
         }));
         Pipe {
@@ -218,6 +222,42 @@ impl PipeTx {
         _ = self.tx.take();
     }
 
+    fn send_bytes(&self, buf: &[u8]) -> std::io::Result<usize> {
+        let Some(ref tx) = self.tx else {
+            return Err(std::io::Error::new(
+                std::io::ErrorKind::BrokenPipe,
+                "PipeTx is closed",
+            ));
+        };
+
+        tx.send(buf.to_vec())
+            .map_err(|_| Into::<std::io::Error>::into(std::io::ErrorKind::BrokenPipe))?;
+
+        if let Some(rx_end) = self.rx_end.upgrade() {
+            let mut rx = rx_end.lock().unwrap();
+            rx.pending_bytes = rx.pending_bytes.saturating_add(buf.len());
+        }
+
+        self.mark_other_end_readable();
+        Ok(buf.len())
+    }
+
+    pub fn try_write_nonblocking(&self, buf: &[u8]) -> std::io::Result<usize> {
+        let Some(rx_end) = self.rx_end.upgrade() else {
+            return Err(std::io::Error::new(
+                std::io::ErrorKind::BrokenPipe,
+                "PipeRx is closed",
+            ));
+        };
+        {
+            let rx = rx_end.lock().unwrap();
+            if rx.pending_bytes.saturating_add(buf.len()) > PIPE_MAX_BYTES {
+                return Err(std::io::ErrorKind::WouldBlock.into());
+            }
+        }
+        self.send_bytes(buf)
+    }
+
     pub fn poll_write_ready(self: Pin<&mut Self>) -> Poll<io::Result<usize>> {
         let Some(ref tx) = self.tx else {
             return Poll::Ready(Err(std::io::Error::new(
@@ -321,17 +361,7 @@ impl std::io::Write for Pipe {
 
 impl std::io::Write for PipeTx {
     fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
-        let Some(ref tx) = self.tx else {
-            return Err(std::io::Error::new(
-                std::io::ErrorKind::BrokenPipe,
-                "PipeTx is closed",
-            ));
-        };
-
-        tx.send(buf.to_vec())
-            .map_err(|_| Into::<std::io::Error>::into(std::io::ErrorKind::BrokenPipe))?;
-        self.mark_other_end_readable();
-        Ok(buf.len())
+        self.send_bytes(buf)
     }
 
     fn flush(&mut self) -> std::io::Result<()> {
@@ -408,22 +438,7 @@ impl AsyncWrite for PipeTx {
         _cx: &mut Context<'_>,
         buf: &[u8],
     ) -> Poll<io::Result<usize>> {
-        let Some(ref tx) = self.tx else {
-            return Poll::Ready(Err(std::io::Error::new(
-                std::io::ErrorKind::BrokenPipe,
-                "PipeTx is closed",
-            )));
-        };
-
-        match tx.send(buf.to_vec()) {
-            Ok(()) => {
-                self.mark_other_end_readable();
-                Poll::Ready(Ok(buf.len()))
-            }
-            Err(_) => Poll::Ready(Err(Into::<std::io::Error>::into(
-                std::io::ErrorKind::BrokenPipe,
-            ))),
-        }
+        Poll::Ready(self.send_bytes(buf))
     }
 
     fn poll_flush(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<io::Result<()>> {

lib/virtual-net/Cargo.toml

@@ -25,6 +25,7 @@ libc = { workspace = true, optional = true }
 mio = { workspace = true, optional = true }
 socket2 = { workspace = true, optional = true }
 derive_more.workspace = true
+interfaces.workspace = true
 virtual-mio = { path = "../virtual-io", version = "0.701.0", default-features = false }
 bincode = { workspace = true, features = ["serde"] }
 serde = { workspace = true, default-features = false, features = ["derive"] }

lib/virtual-net/src/host.rs

@@ -8,7 +8,7 @@ use crate::{
 };
 use crate::{VirtualIoSource, io_err_into_net_error};
 use bytes::{Buf, BytesMut};
-use std::collections::VecDeque;
+use std::collections::{HashSet, VecDeque};
 use std::io::{self, Read, Write};
 use std::mem::MaybeUninit;
 use std::net::{IpAddr, Ipv4Addr, Ipv6Addr, Shutdown, SocketAddr};
@@ -29,6 +29,37 @@ use virtual_mio::{
     HandlerGuardState, InterestGuard, InterestHandler, InterestType, Selector, state_as_waker_map,
 };
 
+fn prefix_from_mask(ip: IpAddr, mask: Option<SocketAddr>) -> u8 {
+    let default_prefix = match ip {
+        IpAddr::V4(_) => 32,
+        IpAddr::V6(_) => 128,
+    };
+    let Some(mask) = mask else {
+        return default_prefix;
+    };
+
+    match (ip, mask.ip()) {
+        (IpAddr::V4(_), IpAddr::V4(m)) => {
+            let bits = u32::from_be_bytes(m.octets()).count_ones();
+            bits as u8
+        }
+        (IpAddr::V6(_), IpAddr::V6(m)) => {
+            let bits: u32 = m.octets().iter().map(|b| b.count_ones()).sum();
+            bits as u8
+        }
+        _ => default_prefix,
+    }
+}
+
+fn default_loopback_cidrs() -> Vec<IpCidr> {
+    vec![
+        IpCidr {
+            ip: IpAddr::V4(Ipv4Addr::LOCALHOST),
+            prefix: 8,
+        },
+    ]
+}
+
 #[derive(Debug)]
 pub struct LocalNetworking {
     selector: Arc<Selector>,
@@ -69,6 +100,44 @@ impl Default for LocalNetworking {
 #[async_trait::async_trait]
 #[allow(unused_variables)]
 impl VirtualNetworking for LocalNetworking {
+    async fn ip_list(&self) -> Result<Vec<IpCidr>> {
+        let mut cidrs = HashSet::new();
+        if let Ok(ifaces) = interfaces::Interface::get_all() {
+            for iface in ifaces {
+                for addr in iface.addresses.iter() {
+                    if let Some(sock) = addr.addr {
+                        let ip = sock.ip();
+                        if ip.is_ipv6() {
+                            continue;
+                        }
+                        let prefix = prefix_from_mask(ip, addr.mask);
+                        cidrs.insert(IpCidr { ip, prefix });
+                    }
+                }
+            }
+        }
+
+        let mut cidrs: Vec<IpCidr> = cidrs.into_iter().collect();
+        if cidrs.is_empty() {
+            cidrs = default_loopback_cidrs();
+        }
+        Ok(cidrs)
+    }
+
+    async fn route_list(&self) -> Result<Vec<IpRoute>> {
+        let cidrs = self.ip_list().await?;
+        let routes = cidrs
+            .into_iter()
+            .map(|cidr| IpRoute {
+                cidr,
+                via_router: cidr.ip,
+                preferred_until: None,
+                expires_at: None,
+            })
+            .collect();
+        Ok(routes)
+    }
+
     async fn listen_tcp(
         &self,
         addr: SocketAddr,
@@ -1033,3 +1102,52 @@ impl VirtualIoSource for LocalUdpSocket {
         Poll::Pending
     }
 }
+
+
+#[cfg(test)]
+mod udp_tests {
+    use super::*;
+    use std::mem::MaybeUninit;
+    use std::net::{Ipv4Addr, SocketAddr};
+
+    #[tokio::test]
+    async fn udp_send_recv_loopback() {
+        let net = LocalNetworking::new();
+        let mut sock = net
+            .bind_udp(SocketAddr::from((Ipv4Addr::LOCALHOST, 0)), false, false)
+            .await
+            .unwrap();
+        let local = sock.addr_local().unwrap();
+
+        // Empty recv should wouldblock
+        let mut buf: [MaybeUninit<u8>; 16] = [MaybeUninit::uninit(); 16];
+        let err = sock.try_recv_from(&mut buf, false).unwrap_err();
+        assert_eq!(err, NetworkError::WouldBlock);
+
+        // Send to self and read back
+        let sent = sock.try_send_to(b"hi", local).unwrap();
+        assert_eq!(sent, 2);
+
+        let mut read = None;
+        let mut addr = None;
+        for _ in 0..50 {
+            match sock.try_recv_from(&mut buf, false) {
+                Ok((n, a)) => {
+                    read = Some(n);
+                    addr = Some(a);
+                    break;
+                }
+                Err(NetworkError::WouldBlock) => {
+                    tokio::time::sleep(std::time::Duration::from_millis(2)).await;
+                }
+                Err(err) => panic!("unexpected recv error: {err:?}"),
+            }
+        }
+        let read = read.expect("no udp data received");
+        let addr = addr.expect("no udp addr received");
+        assert_eq!(read, 2);
+        assert_eq!(addr, local);
+        let got = unsafe { std::slice::from_raw_parts(buf.as_ptr() as *const u8, read) };
+        assert_eq!(got, b"hi");
+    }
+}

lib/virtual-net/src/lib.rs

@@ -802,6 +802,9 @@ pub enum NetworkError {
     /// Insufficient memory
     #[error("Insufficient memory")]
     InsufficientMemory,
+    /// The message is too large for the socket
+    #[error("message too large")]
+    MessageTooLarge,
     /// The connection was aborted
     #[error("connection aborted")]
     ConnectionAborted,
@@ -890,6 +893,7 @@ pub fn io_err_into_net_error(net_error: std::io::Error) -> NetworkError {
                     libc::ENODEV => NetworkError::NoDevice,
                     libc::EINVAL => NetworkError::InvalidInput,
                     libc::EPIPE => NetworkError::BrokenPipe,
+                    libc::EMSGSIZE => NetworkError::MessageTooLarge,
                     err => {
                         tracing::trace!("unknown os error {}", err);
                         NetworkError::UnknownError
@@ -920,6 +924,7 @@ pub fn net_error_into_io_err(net_error: NetworkError) -> std::io::Error {
         NetworkError::Interrupted => ErrorKind::Interrupted.into(),
         NetworkError::InvalidData => ErrorKind::InvalidData.into(),
         NetworkError::InvalidInput => ErrorKind::InvalidInput.into(),
+        NetworkError::MessageTooLarge => ErrorKind::InvalidInput.into(),
         NetworkError::NotConnected => ErrorKind::NotConnected.into(),
         NetworkError::NoDevice => ErrorKind::BrokenPipe.into(),
         NetworkError::PermissionDenied => ErrorKind::PermissionDenied.into(),