arcbox-vmnet 0.6.4

//! Pure L2 relay between a vmnet interface and a socketpair.
//!
//! When the `vmnet` feature is enabled, the bridge NIC uses vmnet.framework
//! directly instead of `VZNATNetworkDeviceAttachment`. This relay bridges
//! the vmnet read/write API (blocking) with the socketpair fd that the
//! `VZFileHandleNetworkDeviceAttachment` consumes.
//!
//! All DHCP, DNS, and ARP processing is handled by vmnet itself — the relay
//! is a transparent L2 pipe.

use std::os::fd::{AsRawFd, FromRawFd, OwnedFd};
use std::sync::Arc;

use tokio::io::Interest;
use tokio::io::unix::AsyncFd;
use tokio_util::sync::CancellationToken;

use crate::interface::Vmnet;

/// Maximum Ethernet frame size (jumbo frame capable).
const MAX_FRAME_SIZE: usize = 9216;

/// Bidirectional L2 relay between vmnet and a socketpair.
pub struct VmnetRelay {
    vmnet: Arc<Vmnet>,
    cancel: CancellationToken,
}

impl VmnetRelay {
    /// Creates a new relay.
    #[must_use]
    pub fn new(vmnet: Arc<Vmnet>, cancel: CancellationToken) -> Self {
        Self { vmnet, cancel }
    }

