gled-core 2.34.0

//! How many Art-Net packets the output threads send a second, fed made-up
//! frames as fast as they take them, against what the kernel takes from
//! plain sending loops on as many threads. No GPU and no engine: this
//! measures the network code alone. Run with
//!
//! ```sh
//! cargo test --release -p gled-core output_bench -- --ignored --nocapture
//! ```
//!
//! Packets are counted by the kernel (`UdpOutDatagrams` in /proc/net/snmp),
//! so the numbers include nothing gled could miscount. They go to 127.0.0.1,
//! where the test holds the Art-Net port and never reads it.

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

use gled_gpu::constants::{UNIVERSE_BUFFER_SIZE, UNIVERSE_SLOTS};
use gled_model::{
    asset::{
        Asset,
        output_device::{OutputDevice, routing::OutputRoutings},
    },
    asset_id::AssetId,
};
use kanal::bounded;

use super::{artdmx::PACKET_LEN, senders::sender_count, start};
use crate::{
    pipeline::extract_output::{OutputConfig, OutputFrame, OutputFrames},
    storage::collections::Collections,
};

const UNIVERSES: [u16; 5] = [256, 1024, 2048, 4096, UNIVERSE_SLOTS as u16];
const ARTNET_PORT: u16 = 6454;
const WARM_UP: Duration = Duration::from_millis(500);
const MEASURE: Duration = Duration::from_secs(3);

#[test]
#[ignore = "a measurement, not a test; sends Art-Net to 127.0.0.1"]
fn output_bench() {
    // Held, never read: gled sends from another port, and nothing echoes.
    let _sink = UdpSocket::bind((Ipv4Addr::UNSPECIFIED, ARTNET_PORT))
        .expect("port 6454 is free; close the running gled");
    let senders = sender_count();
    let kernel_one = kernel_rate(1);
    let kernel_all = kernel_rate(senders);

    let mut device = Asset::<OutputDevice>::new(vec!["bench node".to_owned()]);
    device.data = OutputDevice::Artnet {
        ip: Ipv4Addr::LOCALHOST.into(),
        universes: (0..UNIVERSE_SLOTS as u16).collect(),
    };
    let mut collections = Collections::default();
    collections
        .get_mut::<OutputDevice>()
        .set_asset(device.clone());
    collections.save();

    let (frames_sender, frames) = bounded(6);
    let (bell, wake) = bounded(1);
    let _inputs = start(OutputFrames {
        frames,
        wake,
        bell: bell.clone(),
    })
    .expect("the output starts");

    println!();
    println!(
        "kernel, plain sending loops: {kernel_one:.0} packets/s on 1 thread, {kernel_all:.0} on \
         {senders}"
    );
    println!("| universes | frames/s | packets/s | of the kernel on {senders} threads |");
    println!("|---:|---:|---:|---:|");
    for universes in UNIVERSES {
        let config = OutputConfig {
            universes: Arc::new((0..universes).collect()),
            routings: Arc::new(routings(device.id, universes)),
            color_orders: Arc::default(),
        };
        let data: Vec<u8> = (0..usize::from(universes) * UNIVERSE_BUFFER_SIZE as usize)
            .map(|index| (index % 251) as u8)
            .collect();
        // The output thread takes frames as fast as it sends them; a full
        // queue holds this loop back.
        let feed = |until: Instant| {
            let mut frames = 0u64;
            while Instant::now() < until {
                frames_sender
                    .send(OutputFrame {
                        data: data.clone(),
                        config: config.clone(),
                    })
                    .expect("the output takes frames");
                let _ = bell.try_send(());
                frames += 1;
            }
            frames
        };
        feed(Instant::now() + WARM_UP);
        let before = sent_datagrams();
        let start = Instant::now();
        let frames = feed(start + MEASURE);
        let seconds = start.elapsed().as_secs_f64();
        let packets = (sent_datagrams() - before) as f64 / seconds;
        println!(
            "| {universes} | {:.0} | {packets:.0} | {:.0} % |",
            frames as f64 / seconds,
            packets / kernel_all * 100.0
        );
    }
}

fn routings(device: AssetId<OutputDevice>, universes: u16) -> OutputRoutings {
    let mut routings = OutputRoutings::default();
    for universe in 0..universes {
        let routing = routings.universe_output_routing(universe);
        routing.device = Some(device);
        routing.universe = Some(universe);
    }
    routings
}

/// Packets a second `threads` threads sending ArtDmx-sized packets to the
/// sink get through, one `send_to` each, on one shared socket like gled's.
fn kernel_rate(threads: usize) -> f64 {
    let socket = Arc::new(UdpSocket::bind((Ipv4Addr::UNSPECIFIED, 0)).expect("a socket"));
    socket.set_nonblocking(true).expect("non-blocking");
    let target = SocketAddr::from((Ipv4Addr::LOCALHOST, ARTNET_PORT));
    let before = sent_datagrams();
    let start = Instant::now();
    let workers: Vec<_> = (0..threads)
        .map(|_| {
            let socket = socket.clone();
            thread::spawn(move || {
                let packet = [0x42u8; PACKET_LEN];
                while start.elapsed() < MEASURE {
                    for _ in 0..1000 {
                        let _ = socket.send_to(&packet, target);
                    }
                }
            })
        })
        .collect();
    for worker in workers {
        worker.join().expect("a sending loop");
    }
    (sent_datagrams() - before) as f64 / start.elapsed().as_secs_f64()
}

/// UDP datagrams this machine sent so far.
fn sent_datagrams() -> u64 {
    let snmp = std::fs::read_to_string("/proc/net/snmp").expect("/proc/net/snmp");
    let mut udp = snmp.lines().filter(|line| line.starts_with("Udp:"));
    let (Some(names), Some(values)) = (udp.next(), udp.next()) else {
        panic!("/proc/net/snmp has no UDP counters");
    };
    names
        .split_whitespace()
        .zip(values.split_whitespace())
        .find(|(name, _)| *name == "OutDatagrams")
        .and_then(|(_, value)| value.parse().ok())
        .expect("an OutDatagrams counter")
}