facett-cfd/src/fluid.rs
//! **fluid** — a cheap, convincing **1-D dye-advection fluid** for showing *fluid
//! flowing through pipes* on the facett **system map** (`facett-graph3d::SystemMap`,
//! wired into `facett-demo`). This is `facett-cfd`'s first **gatling** consumer and
//! its reusable **accelerated fluid API**: any facett app that wants "data/activity
//! flowing along edges" drops a [`FluidField`] on its edges and steps it.
//!
//! ## Model
//! One field of dye concentration per **pipe**, `cells` samples along the pipe. Each
//! step is an **upwind advection** along the pipe (signed per-pipe velocity = flow
//! direction/rate), plus gentle diffusion, a slight decay, and a **pulsing dye
//! source** injected at the inlet so you *see* packets enter and travel. Pipes are
//! independent, so the step is embarrassingly parallel.
//!
//! ## The three zeroes (the owner's ask — met on the per-step path)
//! - **ZERO-ALLOCATION** — every buffer ([`FluidField::front`]/`back`, `vel`,
//! `inject`) is allocated **once** in [`FluidField::new`]. [`FluidField::step`]
//! touches only those buffers; it never `Vec::new`/`clone`/`collect`/`format!`s. A
//! test ([`tests::step_is_zero_alloc_double_buffer_ping_pong`]) pins the buffer
//! pointers + capacities across 240 steps — a stray realloc/clone flips a pointer
//! and goes RED.
//! - **ZERO-COPY** — a genuine **double buffer**: the step reads `front`, writes
//! `back`, then `std::mem::swap`s the two `Vec`s (a pointer swap, no element copy).
//! The renderer borrows the current dye with [`FluidField::dye`] — a borrowed slice,
//! never a clone.
//! - **ZERO-WAIT** — the per-pipe work fans across all cores through the znippy
//! **gatling** fork-join ([`gatling_run`]): no barrier, no lock, no `.await`; it
//! returns within the frame budget. `rayon` is FORBIDDEN in facett — gatling only.
//! (On a 1-core host / wasm, gatling degrades to a serial drain — same math.)
use znippy_zoomies::gatling_forkjoin::gatling_run;
/// Diffusion coefficient (per second) — a touch of smear so packets read as fluid,
/// not hard pixels.
const DIFFUSION: f32 = 0.35;
/// Dye decay (per second) — keeps a pipe from saturating to solid colour.
const DECAY: f32 = 0.25;
/// Inlet pulse frequency (Hz) — the source breathes so discrete dye packets form
/// and march down the pipe (visible flow direction + rate).
const PULSE_HZ: f32 = 0.9;
const TAU: f32 = std::f32::consts::TAU;
/// Hard clamp on concentration so a numerical blip can never explode the fill.
const DYE_MAX: f32 = 1.5;
/// A `Send`+`Sync` raw base pointer into the `back` buffer, so gatling workers can
/// each write their **own** disjoint pipe row without a lock. Soundness: worker `p`
/// writes only `back[p*cells .. (p+1)*cells]`, and every `p in 0..pipes` is claimed
/// by exactly one worker (gatling's atomic counter), so the writes never alias — the
/// same disjoint-index argument the gatling engine and `facett_core::barnes_hut` use.
#[derive(Clone, Copy)]
struct RowsMut(*mut f32);
// SAFETY: only ever used to write disjoint, per-worker-unique index ranges.
unsafe impl Send for RowsMut {}
unsafe impl Sync for RowsMut {}
/// A 1-D dye-advection fluid over `pipes` pipes of `cells` samples each. All state is
/// preallocated; [`step`](Self::step) is zero-alloc / zero-copy / zero-wait.
pub struct FluidField {
pipes: usize,
cells: usize,
/// Current dye, row-major `[pipe*cells + i]`. The buffer the renderer reads.
front: Vec<f32>,
/// Scratch back buffer the step writes, then swaps with `front` (double buffer).
back: Vec<f32>,
/// Per-pipe signed flow speed in **cells/second** (sign = direction along the
/// pipe: `>= 0` flows inlet→outlet at index 0→last, `< 0` the other way).
vel: Vec<f32>,
/// Per-pipe inlet dye injection rate.
inject: Vec<f32>,
/// Accumulated sim time (seconds) — drives the pulsing inlet.
t: f32,
}
impl FluidField {
/// Allocate a field for `pipes` pipes × `cells` samples — **the only allocation**.
/// Defaults: a gentle downstream flow + a steady inlet on every pipe, so a freshly
/// built field already flows.
pub fn new(pipes: usize, cells: usize) -> Self {
let cells = cells.max(2);
Self {
pipes,
cells,
front: vec![0.0; pipes * cells],
back: vec![0.0; pipes * cells],
vel: vec![6.0; pipes],
inject: vec![0.8; pipes],
t: 0.0,
}
}
/// Number of pipes.
#[must_use]
pub fn pipes(&self) -> usize {
self.pipes
}
/// Samples per pipe.
#[must_use]
pub fn cells(&self) -> usize {
self.cells
}
/// Accumulated sim time (seconds).
#[must_use]
pub fn time(&self) -> f32 {
self.t
}
/// Set pipe `p`'s signed flow speed (cells/second). Out-of-range is ignored.
pub fn set_flow(&mut self, p: usize, cells_per_sec: f32) {
if let Some(v) = self.vel.get_mut(p) {
*v = cells_per_sec;
}
}
/// Set pipe `p`'s inlet dye injection rate. Out-of-range is ignored.
pub fn set_inject(&mut self, p: usize, rate: f32) {
if let Some(v) = self.inject.get_mut(p) {
*v = rate;
}
}
/// Pipe `p`'s signed flow speed (cells/second), `0.0` if out of range.
#[must_use]
pub fn flow(&self, p: usize) -> f32 {
self.vel.get(p).copied().unwrap_or(0.0)
}
/// **Borrow** pipe `p`'s current dye row (`cells` samples) — the zero-copy read
/// the renderer samples along the pipe. Empty slice if `p` is out of range.
#[must_use]
pub fn dye(&self, p: usize) -> &[f32] {
if p < self.pipes {
&self.front[p * self.cells..(p + 1) * self.cells]
} else {
&[]
}
}
/// Total dye in the field (an observable "how much is flowing" scalar).
#[must_use]
pub fn total_dye(&self) -> f32 {
self.front.iter().copied().sum()
}
/// **The accelerated step (gatling / zero-alloc / zero-copy / zero-wait).** Advance
/// the whole field by `dt` seconds: fan the per-pipe advection across all cores
/// through the znippy gatling fork-join (each worker owns one pipe's disjoint back
/// row), then swap the double buffer. No allocation, no lock, no wait.
pub fn step(&mut self, dt: f32) {
self.t += dt;
let cells = self.cells;
let pipes = self.pipes;
if pipes == 0 {
return;
}
// Disjoint borrows of distinct fields: `front`/`vel`/`inject` shared, `back`
// unique — the borrow checker allows this because they are different fields.
let front: &[f32] = &self.front;
let vel: &[f32] = &self.vel;
let inject: &[f32] = &self.inject;
let t = self.t;
let back = RowsMut(self.back.as_mut_ptr());
// gatling: one worker per core drains the pipe counter; `0` ⇒ default workers.
gatling_run(pipes, 0, |p| {
// Force whole-struct capture of the `Sync` wrapper (not the raw `*mut`
// field) so the closure stays `Sync` under edition-2024 disjoint capture —
// the same guard the gatling engine's own `OutPtr` uses.
#[allow(clippy::redundant_locals)] // not redundant: it moves the WHOLE wrapper in (see above)
let back = back;
let src = &front[p * cells..p * cells + cells];
// SAFETY: pipe `p` writes only its own disjoint row `back[p*cells..]`; each
// `p` is claimed by exactly one gatling worker, so these writes never alias.
let dst = unsafe { std::slice::from_raw_parts_mut(back.0.add(p * cells), cells) };
advance_pipe(src, dst, vel[p], inject[p], t, dt);
});
// ZERO-COPY double-buffer swap: a pointer swap of the two Vecs, not a copy.
std::mem::swap(&mut self.front, &mut self.back);
}
/// The **serial reference** step — identical math, no gatling. The determinism test
/// asserts `step` (gatling, any worker order) equals `step_serial` bit-for-bit;
/// since pipes are independent this must hold, so a broken fan-out (aliasing writes,
/// a dropped pipe) diverges → RED. Also the honest 1-core fallback.
pub fn step_serial(&mut self, dt: f32) {
self.t += dt;
let cells = self.cells;
for p in 0..self.pipes {
// Copy the source row out is NOT needed — read `front`, write `back`.
let (src_lo, src_hi) = (p * cells, p * cells + cells);
// Split the borrow: read front row, write back row (different Vecs).
let src: &[f32] = &self.front[src_lo..src_hi];
let dst_ptr = self.back[src_lo..src_hi].as_mut_ptr();
// SAFETY: `dst` is a fresh unique slice into `back`, disjoint from `src`
// (which borrows `front`); no aliasing.
let dst = unsafe { std::slice::from_raw_parts_mut(dst_ptr, cells) };
advance_pipe(src, dst, self.vel[p], self.inject[p], self.t, dt);
}
std::mem::swap(&mut self.front, &mut self.back);
}
/// The observable surface (for `state_json` / a robot proof): sizes, time, the
/// per-pipe flow, and the total dye + a per-pipe dye sum so a driver can assert the
/// fluid is *present and moving* as data.
#[must_use]
pub fn state_json(&self) -> serde_json::Value {
let per_pipe: Vec<f32> =
(0..self.pipes).map(|p| self.dye(p).iter().copied().sum::<f32>()).collect();
serde_json::json!({
"pipes": self.pipes,
"cells": self.cells,
"time": self.t,
"flow": self.vel,
"total_dye": self.total_dye(),
"pipe_dye": per_pipe,
})
}
}
/// Advance one pipe's dye row by `dt`: upwind advection at signed speed `vel`
/// (cells/s), gentle diffusion, slight decay, and a pulsing inlet injection at the
/// upstream end. Pure — reads `src`, writes `dst` (same length); no allocation. The
/// per-cell op is identical regardless of which worker/order runs it, so the gatling
/// and serial steps agree bit-for-bit.
#[inline]
fn advance_pipe(src: &[f32], dst: &mut [f32], vel: f32, inject: f32, t: f32, dt: f32) {
let cells = src.len();
// CFL-limited Courant number (|c| < 1 keeps upwind stable at these dt).
let c = (vel * dt).clamp(-0.9, 0.9);
let ca = c.abs();
let diff = DIFFUSION * dt;
let keep = 1.0 - DECAY * dt;
for i in 0..cells {
let here = src[i];
// Upwind neighbour = the cell the flow is coming FROM.
let up = if c >= 0.0 {
if i > 0 { src[i - 1] } else { here }
} else if i + 1 < cells {
src[i + 1]
} else {
here
};
let adv = here - ca * (here - up);
// Symmetric diffusion (discrete Laplacian, clamped ends).
let left = if i > 0 { src[i - 1] } else { here };
let right = if i + 1 < cells { src[i + 1] } else { here };
let lap = left + right - 2.0 * here;
dst[i] = ((adv + diff * lap) * keep).clamp(0.0, DYE_MAX);
}
// Pulsing inlet at the upstream end (index 0 for downstream flow, else the last
// cell) — a breathing source so dye packets form and march along the pipe.
let pulse = inject * (0.5 + 0.5 * (t * PULSE_HZ * TAU).sin()).max(0.0);
if pulse > 0.0 {
let inlet = if vel >= 0.0 { 0 } else { cells - 1 };
dst[inlet] = (dst[inlet] + pulse).clamp(0.0, DYE_MAX);
}
}
#[cfg(test)]
mod tests {
use super::*;
fn seeded(pipes: usize, cells: usize) -> FluidField {
let mut f = FluidField::new(pipes, cells);
for p in 0..pipes {
f.set_flow(p, 5.0 + p as f32);
f.set_inject(p, 0.7);
}
f
}
// ── ZERO-COPY / ZERO-ALLOC guard (RED-when-broken) ──────────────────────────────
#[test]
fn step_is_zero_alloc_double_buffer_ping_pong() {
let mut f = seeded(9, 64);
let (pa, ca) = (f.front.as_ptr(), f.front.capacity());
let (pb, cb) = (f.back.as_ptr(), f.back.capacity());
// Many steps: the hot path must reuse the two buffers in place (no realloc, no
// clone/collect). A stray `self.front = something.collect()` flips a pointer.
for _ in 0..240 {
f.step(1.0 / 60.0);
}
// Even number of swaps ⇒ each buffer is back where it started; capacities pinned.
assert_eq!(f.front.as_ptr(), pa, "front buffer reused in place (no realloc/clone)");
assert_eq!(f.back.as_ptr(), pb, "back buffer reused in place (no realloc/clone)");
assert_eq!(f.front.capacity(), ca, "front capacity unchanged (no growth alloc)");
assert_eq!(f.back.capacity(), cb, "back capacity unchanged (no growth alloc)");
// And it really is a ping-pong swap: one step flips front↔back pointers.
let before = f.front.as_ptr();
f.step(1.0 / 60.0);
assert_eq!(f.front.as_ptr(), pb, "one step swaps front to the other buffer (double-buffer)");
assert_ne!(f.front.as_ptr(), before, "the swap actually moved buffers, not copied");
}
// ── the gatling fan-out equals the serial reference, bit-for-bit ────────────────
#[test]
fn gatling_step_matches_serial_step_bit_for_bit() {
let mut g = seeded(11, 48);
let mut s = seeded(11, 48);
for _ in 0..200 {
g.step(0.016);
s.step_serial(0.016);
}
// Independent pipes ⇒ worker order cannot matter; the fields must be identical.
assert_eq!(g.front, s.front, "gatling advection == serial advection (RED if the fan-out aliases/drops a pipe)");
assert!((g.total_dye() - s.total_dye()).abs() < 1e-6);
}
// ── the fluid actually MOVES (a packet advects downstream) ──────────────────────
#[test]
fn dye_advects_downstream_over_time() {
let mut f = FluidField::new(1, 32);
f.set_flow(0, 8.0); // downstream (index 0 → last)
f.set_inject(0, 1.0);
// Warm the inlet, then watch the centre of mass move toward the outlet.
for _ in 0..20 {
f.step(0.03);
}
let com = |f: &FluidField| {
let d = f.dye(0);
let (mut num, mut den) = (0.0f32, 0.0f32);
for (i, &v) in d.iter().enumerate() {
num += i as f32 * v;
den += v;
}
if den > 0.0 { num / den } else { 0.0 }
};
let c0 = com(&f);
for _ in 0..40 {
f.step(0.03);
}
let c1 = com(&f);
assert!(c1 > c0, "dye centre-of-mass advances downstream (c0={c0:.2} → c1={c1:.2})");
assert!(f.total_dye() > 0.0, "there is dye flowing");
}
// ── a later frame differs from an earlier one (it's animated, not frozen) ───────
#[test]
fn field_changes_frame_to_frame() {
let mut f = seeded(6, 40);
for _ in 0..10 {
f.step(0.02);
}
let a: Vec<f32> = f.dye(0).to_vec();
for _ in 0..10 {
f.step(0.02);
}
let b = f.dye(0);
assert_ne!(a.as_slice(), b, "the fluid keeps moving (a later frame differs)");
}
#[test]
fn state_json_reports_flow_and_dye() {
let mut f = seeded(4, 20);
for _ in 0..30 {
f.step(0.02);
}
let j = f.state_json();
assert_eq!(j["pipes"], 4);
assert_eq!(j["cells"], 20);
assert!(j["total_dye"].as_f64().unwrap() > 0.0, "dye is present");
assert_eq!(j["flow"].as_array().unwrap().len(), 4);
assert_eq!(j["pipe_dye"].as_array().unwrap().len(), 4);
}
#[test]
fn out_of_range_access_is_safe() {
let f = FluidField::new(2, 8);
assert!(f.dye(5).is_empty());
assert_eq!(f.flow(9), 0.0);
}
#[test]
fn empty_field_step_is_a_noop() {
let mut f = FluidField::new(0, 8);
f.step(0.016); // must not panic / index
assert_eq!(f.total_dye(), 0.0);
}
}