webrtc 0.20.0-rc.4

//! Async Runtime Abstraction
//!
//! This module provides the [`Runtime`] trait, which abstracts all asynchronous operations
//! and primitives required by the WebRTC stack. This makes the `webrtc` crate runtime-agnostic,
//! allowing it to support multiple async runtimes through feature flags.
//!
//! # Active Runtime
//!
//! The active runtime is selected at compile time via Cargo features:
//! *   **`runtime-tokio` (default)**: Uses the Tokio runtime.
//! *   **`runtime-smol`**: Uses the smol runtime.
//!
//! This module exports concrete type aliases (e.g., [`Mutex`], [`Sender`], [`Receiver`], [`Interval`])
//! which map to the selected runtime's primitives, ensuring zero-cost abstraction without
//! dynamic dispatch in the hot path.

#![allow(clippy::type_complexity)]

use std::sync::atomic::{AtomicUsize, Ordering};
use std::{fmt::Debug, future::Future, io, net::SocketAddr, pin::Pin, sync::Arc, time::Duration};

/// Process-global override for the shared reactor pool's thread count, set via
/// [`set_reactor_pool_size`]. `0` means "unset": fall back to the
/// `WEBRTC_REACTOR_POOL_SIZE` environment variable, then to host parallelism.
static REACTOR_POOL_SIZE: AtomicUsize = AtomicUsize::new(0);

/// Upper clamp on the reactor-pool thread count, guarding against a fat-fingered
/// `WEBRTC_REACTOR_POOL_SIZE` / [`set_reactor_pool_size`] value eagerly allocating a
/// slot table large enough to OOM (or overflow). Far above any sane reactor count.
pub(crate) const MAX_REACTOR_POOL_SIZE: usize = 1024;

/// Set the size of the shared reactor-thread pool used by
/// [`Runtime::spawn_reactor`] (i.e. by connections built with
/// [`with_dedicated_reactor_thread(true)`](crate::peer_connection::PeerConnectionBuilder::with_dedicated_reactor_thread)).
///
/// The pool is process-global and sized **once**, lazily, on first use. Call this
/// before building the first dedicated-reactor `PeerConnection`; later calls (or
/// calls after the pool has been created) have no effect. `0` restores the
/// default resolution (env var, then host parallelism).
///
/// Prefer this or the `WEBRTC_REACTOR_POOL_SIZE` env var for a global default;
/// [`PeerConnectionBuilder::with_reactor_pool_size`](crate::peer_connection::PeerConnectionBuilder::with_reactor_pool_size)
/// is a convenience that forwards here at build time.
pub fn set_reactor_pool_size(size: usize) {
    REACTOR_POOL_SIZE.store(size, Ordering::Relaxed);
}

/// Resolve the reactor-pool size, in precedence order: an explicit override (via
/// [`set_reactor_pool_size`]), then the `WEBRTC_REACTOR_POOL_SIZE` env var, then
/// host parallelism (`available_parallelism`, falling back to 4). Read once by
/// each runtime when it lazily builds its pool.
pub(crate) fn reactor_pool_size() -> usize {
    let override_size = REACTOR_POOL_SIZE.load(Ordering::Relaxed);
    let resolved = if override_size != 0 {
        override_size
    } else {
        std::env::var("WEBRTC_REACTOR_POOL_SIZE")
            .ok()
            .and_then(|v| v.parse::<usize>().ok())
            .filter(|&n| n != 0)
            .unwrap_or_else(|| {
                std::thread::available_parallelism()
                    .map(|n| n.get())
                    .unwrap_or(4)
            })
    };
    resolved.clamp(1, MAX_REACTOR_POOL_SIZE)
}

/// Handle to a spawned task that can be used to manage its lifecycle
pub struct JoinHandle {
    inner: Box<dyn JoinHandleInner>,
}

impl JoinHandle {
    /// Abort the spawned task
    pub fn abort(&self) {
        self.inner.abort();
    }

    /// Check if the task is finished
    pub fn is_finished(&self) -> bool {
        self.inner.is_finished()
    }
}

impl Drop for JoinHandle {
    fn drop(&mut self) {
        self.inner.detach();
    }
}

