WIP: Big endian support and basic raw UDP framebuffer stream

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
 

breakwater-parser/src/framebuffer/mod.rs

@@ -31,13 +31,17 @@ 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
     /// aligned. As the data already is stored in a buffer we can not guarantee it's correctly aligned, so let's just
     /// 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();

breakwater-parser/src/original.rs

@@ -93,8 +93,9 @@ impl<FB: FrameBuffer> Parser for OriginalParser<FB> {
         }
 
         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<FB: FrameBuffer> Parser for OriginalParser<FB> {
                                     // 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<FB: FrameBuffer> Parser for OriginalParser<FB> {
             #[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<FB: FrameBuffer> Parser for OriginalParser<FB> {
             #[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;

breakwater-parser/src/refactored.rs

@@ -87,7 +87,7 @@ impl<FB: FrameBuffer> RefactoredParser<FB> {
         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<FB: FrameBuffer> RefactoredParser<FB> {
                     // 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<FB: FrameBuffer> Parser for RefactoredParser<FB> {
 
         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")

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"]
 

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<usize>,
+
     /// 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<String>,
 
+    /// Stream KFSVS to the given address and UDP port.
+    #[clap(long)]
+    pub kfsvs: Option<SocketAddr>,
+
     #[clap(flatten)]
     pub ffmpeg_sink: crate::sinks::ffmpeg::FfmpegSinkCliArgs,
 

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::<StatisticsEvent>(100);
+    #[allow(unused)]
     let (statistics_information_tx, statistics_information_rx) =
         broadcast::channel::<StatisticsInformationEvent>(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::<Box<dyn DisplaySink<SharedMemoryFrameBuffer> + 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();
 

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<FB: FrameBuffer> {
     network_buffer_size: usize,
     connections_per_ip: HashMap<IpAddr, u64>,
     max_connections_per_ip: Option<u64>,
+    allowed_bytes_per_second: Option<usize>,
 }
 
 impl<FB: FrameBuffer + Send + Sync + 'static> Server<FB> {
@@ -47,6 +48,7 @@ impl<FB: FrameBuffer + Send + Sync + 'static> Server<FB> {
         statistics_tx: mpsc::Sender<StatisticsEvent>,
         network_buffer_size: usize,
         max_connections_per_ip: Option<u64>,
+        allowed_bytes_per_second: Option<usize>,
     ) -> eyre::Result<Self> {
         let mut listener_streams = Vec::with_capacity(listen_addresses.len());
         for addr in listen_addresses {
@@ -68,6 +70,7 @@ impl<FB: FrameBuffer + Send + Sync + 'static> Server<FB> {
             network_buffer_size,
             connections_per_ip: HashMap::new(),
             max_connections_per_ip,
+            allowed_bytes_per_second,
         })
     }
 
@@ -126,6 +129,9 @@ impl<FB: FrameBuffer + Send + Sync + 'static> Server<FB> {
             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<FB: FrameBuffer + Send + Sync + 'static> Server<FB> {
                     statistics_tx_for_thread,
                     network_buffer_size,
                     connection_dropped_tx_clone,
+                    allowed_bytes_per_interval,
                 )
                 .await
                 {
@@ -147,7 +154,13 @@ impl<FB: FrameBuffer + Send + Sync + 'static> Server<FB> {
 }
 
 #[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<FB: FrameBuffer>(
@@ -157,6 +170,7 @@ pub async fn handle_connection<FB: FrameBuffer>(
     statistics_tx: mpsc::Sender<StatisticsEvent>,
     network_buffer_size: usize,
     connection_dropped_tx: Option<mpsc::UnboundedSender<IpAddr>>,
+    allowed_bytes_per_interval: Option<usize>,
 ) -> eyre::Result<()> {
     tracing::debug!("handling new connection");
 
@@ -183,6 +197,8 @@ pub async fn handle_connection<FB: FrameBuffer>(
     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<FB: FrameBuffer>(
             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 {