concinnity-engine 0.19.23

// src/gfx/streaming/texture.rs
//
// The `std`-side half of the asset-streaming subsystem.
//
// This owns the background payload-fetch thread and the channels that carry
// work to it, and wraps the `no_std` policy core in `crate::gfx::streaming`.
// The split is deliberate: `gfx::streaming::StreamPlanner` decides *what* to
// stream using only `core` + `alloc`; everything OS-coupled (threads,
// payload I/O) is confined here so a future `no_std` client runtime only has
// to replace this file.
//
// `PayloadSource` is the seam. `MemPayloadSource` serves compiled texture
// payloads already resident in RAM (used by `cn debug`, which builds payloads
// in memory); `DiskPayloadSource` re-reads them from their blob files on disk
// (used by `cn run`, so the bytes never stay RAM-resident). Both plug into the
// same planner and renderer.

use std::sync::Arc;
use std::sync::mpsc::{Receiver, Sender};

use super::{StreamPlanner, StreamState};
use crate::bake::texture::TextureImage;

// A texture payload decoded to a GPU-ready image (RGBA8 or block-compressed
// with its mip chain).
pub(crate) struct DecodedTexture {
    pub image: TextureImage,
}

// Fetches and decodes a streamable texture payload by item id.
//
// `Send + Sync` so the background worker thread can own one. Implementors do
// the slow part of streaming (disk read, decompression); the renderer only
// ever sees the finished [`DecodedTexture`].
pub(crate) trait PayloadSource: Send + Sync {
    // Decode item `id` into GPU-ready pixels, or return a human-readable
    // error. Called off the main thread.
    fn fetch(&self, id: usize) -> Result<DecodedTexture, String>;
}

// Payload source for the `cn debug` path: compiled texture payloads kept
// resident in RAM.
//
// `cn debug` builds payloads in memory with no blob files on disk, so it
// cannot use [`DiskPayloadSource`]; the bytes stay RAM-resident and this
// streams the *GPU upload* only. `cn run` uses [`DiskPayloadSource`] instead.
pub(crate) struct MemPayloadSource {
    payloads: Vec<Vec<u8>>,
}

impl MemPayloadSource {
    // `payloads[id]` is the compiled texture payload for streamable item `id`.
    pub(crate) fn new(payloads: Vec<Vec<u8>>) -> Self {
        Self { payloads }
    }
}

impl PayloadSource for MemPayloadSource {
    fn fetch(&self, id: usize) -> Result<DecodedTexture, String> {
        let bytes = self
            .payloads
            .get(id)
            .ok_or_else(|| format!("no payload for streamed texture {}", id))?;
        let image = crate::bake::texture::deserialise(bytes)?;
        Ok(DecodedTexture { image })
    }
}

// Locates one streamed texture payload inside a blob file on disk.
//
// `file_offset` is absolute into the file -- the payload-section start (past
// the blob header and defs) is already folded in by the caller, so the
// background worker only seeks and reads.
#[derive(Clone)]
pub(crate) struct DiskTextureLocator {
    pub path: String,
    pub(crate) file_offset: u64,
    pub len: u64,
}

// Disk-backed payload source: re-reads each compiled texture payload
// from its blob file on disk, so the bytes never stay RAM-resident.
//
// This is the counterpart to [`MemPayloadSource`] for the `cn run` path,
// where the world was loaded from blob files that are still on disk. The
// `cn debug` path builds payloads in memory with no blob files, so it must
// keep using [`MemPayloadSource`].
pub(crate) struct DiskPayloadSource {
    // locators[id] points streamed item `id` at its bytes in a blob file.
    locators: Vec<DiskTextureLocator>,
}

impl DiskPayloadSource {
    // `locators[id]` locates the compiled payload for streamable item `id`.
    pub(crate) fn new(locators: Vec<DiskTextureLocator>) -> Self {
        Self { locators }
    }
}

impl PayloadSource for DiskPayloadSource {
    fn fetch(&self, id: usize) -> Result<DecodedTexture, String> {
        let loc = self
            .locators
            .get(id)
            .ok_or_else(|| format!("no disk locator for streamed texture {}", id))?;
        let bytes = super::file_range::read_at(&loc.path, loc.file_offset, loc.len)?;
        let image = crate::bake::texture::deserialise(&bytes)?;
        Ok(DecodedTexture { image })
    }
}

// Outcome of one background load, carried back to the main thread.
struct LoadResult {
    id: usize,
    decoded: Result<DecodedTexture, String>,
}