trait JoinHandleInner: Send + Sync {
    /// Detach the task so it keeps running independently after the handle is dropped.
    fn detach(&self);
    /// Cancel the task cooperatively.
    fn abort(&self);
    fn is_finished(&self) -> bool;
}

/// Abstracts I/O and timer operations for runtime independence
///
/// This trait allows the WebRTC implementation to work with different async runtimes
/// without being tightly coupled to any specific runtime.
pub trait Runtime: Send + Sync + Debug + 'static {
    /// Drive a future to completion in the background
    ///
    /// The future must complete to `()` and will be spawned as a background task.
    /// Returns a handle that can be used to abort or inspect the task.
    /// Dropping the handle detaches the task; the task keeps running until it
    /// completes or the runtime is shut down. Call `.abort()` to cancel explicitly.
    #[track_caller]
    fn spawn(&self, future: Pin<Box<dyn Future<Output = ()> + Send>>) -> JoinHandle;

    /// Drive `future` to completion on a **shared, bounded pool** of
    /// single-threaded reactors, pinned to one pool thread for its lifetime.
    ///
    /// The tokio and smol implementations keep a process-global pool of at most
    /// `N` dedicated OS threads (each hosting its own single-threaded runtime),
    /// created lazily and sized by [`reactor_pool_size`]. Each `future` is
    /// assigned to one pool thread round-robin and never migrates off it, so the
    /// async runtime never moves a peer-connection driver across a shared worker
    /// pool — the dominant cost for in-process data-channel throughput (issue
    /// #101) — while the thread (and per-thread allocator arena) count stays
    /// bounded by `N` regardless of connection count, instead of one OS thread
    /// per connection. The socket wrapping and the whole event loop run inside
    /// `future`, on the pool thread's runtime, so I/O resources bind to it.
    ///
    /// `future` runs as an abortable *task* on its pool thread; the returned
    /// [`JoinHandle`] aborts that task (not a whole thread). Up to a few drivers
    /// cooperatively share a pool thread; they are I/O-bound and yield at await
    /// points, so they interleave without blocking one another.
    ///
    /// This is thread confinement, not CPU-core affinity: the OS scheduler may
    /// still move a pool thread between cores. Pinning pool threads to cores (via
    /// `core_affinity`) is a planned follow-up (issue #101).
    ///
    /// The default implementation falls back to [`Runtime::spawn`] on the ambient
    /// runtime, so custom runtimes keep working (without the confinement benefit).
    fn spawn_reactor(&self, future: Pin<Box<dyn Future<Output = ()> + Send>>) -> JoinHandle {
        self.spawn(future)
    }

    /// Create an async UDP socket from a standard socket
    ///
    /// The socket should be bound and configured before being wrapped.
    fn wrap_udp_socket(&self, socket: std::net::UdpSocket) -> io::Result<Arc<dyn AsyncUdpSocket>>;

    /// Create an async TCP listener from a standard listener
    ///
    /// The listener should be bound and configured before being wrapped.
    fn wrap_tcp_listener(
        &self,
        listener: std::net::TcpListener,
    ) -> io::Result<Arc<dyn AsyncTcpListener>>;

    /// Connect to a remote TCP address.
    fn connect_tcp<'a>(
        &'a self,
        remote_addr: SocketAddr,
    ) -> Pin<Box<dyn Future<Output = io::Result<Arc<dyn AsyncTcpStream>>> + Send + 'a>>;
}

/// Outcome of a batched UDP receive ([`AsyncUdpSocket::recv_gro`]).
///
/// `buf[..len]` holds one or more datagrams received in a single syscall. When the
/// kernel coalesced consecutive same-flow datagrams via UDP GRO, each is `stride`
/// bytes except possibly the last — walk `buf[..len]` in `stride`-sized steps to
/// recover the individual datagrams. Without GRO (or for a lone datagram)
/// `stride == len` and there is exactly one datagram. Every datagram in the batch
/// shares `peer_addr` (GRO only coalesces a single source flow).
#[derive(Debug, Clone, Copy)]
pub struct GroRecv {
    /// Total bytes written to the buffer across all coalesced datagrams.
    pub len: usize,
    /// Size of each datagram in the batch; the final one may be shorter. Always
    /// `>= 1` when `len > 0`.
    pub stride: usize,
    /// Source address shared by every datagram in the batch.
    pub peer_addr: SocketAddr,
}

