diff --git a/lib/virtual-io/src/guard.rs b/lib/virtual-io/src/guard.rs index 60d55960140b..9b258d096c45 100644 --- a/lib/virtual-io/src/guard.rs +++ b/lib/virtual-io/src/guard.rs @@ -72,20 +72,42 @@ pub enum HandlerGuardState { WakerMap(InterestGuard, InterestWakerMap), } +/// The full set of interests we register sources with. Includes priority +/// (EPOLLPRI, e.g. TCP out-of-band data) on platforms that support it. +pub fn all_interests() -> mio::Interest { + #[cfg(any(target_os = "linux", target_os = "android"))] + { + mio::Interest::READABLE | mio::Interest::WRITABLE | mio::Interest::PRIORITY + } + #[cfg(not(any(target_os = "linux", target_os = "android")))] + { + mio::Interest::READABLE | mio::Interest::WRITABLE + } +} + pub fn state_as_waker_map<'a>( state: &'a mut HandlerGuardState, selector: &'_ Arc, source: &'_ mut dyn mio::event::Source, +) -> io::Result<&'a mut InterestWakerMap> { + state_as_waker_map_with_interests( + state, + selector, + source, + mio::Interest::READABLE | mio::Interest::WRITABLE, + ) +} + +pub fn state_as_waker_map_with_interests<'a>( + state: &'a mut HandlerGuardState, + selector: &'_ Arc, + source: &'_ mut dyn mio::event::Source, + interests: mio::Interest, ) -> io::Result<&'a mut InterestWakerMap> { if !matches!(state, HandlerGuardState::WakerMap(_, _)) { let waker_map = InterestWakerMap::default(); *state = HandlerGuardState::WakerMap( - InterestGuard::new( - selector, - Box::new(waker_map.clone()), - source, - mio::Interest::READABLE | mio::Interest::WRITABLE, - )?, + InterestGuard::new(selector, Box::new(waker_map.clone()), source, interests)?, waker_map, ); } diff --git a/lib/virtual-io/src/interest.rs b/lib/virtual-io/src/interest.rs index 7cf0529cddb7..6abc2e9aa783 100644 --- a/lib/virtual-io/src/interest.rs +++ b/lib/virtual-io/src/interest.rs @@ -11,6 +11,8 @@ pub enum InterestType { Writable, Closed, Error, + // e.g. TCP out-of-band data + Priority, } #[derive(Debug)] diff --git a/lib/virtual-io/src/selector.rs b/lib/virtual-io/src/selector.rs index 858f80e9c63c..a737fa804e5f 100644 --- a/lib/virtual-io/src/selector.rs +++ b/lib/virtual-io/src/selector.rs @@ -50,6 +50,7 @@ impl SelectorModification { InterestType::Writable, InterestType::Closed, InterestType::Error, + InterestType::Priority, ]; for interest in interests { if last.has_interest(interest) && !handler.has_interest(interest) { @@ -277,6 +278,10 @@ impl Selector { tracing::trace!(token = ?token, interest = ?InterestType::Error, "host epoll"); handler.push_interest(InterestType::Error); } + if event.is_priority() { + tracing::trace!(token = ?token, interest = ?InterestType::Priority, "host epoll"); + handler.push_interest(InterestType::Priority); + } } } } diff --git a/lib/virtual-net/src/host.rs b/lib/virtual-net/src/host.rs index d39b5d79b9f9..50bdbb69002d 100644 --- a/lib/virtual-net/src/host.rs +++ b/lib/virtual-net/src/host.rs @@ -28,6 +28,7 @@ use tokio::runtime::Handle; use tracing::{debug, error, info, trace, warn}; use virtual_mio::{ HandlerGuardState, InterestGuard, InterestHandler, InterestType, Selector, state_as_waker_map, + state_as_waker_map_with_interests, }; /// Use the platform's maximum listen backlog where available so that @@ -719,6 +720,11 @@ impl VirtualConnectedSocket for LocalTcpStream { ret } + fn try_send_oob(&mut self, data: &[u8]) -> Result { + self.with_sock_ref(|s| s.send_out_of_band(data)) + .map_err(io_err_into_net_error) + } + fn try_flush(&mut self) -> Result<()> { self.stream.flush().map_err(io_err_into_net_error) } @@ -745,6 +751,13 @@ impl VirtualConnectedSocket for LocalTcpStream { } .map_err(io_err_into_net_error) } + + #[cfg(not(target_os = "windows"))] + fn try_recv_oob(&mut self, buf: &mut [MaybeUninit], peek: bool) -> Result { + let flags = libc::MSG_OOB | if peek { libc::MSG_PEEK } else { 0 }; + self.with_sock_ref(|s| s.recv_with_flags(buf, flags)) + .map_err(io_err_into_net_error) + } } impl VirtualSocket for LocalTcpStream { @@ -826,7 +839,7 @@ impl VirtualSocket for LocalTcpStream { &self.selector, handler, &mut self.stream, - mio::Interest::READABLE.add(mio::Interest::WRITABLE), + virtual_mio::all_interests(), ) .map_err(io_err_into_net_error)?; @@ -875,32 +888,76 @@ impl VirtualIoSource for LocalTcpStream { } let (state, selector, stream, buffer) = self.split_borrow(); - let map = state_as_waker_map(state, selector, stream).map_err(io_err_into_net_error)?; - map.pop(InterestType::Readable); + let map = state_as_waker_map_with_interests( + state, + selector, + stream, + virtual_mio::all_interests(), + ) + .map_err(io_err_into_net_error)?; + #[cfg(not(target_os = "windows"))] + let selector_readable = map.pop(InterestType::Readable); map.add(InterestType::Readable, cx.waker()); + map.add(InterestType::Closed, cx.waker()); - buffer.reserve(buffer.len() + 10240); - let uninit: &mut [MaybeUninit] = buffer.spare_capacity_mut(); - let uninit_unsafe: &mut [u8] = unsafe { std::mem::transmute(uninit) }; + #[cfg(not(target_os = "windows"))] + { + // Readiness checks must not consume stream data. In particular, + // pre-reading can cross TCP's urgent mark and discard OOB data. + if map.has_interest(InterestType::Closed) { + return Poll::Ready(Ok(0)); + } - match stream.read(uninit_unsafe) { - Ok(0) => Poll::Ready(Ok(0)), - Ok(amt) => { - unsafe { - buffer.set_len(buffer.len() + amt); + if selector_readable { + let available = + socket_bytes_available(stream.as_raw_fd()).map_err(io_err_into_net_error)?; + if available > 0 { + return Poll::Ready(Ok(available)); } - Poll::Ready(Ok(amt)) } - Err(err) if err.kind() == io::ErrorKind::ConnectionAborted => Poll::Ready(Ok(0)), - Err(err) if err.kind() == io::ErrorKind::ConnectionReset => Poll::Ready(Ok(0)), - Err(err) if err.kind() == io::ErrorKind::WouldBlock => Poll::Pending, - Err(err) => Poll::Ready(Err(io_err_into_net_error(err))), + + match libc_poll(stream.as_raw_fd(), libc::POLLIN | libc::POLLHUP) { + Some(events) if (events & libc::POLLHUP) != 0 => Poll::Ready(Ok(0)), + Some(events) if (events & libc::POLLIN) != 0 => { + let available = socket_bytes_available(stream.as_raw_fd()) + .map_err(io_err_into_net_error)?; + Poll::Ready(Ok(available)) + } + _ => Poll::Pending, + } + } + + #[cfg(target_os = "windows")] + { + buffer.reserve(buffer.len() + 10240); + let uninit: &mut [MaybeUninit] = buffer.spare_capacity_mut(); + let uninit_unsafe: &mut [u8] = unsafe { std::mem::transmute(uninit) }; + + match stream.read(uninit_unsafe) { + Ok(0) => Poll::Ready(Ok(0)), + Ok(amt) => { + unsafe { + buffer.set_len(buffer.len() + amt); + } + Poll::Ready(Ok(amt)) + } + Err(err) if err.kind() == io::ErrorKind::ConnectionAborted => Poll::Ready(Ok(0)), + Err(err) if err.kind() == io::ErrorKind::ConnectionReset => Poll::Ready(Ok(0)), + Err(err) if err.kind() == io::ErrorKind::WouldBlock => Poll::Pending, + Err(err) => Poll::Ready(Err(io_err_into_net_error(err))), + } } } fn poll_write_ready(&mut self, cx: &mut std::task::Context<'_>) -> Poll> { let (state, selector, stream, _) = self.split_borrow(); - let map = state_as_waker_map(state, selector, stream).map_err(io_err_into_net_error)?; + let map = state_as_waker_map_with_interests( + state, + selector, + stream, + virtual_mio::all_interests(), + ) + .map_err(io_err_into_net_error)?; #[cfg(not(target_os = "windows"))] map.pop(InterestType::Writable); map.add(InterestType::Writable, cx.waker()); @@ -929,6 +986,32 @@ impl VirtualIoSource for LocalTcpStream { Poll::Pending } + + fn poll_pri_ready(&mut self, cx: &mut std::task::Context<'_>) -> Poll> { + #[cfg(not(target_os = "windows"))] + { + let (state, selector, stream, _) = self.split_borrow(); + let map = state_as_waker_map_with_interests( + state, + selector, + stream, + virtual_mio::all_interests(), + ) + .map_err(io_err_into_net_error)?; + map.pop(InterestType::Priority); + map.add(InterestType::Priority, cx.waker()); + + if libc_poll(stream.as_raw_fd(), libc::POLLPRI) + .is_some_and(|events| (events & libc::POLLPRI) != 0) + { + return Poll::Ready(Ok(1)); + } + } + + #[cfg(target_os = "windows")] + let _ = cx; + Poll::Pending + } } #[cfg(not(target_os = "windows"))] @@ -946,6 +1029,154 @@ fn libc_poll(fd: RawFd, events: libc::c_short) -> Option { } } +#[cfg(not(target_os = "windows"))] +fn socket_bytes_available(fd: RawFd) -> io::Result { + let mut available: libc::c_int = 0; + let ret = unsafe { libc::ioctl(fd, libc::FIONREAD, &mut available) }; + if ret == -1 { + Err(io::Error::last_os_error()) + } else { + Ok(available.max(0) as usize) + } +} + +#[cfg(all(test, any(target_os = "linux", target_os = "android")))] +mod oob_tests { + use super::*; + use std::{ + io::Write, + net::{TcpListener, TcpStream}, + os::fd::AsRawFd, + thread, + }; + + fn connected_streams() -> (TcpStream, TcpStream) { + let listener = TcpListener::bind((Ipv4Addr::LOCALHOST, 0)).unwrap(); + let addr = listener.local_addr().unwrap(); + let sender = TcpStream::connect(addr).unwrap(); + let (receiver, _) = listener.accept().unwrap(); + sender.set_nodelay(true).unwrap(); + (sender, receiver) + } + + #[test] + fn send_oob_reaches_the_peer_as_urgent_data() { + let (sender, receiver) = connected_streams(); + let peer = sender.peer_addr().unwrap(); + sender.set_nonblocking(true).unwrap(); + let mut sender = + LocalTcpStream::new(Selector::new(), mio::net::TcpStream::from_std(sender), peer); + + assert_eq!(sender.try_send_oob(b"!").unwrap(), 1); + + let mut saw_urgent = false; + for _ in 0..100 { + if libc_poll(receiver.as_raw_fd(), libc::POLLPRI) + .is_some_and(|events| (events & libc::POLLPRI) != 0) + { + saw_urgent = true; + break; + } + thread::sleep(Duration::from_millis(10)); + } + assert!(saw_urgent, "the receiver never observed POLLPRI"); + + let mut byte = 0u8; + assert_eq!( + unsafe { + libc::recv( + receiver.as_raw_fd(), + (&mut byte as *mut u8).cast(), + 1, + libc::MSG_OOB, + ) + }, + 1 + ); + assert_eq!(byte, b'!'); + } + + #[test] + fn read_readiness_check_preserves_oob_and_peek_does_not_consume_it() { + let (mut sender, receiver) = connected_streams(); + let peer = sender.local_addr().unwrap(); + receiver.set_nonblocking(true).unwrap(); + let mut receiver = LocalTcpStream::new( + Selector::new(), + mio::net::TcpStream::from_std(receiver), + peer, + ); + sender.write_all(b"before").unwrap(); + assert_eq!( + unsafe { libc::send(sender.as_raw_fd(), b"!".as_ptr().cast(), 1, libc::MSG_OOB) }, + 1 + ); + + // Wait until the urgent mark has reached the receiver, rather than + // relying on the normal data and OOB byte arriving in one packet. + let mut saw_urgent = false; + for _ in 0..100 { + if libc_poll(receiver.stream.as_raw_fd(), libc::POLLPRI) + .is_some_and(|events| (events & libc::POLLPRI) != 0) + { + saw_urgent = true; + break; + } + thread::sleep(Duration::from_millis(10)); + } + assert!(saw_urgent, "the receiver never observed POLLPRI"); + + let waker = futures_util::task::noop_waker_ref(); + let mut cx = std::task::Context::from_waker(waker); + assert_eq!(receiver.poll_read_ready(&mut cx), Poll::Ready(Ok(6))); + + let mut oob = [MaybeUninit::uninit(); 1]; + assert_eq!(receiver.try_recv_oob(&mut oob, true).unwrap(), 1); + assert_eq!(unsafe { oob[0].assume_init() }, b'!'); + assert_eq!(receiver.try_recv_oob(&mut oob, false).unwrap(), 1); + assert_eq!(unsafe { oob[0].assume_init() }, b'!'); + } + + #[test] + fn priority_pending_before_registration_is_observed() { + let (mut sender, receiver) = connected_streams(); + let peer = sender.local_addr().unwrap(); + receiver.set_nonblocking(true).unwrap(); + sender.write_all(b"before").unwrap(); + assert_eq!( + unsafe { libc::send(sender.as_raw_fd(), b"!".as_ptr().cast(), 1, libc::MSG_OOB) }, + 1 + ); + + let mut saw_urgent = false; + for _ in 0..100 { + if libc_poll(receiver.as_raw_fd(), libc::POLLPRI) + .is_some_and(|events| (events & libc::POLLPRI) != 0) + { + saw_urgent = true; + break; + } + thread::sleep(Duration::from_millis(10)); + } + assert!(saw_urgent, "the receiver never observed POLLPRI"); + + // The selector was not registered while the urgent byte arrived, so + // the synchronous level check must observe it on the first poll. + let mut receiver = LocalTcpStream::new( + Selector::new(), + mio::net::TcpStream::from_std(receiver), + peer, + ); + let waker = futures_util::task::noop_waker_ref(); + let mut cx = std::task::Context::from_waker(waker); + assert_eq!(receiver.poll_pri_ready(&mut cx), Poll::Ready(Ok(1))); + + let mut oob = [MaybeUninit::uninit(); 1]; + assert_eq!(receiver.try_recv_oob(&mut oob, false).unwrap(), 1); + assert_eq!(unsafe { oob[0].assume_init() }, b'!'); + } +} + #[derive(Debug)] pub struct LocalUdpSocket { socket: mio::net::UdpSocket, diff --git a/lib/virtual-net/src/lib.rs b/lib/virtual-net/src/lib.rs index 410a4c36020d..2109770319d4 100644 --- a/lib/virtual-net/src/lib.rs +++ b/lib/virtual-net/src/lib.rs @@ -85,6 +85,14 @@ pub trait VirtualIoSource: fmt::Debug + Send + Sync + 'static { /// Polls the source to see if data can be sent fn poll_write_ready(&mut self, cx: &mut Context<'_>) -> Poll>; + + /// Polls the source for an exceptional/priority condition (e.g. TCP + /// out-of-band data). Sources without a notion of exceptional conditions + /// are simply never ready. + fn poll_pri_ready(&mut self, cx: &mut Context<'_>) -> Poll> { + let _ = cx; + Poll::Pending + } } /// An implementation of virtual networking @@ -365,6 +373,13 @@ pub trait VirtualConnectedSocket: VirtualSocket + fmt::Debug + Send + Sync + 'st /// Tries to send out a datagram or stream of bytes on this socket fn try_send(&mut self, data: &[u8]) -> Result; + /// Tries to send out-of-band (urgent) data on this socket. + /// Only meaningful for stream sockets whose transport supports it. + fn try_send_oob(&mut self, data: &[u8]) -> Result { + let _ = data; + Err(NetworkError::Unsupported) + } + // Tries to flush any data in the local buffers fn try_flush(&mut self) -> Result<()>; @@ -373,6 +388,13 @@ pub trait VirtualConnectedSocket: VirtualSocket + fmt::Debug + Send + Sync + 'st /// Tries to read a packet from the socket fn try_recv(&mut self, buf: &mut [MaybeUninit], peek: bool) -> Result; + + /// Tries to read pending out-of-band (urgent) data from the socket. + /// Only meaningful for stream sockets whose transport supports it. + fn try_recv_oob(&mut self, buf: &mut [MaybeUninit], peek: bool) -> Result { + let _ = (buf, peek); + Err(NetworkError::Unsupported) + } } #[async_trait::async_trait] diff --git a/lib/wasi-types/src/types.rs b/lib/wasi-types/src/types.rs index f84e0788cc11..4684f277ae21 100644 --- a/lib/wasi-types/src/types.rs +++ b/lib/wasi-types/src/types.rs @@ -282,8 +282,10 @@ pub mod net { pub const __WASI_SOCK_RECV_INPUT_WAITALL: RiFlags = 1 << 1; pub const __WASI_SOCK_RECV_INPUT_DATA_TRUNCATED: RiFlags = 1 << 2; pub const __WASI_SOCK_RECV_INPUT_DONT_WAIT: RiFlags = 1 << 3; + pub const __WASI_SOCK_RECV_INPUT_OOB: RiFlags = 1 << 4; pub const __WASI_SOCK_SEND_INPUT_DONT_WAIT: SiFlags = 1 << 0; + pub const __WASI_SOCK_SEND_INPUT_OOB: SiFlags = 1 << 1; pub const __WASI_SOCK_RECV_OUTPUT_DATA_TRUNCATED: RoFlags = 1 << 0; diff --git a/lib/wasi-types/src/wasi/bindings.rs b/lib/wasi-types/src/wasi/bindings.rs index 38e661476214..68185dc953a6 100644 --- a/lib/wasi-types/src/wasi/bindings.rs +++ b/lib/wasi-types/src/wasi/bindings.rs @@ -826,6 +826,9 @@ pub enum Eventtype { #[doc = " File descriptor `subscription_fd_readwrite::fd` has capacity"] #[doc = " available for writing. This event always triggers for regular files."] FdWrite, + #[doc = " File descriptor `subscription_fd_readwrite::fd` has an exceptional"] + #[doc = " condition pending (e.g. TCP out-of-band data)."] + FdExcept, #[doc = " Event type is unknown"] Unknown = 255, } @@ -835,6 +838,7 @@ impl core::fmt::Debug for Eventtype { Eventtype::Clock => f.debug_tuple("Eventtype::Clock").finish(), Eventtype::FdRead => f.debug_tuple("Eventtype::FdRead").finish(), Eventtype::FdWrite => f.debug_tuple("Eventtype::FdWrite").finish(), + Eventtype::FdExcept => f.debug_tuple("Eventtype::FdExcept").finish(), Eventtype::Unknown => f.debug_tuple("Eventtype::Unknown").finish(), } } @@ -3024,6 +3028,7 @@ unsafe impl wasmer::FromToNativeWasmType for Eventtype { 0 => Self::Clock, 1 => Self::FdRead, 2 => Self::FdWrite, + 3 => Self::FdExcept, q => { tracing::debug!("could not serialize number {q} to enum Eventtype"); diff --git a/lib/wasi-types/src/wasi/wasix_manual.rs b/lib/wasi-types/src/wasi/wasix_manual.rs index c91eaae29426..634ae3678326 100644 --- a/lib/wasi-types/src/wasi/wasix_manual.rs +++ b/lib/wasi-types/src/wasi/wasix_manual.rs @@ -87,6 +87,9 @@ impl From for Subscription { Eventtype::FdWrite => SubscriptionUnion { fd_readwrite: unsafe { other.u.fd_readwrite }, }, + Eventtype::FdExcept => SubscriptionUnion { + fd_readwrite: unsafe { other.u.fd_readwrite }, + }, Eventtype::Unknown => SubscriptionUnion { fd_readwrite: SubscriptionFsReadwrite { file_descriptor: u32::MAX, diff --git a/lib/wasix/src/fs/inode_guard.rs b/lib/wasix/src/fs/inode_guard.rs index f1a36d7d2701..522d91be3426 100644 --- a/lib/wasix/src/fs/inode_guard.rs +++ b/lib/wasix/src/fs/inode_guard.rs @@ -163,6 +163,7 @@ impl Future for InodeValFilePollGuardJoin { let waker = cx.waker(); let mut has_read = false; let mut has_write = false; + let mut has_pri = false; let mut has_close = false; let mut has_hangup = false; @@ -175,6 +176,9 @@ impl Future for InodeValFilePollGuardJoin { PollEvent::PollOut => { has_write = true; } + PollEvent::PollPri => { + has_pri = true; + } PollEvent::PollHangUp => { has_hangup = true; has_close = true; @@ -384,6 +388,54 @@ impl Future for InodeValFilePollGuardJoin { Poll::Pending => {} }; } + if has_pri { + // Only sockets have a notion of exceptional/priority conditions + // (e.g. TCP out-of-band data); everything else is never in an + // exceptional state. + let poll_result = match &mut self.mode { + InodeValFilePollGuardMode::Socket { inner } => { + let mut guard = inner.protected.write().unwrap(); + guard.poll_pri_ready(cx) + } + _ => Poll::Pending, + }; + if let Poll::Ready(bytes_available) = poll_result { + let mut error = Errno::Success; + let bytes_available = match bytes_available { + Ok(a) => a, + Err(e) => { + error = map_io_err(e); + 0 + } + }; + let inner = match self.subscription.type_ { + Eventtype::FdRead | Eventtype::FdWrite | Eventtype::FdExcept => { + Some(EventResultType::Fd(EventFdReadwrite { + nbytes: bytes_available as u64, + flags: Eventrwflags::empty(), + })) + } + Eventtype::Clock => Some(EventResultType::Clock(0)), + _ => None, + }; + if let Some(inner) = inner { + ret.push(( + EventResult { + userdata: self.subscription.userdata, + error, + type_: self.subscription.type_, + inner, + }, + if error == Errno::Success { + EpollType::EPOLLPRI + } else { + EpollType::EPOLLERR + }, + )) + .ok(); + } + } + } if !ret.is_empty() { return Poll::Ready(ret); } diff --git a/lib/wasix/src/net/socket.rs b/lib/wasix/src/net/socket.rs index f75b206e4c87..6342ee884cb6 100644 --- a/lib/wasix/src/net/socket.rs +++ b/lib/wasix/src/net/socket.rs @@ -1307,11 +1307,13 @@ impl InodeSocket { buf: &[u8], timeout: Option, nonblocking: bool, + oob: bool, ) -> Result { struct SocketSender<'a, 'b> { inner: &'a InodeSocketInner, data: &'b [u8], nonblocking: bool, + oob: bool, handler_registered: bool, } impl Drop for SocketSender<'_, '_> { @@ -1331,8 +1333,15 @@ impl InodeSocket { loop { let mut inner = self.inner.protected.write().unwrap(); let res = match &mut inner.kind { + InodeSocketKind::Raw(_) if self.oob => Err(NetworkError::Unsupported), InodeSocketKind::Raw(socket) => socket.try_send(self.data), + InodeSocketKind::TcpStream { socket, .. } if self.oob => { + socket.try_send_oob(self.data) + } InodeSocketKind::TcpStream { socket, .. } => socket.try_send(self.data), + InodeSocketKind::UdpSocket { .. } if self.oob => { + Err(NetworkError::Unsupported) + } InodeSocketKind::UdpSocket { socket, peer } => { if let Some(peer) = peer { socket.try_send_to(self.data, *peer) @@ -1343,6 +1352,9 @@ impl InodeSocket { InodeSocketKind::PreSocket { .. } => { return Poll::Ready(Err(Errno::Notconn)); } + InodeSocketKind::RemoteSocket { .. } if self.oob => { + return Poll::Ready(Err(Errno::Notsup)); + } InodeSocketKind::RemoteSocket { is_dead, .. } => { return match is_dead { true => Poll::Ready(Err(Errno::Connreset)), @@ -1375,6 +1387,7 @@ impl InodeSocket { inner: &self.inner, data: buf, nonblocking, + oob, handler_registered: false, }; if let Some(timeout) = timeout { @@ -1479,12 +1492,14 @@ impl InodeSocket { timeout: Option, nonblocking: bool, peek: bool, + oob: bool, ) -> Result { struct SocketReceiver<'a, 'b> { inner: &'a InodeSocketInner, data: &'b mut [MaybeUninit], nonblocking: bool, peek: bool, + oob: bool, handler_registered: bool, } impl Drop for SocketReceiver<'_, '_> { @@ -1505,10 +1520,21 @@ impl InodeSocket { let peek = self.peek; let mut inner = self.inner.protected.write().unwrap(); let res = match &mut inner.kind { + InodeSocketKind::Raw(socket) if self.oob => { + let _ = socket; + Err(NetworkError::Unsupported) + } InodeSocketKind::Raw(socket) => socket.try_recv(self.data, peek), + InodeSocketKind::TcpStream { socket, .. } if self.oob => { + socket.try_recv_oob(self.data, peek) + } InodeSocketKind::TcpStream { socket, .. } => { socket.try_recv(self.data, peek) } + InodeSocketKind::UdpSocket { socket, .. } if self.oob => { + let _ = socket; + Err(NetworkError::Unsupported) + } InodeSocketKind::UdpSocket { socket, peer } => match peer { Some(peer) => { try_recv_from_connected_udp(socket.as_mut(), self.data, peek, peer) @@ -1553,6 +1579,7 @@ impl InodeSocket { data: buf, nonblocking, peek, + oob, handler_registered: false, }; if let Some(timeout) = timeout { @@ -1572,12 +1599,14 @@ impl InodeSocket { timeout: Option, nonblocking: bool, peek: bool, + oob: bool, ) -> Result<(usize, SocketAddr), Errno> { struct SocketReceiver<'a, 'b> { inner: &'a InodeSocketInner, data: &'b mut [MaybeUninit], nonblocking: bool, peek: bool, + oob: bool, handler_registered: bool, } impl Drop for SocketReceiver<'_, '_> { @@ -1598,7 +1627,19 @@ impl InodeSocket { let mut inner = self.inner.protected.write().unwrap(); loop { let res = match &mut inner.kind { + InodeSocketKind::Icmp(_) if self.oob => Err(NetworkError::Unsupported), InodeSocketKind::Icmp(socket) => socket.try_recv_from(self.data, peek), + InodeSocketKind::TcpStream { socket, .. } => { + let received = if self.oob { + socket.try_recv_oob(self.data, peek) + } else { + socket.try_recv(self.data, peek) + }; + received.and_then(|amt| socket.addr_peer().map(|addr| (amt, addr))) + } + InodeSocketKind::UdpSocket { .. } if self.oob => { + Err(NetworkError::Unsupported) + } InodeSocketKind::UdpSocket { socket, peer } => match peer { Some(peer) => { try_recv_from_connected_udp(socket.as_mut(), self.data, peek, peer) @@ -1642,6 +1683,7 @@ impl InodeSocket { data: buf, nonblocking, peek, + oob, handler_registered: false, }; if let Some(timeout) = timeout { @@ -1813,6 +1855,17 @@ impl InodeSocketProtected { .map_err(net_error_into_io_err) } + pub fn poll_pri_ready(&mut self, cx: &mut Context<'_>) -> Poll> { + match &mut self.kind { + // Only connected TCP streams can carry an exceptional condition + // (TCP out-of-band data); all other socket kinds are never in an + // exceptional state. + InodeSocketKind::TcpStream { socket, .. } => socket.poll_pri_ready(cx), + _ => Poll::Pending, + } + .map_err(net_error_into_io_err) + } + pub fn set_handler( &mut self, handler: Box, diff --git a/lib/wasix/src/os/epoll/mod.rs b/lib/wasix/src/os/epoll/mod.rs index a0731d6a1eb7..04e69ab502c7 100644 --- a/lib/wasix/src/os/epoll/mod.rs +++ b/lib/wasix/src/os/epoll/mod.rs @@ -65,6 +65,7 @@ const READABLE_BIT: u8 = 1 << 0; const WRITABLE_BIT: u8 = 1 << 1; const HUP_BIT: u8 = 1 << 2; const ERR_BIT: u8 = 1 << 3; +const PRI_BIT: u8 = 1 << 4; static EPOLL_ENQUEUE_ATTEMPTS: AtomicU64 = AtomicU64::new(0); static EPOLL_ENQUEUE_DEDUPE_HITS: AtomicU64 = AtomicU64::new(0); @@ -389,6 +390,8 @@ pub(crate) fn epoll_type_to_pending_bit(readiness: EpollType) -> Option { Some(HUP_BIT) } else if readiness == EpollType::EPOLLERR { Some(ERR_BIT) + } else if readiness == EpollType::EPOLLPRI { + Some(PRI_BIT) } else { None } @@ -401,6 +404,7 @@ fn interest_to_pending_bit(interest: InterestType) -> u8 { InterestType::Writable => WRITABLE_BIT, InterestType::Closed => HUP_BIT, InterestType::Error => ERR_BIT, + InterestType::Priority => PRI_BIT, } } @@ -419,6 +423,12 @@ fn pending_bits_to_event(bits: u8, mask: EpollType) -> EpollType { { event |= EpollType::EPOLLOUT; } + if let Some(bit) = epoll_type_to_pending_bit(EpollType::EPOLLPRI) + && (bits & bit) != 0 + && mask.contains(EpollType::EPOLLPRI) + { + event |= EpollType::EPOLLPRI; + } if let Some(bit) = epoll_type_to_pending_bit(EpollType::EPOLLHUP) && (bits & bit) != 0 { @@ -597,6 +607,9 @@ pub(crate) fn register_epoll_handler( type_ = Eventtype::FdRead; peb = peb.add(PollEvent::PollIn); } + if event.events().contains(EpollType::EPOLLPRI) { + peb = peb.add(PollEvent::PollPri); + } // EPOLLERR/EPOLLHUP are always delivered by epoll regardless of requested mask. peb = peb.add(PollEvent::PollError); peb = peb.add(PollEvent::PollHangUp); diff --git a/lib/wasix/src/state/types.rs b/lib/wasix/src/state/types.rs index 24f15a3e62da..40f56410c3b5 100644 --- a/lib/wasix/src/state/types.rs +++ b/lib/wasix/src/state/types.rs @@ -23,6 +23,8 @@ pub enum PollEvent { PollHangUp = 8, /// Invalid request. ignored as input PollInvalid = 16, + /// Exceptional condition pending (e.g. TCP out-of-band data) + PollPri = 32, } impl PollEvent { @@ -33,6 +35,7 @@ impl PollEvent { 4 => PollEvent::PollError, 8 => PollEvent::PollHangUp, 16 => PollEvent::PollInvalid, + 32 => PollEvent::PollPri, _ => return None, }) } diff --git a/lib/wasix/src/syscalls/wasi/fd_read.rs b/lib/wasix/src/syscalls/wasi/fd_read.rs index f5de216676dc..77fcd3955cc1 100644 --- a/lib/wasix/src/syscalls/wasi/fd_read.rs +++ b/lib/wasix/src/syscalls/wasi/fd_read.rs @@ -270,6 +270,7 @@ pub(crate) fn fd_read_internal( Some(timeout), nonblocking, false, + false, ) .await?; total_read += local_read; diff --git a/lib/wasix/src/syscalls/wasi/fd_write.rs b/lib/wasix/src/syscalls/wasi/fd_write.rs index 4fcfbf883b76..20d0b0908755 100644 --- a/lib/wasix/src/syscalls/wasi/fd_write.rs +++ b/lib/wasix/src/syscalls/wasi/fd_write.rs @@ -289,7 +289,13 @@ pub(crate) fn fd_write_internal( if socket.is_dgram() { let data = data.coalesce(&memory, MAX_SOCKET_PAYLOAD)?; sent += socket - .send(tasks.deref(), data.as_ref(), Some(timeout), nonblocking) + .send( + tasks.deref(), + data.as_ref(), + Some(timeout), + nonblocking, + false, + ) .await?; return Ok(sent); } @@ -311,6 +317,7 @@ pub(crate) fn fd_write_internal( buf.as_ref(), Some(timeout), nonblocking, + false, ) .await { @@ -326,7 +333,13 @@ pub(crate) fn fd_write_internal( } FdWriteSource::Buffer(data) => { sent += socket - .send(tasks.deref(), data.as_ref(), Some(timeout), nonblocking) + .send( + tasks.deref(), + data.as_ref(), + Some(timeout), + nonblocking, + false, + ) .await?; } } diff --git a/lib/wasix/src/syscalls/wasi/poll_oneoff.rs b/lib/wasix/src/syscalls/wasi/poll_oneoff.rs index 73a96d60ad5e..0242ca226b78 100644 --- a/lib/wasix/src/syscalls/wasi/poll_oneoff.rs +++ b/lib/wasix/src/syscalls/wasi/poll_oneoff.rs @@ -277,6 +277,12 @@ where *peb |= (PollEvent::PollOut as PollEventSet); file_descriptor } + Eventtype::FdExcept => { + let file_descriptor = unsafe { s.data.fd_readwrite.file_descriptor }; + *fd = Some(file_descriptor); + *peb |= (PollEvent::PollPri as PollEventSet); + file_descriptor + } Eventtype::Clock => { let clock_info = unsafe { s.data.clock }; if clock_info.clock_id == Clockid::Realtime diff --git a/lib/wasix/src/syscalls/wasix/sock_recv.rs b/lib/wasix/src/syscalls/wasix/sock_recv.rs index 78bfc428ecf2..19c34fc85bae 100644 --- a/lib/wasix/src/syscalls/wasix/sock_recv.rs +++ b/lib/wasix/src/syscalls/wasix/sock_recv.rs @@ -115,6 +115,7 @@ pub(super) fn sock_recv_internal( let memory = unsafe { env.memory_view(ctx) }; let peek = (ri_flags & __WASI_SOCK_RECV_INPUT_PEEK) != 0; let nonblocking_flag = (ri_flags & __WASI_SOCK_RECV_INPUT_DONT_WAIT) != 0; + let oob = (ri_flags & __WASI_SOCK_RECV_INPUT_OOB) != 0; let data = wasi_try_ok_ok!(__sock_asyncify( env, sock, @@ -148,6 +149,7 @@ pub(super) fn sock_recv_internal( Some(timeout), nonblocking, peek, + oob, ) .await { diff --git a/lib/wasix/src/syscalls/wasix/sock_recv_from.rs b/lib/wasix/src/syscalls/wasix/sock_recv_from.rs index b5b4fb2f3b07..cd85f6b4e106 100644 --- a/lib/wasix/src/syscalls/wasix/sock_recv_from.rs +++ b/lib/wasix/src/syscalls/wasix/sock_recv_from.rs @@ -53,6 +53,7 @@ pub(super) fn sock_recv_from_internal( ) -> Result { let peek = (ri_flags & __WASI_SOCK_RECV_INPUT_PEEK) != 0; let nonblocking_flag = (ri_flags & __WASI_SOCK_RECV_INPUT_DONT_WAIT) != 0; + let oob = (ri_flags & __WASI_SOCK_RECV_INPUT_OOB) != 0; let mut env = ctx.data(); // Check rights first to preserve error precedence @@ -89,6 +90,7 @@ pub(super) fn sock_recv_from_internal( Some(timeout), nonblocking, peek, + oob, ) .await }, @@ -125,6 +127,7 @@ pub(super) fn sock_recv_from_internal( Some(timeout), nonblocking, peek, + oob, ) .await .map(|(amt, addr)| { diff --git a/lib/wasix/src/syscalls/wasix/sock_send.rs b/lib/wasix/src/syscalls/wasix/sock_send.rs index fab371b8f10e..bda584df98c9 100644 --- a/lib/wasix/src/syscalls/wasix/sock_send.rs +++ b/lib/wasix/src/syscalls/wasix/sock_send.rs @@ -93,6 +93,7 @@ pub(crate) fn sock_send_internal( let runtime = env.runtime.clone(); let nonblocking_flag = (si_flags & __WASI_SOCK_SEND_INPUT_DONT_WAIT) != 0; + let oob = (si_flags & __WASI_SOCK_SEND_INPUT_OOB) != 0; let bytes_written = wasi_try_ok_ok!(__sock_asyncify( env, @@ -106,7 +107,7 @@ pub(crate) fn sock_send_internal( .flatten() .unwrap_or(Duration::from_secs(30)); - if socket.is_dgram() { + if socket.is_dgram() || oob { let data = si_data.coalesce(&memory, MAX_SOCKET_PAYLOAD)?; return socket .send( @@ -114,6 +115,7 @@ pub(crate) fn sock_send_internal( data.as_ref(), Some(timeout), nonblocking, + oob, ) .await; } @@ -136,6 +138,7 @@ pub(crate) fn sock_send_internal( buf.as_ref(), Some(timeout), nonblocking, + false, ) .await { @@ -157,6 +160,7 @@ pub(crate) fn sock_send_internal( data.as_ref(), Some(timeout), nonblocking, + false, ) .await } diff --git a/lib/wasix/src/syscalls/wasix/sock_send_file.rs b/lib/wasix/src/syscalls/wasix/sock_send_file.rs index 57910c9d020a..108f4d171a68 100644 --- a/lib/wasix/src/syscalls/wasix/sock_send_file.rs +++ b/lib/wasix/src/syscalls/wasix/sock_send_file.rs @@ -148,6 +148,7 @@ pub(crate) fn sock_send_file_internal( Some(read_timeout), false, false, + false, ) .await .map(|amt| { @@ -234,7 +235,7 @@ pub(crate) fn sock_send_file_internal( .flatten() .unwrap_or(Duration::from_secs(30)); socket - .send(tasks.deref(), &data, Some(write_timeout), true) + .send(tasks.deref(), &data, Some(write_timeout), true, false) .await }, ));