Compare commits

...
Sign in to create a new pull request.

10 commits
master ... test

Author SHA1 Message Date
4cb494576b better ftp 2026-07-20 00:43:10 +02:00
10355620d2 car 2026-07-19 20:53:31 +02:00
c9e10cd2c3 config 2026-07-18 20:57:58 +02:00
28d5d6db14 shuffel for every frame 2026-07-18 20:48:20 +02:00
1e93198c31 noise 2026-07-18 16:46:47 +02:00
495e7e3b07 fix delay 2026-07-18 15:56:19 +02:00
07021aef6b host arg and gif 2026-07-18 15:53:56 +02:00
f70c6c716a wip 2026-07-18 03:05:55 +02:00
73c3462543 wip 2026-07-18 03:05:48 +02:00
f6e64b6e0b wip 2026-07-17 16:16:36 +02:00
20 changed files with 1992 additions and 642 deletions

319
Cargo.lock generated
View file

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

View file

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

14
config.toml Normal file
View file

@ -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

27
flake.lock generated Normal file
View file

@ -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
}

29
flake.nix Normal file
View file

@ -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
];
};
};
}

BIN
images/ftp.png Normal file

Binary file not shown.

After

Width:  |  Height:  |  Size: 115 KiB

BIN
images/ftp2.png Normal file

Binary file not shown.

After

Width:  |  Height:  |  Size: 83 KiB

BIN
images/netto.gif Normal file

Binary file not shown.

After

Width:  |  Height:  |  Size: 842 KiB

85
src/bouncing_image.rs Normal file
View file

@ -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<Pixel>, dx: u16, dy: u16) -> Vec<Pixel> {
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);
}
}
}

109
src/circle.rs Normal file
View file

@ -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<Mutex<u32>>,
y: Arc<Mutex<u32>>,
set: Arc<Mutex<bool>>,
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<Mutex<u32>>, y: Arc<Mutex<u32>>, set: Arc<Mutex<bool>>) -> 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();
}
}

54
src/config.rs Normal file
View file

@ -0,0 +1,54 @@
use serde::Deserialize;
use std::fs;
use std::path::Path;
#[derive(Deserialize)]
pub struct Config {
pub sender: Vec<SenderConfig>,
}
#[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<String>,
/// 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<Self, String> {
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)
}
}

408
src/display.rs Normal file
View file

@ -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<VecDeque<Packet>>,
condvar: Condvar,
pool: Mutex<Vec<Box<[u8; MSGSIZE]>>>,
}
/// 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<Vec<[u8;7]>> 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<Mutex<Arc<Vec<[u8; 7]>>>>,
}
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(&current_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<Mutex<Arc<Vec<[u8; 7]>>>>,
queue: Arc<SenderQueue>,
seed: u64,
) {
let mut rng = Xorshift64(seed);
let mut permutation: Vec<usize> = 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<SenderQueue>) {
let mut batch: Vec<Packet> = Vec::with_capacity(SEND_BATCH_SIZE);
let mut frame_buf: Vec<u8> = 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<SenderQueue>) {
let mut batch: Vec<Packet> = 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;
}
}
}

8
src/drawable.rs Normal file
View file

@ -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);
}

164
src/gif_image.rs Normal file
View file

@ -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<Pixel>,
/// 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<GifFrame>,
/// 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.
}
}

View file

@ -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<u8>,
}
impl PngData {
/// Loads and decodes a PNG image from the given path.
fn open(path: &str) -> Result<Self, png::DecodingError> {
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<usize>, // 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));
}
}

View file

@ -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<u8>,
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<Self, png::DecodingError> {
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<Mutex<u32>>,
y: Arc<Mutex<u32>>,
}
impl Circle {
fn new(x:Arc<Mutex<u32>>, y: Arc<Mutex<u32>>) -> 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<Mutex<u32>> = Arc::new(Mutex::new(0));
let x_thread = x.clone();
let y:Arc<Mutex<u32>>= Arc::new(Mutex::new(0));
let y_thread = y.clone();
let circle = Box::new(Circle::new(x,y));
let mut images:Vec<Box<dyn Drawable>> = 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::<SocketAddr>().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<Mutex<u32>> = Arc::new(Mutex::new(0));
let x_thread = x.clone();
let y: Arc<Mutex<u32>> = Arc::new(Mutex::new(0));
let y_thread = y.clone();
let set: Arc<Mutex<bool>> = Arc::new(Mutex::new(false));
let set_thread = set.clone();
let circle = Box::new(Circle::new(x, y, set));
let mut images: Vec<Box<dyn Drawable>> = 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::<SocketAddr>().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<RwLock<>> 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<Arc<RwLock<Box<dyn Drawable>>>> = 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<Arc<Mutex<PixelBuf>>> = (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));
}
}

22
src/pixel.rs Normal file
View file

@ -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;
}
}

46
src/pixel_buf.rs Normal file
View file

@ -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<Pixel>) {
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())
}
}
}

55
src/png_data.rs Normal file
View file

@ -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<Pixel>,
}
impl PngData {
pub fn open(path: &str) -> Result<Self, png::DecodingError> {
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,
})
}
}

467
src/raw_socket.rs Normal file
View file

@ -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<Self> {
// 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::<libc::sockaddr_ll>() 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::<libc::c_int>() 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::<libc::c_int>() 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<u8>) -> io::Result<usize> {
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<libc::iovec> = Vec::with_capacity(count);
let mut msghdrs: Vec<libc::mmsghdr> = 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::<libc::sockaddr_ll>() 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<i32> {
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<Ipv4Addr> {
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)
}