/// Abstract implementation of a UDP socket for runtime independence
///
/// Simple async wrapper around UDP sockets
pub trait AsyncUdpSocket: Send + Sync + Debug + 'static {
    /// Send data to the specified address
    fn send_to<'a>(
        &'a self,
        buf: &'a [u8],
        target: SocketAddr,
    ) -> Pin<Box<dyn Future<Output = io::Result<usize>> + Send + 'a>>;

    /// Receive a datagram from the socket
    fn recv_from<'a>(
        &'a self,
        buf: &'a mut [u8],
    ) -> Pin<Box<dyn Future<Output = io::Result<(usize, SocketAddr)>> + Send + 'a>>;

    /// Get the local address this socket is bound to
    fn local_addr(&self) -> io::Result<SocketAddr>;

    /// Maximum number of segments a single [`send_segments`](Self::send_segments)
    /// call can emit in one syscall via UDP GSO. Returns `1` when GSO is
    /// unavailable (each segment then costs one syscall).
    fn max_gso_segments(&self) -> usize {
        1
    }

    /// Maximum number of datagrams the kernel may coalesce into one
    /// [`recv_gro`](Self::recv_gro) via UDP GRO. Returns `1` when GRO is
    /// unavailable. Used to size receive buffers.
    fn max_gro_segments(&self) -> usize {
        1
    }

    /// Send `buf` as consecutive datagrams of `segment_size` bytes to `target`
    /// using a single UDP GSO (`UDP_SEGMENT`) syscall — the final datagram may be
    /// shorter than `segment_size`. `ecn`, when `Some`, stamps the ECN codepoint
    /// bits on every segment. Returns the number of payload bytes accepted.
    ///
    /// The default implementation falls back to a loop of [`send_to`](Self::send_to),
    /// so implementors without GSO support (and external impls) need not override it.
    /// Callers should only batch (`buf` spanning more than one segment) when
    /// [`max_gso_segments`](Self::max_gso_segments) reports `> 1`.
    fn send_segments<'a>(
        &'a self,
        buf: &'a [u8],
        segment_size: usize,
        target: SocketAddr,
        ecn: Option<u8>,
    ) -> Pin<Box<dyn Future<Output = io::Result<usize>> + Send + 'a>> {
        Box::pin(async move {
            let _ = ecn;
            // `segment_size == 0` means "no segmentation" — send the whole buffer as a
            // single datagram rather than shredding it into 1-byte sends.
            let step = if segment_size == 0 {
                buf.len().max(1)
            } else {
                segment_size
            };
            let mut sent = 0;
            for chunk in buf.chunks(step) {
                sent += self.send_to(chunk, target).await?;
            }
            Ok(sent)
        })
    }

    /// Receive one or more datagrams into `buf` in a single syscall, using UDP GRO
    /// to coalesce consecutive same-flow datagrams when available. See [`GroRecv`]
    /// for how to split the buffer back into individual datagrams.
    ///
    /// The default implementation receives a single datagram (`stride == len`), so
    /// implementors without GRO support (and external impls) need not override it.
    fn recv_gro<'a>(
        &'a self,
        buf: &'a mut [u8],
    ) -> Pin<Box<dyn Future<Output = io::Result<GroRecv>> + Send + 'a>> {
        Box::pin(async move {
            let (len, peer_addr) = self.recv_from(buf).await?;
            Ok(GroRecv {
                len,
                stride: if len == 0 { 1 } else { len },
                peer_addr,
            })
        })
    }
}

/// Abstract implementation of a TCP listener for runtime independence.
pub trait AsyncTcpListener: Send + Sync + Debug + 'static {
    /// Accept a new TCP stream.
    fn accept<'a>(
        &'a self,
    ) -> Pin<Box<dyn Future<Output = io::Result<(Arc<dyn AsyncTcpStream>, SocketAddr)>> + Send + 'a>>;

    /// Get the local address this listener is bound to.
    fn local_addr(&self) -> io::Result<SocketAddr>;
}

