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 9a68b01..6e42936 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" @@ -795,6 +804,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" @@ -987,6 +1005,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" @@ -1074,6 +1098,22 @@ version = "2.11.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a4ae5f15dda3c708c0ade84bfee31ccab44a3da4f88015ed22f63732abe300c8" +[[package]] +name = "deich" +version = "0.21.0" +dependencies = [ + "breakwater", + "breakwater-parser", + "clap", + "color-eyre", + "postcard", + "serde", + "tokio", + "tracing", + "tracing-subscriber", + "uuid", +] + [[package]] name = "deranged" version = "0.5.8" @@ -1361,6 +1401,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" @@ -1564,6 +1616,12 @@ version = "1.0.7" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "3f9eec918d3f24069decb9af1554cad7c880e2da24a9afd88aca000531ab82c1" +[[package]] +name = "foldhash" +version = "0.1.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d9c4f5dac5e15c24eb999c26181a6ca40b39fe946cbe4c263c7209467bc83af2" + [[package]] name = "foldhash" version = "0.2.0" @@ -1739,8 +1797,21 @@ checksum = "899def5c37c4fd7b2664648c28120ecec138e4d395b459e5ca34f9cce2dd77fd" dependencies = [ "cfg-if", "libc", - "r-efi", + "r-efi 5.3.0", + "wasip2", +] + +[[package]] +name = "getrandom" +version = "0.4.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0de51e6874e94e7bf76d726fc5d13ba782deca734ff60d5bb2fb2607c7406555" +dependencies = [ + "cfg-if", + "libc", + "r-efi 6.0.0", "wasip2", + "wasip3", ] [[package]] @@ -1878,13 +1949,31 @@ 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.15.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9229cfe53dfd69f0609a49f65461bd93001ea1ef889cd5529dd176593f5338a1" +dependencies = [ + "foldhash 0.1.5", +] + [[package]] name = "hashbrown" version = "0.16.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "841d1cc9bed7f9236f321df977030373f4a4163ae1a7dbfe1a51a2c1a51d9100" dependencies = [ - "foldhash", + "foldhash 0.2.0", ] [[package]] @@ -1893,6 +1982,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" @@ -2026,6 +2129,12 @@ dependencies = [ "zerovec", ] +[[package]] +name = "id-arena" +version = "2.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3d3067d79b975e8844ca9eb072e16b31c3c1c36928edf9c6789548c524d0d954" + [[package]] name = "ident_case" version = "1.0.1" @@ -2117,6 +2226,8 @@ checksum = "d466e9454f08e4a911e14806c24e16fba1b4c121d1ea474396f396069cf949d9" dependencies = [ "equivalent", "hashbrown 0.17.1", + "serde", + "serde_core", ] [[package]] @@ -2347,6 +2458,12 @@ version = "1.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "830d08ce1d1d941e6b30645f1a0eb5643013d835ce3779a5fc208261dbe10f55" +[[package]] +name = "leb128fmt" +version = "0.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "09edd9e8b54e49e587e4f6295a7d29c3ea94d469cb40ab8ca70b288248a81db2" + [[package]] name = "lebe" version = "0.5.3" @@ -3479,6 +3596,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" @@ -3645,6 +3775,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" @@ -4322,6 +4458,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" @@ -4850,7 +4995,10 @@ version = "1.23.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d258b83ceec21034727ecee8c382cfa6c3e133699b0742c64571814fb420c9f7" dependencies = [ + "getrandom 0.4.2", + "js-sys", "serde_core", + "wasm-bindgen", ] [[package]] @@ -4940,7 +5088,16 @@ version = "1.0.3+wasi-0.2.9" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "20064672db26d7cdc89c7798c48a0fdfac8213434a1186e5ef29fd560ae223d6" dependencies = [ - "wit-bindgen", + "wit-bindgen 0.57.1", +] + +[[package]] +name = "wasip3" +version = "0.4.0+wasi-0.3.0-rc-2026-01-06" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5428f8bf88ea5ddc08faddef2ac4a67e390b88186c703ce6dbd955e1c145aca5" +dependencies = [ + "wit-bindgen 0.51.0", ] [[package]] @@ -4998,6 +5155,40 @@ dependencies = [ "unicode-ident", ] +[[package]] +name = "wasm-encoder" +version = "0.244.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "990065f2fe63003fe337b932cfb5e3b80e0b4d0f5ff650e6985b1048f62c8319" +dependencies = [ + "leb128fmt", + "wasmparser", +] + +[[package]] +name = "wasm-metadata" +version = "0.244.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bb0e353e6a2fbdc176932bbaab493762eb1255a7900fe0fea1a2f96c296cc909" +dependencies = [ + "anyhow", + "indexmap", + "wasm-encoder", + "wasmparser", +] + +[[package]] +name = "wasmparser" +version = "0.244.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "47b807c72e1bac69382b3a6fb3dbe8ea4c0ed87ff5629b8685ae6b9a611028fe" +dependencies = [ + "bitflags 2.11.1", + "hashbrown 0.15.5", + "indexmap", + "semver", +] + [[package]] name = "wayland-backend" version = "0.3.15" @@ -5786,12 +5977,100 @@ dependencies = [ "memchr", ] +[[package]] +name = "wit-bindgen" +version = "0.51.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d7249219f66ced02969388cf2bb044a09756a083d0fab1e566056b04d9fbcaa5" +dependencies = [ + "wit-bindgen-rust-macro", +] + [[package]] name = "wit-bindgen" version = "0.57.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1ebf944e87a7c253233ad6766e082e3cd714b5d03812acc24c318f549614536e" +[[package]] +name = "wit-bindgen-core" +version = "0.51.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ea61de684c3ea68cb082b7a88508a8b27fcc8b797d738bfc99a82facf1d752dc" +dependencies = [ + "anyhow", + "heck", + "wit-parser", +] + +[[package]] +name = "wit-bindgen-rust" +version = "0.51.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b7c566e0f4b284dd6561c786d9cb0142da491f46a9fbed79ea69cdad5db17f21" +dependencies = [ + "anyhow", + "heck", + "indexmap", + "prettyplease", + "syn", + "wasm-metadata", + "wit-bindgen-core", + "wit-component", +] + +[[package]] +name = "wit-bindgen-rust-macro" +version = "0.51.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0c0f9bfd77e6a48eccf51359e3ae77140a7f50b1e2ebfe62422d8afdaffab17a" +dependencies = [ + "anyhow", + "prettyplease", + "proc-macro2", + "quote", + "syn", + "wit-bindgen-core", + "wit-bindgen-rust", +] + +[[package]] +name = "wit-component" +version = "0.244.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9d66ea20e9553b30172b5e831994e35fbde2d165325bec84fc43dbf6f4eb9cb2" +dependencies = [ + "anyhow", + "bitflags 2.11.1", + "indexmap", + "log", + "serde", + "serde_derive", + "serde_json", + "wasm-encoder", + "wasm-metadata", + "wasmparser", + "wit-parser", +] + +[[package]] +name = "wit-parser" +version = "0.244.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ecc8ac4bc1dc3381b7f59c34f00b67e18f910c2c0f50015669dde7def656a736" +dependencies = [ + "anyhow", + "id-arena", + "indexmap", + "log", + "semver", + "serde", + "serde_derive", + "serde_json", + "unicode-xid", + "wasmparser", +] + [[package]] name = "writeable" version = "0.6.3" diff --git a/Cargo.toml b/Cargo.toml index 1b396d0..bca762e 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,9 +1,10 @@ [workspace] members = [ "breakwater-parser", - "breakwater", "breakwater-egui-overlay", - "breakwater-parser-c-bindings" + "breakwater-parser-c-bindings", + "breakwater", + "deich", ] resolver = "2" @@ -33,6 +34,7 @@ number_prefix = "0.4" ndi-sdk-sys = { git = "https://github.com/kleinesfilmroellchen/rust-ndi-sdk.git", branch = "main" } page_size = "0.6" pixelbomber = "1.1" +postcard = { version = "1", features = ["use-std"] } prometheus_exporter = "0.8" rstest = "0.26" rusttype = "0.9" @@ -47,13 +49,15 @@ 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", 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.21.0" } -breakwater-parser = { path = "breakwater-parser", version = "0.21.0" } +breakwater-parser = { path = "breakwater-parser", version = "0.21.0", default-features = false } breakwater-egui-overlay = { path = "breakwater-egui-overlay", version = "0.21.0" } +breakwater = { path = "breakwater", version = "0.21.0", default-features = false } [profile.dev] opt-level = 3 diff --git a/breakwater-parser/Cargo.toml b/breakwater-parser/Cargo.toml index 7fe8018..d28a202 100644 --- a/breakwater-parser/Cargo.toml +++ b/breakwater-parser/Cargo.toml @@ -27,6 +27,7 @@ rstest.workspace = true alpha = [] binary-set-pixel = [] binary-sync-pixels = [] +time-tracking = [] default = [] diff --git a/breakwater-parser/src/framebuffer/mod.rs b/breakwater-parser/src/framebuffer/mod.rs index bff496f..1ebeb8c 100644 --- a/breakwater-parser/src/framebuffer/mod.rs +++ b/breakwater-parser/src/framebuffer/mod.rs @@ -1,5 +1,7 @@ pub mod shared_memory; pub mod simple; +#[cfg(feature = "time-tracking")] +pub mod time_tracking; pub const FB_BYTES_PER_PIXEL: usize = std::mem::size_of::(); @@ -55,4 +57,15 @@ pub trait FrameBuffer { /// As the pixel memory doesn't necessarily need to be aligned (think of using shared memory for /// that), we can only return it as a list of bytes, not a list of pixels. fn as_bytes(&self) -> &[u8]; + + /// Encode `ns_since_unix_epoch` into the per-pixel timestamp representation. Compute it **once + /// per parse call** and pass the result to [`Self::set_with_pixel_timestamp`] for every pixel + /// (it's the same for all pixels in a call), keeping the encoding out of the hot per-pixel path. + #[cfg(feature = "time-tracking")] + fn pixel_timestamp(&self, ns_since_unix_epoch: u64) -> u64; + + /// Like [`Self::set`], but also records *when* the pixel was written, as the precomputed + /// `pixel_timestamp` from [`Self::pixel_timestamp`]. + #[cfg(feature = "time-tracking")] + fn set_with_pixel_timestamp(&self, x: usize, y: usize, rgba: u32, pixel_timestamp: u64); } diff --git a/breakwater-parser/src/framebuffer/shared_memory.rs b/breakwater-parser/src/framebuffer/shared_memory.rs index 81977e0..02b4630 100644 --- a/breakwater-parser/src/framebuffer/shared_memory.rs +++ b/breakwater-parser/src/framebuffer/shared_memory.rs @@ -222,4 +222,14 @@ impl FrameBuffer for SharedMemoryFrameBuffer { let base_ptr: *const u8 = self.buffer.as_ptr().cast(); unsafe { slice::from_raw_parts(base_ptr, self.bytes) } } + + #[cfg(feature = "time-tracking")] + fn pixel_timestamp(&self, _: u64) -> u64 { + panic!("The shared memory framebuffer does not support time tracking"); + } + + #[cfg(feature = "time-tracking")] + fn set_with_pixel_timestamp(&self, _: usize, _: usize, _: u32, _: u64) { + panic!("The shared memory framebuffer does not support time tracking"); + } } diff --git a/breakwater-parser/src/framebuffer/simple.rs b/breakwater-parser/src/framebuffer/simple.rs index d723111..3c45ca0 100644 --- a/breakwater-parser/src/framebuffer/simple.rs +++ b/breakwater-parser/src/framebuffer/simple.rs @@ -82,6 +82,16 @@ impl FrameBuffer for SimpleFrameBuffer { let ptr = self.buffer.as_ptr() as *const u8; unsafe { std::slice::from_raw_parts(ptr, len) } } + + #[cfg(feature = "time-tracking")] + fn pixel_timestamp(&self, _: u64) -> u64 { + panic!("The simple framebuffer does not support time tracking"); + } + + #[cfg(feature = "time-tracking")] + fn set_with_pixel_timestamp(&self, _: usize, _: usize, _: u32, _: u64) { + panic!("The simple framebuffer does not support time tracking"); + } } #[cfg(test)] diff --git a/breakwater-parser/src/framebuffer/time_tracking.rs b/breakwater-parser/src/framebuffer/time_tracking.rs new file mode 100644 index 0000000..cca3daf --- /dev/null +++ b/breakwater-parser/src/framebuffer/time_tracking.rs @@ -0,0 +1,183 @@ +use std::time::{SystemTime, UNIX_EPOCH}; + +use crate::FrameBuffer; + +/// Number of low bits holding the RGB color; the rest hold the timestamp. +const RGB_BITS: u32 = 24; +const RGB_MASK: u64 = (1 << RGB_BITS) - 1; + +/// Largest timestamp we can store in the remaining `64 - RGB_BITS = 40` bits. In microseconds +/// that's ≈ 12.7 days of collector uptime before it saturates — plenty, and a collector restart +/// resets the epoch anyway. +const TIMESTAMP_MAX: u64 = (1 << (u64::BITS - RGB_BITS)) - 1; + +/// A single pixel packed into one `u64`: the low [`RGB_BITS`] bits are the RGB color, the high bits +/// are an opaque timestamp. Packing both into a `u64` means writes are a single aligned 64-bit +/// store (the hot path). +/// +/// The timestamp is deliberately opaque here — its *meaning* (microseconds since the collector +/// epoch) is assigned solely by [`TimeTrackingFrameBuffer::pixel_timestamp`]. To everyone else it's +/// just a comparable number: larger = more recently written, and `0` (the default) = never written, +/// which compares as the oldest possible, so it never wins a merge. +#[derive(Debug, Default, Clone, Copy)] +#[repr(transparent)] +pub struct TimeTrackingPixel(u64); + +impl TimeTrackingPixel { + pub fn new(rgb: u32, timestamp: u64) -> Self { + Self((timestamp << RGB_BITS) | (u64::from(rgb) & RGB_MASK)) + } + + /// The 24-bit RGB color (the implicit alpha byte is always zero). + pub fn rgb(self) -> u32 { + (self.0 & RGB_MASK) as u32 + } + + /// The opaque write timestamp (see the type docs); larger is newer, `0` is never written. + pub fn timestamp(self) -> u64 { + self.0 >> RGB_BITS + } +} + +/// Views a slice of pixels as their raw bytes (8 per pixel), e.g. to read a frame straight off the +/// wire into a `Vec`. Same layout as [`FrameBuffer::as_bytes`]. +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) } +} + +pub struct TimeTrackingFrameBuffer { + width: usize, + height: usize, + buffer: Vec, + /// Per-pixel timestamps are microseconds since this epoch — the collector's startup time, in ns + /// since the UNIX epoch. Fixed for the framebuffer's life, so there is no re-basing. + epoch_ns_since_unix_epoch: u64, +} + +impl TimeTrackingFrameBuffer { + pub fn new(width: usize, height: usize, epoch_ns_since_unix_epoch: u64) -> Self { + let mut buffer = Vec::with_capacity(width * height); + buffer.resize_with(width * height, TimeTrackingPixel::default); + + Self { + width, + height, + buffer, + epoch_ns_since_unix_epoch, + } + } + + #[inline(always)] + fn pixel_index(&self, x: usize, y: usize) -> usize { + x + y * self.width + } +} + +impl FrameBuffer for TimeTrackingFrameBuffer { + #[inline(always)] + fn get_width(&self) -> usize { + self.width + } + + #[inline(always)] + fn get_height(&self) -> usize { + self.height + } + + #[inline(always)] + unsafe fn get_unchecked(&self, x: usize, y: usize) -> u32 { + let pixel_index = self.pixel_index(x, y); + unsafe { self.buffer.get_unchecked(pixel_index).rgb() } + } + + fn set(&self, _: usize, _: usize, _: u32) { + panic!( + "The time tracking framebuffer requires you to use the set_with_pixel_timestamp function!" + ); + } + + fn pixel_timestamp(&self, ns_since_unix_epoch: u64) -> u64 { + // The per-parse-call timestamp, computed once. `saturating_sub` guards against a worker + // whose clock trails the collector's epoch slightly; `.min` clamps the ~12.7 day range. + (ns_since_unix_epoch.saturating_sub(self.epoch_ns_since_unix_epoch) / 1_000) + .min(TIMESTAMP_MAX) + } + + fn set_with_pixel_timestamp(&self, x: usize, y: usize, rgba: u32, pixel_timestamp: u64) { + if x < self.width && y < self.height { + let pixel_index = self.pixel_index(x, y); + // A single aligned 64-bit store (the whole point of packing the pixel into a `u64`). + unsafe { + let ptr: *mut TimeTrackingPixel = self.buffer.as_ptr().add(pixel_index).cast_mut(); + *ptr = TimeTrackingPixel::new(rgba, pixel_timestamp); + } + } + } + + fn set_multi_from_start_index(&self, _: usize, _: &[u8]) -> usize { + panic!("The time tracking framebuffer does not implement set_multi_from_start_index"); + } + + fn as_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::(); + unsafe { std::slice::from_raw_parts(ptr, len) } + } +} + +/// Don't call too often, there is a cost involved! +pub fn get_current_ns_since_unix_epoch() -> u64 { + let ns_since_unix_epoch = SystemTime::now() + .duration_since(UNIX_EPOCH) + // TODO: If this turns out to be correct, convert to `unwrap_unchecked` + .expect("your system clock must be after UNIX EPOCH (so greater than 0)") + .as_nanos(); + + // u64::MAX allows us 18446744073709551615 ns since UNIX_EPOCH, which is + // some time in the year 2554, well beyond any reasonable timestamp. + u64::try_from(ns_since_unix_epoch).expect( + "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?", + ) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn pixel_packs_rgb_and_timestamp() { + let pixel = TimeTrackingPixel::new(0x12_3456, 1_000_000); + assert_eq!(pixel.rgb(), 0x12_3456); + assert_eq!(pixel.timestamp(), 1_000_000); + } + + #[test] + fn pixel_timestamp_is_micros_since_epoch_clamped() { + let fb = TimeTrackingFrameBuffer::new(1, 1, 1_000_000); + + // 1_000_000 ns after the epoch -> 1000 µs. + assert_eq!(fb.pixel_timestamp(2_000_000), 1_000); + // A timestamp before the epoch saturates to 0 (clock skew guard). + assert_eq!(fb.pixel_timestamp(0), 0); + // Beyond the 40-bit range it clamps. + assert_eq!(fb.pixel_timestamp(u64::MAX), TIMESTAMP_MAX); + } + + #[test] + fn set_and_read_back_pixel() { + let fb = TimeTrackingFrameBuffer::new(4, 4, 0); + fb.set_with_pixel_timestamp(1, 2, 0x00_ff00, 42); + + let pixel = fb.buffer[fb.pixel_index(1, 2)]; + assert_eq!(pixel.rgb(), 0x00_ff00); + assert_eq!(pixel.timestamp(), 42); + assert_eq!(fb.get(1, 2), Some(0x00_ff00)); + } +} diff --git a/breakwater-parser/src/lib.rs b/breakwater-parser/src/lib.rs index dceadd5..f4f6245 100644 --- a/breakwater-parser/src/lib.rs +++ b/breakwater-parser/src/lib.rs @@ -12,6 +12,11 @@ mod refactored; #[cfg(target_arch = "x86_64")] pub use assembler::AssemblerParser; +#[cfg(feature = "time-tracking")] +pub use framebuffer::time_tracking::{ + TimeTrackingFrameBuffer, TimeTrackingPixel, get_current_ns_since_unix_epoch, + pixels_as_bytes_mut, +}; pub use framebuffer::{ FB_BYTES_PER_PIXEL, FrameBuffer, shared_memory::SharedMemoryFrameBuffer, simple::SimpleFrameBuffer, @@ -57,3 +62,8 @@ pub trait Parser { // Sadly this cant be const (yet?) (https://github.com/rust-lang/rust/issues/71971 and https://github.com/rust-lang/rfcs/pull/2632) fn parser_lookahead(&self) -> usize; } + +#[cfg(all(feature = "time-tracking", feature = "alpha"))] +compile_error!("The features time-tracking and alpha can not be combined"); +#[cfg(all(feature = "time-tracking", feature = "binary-sync-pixels"))] +compile_error!("The features time-tracking and binary-sync-pixels can not be combined"); diff --git a/breakwater-parser/src/original.rs b/breakwater-parser/src/original.rs index 06de549..ac14fa7 100644 --- a/breakwater-parser/src/original.rs +++ b/breakwater-parser/src/original.rs @@ -53,6 +53,15 @@ impl Parser for OriginalParser { let mut i = 0; // We can't use a for loop here because Rust don't lets use skip characters by incrementing i let loop_end = buffer.len().saturating_sub(PARSER_LOOKAHEAD); // Let's extract the .len() call and the subtraction into it's own variable so we only compute it once + // As this is a potentially(?) expensive operation we only call it one in this parsing loop + // All the pixels likely where in the same TCP packets (+- 1/2 or so) it doesn't matter after all + // Encode the timestamp exactly once here, not per pixel: it's constant for the whole parse + // call, so computing it per write would just waste throughput on the hot path. + #[cfg(feature = "time-tracking")] + let pixel_timestamp = self + .fb + .pixel_timestamp(crate::framebuffer::time_tracking::get_current_ns_since_unix_epoch()); + #[cfg(feature = "binary-sync-pixels")] if let Some(remaining) = &self.remaining_pixel_sync { let buffer = &buffer[0..loop_end]; @@ -118,7 +127,16 @@ impl Parser for OriginalParser { let rgba: u32 = simd_unhex(unsafe { buffer.as_ptr().add(i - 7) }); + #[cfg(not(feature = "time-tracking"))] self.fb.set(x, y, rgba & 0x00ff_ffff); + #[cfg(feature = "time-tracking")] + self.fb.set_with_pixel_timestamp( + x, + y, + rgba & 0x00ff_ffff, + pixel_timestamp, + ); + continue; } @@ -130,7 +148,16 @@ impl Parser for OriginalParser { let rgba: u32 = simd_unhex(unsafe { buffer.as_ptr().add(i - 9) }); + #[cfg(not(feature = "time-tracking"))] self.fb.set(x, y, rgba & 0x00ff_ffff); + #[cfg(feature = "time-tracking")] + self.fb.set_with_pixel_timestamp( + x, + y, + rgba & 0x00ff_ffff, + pixel_timestamp, + ); + continue; } #[cfg(feature = "alpha")] @@ -159,6 +186,7 @@ impl Parser for OriginalParser { let blue: u32 = (((current >> 8) & 0xff) * alpha_comp + blue * alpha) / 0xff; + // Note: We have a compile check to make sure time-tracking and alpha can not be combined self.fb.set(x, y, (red << 16) | (green << 8) | blue); continue; } @@ -172,7 +200,11 @@ impl Parser for OriginalParser { let rgba: u32 = (base << 16) | (base << 8) | base; + #[cfg(not(feature = "time-tracking"))] self.fb.set(x, y, rgba); + #[cfg(feature = "time-tracking")] + self.fb + .set_with_pixel_timestamp(x, y, rgba, pixel_timestamp); continue; } @@ -208,7 +240,16 @@ impl Parser for OriginalParser { let rgba = u32::from_le((command_bytes >> 32) as u32); // TODO: Support alpha channel (behind alpha feature flag) + #[cfg(not(feature = "time-tracking"))] self.fb.set(x as usize, y as usize, rgba & 0x00ff_ffff); + #[cfg(feature = "time-tracking")] + self.fb.set_with_pixel_timestamp( + x as usize, + y as usize, + rgba & 0x00ff_ffff, + pixel_timestamp, + ); + // P B XX YY RGBA last_byte_parsed = i + 1 + 2 + 2 + 4; i += 10; @@ -228,6 +269,7 @@ impl Parser for OriginalParser { if len_in_bytes <= bytes_left_in_buffer { // Easy going here + // Note: We have a compile check to make sure time-tracking and binary-sync-pixels can not be combined self.fb .set_multi(start_x as usize, start_y as usize, unsafe { slice::from_raw_parts(buffer.as_ptr().add(i), len_in_bytes) diff --git a/breakwater/src/lib.rs b/breakwater/src/lib.rs new file mode 100644 index 0000000..b20edd3 --- /dev/null +++ b/breakwater/src/lib.rs @@ -0,0 +1,28 @@ +use color_eyre::eyre::{self, Context}; +use tokio::sync::broadcast; + +pub mod cli_args; +pub mod prometheus_exporter; +pub mod server; +pub mod sinks; +pub mod statistics; + +mod connection_buffer; + +#[cfg(test)] +mod test_helpers; + +#[cfg(test)] +mod tests; + +pub async fn handle_ctrl_c(terminate_signal_tx: broadcast::Sender<()>) -> eyre::Result<()> { + tokio::signal::ctrl_c() + .await + .context("failed to wait for ctrl + c")?; + + terminate_signal_tx + .send(()) + .context("failed to signal termination")?; + + Ok(()) +} diff --git a/breakwater/src/main.rs b/breakwater/src/main.rs index e48b6da..c634e17 100644 --- a/breakwater/src/main.rs +++ b/breakwater/src/main.rs @@ -1,29 +1,17 @@ use std::sync::Arc; -use breakwater_parser::SharedMemoryFrameBuffer; -use clap::Parser; -use color_eyre::eyre::{self, Context}; -use server::Server; -use tokio::sync::{broadcast, mpsc}; - -use crate::{ +use breakwater::{ cli_args::CliArgs, + handle_ctrl_c, prometheus_exporter::PrometheusExporter, + server::Server, sinks::{DisplaySink, ffmpeg::FfmpegSink}, statistics::{Statistics, StatisticsEvent, StatisticsInformationEvent, StatisticsSaveMode}, }; - -mod cli_args; -mod connection_buffer; -mod prometheus_exporter; -mod server; -mod sinks; -mod statistics; -#[cfg(test)] -mod test_helpers; - -#[cfg(test)] -mod tests; +use breakwater_parser::SharedMemoryFrameBuffer; +use clap::Parser; +use color_eyre::eyre::{self, Context}; +use tokio::sync::{broadcast, mpsc}; #[tokio::main] #[allow(clippy::too_many_lines)] @@ -97,17 +85,12 @@ async fn main() -> eyre::Result<()> { #[cfg(all(feature = "native-display", not(feature = "egui")))] { - use crate::sinks::native_display::NativeDisplaySink; + use breakwater::sinks::native_display::NativeDisplaySink; - if let Some(native_display_sink) = NativeDisplaySink::new( - fb.clone(), - &args, - statistics_tx.clone(), - statistics_information_rx.resubscribe(), - terminate_signal_rx.resubscribe(), - ) - .await - .context("failed to create native display sink")? + if let Some(native_display_sink) = + NativeDisplaySink::new(fb.clone(), terminate_signal_rx.resubscribe()) + .await + .context("failed to create native display sink")? { display_sinks.push(Box::new(native_display_sink)); } @@ -115,7 +98,7 @@ async fn main() -> eyre::Result<()> { #[cfg(feature = "vnc")] { - use crate::sinks::vnc::VncSink; + use breakwater::sinks::vnc::VncSink; if let Some(vnc_sink) = VncSink::new( fb.clone(), @@ -134,7 +117,7 @@ async fn main() -> eyre::Result<()> { #[cfg(feature = "ndi")] { - use crate::sinks::ndi::NdiSink; + use breakwater::sinks::ndi::NdiSink; if let Some(ndi_sink) = NdiSink::new( fb.clone(), @@ -171,7 +154,7 @@ async fn main() -> eyre::Result<()> { #[cfg(feature = "egui")] { - use sinks::egui::EguiSink; + use breakwater::sinks::egui::EguiSink; match EguiSink::new( fb.clone(), @@ -223,15 +206,3 @@ async fn main() -> eyre::Result<()> { Ok(()) } - -async fn handle_ctrl_c(terminate_signal_tx: broadcast::Sender<()>) -> eyre::Result<()> { - tokio::signal::ctrl_c() - .await - .context("failed to wait for ctrl + c")?; - - terminate_signal_tx - .send(()) - .context("failed to signal termination")?; - - Ok(()) -} diff --git a/breakwater/src/sinks/mod.rs b/breakwater/src/sinks/mod.rs index 198b070..cb02955 100644 --- a/breakwater/src/sinks/mod.rs +++ b/breakwater/src/sinks/mod.rs @@ -13,6 +13,9 @@ pub mod vnc; // The stabilization of async functions in traits in Rust 1.75 did not include support for using traits containing async // functions as dyn Trait, so we still need to use async_trait here. +// +// Each sink has its own inherent `new(..)` constructor (their arguments differ), returning +// `Ok(None)` when the sink isn't configured. This trait only carries the shared run loop. #[async_trait] pub trait DisplaySink { async fn run(&mut self) -> eyre::Result<()>; diff --git a/breakwater/src/sinks/vnc.rs b/breakwater/src/sinks/vnc.rs index 0bddece..544d6db 100644 --- a/breakwater/src/sinks/vnc.rs +++ b/breakwater/src/sinks/vnc.rs @@ -61,6 +61,7 @@ pub struct VncSink<'a, FB: FrameBuffer> { } impl VncSink<'_, FB> { + /// Returns `Ok(None)` if no VNC listen addresses are configured (the sink is then disabled). pub fn new( fb: Arc, VncSinkCliArgs { @@ -79,7 +80,7 @@ impl VncSink<'_, FB> { } // We ship our own copy of Arial.ttf, so that users don't need to download and provide it - let font = if font.as_str() == "Arial.ttf" { + let font = if font == "Arial.ttf" { let font_bytes = include_bytes!("../../../Arial.ttf"); Font::try_from_bytes(font_bytes).context("failed to load default font")? } else { @@ -129,7 +130,6 @@ impl VncSink<'_, FB> { } } - // FIXME: Only return Some in case VNC is enabled Ok(Some(Self { fb, statistics_tx, diff --git a/deich/Cargo.toml b/deich/Cargo.toml new file mode 100644 index 0000000..a4c75fa --- /dev/null +++ b/deich/Cargo.toml @@ -0,0 +1,33 @@ +[package] +name = "deich" +description = "Distributed Pixelflut server" +version.workspace = true +authors.workspace = true +license.workspace = true +edition.workspace = true +repository.workspace = true + +[[bin]] +name = "deich" +path = "src/main.rs" + +[features] +egui = ["breakwater/egui"] +native-display = ["breakwater/native-display"] +vnc = ["breakwater/vnc"] + +[dependencies] +breakwater-parser = { workspace = true, features = ["time-tracking"] } +breakwater.workspace = true + +clap.workspace = true +color-eyre.workspace = true +postcard.workspace = true +serde.workspace = true +tokio.workspace = true +tracing.workspace = true +tracing-subscriber.workspace = true +uuid.workspace = true + +[lints] +workspace = true diff --git a/deich/src/cli_args.rs b/deich/src/cli_args.rs new file mode 100644 index 0000000..9903e15 --- /dev/null +++ b/deich/src/cli_args.rs @@ -0,0 +1,86 @@ +use std::{net::SocketAddr, path::PathBuf}; + +use breakwater::cli_args::DEFAULT_NETWORK_BUFFER_SIZE_STR; +use clap::{Args, Parser, Subcommand}; + +/// `deich` is breakwater's distributed mode: many workers each run a Pixelflut server and sync +/// their framebuffer to a single collector. +#[derive(Parser, Debug)] +#[clap(author, version, about, long_about = None)] +pub struct Cli { + #[command(subcommand)] + pub role: Role, +} + +#[derive(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(WorkerArgs), + + /// Run the collector: gathers and merges the framebuffers of all workers. + Collector(CollectorArgs), +} + +#[derive(Args, Debug)] +pub struct WorkerArgs { + /// Listen address to bind the Pixelflut server to (multiple can be specified). + /// The default value will listen on all interfaces for IPv4 and IPv6 packets. + #[clap(short, long = "listen-address", default_value = "[::]:1234")] + pub listen_addresses: Vec, + + /// The size in bytes of the network buffer used for each open TCP connection. + /// Please use at least 64 KB (64_000 bytes). + #[clap( + long, + default_value = DEFAULT_NETWORK_BUFFER_SIZE_STR, + value_parser = 64_000..100_000_000, + )] + pub network_buffer_size: i64, + + /// Address of the collector to fetch the config from and stream the framebuffer to. + #[clap(long, default_value = "127.0.0.1:1235")] + pub collector_address: String, + + /// 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, +} + +#[derive(Args, Debug)] +pub struct CollectorArgs { + /// 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 ffmpeg_sink: breakwater::sinks::ffmpeg::FfmpegSinkCliArgs, + + /// Enable native display output. This requires some form of graphical system (so will probably not work on your + /// server). + #[cfg(any(feature = "native-display", feature = "egui"))] + #[clap(long)] + pub native_display: bool, + + #[cfg(feature = "egui")] + #[clap(flatten)] + pub egui_sink: breakwater::sinks::egui::EguiSinkCliArgs, + + #[cfg(feature = "vnc")] + #[clap(flatten)] + pub vnc_sink: breakwater::sinks::vnc::VncSinkCliArgs, +} diff --git a/deich/src/collector.rs b/deich/src/collector.rs new file mode 100644 index 0000000..8405628 --- /dev/null +++ b/deich/src/collector.rs @@ -0,0 +1,597 @@ +//! The collector role: identifies each connecting worker (via its [`Hello`]), tracks the connected +//! set, and stores the frames they stream. +//! +//! A **master** task ticks on the shared [`FrameSchedule`]. Each tick it publishes the window of +//! frame numbers it currently wants — `[render_frame ..= current]`, where `render_frame` is the +//! oldest frame that has had the full streaming budget to arrive — into the shared [`FrameStore`]. +//! Worker connections check that window per incoming frame: in-window frames are stored, others are +//! warned about and discarded. The master then merges `render_frame`'s stored frames into a +//! long-term [`Canvas`] by exact (full-`u64`-timestamp) last-write-per-pixel. +//! +//! [`Hello`]: crate::sync::WorkerMessage::Hello + +use std::{ + collections::HashMap, + io, + net::SocketAddr, + ops::RangeInclusive, + sync::{Arc, Mutex, RwLock}, + time::Duration, +}; + +use breakwater::{ + sinks::{DisplaySink, ffmpeg::FfmpegSink}, + statistics::{StatisticsEvent, StatisticsInformationEvent}, +}; +use breakwater_parser::{ + FrameBuffer, SimpleFrameBuffer, TimeTrackingPixel, get_current_ns_since_unix_epoch, + pixels_as_bytes_mut, +}; +use color_eyre::eyre::{self, Context}; +use tokio::{ + net::{TcpListener, TcpStream}, + sync::{broadcast, mpsc}, +}; +use tracing::{debug, info, warn}; +use uuid::Uuid; + +use crate::{ + cli_args::CollectorArgs, + sync::{self, FrameSchedule, WorkerConfig}, +}; + +/// How long we assume a worker needs to stream a frame to us (on top of the slot the worker spends +/// finishing it). Together with that slot, this sets how far behind "now" the master renders, so a +/// frame has had time to arrive — see the `margin` computation in [`run_master`]. +const FRAME_STREAM_BUDGET: Duration = Duration::from_millis(50); + +/// The workers currently connected, keyed by their UUID (value: the connection's peer address, for +/// logging). A duplicate UUID is refused at connect time, so each entry maps to exactly one live +/// connection. +/// +/// Reads happen every master tick while writes happen only on (dis)connect, hence a read-optimized +/// `RwLock`. It's a `std` (sync) lock on purpose: the critical sections are a single map op and +/// never cross an `.await`, so an async `tokio::sync::RwLock` would only add overhead. +type ConnectedWorkers = Arc>>; + +/// The long-term merged canvas, held by the master for the whole process lifetime. +/// +/// Each pixel keeps its full write timestamp (the high bits of [`TimeTrackingPixel`]), so +/// last-write-wins is exact over arbitrary time and survives traffic gaps and worker restarts. +/// It's a flat pixel vector — no width/height; merging is purely per-index. +struct Canvas { + pixels: Vec, +} + +impl Canvas { + fn new(width: usize, height: usize) -> Self { + Self { + pixels: vec![TimeTrackingPixel::default(); width * height], + } + } + + /// Folds `frame` in, keeping for each pixel whichever write has the larger timestamp. Since the + /// default timestamp is `0` (oldest possible), a never-written pixel — a blank or restarted + /// worker's frame — never clobbers fresher content. + 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; + } + } + } + + fn draw_to_framebuffer(&self, fb: &Arc) { + let pixels = self + .pixels + .iter() + .flat_map(|pixel| pixel.rgb().to_le_bytes()) + .collect::>(); + fb.set_multi_from_start_index(0, &pixels); + } +} + +/// How an incoming frame relates to the window of frames the master currently wants. +enum FrameInterest { + /// In the window — store it. + Wanted, + /// Outside the window. Carries the window so the caller can tell whether the frame is too old + /// (arrived late) or from the future, and by how much. + OutsideWindow { window: RangeInclusive }, + /// The master hasn't published a window yet (just started up). + NoWindowYet, +} + +/// Shared between the master task and the worker-connection tasks. +struct FrameStore { + /// Frame numbers the master currently wants: the inclusive window `[render_frame ..= current]`. + /// Read by every worker connection per frame; written by the master once per tick. `None` + /// until the master's first tick. + interesting: RwLock>>, + + /// Stored frames: `frame_number -> (worker_id -> framebuffer)`. Written by workers, read and + /// evicted by the master. + frames: Mutex>>>, +} + +impl FrameStore { + fn new() -> Self { + Self { + interesting: RwLock::new(None), + frames: Mutex::new(HashMap::new()), + } + } + + /// Classifies an incoming frame against the window the master currently wants. + fn classify(&self, frame_number: u64) -> FrameInterest { + match self + .interesting + .read() + .expect("interesting-window lock poisoned") + .as_ref() + { + None => FrameInterest::NoWindowYet, + Some(window) if window.contains(&frame_number) => FrameInterest::Wanted, + Some(window) => FrameInterest::OutsideWindow { + window: window.clone(), + }, + } + } + + /// Stores a worker's frame (overwriting any previous frame from the same worker for that slot). + fn store(&self, frame_number: u64, worker_id: Uuid, frame: Vec) { + self.frames + .lock() + .expect("frame store lock poisoned") + .entry(frame_number) + .or_default() + .insert(worker_id, frame); + } + + /// Publishes the new interesting `window`, evicts frames below it (they missed their render), + /// and removes & returns `render_frame`'s frames so the caller can merge them without holding + /// the lock. + fn advance( + &self, + window: RangeInclusive, + render_frame: u64, + ) -> HashMap> { + *self + .interesting + .write() + .expect("interesting-window lock poisoned") = Some(window); + + let mut frames = self.frames.lock().expect("frame store lock poisoned"); + frames.retain(|&frame_number, _| frame_number >= render_frame); + frames.remove(&render_frame).unwrap_or_default() + } +} + +/// Runs the collector until Ctrl-C: accepts worker connections, configures them, stores frames, and +/// runs the master render-scheduling task. +pub async fn run(args: CollectorArgs) -> eyre::Result<()> { + let config = WorkerConfig { + width: args.width, + height: args.height, + sync_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 connected_workers: ConnectedWorkers = Arc::new(RwLock::new(HashMap::new())); + let frame_store = Arc::new(FrameStore::new()); + + // Most of the time we want to render the screen to *something*, otherwise the game is boring + // As the breakwater sinks expect the memory layout of the [`SimpleFrameBuffer`] we need to + // create and maintain one. + let render_fb = Arc::new(SimpleFrameBuffer::new( + args.width as usize, + args.height as usize, + )); + + // 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 + #[allow(unused_variables)] + let (statistics_tx, mut statistics_rx) = mpsc::channel::(100); + #[allow(unused_variables)] + let (_statistics_information_tx, statistics_information_rx) = + broadcast::channel::(2); + let (_terminate_signal_tx, terminate_signal_rx) = broadcast::channel::<()>(1); + + // FIXME: For now we need to drain the statistics + tokio::spawn(async move { while statistics_rx.recv().await.is_some() {} }); + + let mut display_sinks = Vec:: + Send>>::new(); + + #[cfg(feature = "vnc")] + { + use breakwater::sinks::vnc::VncSink; + + if let Some(vnc_sink) = VncSink::new( + render_fb.clone(), + &args.vnc_sink, + args.fps, + "deich", // FIXME + statistics_tx.clone(), + statistics_information_rx.resubscribe(), + terminate_signal_rx.resubscribe(), + ) + .context("failed to create vnc sink")? + { + display_sinks.push(Box::new(vnc_sink)); + } + } + + if let Some(ffmpeg_sink) = FfmpegSink::new( + render_fb.clone(), + &args.ffmpeg_sink, + args.fps, + terminate_signal_rx.resubscribe(), + ) + .context("failed to create ffmpeg sink")? + { + display_sinks.push(Box::new(ffmpeg_sink)); + } + + let mut sink_threads = Vec::new(); + for mut sink in display_sinks { + sink_threads.push(tokio::spawn(async move { + sink.run().await?; + eyre::Result::<()>::Ok(()) + })); + } + + { + let frame_store = frame_store.clone(); + let render_fb = render_fb.clone(); + let connected_workers = connected_workers.clone(); + tokio::spawn(async move { + run_master(&frame_store, &connected_workers, config, render_fb).await; + }); + } + + let accept_thread = tokio::spawn(accept_workers( + listener, + connected_workers, + frame_store, + config, + )); + + #[cfg(feature = "egui")] + { + use breakwater::sinks::egui::EguiSink; + + if let Some(mut egui_sink) = EguiSink::new( + render_fb, + &args.egui_sink, + // There are no listen addresses we know about (the traffic comes in via a complicated + // path - virtual IP and stuff) + &[], + args.native_display, + statistics_information_rx.resubscribe(), + terminate_signal_rx.resubscribe(), + ) + .context("failed to create egui sink")? + { + // Some platforms require opening windows from the main thread. + // The tokio::main macro uses Runtime::block_on(future) which runs the future on + // the current thread, which should be the main thread right now. Workers keep being + // accepted by the background task above while egui owns this thread. + egui_sink.run().await.context("failed to run egui sink")?; + return Ok(()); + } + } + + // No main-thread sink is running, so block on the accept task to keep the collector alive. + accept_thread + .await + .context("the worker-accept task panicked")? +} + +/// Accepts worker connections forever, spawning a handler per connection. Only returns if accepting +/// itself fails. +async fn accept_workers( + listener: TcpListener, + connected_workers: ConnectedWorkers, + frame_store: Arc, + config: WorkerConfig, +) -> eyre::Result<()> { + loop { + let (stream, peer) = listener + .accept() + .await + .context("failed to accept worker connection")?; + + let connected_workers = connected_workers.clone(); + let frame_store = frame_store.clone(); + tokio::spawn(async move { + match handle_worker(stream, peer, config, &connected_workers, &frame_store).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(()) => {} + } + }); + } +} + +/// The master render-scheduling task. Ticks on the shared schedule; each tick it takes the workers' +/// frames for the frame it is rendering and merges them (latest write per pixel) into the long-term +/// canvas. +/// +/// The canvas is held here for the whole process lifetime, so it survives traffic gaps and worker +/// restarts: with no new frames it simply 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 run_master( + frame_store: &FrameStore, + connected_workers: &ConnectedWorkers, + config: WorkerConfig, + render_fb: Arc, +) { + let schedule = FrameSchedule::new(config.sync_fps); + // How many slots behind "now" the master renders. A worker only sends frame N once slot N has + // *ended* — one period after it began — so a frame is already a full slot old before streaming + // even starts. The `1 +` covers that slot-completion delay; `frames_spanning(..)` adds the + // streaming budget on top. (Without the `1 +`, `--fps 1` rendered a slot whose frame was still + // in flight, so nothing ever merged.) + let margin = 1 + schedule.frames_spanning(FRAME_STREAM_BUDGET); + + let mut canvas = Canvas::new(config.width as usize, config.height as usize); + + let mut current_frame = schedule.frame_number_at(get_current_ns_since_unix_epoch()); + loop { + // Tick at the start of `current_frame`'s slot. + let now = get_current_ns_since_unix_epoch(); + let slot_start = schedule.frame_start_ns(current_frame); + tokio::time::sleep(Duration::from_nanos(slot_start.saturating_sub(now))).await; + + let render_frame = current_frame.saturating_sub(margin); + let frames = frame_store.advance(render_frame..=current_frame, render_frame); + let connected = connected_workers + .read() + .expect("connected workers lock poisoned") + .len(); + + // Fold each worker's frame for this slot into the canvas (exact latest-write-per-pixel). + for frame in frames.values() { + canvas.merge(frame); + } + + canvas.draw_to_framebuffer(&render_fb); + + info!( + render_frame, + frames_merged = frames.len(), + connected_workers = connected, + "Master tick: merged frames into the canvas" + ); + + // Advance to the current slot, skipping any we fell behind on, always moving forward. + current_frame = schedule + .frame_number_at(get_current_ns_since_unix_epoch()) + .max(current_frame + 1); + } +} + +/// Reads the worker's hello, registers it, then stores its frames until the connection drops, +/// finally deregistering it. +async fn handle_worker( + mut stream: TcpStream, + peer: SocketAddr, + config: WorkerConfig, + connected_workers: &ConnectedWorkers, + frame_store: &FrameStore, +) -> io::Result<()> { + let worker_id = sync::accept_worker(&mut stream).await?; + + if !register(connected_workers, worker_id, peer) { + warn!( + %worker_id, + %peer, + "Worker id is already connected; refusing this duplicate connection" + ); + // Returning drops `stream`, which closes (slams) the connection. + return Ok(()); + } + + let result = serve_frames(&mut stream, peer, worker_id, config, frame_store).await; + + deregister(connected_workers, worker_id, peer); + result +} + +/// Sends the worker its config, then reads its frames forever, storing the ones the master wants. +async fn serve_frames( + stream: &mut TcpStream, + peer: SocketAddr, + worker_id: Uuid, + config: WorkerConfig, + frame_store: &FrameStore, +) -> io::Result<()> { + sync::send_config(stream, config).await?; + + let schedule = FrameSchedule::new(config.sync_fps); + let pixel_count = config.width as usize * config.height as usize; + // Reused only for discarding frames the master doesn't want. + let mut discard = vec![0u8; config.frame_size_bytes()]; + + loop { + let frame_number = sync::receive_frame_number(stream).await?; + + match frame_store.classify(frame_number) { + FrameInterest::Wanted => { + // Read straight into a fresh pixel buffer (outside any lock) so we can hand + // ownership to the store; `read_exact` writes directly into its bytes, no extra copy. + let mut buffer = vec![TimeTrackingPixel::default(); pixel_count]; + sync::receive_frame_body(stream, pixels_as_bytes_mut(&mut buffer)).await?; + frame_store.store(frame_number, worker_id, buffer); + } + other => { + // Still consume the blob so the stream stays aligned for the next message. + sync::receive_frame_body(stream, &mut discard).await?; + log_dropped_frame(peer, frame_number, &other, schedule); + } + } + } +} + +/// Logs a frame the master didn't want, explaining whether it was outdated or from the future and +/// by how much, alongside the tolerance (the size of the window the master accepts). +fn log_dropped_frame( + peer: SocketAddr, + frame_number: u64, + interest: &FrameInterest, + schedule: FrameSchedule, +) { + match interest { + // Can't reach here for `Wanted`, but keep the match exhaustive and cheap. + FrameInterest::Wanted => {} + FrameInterest::NoWindowYet => { + debug!(%peer, frame_number, "Dropping frame: the master hasn't started rendering yet"); + } + FrameInterest::OutsideWindow { window } => { + // The window is `[oldest ..= newest]`; `newest` is the master's current slot and the + // span is how far back it still accepts frames. + let tolerance = schedule.duration_of_frames(window.end() - window.start()); + if frame_number < *window.start() { + // Older than the oldest slot we still accept: it arrived too late. + let delayed_by = schedule.duration_of_frames(window.end() - frame_number); + warn!( + %peer, + frame_number, + ?delayed_by, + ?tolerance, + "Dropping outdated frame: it arrived later than the collector tolerates" + ); + } else { + // Newer than the master's current slot: the worker's clock is ahead of ours. + let ahead_by = schedule.duration_of_frames(frame_number - window.end()); + warn!( + %peer, + frame_number, + ?ahead_by, + ?tolerance, + "Dropping frame from the future (is the worker's clock ahead of the collector's?)" + ); + } + } + } +} + +/// Registers a worker. Returns `false` (without inserting) if its UUID is already connected. +fn register(connected_workers: &ConnectedWorkers, worker_id: Uuid, peer: SocketAddr) -> bool { + let mut workers = connected_workers + .write() + .expect("connected workers lock poisoned"); + if workers.contains_key(&worker_id) { + return false; + } + + workers.insert(worker_id, peer); + info!( + %worker_id, + %peer, + connected_workers = workers.len(), + "Worker connected" + ); + true +} + +fn deregister(connected_workers: &ConnectedWorkers, worker_id: Uuid, peer: SocketAddr) { + let mut workers = connected_workers + .write() + .expect("connected 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" + ); +} + +#[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 classify_distinguishes_wanted_old_and_new() { + let store = FrameStore::new(); + + // Before the master ticks, there's no window yet. + assert!(matches!(store.classify(42), FrameInterest::NoWindowYet)); + + *store.interesting.write().unwrap() = Some(40..=42); + + assert!(matches!(store.classify(41), FrameInterest::Wanted)); + // Older than the window -> outdated; newer than the window -> from the future. + assert!(matches!( + store.classify(38), + FrameInterest::OutsideWindow { .. } + )); + assert!(matches!( + store.classify(50), + FrameInterest::OutsideWindow { .. } + )); + } + + #[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/deich/src/lib.rs b/deich/src/lib.rs new file mode 100644 index 0000000..c0c5a0b --- /dev/null +++ b/deich/src/lib.rs @@ -0,0 +1,30 @@ +//! `deich` is breakwater's distributed mode: a fleet of **workers**, each running a Pixelflut +//! server, that continuously sync their framebuffer to a central **collector** which merges them +//! into the final picture. +//! +//! A single binary runs either role (see [`cli_args::Role`]). Right now only the [`worker`] is +//! functional. The [`sync`] protocol is still just a raw framebuffer blob, and the [`collector`] +//! is not implemented yet — both are next on the list. + +use color_eyre::eyre; + +pub mod cli_args; +pub mod collector; +pub mod sync; +pub mod worker; + +/// Shared process setup for the `deich` 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(()) +} diff --git a/deich/src/main.rs b/deich/src/main.rs new file mode 100644 index 0000000..2d1b16b --- /dev/null +++ b/deich/src/main.rs @@ -0,0 +1,14 @@ +use clap::Parser; +use color_eyre::eyre; +use deich::cli_args::{Cli, Role}; + +#[tokio::main] +async fn main() -> eyre::Result<()> { + let cli = Cli::parse(); + deich::init_telemetry()?; + + match cli.role { + Role::Worker(args) => deich::worker::run(args).await, + Role::Collector(args) => deich::collector::run(args).await, + } +} diff --git a/deich/src/sync.rs b/deich/src/sync.rs new file mode 100644 index 0000000..2bda4dc --- /dev/null +++ b/deich/src/sync.rs @@ -0,0 +1,258 @@ +//! Worker ↔ collector sync protocol. +//! +//! Flow: a worker connects to the collector and announces itself with a handshake (magic) plus a +//! [`WorkerMessage::Hello`] carrying its persistent UUID, so the collector knows which worker the +//! connection belongs to. The collector replies with a [`CollectorMessage::Config`] — it owns +//! canvas geometry and frame rate — and the worker then streams frames 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. The 12 MB framebuffer +//! blob is deliberately *not* serialized: a [`WorkerMessage::Frame`] header is immediately followed +//! on the wire by `frame_size` raw bytes, keeping serialization off the hot payload. + +use std::{io, time::Duration}; + +use breakwater_parser::{ + FrameBuffer, TimeTrackingFrameBuffer, TimeTrackingPixel, get_current_ns_since_unix_epoch, +}; +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 }, + + /// A framebuffer frame. On the wire this is immediately followed by + /// [`WorkerConfig::frame_size_bytes`] raw framebuffer bytes (not part of this serialized + /// message). `frame_number` is the scheduled slot this frame is for (see [`FrameSchedule`]); + /// every node derives it from the same wall-clock schedule, so the collector knows which slot a + /// frame belongs to. The per-pixel write timestamps live in the blob itself (absolute, set by + /// the framebuffer), so no base needs to travel with the frame. + Frame { frame_number: u64 }, +} + +/// The wall-clock frame schedule shared by every worker and the collector. Frames are numbered by +/// how many `1/fps`-long slots have elapsed since the UNIX epoch, so each node maps an instant to a +/// slot independently and still agrees on the number — no node is the master clock. (Server clocks +/// here are within ~µs of each other, far below a tens-of-milliseconds slot.) +#[derive(Debug, Clone, Copy)] +pub struct FrameSchedule { + frame_period_ns: u64, +} + +impl FrameSchedule { + pub fn new(fps: u32) -> Self { + Self { + frame_period_ns: (1_000_000_000 / u64::from(fps)).max(1), + } + } + + /// The frame slot the given UNIX-epoch instant falls into. + pub fn frame_number_at(&self, ns_since_unix_epoch: u64) -> u64 { + ns_since_unix_epoch / self.frame_period_ns + } + + /// The UNIX-epoch nanosecond timestamp at which `frame_number` begins. + pub fn frame_start_ns(&self, frame_number: u64) -> u64 { + frame_number * self.frame_period_ns + } + + /// How many whole frames `duration` spans, rounded up. Used to size delay margins. + pub fn frames_spanning(&self, duration: Duration) -> u64 { + let ns = u64::try_from(duration.as_nanos()).unwrap_or(u64::MAX); + ns.div_ceil(self.frame_period_ns) + } + + /// Wall-clock duration of `frames` frame slots. + pub fn duration_of_frames(&self, frames: u64) -> Duration { + Duration::from_nanos(frames.saturating_mul(self.frame_period_ns)) + } +} + +/// Configuration the collector hands to each worker. The collector is the single source of truth +/// for canvas geometry and frame rate; 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 sync_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 bytes in one framebuffer blob. + pub fn frame_size_bytes(&self) -> usize { + self.width as usize * self.height as usize * size_of::() + } +} + +/// 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: &str, + worker_id: Uuid, +) -> io::Result<(TcpStream, WorkerConfig)> { + let mut stream = TcpStream::connect(collector_address).await?; + write_handshake(&mut stream).await?; + write_message(&mut stream, &WorkerMessage::Hello { worker_id }).await?; + + let CollectorMessage::Config(config) = read_message(&mut stream).await?; + Ok((stream, config)) +} + +/// Collector side: validate the handshake and read the worker's [`WorkerMessage::Hello`], returning +/// the worker's UUID so the connection can be attributed to it. +pub async fn accept_worker(reader: &mut R) -> io::Result { + read_handshake(reader).await?; + match read_message(reader).await? { + WorkerMessage::Hello { worker_id } => Ok(worker_id), + WorkerMessage::Frame { .. } => Err(io::Error::new( + io::ErrorKind::InvalidData, + "expected a hello message but got a frame", + )), + } +} + +/// Collector side: read a frame's `frame_number`. The caller must then read exactly +/// [`WorkerConfig::frame_size_bytes`] bytes with [`receive_frame_body`] before the next message, or +/// discard them — splitting the two lets the caller decide whether to keep the blob. +pub async fn receive_frame_number(reader: &mut R) -> io::Result { + match read_message(reader).await? { + WorkerMessage::Frame { frame_number } => Ok(frame_number), + WorkerMessage::Hello { worker_id } => Err(io::Error::new( + io::ErrorKind::InvalidData, + format!("expected a frame but got another hello from {worker_id}"), + )), + } +} + +/// Collector side: send the worker its config in reply to the hello. +pub async fn send_config( + writer: &mut W, + config: WorkerConfig, +) -> io::Result<()> { + write_message(writer, &CollectorMessage::Config(config)).await +} + +/// Collector side: read a frame's raw blob into `pixels` (which must be +/// [`WorkerConfig::frame_size_bytes`] long), following a [`receive_frame_header`]. +pub async fn receive_frame_body( + reader: &mut R, + pixels: &mut [u8], +) -> io::Result<()> { + reader.read_exact(pixels).await?; + Ok(()) +} + +/// Worker side: stream the framebuffer to the collector forever, reconnecting as needed. +/// +/// `stream`/`config` are the live connection and config from the initial [`connect`]. On every +/// (re)connect the collector re-sends the config; we honour the frame rate, but since the +/// framebuffer is already allocated we only warn if the geometry changed (live resize is not +/// supported yet). +/// Streams frames aligned to the shared [`FrameSchedule`] until the connection fails, returning the +/// error. The worker treats that as a signal to tear down and start a fresh session (reconnect, get +/// new config, rebuild the framebuffer), so there is no reconnection logic here. +/// +/// Each iteration sleeps until the current slot ends and sends the framebuffer, tagged with the +/// just-completed slot number. Per-pixel timestamps are absolute (set by the framebuffer at write +/// time), so there's nothing to re-base — the worker only has to pick the right slot number, which +/// it recomputes from the clock so a worker that falls behind simply skips missed slots. +pub async fn sync_framebuffer( + fb: &TimeTrackingFrameBuffer, + stream: &mut TcpStream, + config: WorkerConfig, +) -> io::Result<()> { + let schedule = FrameSchedule::new(config.sync_fps); + let mut frame_number = schedule.frame_number_at(get_current_ns_since_unix_epoch()); + + loop { + // Sleep until the current slot ends. + let slot_end = schedule.frame_start_ns(frame_number + 1); + let now = get_current_ns_since_unix_epoch(); + tokio::time::sleep(Duration::from_nanos(slot_end.saturating_sub(now))).await; + + write_message(stream, &WorkerMessage::Frame { frame_number }).await?; + stream.write_all(fb.as_bytes()).await?; + stream.flush().await?; + + // Advance to the current slot, skipping any we fell behind on, always moving forward. + frame_number = schedule + .frame_number_at(get_current_ns_since_unix_epoch()) + .max(frame_number + 1); + } +} + +async fn write_handshake(writer: &mut W) -> io::Result<()> { + writer.write_u32_le(MAGIC).await?; + writer.flush().await +} + +async fn read_handshake(reader: &mut R) -> io::Result<()> { + let magic = reader.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(()) +} + +/// Writes a length-delimited, postcard-encoded control message. +async fn write_message( + writer: &mut W, + 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"))?; + + writer.write_u32_le(len).await?; + writer.write_all(&bytes).await?; + writer.flush().await +} + +/// Reads a length-delimited, postcard-encoded control message. +async fn read_message(reader: &mut R) -> io::Result { + let len = reader.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]; + reader.read_exact(&mut bytes).await?; + postcard::from_bytes(&bytes).map_err(|error| io::Error::new(io::ErrorKind::InvalidData, error)) +} diff --git a/deich/src/worker.rs b/deich/src/worker.rs new file mode 100644 index 0000000..3d40962 --- /dev/null +++ b/deich/src/worker.rs @@ -0,0 +1,146 @@ +//! The worker role: runs a Pixelflut server into a time-tracking framebuffer and syncs 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, path::Path, sync::Arc, time::Duration}; + +use breakwater::{server::Server, statistics::StatisticsEvent}; +use breakwater_parser::TimeTrackingFrameBuffer; +use color_eyre::eyre::{self, Context}; +use tokio::{net::TcpStream, sync::mpsc}; +use tracing::{info, warn}; +use uuid::Uuid; + +use crate::{ + cli_args::WorkerArgs, + sync::{self, WorkerConfig}, +}; + +/// 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: WorkerArgs) -> eyre::Result<()> { + let worker_id = load_or_create_worker_id(&args.worker_id_file)?; + info!(%worker_id, "Starting worker"); + + tokio::select! { + // `session_loop` never returns on its own — it just keeps (re)starting sessions. + result = session_loop(&args, worker_id) => result, + result = tokio::signal::ctrl_c() => { + result.context("failed to wait for ctrl + c")?; + info!("Received Ctrl-C, shutting down"); + Ok(()) + } + } +} + +/// 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: &WorkerArgs, worker_id: Uuid) -> eyre::Result<()> { + loop { + match run_session(args, worker_id).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 sync it — until the collector connection drops (or the server stops). +/// +/// The server, stats drain and sync 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: &WorkerArgs, worker_id: Uuid) -> eyre::Result<()> { + let (mut stream, 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); + + let network_buffer_size = args + .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_buffer_size))?; + + let mut server = Server::new( + &args.listen_addresses, + fb.clone(), + statistics_tx, + network_buffer_size, + None, + ) + .await + .context("failed to start pixelflut server")?; + + tokio::select! { + result = server.start() => result.context("pixelflut server stopped")?, + () = drain_stats(statistics_rx) => {} + result = sync::sync_framebuffer(&fb, &mut stream, config) => { + result.context("framebuffer sync to the collector stopped")?; + } + } + + Ok(()) +} + +/// Connects to the collector, retrying with a backoff until it succeeds. +async fn connect_to_collector( + collector_address: &str, + worker_id: Uuid, +) -> (TcpStream, WorkerConfig) { + loop { + match sync::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())) + } + } +} + +/// Currently we don't care about stats, so let's just drain them +async fn drain_stats(mut statistics_rx: mpsc::Receiver) { + loop { + if statistics_rx.recv().await.is_none() { + return; + } + } +}