gled-core 2.34.0

//! Sending a frame's Art-Net packets: one system call per packet, spread
//! over a few threads once a frame has many universes, all on gled's one
//! Art-Net socket, so every packet still leaves from the Art-Net port.
//!
//! Each sender thread has two batches: while it sends one, the output thread
//! fills the other. A frame waits for a sender only when that sender still
//! has both, which holds the frames back at the GPU readback instead of
//! queueing them.

use std::{
    io,
    net::{SocketAddr, UdpSocket},
    sync::Arc,
    thread,
    time::{Duration, Instant},
};

use anyhow::{Context, Result};
use kanal::{Receiver, Sender, bounded};
use tracing::{trace, warn};

use super::artdmx::{self, PACKET_LEN};

/// Packets one sender thread takes of a frame before the next one helps. One
/// thread sends about 700 000 packets a second on a laptop.
const PACKETS_PER_SENDER: usize = 128;
/// Sender threads at most. Eight send about 2.8 million packets a second on
/// a laptop's loopback, over 90 % of what the kernel takes from as many
/// plain sending loops (`bench.rs`) and more than 10 Gbit/s Ethernet carries.
const MAX_SENDERS: usize = 8;
const BATCHES_PER_SENDER: usize = 2;
/// A full send buffer is retried this often before the packet is given up.
const SEND_TRIES: u32 = 10;
/// How often packets the network did not take are reported.
const DROP_REPORT_INTERVAL: Duration = Duration::from_secs(10);

/// ArtDmx packets back to back, and where each goes.
#[derive(Debug, Default)]
pub(super) struct Batch {
    bytes: Vec<u8>,
    addrs: Vec<SocketAddr>,
}

impl Batch {
    /// Adds the packet sending `data` to `universe` at `addr`. A universe
    /// Art-Net cannot address is left out.
    pub(super) fn push(&mut self, addr: SocketAddr, universe: u16, data: &[u8; 512]) {
        if artdmx::write(&mut self.bytes, universe, data) {
            self.addrs.push(addr);
        } else {
            trace!("Universe {universe} has no Art-Net port address");
        }
    }

    fn packets(&self) -> impl Iterator<Item = (&[u8], &SocketAddr)> {
        self.bytes
            .as_chunks::<PACKET_LEN>()
            .0
            .iter()
            .map(|packet| packet.as_slice())
            .zip(&self.addrs)
    }

    fn clear(&mut self) {
        self.bytes.clear();
        self.addrs.clear();
    }
}

/// One sender thread: the batches to send, and the sent ones to fill again.
#[derive(Debug)]
struct Lane {
    work: Sender<Batch>,
    free: Receiver<Batch>,
}

#[derive(Debug)]
pub(super) struct Senders {
    lanes: Vec<Lane>,
}

impl Senders {
    /// Starts the sender threads on `socket`.
    pub(super) fn start(socket: &Arc<UdpSocket>) -> Result<Self> {
        let lanes = (0..sender_count())
            .map(|index| {
                let (work, batches) = bounded(BATCHES_PER_SENDER);
                let (sent, free) = bounded(BATCHES_PER_SENDER);
                for _ in 0..BATCHES_PER_SENDER {
                    sent.try_send(Batch::default())
                        .context("Could not prepare an Art-Net batch")?;
                }
                let socket = socket.clone();
                thread::Builder::new()
                    .name(format!("gled:output:tx{index}"))
                    .spawn(move || send_batches(&socket, &batches, &sent))
                    .context("Could not spawn an Art-Net sender thread")?;
                Ok(Lane { work, free })
            })
            .collect::<Result<_>>()?;
        Ok(Self { lanes })
    }

    /// Empty batches for a frame of `packets` packets, one per sender it
    /// takes. Waits for a sender that still sends its last two.
    pub(super) fn take(&self, packets: usize) -> Vec<Batch> {
        let count = lanes_for(packets, self.lanes.len());
        self.lanes[..count]
            .iter()
            .map(|lane| lane.free.recv().unwrap_or_default())
            .collect()
    }

    /// Hands every batch to its sender.
    pub(super) fn send(&self, batches: Vec<Batch>) {
        for (lane, batch) in self.lanes.iter().zip(batches) {
            if lane.work.send(batch).is_err() {
                warn!("An Art-Net sender thread is gone");
            }
        }
    }
}