// Drives streaming of one of the renderer's texture pools.
//
// One instance drives the albedo pool, a second the normal-map pool; the
// type is pool-agnostic -- the caller's `upload` callback routes a completed
// load to the right pool slot.
//
// Owns the [`StreamPlanner`] policy core plus the background fetch thread.
// Each frame the renderer calls [`update_scores`], [`plan_and_dispatch`], and
// [`drain_completed`] in that order.
//
// [`update_scores`]: TextureStreamer::update_scores
// [`plan_and_dispatch`]: TextureStreamer::plan_and_dispatch
// [`drain_completed`]: TextureStreamer::drain_completed
pub(crate) struct TextureStreamer {
    planner: StreamPlanner,
    // centers[id] holds the world-space positions of every draw object that
    // samples texture slot `id`; the streaming priority is the squared
    // distance from the camera to the nearest of them.
    centers: Vec<Vec<[f32; 3]>>,
    worker: super::worker::Worker<usize>,
    result_rx: Receiver<LoadResult>,
}

impl TextureStreamer {
    // Spawn the background worker and build a streamer for `centers.len()`
    // texture slots.
    //
    // `centers[id]` lists the draw-object positions that reference slot `id`.
    // `load_budget` caps loads dispatched per frame; `resident_cap` caps how
    // many textures stay resident at once before LRU eviction kicks in.
    pub(crate) fn new(
        source: Arc<dyn PayloadSource>,
        centers: Vec<Vec<[f32; 3]>>,
        load_budget: usize,
        resident_cap: usize,
    ) -> Self {
        let planner = StreamPlanner::new(centers.len(), load_budget, resident_cap);
        let (request_tx, request_rx) = std::sync::mpsc::channel::<usize>();
        let (result_tx, result_rx) = std::sync::mpsc::channel::<LoadResult>();

        let worker =
            super::worker::Worker::spawn("cn-texture-stream", request_rx, request_tx, move |rx| {
                worker_loop(source, rx, result_tx)
            });

        Self {
            planner,
            centers,
            result_rx,
            worker,
        }
    }

    // Number of streamed texture slots.
    pub(crate) fn len(&self) -> usize {
        self.planner.len()
    }

    // Set (or clear with `None`) the resident-byte budget for this pool. When
    // set, the planner evicts farther-from-camera textures to hold resident
    // bytes at or under the budget, on top of the item-count cap.
    pub(crate) fn set_byte_budget(&mut self, budget: Option<u64>) {
        self.planner.set_byte_budget(budget);
    }

    // Total resident texture bytes, for diagnostics.
    pub(crate) fn resident_bytes(&self) -> u64 {
        self.planner.resident_bytes()
    }

    // Block or unblock a slot for scene residency: a blocked slot never loads
    // and is evicted by the next plan if resident.
    pub(crate) fn set_blocked(&mut self, slot: usize, blocked: bool) {
        self.planner.set_blocked(slot, blocked);
    }

    // The active resident-byte budget, or `None` when byte accounting is off.
    pub(crate) fn byte_budget(&self) -> Option<u64> {
        self.planner.byte_budget()
    }

    // Re-score every slot from the camera position and refresh the LRU
    // timestamp of resident slots. Call once per frame before
    // [`plan_and_dispatch`](Self::plan_and_dispatch).
    pub(crate) fn update_scores(&mut self, camera: [f32; 3], frame: u64) {
        for id in 0..self.planner.len() {
            self.planner
                .set_score(id, nearest_sq_distance(&self.centers[id], camera));
            if self.planner.state(id) == Some(StreamState::Resident) {
                self.planner.touch(id, frame);
            }
        }
    }

    // Run the planner: dispatch this frame's loads to the worker and return
    // the slots the caller must evict from the GPU.
    pub(crate) fn plan_and_dispatch(&mut self) -> Vec<usize> {
        let plan = self.planner.plan();
        for &id in &plan.to_load {
            let sent = self.worker.send(id);
            if !sent {
                // Worker gone: revert so the slot is retried rather than
                // stuck Pending forever.
                self.planner.mark_unloaded(id);
            }
        }
        plan.to_evict
    }

    // Apply every completed background load via `upload`, which receives the
    // decoded image by value so it can carry it into a recorded backend op.
    // Returns the number of slots brought resident this call.
    pub(crate) fn drain_completed(
        &mut self,
        frame: u64,
        mut upload: impl FnMut(usize, TextureImage),
    ) -> usize {
        let mut applied = 0;
        while let Ok(result) = self.result_rx.try_recv() {
            match result.decoded {
                Ok(tex) => {
                    // Resident footprint is the sum of every mip level's bytes.
                    let bytes = tex.image.byte_len() as u64;
                    upload(result.id, tex.image);
                    self.planner.mark_resident(result.id, frame, bytes);
                    applied += 1;
                }
                Err(e) => {
                    tracing::warn!("texture stream: load of slot {} failed: {}", result.id, e);
                    // Treat a failed fetch as terminally resident so the
                    // planner stops retrying; the slot keeps its placeholder,
                    // which occupies no streamed bytes.
                    self.planner.mark_resident(result.id, frame, 0);
                }
            }
        }
        applied
    }

