diff --git a/README.md b/README.md index 31ebadd..22daf8a 100644 --- a/README.md +++ b/README.md @@ -140,6 +140,7 @@ As of writing the following features are supported: * `alpha` (disabled by default): Respect alpha values during `PX` commands. Disabled by default as this can cause performance degradation. * `binary-set-pixel` (disabled by default): Allows use of the `PB` command. * `binary-sync-pixels`(disabled by default): Allows use of the `PXMULTI` command. +* `prometheus` (enabled by default): Enables the Prometheus metrics exporter. Can be disabled to compile for 32-bit targets. To e.g. turn the VNC server off, build with diff --git a/breakwater-parser/src/framebuffer/mod.rs b/breakwater-parser/src/framebuffer/mod.rs index bff496f..23324ba 100644 --- a/breakwater-parser/src/framebuffer/mod.rs +++ b/breakwater-parser/src/framebuffer/mod.rs @@ -31,6 +31,9 @@ pub trait FrameBuffer { /// make sure x and y are in bounds unsafe fn get_unchecked(&self, x: usize, y: usize) -> u32; + /// RGBA is always in this order within the byte. + /// However, the memory order is reversed for performance reasons (so native u32’s are laid out in memory). + /// Big endian in memory: RGBA, little endian in memory: ABGR fn set(&self, x: usize, y: usize, rgba: u32); /// We can *not* take an `&[u32]` for the pixel here, as `std::slice::from_raw_parts` requires the data to be @@ -38,6 +41,7 @@ pub trait FrameBuffer { /// treat the pixels as raw bytes. /// /// Returns the coordinates where we landed after filling + // FIXME: Broken on big-endian #[inline(always)] fn set_multi(&self, start_x: usize, start_y: usize, pixels: &[u8]) -> (usize, usize) { let starting_index = start_x + start_y * self.get_width(); diff --git a/breakwater-parser/src/original.rs b/breakwater-parser/src/original.rs index 06de549..a34aff8 100644 --- a/breakwater-parser/src/original.rs +++ b/breakwater-parser/src/original.rs @@ -93,8 +93,9 @@ impl Parser for OriginalParser { } while i < loop_end { + // BUG: byte order is inverted on big-endian let current_command = - unsafe { (buffer.as_ptr().add(i) as *const u64).read_unaligned() }; + unsafe { (buffer.as_ptr().add(i) as *const u64).read_unaligned() }.to_le(); if current_command & 0x00ff_ffff == PX_PATTERN { i += 3; @@ -189,7 +190,7 @@ impl Parser for OriginalParser { // We don't want to return the actual (absolute) coordinates, the client should also get the result offseted x - self.connection_x_offset, y - self.connection_y_offset, - rgb.to_be() >> 8 + rgb.swap_bytes() >> 8 ) .as_bytes(), ); @@ -201,7 +202,7 @@ impl Parser for OriginalParser { #[cfg(feature = "binary-set-pixel")] if current_command & 0x0000_ffff == PB_PATTERN { let command_bytes = - unsafe { (buffer.as_ptr().add(i + 2) as *const u64).read_unaligned() }; + unsafe { (buffer.as_ptr().add(i + 2) as *const u64).read_unaligned() }.to_le(); let x = u16::from_le((command_bytes) as u16); let y = u16::from_le((command_bytes >> 16) as u16); @@ -217,7 +218,7 @@ impl Parser for OriginalParser { #[cfg(feature = "binary-sync-pixels")] if current_command & 0x00ff_ffff_ffff_ffff == PXMULTI_PATTERN { i += "PXMULTI".len(); - let header = unsafe { (buffer.as_ptr().add(i) as *const u64).read_unaligned() }; + let header = unsafe { (buffer.as_ptr().add(i) as *const u64).read_unaligned() }.to_le(); i += 8; let start_x = u16::from_le((header) as u16); @@ -356,7 +357,7 @@ pub(crate) fn simd_unhex(value: *const u8) -> u32 { #[inline(always)] fn parse_coordinate(buffer: *const u8, current_index: &mut usize) -> (usize, bool) { - let digits = unsafe { (buffer.add(*current_index) as *const usize).read_unaligned() }; + let digits = unsafe { (buffer.add(*current_index) as *const usize).read_unaligned() }.to_le(); let mut result = 0; let mut visited = false; diff --git a/breakwater-parser/src/refactored.rs b/breakwater-parser/src/refactored.rs index ee953f3..02c4be2 100644 --- a/breakwater-parser/src/refactored.rs +++ b/breakwater-parser/src/refactored.rs @@ -87,7 +87,7 @@ impl RefactoredParser { let previous = idx; idx += 2; - let command_bytes = unsafe { (buffer.as_ptr().add(idx) as *const u64).read_unaligned() }; + let command_bytes = unsafe { (buffer.as_ptr().add(idx) as *const u64).read_unaligned() }.to_le(); let x = u16::from_le((command_bytes) as u16); let y = u16::from_le((command_bytes >> 16) as u16); @@ -182,7 +182,8 @@ impl RefactoredParser { // We don't want to return the actual (absolute) coordinates, the client should also get the result offseted x - self.connection_x_offset, y - self.connection_y_offset, - rgb.to_be() >> 8 + // We want big-endian order on all systems + rgb.swap_bytes() >> 8 ) .as_bytes(), ); @@ -199,7 +200,7 @@ impl Parser for RefactoredParser { while i < loop_end { let current_command = - unsafe { (buffer.as_ptr().add(i) as *const u64).read_unaligned() }; + unsafe { (buffer.as_ptr().add(i) as *const u64).read_unaligned() }.to_le(); if current_command & 0x00ff_ffff == PX_PATTERN { (i, last_byte_parsed) = self.handle_pixel(buffer, i, response); } else if cfg!(feature = "binary-set-pixel") diff --git a/breakwater/Cargo.toml b/breakwater/Cargo.toml index 741d70d..ddec793 100644 --- a/breakwater/Cargo.toml +++ b/breakwater/Cargo.toml @@ -29,7 +29,7 @@ memadvise.workspace = true ndi-sdk-sys = { workspace = true, optional = true } number_prefix.workspace = true page_size.workspace = true -prometheus_exporter.workspace = true +prometheus_exporter = { workspace = true, optional = true } rusttype.workspace = true serde_json.workspace = true serde.workspace = true @@ -49,13 +49,14 @@ rstest.workspace = true [features] # We don't enable binary-sync-pixels and binary-set-pixel by default to make it a bit harder for clients ;) -default = ["egui", "vnc"] +default = ["egui", "prometheus", "vnc"] alpha = ["breakwater-parser/alpha"] binary-set-pixel = ["breakwater-parser/binary-set-pixel"] binary-sync-pixels = ["breakwater-parser/binary-sync-pixels"] egui = ["dep:breakwater-egui-overlay", "dep:bytemuck", "dep:eframe", "dep:egui", "dep:libloading"] native-display = ["dep:softbuffer", "dep:winit"] +prometheus = ["dep:prometheus_exporter"] vnc = ["dep:vncserver"] ndi = ["dep:ndi-sdk-sys"] diff --git a/breakwater/src/cli_args.rs b/breakwater/src/cli_args.rs index 55366f8..b1032bf 100644 --- a/breakwater/src/cli_args.rs +++ b/breakwater/src/cli_args.rs @@ -35,6 +35,11 @@ pub struct CliArgs { )] pub network_buffer_size: i64, + /// Allowed number of bytes a client can send per second. + /// Any additional bytes a client sends are silently dropped. + #[clap(long)] + pub allowed_bytes_per_second: Option, + /// Text to display on the screen. #[clap(short, long, default_value = "Pixelflut server (breakwater)")] pub text: String, @@ -73,6 +78,10 @@ pub struct CliArgs { #[clap(long)] pub shared_memory_name: Option, + /// Stream KFSVS to the given address and UDP port. + #[clap(long)] + pub kfsvs: Option, + #[clap(flatten)] pub ffmpeg_sink: crate::sinks::ffmpeg::FfmpegSinkCliArgs, diff --git a/breakwater/src/main.rs b/breakwater/src/main.rs index e48b6da..2cad671 100644 --- a/breakwater/src/main.rs +++ b/breakwater/src/main.rs @@ -6,15 +6,17 @@ use color_eyre::eyre::{self, Context}; use server::Server; use tokio::sync::{broadcast, mpsc}; +#[cfg(feature = "prometheus")] +use crate::prometheus_exporter::PrometheusExporter; use crate::{ cli_args::CliArgs, - prometheus_exporter::PrometheusExporter, - sinks::{DisplaySink, ffmpeg::FfmpegSink}, + sinks::{DisplaySink, ffmpeg::FfmpegSink, kfsvs::KfsvsSink}, statistics::{Statistics, StatisticsEvent, StatisticsInformationEvent, StatisticsSaveMode}, }; mod cli_args; mod connection_buffer; +#[cfg(feature = "prometheus")] mod prometheus_exporter; mod server; mod sinks; @@ -50,6 +52,7 @@ async fn main() -> eyre::Result<()> { // If we make the channel to big, stats will start to lag behind // TODO: Check performance impact in real-world scenario. Maybe the statistics thread blocks the other threads let (statistics_tx, statistics_rx) = mpsc::channel::(100); + #[allow(unused)] let (statistics_information_tx, statistics_information_rx) = broadcast::channel::(2); let (terminate_signal_tx, terminate_signal_rx) = broadcast::channel::<()>(1); @@ -79,18 +82,23 @@ async fn main() -> eyre::Result<()> { format!("invalid network buffer size: {}", args.network_buffer_size) })?, args.connections_per_ip, + args.allowed_bytes_per_second, ) .await .context("failed to start pixelflut server")?; - let mut prometheus_exporter = PrometheusExporter::new( - &args.prometheus_listen_address, - statistics_information_rx.resubscribe(), - ) - .context("failed to start prometheus exporter")?; + #[cfg(feature = "prometheus")] + { + let mut prometheus_exporter = PrometheusExporter::new( + &args.prometheus_listen_address, + statistics_information_rx.resubscribe(), + ) + .context("failed to start prometheus exporter")?; + } let server_listener_thread = tokio::spawn(async move { server.start().await }); let statistics_thread = tokio::spawn(async move { statistics.run().await }); + #[cfg(feature = "prometheus")] let prometheus_exporter_thread = tokio::spawn(async move { prometheus_exporter.run().await }); let mut display_sinks = Vec:: + Send>>::new(); @@ -161,6 +169,12 @@ async fn main() -> eyre::Result<()> { ffmpeg_thread_present = true; } + if let Some(kfsvs_sink) = KfsvsSink::new(fb.clone(), &args, terminate_signal_rx.resubscribe()) + .context("failed to create kfsvs sink")? + { + display_sinks.push(Box::new(kfsvs_sink)); + } + let mut sink_threads = Vec::new(); for mut sink in display_sinks { sink_threads.push(tokio::spawn(async move { @@ -200,6 +214,7 @@ async fn main() -> eyre::Result<()> { #[cfg(not(feature = "egui"))] handle_ctrl_c(terminate_signal_tx).await?; + #[cfg(feature = "prometheus")] prometheus_exporter_thread.abort(); server_listener_thread.abort(); diff --git a/breakwater/src/server.rs b/breakwater/src/server.rs index ead1cb5..07f1214 100644 --- a/breakwater/src/server.rs +++ b/breakwater/src/server.rs @@ -18,7 +18,7 @@ use tokio::{ time::Instant, }; use tokio_stream::wrappers::TcpListenerStream; -use tracing::instrument; +use tracing::{instrument, warn}; use crate::{ connection_buffer::ConnectionBuffer, @@ -37,6 +37,7 @@ pub struct Server { network_buffer_size: usize, connections_per_ip: HashMap, max_connections_per_ip: Option, + allowed_bytes_per_second: Option, } impl Server { @@ -47,6 +48,7 @@ impl Server { statistics_tx: mpsc::Sender, network_buffer_size: usize, max_connections_per_ip: Option, + allowed_bytes_per_second: Option, ) -> eyre::Result { let mut listener_streams = Vec::with_capacity(listen_addresses.len()); for addr in listen_addresses { @@ -68,6 +70,7 @@ impl Server { network_buffer_size, connections_per_ip: HashMap::new(), max_connections_per_ip, + allowed_bytes_per_second, }) } @@ -126,6 +129,9 @@ impl Server { let statistics_tx_for_thread = self.statistics_tx.clone(); let network_buffer_size = self.network_buffer_size; let connection_dropped_tx_clone = connection_dropped_tx.clone(); + let allowed_bytes_per_interval = self + .allowed_bytes_per_second + .map(|ab| (ab as usize * 1000) / STATISTICS_REPORT_INTERVAL.as_millis() as usize); tokio::spawn(async move { if let Err(error) = handle_connection( stream, @@ -134,6 +140,7 @@ impl Server { statistics_tx_for_thread, network_buffer_size, connection_dropped_tx_clone, + allowed_bytes_per_interval, ) .await { @@ -147,7 +154,13 @@ impl Server { } #[instrument( - skip(stream, fb, statistics_tx, connection_dropped_tx), + skip( + stream, + fb, + statistics_tx, + connection_dropped_tx, + allowed_bytes_per_interval + ), err(level = "debug") )] pub async fn handle_connection( @@ -157,6 +170,7 @@ pub async fn handle_connection( statistics_tx: mpsc::Sender, network_buffer_size: usize, connection_dropped_tx: Option>, + allowed_bytes_per_interval: Option, ) -> eyre::Result<()> { tracing::debug!("handling new connection"); @@ -183,6 +197,8 @@ pub async fn handle_connection( let mut last_statistics = Instant::now(); let mut statistics_bytes_read: u64 = 0; + let mut has_warned_for_connection = false; + // Fill the buffer up with new data from the socket // If there are any bytes left over from the previous loop iteration leave them as is and put the new data behind while let Ok(bytes_read) = stream @@ -205,6 +221,18 @@ pub async fn handle_connection( statistics_bytes_read = 0; } + if let Some(allowed_bytes_per_interval) = allowed_bytes_per_interval + && statistics_bytes_read as usize > allowed_bytes_per_interval + { + if !has_warned_for_connection { + warn!("Rate-limiting client due to send capacity exceeded."); + has_warned_for_connection = true; + } + // Discard pending data, since we ignore the data that continues it + leftover_bytes_in_buffer = 0; + continue; + } + let data_end = leftover_bytes_in_buffer + bytes_read; if bytes_read == 0 { if leftover_bytes_in_buffer == 0 { diff --git a/breakwater/src/sinks/kfsvs.rs b/breakwater/src/sinks/kfsvs.rs new file mode 100644 index 0000000..f80e5f3 --- /dev/null +++ b/breakwater/src/sinks/kfsvs.rs @@ -0,0 +1,116 @@ +//! kleines Filmröllchen’s Shitty Video Streaming (kfsvs) +//! +//! ## Format +//! +//! 24-byte header for each frame: +//! - Fixed prelude, see [`DATA_PRELUDE`]. +//! - 4 bytes network-endian frame length (unsigned integer) +//! - 4 bytes framebuffer endianness (1 = little-endian, RGBA order, 0 = big-endian, ABGR order) +//! - raw framebuffer with stride 4, length as per header field + +use std::{net::SocketAddr, sync::Arc, time::Duration}; + +use async_trait::async_trait; +use breakwater_parser::FrameBuffer; +use color_eyre::eyre; +use tokio::{ + io, + net::UdpSocket, + sync::broadcast, + time::{Instant, sleep_until}, +}; +use tracing::{debug, info, instrument}; + +use crate::{cli_args::CliArgs, sinks::DisplaySink}; + +pub struct KfsvsSink { + fb: Arc, + terminate_signal_rx: broadcast::Receiver<()>, + + addr: SocketAddr, + target_fps: u32, +} + +impl KfsvsSink { + #[instrument(skip_all, err)] + pub fn new( + fb: Arc, + cli_args: &CliArgs, + terminate_signal_rx: broadcast::Receiver<()>, + ) -> eyre::Result> { + if let Some(addr) = cli_args.kfsvs { + Ok(Some(Self { + fb, + terminate_signal_rx, + addr, + target_fps: cli_args.fps, + })) + } else { + Ok(None) + } + } +} + +// A basic string of convenient size containing the protocol name and a bunch of data that shouldn’t just appear within a frame. +const DATA_PRELUDE: &[u8; 16] = b"%%KFSVS%%\xaa\xbb\xcc\xdd\xee\xff\0"; +const MAX_SAFE_UDP_SIZE: usize = 1280; +const ENDIAN_MARKER: u32 = + cfg_select! {target_endian = "big" => 0u32, target_endian = "little" => 1u32}; + +#[async_trait] +impl DisplaySink for KfsvsSink { + #[instrument(skip(self), err)] + async fn run(&mut self) -> eyre::Result<()> { + let socket = UdpSocket::bind("[::]:0").await?; + socket.connect(self.addr).await?; + info!(?self.addr, "Connected KFSVS stream"); + let target_wait_time = Duration::from_secs(1) / self.target_fps; + debug!( + ?target_wait_time, + target_fps = self.target_fps, + "established inter-frame wait time" + ); + + loop { + if self.terminate_signal_rx.try_recv().is_ok() { + return eyre::Ok(()); + } + let start_time = Instant::now(); + let next_start_time = start_time + target_wait_time; + let res = (async || { + let mut header: [u8; 24] = [0; _]; + header[0..16].copy_from_slice(DATA_PRELUDE); + let source_frame_data = self.fb.as_bytes(); + let source_frame_len = source_frame_data.len() as u32; + header[16..20].copy_from_slice(&source_frame_len.to_be_bytes()); + header[20..].copy_from_slice(&ENDIAN_MARKER.to_be_bytes()); + socket.send(&header).await?; + + let mut remaining_data_slice = source_frame_data; + while remaining_data_slice.len() > MAX_SAFE_UDP_SIZE { + let (safe_slice, remaining) = remaining_data_slice.split_at(MAX_SAFE_UDP_SIZE); + remaining_data_slice = remaining; + socket.send(safe_slice).await?; + } + if remaining_data_slice.len() > 0 { + socket.send(remaining_data_slice).await?; + } + + sleep_until(next_start_time).await; + eyre::Ok(()) + })() + .await; + if let Err(err) = res { + match err.downcast::() { + Ok(inner_e) if inner_e.kind() == io::ErrorKind::ConnectionRefused => { + // No receiver yet connected + sleep_until(next_start_time).await; + continue; + } + Ok(other) => return Err(other.into()), + Err(e) => return Err(e), + } + } + } + } +} diff --git a/breakwater/src/sinks/mod.rs b/breakwater/src/sinks/mod.rs index 198b070..6d8460b 100644 --- a/breakwater/src/sinks/mod.rs +++ b/breakwater/src/sinks/mod.rs @@ -10,6 +10,7 @@ pub mod native_display; pub mod ndi; #[cfg(feature = "vnc")] pub mod vnc; +pub mod kfsvs; // The stabilization of async functions in traits in Rust 1.75 did not include support for using traits containing async // functions as dyn Trait, so we still need to use async_trait here. diff --git a/kfsvs-receive.rs b/kfsvs-receive.rs new file mode 100755 index 0000000..bb14666 --- /dev/null +++ b/kfsvs-receive.rs @@ -0,0 +1,88 @@ +#!/usr/bin/env -S cargo +nightly -Zscript +--- +package.edition = "2024" + +[profile.dev] +opt-level = 3 +--- + +use std::env::args; +use std::io::Write; +use std::net::{SocketAddr, UdpSocket}; +use std::slice; + +const DATA_PRELUDE: &[u8; 16] = b"%%KFSVS%%\xaa\xbb\xcc\xdd\xee\xff\0"; +const ENDIAN_MARKER: u32 = + cfg_select! {target_endian = "big" => 0u32, target_endian = "little" => 1u32}; + +pub fn main() { + let listen_addr: SocketAddr = args() + .nth(1) + .expect("usage: kfsvs-receive LISTEN_ADDR:LISTEN_PORT") + .parse() + .expect("invalid socket address"); + + let socket = UdpSocket::bind(listen_addr).expect("could not bind to UDP socket"); + let mut buffer = vec![0; 24]; + + let mut stdout = std::io::stdout(); + + let mut frame = Vec::new(); + 'frame: loop { + let (size, remote_addr) = socket.recv_from(buffer.as_mut_slice()).unwrap(); + let data = &buffer[..size]; + + if data.starts_with(DATA_PRELUDE) && data.len() >= 24 { + // Correct new header received. + let frame_size = u32::from_be_bytes(*data[16..20].as_array().unwrap()); + let endianness = u32::from_be_bytes(*data[20..24].as_array().unwrap()); + eprintln!( + "received frame of size {frame_size} endianness {endianness} from {remote_addr:?}" + ); + if frame_size == 0 { + continue; + } + + if !frame_size.is_multiple_of(4) { + eprintln!("error: frame size is not multiple of 4"); + continue; + } + + // Receive rest of frame. + frame.resize((frame_size / 4) as usize, 0u32); + // SAFETY: idk, things are broken sometimes + let raw_frame = unsafe { + slice::from_raw_parts_mut(frame.as_mut_ptr() as *mut u8, frame.len() * 4) + }; + let mut write_index = 0; + if data.len() > 24 { + raw_frame.copy_from_slice(&data[24..]); + write_index += data.len() - 24; + } + + while write_index < frame_size as usize { + let (size, _) = socket.recv_from(&mut raw_frame[write_index..]).unwrap(); + // This condition fixes all visual glitches, even when it doesn’t trigger. + if raw_frame[write_index..].starts_with(b"%%KF") { + eprintln!("transport got messed up, discarding frame"); + continue 'frame; + } + write_index += size; + } + // eprintln!("finished receiving frame"); + + // Send out frame over stdout + for pixel in &frame { + let correct_endianness_pixel = if endianness != ENDIAN_MARKER { + pixel.swap_bytes() + } else { + *pixel + }; + stdout + .write_all(&correct_endianness_pixel.to_le_bytes()[0..3]) + .expect("ffmpeg exited"); + } + stdout.flush().unwrap(); + } + } +}