ferranet 0.1.0

//! Tokio-based asynchronous datalink channel (feature `tokio`).
//!
//! [`AsyncSender`] and [`AsyncReceiver`] mirror the synchronous [`Sender`](crate::channel::Sender)
//! / [`Receiver`](crate::channel::Receiver), but await readiness on the Tokio reactor instead of
//! blocking in `poll`. Readiness is tracked with [`tokio::io::unix::AsyncFd`] registered on the
//! channel's file descriptor; the actual zero-copy reads/writes go straight through the backend.

use std::io;
use std::os::fd::{AsFd, AsRawFd, RawFd};

use tokio::io::Interest;
use tokio::io::unix::AsyncFd;

use crate::block::{Block, Frame, FrameMeta, PacketType};
use crate::channel::{ChannelBuilder, RxBackend, TxBackend};
use crate::error::Result;
use crate::sys::Stats;

/// A bare file descriptor wrapper so [`AsyncFd`] can register interest without owning (and later
/// closing) the descriptor — ownership stays with the backend.
struct FdHolder(RawFd);

impl AsRawFd for FdHolder {
    fn as_raw_fd(&self) -> RawFd {
        self.0
    }
}

impl ChannelBuilder {
    /// Opens the channel for asynchronous use on the current Tokio runtime.
    ///
    /// Must be called from within a Tokio runtime context (the file descriptors are registered
    /// with the reactor on construction).
    pub fn build_async(&self) -> Result<(AsyncSender, AsyncReceiver)> {
        let parts = self.open()?;
        let tx = AsyncSender::new(parts.tx)?;
        let rx = AsyncReceiver::new(parts.rx)?;
        Ok((tx, rx))
    }

    /// Opens `count` async receivers load-balanced across a `PACKET_FANOUT` group for multi-core
    /// RX. Each [`AsyncReceiver`] is registered with the current Tokio runtime; spawn a task per
    /// receiver. Fanout is receive-only.
    pub fn build_fanout_rx_async(&self, count: usize) -> Result<Vec<AsyncReceiver>> {
        self.build_fanout_backends(count)?.into_iter().map(AsyncReceiver::new).collect()
    }
}

/// The asynchronous transmit half of a datalink channel.
pub struct AsyncSender {
    // `readiness` is declared first so it deregisters from the reactor before `backend` closes
    // the underlying fd on drop.
    readiness: AsyncFd<FdHolder>,
    backend: TxBackend,
}

impl AsyncSender {
    fn new(backend: TxBackend) -> Result<Self> {
        let fd = backend.borrow_fd().as_raw_fd();
        let readiness = AsyncFd::with_interest(FdHolder(fd), Interest::WRITABLE)
            .map_err(crate::Error::Io)?;
        Ok(AsyncSender { readiness, backend })
    }

    /// Sends a single raw frame, awaiting transmit capacity if the ring/queue is full.
    pub async fn send(&mut self, frame: &[u8]) -> Result<()> {
        if let Some(max) = self.backend.max_frame_len() {
            if frame.len() > max {
                return Err(crate::Error::FrameTooLarge { len: frame.len(), max });
            }
        }
        loop {
            match self.backend.try_send(frame) {
                Ok(_) => return Ok(()),
                Err(e) if e.kind() == io::ErrorKind::WouldBlock => {
                    let mut guard = self.readiness.writable().await.map_err(crate::Error::Io)?;
                    guard.clear_ready();
                }
                Err(e) => return Err(e.into()),
            }
        }
    }

    /// Sends a batch of frames, awaiting capacity if needed, returning how many were accepted.
    pub async fn send_batch(&mut self, frames: &[&[u8]]) -> Result<usize> {
        loop {
            match self.backend.try_send_batch(frames) {
                Ok(0) => {
                    let mut guard = self.readiness.writable().await.map_err(crate::Error::Io)?;
                    guard.clear_ready();
                }
                Ok(n) => return Ok(n),
                Err(e) if e.kind() == io::ErrorKind::WouldBlock => {
                    let mut guard = self.readiness.writable().await.map_err(crate::Error::Io)?;
                    guard.clear_ready();
                }
                Err(e) => return Err(e.into()),
            }
        }
    }

    /// Borrows the underlying file descriptor.
    pub fn as_fd(&self) -> std::os::fd::BorrowedFd<'_> {
        self.backend.borrow_fd()
    }
}

/// The asynchronous receive half of a datalink channel.
pub struct AsyncReceiver {
    readiness: AsyncFd<FdHolder>,
    backend: RxBackend,
    stats_total: Stats,
}

impl AsyncReceiver {
    fn new(backend: RxBackend) -> Result<Self> {
        let fd = backend.borrow_fd().as_raw_fd();
        let readiness = AsyncFd::with_interest(FdHolder(fd), Interest::READABLE)
            .map_err(crate::Error::Io)?;
        Ok(AsyncReceiver { readiness, backend, stats_total: Stats::default() })
    }