    /// Runs the relay until cancellation.
    ///
    /// `guest_fd` is one end of a `SOCK_DGRAM` socketpair. The other end is
    /// given to `VZFileHandleNetworkDeviceAttachment`.
    ///
    /// Two directions:
    /// - **vmnet → guest**: dedicated blocking thread (vmnet read is blocking)
    /// - **guest → vmnet**: async via `AsyncFd` on the socketpair
    ///
    /// # Errors
    ///
    /// Returns an error if the `AsyncFd` cannot be created.
    pub async fn run(self, guest_fd: OwnedFd) -> std::io::Result<()> {
        // The fd MUST be non-blocking: `AsyncFd` requires it, and a blocking
        // `read` here wedges the task inside the syscall — the select below
        // never regains control, cancellation cannot propagate to the
        // vmnet→guest worker, and the tokio blocking pool then waits forever
        // in `Runtime::drop`, hanging daemon shutdown. (The write side
        // already assumes non-blocking: it treats `WouldBlock` as guest
        // backpressure.) `dup` shares the file status flags, so this covers
        // the cloned writer fd too.
        let raw_fd = guest_fd.as_raw_fd();
        // SAFETY: fcntl F_GETFL/F_SETFL on a valid, owned fd.
        unsafe {
            let flags = libc::fcntl(raw_fd, libc::F_GETFL);
            if flags < 0 || libc::fcntl(raw_fd, libc::F_SETFL, flags | libc::O_NONBLOCK) < 0 {
                return Err(std::io::Error::last_os_error());
            }
        }

        // Clone the fd for the blocking thread (vmnet → guest direction).
        // SAFETY: dup is safe on a valid fd.
        let dup_fd = unsafe { libc::dup(raw_fd) };
        if dup_fd < 0 {
            return Err(std::io::Error::last_os_error());
        }
        // SAFETY: dup_fd is a valid fd from dup().
        let reader_fd: OwnedFd = unsafe { OwnedFd::from_raw_fd(dup_fd) };

        let async_fd = AsyncFd::new(guest_fd)?;

        // vmnet → guest: blocking thread
        let vmnet_read = Arc::clone(&self.vmnet);
        let cancel_read = self.cancel.clone();
        let mut vmnet_to_guest = tokio::task::spawn_blocking(move || {
            let mut buf = vec![0u8; MAX_FRAME_SIZE];
            loop {
                if cancel_read.is_cancelled() {
                    break;
                }
                match vmnet_read.read_packet(&mut buf) {
                    Ok(0) => {
                        // No data available, brief yield.
                        std::thread::sleep(std::time::Duration::from_millis(1));
                    }
                    Ok(n) => {
                        let fd = reader_fd.as_raw_fd();
                        // Retry EINTR (a signal interrupted the write before any
                        // bytes were sent); surface any other error to classify.
                        let write_err = loop {
                            // SAFETY: write to a valid socketpair fd with valid buffer.
                            let written =
                                unsafe { libc::write(fd, buf.as_ptr().cast::<libc::c_void>(), n) };
                            if written >= 0 {
                                break None;
                            }
                            let err = std::io::Error::last_os_error();
                            if err.kind() == std::io::ErrorKind::Interrupted {
                                continue;
                            }
                            break Some(err);
                        };
                        if let Some(err) = write_err {
                            if err.kind() == std::io::ErrorKind::BrokenPipe {
                                break;
                            }
                            // WouldBlock (EAGAIN) or ENOBUFS: the guest RX ring
                            // is momentarily full — drop this frame; the
                            // transport layer retransmits. Anything else is
                            // unexpected but non-fatal.
                            if err.kind() != std::io::ErrorKind::WouldBlock
                                && err.raw_os_error() != Some(libc::ENOBUFS)
                            {
                                tracing::debug!("vmnet→guest write error: {err}");
                            }
                        }
                    }
                    Err(e) => {
                        if cancel_read.is_cancelled() {
                            break;
                        }
                        tracing::debug!("vmnet read error: {e}");
                        std::thread::sleep(std::time::Duration::from_millis(1));
                    }
                }
            }
        });

        // guest → vmnet: async
        let vmnet_write = Arc::clone(&self.vmnet);
        let cancel_write = self.cancel.clone();
        let guest_to_vmnet = async move {
            let mut buf = vec![0u8; MAX_FRAME_SIZE];
            loop {
                tokio::select! {
                    () = cancel_write.cancelled() => break,
                    ready = async_fd.ready(Interest::READABLE) => {
                        let mut guard = match ready {
                            Ok(g) => g,
                            Err(e) => {
                                tracing::debug!("AsyncFd ready error: {e}");
                                break;
                            }
                        };

                        // Try to read from the socketpair.
                        let fd = async_fd.as_raw_fd();
                        // SAFETY: read from a valid socketpair fd with valid buffer.
                        let n = unsafe {
                            libc::read(
                                fd,
                                buf.as_mut_ptr().cast::<libc::c_void>(),
                                buf.len(),
                            )
                        };

                        match n.cmp(&0) {
                            std::cmp::Ordering::Greater => {
                                if let Err(e) = vmnet_write.write_packet(&buf[..n as usize]) {
                                    tracing::debug!("guest→vmnet write error: {e}");
                                }
                            }
                            std::cmp::Ordering::Equal => break, // Peer closed
                            std::cmp::Ordering::Less => {
                                let err = std::io::Error::last_os_error();
                                match err.kind() {
                                    std::io::ErrorKind::WouldBlock => guard.clear_ready(),
                                    // A signal interrupted the read before a
                                    // frame arrived (EINTR): retry on the next
                                    // iteration instead of tearing down the
                                    // whole bridge-NIC relay.
                                    std::io::ErrorKind::Interrupted => {}
                                    _ => {
                                        tracing::debug!("guest→vmnet read error: {err}");
                                        break;
                                    }
                                }
                            }
                        }
                    }
                }
            }
        };

        tokio::select! {
            () = self.cancel.cancelled() => {}
            _ = &mut vmnet_to_guest => {}
            () = guest_to_vmnet => {}
        }

        // Ensure the blocking thread exits regardless of which branch won.
        self.cancel.cancel();
        if let Err(e) = vmnet_to_guest.await {
            if e.is_panic() {
                tracing::error!("vmnet→guest blocking task panicked: {e}");
            } else {
                tracing::debug!("vmnet→guest blocking task join error: {e}");
            }
        }

        Ok(())
    }
}