/// How many sender threads run: half the cores, at most [`MAX_SENDERS`], so
/// rendering keeps cores of its own.
pub(super) fn sender_count() -> usize {
    thread::available_parallelism()
        .map_or(1, |cores| cores.get() / 2)
        .clamp(1, MAX_SENDERS)
}

/// How many of `senders` senders a frame of `packets` packets takes.
pub(super) fn lanes_for(packets: usize, senders: usize) -> usize {
    packets
        .div_ceil(PACKETS_PER_SENDER)
        .clamp(1, senders.max(1))
}

/// The batch of `batches` that the `index`-th of `packets` packets goes into.
pub(super) fn batch_for(index: usize, packets: usize, batches: usize) -> usize {
    (index * batches / packets.max(1)).min(batches.saturating_sub(1))
}

fn send_batches(socket: &UdpSocket, batches: &Receiver<Batch>, sent: &Sender<Batch>) {
    #[cfg(feature = "profiling")]
    profiling::register_thread!("output:tx");

    let mut dropped = Dropped::default();
    while let Ok(mut batch) = batches.recv() {
        let mut bytes = 0;
        for (packet, addr) in batch.packets() {
            bytes += send(socket, packet, *addr, &mut dropped);
        }
        crate::network_stats::add_outgoing_bytes(bytes);
        batch.clear();
        if sent.send(batch).is_err() {
            return;
        }
    }
}

/// Sends one packet, retrying while the send buffer is full. Returns the
/// bytes sent.
pub(super) fn send(
    socket: &UdpSocket,
    packet: &[u8],
    addr: SocketAddr,
    dropped: &mut Dropped,
) -> usize {
    for attempt in 1..=SEND_TRIES {
        match socket.send_to(packet, addr) {
            Ok(bytes) => return bytes,
            Err(err) if attempt == SEND_TRIES => dropped.add(addr, &err),
            // Yield, not sleep: a sleep rounds up to about 50 µs on Linux,
            // which throttled the output to about 5000 packets a second.
            Err(_) => thread::yield_now(),
        }
    }
    0
}

/// Art-Net packets given up on. At hundreds of frames a second a full send
/// buffer drops many per second; one warning per interval says the same.
#[derive(Debug, Default)]
pub(super) struct Dropped {
    /// Packets dropped since the last warning.
    count: u64,
    /// When the last warning was logged.
    reported: Option<Instant>,
}

impl Dropped {
    fn add(&mut self, addr: SocketAddr, err: &io::Error) {
        self.count += 1;
        if self
            .reported
            .is_none_or(|reported| reported.elapsed() >= DROP_REPORT_INTERVAL)
        {
            warn!(
                "Could not send Art-Net to {addr}: {err}; {} packets dropped since the last \
                 warning",
                self.count
            );
            self.count = 0;
            self.reported = Some(Instant::now());
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn a_small_frame_takes_one_sender_and_a_large_one_all() {
        assert_eq!(lanes_for(0, 4), 1);
        assert_eq!(lanes_for(PACKETS_PER_SENDER, 4), 1);
        assert_eq!(lanes_for(PACKETS_PER_SENDER + 1, 4), 2);
        assert_eq!(lanes_for(4096, 4), 4);
        assert_eq!(lanes_for(4096, 0), 1);
    }

    #[test]
    fn packets_spread_evenly_over_the_batches_in_order() {
        let batches: Vec<usize> = (0..8).map(|index| batch_for(index, 8, 3)).collect();
        assert_eq!(batches, [0, 0, 0, 1, 1, 1, 2, 2]);
        // Packets past the estimate go into the last batch.
        assert_eq!(batch_for(9, 8, 3), 2);
        assert_eq!(batch_for(0, 0, 1), 0);
    }

    #[test]
    fn a_batch_sends_each_packet_to_its_address() {
        let mut batch = Batch::default();
        let first: SocketAddr = "10.0.0.1:6454".parse().expect("an address");
        let second: SocketAddr = "10.0.0.2:6454".parse().expect("an address");
        batch.push(first, 1, &[1; 512]);
        batch.push(second, 0x8000, &[2; 512]);
        batch.push(second, 2, &[3; 512]);

        let packets: Vec<_> = batch.packets().collect();

        assert_eq!(packets.len(), 2);
        assert_eq!((packets[0].1, packets[0].0[18]), (&first, 1));
        assert_eq!((packets[1].1, packets[1].0[18]), (&second, 3));
    }
}