    /// Receives the next batch of frames, awaiting until at least one is available.
    ///
    /// The returned [`Block`] borrows zero-copy from kernel memory (ring backend) or an internal
    /// buffer (basic backend); it must be dropped before the next call.
    pub async fn recv_block(&mut self) -> Result<Block<'_>> {
        match &self.backend {
            RxBackend::Ring(_) => {
                // Wait until the current block is owned by user space, then borrow it.
                loop {
                    if let RxBackend::Ring(ring) = &self.backend {
                        if ring.block_ready() {
                            break;
                        }
                    }
                    let mut guard = self.readiness.readable().await.map_err(crate::Error::Io)?;
                    // If readiness was spurious (block still not closed), clear so epoll re-arms.
                    let ready = matches!(&self.backend, RxBackend::Ring(r) if r.block_ready());
                    if !ready {
                        guard.clear_ready();
                    }
                }
                match &mut self.backend {
                    RxBackend::Ring(ring) => Ok(ring.recv_block().expect("block reported ready")),
                    _ => unreachable!("backend kind is stable"),
                }
            }
            RxBackend::Basic { .. } => {
                // Read once we are readable; loop only over the wait, then build the frame.
                let (n, pkttype) = loop {
                    let attempt = match &mut self.backend {
                        RxBackend::Basic { sock, buf } => sock.recv_into(buf),
                        _ => unreachable!("backend kind is stable"),
                    };
                    match attempt {
                        Ok(v) => break v,
                        Err(e) if e.kind() == io::ErrorKind::WouldBlock => {
                            let mut g = self.readiness.readable().await.map_err(crate::Error::Io)?;
                            g.clear_ready();
                        }
                        Err(e) => return Err(e.into()),
                    }
                };
                let RxBackend::Basic { buf, .. } = &self.backend else {
                    unreachable!("backend kind is stable")
                };
                let captured = n.min(buf.len());
                let meta = FrameMeta {
                    wire_len: n,
                    timestamp: None,
                    vlan: None,
                    packet_type: PacketType::from_raw(pkttype),
                };
                Ok(Block::single(Some(Frame::new(&buf[..captured], meta))))
            }
            RxBackend::Dummy { .. } => {
                // Await an inject (signalled via the pipe), then deliver the queued frame.
                let frame = loop {
                    let popped = match &mut self.backend {
                        RxBackend::Dummy { queue, .. } => {
                            queue.lock().expect("dummy queue poisoned").pop_front()
                        }
                        _ => unreachable!("backend kind is stable"),
                    };
                    match popped {
                        Some(Ok(v)) => break v,
                        Some(Err(e)) => return Err(e.into()),
                        None => {
                            let mut g = self.readiness.readable().await.map_err(crate::Error::Io)?;
                            let eof = match &self.backend {
                                RxBackend::Dummy { readiness, .. } => {
                                    crate::dummy::drain_readiness(readiness.as_fd())
                                }
                                _ => false,
                            };
                            g.clear_ready();
                            let empty = match &self.backend {
                                RxBackend::Dummy { queue, .. } => {
                                    queue.lock().expect("dummy queue poisoned").is_empty()
                                }
                                _ => true,
                            };
                            if eof && empty {
                                return Err(io::Error::from(io::ErrorKind::UnexpectedEof).into());
                            }
                        }
                    }
                };
                let RxBackend::Dummy { last, .. } = &mut self.backend else {
                    unreachable!("backend kind is stable")
                };
                *last = frame;
                let meta = FrameMeta {
                    wire_len: last.len(),
                    timestamp: None,
                    vlan: None,
                    packet_type: crate::channel::dummy_packet_type(last),
                };
                Ok(Block::single(Some(Frame::new(&last[..], meta))))
            }
        }
    }

    /// Returns cumulative receive/drop statistics for this channel since it was opened.
    ///
    /// See [`Receiver::stats`](crate::Receiver::stats); reading also clears the kernel
    /// `TP_STATUS_LOSING` flag observed via [`Block::is_losing`](crate::Block::is_losing).
    pub fn stats(&mut self) -> Result<Stats> {
        let delta = match &mut self.backend {
            RxBackend::Ring(ring) => ring.stats()?,
            RxBackend::Basic { sock, .. } => sock.statistics()?,
            RxBackend::Dummy { .. } => Stats::default(),
        };
        self.stats_total.accumulate(delta);
        Ok(self.stats_total)
    }

    /// Enables or disables promiscuous mode on the interface.
    pub fn set_promiscuous(&self, on: bool) -> Result<()> {
        match &self.backend {
            RxBackend::Ring(ring) => ring.set_promiscuous(on),
            RxBackend::Basic { sock, .. } => sock.set_promiscuous(on),
            RxBackend::Dummy { .. } => Ok(()),
        }
    }

    /// Borrows the underlying file descriptor.
    pub fn as_fd(&self) -> std::os::fd::BorrowedFd<'_> {
        self.backend.borrow_fd()
    }
}