diff --git a/Cargo.lock b/Cargo.lock index 09fb07b..59a8386 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -8,6 +8,56 @@ version = "2.0.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "320119579fcad9c21884f5c4861d16174d0e06250625266f50fe6898340abefa" +[[package]] +name = "anstream" +version = "1.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "824a212faf96e9acacdbd09febd34438f8f711fb84e09a8916013cd7815ca28d" +dependencies = [ + "anstyle", + "anstyle-parse", + "anstyle-query", + "anstyle-wincon", + "colorchoice", + "is_terminal_polyfill", + "utf8parse", +] + +[[package]] +name = "anstyle" +version = "1.0.14" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "940b3a0ca603d1eade50a4846a2afffd5ef57a9feac2c0e2ec2e14f9ead76000" + +[[package]] +name = "anstyle-parse" +version = "1.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "52ce7f38b242319f7cabaa6813055467063ecdc9d355bbb4ce0c68908cd8130e" +dependencies = [ + "utf8parse", +] + +[[package]] +name = "anstyle-query" +version = "1.1.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "40c48f72fd53cd289104fc64099abca73db4166ad86ea0b4341abe65af83dadc" +dependencies = [ + "windows-sys", +] + +[[package]] +name = "anstyle-wincon" +version = "3.0.11" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "291e6a250ff86cd4a820112fb8898808a366d8f9f58ce16d1f538353ad55747d" +dependencies = [ + "anstyle", + "once_cell_polyfill", + "windows-sys", +] + [[package]] name = "autocfg" version = "1.5.0" @@ -38,6 +88,58 @@ version = "0.2.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "613afe47fcd5fac7ccf1db93babcb082c5994d996f20b8b159f2ad1658eb5724" +[[package]] +name = "clap" +version = "4.6.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "dd059f9da4f5c36b3787f65d38ccaab1cc315f07b01f89abc8359ee6a8205011" +dependencies = [ + "clap_builder", + "clap_derive", +] + +[[package]] +name = "clap_builder" +version = "4.6.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f09628afdcc538b57f3c6341e9c8e9970f18e4a481690a64974d7023bd33548b" +dependencies = [ + "anstream", + "anstyle", + "clap_lex", + "strsim", +] + +[[package]] +name = "clap_derive" +version = "4.6.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f2ce8604710f6733aa641a2b3731eaa1e8b3d9973d5e3565da11800813f997a9" +dependencies = [ + "heck", + "proc-macro2", + "quote", + "syn", +] + +[[package]] +name = "clap_lex" +version = "1.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c8d4a3bb8b1e0c1050499d1815f5ab16d04f0959b233085fb31653fbfc9d98f9" + +[[package]] +name = "color_quant" +version = "1.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3d7b894f5411737b7867f4827955924d7c254fc9f4d91a6aad6b097804b1018b" + +[[package]] +name = "colorchoice" +version = "1.0.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1d07550c9036bf2ae0c684c4297d503f838287c83c53686d05370d0e139ae570" + [[package]] name = "crc32fast" version = "1.5.0" @@ -47,6 +149,12 @@ dependencies = [ "cfg-if", ] +[[package]] +name = "equivalent" +version = "1.0.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "877a4ace8713b0bcf2a4e7eec82529c029f1d0619886d18145fea96c3ffe5c0f" + [[package]] name = "fdeflate" version = "0.3.7" @@ -70,18 +178,66 @@ dependencies = [ name = "flood-rs" version = "0.1.0" dependencies = [ + "clap", + "gif", "libc", "nix", "png", + "serde", "socket2", + "toml", ] +[[package]] +name = "gif" +version = "0.13.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4ae047235e33e2829703574b54fdec96bfbad892062d97fed2f76022287de61b" +dependencies = [ + "color_quant", + "weezl", +] + +[[package]] +name = "hashbrown" +version = "0.17.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ed5909b6e89a2db4456e54cd5f673791d7eca6732202bbf2a9cc504fe2f9b84a" + +[[package]] +name = "heck" +version = "0.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2304e00983f87ffb38b55b444b5e3b60a884b5d30c0fca7d82fe33449bbe55ea" + +[[package]] +name = "indexmap" +version = "2.14.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d466e9454f08e4a911e14806c24e16fba1b4c121d1ea474396f396069cf949d9" +dependencies = [ + "equivalent", + "hashbrown", +] + +[[package]] +name = "is_terminal_polyfill" +version = "1.70.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a6cb138bb79a146c1bd460005623e142ef0181e3d0219cb493e02f7d08a35695" + [[package]] name = "libc" version = "0.2.174" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1171693293099992e19cddea4e8b849964e9846f4acee11b3948bcc337be8776" +[[package]] +name = "memchr" +version = "2.8.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cf8baf1c55e62ffcace7a9f06f4bd9cd3f0c4beb022d3b367256b91b87513d98" + [[package]] name = "memoffset" version = "0.9.1" @@ -114,6 +270,12 @@ dependencies = [ "memoffset", ] +[[package]] +name = "once_cell_polyfill" +version = "1.70.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "384b8ab6d37215f3c5301a95a4accb5d64aa607f1fcb26a11b5303878451b4fe" + [[package]] name = "png" version = "0.17.16" @@ -127,6 +289,63 @@ dependencies = [ "miniz_oxide", ] +[[package]] +name = "proc-macro2" +version = "1.0.106" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8fd00f0bb2e90d81d1044c2b32617f68fcb9fa3bb7640c23e9c748e53fb30934" +dependencies = [ + "unicode-ident", +] + +[[package]] +name = "quote" +version = "1.0.46" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "dfbc457d0c7a0759a614551b11a6409e5951f6c7537be1f1b7682b9ae9230368" +dependencies = [ + "proc-macro2", +] + +[[package]] +name = "serde" +version = "1.0.228" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9a8e94ea7f378bd32cbbd37198a4a91436180c5bb472411e48b5ec2e2124ae9e" +dependencies = [ + "serde_core", + "serde_derive", +] + +[[package]] +name = "serde_core" +version = "1.0.228" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "41d385c7d4ca58e59fc732af25c3983b67ac852c1a25000afe1175de458b67ad" +dependencies = [ + "serde_derive", +] + +[[package]] +name = "serde_derive" +version = "1.0.228" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d540f220d3187173da220f885ab66608367b6574e925011a9353e4badda91d79" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + +[[package]] +name = "serde_spanned" +version = "0.6.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bf41e0cfaf7226dca15e8197172c295a782857fcb97fad1808a166870dee75a3" +dependencies = [ + "serde", +] + [[package]] name = "simd-adler32" version = "0.3.7" @@ -143,6 +362,82 @@ dependencies = [ "winapi", ] +[[package]] +name = "strsim" +version = "0.11.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7da8b5736845d9f2fcb837ea5d9e2628564b3b043a70948a3f0b778838c5fb4f" + +[[package]] +name = "syn" +version = "2.0.119" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "872831b642d1a07999a962a351ed35b955ea2cfc8f3862091e2a240a84f17297" +dependencies = [ + "proc-macro2", + "quote", + "unicode-ident", +] + +[[package]] +name = "toml" +version = "0.8.23" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "dc1beb996b9d83529a9e75c17a1686767d148d70663143c7854d8b4a09ced362" +dependencies = [ + "serde", + "serde_spanned", + "toml_datetime", + "toml_edit", +] + +[[package]] +name = "toml_datetime" +version = "0.6.11" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "22cddaf88f4fbc13c51aebbf5f8eceb5c7c5a9da2ac40a13519eb5b0a0e8f11c" +dependencies = [ + "serde", +] + +[[package]] +name = "toml_edit" +version = "0.22.27" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "41fe8c660ae4257887cf66394862d21dbca4a6ddd26f04a3560410406a2f819a" +dependencies = [ + "indexmap", + "serde", + "serde_spanned", + "toml_datetime", + "toml_write", + "winnow", +] + +[[package]] +name = "toml_write" +version = "0.1.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5d99f8c9a7727884afe522e9bd5edbfc91a3312b36a77b5fb8926e4c31a41801" + +[[package]] +name = "unicode-ident" +version = "1.0.24" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e6e4313cd5fcd3dad5cafa179702e2b244f760991f45397d14d4ebf38247da75" + +[[package]] +name = "utf8parse" +version = "0.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "06abde3611657adf66d383f00b093d7faecc7fa57071cce2578660c9f1010821" + +[[package]] +name = "weezl" +version = "0.1.12" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a28ac98ddc8b9274cb41bb4d9d4d5c425b6020c50c46f25559911905610b4a88" + [[package]] name = "winapi" version = "0.3.9" @@ -164,3 +459,27 @@ name = "winapi-x86_64-pc-windows-gnu" version = "0.4.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "712e227841d057c1ee1cd2fb22fa7e5a5461ae8e48fa2ca79ec42cfc1931183f" + +[[package]] +name = "windows-link" +version = "0.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f0805222e57f7521d6a62e36fa9163bc891acd422f971defe97d64e70d0a4fe5" + +[[package]] +name = "windows-sys" +version = "0.61.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ae137229bcbd6cdf0f7b80a31df61766145077ddf49416a728b02cb3921ff3fc" +dependencies = [ + "windows-link", +] + +[[package]] +name = "winnow" +version = "0.7.15" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "df79d97927682d2fd8adb29682d1140b343be4ac0f08fd68b7765d9c059d3945" +dependencies = [ + "memchr", +] diff --git a/Cargo.toml b/Cargo.toml index 5afe95d..5d5a765 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -3,8 +3,15 @@ name = "flood-rs" version = "0.1.0" edition = "2024" +[profile.release] +debug = true + [dependencies] png = "0.17" libc = "0.2" nix = { version = "0.29", features = ["socket", "uio"] } socket2 = "0.4.7" +clap = { version = "4", features = ["derive"] } +gif = "0.13" +serde = { version = "1", features = ["derive"] } +toml = "0.8" diff --git a/config.toml b/config.toml new file mode 100644 index 0000000..fcbebd5 --- /dev/null +++ b/config.toml @@ -0,0 +1,14 @@ +# Each [[sender]] block defines a destination with its own thread group. +# All senders share the same frame data. + +[[sender]] +address = "100.65.0.2" +port = 5005 +interface = "eth0" +threads = 5 + +# Example: send to a second display from a different interface +# [[sender]] +# address = "192.168.1.50" +# port = 5005 +# threads = 3 diff --git a/flake.lock b/flake.lock new file mode 100644 index 0000000..7dac337 --- /dev/null +++ b/flake.lock @@ -0,0 +1,27 @@ +{ + "nodes": { + "nixpkgs": { + "locked": { + "lastModified": 1759733170, + "narHash": "sha256-TXnlsVb5Z8HXZ6mZoeOAIwxmvGHp1g4Dw89eLvIwKVI=", + "owner": "nixos", + "repo": "nixpkgs", + "rev": "8913c168d1c56dc49a7718685968f38752171c3b", + "type": "github" + }, + "original": { + "owner": "nixos", + "ref": "nixos-unstable", + "repo": "nixpkgs", + "type": "github" + } + }, + "root": { + "inputs": { + "nixpkgs": "nixpkgs" + } + } + }, + "root": "root", + "version": 7 +} diff --git a/flake.nix b/flake.nix new file mode 100644 index 0000000..004750b --- /dev/null +++ b/flake.nix @@ -0,0 +1,29 @@ +{ + description = "A very basic flake"; + + inputs = { + nixpkgs.url = "github:nixos/nixpkgs?ref=nixos-unstable"; + }; + + outputs = { self, nixpkgs }: { + devShells.x86_64-linux.default = + let + insecure-package-overlay = final: prev: { + nixpkgs.config.permittedInsecurePackages = [ + # Add the full name and version of the package here + "openssl-1.1.1w" + # "another-insecure-package-2.0" + ]; + }; + pkgs = import nixpkgs { + system = "x86_64-linux"; + overlays = []; + }; + in + pkgs.mkShell { + buildInputs = [ + pkgs.cargo + ]; + }; + }; +} diff --git a/images/ftp.png b/images/ftp.png new file mode 100644 index 0000000..052a042 Binary files /dev/null and b/images/ftp.png differ diff --git a/images/ftp2.png b/images/ftp2.png new file mode 100644 index 0000000..443406a Binary files /dev/null and b/images/ftp2.png differ diff --git a/images/netto.gif b/images/netto.gif new file mode 100644 index 0000000..bf76d3a Binary files /dev/null and b/images/netto.gif differ diff --git a/src/bouncing_image.rs b/src/bouncing_image.rs new file mode 100644 index 0000000..bd4544c --- /dev/null +++ b/src/bouncing_image.rs @@ -0,0 +1,85 @@ +use crate::drawable::Drawable; +use crate::pixel_buf::PixelBuf; +use crate::png_data::PngData; +use crate::{DISPLAY_HEIGHT, DISPLAY_WIDTH}; +use crate::pixel::Pixel; +/// Represents a bouncing image on the screen. +pub struct BouncingImage { + img: PngData, + x: f32, + y: f32, + x1: f32, + y1: f32, + x2: f32, + y2: f32, + /// Horizontal velocity in pixels per second. + move_x: f32, + /// Vertical velocity in pixels per second. + move_y: f32, +} + +impl BouncingImage { + /// Initializes a new BouncingImage. + /// + /// - `move_x`, `move_y`: velocity in pixels per second. + /// - `start_x`, `start_y`: initial position (-1 to center on that axis). + pub fn new(img_file: &str, move_x: f32, move_y: f32, start_x: i32, start_y: i32) -> Self { + let img = PngData::open(img_file).expect("Could not load image"); + let x2 = DISPLAY_WIDTH as f32 - img.width as f32; + let y2 = DISPLAY_HEIGHT as f32 - img.height as f32; + let x1: f32 = 0.0; + let y1: f32 = 0.0; + + let x = if start_x == -1 { + (x1 + x2) / 2.0 + } else { + start_x as f32 + }; + let y = if start_y == -1 { + (y1 + y2) / 2.0 + } else { + start_y as f32 + }; + + BouncingImage { + img, + x, + y, + x1, + y1, + x2, + y2, + move_x, + move_y, + } + } +} +fn offset_pixels(pixels: &Vec, dx: u16, dy: u16) -> Vec { + let mut out = pixels.clone(); // memcpy, since Pixel is Copy + for p in out.iter_mut() { + p.x = p.x.saturating_add(dx); + p.y = p.y.saturating_add(dy); + } + out +} +impl Drawable for BouncingImage { + fn draw(&self, buf: &mut PixelBuf, _elapsed: f32) { + + let pixels = offset_pixels(&self.img.packed_pixels,self.x as u16,self.y as u16); + buf.set_pixels(pixels); + } + + fn update(&mut self, dt: f32, _elapsed: f32) { + self.x += self.move_x * dt; + self.y += self.move_y * dt; + + if self.x < self.x1 || self.x > self.x2 { + self.move_x *= -1.0; + self.x = self.x.clamp(self.x1, self.x2); + } + if self.y < self.y1 || self.y > self.y2 { + self.move_y *= -1.0; + self.y = self.y.clamp(self.y1, self.y2); + } + } +} diff --git a/src/circle.rs b/src/circle.rs new file mode 100644 index 0000000..c975228 --- /dev/null +++ b/src/circle.rs @@ -0,0 +1,109 @@ +use std::sync::{Arc, Mutex}; + +use crate::color; +use crate::pixel::Pixel; +use crate::drawable::Drawable; +use crate::pixel_buf::PixelBuf; + +/// Speed at which the radius shrinks, in pixels per second. +const RADIUS_SHRINK_RATE: f32 = 30.0; + +/// Speed at which the hue cycles, in degrees per second. +const HUE_CYCLE_RATE: f32 = 180.0; + +/// Represents a circle drawn at coordinates received via shared state. +pub struct Circle { + x: Arc>, + y: Arc>, + set: Arc>, + radius: f32, + /// Cached draw coordinates (read from mutex in update, used in draw). + draw_x: u32, + draw_y: u32, + color: color::Rgb +} + +impl Circle { + pub fn new(x: Arc>, y: Arc>, set: Arc>) -> Self { + Circle { + x, + y, + set, + radius: 0.0, + draw_x: 0, + draw_y: 0, + color: color::Rgb { r:0,g:0,b:0} + } + } + + /// Draws a circle using the Midpoint Circle Algorithm, writing 8 octant + /// points per step. + fn draw_circle( + buf: &mut PixelBuf, + cx: i32, + cy: i32, + radius: u32, + r: u8, + g: u8, + b: u8, + ) { + let rad = radius as i32; + let mut x: i32 = 0; + let mut y: i32 = rad; + let mut d: i32 = 3 - 2 * rad; + + while y >= x { + // 8 octant points. + buf.set_pixel(Pixel {x:(cx + x) as u16, y:(cy + y) as u16, r, g, b}); + buf.set_pixel(Pixel {x:(cx - x) as u16, y:(cy + y) as u16, r, g, b}); + buf.set_pixel(Pixel {x:(cx + x) as u16, y:(cy - y) as u16, r, g, b}); + buf.set_pixel(Pixel {x:(cx - x) as u16, y:(cy - y) as u16, r, g, b}); + buf.set_pixel(Pixel {x:(cx + y) as u16, y:(cy + x) as u16, r, g, b}); + buf.set_pixel(Pixel {x:(cx - y) as u16, y:(cy + x) as u16, r, g, b}); + buf.set_pixel(Pixel {x:(cx + y) as u16, y:(cy - x) as u16, r, g, b}); + buf.set_pixel(Pixel {x:(cx - y) as u16, y:(cy - x) as u16, r, g, b}); + + x += 1; + if d > 0 { + y -= 1; + d = d + 4 * (x - y) + 10; + } else { + d = d + 4 * x + 6; + } + } + } +} + +impl Drawable for Circle { + fn draw(&self, buf: &mut PixelBuf, _elapsed: f32) { + + + let cx = self.draw_y as i32; + let cy = self.draw_x as i32; + let radius = self.radius as u32; + + for i in 0..10 { + if radius > i { + Self::draw_circle(buf, cx, cy, radius - i, self.color.r, self.color.g, self.color.b); + } + } + } + + fn update(&mut self, dt: f32, elapsed: f32) { + self.radius = 150.0; + self.draw_y = (*self.x.lock().unwrap() + (elapsed as u32))%1080; + self.draw_x = (*self.y.lock().unwrap() + (elapsed as u32))%1920; + + self.radius -= RADIUS_SHRINK_RATE * dt; + if self.radius < 0.0 { + self.radius = 0.0; + } + let hue = ((elapsed * HUE_CYCLE_RATE) % 360.0) as u16; + let hsv_color = color::Hsv { + h: hue, + s: 1.0, + v: 1.0, + }; + self.color = hsv_color.into(); + } +} diff --git a/src/config.rs b/src/config.rs new file mode 100644 index 0000000..a8a14fa --- /dev/null +++ b/src/config.rs @@ -0,0 +1,54 @@ +use serde::Deserialize; +use std::fs; +use std::path::Path; + +#[derive(Deserialize)] +pub struct Config { + pub sender: Vec, +} + +#[derive(Deserialize, Clone)] +pub struct SenderConfig { + /// Destination IP address or hostname. + pub address: String, + + /// Destination port (default: 5005). + #[serde(default = "default_port")] + pub port: u16, + + /// Source network interface (e.g. "eth0"). Auto-discovered if omitted. + pub interface: Option, + + /// Number of packer/sender thread pairs (default: 5). + #[serde(default = "default_threads")] + pub threads: usize, +} + +fn default_port() -> u16 { + 5005 +} + +fn default_threads() -> usize { + 5 +} + +impl Config { + pub fn load(path: &Path) -> Result { + let content = fs::read_to_string(path) + .map_err(|e| format!("Failed to read config file '{}': {}", path.display(), e))?; + let config: Config = toml::from_str(&content) + .map_err(|e| format!("Failed to parse config file '{}': {}", path.display(), e))?; + + if config.sender.is_empty() { + return Err("Config must define at least one [[sender]]".into()); + } + + for (i, s) in config.sender.iter().enumerate() { + if s.threads == 0 { + return Err(format!("sender[{}]: threads must be >= 1", i)); + } + } + + Ok(config) + } +} diff --git a/src/display.rs b/src/display.rs new file mode 100644 index 0000000..78901d3 --- /dev/null +++ b/src/display.rs @@ -0,0 +1,408 @@ +use std::collections::VecDeque; +use std::net::{ToSocketAddrs, UdpSocket}; +use std::sync::{Arc, Condvar, Mutex}; +use std::thread; +use std::time::Instant; + +use crate::config::SenderConfig; +use crate::raw_socket::RawSender; + +/// Per-thread packet queue capacity. +const QUEUE_LEN: usize = 200; + +/// Max pixels per packet. Constrained by MTU 1500: +/// 1500 (IP payload max) - 20 (IP hdr) - 8 (UDP hdr) - 2 (pixel hdr) = 1470 +/// 1470 / 7 bytes per pixel = 210 pixels. +const PIXELS_PER_PACKET: usize = 210; +const MSGSIZE: usize = 2 + 7 * PIXELS_PER_PACKET; + +/// Maximum number of packets to batch into a single sendmmsg call. +const SEND_BATCH_SIZE: usize = 26; + +// --------------------------------------------------------------------------- +// Packet + per-thread queue +// --------------------------------------------------------------------------- + +/// A filled buffer ready to be sent, carrying its valid data length. +struct Packet { + buf: Box<[u8; MSGSIZE]>, + len: usize, +} + +/// Per-thread queue shared between a packer and its sender. +struct SenderQueue { + pending: Mutex>, + condvar: Condvar, + pool: Mutex>>, +} + +/// Allocates a fresh buffer with the protocol header bytes set. +fn alloc_buf() -> Box<[u8; MSGSIZE]> { + let mut buf = Box::new([0u8; MSGSIZE]); + buf[0] = 0x00; + buf[1] = 0x01; + buf +} + +/// Grabs an empty buffer: try pool first, steal oldest pending, or allocate. +fn grab_buf(queue: &SenderQueue) -> Box<[u8; MSGSIZE]> { + if let Some(buf) = queue.pool.lock().unwrap().pop() { + return buf; + } + if let Some(oldest) = queue.pending.lock().unwrap().pop_front() { + return oldest.buf; + } + alloc_buf() +} + +/// Submits a filled packet to the sender queue and wakes the sender. +fn submit_packet(queue: &SenderQueue, buf: Box<[u8; MSGSIZE]>, pixel_count: usize) { + let len = 2 + pixel_count * 7; + queue.pending.lock().unwrap().push_back(Packet { buf, len }); + queue.condvar.notify_one(); +} + +// --------------------------------------------------------------------------- +// Simple PRNG for deterministic permutation generation +// --------------------------------------------------------------------------- + +struct Xorshift64(u64); + +impl Xorshift64 { + fn next(&mut self) -> u64 { + let mut x = self.0; + x ^= x << 13; + x ^= x >> 7; + x ^= x << 17; + self.0 = x; + x + } +} + +fn fisher_yates_shuffle(arr: &mut [usize], rng: &mut Xorshift64) { + for i in (1..arr.len()).rev() { + let j = (rng.next() as usize) % (i + 1); + arr.swap(i, j); + } +} + +// --------------------------------------------------------------------------- +// Display +// --------------------------------------------------------------------------- + +/// Collects pixels into a frame buffer and publishes completed frames for +/// packer threads to consume. +/// +/// Architecture: +/// main thread -> write_raw_pixels() -> flush_frame() +/// | +/// publishes Arc> frame snapshot +/// | +/// N packer threads (each with unique noise permutation) +/// continuously iterate the frame, packing pixels into packets +/// | +/// N sender threads (one per packer, own packet queue) +/// send via RawSender (AF_PACKET) or fallback UDP +/// +/// Multiple sender configs are supported -- each config spawns its own group +/// of packer/sender thread pairs targeting a different destination. +pub struct Display { + /// Frame being assembled by the main thread. + building_frame: Vec<[u8; 7]>, + /// Shared pointer to the latest completed frame. + /// Packer threads clone the Arc to read it without holding the lock. + current_frame: Arc>>>, +} + +impl Display { + /// Creates a new Display and spawns packer + sender thread pairs for + /// each sender in the config. + pub fn new(senders: &[SenderConfig]) -> Self { + let current_frame = Arc::new(Mutex::new(Arc::new(Vec::new()))); + + let mut global_thread_idx: usize = 0; + + for (sender_idx, sender) in senders.iter().enumerate() { + let host = &sender.address; + let port = sender.port; + let interface = sender.interface.as_deref(); + let num_threads = sender.threads; + + eprintln!( + "[display] Sender {}: {}:{}, interface={}, threads={}", + sender_idx, + host, + port, + interface.unwrap_or("auto"), + num_threads, + ); + + for t in 0..num_threads { + // Each packer-sender pair gets its own queue. + let queue = Arc::new(SenderQueue { + pending: Mutex::new(VecDeque::with_capacity(QUEUE_LEN)), + condvar: Condvar::new(), + pool: Mutex::new({ + let mut v = Vec::with_capacity(QUEUE_LEN); + for _ in 0..QUEUE_LEN { + v.push(alloc_buf()); + } + v + }), + }); + + // Spawn packer thread. + let frame_source = Arc::clone(¤t_frame); + let packer_queue = Arc::clone(&queue); + // Each thread gets a globally unique seed. + let seed = (global_thread_idx as u64 + 1).wrapping_mul(0x517cc1b727220a95); + + let packer_name = format!("packer-{}-{}", sender_idx, t); + thread::Builder::new() + .name(packer_name) + .spawn(move || { + packer_loop(frame_source, packer_queue, seed); + }) + .unwrap(); + + // Spawn sender thread. + let sender_queue = Arc::clone(&queue); + let sender_name = format!("sender-{}-{}", sender_idx, t); + match RawSender::new(host, port, 0, interface) { + Ok(raw_sender) => { + eprintln!("[display] {}: raw AF_PACKET sender", sender_name); + thread::Builder::new() + .name(sender_name) + .spawn(move || { + sender_loop_raw(raw_sender, sender_queue); + }) + .unwrap(); + } + Err(e) => { + eprintln!( + "[display] {}: raw unavailable ({}), falling back to UDP", + sender_name, e + ); + let remote_addr = (host.as_str(), port) + .to_socket_addrs() + .expect("Invalid remote address") + .next() + .expect("Could not resolve host"); + let socket = UdpSocket::bind("0.0.0.0:0").expect("Could not bind"); + socket.connect(remote_addr).expect("Could not connect"); + thread::Builder::new() + .name(sender_name) + .spawn(move || { + sender_loop_udp(socket, sender_queue); + }) + .unwrap(); + } + } + + global_thread_idx += 1; + } + } + + Display { + building_frame: Vec::with_capacity(60_000), + current_frame, + } + } + + /// Appends raw 7-byte pixel entries to the frame being built. + #[inline] + pub fn write_raw_pixels(&mut self, pixels: &[[u8; 7]]) { + self.building_frame.extend_from_slice(pixels); + } + + /// Publishes the current frame for packer threads and starts a new one. + pub fn flush_frame(&mut self) { + let frame = std::mem::take(&mut self.building_frame); + let capacity = frame.len(); + let new_frame = Arc::new(frame); + *self.current_frame.lock().unwrap() = new_frame; + // Pre-allocate next frame to the same size. + self.building_frame.reserve(capacity); + } +} + +// --------------------------------------------------------------------------- +// Packer thread +// --------------------------------------------------------------------------- + +/// Continuously reads the current frame and packs pixels in a noise-shuffled +/// order into packets, feeding them to the paired sender thread. +/// +/// Each packer has a unique fixed permutation so all 5 threads naturally +/// cover different regions of the image at any given moment. +fn packer_loop( + frame_source: Arc>>>, + queue: Arc, + seed: u64, +) { + let mut rng = Xorshift64(seed); + let mut permutation: Vec = Vec::new(); + let mut last_frame_len: usize = 0; + + loop { + // Snapshot the current frame (just an Arc clone -- very cheap). + let frame = frame_source.lock().unwrap().clone(); + + if frame.is_empty() { + thread::yield_now(); + continue; + } + + // Rebuild index array only when frame size changes; otherwise + // re-shuffle the existing permutation in place. + if frame.len() != last_frame_len { + permutation = (0..frame.len()).collect(); + last_frame_len = frame.len(); + } + fisher_yates_shuffle(&mut permutation, &mut rng); + + // Walk the permutation, packing pixels into packets. + let mut buf = grab_buf(&queue); + let mut pos: usize = 0; + + for &idx in &permutation { + if idx >= frame.len() { + continue; + } + + let dst_offset = 2 + pos * 7; + buf[dst_offset..dst_offset + 7].copy_from_slice(&frame[idx]); + pos += 1; + + if pos == PIXELS_PER_PACKET { + submit_packet(&queue, buf, pos); + buf = grab_buf(&queue); + pos = 0; + } + } + + // Flush any remaining partial packet. + if pos > 0 { + submit_packet(&queue, buf, pos); + } + } +} + +// --------------------------------------------------------------------------- +// Sender threads +// --------------------------------------------------------------------------- + +/// Raw AF_PACKET sender loop. +fn sender_loop_raw(raw_sender: RawSender, queue: Arc) { + let mut batch: Vec = Vec::with_capacity(SEND_BATCH_SIZE); + let mut frame_buf: Vec = Vec::with_capacity(SEND_BATCH_SIZE * (MSGSIZE + 42)); + + let mut stats_sent: u64 = 0; + let mut stats_bytes: u64 = 0; + let mut stats_errors: u64 = 0; + let mut stats_send_us: u128 = 0; + let mut stats_send_calls: u64 = 0; + let mut stats_last_report = Instant::now(); + + loop { + { + let mut pending = queue.pending.lock().unwrap(); + while pending.is_empty() { + pending = queue.condvar.wait(pending).unwrap(); + } + let n = pending.len().min(SEND_BATCH_SIZE); + batch.extend(pending.drain(..n)); + } + + let payload_ptrs: Vec<&[u8]> = batch.iter().map(|p| &p.buf[..p.len]).collect(); + + let send_start = Instant::now(); + match raw_sender.send_batch(&payload_ptrs, &mut frame_buf) { + Ok(n) => { + stats_sent += n as u64; + for packet in batch.iter().take(n) { + stats_bytes += (packet.len + raw_sender.header_size()) as u64; + } + } + Err(e) => { + if stats_errors == 0 { + eprintln!("sendmmsg (raw) failed: {}", e); + } + stats_errors += batch.len() as u64; + } + } + stats_send_us += send_start.elapsed().as_micros(); + stats_send_calls += 1; + + // Return buffers to the pool. + { + let mut pool = queue.pool.lock().unwrap(); + for packet in batch.drain(..) { + pool.push(packet.buf); + } + } + + let now = Instant::now(); + let elapsed = (now - stats_last_report).as_secs_f64(); + if elapsed >= 5.0 { + let pps = stats_sent as f64 / elapsed; + let mbps = stats_bytes as f64 * 8.0 / elapsed / 1_000_000.0; + eprintln!( + "[sender] {:.0} pkt/s, {:.1} Mbit/s sent | errors: {} | send: {:.0}us avg over {} calls", + pps, mbps, stats_errors, + stats_send_us as f64 / stats_send_calls.max(1) as f64, + stats_send_calls, + ); + stats_sent = 0; + stats_bytes = 0; + stats_errors = 0; + stats_send_us = 0; + stats_send_calls = 0; + stats_last_report = now; + } + } +} + +/// Fallback UDP sender loop. +fn sender_loop_udp(socket: UdpSocket, queue: Arc) { + let mut batch: Vec = Vec::with_capacity(SEND_BATCH_SIZE); + let mut stats_sent: u64 = 0; + let mut stats_bytes: u64 = 0; + let mut stats_last_report = Instant::now(); + + loop { + { + let mut pending = queue.pending.lock().unwrap(); + while pending.is_empty() { + pending = queue.condvar.wait(pending).unwrap(); + } + let n = pending.len().min(SEND_BATCH_SIZE); + batch.extend(pending.drain(..n)); + } + + for packet in &batch { + if let Ok(n) = socket.send(&packet.buf[..packet.len]) { + stats_sent += 1; + stats_bytes += n as u64; + } + } + + { + let mut pool = queue.pool.lock().unwrap(); + for packet in batch.drain(..) { + pool.push(packet.buf); + } + } + + let now = Instant::now(); + let elapsed = (now - stats_last_report).as_secs_f64(); + if elapsed >= 5.0 { + let pps = stats_sent as f64 / elapsed; + let mbps = stats_bytes as f64 * 8.0 / elapsed / 1_000_000.0; + eprintln!("[sender] {:.0} pkt/s, {:.1} Mbit/s (UDP fallback)", pps, mbps); + stats_sent = 0; + stats_bytes = 0; + stats_last_report = now; + } + } +} diff --git a/src/drawable.rs b/src/drawable.rs new file mode 100644 index 0000000..3850860 --- /dev/null +++ b/src/drawable.rs @@ -0,0 +1,8 @@ +use crate::pixel_buf::PixelBuf; + +pub trait Drawable: Send + Sync { + fn draw(&self, buf: &mut PixelBuf, elapsed: f32); + + + fn update(&mut self, dt: f32, elapsed: f32); +} diff --git a/src/gif_image.rs b/src/gif_image.rs new file mode 100644 index 0000000..700a87e --- /dev/null +++ b/src/gif_image.rs @@ -0,0 +1,164 @@ +use std::fs::File; + +use crate::drawable::Drawable; +use crate::pixel::Pixel; +use crate::pixel_buf::PixelBuf; + +/// A precomputed frame: the offset pixels ready to draw and its duration. +struct GifFrame { + /// Precomputed pixels at the fixed screen position. + pixels: Vec, + /// Duration of this frame in seconds. + duration_secs: f32, +} + +/// A fixed-position animated GIF drawable. +/// +/// All frames are decoded and their opaque pixels precomputed at load time. +/// During playback the elapsed time selects which frame to display. +pub struct GifImage { + frames: Vec, + /// Total loop duration (sum of all frame durations) in seconds. + total_duration: f32, +} + +impl GifImage { + /// Loads a GIF file and precomputes the pixel data for every frame. + /// + /// `x`, `y` set the fixed screen position of the top-left corner. + pub fn new(path: &str, x: i32, y: i32) -> Self { + let file = File::open(path).expect("Failed to open GIF file"); + let mut decoder = gif::DecodeOptions::new(); + decoder.set_color_output(gif::ColorOutput::RGBA); + let mut reader = decoder.read_info(file).expect("Failed to read GIF info"); + + let width = reader.width() as u32; + let height = reader.height() as u32; + + // Canvas holds the composited RGBA image, built up across frames + // to handle disposal methods correctly. + let mut canvas = vec![0u8; (width * height * 4) as usize]; + let mut frames = Vec::new(); + + while let Some(frame) = reader.read_next_frame().expect("Failed to decode GIF frame") { + let fw = frame.width as u32; + let fh = frame.height as u32; + let fx = frame.left as u32; + let fy = frame.top as u32; + + // Composite this frame onto the canvas. + for row in 0..fh { + for col in 0..fw { + let src_idx = ((row * fw + col) * 4) as usize; + let dst_x = fx + col; + let dst_y = fy + row; + if dst_x >= width || dst_y >= height { + continue; + } + let dst_idx = ((dst_y * width + dst_x) * 4) as usize; + let a = frame.buffer[src_idx + 3]; + if a > 0 { + canvas[dst_idx] = frame.buffer[src_idx]; + canvas[dst_idx + 1] = frame.buffer[src_idx + 1]; + canvas[dst_idx + 2] = frame.buffer[src_idx + 2]; + canvas[dst_idx + 3] = a; + } + } + } + + // Extract opaque pixels from the composited canvas. + let mut pixels = Vec::new(); + for sy in 0..height { + for sx in 0..width { + let idx = ((sy * width + sx) * 4) as usize; + let a = canvas[idx + 3]; + if a > 0 { + pixels.push(Pixel { + x: x as u16 + sx as u16, + y: y as u16 + sy as u16, + r: canvas[idx], + g: canvas[idx + 1], + b: canvas[idx + 2], + }); + } + } + } + + // GIF frame delay is in centiseconds; 0 is conventionally ~100ms. + let delay_cs = frame.delay; + let duration_secs = if delay_cs == 0 { + 0.1 + } else { + delay_cs as f32 / 100.0 + }; + + frames.push(GifFrame { + pixels, + duration_secs, + }); + + // Handle disposal: if RestoreToBackground, clear the frame region. + if frame.dispose == gif::DisposalMethod::Background { + for row in 0..fh { + for col in 0..fw { + let dst_x = fx + col; + let dst_y = fy + row; + if dst_x < width && dst_y < height { + let idx = ((dst_y * width + dst_x) * 4) as usize; + canvas[idx] = 0; + canvas[idx + 1] = 0; + canvas[idx + 2] = 0; + canvas[idx + 3] = 0; + } + } + } + } + // DisposalMethod::Previous is rare and complex; we ignore it + // (treat as Keep, which is usually fine). + } + + let total_duration: f32 = frames.iter().map(|f| f.duration_secs).sum(); + + eprintln!( + "[gif] {}: {}x{}, {} frames, {:.2}s total loop", + path, + width, + height, + frames.len(), + total_duration, + ); + + GifImage { + frames, + total_duration, + } + } +} + +impl Drawable for GifImage { + fn draw(&self, buf: &mut PixelBuf, elapsed: f32) { + if self.frames.is_empty() || self.total_duration <= 0.0 { + return; + } + + // Loop the animation. + let t = elapsed % self.total_duration; + + // Find which frame corresponds to time t. + let mut accumulated = 0.0; + let mut frame_idx = 0; + for (i, frame) in self.frames.iter().enumerate() { + accumulated += frame.duration_secs; + if t < accumulated { + frame_idx = i; + break; + } + } + + buf.set_pixels_ref(&self.frames[frame_idx].pixels); + } + + fn update(&mut self, _dt: f32, _elapsed: f32) { + // Fixed position, nothing to update. + } +} diff --git a/src/main-new.rs b/src/main-new.rs deleted file mode 100644 index 7af1867..0000000 --- a/src/main-new.rs +++ /dev/null @@ -1,272 +0,0 @@ - -use std::fs::File; -use std::io::BufReader; -use std::net::{ToSocketAddrs, UdpSocket}; -use std::os::unix::io::{AsRawFd, RawFd}; -use std::time::Duration; - -// Nix crate for sendmmsg -use nix::sys::socket::{sendmmsg, MsgFlags, SendMmsgData}; -use nix::sys::uio::IoVec; - -// Constants from the C code -const QUEUE_LEN: usize = 1000; -const MSG_PAYLOAD_SIZE: usize = 7 * 160; -const MSGSIZE: usize = 2 + MSG_PAYLOAD_SIZE; - -const DISPLAY_HOST: &str = "100.65.0.2"; -const DISPLAY_PORT: u16 = 5005; -const DISPLAY_WIDTH: i32 = 1920; -const DISPLAY_HEIGHT: i32 = 1080; - -/// Represents the data decoded from a PNG file. -struct PngData { - width: u32, - height: u32, - pixels: Vec, -} - -impl PngData { - /// Loads and decodes a PNG image from the given path. - fn open(path: &str) -> Result { - let file = File::open(path).expect("Failed to open PNG file"); - let decoder = png::Decoder::new(BufReader::new(file)); - let mut reader = decoder.read_info()?; - let mut buf = vec![0; reader.output_buffer_size()]; - let info = reader.next_frame(&mut buf)?; - - Ok(PngData { - width: info.width, - height: info.height, - pixels: buf, - }) - } -} - -/// Represents a bouncing image on the screen. -struct BouncingImage { - img: PngData, - x: i32, - y: i32, - x1: i32, - y1: i32, - x2: i32, - y2: i32, - move_x: i32, - move_y: i32, - rate: u32, -} - -impl BouncingImage { - /// Initializes a new BouncingImage. - fn new(img_file: &str, move_x: i32, move_y: i32, rate: u32, start_x: i32, start_y: i32) -> Self { - let img = PngData::open(img_file).expect("Could not load image"); - let mut bb = BouncingImage { - x2: DISPLAY_WIDTH - img.width as i32, - y2: DISPLAY_HEIGHT - img.height as i32, - img, - x: start_x, - y: start_y, - x1: 0, - y1: 0, - move_x, - move_y, - rate, - }; - if bb.x == -1 { - bb.x = (bb.x1 + bb.x2) / 2; - } - if bb.y == -1 { - bb.y = (bb.y1 + bb.y2) / 2; - } - bb - } - - /// Draws the image and updates its position. - fn draw_and_move(&mut self, display: &mut Display) { - display.draw_png(&self.img, self.x, self.y); - - self.x += self.move_x; - self.y += self.move_y; - - if self.x < self.x1 || self.x > self.x2 { - self.move_x *= -1; - } - if self.y < self.y1 || self.y > self.y2 { - self.move_y *= -1; - } - } -} - -/// Manages the connection and data sent to the display. -struct Display { - fd: RawFd, - bufs: Vec<[u8; MSGSIZE]>, - lens: Vec, // Stores the actual length of data in each buffer - next_buf: usize, // The next buffer in the queue to be filled - send_next: usize, // The next buffer in the queue to be sent - pos_in_buf: usize, - // Keep the socket alive to keep the file descriptor valid - _socket: UdpSocket, -} - -impl Display { - /// Creates a new Display and connects to the specified host and port. - fn new(host: &str, port: u16) -> Self { - let remote_addr = (host, port) - .to_socket_addrs() - .expect("Invalid remote address") - .next() - .expect("Could not resolve host"); - - let socket = UdpSocket::bind("0.0.0.0:0").expect("Could not bind to local port"); - socket.connect(remote_addr).expect("Could not connect to remote"); - let fd = socket.as_raw_fd(); - - let mut bufs = vec![[0; MSGSIZE]; QUEUE_LEN]; - for buf in bufs.iter_mut() { - buf[0] = 0x00; - buf[1] = 0x01; - } - - Display { - fd, - bufs, - lens: vec![0; QUEUE_LEN], - next_buf: 0, - send_next: 0, - pos_in_buf: 0, - _socket: socket, - } - } - - /// Marks the current buffer as ready to be sent and moves to the next one. - fn mark_buffer_ready(&mut self) { - if self.pos_in_buf > 0 { - self.lens[self.next_buf] = 2 + self.pos_in_buf * 7; - self.next_buf = (self.next_buf + 1) % QUEUE_LEN; - self.pos_in_buf = 0; - - // If we've wrapped around and caught up to the send queue, - // we must flush to avoid overwriting data that hasn't been sent. - if self.next_buf == self.send_next { - eprintln!("Warning: Buffer queue full. Forcing a flush."); - self.flush_all_pending(); - } - } - } - - /// Sends all queued packets using the efficient `sendmmsg` syscall. - fn flush_all_pending(&mut self) { - // First, ensure the current, partially-filled buffer is marked as ready. - self.mark_buffer_ready(); - - if self.send_next == self.next_buf { - return; // Nothing to send. - } - - // We build a temporary list of message headers to pass to sendmmsg. - // This is the cleanest way to handle the circular buffer. - let mut iovecs_storage = Vec::new(); - let mut messages_to_send = Vec::new(); - - let mut current_idx = self.send_next; - while current_idx != self.next_buf { - let data_slice = &self.bufs[current_idx][..self.lens[current_idx]]; - iovecs_storage.push(IoVec::from_slice(data_slice)); - current_idx = (current_idx + 1) % QUEUE_LEN; - } - - // Since we used `connect()`, the kernel knows the destination address, - // so we can pass `None` for the address in `SendMmsgData`. - for iov in &iovecs_storage { - messages_to_send.push(SendMmsgData { - iov: &[*iov], - addr: None, - cmsgs: &[], - _phantom: std::marker::PhantomData, - }); - } - - if messages_to_send.is_empty() { - return; - } - - // Perform the `sendmmsg` syscall - match sendmmsg(self.fd, &messages_to_send, MsgFlags::empty()) { - Ok(num_sent) => { - // Advance the send queue by the number of packets actually sent. - self.send_next = (self.send_next + num_sent) % QUEUE_LEN; - } - Err(e) => { - // Non-blocking sockets might return an error indicating to try again. - // For this example, we'll just log other errors. - if e != nix::errno::Errno::EAGAIN && e != nix::errno::Errno::EWOULDBLOCK { - eprintln!("Failed to send messages with sendmmsg: {}", e); - } - } - } - } - - /// Sets a pixel color at a specific coordinate. - fn set_pixel(&mut self, x: u16, y: u16, r: u8, g: u8, b: u8) { - let offset = 2 + self.pos_in_buf * 7; - let buf = &mut self.bufs[self.next_buf][offset..offset + 7]; - buf[0] = x as u8; - buf[1] = (x >> 8) as u8; - buf[2] = y as u8; - buf[3] = (y >> 8) as u8; - buf[4] = r; - buf[5] = g; - buf[6] = b; - - self.pos_in_buf += 1; - if self.pos_in_buf == 160 { - self.mark_buffer_ready(); - } - } - - /// Draws a PNG image at the given coordinates. - fn draw_png(&mut self, png: &PngData, x: i32, y: i32) { - for sy in 0..png.height { - for sx in 0..png.width { - let index = (sy * png.width + sx) as usize * 4; - let rgba = &png.pixels[index..index + 4]; - if rgba[3] > 0 { // Check alpha channel - self.set_pixel((x + sx as i32) as u16, (y + sy as i32) as u16, rgba[0], rgba[1], rgba[2]); - } - } - } - } -} - -fn main() { - let mut images = vec![ - BouncingImage::new("images/unicorn_cc.png", 13, -10, 1, -1, -1), - BouncingImage::new("images/windows_logo.png", -8, 3, 2, -1, -1), - BouncingImage::new("images/spade.png", 32, -12, 1, 0, 0), - BouncingImage::new("images/dvdvideo.png", 20, 6, 5, 1000, 800), - BouncingImage::new("images/hackaday.png", 40, 18, 3, 500, 800), - ]; - - let mut display = Display::new(DISPLAY_HOST, DISPLAY_PORT); - let mut frame_counter: u32 = 0; - - loop { - for bb in images.iter_mut() { - if bb.rate > 0 && frame_counter % bb.rate != 0 { - continue; - } - bb.draw_and_move(&mut display); - } - - // Send all queued packets for this frame in a single batch. - display.flush_all_pending(); - - frame_counter += 1; - - // A small delay to control the frame rate (approx 60 FPS). - std::thread::sleep(Duration::from_millis(16)); - } -} - diff --git a/src/main.rs b/src/main.rs index 8585c0b..c15791f 100644 --- a/src/main.rs +++ b/src/main.rs @@ -1,396 +1,204 @@ -use std::fs::File; -use std::io::BufReader; -use std::net::{ToSocketAddrs, UdpSocket}; -use std::time::Duration; +use std::net::{SocketAddr, UdpSocket}; +use std::path::PathBuf; +use std::sync::atomic::{AtomicU32, Ordering}; +use std::sync::{Arc, Barrier, Mutex, RwLock}; use std::thread; -use std::sync::{Mutex,Arc}; +use std::time::{Duration, Instant}; + +use clap::Parser; use socket2::{Domain, Socket, Type}; -use std::net::{Ipv4Addr, SocketAddr}; + +mod bouncing_image; +mod circle; mod color; -// Constants from the C code -const QUEUE_LEN: usize = 1000; -const MSG_PAYLOAD_SIZE: usize = 7 * 211; -const MSGSIZE: usize = 2 + MSG_PAYLOAD_SIZE; +mod config; +mod display; +mod drawable; +mod pixel_buf; +mod png_data; +mod raw_socket; +mod pixel; +mod gif_image; -const DISPLAY_HOST: &str = "100.65.0.2"; -const DISPLAY_PORT: u16 = 5005; -const DISPLAY_WIDTH: i32 = 1920; -const DISPLAY_HEIGHT: i32 = 1080; +use bouncing_image::BouncingImage; +use circle::Circle; +use config::Config; +use display::Display; +use drawable::Drawable; +use gif_image::GifImage; +use pixel_buf::PixelBuf; -/// Represents the data decoded from a PNG file. -struct PngData { - width: u32, - height: u32, - pixels: Vec, +pub const DISPLAY_WIDTH: i32 = 1920; +pub const DISPLAY_HEIGHT: i32 = 1080; + +#[derive(Parser)] +#[command(name = "flood-rs", about = "Pixel flooding tool")] +struct Args { + /// Path to the config file + config: PathBuf, } -impl PngData { - /// Loads and decodes a PNG image from the given path. - fn open(path: &str) -> Result { - let file = File::open(path).expect("Failed to open PNG file"); - let decoder = png::Decoder::new(BufReader::new(file)); - let mut reader = decoder.read_info()?; - let mut buf = vec![0; reader.output_buffer_size()]; - let info = reader.next_frame(&mut buf)?; - - Ok(PngData { - width: info.width, - height: info.height, - pixels: buf, - }) - } -} - -/// Represents a bouncing image on the screen. -struct BouncingImage { - img: PngData, - x: i32, - y: i32, - x1: i32, - y1: i32, - x2: i32, - y2: i32, - move_x: i32, - move_y: i32, - rate: u32, -} - -trait Drawable { - // Associated function signature; `Self` refers to the implementor type. - fn rate(&self) -> u32; - fn draw_and_move(&mut self, display: &mut Display,tick:u32); - -} -impl BouncingImage { - /// Initializes a new BouncingImage. - fn new(img_file: &str, move_x: i32, move_y: i32, rate: u32, start_x: i32, start_y: i32) -> Self { - let img = PngData::open(img_file).expect("Could not load image"); - let mut bb = BouncingImage { - x2: DISPLAY_WIDTH - img.width as i32, - y2: DISPLAY_HEIGHT - img.height as i32, - img, - x: start_x, - y: start_y, - x1: 0, - y1: 0, - move_x, - move_y, - rate, - }; - if bb.x == -1 { - bb.x = (bb.x1 + bb.x2) / 2; - } - if bb.y == -1 { - bb.y = (bb.y1 + bb.y2) / 2; - } - bb - } - - - /// Draws a PNG image at the given coordinates. - fn draw_png(&mut self, display: &mut Display, x: i32, y: i32) { - - for sy in 0..self.img.height { - for sx in 0..self.img.width { - let index = (sy * self.img.width + sx) as usize * 4; - let rgba = &self.img.pixels[index..index + 4]; - if rgba[3] > 0 { // Check alpha channel - //display.set_pixel((x + sx as i32) as u16, (y + sy as i32) as u16, rgba[0], rgba[1], rgba[2]); - } - } - } - } - -} -impl Drawable for BouncingImage { - fn rate(&self) -> u32 { - return self.rate; - } - - /// Draws the image and updates its position. - fn draw_and_move(&mut self, display: &mut Display,_: u32) { - self.draw_png(display, self.x, self.y); - - self.x += self.move_x; - self.y += self.move_y; - - if self.x < self.x1 || self.x > self.x2 { - self.move_x *= -1; - } - if self.y < self.y1 || self.y > self.y2 { - self.move_y *= -1; - } - } -} - -struct Circle { - x: Arc>, - y: Arc>, -} -impl Circle { - - fn new(x:Arc>, y: Arc>) -> Self { - Circle { - x, - y - } - } - - /// This exploits the eight-way symmetry of a circle. - fn draw_circle_octants(&mut self, display: &mut Display, cx: i32, cy: i32, x: i32, y: i32, r: u8, g: u8, b: u8) { - display.set_pixel(cx + x, cy + y, r, g, b); - display.set_pixel(cx - x, cy + y, r, g, b); - display.set_pixel(cx + x, cy - y, r, g, b); - display.set_pixel(cx - x, cy - y, r, g, b); - display.set_pixel(cx + y, cy + x, r, g, b); - display.set_pixel(cx - y, cy + x, r, g, b); - display.set_pixel(cx + y, cy - x, r, g, b); - display.set_pixel(cx - y, cy - x, r, g, b); - } - - /// Draws a circle using the Midpoint Circle Algorithm. - /// - /// # Arguments - /// * `center_x`: The x-coordinate of the circle's center. - /// * `center_y`: The y-coordinate of the circle's center. - /// * `radius`: The radius of the circle. Must be non-negative. - /// * `r`, `g`, `b`: The RGB color components for the circle. - pub fn draw_circle(&mut self, display: &mut Display, center_x: u32, center_y: u32, radius: u32, r: u8, g: u8, b: u8) { - if radius < 0 { - // Or return an error: Err("Radius cannot be negative".into()) - return; - } - let radius_i32:i32 = radius.try_into().unwrap(); - let mut x:i32 = 0; - let mut y:i32 = radius_i32; - // Initial decision parameter - let mut d:i32 = 3 - 2 * radius_i32; - - // Iterate through the first octant and draw points in all 8 octants - while y >= x { - self.draw_circle_octants(display,center_x.try_into().unwrap(), center_y.try_into().unwrap(), x, y, r, g, b); - - x += 1; - - // Update the decision parameter - if d > 0 { - y -= 1; - d = d + 4 * (x - y) + 10; - } else { - d = d + 4 * x + 6; - } - } - } - -} -impl Drawable for Circle { - - fn rate(&self) -> u32 { - 1 - } - - /// Helper method to draw the 8 symmetric points for a given (x, y) offset. - fn draw_and_move(&mut self, display: &mut Display,tick:u32) { - let hsv_color = color::Hsv { - h: ((tick/200)%360).try_into().unwrap(), - s: 1.0, - v: 1.0, - }; - let rgb: color::Rgb = hsv_color.into(); - let draw_y =*self.x.lock().unwrap(); - let draw_x = *self.y.lock().unwrap(); - let radius = (tick/30) % (300/2); - self.draw_circle(display,draw_y,draw_x,radius.try_into().unwrap(),rgb.r,rgb.g,rgb.b); - } -} - -/// Manages the connection and data sent to the display. -struct Display { - socket: UdpSocket, - bufs: Vec<[u8; MSGSIZE]>, - next_buf: usize, - pos_in_buf: usize, -} - -impl Display { - /// Creates a new Display and connects to the specified host and port. - fn new(host: &str, port: u16) -> Self { - let remote_addr = (host, port) - .to_socket_addrs() - .expect("Invalid remote address") - .next() - .expect("Could not resolve host"); - - let socket = UdpSocket::bind("0.0.0.0:0").expect("Could not bind to local port"); - socket.connect(remote_addr).expect("Could not connect to remote"); - - let mut bufs = vec![[0; MSGSIZE]; QUEUE_LEN]; - for buf in bufs.iter_mut() { - buf[0] = 0x00; - buf[1] = 0x01; - } - - Display { - socket, - bufs, - next_buf: 0, - pos_in_buf: 0, - } - } - - /// Flushes the current buffer if it contains pixel data. - fn flush_frame(&mut self) { - if self.pos_in_buf > 0 { - let len = 2 + self.pos_in_buf * 7; - let buf_to_send = &self.bufs[self.next_buf][..len]; - self.socket.send(buf_to_send).expect("Failed to send data"); - self.next_buf = (self.next_buf + 1) % QUEUE_LEN; - self.pos_in_buf = 0; - } - } - - /// Sets a pixel color at a specific coordinate. - fn set_pixel(&mut self, x: i32, y: i32, r: u8, g: u8, b: u8) { - if let (Ok(output_x),Ok(output_y)) = (u16::try_from(x), u16::try_from(y)) { - let offset = 2 + self.pos_in_buf * 7; - let buf = &mut self.bufs[self.next_buf][offset..offset + 7]; - buf[0] = output_x as u8; - buf[1] = (output_x >> 8) as u8; - buf[2] = output_y as u8; - buf[3] = (output_y >> 8) as u8; - buf[4] = r; - buf[5] = g; - buf[6] = b; - - self.pos_in_buf += 1; - if self.pos_in_buf == 211 { - self.flush_frame(); - } - } - - - } - - - /// Clears the entire screen to black. - #[allow(dead_code)] - fn blank_screen(&mut self) { - for x in 0..DISPLAY_WIDTH { - for y in 0..DISPLAY_HEIGHT { - //self.set_pixel(x as u16, y as u16, 0, 0, 0); - } - } - self.flush_frame(); - } -} - - /// Unpacks a 4-byte slice into two u16 values (little-endian). fn unpack_coordinates(buffer: &[u8]) -> Option<(u16, u16)> { if buffer.len() != 4 { return None; } - // Try to convert the first 2 bytes to a u16 for x. let x_bytes: [u8; 2] = buffer[0..2].try_into().ok()?; - // Try to convert the next 2 bytes to a u16 for y. let y_bytes: [u8; 2] = buffer[2..4].try_into().ok()?; - - // Reconstruct the u16 values from their little-endian byte representation. let x = u16::from_le_bytes(x_bytes); let y = u16::from_le_bytes(y_bytes); - Some((x, y)) } fn main() { - - - let x:Arc> = Arc::new(Mutex::new(0)); - let x_thread = x.clone(); - - let y:Arc>= Arc::new(Mutex::new(0)); - let y_thread = y.clone(); - - let circle = Box::new(Circle::new(x,y)); - let mut images:Vec> = vec![ -// Box::new(BouncingImage::new("images/unicorn_cc.png", 13, -10, 1, -1, -1)), -// Box::new(BouncingImage::new("images/windows_logo.png", -8, 3, 2, -1, -1)), -// Box::new(BouncingImage::new("images/spade.png", 32, -12, 1, 0, 0)), -// Box::new(BouncingImage::new("images/dvdvideo.png", 20, 6, 5, 1000, 800)), -// Box::new(BouncingImage::new("images/hackaday.png", 40, 18, 3, 500, 800)), - circle - ]; - - let mut display = Display::new(DISPLAY_HOST, DISPLAY_PORT); - let mut frame_counter: u32 = 0; - thread::spawn(move || { - let bind_address = format!("0.0.0.0:12345"); - let socket = Socket::new(Domain::IPV4, Type::DGRAM, None).unwrap(); - socket.set_reuse_address(true).unwrap(); - //socket.set_nonblocking(true).unwrap(); - socket.join_multicast_v4(&Ipv4Addr::new(239, 1, 1, 1), &Ipv4Addr::new(0, 0, 0, 0)).unwrap(); - socket.bind(&"0.0.0.0:1234".parse::().unwrap().into()).unwrap(); - // Bind the UDP socket to the specified address and port. - let socket: UdpSocket = socket.into(); - - println!("Listening for UDP packets on {}", bind_address); - - // Create a buffer to hold incoming data. 4 bytes for two u16 values. - let mut buf = [0u8; 4]; - - loop { - // Wait for a packet to arrive. - match socket.recv_from(&mut buf) { - Ok((number_of_bytes, src_addr)) => { - println!("\nReceived {} bytes from {}", number_of_bytes, src_addr); - - // Ensure we received the correct number of bytes. - if number_of_bytes == 4 { - // Unpack the buffer into coordinates. - if let Some((x_rev, y_rev)) = unpack_coordinates(&buf) { - - println!("Received Coordinates: X = {}, Y = {}", x_rev, y_rev); let x_32:u32 = x_rev.into(); - let y_32:u32 = y_rev.into(); - *x_thread.lock().unwrap() = x_32; - *y_thread.lock().unwrap() = y_32; - - - } else { - // This case should ideally not be reached if number_of_bytes is 4. - eprintln!("Error: Failed to unpack coordinate data."); - } - } else { - eprintln!( - "Warning: Received packet with incorrect size ({} bytes). Expected 4.", - number_of_bytes - ); - } - } - Err(e) => { - eprintln!("Error receiving data: {}", e); - // Decide if you want to break the loop on an error. - // For a continuous server, you might just log and continue. - } - } - } - + let args = Args::parse(); + let config = Config::load(&args.config).unwrap_or_else(|e| { + eprintln!("Error: {}", e); + std::process::exit(1); }); - // display.blank_screen(); - let mut tick:u32 = 0; - loop { - for (i, bb) in images.iter_mut().enumerate() { - if bb.rate() > 0 && frame_counter % bb.rate() != 0 { - continue; - } - bb.draw_and_move(&mut display,tick); + let x: Arc> = Arc::new(Mutex::new(0)); + let x_thread = x.clone(); + + let y: Arc> = Arc::new(Mutex::new(0)); + let y_thread = y.clone(); + + let set: Arc> = Arc::new(Mutex::new(false)); + let set_thread = set.clone(); + + let circle = Box::new(Circle::new(x, y, set)); + let mut images: Vec> = vec![ + // Velocities are in pixels per second. + Box::new(BouncingImage::new("images/ftp.png", 30.0, -30.0, -1, -1)), +// Box::new(BouncingImage::new("images/windows_logo.png", -20.0, 20.0, -1, -1)), +// Box::new(BouncingImage::new("images/spade.png", 90.0, -60.0, 0, 0)), +// Box::new(BouncingImage::new("images/dvdvideo.png", 60.0, 18.0, 1000, 800)), +// Box::new(BouncingImage::new("images/hackaday.png", 40.0, 50.0, 500, 800)), + circle, + Box::new(GifImage::new("images/netto.gif", 200, 200)), + ]; + + let mut display = Display::new(&config.sender); + + // Spawn a UDP listener thread for receiving coordinates + thread::spawn(move || { + let bind_address = "0.0.0.0:12345"; + let socket = Socket::new(Domain::IPV4, Type::DGRAM, None).unwrap(); + socket.set_reuse_address(true).unwrap(); + socket + .bind(&"0.0.0.0:1234".parse::().unwrap().into()) + .unwrap(); + let socket: UdpSocket = socket.into(); + + println!("Listening for UDP packets on {}", bind_address); + + let mut buf = [0u8; 4]; + + loop { + match socket.recv_from(&mut buf) { + Ok((number_of_bytes, src_addr)) => { + println!("\nReceived {} bytes from {}", number_of_bytes, src_addr); + + if number_of_bytes == 4 { + if let Some((x_rev, y_rev)) = unpack_coordinates(&buf) { + println!("Received Coordinates: X = {}, Y = {}", x_rev, y_rev); + *x_thread.lock().unwrap() = x_rev.into(); + *y_thread.lock().unwrap() = y_rev.into(); + *set_thread.lock().unwrap() = true; + } else { + eprintln!("Error: Failed to unpack coordinate data."); + } + } else { + eprintln!( + "Warning: Received packet with incorrect size ({} bytes). Expected 4.", + number_of_bytes + ); } + } + Err(e) => { + eprintln!("Error receiving data: {}", e); + } + } + } + }); + + let num_drawables = images.len(); + + // Wrap drawables in Arc> for shared access: + // - Workers take read locks for draw(&self) -- parallel, no contention + // - Main takes write locks for update(&mut self) -- exclusive, 30fps only + let images: Vec>>> = images + .into_iter() + .map(|b| Arc::new(RwLock::new(b))) + .collect(); + + // Shared elapsed time for draw threads (as u32 bits of f32). + let elapsed_bits = Arc::new(AtomicU32::new(0)); + + // Barriers: main signals "go", workers draw, then signal "done". + let go_barrier = Arc::new(Barrier::new(num_drawables + 1)); + let done_barrier = Arc::new(Barrier::new(num_drawables + 1)); + + // Each worker gets its own PixelBuf wrapped in Mutex so the main + // thread can read the results after the done barrier. + let pixel_bufs: Vec>> = (0..num_drawables) + .map(|_| Arc::new(Mutex::new(PixelBuf::new(60_000)))) + .collect(); + + // Spawn persistent worker threads. + for i in 0..num_drawables { + let img = Arc::clone(&images[i]); + let buf = Arc::clone(&pixel_bufs[i]); + let elapsed = Arc::clone(&elapsed_bits); + let go = Arc::clone(&go_barrier); + let done = Arc::clone(&done_barrier); + + thread::spawn(move || { + loop { + go.wait(); + + let e = f32::from_bits(elapsed.load(Ordering::Relaxed)); + let drawable = img.read().unwrap(); + let mut pb = buf.lock().unwrap(); + pb.clear(); + drawable.draw(&mut pb, e); + drop(pb); + drop(drawable); + + done.wait(); + } + }); + } + + let start_time = Instant::now(); + let mut last_update = Instant::now(); + let update_interval = Duration::from_secs_f32(1.0 / 30.0); + + loop { + let now = Instant::now(); + let elapsed = (now - start_time).as_secs_f32(); + + // Update positions at 30 fps (main thread only, workers are idle). + if now - last_update >= update_interval { + let dt = (now - last_update).as_secs_f32(); + for img in &images { + img.write().unwrap().update(dt, elapsed); + } + last_update = now; + } + + // Publish elapsed time and signal workers to draw. + elapsed_bits.store(elapsed.to_bits(), Ordering::Relaxed); + go_barrier.wait(); + + // Wait for all workers to finish drawing. + done_barrier.wait(); + + // Merge all pixel bufs into the single Display for efficient packing. + for pb in &pixel_bufs { + let buf = pb.lock().unwrap(); + display.write_raw_pixels(&buf.pixels); + } display.flush_frame(); - tick+=1; - frame_counter += 1; - - // A small delay to control the frame rate - //std::thread::sleep(Duration::from_millis(16)); } } - diff --git a/src/pixel.rs b/src/pixel.rs new file mode 100644 index 0000000..ab3a3af --- /dev/null +++ b/src/pixel.rs @@ -0,0 +1,22 @@ +#[derive(Copy, Clone)] +pub struct Pixel { + pub x: u16, + pub y: u16, + pub r: u8, + pub g: u8, + pub b: u8 +} +impl Pixel { + pub fn format0(&self) -> [u8; 7] { + + let buf:[u8; 7] = [ self.x as u8 + , (self.x >> 8) as u8 + ,self.y as u8 + ,(self.y >> 8) as u8 + , self.r + , self.g + , self.b]; + return buf; + } + +} \ No newline at end of file diff --git a/src/pixel_buf.rs b/src/pixel_buf.rs new file mode 100644 index 0000000..83721fa --- /dev/null +++ b/src/pixel_buf.rs @@ -0,0 +1,46 @@ +/// A lightweight per-thread pixel accumulator. +/// +/// Stores already-adjusted 7-byte pixel entries `[x_lo, x_hi, y_lo, y_hi, r, g, b]` +/// ready to be merged into the main Display buffer. No packets, no flushing, +/// no queue -- just a growable buffer that each draw thread fills independently. +use crate::pixel::Pixel; +pub struct PixelBuf { + pub pixels: Vec<[u8;7]>, +} + +impl PixelBuf { + /// Creates a new empty PixelBuf with the given initial capacity. + pub fn new(capacity: usize) -> Self { + PixelBuf { + pixels: Vec::with_capacity(capacity), + } + } + + /// Clears the buffer for reuse without deallocating. + #[inline] + pub fn clear(&mut self) { + self.pixels.clear(); + } + + #[inline] + pub fn set_pixel(&mut self, pixel: Pixel) { + self.pixels.push(pixel.format0()); + } + + #[inline] + pub fn set_pixels(&mut self, pixels: Vec) { + self.pixels.reserve(pixels.len()); + for pixel in pixels { + self.pixels.push(pixel.format0()) + } + } + + /// Appends pixels from a borrowed slice (avoids cloning precomputed data). + #[inline] + pub fn set_pixels_ref(&mut self, pixels: &[Pixel]) { + self.pixels.reserve(pixels.len()); + for pixel in pixels { + self.pixels.push(pixel.format0()) + } + } +} diff --git a/src/png_data.rs b/src/png_data.rs new file mode 100644 index 0000000..aa7398f --- /dev/null +++ b/src/png_data.rs @@ -0,0 +1,55 @@ +use std::fs::File; +use std::io::BufReader; +use crate::pixel::Pixel; +/// Represents the data decoded from a PNG file, with pre-computed pixel data. +pub struct PngData { + pub width: u32, + pub height: u32, + pub packed_pixels: Vec, +} +impl PngData { + + pub fn open(path: &str) -> Result { + let file = File::open(path).expect("Failed to open PNG file"); + let decoder = png::Decoder::new(BufReader::new(file)); + let mut reader = decoder.read_info()?; + let mut buf = vec![0; reader.output_buffer_size()]; + let info = reader.next_frame(&mut buf)?; + + let width = info.width; + let height = info.height; + + let mut packed_pixels = Vec::new(); + for sy in 0..height { + for sx in 0..width { + let index = (sy * width + sx) as usize * 4; + let a = buf[index + 3]; + if a > 0 { + let r = buf[index]; + let g = buf[index + 1]; + let b = buf[index + 2]; + let x = sx as u16; + let y = sy as u16; + packed_pixels.push(Pixel { + x, + y, + r, + g, + b + }) + } + } + } + + eprintln!( + "[png] {}: {}x{}, {} opaque pixels pre-computed", + path, width, height, packed_pixels.len() + ); + + Ok(PngData { + width, + height, + packed_pixels, + }) + } +} diff --git a/src/raw_socket.rs b/src/raw_socket.rs new file mode 100644 index 0000000..bc022cd --- /dev/null +++ b/src/raw_socket.rs @@ -0,0 +1,467 @@ +//! Raw AF_PACKET sender that constructs complete Ethernet + IP + UDP frames +//! in userspace and sends them directly to the NIC, bypassing the kernel's +//! UDP/IP stack entirely. + +use std::fs; +use std::io::{self, BufRead, BufReader}; +use std::net::{Ipv4Addr, ToSocketAddrs, UdpSocket}; +use std::os::unix::io::AsRawFd; + +/// Size of Ethernet + IPv4 + UDP headers combined. +const HEADER_SIZE: usize = 14 + 20 + 8; // 42 bytes + +/// PACKET_QDISC_BYPASS socket option (skip the qdisc layer). +const PACKET_QDISC_BYPASS: libc::c_int = 20; + +/// A raw AF_PACKET sender that constructs full Ethernet frames. +pub struct RawSender { + fd: libc::c_int, + ifindex: i32, + /// Pre-built header template (Ethernet + IP + UDP). + /// Only the IP total_length, IP checksum, and UDP length fields + /// need to be patched per packet. + header_template: [u8; HEADER_SIZE], +} + +impl RawSender { + /// Creates a new RawSender. + /// + /// Auto-discovers the network interface, MAC addresses, and source IP + /// from the routing table and ARP cache. + /// + /// Requires `CAP_NET_RAW` or root. + pub fn new(dst_host: &str, dst_port: u16, src_port: u16, interface: Option<&str>) -> io::Result { + // Resolve destination IP. + let dst_ip: Ipv4Addr = dst_host + .parse() + .or_else(|_| { + (dst_host, dst_port) + .to_socket_addrs()? + .find_map(|a| match a { + std::net::SocketAddr::V4(v4) => Some(*v4.ip()), + _ => None, + }) + .ok_or_else(|| io::Error::new(io::ErrorKind::Other, "Could not resolve host")) + })?; + + // Discover or use the specified source interface and IP. + let (ifname, ifindex, src_ip) = if let Some(ifname) = interface { + let ifindex = get_ifindex(ifname)?; + let src_ip = get_interface_ip(ifname)?; + (ifname.to_owned(), ifindex, src_ip) + } else { + let probe = UdpSocket::bind("0.0.0.0:0")?; + probe.connect((dst_ip, dst_port))?; + let src_ip: Ipv4Addr = match probe.local_addr()? { + std::net::SocketAddr::V4(a) => *a.ip(), + _ => return Err(io::Error::new(io::ErrorKind::Other, "Not IPv4")), + }; + let (ifname, ifindex) = discover_interface(dst_ip)?; + (ifname, ifindex, src_ip) + }; + eprintln!("[raw] interface: {} (index {})", ifname, ifindex); + + // Read source MAC from sysfs. + let src_mac = read_mac_from_sysfs(&ifname)?; + eprintln!("[raw] src MAC: {:02x}:{:02x}:{:02x}:{:02x}:{:02x}:{:02x}", + src_mac[0], src_mac[1], src_mac[2], src_mac[3], src_mac[4], src_mac[5]); + + // Look up destination MAC from ARP cache. + let dst_mac = lookup_arp(dst_ip, &ifname)?; + eprintln!("[raw] dst MAC: {:02x}:{:02x}:{:02x}:{:02x}:{:02x}:{:02x}", + dst_mac[0], dst_mac[1], dst_mac[2], dst_mac[3], dst_mac[4], dst_mac[5]); + + eprintln!("[raw] {}:{} -> {}:{}", src_ip, src_port, dst_ip, dst_port); + + // Create AF_PACKET raw socket. + let fd = unsafe { + libc::socket( + libc::AF_PACKET, + libc::SOCK_RAW, + (libc::ETH_P_IP as u16).to_be() as i32, + ) + }; + if fd < 0 { + return Err(io::Error::last_os_error()); + } + + // Bind to the interface. + let mut sll: libc::sockaddr_ll = unsafe { std::mem::zeroed() }; + sll.sll_family = libc::AF_PACKET as u16; + sll.sll_protocol = (libc::ETH_P_IP as u16).to_be(); + sll.sll_ifindex = ifindex; + let ret = unsafe { + libc::bind( + fd, + &sll as *const libc::sockaddr_ll as *const libc::sockaddr, + std::mem::size_of::() as libc::socklen_t, + ) + }; + if ret < 0 { + unsafe { libc::close(fd); } + return Err(io::Error::last_os_error()); + } + + // Skip qdisc for lower latency and higher throughput. + let bypass: libc::c_int = 1; + unsafe { + libc::setsockopt( + fd, + libc::SOL_PACKET, + PACKET_QDISC_BYPASS, + &bypass as *const libc::c_int as *const libc::c_void, + std::mem::size_of::() as libc::socklen_t, + ); + } + + // Increase send buffer. + let sndbuf: libc::c_int = 25 * 1024 * 1024; + unsafe { + libc::setsockopt( + fd, + libc::SOL_SOCKET, + libc::SO_SNDBUF, + &sndbuf as *const libc::c_int as *const libc::c_void, + std::mem::size_of::() as libc::socklen_t, + ); + } + + // Build the header template. + let header_template = build_header_template( + &src_mac, &dst_mac, src_ip, dst_ip, src_port, dst_port, + ); + + Ok(RawSender { + fd, + ifindex, + header_template, + }) + } + + /// Sends a batch of payloads as raw Ethernet frames using sendmmsg. + /// + /// `frames` is a pre-allocated buffer of frame data. + /// Each frame is HEADER_SIZE + payload bytes. + /// Returns the number of frames successfully sent. + pub fn send_batch(&self, payloads: &[&[u8]], frame_buf: &mut Vec) -> io::Result { + let count = payloads.len(); + if count == 0 { + return Ok(0); + } + + // Build all frames into the contiguous buffer. + frame_buf.clear(); + let mut offsets: Vec<(usize, usize)> = Vec::with_capacity(count); + + for payload in payloads { + let frame_start = frame_buf.len(); + let total_len = (20 + 8 + payload.len()) as u16; // IP total length + let udp_len = (8 + payload.len()) as u16; + + // Copy header template. + frame_buf.extend_from_slice(&self.header_template); + + // Patch IP total length (bytes 16-17 of the frame = offset 2-3 in IP header). + let ip_start = frame_start + 14; + frame_buf[ip_start + 2] = (total_len >> 8) as u8; + frame_buf[ip_start + 3] = total_len as u8; + + // Patch IP header checksum (bytes 10-11 of IP header). + // Zero the checksum field first, then compute. + frame_buf[ip_start + 10] = 0; + frame_buf[ip_start + 11] = 0; + let cksum = ip_checksum(&frame_buf[ip_start..ip_start + 20]); + frame_buf[ip_start + 10] = (cksum >> 8) as u8; + frame_buf[ip_start + 11] = cksum as u8; + + // Patch UDP length (bytes 4-5 of UDP header). + let udp_start = frame_start + 14 + 20; + frame_buf[udp_start + 4] = (udp_len >> 8) as u8; + frame_buf[udp_start + 5] = udp_len as u8; + + // UDP checksum = 0 (optional for IPv4). + frame_buf[udp_start + 6] = 0; + frame_buf[udp_start + 7] = 0; + + // Append payload. + frame_buf.extend_from_slice(payload); + + let frame_end = frame_buf.len(); + offsets.push((frame_start, frame_end)); + } + + // Build iovec and mmsghdr arrays for sendmmsg. + let mut iovecs: Vec = Vec::with_capacity(count); + let mut msghdrs: Vec = Vec::with_capacity(count); + + // Destination sockaddr_ll. + let mut sll: libc::sockaddr_ll = unsafe { std::mem::zeroed() }; + sll.sll_family = libc::AF_PACKET as u16; + sll.sll_protocol = (libc::ETH_P_IP as u16).to_be(); + sll.sll_ifindex = self.ifindex; + sll.sll_halen = 6; + // dst MAC is in the first 6 bytes of each frame, but sockaddr_ll + // also needs it. Copy from the template. + sll.sll_addr[..6].copy_from_slice(&self.header_template[..6]); + + for &(start, end) in &offsets { + iovecs.push(libc::iovec { + iov_base: frame_buf[start..end].as_ptr() as *mut libc::c_void, + iov_len: end - start, + }); + } + + for iov in &iovecs { + let mut mhdr: libc::mmsghdr = unsafe { std::mem::zeroed() }; + mhdr.msg_hdr.msg_iov = iov as *const libc::iovec as *mut libc::iovec; + mhdr.msg_hdr.msg_iovlen = 1; + mhdr.msg_hdr.msg_name = &sll as *const libc::sockaddr_ll as *mut libc::c_void; + mhdr.msg_hdr.msg_namelen = std::mem::size_of::() as u32; + msghdrs.push(mhdr); + } + + let ret = unsafe { + libc::sendmmsg( + self.fd, + msghdrs.as_mut_ptr(), + count as libc::c_uint, + 0, + ) + }; + + if ret < 0 { + Err(io::Error::last_os_error()) + } else { + Ok(ret as usize) + } + } + + /// Returns the header size so callers can compute total frame sizes. + pub fn header_size(&self) -> usize { + HEADER_SIZE + } +} + +impl Drop for RawSender { + fn drop(&mut self) { + unsafe { + libc::close(self.fd); + } + } +} + +/// Builds the 42-byte header template: Ethernet (14) + IP (20) + UDP (8). +/// +/// Fields that vary per packet (IP total_length, IP checksum, UDP length) +/// are filled with placeholder values and must be patched before sending. +fn build_header_template( + src_mac: &[u8; 6], + dst_mac: &[u8; 6], + src_ip: Ipv4Addr, + dst_ip: Ipv4Addr, + src_port: u16, + dst_port: u16, +) -> [u8; HEADER_SIZE] { + let mut h = [0u8; HEADER_SIZE]; + + // -- Ethernet header (14 bytes) -- + h[0..6].copy_from_slice(dst_mac); + h[6..12].copy_from_slice(src_mac); + h[12] = 0x08; // EtherType: IPv4 + h[13] = 0x00; + + // -- IPv4 header (20 bytes, no options) -- + let ip = &mut h[14..34]; + ip[0] = 0x45; // Version 4, IHL 5 (20 bytes) + ip[1] = 0x00; // DSCP/ECN + // ip[2..4] = total length (patched per packet) + ip[4] = 0x00; // Identification + ip[5] = 0x00; + ip[6] = 0x40; // Flags: Don't Fragment + ip[7] = 0x00; // Fragment offset + ip[8] = 64; // TTL + ip[9] = 17; // Protocol: UDP + // ip[10..12] = header checksum (patched per packet) + ip[12..16].copy_from_slice(&src_ip.octets()); + ip[16..20].copy_from_slice(&dst_ip.octets()); + + // -- UDP header (8 bytes) -- + let udp = &mut h[34..42]; + udp[0] = (src_port >> 8) as u8; + udp[1] = src_port as u8; + udp[2] = (dst_port >> 8) as u8; + udp[3] = dst_port as u8; + // udp[4..6] = length (patched per packet) + // udp[6..8] = checksum (set to 0, optional for IPv4) + + h +} + +/// Computes the IPv4 header checksum (one's complement of the one's +/// complement sum of all 16-bit words in the header). +fn ip_checksum(header: &[u8]) -> u16 { + let mut sum: u32 = 0; + for i in (0..header.len()).step_by(2) { + let word = if i + 1 < header.len() { + ((header[i] as u32) << 8) | (header[i + 1] as u32) + } else { + (header[i] as u32) << 8 + }; + sum += word; + } + // Fold carry bits. + while sum > 0xFFFF { + sum = (sum & 0xFFFF) + (sum >> 16); + } + !(sum as u16) +} + +/// Gets the ifindex for a named interface from sysfs. +fn get_ifindex(ifname: &str) -> io::Result { + let path = format!("/sys/class/net/{}/ifindex", ifname); + let s = fs::read_to_string(&path).map_err(|e| { + io::Error::new(e.kind(), format!("Interface '{}' not found: {}", ifname, e)) + })?; + s.trim().parse().map_err(|e| { + io::Error::new(io::ErrorKind::InvalidData, format!("Bad ifindex: {}", e)) + }) +} + +/// Gets the IPv4 address of a named interface using ioctl. +fn get_interface_ip(ifname: &str) -> io::Result { + let sock = UdpSocket::bind("0.0.0.0:0")?; + let fd = sock.as_raw_fd(); + let mut ifr: libc::ifreq = unsafe { std::mem::zeroed() }; + let name_bytes = ifname.as_bytes(); + let copy_len = name_bytes.len().min(libc::IFNAMSIZ - 1); + unsafe { + std::ptr::copy_nonoverlapping( + name_bytes.as_ptr(), + ifr.ifr_name.as_mut_ptr() as *mut u8, + copy_len, + ); + } + let ret = unsafe { libc::ioctl(fd, libc::SIOCGIFADDR, &mut ifr) }; + if ret < 0 { + return Err(io::Error::new( + io::ErrorKind::Other, + format!("Could not get IP for interface '{}'", ifname), + )); + } + let addr = unsafe { &*(&ifr.ifr_ifru as *const _ as *const libc::sockaddr_in) }; + Ok(Ipv4Addr::from(u32::from_be(addr.sin_addr.s_addr))) +} + +/// Discovers the outbound interface name and index for a destination IP. +fn discover_interface(dst_ip: Ipv4Addr) -> io::Result<(String, i32)> { + // Use a temporary UDP socket to discover the interface. + let sock = UdpSocket::bind("0.0.0.0:0")?; + sock.connect((dst_ip, 80))?; + + // Get the interface index via SO_BINDTODEVICE or by reading /proc/net/route. + // Simpler: use the socket's bound address to find the matching interface. + let local_ip = match sock.local_addr()? { + std::net::SocketAddr::V4(a) => *a.ip(), + _ => return Err(io::Error::new(io::ErrorKind::Other, "Not IPv4")), + }; + + // Scan /proc/net/if_inet6 or /proc/net/fib_trie... actually, simplest + // is to iterate /sys/class/net/*/address and check IPs. + // But even simpler: parse `ip route get` output... that's ugly. + // + // Let's iterate interfaces and match by IP. + let entries = fs::read_dir("/sys/class/net/")?; + for entry in entries { + let entry = entry?; + let ifname = entry.file_name().to_string_lossy().into_owned(); + + // Get the interface index. + let ifindex_path = format!("/sys/class/net/{}/ifindex", ifname); + let ifindex_str = match fs::read_to_string(&ifindex_path) { + Ok(s) => s, + Err(_) => continue, + }; + let ifindex: i32 = match ifindex_str.trim().parse() { + Ok(i) => i, + Err(_) => continue, + }; + + // Check if this interface has the local IP using ioctl. + let fd = sock.as_raw_fd(); + let mut ifr: libc::ifreq = unsafe { std::mem::zeroed() }; + let name_bytes = ifname.as_bytes(); + let copy_len = name_bytes.len().min(libc::IFNAMSIZ - 1); + unsafe { + std::ptr::copy_nonoverlapping( + name_bytes.as_ptr(), + ifr.ifr_name.as_mut_ptr() as *mut u8, + copy_len, + ); + } + + let ret = unsafe { + libc::ioctl(fd, libc::SIOCGIFADDR, &mut ifr) + }; + if ret < 0 { + continue; + } + + let addr = unsafe { &*(&ifr.ifr_ifru as *const _ as *const libc::sockaddr_in) }; + let iface_ip = Ipv4Addr::from(u32::from_be(addr.sin_addr.s_addr)); + + if iface_ip == local_ip { + return Ok((ifname, ifindex)); + } + } + + Err(io::Error::new( + io::ErrorKind::NotFound, + format!("Could not find interface for IP {}", local_ip), + )) +} + +/// Reads the MAC address of a network interface from sysfs. +fn read_mac_from_sysfs(ifname: &str) -> io::Result<[u8; 6]> { + let path = format!("/sys/class/net/{}/address", ifname); + let mac_str = fs::read_to_string(path)?; + parse_mac(mac_str.trim()) +} + +/// Looks up the MAC address for an IP in the kernel ARP cache. +fn lookup_arp(ip: Ipv4Addr, ifname: &str) -> io::Result<[u8; 6]> { + let file = fs::File::open("/proc/net/arp")?; + let reader = BufReader::new(file); + + let ip_str = ip.to_string(); + + for line in reader.lines().skip(1) { + let line = line?; + let fields: Vec<&str> = line.split_whitespace().collect(); + if fields.len() >= 6 && fields[0] == ip_str && fields[5] == ifname { + return parse_mac(fields[3]); + } + } + + Err(io::Error::new( + io::ErrorKind::NotFound, + format!("No ARP entry for {} on {}", ip, ifname), + )) +} + +/// Parses a MAC address string like "50:6b:4b:c3:fa:9c" into 6 bytes. +fn parse_mac(s: &str) -> io::Result<[u8; 6]> { + let parts: Vec<&str> = s.split(':').collect(); + if parts.len() != 6 { + return Err(io::Error::new( + io::ErrorKind::InvalidData, + format!("Invalid MAC: {}", s), + )); + } + let mut mac = [0u8; 6]; + for (i, part) in parts.iter().enumerate() { + mac[i] = u8::from_str_radix(part, 16).map_err(|_| { + io::Error::new(io::ErrorKind::InvalidData, format!("Invalid MAC byte: {}", part)) + })?; + } + Ok(mac) +}