    // `(resident, pending, unloaded)` slot counts, for diagnostics.
    pub(crate) fn stats(&self) -> (usize, usize, usize) {
        self.planner.counts()
    }
}

// Background worker: fetch each requested payload and ship the result back.
// Exits when the request channel closes (the streamer was dropped).
fn worker_loop(
    source: Arc<dyn PayloadSource>,
    requests: Receiver<usize>,
    results: Sender<LoadResult>,
) {
    while let Ok(id) = requests.recv() {
        let decoded = source.fetch(id);
        if results.send(LoadResult { id, decoded }).is_err() {
            break;
        }
    }
}

// Squared distance from `camera` to the nearest position in `centers`.
//
// Squared (not true) distance keeps the math `sqrt`-free: ordering is all
// the planner needs. An empty `centers` (a texture referenced by no draw)
// scores 0 so it still streams in promptly rather than stalling forever.
fn nearest_sq_distance(centers: &[[f32; 3]], camera: [f32; 3]) -> f32 {
    let mut nearest = f32::MAX;
    for c in centers {
        let dx = c[0] - camera[0];
        let dy = c[1] - camera[1];
        let dz = c[2] - camera[2];
        let d = dx * dx + dy * dy + dz * dz;
        if d < nearest {
            nearest = d;
        }
    }
    if centers.is_empty() { 0.0 } else { nearest }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn nearest_sq_distance_picks_the_closest_center() {
        let centers = [[10.0, 0.0, 0.0], [3.0, 0.0, 0.0], [7.0, 0.0, 0.0]];
        // Closest center is at x=3, so squared distance from origin is 9.
        assert_eq!(nearest_sq_distance(&centers, [0.0, 0.0, 0.0]), 9.0);
    }

    #[test]
    fn nearest_sq_distance_of_no_centers_is_zero() {
        assert_eq!(nearest_sq_distance(&[], [5.0, 5.0, 5.0]), 0.0);
    }

    // Build a minimal compiled RGBA8 texture payload via the shared serialiser.
    fn make_payload(w: u32, h: u32, fill: u8) -> Vec<u8> {
        let pixels = std::iter::repeat_n(fill, (w * h * 4) as usize).collect();
        crate::bake::texture::serialise(&TextureImage::rgba8(w, h, pixels))
    }

    #[test]
    fn mem_payload_source_decodes_a_payload() {
        let source = MemPayloadSource::new(vec![make_payload(2, 1, 0xAB)]);
        let tex = source.fetch(0).expect("fetch ok");
        assert_eq!((tex.image.width(), tex.image.height()), (2, 1));
        assert_eq!(tex.image.mips[0].data.len(), 2 * 4);
        assert!(tex.image.mips[0].data.iter().all(|&b| b == 0xAB));
    }

    #[test]
    fn mem_payload_source_errors_on_unknown_id() {
        let source = MemPayloadSource::new(vec![make_payload(1, 1, 0)]);
        assert!(source.fetch(9).is_err());
    }

    #[test]
    fn disk_payload_source_reads_a_payload_at_offset() {
        let tree = concinnity_testing::TempTree::new();
        let payload = make_payload(2, 1, 0xCD);
        // arbitrary leading bytes standing in for a blob header + defs section
        let prefix = vec![0u8; 37];
        let mut bytes = prefix.clone();
        bytes.extend_from_slice(&payload);
        let path = tree.write("payload.bin", &bytes);
        let source = DiskPayloadSource::new(vec![DiskTextureLocator {
            path: path.to_string_lossy().into_owned(),
            file_offset: prefix.len() as u64,
            len: payload.len() as u64,
        }]);
        let tex = source.fetch(0).expect("fetch ok");
        assert_eq!((tex.image.width(), tex.image.height()), (2, 1));
        assert_eq!(tex.image.mips[0].data.len(), 2 * 4);
        assert!(tex.image.mips[0].data.iter().all(|&b| b == 0xCD));
        let _ = std::fs::remove_file(&path);
    }

    #[test]
    fn disk_payload_source_errors_on_unknown_id() {
        let source = DiskPayloadSource::new(vec![]);
        assert!(source.fetch(0).is_err());
    }

    #[test]
    fn disk_payload_source_errors_on_missing_file() {
        let source = DiskPayloadSource::new(vec![DiskTextureLocator {
            path: "/nonexistent/cn_disk_payload_missing.bin".to_string(),
            file_offset: 0,
            len: 4,
        }]);
        assert!(source.fetch(0).is_err());
    }

    // A source that yields a fixed 1x1 texture for any id, used to exercise
    // the worker thread without the build pipeline.
    struct ConstSource;
    impl PayloadSource for ConstSource {
        fn fetch(&self, _id: usize) -> Result<DecodedTexture, String> {
            Ok(DecodedTexture {
                image: TextureImage::rgba8(1, 1, vec![1, 2, 3, 4]),
            })
        }
    }

    // Pump drain_completed until `want` slots are resident or a deadline hits.
    fn drain_until(streamer: &mut TextureStreamer, frame: u64, want: usize) -> usize {
        let deadline = std::time::Instant::now() + std::time::Duration::from_secs(2);
        let mut uploads = 0;
        while std::time::Instant::now() < deadline {
            uploads += streamer.drain_completed(frame, |_, _| {});
            if streamer.stats().0 >= want {
                break;
            }
            std::thread::sleep(std::time::Duration::from_millis(1));
        }
        uploads
    }

    #[test]
    fn streamer_loads_nearest_slots_within_budget() {
        let centers = vec![
            vec![[100.0, 0.0, 0.0]], // slot 0: far
            vec![[2.0, 0.0, 0.0]],   // slot 1: near
            vec![[50.0, 0.0, 0.0]],  // slot 2: mid
        ];
        // Budget 1/frame, generous cap.
        let mut streamer = TextureStreamer::new(Arc::new(ConstSource), centers, 1, 8);
        assert_eq!(streamer.len(), 3);

        // Frame 1: nearest slot (1) is dispatched first.
        streamer.update_scores([0.0, 0.0, 0.0], 1);
        let evict = streamer.plan_and_dispatch();
        assert!(evict.is_empty());
        drain_until(&mut streamer, 1, 1);
        assert_eq!(streamer.stats().0, 1);

        // Frame 2: next-nearest (slot 2) follows.
        streamer.update_scores([0.0, 0.0, 0.0], 2);
        streamer.plan_and_dispatch();
        drain_until(&mut streamer, 2, 2);
        assert_eq!(streamer.stats().0, 2);

        // Frame 3: the far slot finishes the set.
        streamer.update_scores([0.0, 0.0, 0.0], 3);
        streamer.plan_and_dispatch();
        drain_until(&mut streamer, 3, 3);
        assert_eq!(streamer.stats(), (3, 0, 0));
    }

    // A payload that will never decode.
    struct FailingSource;
    impl PayloadSource for FailingSource {
        fn fetch(&self, _id: usize) -> Result<DecodedTexture, String> {
            Err("undecodable payload".to_string())
        }
    }

    // A failed fetch is terminal, not transient: the slot is marked resident
    // (keeping its placeholder, occupying no streamed bytes) so the planner
    // stops re-dispatching a payload that will never decode.
    #[test]
    fn a_failed_fetch_is_not_retried() {
        let centers = vec![vec![[1.0, 0.0, 0.0]]];
        let mut streamer = TextureStreamer::new(Arc::new(FailingSource), centers, 4, 8);
        streamer.update_scores([0.0, 0.0, 0.0], 1);
        streamer.plan_and_dispatch();

        let uploads = drain_until(&mut streamer, 1, 1);
        assert_eq!(uploads, 0, "a failed load uploads no pixels");
        assert_eq!(streamer.stats(), (1, 0, 0));
        assert_eq!(streamer.resident_bytes(), 0);
    }

    #[test]
    fn upload_callback_receives_decoded_pixels() {
        let centers = vec![vec![[1.0, 0.0, 0.0]]];
        let mut streamer = TextureStreamer::new(Arc::new(ConstSource), centers, 4, 8);
        streamer.update_scores([0.0, 0.0, 0.0], 1);
        streamer.plan_and_dispatch();

        let mut seen: Option<(usize, u32, u32, Vec<u8>)> = None;
        let deadline = std::time::Instant::now() + std::time::Duration::from_secs(2);
        while std::time::Instant::now() < deadline && seen.is_none() {
            streamer.drain_completed(1, |id, image| {
                seen = Some((
                    id,
                    image.width(),
                    image.height(),
                    image.mips[0].data.clone(),
                ));
            });
            std::thread::sleep(std::time::Duration::from_millis(1));
        }
        assert_eq!(seen, Some((0, 1, 1, vec![1, 2, 3, 4])));
    }
}