diff --git a/.github/workflows/build.yml b/.github/workflows/build.yml index 4ab979f..f1cbacd 100644 --- a/.github/workflows/build.yml +++ b/.github/workflows/build.yml @@ -86,14 +86,16 @@ jobs: uses: actions-rs/clippy-check@v1 if: env.GITHUB_TOKEN != null with: - args: --all-targets --features prometheus,alpha,binary-set-pixel,binary-sync-pixels,egui,winit,vnc,ndi -- -D warnings + args: --all-targets --features prometheus,binary-set-pixel,egui,winit,vnc,ndi -- -D warnings token: ${{ secrets.GITHUB_TOKEN }} - name: Run clippy (no default features) run: cargo clippy --all-targets --no-default-features -- -D warnings - name: Run clippy (all compatible features) - run: cargo clippy --all-targets --features prometheus,alpha,binary-set-pixel,binary-sync-pixels,egui,winit,vnc,ndi -- -D warnings + run: cargo clippy --all-targets --features prometheus,binary-set-pixel,egui,winit,vnc,ndi -- -D warnings - name: Run clippy on the time-tracking feature (mutually exclusive with alpha and binary-sync-pixels) run: cargo clippy --all-targets --features prometheus,binary-set-pixel,egui,winit,vnc,ndi,time-tracking -- -D warnings + - name: Run clippy (all compatible features for breakwater) + run: cargo clippy -p breakwater --features prometheus,alpha,binary-set-pixel,binary-sync-pixels,egui,winit,vnc,ndi -- -D warnings run_rustdoc: name: Run RustDoc @@ -127,23 +129,16 @@ jobs: run: sudo apt update && sudo apt install -y libvncserver-dev - name: Install NDI SDK uses: ./.github/actions/install-ndi - - name: Test with all features turned off - uses: actions-rs/cargo@v1 - with: - command: test - args: --no-default-features --all-targets - # We can't use `--all-features`: `time-tracking` is mutually exclusive with `alpha` and - # `binary-sync-pixels`. So we test explicit, valid combinations. + - name: Run tests (no default features) + run: cargo test --all-targets --no-default-features + - name: Run tests (all compatible features) + run: cargo test --all-targets --features prometheus,binary-set-pixel,egui,winit,vnc,ndi + - name: Run tests on the time-tracking feature (mutually exclusive with alpha and binary-sync-pixels) + run: cargo test --all-targets --features prometheus,binary-set-pixel,egui,winit,vnc,ndi,time-tracking - name: Test breakwater with all compatible features - uses: actions-rs/cargo@v1 - with: - command: test - args: --features prometheus,alpha,binary-set-pixel,binary-sync-pixels,egui,winit,vnc,ndi --all-targets + run: cargo test -p breakwater --features prometheus,alpha,binary-set-pixel,binary-sync-pixels,egui,winit,vnc,ndi - name: Test breakwater-parser with the time-tracking feature (mutually exclusive with alpha and binary-sync-pixels) - uses: actions-rs/cargo@v1 - with: - command: test - args: --features prometheus,binary-set-pixel,egui,winit,vnc,ndi,time-tracking --all-targets + run: cargo test -p breakwater-parser --features time-tracking,binary-set-pixel run_tests_for_feature_combinations: name: Run tests for all feature combinations @@ -209,16 +204,18 @@ jobs: with: command: build args: --target=${{ matrix.target }} --no-default-features - - if: runner.os == 'Linux' + - name: Build on Linux (all compatible features) + if: runner.os == 'Linux' uses: actions-rs/cargo@v1 with: command: build # ndi-sdk-sys (behind the ndi features) only supports Linux x86 (no aarch64) so far. # We already run the tests on x86 Linux with all features above. - args: --target=${{ matrix.target }} --features prometheus,alpha,binary-set-pixel,binary-sync-pixels,egui,winit,vnc + args: --target=${{ matrix.target }} --features prometheus,binary-set-pixel,egui,winit,vnc # I couldn't get pkg-config to work on macOS and Windows, so we omit the vnc and ndi feature - - if: runner.os == 'macOS' || runner.os == 'Windows' + - name: Build on Windows/MacOS (all compatible features) + if: runner.os == 'Windows' || runner.os == 'macOS' uses: actions-rs/cargo@v1 with: command: build - args: --target=${{ matrix.target }} --no-default-features --features prometheus,alpha,binary-set-pixel,binary-sync-pixels,egui,winit + args: --target=${{ matrix.target }} --no-default-features --features prometheus,binary-set-pixel,egui,winit diff --git a/.gitignore b/.gitignore index d84f564..da79167 100644 --- a/.gitignore +++ b/.gitignore @@ -3,3 +3,4 @@ .vscode/ statistics.json /pixelflut/ +worker-id diff --git a/Cargo.lock b/Cargo.lock index 06f943f..5178045 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -271,6 +271,15 @@ dependencies = [ "syn", ] +[[package]] +name = "atomic-polyfill" +version = "1.0.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8cf2bce30dfe09ef0bfaef228b9d414faaf7e563035494d7fe092dba54b300f4" +dependencies = [ + "critical-section", +] + [[package]] name = "atomic-waker" version = "1.1.2" @@ -505,6 +514,22 @@ dependencies = [ "winit", ] +[[package]] +name = "breakwater-deich" +version = "0.22.0" +dependencies = [ + "breakwater", + "breakwater-parser", + "clap", + "color-eyre", + "postcard", + "prometheus_exporter", + "serde", + "tokio", + "tracing", + "uuid", +] + [[package]] name = "breakwater-egui-overlay" version = "0.22.0" @@ -805,6 +830,15 @@ dependencies = [ "error-code", ] +[[package]] +name = "cobs" +version = "0.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0fa961b519f0b462e3a3b4a34b64d119eeaca1d59af726fe450bbba07a9fc0a1" +dependencies = [ + "thiserror 2.0.18", +] + [[package]] name = "codespan-reporting" version = "0.13.1" @@ -997,6 +1031,12 @@ dependencies = [ "itertools 0.13.0", ] +[[package]] +name = "critical-section" +version = "1.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "790eea4361631c5e7d22598ecd5723ff611904e3344ce8720784c93e3d83d40b" + [[package]] name = "crossbeam-deque" version = "0.8.6" @@ -1400,6 +1440,18 @@ dependencies = [ "serde", ] +[[package]] +name = "embedded-io" +version = "0.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ef1a6892d9eef45c8fa6b9e0086428a2cca8491aca8f787c534a3d6d0bcb3ced" + +[[package]] +name = "embedded-io" +version = "0.6.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "edd0f118536f44f5ccd48bcb8b111bdc3de888b58c74639dfb034a357d0f206d" + [[package]] name = "enum-as-inner" version = "0.6.1" @@ -1778,10 +1830,21 @@ checksum = "899def5c37c4fd7b2664648c28120ecec138e4d395b459e5ca34f9cce2dd77fd" dependencies = [ "cfg-if", "libc", - "r-efi", + "r-efi 5.3.0", "wasip2", ] +[[package]] +name = "getrandom" +version = "0.4.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "300e883d756b2e4ec94e02791f39b04b522276138852cfc41d9fb7e904106099" +dependencies = [ + "cfg-if", + "libc", + "r-efi 6.0.0", +] + [[package]] name = "getset" version = "0.1.7" @@ -1916,6 +1979,15 @@ dependencies = [ "zerocopy", ] +[[package]] +name = "hash32" +version = "0.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b0c35f58762feb77d74ebe43bdbc3210f09be9fe6742234d573bacc26ed92b67" +dependencies = [ + "byteorder", +] + [[package]] name = "hashbrown" version = "0.16.1" @@ -1931,6 +2003,20 @@ version = "0.17.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ed5909b6e89a2db4456e54cd5f673791d7eca6732202bbf2a9cc504fe2f9b84a" +[[package]] +name = "heapless" +version = "0.7.17" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cdc6457c0eb62c71aac4bc17216026d8410337c4126773b9c5daba343f17964f" +dependencies = [ + "atomic-polyfill", + "hash32", + "rustc_version", + "serde", + "spin", + "stable_deref_trait", +] + [[package]] name = "heck" version = "0.5.0" @@ -3525,6 +3611,19 @@ dependencies = [ "portable-atomic", ] +[[package]] +name = "postcard" +version = "1.1.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6764c3b5dd454e283a30e6dfe78e9b31096d9e32036b5d1eaac7a6119ccb9a24" +dependencies = [ + "cobs", + "embedded-io 0.4.0", + "embedded-io 0.6.1", + "heapless", + "serde", +] + [[package]] name = "potential_utf" version = "0.1.5" @@ -3691,6 +3790,12 @@ version = "5.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "69cdb34c158ceb288df11e18b4bd39de994f6657d83847bdffdbd7f346754b0f" +[[package]] +name = "r-efi" +version = "6.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f8dcc9c7d52a811697d2151c701e0d08956f92b0e24136cf4cf27b57a6a0d9bf" + [[package]] name = "rand" version = "0.8.6" @@ -4368,6 +4473,15 @@ dependencies = [ "x11rb", ] +[[package]] +name = "spin" +version = "0.9.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6980e8d7511241f8acf4aebddbb1ff938df5eebe98691418c4468d0b72a96a67" +dependencies = [ + "lock_api", +] + [[package]] name = "stable_deref_trait" version = "1.2.1" @@ -4895,7 +5009,10 @@ version = "1.23.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "bf80a72845275afea99e7f2b434723d3bc7e38470fcd1c7ed39a599c73319a53" dependencies = [ + "getrandom 0.4.3", + "js-sys", "serde_core", + "wasm-bindgen", ] [[package]] diff --git a/Cargo.toml b/Cargo.toml index 0d23347..bbef80b 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -3,7 +3,8 @@ members = [ "breakwater-parser", "breakwater", "breakwater-egui-overlay", - "breakwater-parser-c-bindings" + "breakwater-parser-c-bindings", + "breakwater-deich", ] resolver = "2" @@ -34,6 +35,7 @@ number_prefix = "0.4" ndi-sdk-sys = "0.1.1" page_size = "0.6" pixelbomber = "1.1" +postcard = { version = "1.1", features = ["use-std"] } prometheus_exporter = "0.8" rstest = "0.26" rusttype = "0.9" @@ -48,12 +50,14 @@ tokio-stream = { version = "0.1", features = ["net"] } tracing = "0.1" tracing-subscriber = { version = "0.3", features = ["env-filter"] } trait-variant = "0.1" +uuid = { version = "1.23", features = ["v4", "serde"] } vncserver = "0.2" winit = "0.30" # Uses the given path when used locally, and uses the specified version from crates.io when published. breakwater-core = { path = "breakwater-core", version = "0.22.0", default-features = false } breakwater-parser = { path = "breakwater-parser", version = "0.22.0", default-features = false } +breakwater = { path = "breakwater", version = "0.22.0", default-features = false } breakwater-egui-overlay = { path = "breakwater-egui-overlay", version = "0.22.0", default-features = false } [profile.dev] diff --git a/breakwater-deich/Cargo.toml b/breakwater-deich/Cargo.toml new file mode 100644 index 0000000..ed49f6a --- /dev/null +++ b/breakwater-deich/Cargo.toml @@ -0,0 +1,40 @@ +[package] +name = "breakwater-deich" +description = "Distributed Pixelflut server" +version.workspace = true +authors.workspace = true +license.workspace = true +edition.workspace = true +repository.workspace = true + +[[bin]] +name = "breakwater-deich" +path = "src/main.rs" + +[dependencies] +breakwater-parser = { workspace = true, features = ["time-tracking"] } +breakwater = { workspace = true, features = ["prometheus"] } + +clap.workspace = true +color-eyre.workspace = true +postcard.workspace = true +prometheus_exporter.workspace = true +serde.workspace = true +tokio.workspace = true +tracing.workspace = true +uuid.workspace = true + +[features] +# VNC and NDI require shared libraries to be installed, so we don't enable them by default +# Yes, egui and winit require a graphical environment, but most people getting started with +# breakwater will likely do that on a desktop, servers admins generally know what they are doing (tm). +default = ["egui", "winit"] + +binary-set-pixel = ["breakwater/binary-set-pixel", "breakwater-parser/binary-set-pixel"] +egui = ["breakwater/egui"] +winit = ["breakwater/winit"] +vnc = ["breakwater/vnc"] +ndi = ["breakwater/ndi"] + +[lints] +workspace = true diff --git a/breakwater-deich/src/cli_args.rs b/breakwater-deich/src/cli_args.rs new file mode 100644 index 0000000..f551f3f --- /dev/null +++ b/breakwater-deich/src/cli_args.rs @@ -0,0 +1,82 @@ +use std::{net::SocketAddr, path::PathBuf}; + +use breakwater::{ + cli_args::{NetworkListenerCliArgs, StatisticsSaveFileCliArgs}, + sinks::cli_args::SinkCliArgs, +}; + +#[derive(clap::Parser, Debug)] +#[clap(author, version, about, long_about = None)] +pub struct CliArgs { + #[command(subcommand)] + pub role: Role, +} + +#[derive(clap::Subcommand, Debug)] +pub enum Role { + /// Run a worker: A Pixelflut server that continuously syncs its framebuffer to the collector. + /// Canvas geometry and frame rate are dictated by the collector. + Worker(Box), + + /// Run the collector: Gathers and merges the framebuffers of all workers. + Collector(Box), +} + +#[derive(clap::Args, Debug)] +pub struct WorkerCliArgs { + #[clap(flatten)] + pub network_listener: NetworkListenerCliArgs, + + /// Address of the collector to fetch the config from and stream the framebuffer to. + #[clap(long, default_value = "[::1]:1235")] + pub collector_address: SocketAddr, + + /// File storing this worker's persistent UUID, created on first run. The UUID identifies the + /// worker to the collector across restarts, which keeps per-worker stats stable over time. + #[clap(long, default_value = "worker-id")] + pub worker_id_file: PathBuf, + + #[clap(flatten)] + pub prometheus: PrometheusCliArgs, +} + +#[derive(clap::Args, Debug)] +pub struct CollectorCliArgs { + /// Listen address the workers connect to. + #[clap(short, long, default_value = "[::]:1235")] + pub listen_address: SocketAddr, + + /// Width of the canvas. Sent to every worker as part of its config. + #[clap(long, default_value_t = 1920)] + pub width: u32, + + /// Height of the canvas. Sent to every worker as part of its config. + #[clap(long, default_value_t = 1080)] + pub height: u32, + + /// Frames per second the workers should sync at. Sent to every worker as part of its config. + /// Must stay well below the ~536 ms window the per-pixel timestamp can represent. + #[clap(long, default_value_t = 30, value_parser = clap::value_parser!(u32).range(1..=60))] + pub fps: u32, + + #[clap(flatten)] + pub prometheus: PrometheusCliArgs, + + #[clap(flatten)] + pub statistics_save_file: StatisticsSaveFileCliArgs, + + #[clap(flatten)] + pub sinks: SinkCliArgs, +} + +/// Prometheus metrics options, shared by both roles. Enabled by default (like breakwater): each node +/// serves `/metrics` on this address. Metrics are also summarized to the log on an interval, so +/// they're available even without a scraper. +#[derive(clap::Args, Debug)] +#[command(next_help_heading = "Prometheus options")] +pub struct PrometheusCliArgs { + /// Address to serve Prometheus metrics on. Workers and the collector run on separate hosts, so + /// the same default port is fine for all of them; override it if you colocate roles on one host. + #[clap(long, default_value = "[::]:9100")] + pub prometheus_listen_address: SocketAddr, +} diff --git a/breakwater-deich/src/collector/canvas.rs b/breakwater-deich/src/collector/canvas.rs new file mode 100644 index 0000000..5d93f96 --- /dev/null +++ b/breakwater-deich/src/collector/canvas.rs @@ -0,0 +1,101 @@ +//! The long-term merged canvas. +//! +//! Every pixel keeps its full write timestamp (the high bits of [`TimeTrackingPixel`]), set by the +//! worker relative to the shared epoch. Merging is therefore exact last-write-wins per pixel — and, +//! crucially, *commutative*: merge order doesn't matter, so the collector can fold each worker's +//! latest framebuffer in whenever it arrives without any frame numbering or windowing. A +//! never-written pixel has timestamp `0` (the oldest possible), so a blank or restarted worker's +//! frame can never clobber live content, and the canvas keeps its contents across traffic gaps. + +use std::sync::Arc; + +use breakwater_parser::{FB_BYTES_PER_PIXEL, FrameBuffer, MultiPixelSet, TimeTrackingPixel}; + +/// A flat pixel vector — no width/height; merging is purely per-index. +pub struct Canvas { + pixels: Vec, + /// Reused scratch for [`Self::draw_to_framebuffer`]'s RGB byte layout, so the per-tick draw + /// doesn't allocate a fresh multi-megabyte buffer at the frame rate. + rgb_scratch: Vec, +} + +impl Canvas { + pub fn new(width: usize, height: usize) -> Self { + let pixel_count = width * height; + Self { + pixels: vec![TimeTrackingPixel::default(); pixel_count], + rgb_scratch: Vec::with_capacity(pixel_count * FB_BYTES_PER_PIXEL), + } + } + + /// Folds `frame` in, keeping for each pixel whichever write has the larger timestamp. + pub fn merge(&mut self, frame: &[TimeTrackingPixel]) { + for (canvas_pixel, &frame_pixel) in self.pixels.iter_mut().zip(frame) { + if frame_pixel.timestamp() > canvas_pixel.timestamp() { + *canvas_pixel = frame_pixel; + } + } + } + + pub fn draw_to_framebuffer(&mut self, fb: &Arc) { + self.rgb_scratch.clear(); + self.rgb_scratch.extend( + self.pixels + .iter() + .flat_map(|pixel| pixel.rgb().to_le_bytes()), + ); + fb.set_multi_from_start_index(0, &self.rgb_scratch); + } +} + +#[cfg(test)] +mod tests { + use super::*; + + /// Builds a frame from `(rgb, timestamp)` pairs. + fn frame(pixels: &[(u32, u64)]) -> Vec { + pixels + .iter() + .map(|&(rgb, timestamp)| TimeTrackingPixel::new(rgb, timestamp)) + .collect() + } + + #[test] + fn merges_latest_timestamp_per_pixel() { + let mut canvas = Canvas::new(1, 2); + + // Pixel 0: A timestamp 1320, B timestamp 5032 -> B wins. + // Pixel 1: A timestamp 4200, B timestamp 0 (never written) -> A stays. + canvas.merge(&frame(&[(0xaa_0000, 1_320), (0xaa_0001, 4_200)])); + canvas.merge(&frame(&[(0x00_00bb, 5_032), (0x00_00bc, 0)])); + + assert_eq!(canvas.pixels[0].rgb(), 0x00_00bb); + assert_eq!(canvas.pixels[0].timestamp(), 5_032); + assert_eq!(canvas.pixels[1].rgb(), 0xaa_0001); + assert_eq!(canvas.pixels[1].timestamp(), 4_200); + } + + #[test] + fn blank_frame_never_overwrites_live_content() { + let mut canvas = Canvas::new(1, 1); + canvas.merge(&frame(&[(0x12_3456, 50)])); + + // A never-written pixel has timestamp 0 (oldest possible), so it can't clobber live content + // (the restarted-worker / blank-canvas case). + canvas.merge(&frame(&[(0, 0)])); + + assert_eq!(canvas.pixels[0].rgb(), 0x12_3456); + assert_eq!(canvas.pixels[0].timestamp(), 50); + } + + #[test] + fn older_write_does_not_replace_newer() { + let mut canvas = Canvas::new(1, 1); + // Merge the newer write first, then an older one — order must not matter. + canvas.merge(&frame(&[(0x00_00bb, 5_032)])); + canvas.merge(&frame(&[(0xaa_0000, 1_320)])); + + assert_eq!(canvas.pixels[0].rgb(), 0x00_00bb); + assert_eq!(canvas.pixels[0].timestamp(), 5_032); + } +} diff --git a/breakwater-deich/src/collector/mod.rs b/breakwater-deich/src/collector/mod.rs new file mode 100644 index 0000000..4738bc8 --- /dev/null +++ b/breakwater-deich/src/collector/mod.rs @@ -0,0 +1,433 @@ +//! The collector role: accepts worker connections, keeps each worker's latest framebuffer, and on a +//! fixed render cadence merges them into a long-term [`Canvas`] that the sinks display. +//! +//! There is no frame numbering, window or shared schedule. Because the canvas merges by absolute +//! per-pixel timestamp (last-write-wins, which is *commutative*) and each worker sends a *full* +//! framebuffer that supersedes its earlier ones, the collector can simply keep only the newest frame +//! per worker and fold them all in every render tick — order and arrival timing don't matter. See +//! [`Canvas`] for the correctness argument. + +pub mod canvas; +pub mod stats; + +use std::{ + collections::HashMap, + io, + net::SocketAddr, + sync::{Arc, Mutex}, + time::Duration, +}; + +use breakwater::{ + sinks::start_sinks, + statistics::{ + STATS_REPORT_INTERVAL, StatisticsEvent, StatisticsInformationEvent, StatisticsSaveMode, + }, +}; +use breakwater_parser::{SimpleFrameBuffer, TimeTrackingPixel, get_current_ns_since_unix_epoch}; +use color_eyre::eyre::{self, Context}; +use tokio::{ + net::{TcpListener, TcpStream}, + sync::{broadcast, mpsc}, + time::{MissedTickBehavior, interval}, +}; +use tracing::{debug, info, warn}; +use uuid::Uuid; + +use crate::{ + cli_args::CollectorCliArgs, + collector::{canvas::Canvas, stats::CollectorStatistics}, + metrics::{self, CollectorMetrics, METRICS_LOG_INTERVAL, StatisticsMetrics}, + protocol::{self, Connection, WorkerConfig, WorkerData, frame_period}, +}; + +/// Each connected worker's latest full framebuffer (`None` until its first frame arrives). +type LatestFrame = Option>>; + +/// The connected workers, keyed by UUID. A duplicate UUID is refused at connect time, so each entry +/// maps to exactly one live connection. Read by the render loop every tick and written by the worker +/// connections on (dis)connect and per frame — a `std` (sync) `Mutex` because every critical section +/// is a short map op that never crosses an `.await`. +type Workers = Arc>>; + +/// Collector-wide statistics, shared between the worker-connection tasks (which feed it) and +/// [`publish_statistics`] (which reads it). +type SharedStatistics = Arc>; + +/// Runs the collector until Ctrl-C (or a sink error): accepts worker connections, stores their +/// latest frames, renders the merged canvas, and publishes aggregated statistics. +pub async fn run(args: CollectorCliArgs) -> eyre::Result<()> { + let config = WorkerConfig { + width: args.width, + height: args.height, + fps: args.fps, + // Our startup time: the zero point for every worker's per-pixel timestamps. + epoch_ns_since_unix_epoch: get_current_ns_since_unix_epoch(), + }; + + let listener = TcpListener::bind(args.listen_address) + .await + .with_context(|| format!("failed to bind collector to {}", args.listen_address))?; + info!( + listen_address = %args.listen_address, + ?config, + frame_size = config.frame_size_bytes(), + "Collector listening for workers" + ); + + let workers: Workers = Arc::new(Mutex::new(HashMap::new())); + + // Persistent per-IP statistics: seed the grand totals from the save file (if any) so the "big + // numbers" survive restarts; the live per-worker view is rebuilt as workers (re)connect. + let statistics_save_mode = StatisticsSaveMode::from(args.statistics_save_file); + let statistics = match &statistics_save_mode { + StatisticsSaveMode::Enabled { save_file, .. } => { + match StatisticsInformationEvent::load_from_file(save_file)? { + Some(save_point) => { + info!(%save_file, "Restored statistics from save file"); + CollectorStatistics::from_save_point(save_point) + } + None => CollectorStatistics::default(), + } + } + StatisticsSaveMode::Disabled => CollectorStatistics::default(), + }; + let statistics: SharedStatistics = Arc::new(Mutex::new(statistics)); + + let metrics = + Arc::new(CollectorMetrics::new().context("failed to register collector metrics")?); + metrics::serve(args.prometheus.prometheus_listen_address)?; + + // The breakwater sinks expect the memory layout of a [`SimpleFrameBuffer`], so the render loop + // draws the merged canvas into one for them to consume. + let render_fb = Arc::new(SimpleFrameBuffer::new( + args.width as usize, + args.height as usize, + )); + + // If we make the channel too big, stats will start to lag behind. + let (statistics_tx, mut statistics_rx) = mpsc::channel::(100); + let (statistics_information_tx, statistics_information_rx) = + broadcast::channel::(2); + + // Expose the aggregated per-IP statistics as Prometheus metrics, using the same metric names as + // breakwater. Subscribe before the sender is moved into `publish_statistics` below. + let mut statistics_metrics = StatisticsMetrics::new(statistics_information_tx.subscribe()) + .context("failed to register statistics metrics")?; + + // Support tasks: every handle is kept (no detached tasks) so shutdown is deterministic. The + // collector produces no raw statistics events itself; only sinks (e.g. the VNC sink's rendered + // -frame counter) might, so we drain them. Sinks can still emit during their own teardown, so + // the drain is aborted *last*, after the sink tasks are joined. + let drain_task = tokio::spawn(async move { while statistics_rx.recv().await.is_some() {} }); + let accept_task = tokio::spawn(accept_workers( + listener, + workers.clone(), + statistics.clone(), + metrics.clone(), + config, + )); + let render_task = tokio::spawn(render_loop(workers.clone(), render_fb.clone(), config)); + let publish_task = tokio::spawn(publish_statistics( + statistics.clone(), + statistics_information_tx, + statistics_save_mode, + )); + let metrics_task = tokio::spawn(log_metrics(metrics.clone(), workers.clone())); + let statistics_metrics_task = tokio::spawn(async move { statistics_metrics.run().await }); + + let (sink_tasks, ffmpeg_thread_present) = start_sinks( + &args.sinks, + render_fb.clone(), + // There are no listen addresses we know about — traffic reaches the workers via a virtual + // IP and stuff, not the collector. + &[], + config.fps, + statistics_tx, + statistics_information_rx, + ) + .await + .context("failed to start sinks")?; + + // `start_sinks` returns once shutdown is triggered (a sink erred, or Ctrl-C). Stop the support + // tasks, join the sinks, then stop the stats drain last (see above). + accept_task.abort(); + render_task.abort(); + publish_task.abort(); + metrics_task.abort(); + statistics_metrics_task.abort(); + + for sink_task in sink_tasks { + sink_task + .await + .context("failed to join sink task")? + .context("failed to stop sink")?; + } + + drain_task.abort(); + + if ffmpeg_thread_present { + info!( + "successfully shut down (there might still be a ffmpeg process running - it's complicated)" + ); + } else { + info!("successfully shut down"); + } + + Ok(()) +} + +/// Accepts worker connections forever, spawning a handler per connection. Only returns if accepting +/// itself fails. +async fn accept_workers( + listener: TcpListener, + workers: Workers, + statistics: SharedStatistics, + metrics: Arc, + config: WorkerConfig, +) -> eyre::Result<()> { + loop { + let (stream, peer) = listener + .accept() + .await + .context("failed to accept worker connection")?; + + let workers = workers.clone(); + let statistics = statistics.clone(); + let metrics = metrics.clone(); + tokio::spawn(async move { + match handle_worker(stream, peer, config, &workers, &statistics, &metrics).await { + // The read loop only ever returns on error; a clean disconnect shows up as EOF. + Err(error) if error.kind() == io::ErrorKind::UnexpectedEof => { + debug!(%peer, "Worker connection closed"); + } + Err(error) => warn!(%peer, %error, "Worker connection error"), + Ok(()) => {} + } + }); + } +} + +/// Reads the worker's hello, registers it, serves it until the connection drops, then deregisters +/// it (and drops its statistics baseline). +async fn handle_worker( + stream: TcpStream, + peer: SocketAddr, + config: WorkerConfig, + workers: &Workers, + statistics: &SharedStatistics, + metrics: &CollectorMetrics, +) -> io::Result<()> { + let (mut connection, worker_id) = protocol::accept(stream).await?; + + if !register(workers, worker_id, peer) { + warn!( + %worker_id, + %peer, + "Worker id is already connected; refusing this duplicate connection" + ); + // Returning drops `connection`, which closes it. + return Ok(()); + } + + let result = serve_worker( + &mut connection, + worker_id, + config, + workers, + statistics, + metrics, + ) + .await; + + deregister(workers, worker_id, peer); + // Drop this worker's baseline so its next session accumulates from zero; its already-folded + // grand totals stay. + statistics + .lock() + .expect("collector statistics lock poisoned") + .forget(worker_id); + result +} + +/// Sends the worker its config, then reads its messages forever: keeping its latest framebuffer and +/// recording each statistics snapshot, until the connection drops. +async fn serve_worker( + connection: &mut Connection, + worker_id: Uuid, + config: WorkerConfig, + workers: &Workers, + statistics: &SharedStatistics, + metrics: &CollectorMetrics, +) -> io::Result<()> { + connection.send_config(config).await?; + let pixel_count = config.pixel_count(); + let frame_size_bytes = config.frame_size_bytes() as u64; + // Cache this worker's metric handles once, so the per-frame path skips the label lookup. + let recorder = metrics.recorder(worker_id); + + loop { + match connection.recv_worker_data(pixel_count).await? { + WorkerData::Framebuffer { + pixels, + recv_duration, + latency, + } => { + recorder.observe_frame(recv_duration, latency, frame_size_bytes); + // Keep only this worker's latest frame; the render loop merges it. A full frame + // supersedes the worker's earlier ones, so there is nothing to accumulate here. + if let Some(slot) = workers + .lock() + .expect("workers lock poisoned") + .get_mut(&worker_id) + { + *slot = Some(Arc::new(pixels)); + } + } + WorkerData::Statistics(event) => { + // Fold this snapshot's increase into the grand totals; the publisher reads the + // result on its next tick. + statistics + .lock() + .expect("collector statistics lock poisoned") + .record(worker_id, event); + } + } + } +} + +/// Renders at the configured frame rate: folds every connected worker's latest framebuffer into the +/// long-term canvas (exact latest-write-per-pixel) and draws it to the framebuffer the sinks read. +/// +/// The canvas is persistent for the whole process lifetime, so it survives traffic gaps and worker +/// restarts: with no new frames it keeps its last contents, and a restarted worker's blank frame +/// can't overwrite live pixels (a blank pixel is never "newer" than real content). +async fn render_loop(workers: Workers, render_fb: Arc, config: WorkerConfig) { + let mut canvas = Canvas::new(config.width as usize, config.height as usize); + let mut ticker = interval(frame_period(config.fps)); + // Under load, pace renders a period apart rather than bursting to catch up on missed ticks. + ticker.set_missed_tick_behavior(MissedTickBehavior::Delay); + + loop { + ticker.tick().await; + + // Snapshot the current per-worker frames (cheap `Arc` clones) under a short lock, then merge + // outside it so worker connections never block on rendering. + let frames: Vec>> = workers + .lock() + .expect("workers lock poisoned") + .values() + .filter_map(Clone::clone) + .collect(); + + for frame in &frames { + canvas.merge(frame); + } + canvas.draw_to_framebuffer(&render_fb); + } +} + +/// Logs a per-interval summary of collector ingress (bytes/s and the implied link utilisation) plus +/// the live worker count. Per-worker receive time and end-to-end latency live in the Prometheus +/// histograms; this is just the at-a-glance "is the collector link keeping up" line. Never returns. +async fn log_metrics(metrics: Arc, workers: Workers) { + let mut ticker = interval(METRICS_LOG_INTERVAL); + let interval_secs = METRICS_LOG_INTERVAL.as_secs().max(1); + let mut previous_bytes = 0u64; + + loop { + ticker.tick().await; + + let bytes = metrics.ingress_bytes_total(); + let delta = bytes.saturating_sub(previous_bytes); + previous_bytes = bytes; + let mb_per_s = delta as f64 / interval_secs as f64 / 1_000_000.0; + let connected = workers.lock().expect("workers lock poisoned").len(); + + info!( + connected_workers = connected, + ingress_mb_s = format!("{mb_per_s:.1}"), + ingress_gbit_s = format!("{:.2}", mb_per_s * 8.0 / 1000.0), + "Collector ingress metrics (last interval)" + ); + } +} + +/// Runs for the whole process lifetime, doing two things on their own intervals: +/// - every [`STATS_REPORT_INTERVAL`], publishes the aggregated event on the broadcast channel the +/// sinks already consume — so the overlay renders the merged, per-IP view with no extra work; +/// - every configured save interval (when enabled), persists the grand totals to the save file so +/// the "big numbers" survive a collector restart. +async fn publish_statistics( + statistics: SharedStatistics, + statistics_information_tx: broadcast::Sender, + save_mode: StatisticsSaveMode, +) { + let mut report = interval(STATS_REPORT_INTERVAL); + // Mirror breakwater's `Statistics`: when saving is disabled, an effectively-never timer keeps + // the `select!` arm valid without firing. + let (mut save, save_file) = match &save_mode { + StatisticsSaveMode::Disabled => (interval(Duration::MAX), None), + StatisticsSaveMode::Enabled { + save_file, + interval: save_interval, + } => (interval(*save_interval), Some(save_file.clone())), + }; + + // Previous tick's total byte count, so we can derive a per-second rate at the collector. + let mut previous_bytes = 0u64; + + loop { + tokio::select! { + _ = report.tick() => { + let event = statistics + .lock() + .expect("collector statistics lock poisoned") + .published_event(&mut previous_bytes); + // A send error just means no sink is currently subscribed; nothing to do about it. + let _ = statistics_information_tx.send(event); + } + _ = save.tick() => { + if let Some(save_file) = &save_file { + let save_point = statistics + .lock() + .expect("collector statistics lock poisoned") + .save_point(); + if let Err(error) = save_point.save_to_file(save_file) { + warn!(%save_file, %error, "Failed to save statistics"); + } + } + } + } + } +} + +/// Registers a worker (with no frame yet). Returns `false` (without inserting) if its UUID is +/// already connected. +fn register(workers: &Workers, worker_id: Uuid, peer: SocketAddr) -> bool { + let mut workers = workers.lock().expect("workers lock poisoned"); + if workers.contains_key(&worker_id) { + return false; + } + + workers.insert(worker_id, None); + info!( + %worker_id, + %peer, + connected_workers = workers.len(), + "Worker connected" + ); + true +} + +fn deregister(workers: &Workers, worker_id: Uuid, peer: SocketAddr) { + let mut workers = workers.lock().expect("workers lock poisoned"); + // Duplicates are refused at connect time, so the entry is guaranteed to be ours to remove. + workers.remove(&worker_id); + info!( + %worker_id, + %peer, + connected_workers = workers.len(), + "Worker disconnected" + ); +} diff --git a/breakwater-deich/src/collector/stats.rs b/breakwater-deich/src/collector/stats.rs new file mode 100644 index 0000000..ad8194c --- /dev/null +++ b/breakwater-deich/src/collector/stats.rs @@ -0,0 +1,251 @@ +//! Collector-side statistics: folds each worker's periodic snapshot into persistent, per-IP grand +//! totals that survive both worker and collector restarts. +//! +//! # Contract with the worker's aggregator +//! +//! Each worker runs breakwater's [`Statistics`] aggregator and streams its +//! [`StatisticsInformationEvent`] snapshots. We rely on two properties of that aggregator, which +//! hold as long as the worker keeps the connection open (one "session"): +//! +//! - `bytes_for_ip` and `denied_connections_for_ip` are **cumulative and monotonic within a +//! session** — they only ever grow. We therefore fold the *delta* since the worker's previous +//! snapshot into the grand totals, so a cumulative counter isn't counted many times over. +//! - `connections_for_ip` is a **live gauge** (it goes up and down as connections open/close), so +//! we sum the latest snapshots across workers rather than accumulating deltas. +//! +//! On disconnect a worker's baseline is dropped ([`CollectorStatistics::forget`]); its counters +//! reset to zero when it reconnects, and the first post-reconnect snapshot then counts in full from +//! that zero. +//! +//! [`Statistics`]: breakwater::statistics::Statistics + +use std::{collections::HashMap, net::IpAddr}; + +use breakwater::statistics::{STATS_REPORT_INTERVAL, StatisticsInformationEvent}; +use uuid::Uuid; + +/// Holds both the live, per-worker view and the persistent grand totals. +#[derive(Default)] +pub struct CollectorStatistics { + /// The latest snapshot from each connected worker, keyed by UUID. Used for the live connection + /// gauge and as the per-worker baseline for delta accumulation. Removed on disconnect. + latest_per_worker: HashMap, + + /// Monotonic grand totals, accumulated from per-worker deltas. Persisted to the save file and + /// seeded from it on startup, so the "big numbers" outlive any restart. + total_bytes_for_ip: HashMap, + total_denied_for_ip: HashMap, +} + +impl CollectorStatistics { + /// Seeds the grand totals from a previously saved snapshot (the live per-worker view always + /// starts empty — it's rebuilt as workers (re)connect). + pub fn from_save_point(save_point: StatisticsInformationEvent) -> Self { + Self { + total_bytes_for_ip: save_point.bytes_for_ip, + total_denied_for_ip: save_point.denied_connections_for_ip, + ..Default::default() + } + } + + /// Records a worker's fresh snapshot: folds the per-IP increase since its previous snapshot + /// (zero if this is its first) into the grand totals, then stores it as the new baseline. + pub fn record(&mut self, worker_id: Uuid, event: StatisticsInformationEvent) { + let previous = self.latest_per_worker.get(&worker_id); + + for (&ip, &bytes) in &event.bytes_for_ip { + let baseline = previous + .and_then(|p| p.bytes_for_ip.get(&ip)) + .copied() + .unwrap_or(0); + // Monotonic within a session, so `bytes >= baseline`; `saturating_sub` only guards the + // (shouldn't-happen) case of a counter going backwards without a disconnect in between. + *self.total_bytes_for_ip.entry(ip).or_default() += bytes.saturating_sub(baseline); + } + for (&ip, &denied) in &event.denied_connections_for_ip { + let baseline = previous + .and_then(|p| p.denied_connections_for_ip.get(&ip)) + .copied() + .unwrap_or(0); + let total = self.total_denied_for_ip.entry(ip).or_default(); + *total = total.saturating_add(denied.saturating_sub(baseline)); + } + + self.latest_per_worker.insert(worker_id, event); + } + + /// Drops a disconnected worker's baseline so its next session accumulates from zero again. Its + /// already-folded bytes stay in the grand totals. + pub fn forget(&mut self, worker_id: Uuid) { + self.latest_per_worker.remove(&worker_id); + } + + /// Builds the event published to the sinks: persistent grand totals for bytes/denied, plus the + /// live connection gauge summed across currently-connected workers. `previous_bytes` carries + /// the last tick's total so we can derive a per-second rate at the collector. + pub fn published_event(&self, previous_bytes: &mut u64) -> StatisticsInformationEvent { + let mut connections_for_ip: HashMap = HashMap::new(); + let mut statistic_events = 0; + for snapshot in self.latest_per_worker.values() { + for (&ip, &connections) in &snapshot.connections_for_ip { + *connections_for_ip.entry(ip).or_default() += connections; + } + statistic_events += snapshot.statistic_events; + } + + let connections = connections_for_ip.values().sum(); + let [ips_v6, ips_v4] = + connections_for_ip + .keys() + .fold([0u32, 0u32], |[v6, v4], ip| match ip { + IpAddr::V6(_) => [v6 + 1, v4], + IpAddr::V4(_) => [v6, v4 + 1], + }); + + let bytes: u64 = self.total_bytes_for_ip.values().sum(); + // Rate over one report interval, saturating since a worker dropping out can't shrink the + // (monotonic) total, but a freshly seeded total on startup can jump the first `previous`. + let elapsed_secs = STATS_REPORT_INTERVAL.as_secs().max(1); + let bytes_per_s = bytes.saturating_sub(*previous_bytes) / elapsed_secs; + *previous_bytes = bytes; + + StatisticsInformationEvent { + connections, + ips_v6, + ips_v4, + bytes, + bytes_per_s, + connections_for_ip, + denied_connections_for_ip: self.total_denied_for_ip.clone(), + bytes_for_ip: self.total_bytes_for_ip.clone(), + statistic_events, + // Workers don't render, so there's no frame/fps to report at the collector. + frame: 0, + fps: 0, + } + } + + /// Builds the event written to the save file: only the persistent grand totals matter (the live + /// view is rebuilt from reconnecting workers), so the other fields stay at their defaults. + pub fn save_point(&self) -> StatisticsInformationEvent { + StatisticsInformationEvent { + bytes: self.total_bytes_for_ip.values().sum(), + bytes_for_ip: self.total_bytes_for_ip.clone(), + denied_connections_for_ip: self.total_denied_for_ip.clone(), + ..Default::default() + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + /// Builds a worker snapshot carrying per-IP byte and (live) connection counts. + fn snapshot( + bytes_for_ip: &[(IpAddr, u64)], + connections_for_ip: &[(IpAddr, u32)], + ) -> StatisticsInformationEvent { + StatisticsInformationEvent { + bytes_for_ip: bytes_for_ip.iter().copied().collect(), + connections_for_ip: connections_for_ip.iter().copied().collect(), + statistic_events: 1, + ..Default::default() + } + } + + #[test] + fn accumulates_per_ip_totals_and_live_connections_across_workers() { + let v4: IpAddr = "1.2.3.4".parse().unwrap(); + let v6: IpAddr = "::1".parse().unwrap(); + let (w1, w2) = (Uuid::from_u128(1), Uuid::from_u128(2)); + + // `v4` hit both workers; `v6` only the second. Totals and the shared IP must add up. + let mut stats = CollectorStatistics::default(); + stats.record(w1, snapshot(&[(v4, 100)], &[(v4, 2)])); + stats.record(w2, snapshot(&[(v4, 50), (v6, 7)], &[(v4, 1), (v6, 3)])); + + let mut previous_bytes = 0; + let event = stats.published_event(&mut previous_bytes); + + // Bytes are the persistent grand total. + assert_eq!(event.bytes_for_ip[&v4], 150); + assert_eq!(event.bytes_for_ip[&v6], 7); + assert_eq!(event.bytes, 157); + // Connections are the live gauge, summed across currently-connected workers. + assert_eq!(event.connections_for_ip[&v4], 3); + assert_eq!(event.connections, 6); + assert_eq!(event.ips_v4, 1); + assert_eq!(event.ips_v6, 1); + assert_eq!(event.statistic_events, 2); + // First tick: the whole total counts as this interval's throughput. + assert_eq!( + event.bytes_per_s, + 157 / STATS_REPORT_INTERVAL.as_secs().max(1) + ); + assert_eq!(previous_bytes, 157); + } + + #[test] + fn folds_deltas_so_a_cumulative_counter_is_not_double_counted() { + let v4: IpAddr = "1.2.3.4".parse().unwrap(); + let w = Uuid::from_u128(1); + let mut stats = CollectorStatistics::default(); + + // A worker's counter is cumulative within a session: only the increase between consecutive + // snapshots is folded into the grand total. + stats.record(w, snapshot(&[(v4, 100)], &[])); + stats.record(w, snapshot(&[(v4, 175)], &[])); + assert_eq!(stats.total_bytes_for_ip[&v4], 175); + } + + #[test] + fn worker_restart_keeps_totals_without_dipping_or_double_counting() { + let v4: IpAddr = "1.2.3.4".parse().unwrap(); + let w = Uuid::from_u128(1); + let mut stats = CollectorStatistics::default(); + + stats.record(w, snapshot(&[(v4, 100)], &[(v4, 1)])); + // Worker disconnects: its baseline is dropped, but its folded bytes stay in the total. + stats.forget(w); + assert_eq!(stats.total_bytes_for_ip[&v4], 100); + + // It reconnects (same UUID) with a counter reset to zero. The first post-restart snapshot + // counts in full, so the total grows by exactly the new traffic — no dip, no double count. + stats.record(w, snapshot(&[(v4, 30)], &[(v4, 1)])); + assert_eq!(stats.total_bytes_for_ip[&v4], 130); + } + + #[test] + fn accumulates_denied_connections() { + let v4: IpAddr = "1.2.3.4".parse().unwrap(); + let w = Uuid::from_u128(1); + let denied = |n: u32| StatisticsInformationEvent { + denied_connections_for_ip: [(v4, n)].into_iter().collect(), + ..Default::default() + }; + + let mut stats = CollectorStatistics::default(); + stats.record(w, denied(3)); + stats.record(w, denied(5)); // cumulative -> grand total +2 + assert_eq!(stats.total_denied_for_ip[&v4], 5); + assert_eq!( + stats.published_event(&mut 0).denied_connections_for_ip[&v4], + 5 + ); + } + + #[test] + fn save_point_seeds_grand_totals_on_restart() { + let v4: IpAddr = "1.2.3.4".parse().unwrap(); + let w = Uuid::from_u128(1); + let mut stats = CollectorStatistics::default(); + stats.record(w, snapshot(&[(v4, 4096)], &[])); + + // Collector restart: persist, then seed a fresh instance from the save point. + let reseeded = CollectorStatistics::from_save_point(stats.save_point()); + assert_eq!(reseeded.total_bytes_for_ip[&v4], 4096); + // The live view starts empty; workers reconnect from zero and accumulate on top of the seed. + assert!(reseeded.latest_per_worker.is_empty()); + } +} diff --git a/breakwater-deich/src/lib.rs b/breakwater-deich/src/lib.rs new file mode 100644 index 0000000..1e65c43 --- /dev/null +++ b/breakwater-deich/src/lib.rs @@ -0,0 +1,5 @@ +pub mod cli_args; +pub mod collector; +pub mod metrics; +pub mod protocol; +pub mod worker; diff --git a/breakwater-deich/src/main.rs b/breakwater-deich/src/main.rs new file mode 100644 index 0000000..7b4d7fd --- /dev/null +++ b/breakwater-deich/src/main.rs @@ -0,0 +1,24 @@ +use breakwater_deich::cli_args::{CliArgs, Role}; +use clap::{CommandFactory, FromArgMatches}; +use color_eyre::eyre; + +#[tokio::main] +async fn main() -> eyre::Result<()> { + // We parse via `ArgMatches` (instead of `CliArgs::parse()`) so that `SinkCliArgs::validate` can use + // `value_source` to tell which sink options were actually passed on the command line. + let mut cmd = CliArgs::command(); + let matches = cmd.get_matches_mut(); + let args = CliArgs::from_arg_matches(&matches).unwrap_or_else(|e| e.exit()); + + breakwater::init_telemetry()?; + + match args.role { + Role::Worker(args) => breakwater_deich::worker::run(*args).await, + Role::Collector(args) => { + if let Err(e) = args.sinks.validate(&mut cmd, &matches) { + e.exit(); + } + breakwater_deich::collector::run(*args).await + } + } +} diff --git a/breakwater-deich/src/metrics.rs b/breakwater-deich/src/metrics.rs new file mode 100644 index 0000000..bee7072 --- /dev/null +++ b/breakwater-deich/src/metrics.rs @@ -0,0 +1,260 @@ +//! Per-frame performance metrics shared by the worker and collector. +//! +//! Everything here is measured once per *frame* (≈ fps × workers events/s), never per pixel, so it +//! stays entirely off the Pixelflut hot path — a couple of clock reads and a histogram observe per +//! frame. Metrics register into the global Prometheus registry; [`serve`] exposes it for scraping, +//! and each role also logs a periodic summary (see the `log_metrics` loops in `worker`/`collector`), +//! so the numbers are available with or without Prometheus. +//! +//! The worker and collector measure genuinely different things, so they keep separate metric sets +//! ([`WorkerMetrics`] vs [`CollectorMetrics`]); the plumbing they share — the exporter and the timing +//! buckets — lives here. + +use std::{net::SocketAddr, time::Duration}; + +use breakwater::statistics::StatisticsInformationEvent; +use color_eyre::eyre::{self, Context}; +use prometheus_exporter::prometheus::{ + Histogram, HistogramOpts, HistogramVec, IntCounter, IntGauge, IntGaugeVec, register_histogram, + register_histogram_vec, register_int_counter, register_int_gauge, register_int_gauge_vec, +}; +use tokio::sync::broadcast; +use tracing::info; +use uuid::Uuid; + +/// How often each role logs a metrics summary. +pub const METRICS_LOG_INTERVAL: Duration = Duration::from_secs(10); + +/// Buckets for frame-timing histograms (seconds). A coarse spread covers the fast path (sub-ms +/// frames) and the pathological tail, but the bulk is a fine 20 ms grid across 60–300 ms — the range +/// where real send/receive/latency times actually cluster. +/// +/// Without that grid nearly every sample lands in a single 64–128 ms bucket, so `histogram_quantile` +/// interpolates across its whole 64 ms width: the quantile curve sits pinned near the bucket edge +/// and then jumps in coarse steps instead of tracking the real latency. The 20 ms grid resolves it. +fn frame_timing_buckets() -> Vec { + let mut buckets = vec![0.001, 0.005, 0.01, 0.025, 0.05]; + // Fine 20 ms grid, 60 ms ..= 300 ms. + let mut ms: u32 = 60; + while ms <= 300 { + buckets.push(f64::from(ms) / 1000.0); + ms += 20; + } + buckets.extend([0.4, 0.5, 0.75, 1.0, 2.0]); + buckets +} + +/// Histogram options for a per-frame duration metric (in seconds), with the shared timing buckets. +fn frame_timing_opts(name: &str, help: &str) -> HistogramOpts { + HistogramOpts::new(name, help).buckets(frame_timing_buckets()) +} + +/// Starts the Prometheus HTTP endpoint serving the global registry. Shared by both roles. +pub fn serve(listen_address: SocketAddr) -> eyre::Result<()> { + prometheus_exporter::start(listen_address) + .with_context(|| format!("failed to start Prometheus exporter on {listen_address}"))?; + info!(%listen_address, "Serving Prometheus metrics"); + Ok(()) +} + +/// Worker-side metrics: how long each framebuffer push takes and how many frames we fell behind on. +pub struct WorkerMetrics { + frame_send: Histogram, + frames_lagged: IntCounter, +} + +impl WorkerMetrics { + pub fn new() -> eyre::Result { + Ok(Self { + frame_send: register_histogram!(frame_timing_opts( + "deich_worker_frame_send_seconds", + "Wall-clock time to push one framebuffer to the collector (marker + blob + flush)", + ))?, + frames_lagged: register_int_counter!( + "deich_worker_frames_lagged_total", + "Frames skipped versus the target FPS because a prior send overran its slot", + )?, + }) + } + + pub fn observe_send(&self, duration: Duration) { + self.frame_send.observe(duration.as_secs_f64()); + } + + pub fn add_lagged(&self, frames: u64) { + self.frames_lagged.inc_by(frames); + } + + /// Cumulative (send count, total send seconds); the log loop diffs these into a per-interval mean. + pub fn send_totals(&self) -> (u64, f64) { + ( + self.frame_send.get_sample_count(), + self.frame_send.get_sample_sum(), + ) + } + + pub fn lagged_total(&self) -> u64 { + self.frames_lagged.get() + } +} + +/// Collector-side metrics: per-worker receive time and end-to-end latency, plus total ingress bytes. +pub struct CollectorMetrics { + frame_recv: HistogramVec, + frame_latency: HistogramVec, + ingress_bytes: IntCounter, +} + +impl CollectorMetrics { + pub fn new() -> eyre::Result { + Ok(Self { + frame_recv: register_histogram_vec!( + frame_timing_opts( + "deich_collector_frame_recv_seconds", + "Time to read one framebuffer body off the wire, per worker", + ), + &["worker"], + )?, + frame_latency: register_histogram_vec!( + frame_timing_opts( + "deich_collector_frame_latency_seconds", + "Worker send-start to collector recv-done latency, per worker (needs clock sync)", + ), + &["worker"], + )?, + ingress_bytes: register_int_counter!( + "deich_collector_ingress_bytes_total", + "Total framebuffer bytes received from all workers", + )?, + }) + } + + /// Returns a recorder with this worker's per-label histogram handles cached, so the per-frame hot + /// path avoids a label-map lookup. + pub fn recorder(&self, worker_id: Uuid) -> FrameRecorder { + let label = worker_id.to_string(); + FrameRecorder { + recv: self.frame_recv.with_label_values(&[&label]), + latency: self.frame_latency.with_label_values(&[&label]), + ingress_bytes: self.ingress_bytes.clone(), + } + } + + pub fn ingress_bytes_total(&self) -> u64 { + self.ingress_bytes.get() + } +} + +/// Per-connection recorder holding one worker's cached metric handles. +pub struct FrameRecorder { + recv: Histogram, + latency: Histogram, + ingress_bytes: IntCounter, +} + +impl FrameRecorder { + pub fn observe_frame(&self, recv: Duration, latency: Duration, bytes: u64) { + self.recv.observe(recv.as_secs_f64()); + self.latency.observe(latency.as_secs_f64()); + self.ingress_bytes.inc_by(bytes); + } +} + +/// Collector-side per-IP client statistics — bytes, connections and denied connections per IP, plus +/// the connected-address counts. Exposed with the **same metric names and labels as breakwater** +/// (`breakwater_bytes`, `breakwater_connections`, …) so existing breakwater dashboards work against +/// the collector unchanged. +/// +/// Fed by the aggregated [`StatisticsInformationEvent`] the collector already publishes: `bytes` and +/// `denied_connections` are the persistent grand totals (monotonic, survive restarts), while +/// `connections` is the live gauge. (breakwater's VNC-only `breakwater_frame` has no meaning here — +/// workers don't render — so it is deliberately omitted.) +pub struct StatisticsMetrics { + statistics_information_rx: broadcast::Receiver, + + ips_v6: IntGauge, + ips_v4: IntGauge, + statistic_events: IntGauge, + + connections_for_ip: IntGaugeVec, + denied_connections_for_ip: IntGaugeVec, + bytes_for_ip: IntGaugeVec, +} + +impl StatisticsMetrics { + pub fn new( + statistics_information_rx: broadcast::Receiver, + ) -> eyre::Result { + Ok(Self { + statistics_information_rx, + ips_v6: register_int_gauge!( + "breakwater_ips_v6", + "Total number of connected IPv6 addresses" + )?, + ips_v4: register_int_gauge!( + "breakwater_ips_v4", + "Total number of connected IPv4 addresses" + )?, + statistic_events: register_int_gauge!( + "breakwater_statistic_events", + "Number of statistics events send internally" + )?, + connections_for_ip: register_int_gauge_vec!( + "breakwater_connections", + "Number of client connections per IP address", + &["ip"], + )?, + denied_connections_for_ip: register_int_gauge_vec!( + "breakwater_denied_connections", + "Number of denied connections per IP address because it tried to open too many connections", + &["ip"], + )?, + bytes_for_ip: register_int_gauge_vec!( + "breakwater_bytes", + "Number of bytes received per IP address", + &["ip"], + )?, + }) + } + + /// Updates the gauges from each published statistics event until the channel closes. A lag just + /// means intermediate snapshots were dropped — the next event is fully authoritative, so we skip + /// the gap and carry on. + pub async fn run(&mut self) { + loop { + match self.statistics_information_rx.recv().await { + Ok(event) => self.update(&event), + Err(broadcast::error::RecvError::Lagged(_)) => {} + Err(broadcast::error::RecvError::Closed) => return, + } + } + } + + fn update(&self, event: &StatisticsInformationEvent) { + self.ips_v6.set(i64::from(event.ips_v6)); + self.ips_v4.set(i64::from(event.ips_v4)); + self.statistic_events.set(event.statistic_events as i64); + + // Reset the per-IP vectors before repopulating: an IP that dropped out of the event (e.g. + // all its connections closed) would otherwise linger at its last value forever. This mirrors + // breakwater's exporter. + self.connections_for_ip.reset(); + for (ip, connections) in &event.connections_for_ip { + self.connections_for_ip + .with_label_values(&[&ip.to_string()]) + .set(i64::from(*connections)); + } + self.denied_connections_for_ip.reset(); + for (ip, denied) in &event.denied_connections_for_ip { + self.denied_connections_for_ip + .with_label_values(&[&ip.to_string()]) + .set(i64::from(*denied)); + } + self.bytes_for_ip.reset(); + for (ip, bytes) in &event.bytes_for_ip { + self.bytes_for_ip + .with_label_values(&[&ip.to_string()]) + .set(*bytes as i64); + } + } +} diff --git a/breakwater-deich/src/protocol.rs b/breakwater-deich/src/protocol.rs new file mode 100644 index 0000000..3d8cc4e --- /dev/null +++ b/breakwater-deich/src/protocol.rs @@ -0,0 +1,317 @@ +//! Worker ↔ collector sync protocol. +//! +//! A worker connects to the collector and announces itself with a handshake (magic) plus a +//! [`WorkerMessage::Hello`] carrying its persistent UUID. The collector replies with a +//! [`CollectorMessage::Config`] — it owns canvas geometry, frame rate and the timestamp epoch — and +//! the worker then streams full framebuffers (and periodic statistics) until the connection drops. +//! +//! Control messages are serialized with [`postcard`] and length-delimited (a little-endian `u32` +//! length prefix), so the message set can grow into richer enums over time. A framebuffer blob +//! (~12 MB) is deliberately *not* serialized: a [`WorkerMessage::Framebuffer`] marker is immediately +//! followed on the wire by [`WorkerConfig::frame_size_bytes`] raw bytes. [`Connection`] owns that +//! "marker then raw blob" contract, so no caller has to remember it. + +use std::{ + io, + net::SocketAddr, + time::{Duration, Instant}, +}; + +use breakwater::statistics::StatisticsInformationEvent; +use breakwater_parser::{TimeTrackingPixel, get_current_ns_since_unix_epoch, pixels_as_bytes_mut}; +use serde::{Deserialize, Serialize, de::DeserializeOwned}; +use tokio::{ + io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt}, + net::TcpStream, +}; +use uuid::Uuid; + +/// Identifies the deich sync protocol on the wire ("deic" in ASCII). The collector and workers are +/// always deployed together, so a single magic to reject obviously-wrong connections is enough; we +/// don't bother with version negotiation. +const MAGIC: u32 = 0x6465_6963; + +/// Upper bound on a (length-delimited) control message, to reject garbage before allocating. +/// Control messages are tiny; the big framebuffer blob is sent raw, outside this path. +const MAX_MESSAGE_SIZE: usize = 1024 * 1024; + +/// Messages the collector sends to a worker. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub enum CollectorMessage { + /// Reply to the worker's [`WorkerMessage::Hello`]; tells the worker how to configure itself. + Config(WorkerConfig), +} + +/// Messages a worker sends to the collector. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub enum WorkerMessage { + /// First message after the handshake: identifies which worker this connection belongs to. + Hello { worker_id: Uuid }, + + /// Marker that a full framebuffer follows on the wire as [`WorkerConfig::frame_size_bytes`] raw + /// bytes (not part of this serialized message). The per-pixel write timestamps live in the blob + /// itself (absolute, relative to the shared epoch), so nothing else needs to travel with it. + /// + /// `sent_at_ns` is the worker's wall-clock send-start time (ns since the UNIX epoch), used by the + /// collector to measure end-to-end latency. It relies on the same worker/collector clock sync the + /// per-pixel timestamps already assume. + Framebuffer { sent_at_ns: u64 }, + + /// A periodic statistics snapshot, already aggregated per IP by the worker (~once per second) + /// and cumulative within this connection's session. Self-contained — no raw bytes follow. + Statistics(StatisticsInformationEvent), +} + +/// Configuration the collector hands to each worker. The collector is the single source of truth for +/// canvas geometry, frame rate and the timestamp epoch; workers configure themselves from it. Extend +/// this when workers need more, e.g. their offset within a larger canvas. +#[derive(Debug, Clone, Copy, Serialize, Deserialize)] +pub struct WorkerConfig { + pub width: u32, + pub height: u32, + pub fps: u32, + + /// The collector's startup time (ns since the UNIX epoch). Workers use it as the zero point for + /// their per-pixel timestamps, so every worker's timestamps are mutually comparable at the + /// collector. A 40-bit µs offset from here lasts ~12.7 days of collector uptime. + pub epoch_ns_since_unix_epoch: u64, +} + +impl WorkerConfig { + /// Number of pixels in one framebuffer. + pub fn pixel_count(&self) -> usize { + self.width as usize * self.height as usize + } + + /// Number of bytes in one framebuffer blob. + pub fn frame_size_bytes(&self) -> usize { + self.pixel_count() * size_of::() + } +} + +/// The interval between framebuffer pushes (worker side) and renders (collector side) for a given +/// frame rate. Each node just runs a local timer at this cadence — there is no shared wall-clock +/// schedule and no cross-node clock coupling; ordering is resolved entirely by the per-pixel +/// timestamps in the framebuffer. +pub fn frame_period(fps: u32) -> Duration { + Duration::from_nanos(1_000_000_000 / u64::from(fps.max(1))) +} + +/// What a worker sent us: either a full framebuffer (already read off the wire into an owned buffer) +/// or a statistics snapshot. The one-shot [`WorkerMessage::Hello`] is consumed at [`accept`] time. +pub enum WorkerData { + Framebuffer { + pixels: Vec, + /// Time spent reading the blob off the wire — the transfer itself, not the idle wait for the + /// next frame that precedes it. + recv_duration: Duration, + /// Worker send-start to collector recv-done latency. Saturates to zero under clock skew (a + /// worker clock ahead of the collector's). + latency: Duration, + }, + Statistics(StatisticsInformationEvent), +} + +/// A length-delimited, postcard-framed connection over some byte stream. It owns the deich wire +/// format — the magic handshake, control-message framing, and the raw framebuffer-blob contract — so +/// the worker and collector only ever deal in typed messages. +pub struct Connection { + stream: S, +} + +impl Connection { + pub fn new(stream: S) -> Self { + Self { stream } + } + + /// Collector side: send the worker its config in reply to the hello. + pub async fn send_config(&mut self, config: WorkerConfig) -> io::Result<()> { + self.send_message(&CollectorMessage::Config(config)).await + } + + /// Worker side: push a full framebuffer — the marker (stamped with `sent_at_ns`) immediately + /// followed by its raw bytes, flushed together so a statistics message can never slip between the + /// two. + pub async fn send_framebuffer(&mut self, sent_at_ns: u64, bytes: &[u8]) -> io::Result<()> { + self.write_message(&WorkerMessage::Framebuffer { sent_at_ns }) + .await?; + self.stream.write_all(bytes).await?; + self.stream.flush().await + } + + /// Worker side: send a statistics snapshot. + pub async fn send_statistics(&mut self, event: StatisticsInformationEvent) -> io::Result<()> { + self.send_message(&WorkerMessage::Statistics(event)).await + } + + /// Collector side: read the next framebuffer or statistics message. For a framebuffer the raw + /// blob is read here into a fresh `pixel_count`-long buffer (allocated only when a framebuffer + /// actually arrives), so callers never touch the "raw bytes follow the marker" contract. A + /// second [`WorkerMessage::Hello`] mid-stream is a protocol error. + pub async fn recv_worker_data(&mut self, pixel_count: usize) -> io::Result { + match self.recv_message::().await? { + WorkerMessage::Framebuffer { sent_at_ns } => { + let mut pixels = vec![TimeTrackingPixel::default(); pixel_count]; + // Time only the blob read (the transfer), not the idle wait that `recv_message` + // above spent blocking for the next frame. + let started = Instant::now(); + self.stream + .read_exact(pixels_as_bytes_mut(&mut pixels)) + .await?; + let recv_duration = started.elapsed(); + let latency = Duration::from_nanos( + get_current_ns_since_unix_epoch().saturating_sub(sent_at_ns), + ); + Ok(WorkerData::Framebuffer { + pixels, + recv_duration, + latency, + }) + } + WorkerMessage::Statistics(event) => Ok(WorkerData::Statistics(event)), + WorkerMessage::Hello { worker_id } => Err(io::Error::new( + io::ErrorKind::InvalidData, + format!("unexpected second hello (worker id {worker_id}) mid-stream"), + )), + } + } + + /// Writes a length-delimited, postcard-encoded control message and flushes it. + async fn send_message(&mut self, message: &M) -> io::Result<()> { + self.write_message(message).await?; + self.stream.flush().await + } + + /// Writes a length-delimited, postcard-encoded control message *without* flushing (so a + /// framebuffer blob can be appended and the pair flushed as one). + async fn write_message(&mut self, message: &M) -> io::Result<()> { + let bytes = postcard::to_allocvec(message) + .map_err(|error| io::Error::new(io::ErrorKind::InvalidData, error))?; + let len = u32::try_from(bytes.len()) + .map_err(|_| io::Error::new(io::ErrorKind::InvalidData, "control message too large"))?; + + self.stream.write_u32_le(len).await?; + self.stream.write_all(&bytes).await + } + + /// Reads a length-delimited, postcard-encoded control message. + async fn recv_message(&mut self) -> io::Result { + let len = self.stream.read_u32_le().await? as usize; + if len > MAX_MESSAGE_SIZE { + return Err(io::Error::new( + io::ErrorKind::InvalidData, + format!("control message of {len} bytes exceeds the {MAX_MESSAGE_SIZE} byte cap"), + )); + } + + let mut bytes = vec![0; len]; + self.stream.read_exact(&mut bytes).await?; + postcard::from_bytes(&bytes) + .map_err(|error| io::Error::new(io::ErrorKind::InvalidData, error)) + } + + async fn send_magic(&mut self) -> io::Result<()> { + self.stream.write_u32_le(MAGIC).await?; + self.stream.flush().await + } + + async fn expect_magic(&mut self) -> io::Result<()> { + let magic = self.stream.read_u32_le().await?; + if magic != MAGIC { + return Err(io::Error::new( + io::ErrorKind::InvalidData, + format!("unexpected protocol magic {magic:#010x}, expected {MAGIC:#010x}"), + )); + } + + Ok(()) + } +} + +/// Worker side: connect to the collector, announce ourselves with `worker_id`, and read the config +/// the collector replies with. +pub async fn connect( + collector_address: SocketAddr, + worker_id: Uuid, +) -> io::Result<(Connection, WorkerConfig)> { + let mut connection = Connection::new(TcpStream::connect(collector_address).await?); + connection.send_magic().await?; + connection + .send_message(&WorkerMessage::Hello { worker_id }) + .await?; + + let CollectorMessage::Config(config) = connection.recv_message().await?; + Ok((connection, config)) +} + +/// Collector side: validate the handshake and read the worker's [`WorkerMessage::Hello`], returning +/// the connection and the worker's UUID so it can be attributed. +pub async fn accept(stream: TcpStream) -> io::Result<(Connection, Uuid)> { + let mut connection = Connection::new(stream); + connection.expect_magic().await?; + + let worker_id = match connection.recv_message::().await? { + WorkerMessage::Hello { worker_id } => worker_id, + WorkerMessage::Framebuffer { .. } => { + return Err(io::Error::new( + io::ErrorKind::InvalidData, + "expected a hello message but got a framebuffer", + )); + } + WorkerMessage::Statistics(_) => { + return Err(io::Error::new( + io::ErrorKind::InvalidData, + "expected a hello message but got statistics", + )); + } + }; + + Ok((connection, worker_id)) +} + +#[cfg(test)] +mod tests { + use breakwater_parser::{FrameBuffer, TimeTrackingFrameBuffer}; + + use super::*; + + /// A framebuffer push and a statistics snapshot survive the wire round-trip, and the collector + /// reads each back as the right [`WorkerData`] variant (with the blob contract handled for it). + #[tokio::test] + async fn framebuffer_and_statistics_round_trip() { + let (client, server) = tokio::io::duplex(1 << 16); + let mut worker = Connection::new(client); + let mut collector = Connection::new(server); + + // A 2x1 framebuffer with one pixel set, so we have a real blob to compare. + let fb = TimeTrackingFrameBuffer::new(2, 1, 0); + fb.set(0, 0, 0x00_00ff, 42); + + let sent_at_ns = get_current_ns_since_unix_epoch(); + worker + .send_framebuffer(sent_at_ns, fb.as_raw_bytes()) + .await + .unwrap(); + worker + .send_statistics(StatisticsInformationEvent { + bytes: 1234, + ..Default::default() + }) + .await + .unwrap(); + + match collector.recv_worker_data(2).await.unwrap() { + WorkerData::Framebuffer { pixels, .. } => { + assert_eq!(pixels.len(), 2); + assert_eq!(pixels[0].rgb(), 0x00_00ff); + assert_eq!(pixels[0].timestamp(), 42); + } + WorkerData::Statistics(_) => panic!("expected a framebuffer first"), + } + + match collector.recv_worker_data(2).await.unwrap() { + WorkerData::Statistics(event) => assert_eq!(event.bytes, 1234), + WorkerData::Framebuffer { .. } => panic!("expected statistics second"), + } + } +} diff --git a/breakwater-deich/src/worker.rs b/breakwater-deich/src/worker.rs new file mode 100644 index 0000000..94facbf --- /dev/null +++ b/breakwater-deich/src/worker.rs @@ -0,0 +1,256 @@ +//! The worker role: runs a Pixelflut server into a time-tracking framebuffer and streams that +//! framebuffer to the collector. +//! +//! The collector owns the canvas geometry, frame rate and timestamp epoch, so the worker fetches its +//! config from the collector before it can allocate the framebuffer and start serving. Each +//! `run_session` runs until the collector connection drops; the worker then tears everything down +//! and starts a fresh session, which transparently picks up a changed config (geometry, fps, or a +//! new epoch after a collector restart). + +use std::{ + fs, io, + net::SocketAddr, + path::Path, + sync::Arc, + time::{Duration, Instant}, +}; + +use breakwater::{ + server::Server, + statistics::{Statistics, StatisticsEvent, StatisticsInformationEvent, StatisticsSaveMode}, +}; +use breakwater_parser::{TimeTrackingFrameBuffer, get_current_ns_since_unix_epoch}; +use color_eyre::eyre::{self, Context}; +use tokio::{ + net::TcpStream, + sync::{broadcast, mpsc}, + time::{MissedTickBehavior, interval}, +}; +use tracing::{info, warn}; +use uuid::Uuid; + +use crate::{ + cli_args::WorkerCliArgs, + metrics::{self, METRICS_LOG_INTERVAL, WorkerMetrics}, + protocol::{self, Connection, WorkerConfig, frame_period}, +}; + +/// Backoff between worker sessions (also used while waiting for the collector to become reachable). +const SESSION_BACKOFF: Duration = Duration::from_secs(1); + +/// Runs the worker until Ctrl-C, restarting the session whenever the collector connection drops. +pub async fn run(args: WorkerCliArgs) -> eyre::Result<()> { + let worker_id = load_or_create_worker_id(&args.worker_id_file)?; + info!(%worker_id, "Starting worker"); + + // Created once for the whole process so the metrics accumulate across sessions. + let metrics = WorkerMetrics::new().context("failed to register worker metrics")?; + metrics::serve(args.prometheus.prometheus_listen_address)?; + + tokio::select! { + // `session_loop` never returns on its own — it just keeps (re)starting sessions. + result = session_loop(&args, worker_id, &metrics) => result, + // Neither does the metrics logger; it just runs alongside. + () = log_metrics(&metrics) => unreachable!("metrics log loop never returns"), + result = tokio::signal::ctrl_c() => { + result.context("failed to wait for CTRL + C")?; + info!("Received CTRL + C, shutting down"); + Ok(()) + } + } +} + +/// Logs a per-interval summary of the worker's send metrics (mean push time and frames lagged), +/// diffing the cumulative Prometheus counters between ticks. Never returns. +async fn log_metrics(metrics: &WorkerMetrics) { + let mut ticker = interval(METRICS_LOG_INTERVAL); + let (mut prev_count, mut prev_sum, mut prev_lagged) = (0u64, 0.0, 0u64); + + loop { + ticker.tick().await; + let (count, sum) = metrics.send_totals(); + let lagged = metrics.lagged_total(); + let frames = count - prev_count; + let mean_ms = if frames > 0 { + (sum - prev_sum) / frames as f64 * 1000.0 + } else { + 0.0 + }; + info!( + frames, + mean_send_ms = format!("{mean_ms:.1}"), + lagged = lagged - prev_lagged, + "Worker send metrics (last interval)" + ); + (prev_count, prev_sum, prev_lagged) = (count, sum, lagged); + } +} + +/// Runs sessions back-to-back forever; a session ending (connection lost, config change, error) is +/// just logged and followed by a fresh one after a short backoff. +async fn session_loop( + args: &WorkerCliArgs, + worker_id: Uuid, + metrics: &WorkerMetrics, +) -> eyre::Result<()> { + loop { + match run_session(args, worker_id, metrics).await { + Ok(()) => info!("Collector connection ended; restarting worker session"), + Err(error) => warn!(%error, "Worker session failed; restarting"), + } + tokio::time::sleep(SESSION_BACKOFF).await; + } +} + +/// One worker session: connect to the collector, build the framebuffer from its config, serve +/// Pixelflut into it, and stream it — until the collector connection drops (or the server stops). +/// +/// The server, stats aggregator and push loop all run as `select!` arms (not detached tasks), so +/// when the session ends — here, or because the whole worker is cancelled on Ctrl-C — they're all +/// dropped together. No teardown bookkeeping, no leaked tasks. +async fn run_session( + args: &WorkerCliArgs, + worker_id: Uuid, + metrics: &WorkerMetrics, +) -> eyre::Result<()> { + let (mut connection, config) = connect_to_collector(args.collector_address, worker_id).await; + info!(?config, "Received configuration from collector"); + + let fb = Arc::new(TimeTrackingFrameBuffer::new( + config.width as usize, + config.height as usize, + config.epoch_ns_since_unix_epoch, + )); + let (statistics_tx, statistics_rx) = mpsc::channel::(100); + + // Worker-local aggregator: it folds the server's raw per-connection events into a periodic, + // per-IP snapshot (~once per second) that we forward to the collector, which in turn merges the + // snapshots across all workers. Persisting/merging across workers is the collector's job, so the + // worker runs the aggregator with saving disabled. + let (statistics_information_tx, _) = broadcast::channel::(2); + let mut statistics = Statistics::new( + statistics_rx, + statistics_information_tx.clone(), + StatisticsSaveMode::Disabled, + ) + .context("failed to create statistics aggregator")?; + + let network_buffer_size = args + .network_listener + .network_buffer_size + .try_into() + // This should never happen as clap checks the range for us + .with_context(|| { + format!( + "invalid network buffer size: {}", + args.network_listener.network_buffer_size + ) + })?; + + let mut server = Server::new( + &args.network_listener.listen_addresses, + fb.clone(), + statistics_tx, + network_buffer_size, + args.network_listener.connections_per_ip, + ) + .await + .context("failed to start pixelflut server")?; + + tokio::select! { + result = server.start() => result.context("pixelflut server stopped")?, + result = statistics.run() => result.context("statistics aggregator stopped")?, + result = push_frames(&mut connection, &fb, config.fps, statistics_information_tx.subscribe(), metrics) => { + result.context("framebuffer and statistics sync to the collector stopped")?; + } + } + + Ok(()) +} + +/// Streams the framebuffer and periodic statistics to the collector over the one connection, until +/// it fails. Frames go out on a local timer at the configured rate; per-pixel timestamps are +/// absolute (set by the framebuffer at write time), so nothing needs re-basing and the collector +/// resolves ordering itself. Statistics snapshots (~once per second) are forwarded as they arrive. +/// +/// Both message kinds share this one task, so their writes can never interleave — a statistics +/// message can't slip between a framebuffer's marker and its raw blob. +async fn push_frames( + connection: &mut Connection, + fb: &TimeTrackingFrameBuffer, + fps: u32, + mut statistics_rx: broadcast::Receiver, + metrics: &WorkerMetrics, +) -> io::Result<()> { + let period = frame_period(fps); + let mut ticker = interval(period); + // If a push runs long (slow network), don't burst to catch up — just space the next one a full + // period out. The collector only cares about the latest frame anyway. + ticker.set_missed_tick_behavior(MissedTickBehavior::Delay); + + loop { + tokio::select! { + _ = ticker.tick() => { + let sent_at_ns = get_current_ns_since_unix_epoch(); + let started = Instant::now(); + connection.send_framebuffer(sent_at_ns, fb.as_raw_bytes()).await?; + let elapsed = started.elapsed(); + + metrics.observe_send(elapsed); + // A send spanning N whole frame-periods means the N-1 slots after the first went + // unsent — i.e. we couldn't keep up with the target FPS. + let spanned = (elapsed.as_nanos() / period.as_nanos().max(1)) as u64; + metrics.add_lagged(spanned.saturating_sub(1)); + } + event = statistics_rx.recv() => match event { + Ok(event) => connection.send_statistics(event).await?, + // Lagged: snapshots are emitted ~once per second into a small buffer with a single + // consumer, so lagging shouldn't happen; if it somehow does, skip the dropped ones. + // Closed: the aggregator is gone, but the whole session is torn down at that point + // (it runs as a sibling `select!` arm), so this is effectively unreachable. Either + // way: keep streaming frames rather than failing the sync. + Err(broadcast::error::RecvError::Lagged(_) | broadcast::error::RecvError::Closed) => {} + }, + } + } +} + +/// Connects to the collector, retrying with a backoff until it succeeds. +async fn connect_to_collector( + collector_address: SocketAddr, + worker_id: Uuid, +) -> (Connection, WorkerConfig) { + loop { + match protocol::connect(collector_address, worker_id).await { + Ok(result) => return result, + Err(error) => { + warn!( + %collector_address, + %error, + "Waiting for the collector to become reachable" + ); + tokio::time::sleep(SESSION_BACKOFF).await; + } + } + } +} + +/// Loads this worker's persistent UUID from `path`, generating and saving a fresh one on first run. +fn load_or_create_worker_id(path: &Path) -> eyre::Result { + match fs::read_to_string(path) { + Ok(contents) => contents + .trim() + .parse() + .with_context(|| format!("failed to parse worker id from {}", path.display())), + Err(error) if error.kind() == io::ErrorKind::NotFound => { + let worker_id = Uuid::new_v4(); + fs::write(path, worker_id.to_string()) + .with_context(|| format!("failed to persist worker id to {}", path.display()))?; + info!(%worker_id, path = %path.display(), "Generated and persisted a new worker id"); + Ok(worker_id) + } + Err(error) => { + Err(error).with_context(|| format!("failed to read worker id from {}", path.display())) + } + } +} diff --git a/breakwater-parser/src/framebuffer/time_tracking.rs b/breakwater-parser/src/framebuffer/time_tracking.rs index 612e9e9..656b726 100644 --- a/breakwater-parser/src/framebuffer/time_tracking.rs +++ b/breakwater-parser/src/framebuffer/time_tracking.rs @@ -37,6 +37,18 @@ impl TimeTrackingFrameBuffer { fn pixel_index(&self, x: usize, y: usize) -> usize { x + y * self.width } + + #[inline(always)] + pub fn as_raw_bytes(&self) -> &[u8] { + // The buffer is a contiguous `Vec`, so its raw bytes are exactly the + // 8-bytes-per-pixel wire layout. Like the other framebuffers, this reads memory that writers + // may be mutating concurrently — fine for a lossy, best-effort sync. + let len = self.buffer.len() * size_of::(); + let ptr = self.buffer.as_ptr().cast::(); + // SAFETY: `TimeTrackingPixel` is `repr(transparent)` over a `u64` (all bit patterns valid), + // so its bytes are a valid `[u8]` of the same length and lifetime as the shared borrow. + unsafe { std::slice::from_raw_parts(ptr, len) } + } } impl FrameBuffer for TimeTrackingFrameBuffer { @@ -123,3 +135,13 @@ pub fn get_current_ns_since_unix_epoch() -> u64 { "your system time is >= year 2554. I'm developing this in 2026, I'm very likely dead now. And did no one write a better server to use in all that years?", ) } + +/// Return a mutable slice of pixels as their raw bytes (8 per pixel), e.g. to read a frame straight +/// off the wire into a `Vec`. +pub fn pixels_as_bytes_mut(pixels: &mut [TimeTrackingPixel]) -> &mut [u8] { + let len = size_of_val(pixels); + let ptr = pixels.as_mut_ptr().cast::(); + // SAFETY: `TimeTrackingPixel` is `repr(transparent)` over a `u64` (all bit patterns valid), so + // its bytes are a valid `[u8]` of the same lifetime and exclusive borrow. + unsafe { std::slice::from_raw_parts_mut(ptr, len) } +} diff --git a/breakwater-parser/src/lib.rs b/breakwater-parser/src/lib.rs index 89a2f93..6588294 100644 --- a/breakwater-parser/src/lib.rs +++ b/breakwater-parser/src/lib.rs @@ -16,13 +16,14 @@ mod refactored; #[cfg(target_arch = "x86_64")] pub use assembler::AssemblerParser; -pub use framebuffer::time_tracking::{ - RGB_BITS, RGB_MASK, TIMESTAMP_BITS, TIMESTAMP_MAX, TimeTrackingFrameBuffer, TimeTrackingPixel, - get_current_ns_since_unix_epoch, -}; pub use framebuffer::{ FB_BYTES_PER_PIXEL, FrameBuffer, MultiPixelSet, PixelColorBytes, - shared_memory::SharedMemoryFrameBuffer, simple::SimpleFrameBuffer, + shared_memory::SharedMemoryFrameBuffer, + simple::SimpleFrameBuffer, + time_tracking::{ + RGB_BITS, RGB_MASK, TIMESTAMP_BITS, TIMESTAMP_MAX, TimeTrackingFrameBuffer, + TimeTrackingPixel, get_current_ns_since_unix_epoch, pixels_as_bytes_mut, + }, }; pub use memchr::MemchrParser; pub use original::{OriginalParser, OriginalParserFrameBuffer}; diff --git a/breakwater/Cargo.toml b/breakwater/Cargo.toml index 1e91929..199a14e 100644 --- a/breakwater/Cargo.toml +++ b/breakwater/Cargo.toml @@ -7,6 +7,8 @@ license.workspace = true edition.workspace = true repository.workspace = true +default-run = "breakwater" + [[bin]] name = "breakwater" path = "src/main.rs" diff --git a/breakwater/src/lib.rs b/breakwater/src/lib.rs index e840da9..45792ef 100644 --- a/breakwater/src/lib.rs +++ b/breakwater/src/lib.rs @@ -1,3 +1,5 @@ +use color_eyre::eyre; + pub mod cli_args; pub mod connection_buffer; #[cfg(feature = "prometheus")] @@ -6,6 +8,22 @@ pub mod server; pub mod sinks; pub mod statistics; +/// Shared process setup for the breakwater binaries: error reporting and logging. +pub fn init_telemetry() -> eyre::Result<()> { + color_eyre::install()?; + + let filter = tracing_subscriber::EnvFilter::builder() + .with_default_directive(if cfg!(debug_assertions) { + tracing::Level::DEBUG.into() + } else { + tracing::Level::INFO.into() + }) + .from_env()?; + tracing_subscriber::fmt().with_env_filter(filter).init(); + + Ok(()) +} + #[cfg(test)] pub mod test_helpers; #[cfg(test)] diff --git a/breakwater/src/main.rs b/breakwater/src/main.rs index 09cfdef..3ac0eb7 100644 --- a/breakwater/src/main.rs +++ b/breakwater/src/main.rs @@ -15,16 +15,7 @@ use breakwater::{ #[tokio::main] #[allow(clippy::too_many_lines)] async fn main() -> eyre::Result<()> { - color_eyre::install()?; - - let filter = tracing_subscriber::EnvFilter::builder() - .with_default_directive(if cfg!(debug_assertions) { - tracing::Level::DEBUG.into() - } else { - tracing::Level::INFO.into() - }) - .from_env()?; - tracing_subscriber::fmt().with_env_filter(filter).init(); + breakwater::init_telemetry()?; // We parse via `ArgMatches` (instead of `CliArgs::parse()`) so that `SinkCliArgs::validate` can use // `value_source` to tell which sink options were actually passed on the command line. diff --git a/breakwater/src/statistics.rs b/breakwater/src/statistics.rs index 8d49df6..9aa5de0 100644 --- a/breakwater/src/statistics.rs +++ b/breakwater/src/statistics.rs @@ -99,7 +99,7 @@ pub struct Statistics { } impl StatisticsInformationEvent { - fn save_to_file(&self, file_name: &str) -> eyre::Result<()> { + pub fn save_to_file(&self, file_name: &str) -> eyre::Result<()> { // TODO Check if we can use tokio's File here. This needs some integration with serde_json though // This operation is also called very infrequently let file = File::create(file_name) @@ -116,7 +116,7 @@ impl StatisticsInformationEvent { /// the file exists but can not be read or parsed. Otherwise a corrupted or outdated save file /// (e.g. after adding a new mandatory field) would be silently ignored and subsequently /// overwritten, losing all previously collected statistics. - fn load_from_file(file_name: &str) -> eyre::Result> { + pub fn load_from_file(file_name: &str) -> eyre::Result> { let file = match File::open(file_name) { Ok(file) => file, Err(e) if e.kind() == io::ErrorKind::NotFound => return Ok(None),