/// Abstract implementation of a TCP stream for runtime independence.
pub trait AsyncTcpStream: Send + Sync + Debug + 'static {
    /// Read bytes from the stream.
    fn read<'a, 'b>(
        &'a self,
        buf: &'b mut [u8],
    ) -> Pin<Box<dyn Future<Output = io::Result<usize>> + Send + 'b>>
    where
        'a: 'b;

    /// Write all bytes to the stream.
    fn write_all<'a, 'b>(
        &'a self,
        buf: &'b [u8],
    ) -> Pin<Box<dyn Future<Output = io::Result<()>> + Send + 'b>>
    where
        'a: 'b;

    /// Get the local address of the stream.
    fn local_addr(&self) -> io::Result<SocketAddr>;

    /// Get the peer address of the stream.
    fn peer_addr(&self) -> io::Result<SocketAddr>;
}

/// An async mutex that works across different runtimes
pub trait AsyncMutex<T: ?Sized>: Send + Sync {
    /// The guard type returned by lock()
    type Guard<'a>: std::ops::Deref<Target = T> + std::ops::DerefMut + Send + 'a
    where
        Self: 'a,
        T: 'a;

    /// Lock the mutex asynchronously
    fn lock(&self) -> Pin<Box<dyn Future<Output = Self::Guard<'_>> + Send + '_>>;
}

/// An async notification primitive
pub trait AsyncNotify: Send + Sync {
    /// Notify one waiting task
    fn notify_one(&self);

    /// Notify all waiting tasks
    fn notify_waiters(&self);

    /// Wait for a notification
    fn notified(&self) -> Pin<Box<dyn Future<Output = ()> + Send + '_>>;
}

/// Sender half of an async channel
pub trait AsyncSender<T>: Send + Sync {
    /// Send a value, waiting if the channel is full
    fn send(&self, value: T)
    -> Pin<Box<dyn Future<Output = Result<(), SendError<T>>> + Send + '_>>;

    /// Try to send a value without blocking
    fn try_send(&self, value: T) -> Result<(), TrySendError<T>>;
}

/// Receiver half of an async channel
pub trait AsyncReceiver<T>: Send {
    /// Receive a value, waiting if the channel is empty
    fn recv(&mut self) -> Pin<Box<dyn Future<Output = Option<T>> + Send + '_>>;

    /// Try to receive a value without blocking
    fn try_recv(&mut self) -> Result<T, TryRecvError>;
}

/// Error returned when send fails
#[derive(Debug)]
pub struct SendError<T>(pub T);

/// Error returned when try_send fails
#[derive(Debug)]
pub enum TrySendError<T> {
    /// The channel is full.
    Full(T),
    /// The channel is disconnected.
    Disconnected(T),
}

/// Error returned when try_recv fails
#[derive(Debug)]
pub enum TryRecvError {
    /// The channel is empty.
    Empty,
    /// The channel is disconnected.
    Disconnected,
}

impl<T> std::fmt::Display for SendError<T> {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "channel disconnected")
    }
}

impl<T: std::fmt::Debug> std::error::Error for SendError<T> {}

/// Error returned when a broadcast send fails (no receivers)
#[derive(Debug)]
pub struct BroadcastSendError<T>(pub T);

/// Error returned when a broadcast receive fails
#[derive(Debug)]
pub enum BroadcastRecvError {
    /// Channel closed, no more senders
    Closed,
    /// Receiver lagged behind; this many messages were skipped
    Lagged(u64),
}

impl<T> std::fmt::Display for BroadcastSendError<T> {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "broadcast send failed: no receivers")
    }
}

impl<T: std::fmt::Debug> std::error::Error for BroadcastSendError<T> {}

impl std::fmt::Display for BroadcastRecvError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            BroadcastRecvError::Closed => write!(f, "broadcast channel closed"),
            BroadcastRecvError::Lagged(n) => write!(f, "broadcast receiver lagged by {n}"),
        }
    }
}

impl std::error::Error for BroadcastRecvError {}

/// Get the default runtime for the current build configuration
///
/// Returns the runtime for whichever runtime feature is enabled.
/// If multiple runtimes are enabled, tokio takes precedence.
#[cfg(any(feature = "runtime-tokio", feature = "runtime-smol"))]
pub fn default_runtime() -> Option<std::sync::Arc<dyn Runtime>> {
    #[cfg(feature = "runtime-tokio")]
    {
        Some(std::sync::Arc::new(TokioRuntime))
    }

    #[cfg(all(not(feature = "runtime-tokio"), feature = "runtime-smol"))]
    {
        Some(std::sync::Arc::new(smol::SmolRuntime))
    }
}

#[cfg(not(any(feature = "runtime-tokio", feature = "runtime-smol")))]
pub fn default_runtime() -> Option<std::sync::Arc<dyn Runtime>> {
    None
}

/// Get smol runtime if enabled
#[cfg(any(feature = "runtime-tokio", feature = "runtime-smol"))]
pub fn smol_runtime() -> Option<std::sync::Arc<dyn Runtime>> {
    #[cfg(feature = "runtime-smol")]
    {
        Some(std::sync::Arc::new(smol::SmolRuntime))
    }

    #[cfg(not(feature = "runtime-smol"))]
    None
}

// Runtime implementations
#[cfg(feature = "runtime-tokio")]
mod tokio;
#[cfg(feature = "runtime-tokio")]
pub use tokio::TokioRuntime;
#[cfg(feature = "runtime-tokio")]
pub use tokio::{
    TokioInterval, block_on, broadcast_channel, channel, interval, resolve_host, sleep, timeout,
    yield_now,
};
/// The concrete Interval type for the active runtime.
#[cfg(feature = "runtime-tokio")]
pub type Interval = TokioInterval;
/// The concrete Mutex type for the active runtime.
#[cfg(feature = "runtime-tokio")]
pub type Mutex<T> = tokio::TokioMutex<T>;
/// The concrete Notify type for the active runtime.
#[cfg(feature = "runtime-tokio")]
pub type Notify = tokio::TokioNotify;
/// The concrete channel Sender type for the active runtime.
#[cfg(feature = "runtime-tokio")]
pub type Sender<T> = tokio::TokioSender<T>;
/// The concrete channel Receiver type for the active runtime.
#[cfg(feature = "runtime-tokio")]
pub type Receiver<T> = tokio::TokioReceiver<T>;
/// The concrete broadcast channel Sender type for the active runtime.
#[cfg(feature = "runtime-tokio")]
pub type BroadcastSender<T> = tokio::TokioBroadcastSender<T>;
/// The concrete broadcast channel Receiver type for the active runtime.
#[cfg(feature = "runtime-tokio")]
pub type BroadcastReceiver<T> = tokio::TokioBroadcastReceiver<T>;

#[cfg(feature = "runtime-smol")]
mod smol;

#[cfg(all(not(feature = "runtime-tokio"), feature = "runtime-smol"))]
pub use smol::SmolRuntime;
#[cfg(all(not(feature = "runtime-tokio"), feature = "runtime-smol"))]
pub use smol::{
    SmolInterval, block_on, broadcast_channel, channel, interval, resolve_host, sleep, timeout,
    yield_now,
};
/// The concrete Interval type for the active runtime.
#[cfg(all(not(feature = "runtime-tokio"), feature = "runtime-smol"))]
pub type Interval = SmolInterval;
/// The concrete Mutex type for the active runtime.
#[cfg(all(not(feature = "runtime-tokio"), feature = "runtime-smol"))]
pub type Mutex<T> = smol::SmolMutex<T>;
/// The concrete Notify type for the active runtime.
#[cfg(all(not(feature = "runtime-tokio"), feature = "runtime-smol"))]
pub type Notify = smol::SmolNotify;
/// The concrete channel Sender type for the active runtime.
#[cfg(all(not(feature = "runtime-tokio"), feature = "runtime-smol"))]
pub type Sender<T> = smol::SmolSender<T>;
/// The concrete channel Receiver type for the active runtime.
#[cfg(all(not(feature = "runtime-tokio"), feature = "runtime-smol"))]
pub type Receiver<T> = smol::SmolReceiver<T>;
/// The concrete broadcast channel Sender type for the active runtime.
#[cfg(all(not(feature = "runtime-tokio"), feature = "runtime-smol"))]
pub type BroadcastSender<T> = smol::SmolBroadcastSender<T>;
/// The concrete broadcast channel Receiver type for the active runtime.
#[cfg(all(not(feature = "runtime-tokio"), feature = "runtime-smol"))]
pub type BroadcastReceiver<T> = smol::SmolBroadcastReceiver<T>;

#[cfg(test)]
mod default_impl_tests {
    //! Cover the `AsyncUdpSocket` DEFAULT method bodies (send_segments / recv_gro /
    //! max_gso_segments / max_gro_segments). The concrete tokio/smol impls override
    //! them, so nothing else exercises the defaults — a minimal fake that implements
    //! only the required methods does.
    use super::*;
    use std::sync::Mutex;

    #[derive(Debug, Default)]
    struct FakeUdp {
        sent: Mutex<Vec<Vec<u8>>>,
        to_recv: Mutex<Vec<u8>>,
    }

    impl AsyncUdpSocket for FakeUdp {
        fn send_to<'a>(
            &'a self,
            buf: &'a [u8],
            _target: SocketAddr,
        ) -> Pin<Box<dyn Future<Output = io::Result<usize>> + Send + 'a>> {
            Box::pin(async move {
                self.sent.lock().unwrap().push(buf.to_vec());
                Ok(buf.len())
            })
        }
        fn recv_from<'a>(
            &'a self,
            buf: &'a mut [u8],
        ) -> Pin<Box<dyn Future<Output = io::Result<(usize, SocketAddr)>> + Send + 'a>> {
            Box::pin(async move {
                let data = self.to_recv.lock().unwrap();
                let n = data.len().min(buf.len());
                buf[..n].copy_from_slice(&data[..n]);
                Ok((n, "127.0.0.1:9".parse::<SocketAddr>().unwrap()))
            })
        }
        fn local_addr(&self) -> io::Result<SocketAddr> {
            Ok("127.0.0.1:0".parse::<SocketAddr>().unwrap())
        }
    }

    fn addr() -> SocketAddr {
        "127.0.0.1:5".parse::<SocketAddr>().unwrap()
    }

    #[test]
    fn default_caps_are_one() {
        let s = FakeUdp::default();
        assert_eq!(s.max_gso_segments(), 1);
        assert_eq!(s.max_gro_segments(), 1);
    }

    #[test]
    fn default_send_segments_loops_send_to() {
        let s = FakeUdp::default();
        // 11 bytes, segment_size 3 -> datagrams of 3,3,3,2.
        let buf = [1u8, 1, 1, 2, 2, 2, 3, 3, 3, 4, 4];
        let sent = futures::executor::block_on(s.send_segments(&buf, 3, addr(), None)).unwrap();
        assert_eq!(sent, 11);
        let calls = s.sent.lock().unwrap();
        assert_eq!(calls.len(), 4);
        assert_eq!(calls[0], vec![1, 1, 1]);
        assert_eq!(calls[3], vec![4, 4]);
    }

    #[test]
    fn default_send_segments_zero_size_is_one_datagram() {
        let s = FakeUdp::default();
        let buf = [7u8; 10];
        futures::executor::block_on(s.send_segments(&buf, 0, addr(), Some(2))).unwrap();
        let calls = s.sent.lock().unwrap();
        assert_eq!(
            calls.len(),
            1,
            "segment_size 0 must send one datagram, not shred"
        );
        assert_eq!(calls[0].len(), 10);
    }

    #[test]
    fn default_recv_gro_is_single_datagram() {
        let s = FakeUdp::default();
        *s.to_recv.lock().unwrap() = vec![9, 9, 9, 9, 9];
        let mut buf = [0u8; 32];
        let gro = futures::executor::block_on(s.recv_gro(&mut buf)).unwrap();
        assert_eq!(gro.len, 5);
        assert_eq!(
            gro.stride, 5,
            "stride == len for a single (non-GRO) datagram"
        );
        assert_eq!(gro.peer_addr, "127.0.0.1:9".parse::<SocketAddr>().unwrap());
    }
}