From a5af42f119030acdc4befbc7b1fd7fedbd2c5ba8 Mon Sep 17 00:00:00 2001 From: Serkyo Date: Sun, 26 Jul 2026 20:19:59 +0200 Subject: [PATCH] feat(client): mesh chunks on a background worker pool --- crates/client/src/chunks.rs | 202 +++++++++++++++++++----------- crates/client/src/main.rs | 1 + crates/client/src/mesh_pool.rs | 150 ++++++++++++++++++++++ crates/client/src/tests/chunks.rs | 154 +++++++++++++++++++++++ 4 files changed, 436 insertions(+), 71 deletions(-) create mode 100644 crates/client/src/mesh_pool.rs diff --git a/crates/client/src/chunks.rs b/crates/client/src/chunks.rs index f484112..1d7405b 100644 --- a/crates/client/src/chunks.rs +++ b/crates/client/src/chunks.rs @@ -5,13 +5,14 @@ use std::collections::{HashMap, HashSet}; use std::sync::Arc; -use renderer::meshing::Neighbors; use renderer::vertex::Vertex; use renderer::{MeshKey, RendererError}; use shared::protocol::chunk::ChunkMessage; use shared::world::{CHUNK_SIZE, Chunk, ChunkPos}; use tracing::{debug, error}; +use crate::mesh_pool::{JobGen, MeshJob, MeshPool, MeshResult}; + /// Radius, in chunks, of the region kept resident around the camera center. Also the radius the client subscribes with, so the server's resident set matches the client's. // TODO: make configurable / drive from view-distance setting. pub const LOAD_RADIUS: i32 = 8; @@ -19,11 +20,10 @@ pub const LOAD_RADIUS: i32 = 8; /// Maximum number of chunk deliveries materialized in a single call to [`ChunkManager::update`], bounding per-frame materialization work. Deliveries beyond the budget remain queued in the transport for the next frame. const LOADS_PER_UPDATE: usize = 4; -/// Maximum number of chunks (re)meshed and uploaded per call to [`ChunkManager::update`], draining the pending re-mesh set under a bound so the winit loop stays responsive. One delivery can enqueue up to seven mesh jobs (itself plus six neighbours), so this budget exceeds [`LOADS_PER_UPDATE`]. -// TODO: move meshing to a worker pool; until then this budget caps meshing work on the winit thread. +/// Maximum number of mesh jobs dispatched to the worker pool per call to [`ChunkManager::update`], draining the pending re-mesh set under a bound so a burst of deliveries does not flood the pool in a single frame. One delivery can enqueue up to seven mesh jobs (itself plus six neighbours), so this budget exceeds [`LOADS_PER_UPDATE`]. Finished meshes are ingested without a per-frame bound, since uploading already-computed geometry is cheap relative to generating it. const MESHES_PER_UPDATE: usize = 16; -/// The six face-adjacent neighbour offsets, in chunk coordinates. +/// The six face-adjacent neighbour offsets, in chunk coordinates. The order matches the neighbour array carried by [`MeshJob`]: `[+X, -X, +Y, -Y, +Z, -Z]`. const NEIGHBOR_OFFSETS: [(i32, i32, i32); 6] = [ (1, 0, 0), (-1, 0, 0), @@ -35,7 +35,7 @@ const NEIGHBOR_OFFSETS: [(i32, i32, i32); 6] = [ /// Sink that receives finished chunk meshes for upload. /// -/// The production sink is the Vulkan [`Renderer`](renderer::Renderer); the abstraction exists so the meshing pipeline can be exercised against a recording double in tests, which have no GPU. Method signatures mirror the renderer's exactly so the production `impl` is a direct forward. +/// The production sink is the Vulkan [`Renderer`](renderer::Renderer); the abstraction exists so the ingest pipeline can be exercised against a recording double in tests, which have no GPU. Method signatures mirror the renderer's exactly so the production `impl` is a direct forward. pub trait MeshSink { /// Uploads (or replaces) the mesh identified by `key`. /// @@ -70,29 +70,38 @@ impl MeshSink for renderer::Renderer { } } -/// Tracks which server-streamed chunks are resident and orchestrates neighbour-aware meshing. +/// Tracks which server-streamed chunks are resident and orchestrates neighbour-aware background meshing. pub struct ChunkManager { - /// Resident chunks keyed by position, retained so the mesher can sample voxels across chunk boundaries. Stored behind [`Arc`] so a future worker pool can hand a chunk to a thread without copying it. + /// Resident chunks keyed by position, retained so the mesher can sample voxels across chunk boundaries. Stored behind [`Arc`] so a chunk can be handed to a worker thread without copying its 64 KiB volume. // TODO: a resident Chunk is 32³ × 2 bytes = 64 KiB; at LOAD_RADIUS = 8 the resident set is thousands of chunks (hundreds of MiB). A follow-up can store only the six 32×32 boundary planes per chunk instead of the full volume. resident: HashMap>, - /// Positions whose mesh must be rebuilt, accumulated across frames and drained under [`MESHES_PER_UPDATE`]. Held as a set so a burst of deliveries re-meshes each affected neighbour at most once per frame. + /// Positions whose mesh must be rebuilt, accumulated across frames and dispatched under [`MESHES_PER_UPDATE`]. Held as a set so a burst of deliveries re-meshes each affected neighbour at most once. pending_remesh: HashSet, + /// Positions with a mesh job currently outstanding, mapped to the generation of that job. A returned mesh is applied only when its generation still matches, so meshes superseded by a re-dispatch (or by eviction) are discarded rather than uploaded stale. + in_flight: HashMap, + /// The generation stamped on the next dispatched job. Global and strictly increasing across all positions, so no two dispatches ever share a token; see [`JobGen`]. + next_gen: JobGen, + /// Background worker pool that turns chunks into CPU geometry off the winit thread. + pool: MeshPool, /// Reused all-air baseline that server [`ChunkData`](shared::world::ChunkData) diffs are materialized against. baseline: Chunk, } impl ChunkManager { - /// Creates a manager with no chunks yet resident. + /// Creates a manager with no chunks yet resident, spawning the background mesh worker pool. #[must_use] pub fn new() -> Self { Self { resident: HashMap::new(), pending_remesh: HashSet::new(), + in_flight: HashMap::new(), + next_gen: JobGen::FIRST, + pool: MeshPool::new(), baseline: Chunk::default(), } } - /// Reconciles the resident chunk set: evicts chunks outside the load radius around `center`, applies queued server deliveries under a materialization budget, then drains the pending re-mesh set under a meshing budget. + /// Advances the streaming pipeline for one frame: ingests finished meshes from the pool, evicts chunks outside the load radius around `center`, applies queued server deliveries under a materialization budget, then dispatches pending re-mesh jobs under a dispatch budget. /// /// The client's own radius eviction runs independently of the server's authoritative `Drop`, so memory stays bounded even if the server is slow to drop chunks that leave the region. pub fn update( @@ -101,26 +110,57 @@ impl ChunkManager { deliveries: &mut net::ChunkStream, sink: &mut impl MeshSink, ) { + let applied = self.drain_results(sink); let unloaded = self.unload_outside(center, sink); let (loaded, dropped) = self.apply_deliveries(deliveries, sink); - let meshed = self.drain_remesh(sink); + let dispatched = self.dispatch_pending(); - if loaded > 0 || dropped > 0 || unloaded > 0 || meshed > 0 { + if loaded > 0 || dropped > 0 || unloaded > 0 || applied > 0 || dispatched > 0 { debug!( loaded, dropped, unloaded, - meshed, + dispatched, + applied, pending = self.pending_remesh.len(), + in_flight = self.in_flight.len(), resident = self.resident.len(), "chunk stream reconciled" ); } } + /// Ingests every finished mesh currently available from the pool, uploading the ones that are still current and discarding superseded or evicted ones. Returns the number uploaded. + fn drain_results(&mut self, sink: &mut impl MeshSink) -> usize { + let mut applied = 0; + while let Some(result) = self.pool.poll() { + if self.apply_result(&result, sink) { + applied += 1; + } + } + applied + } + + /// Uploads a single finished mesh when it is still current, returning whether it was uploaded. + /// + /// A result is current when its position is still resident and its generation matches the latest job dispatched for that position (see [`should_apply`]). On a match the in-flight entry is cleared; otherwise the result is dropped and any newer outstanding job for the position is left untouched. + fn apply_result(&mut self, result: &MeshResult, sink: &mut impl MeshSink) -> bool { + if !should_apply( + result.pos, + result.generation, + |pos| self.resident.contains_key(&pos), + &self.in_flight, + ) { + return false; + } + self.in_flight.remove(&result.pos); + upload_result(result, sink); + true + } + /// Applies up to [`LOADS_PER_UPDATE`] chunk deliveries plus any interleaved drops, returning the counts of chunks loaded and dropped. /// - /// Delivered chunks are materialized and retained; drops remove the chunk from residency and from the renderer. Both kinds enqueue the affected neighbourhood for re-meshing. + /// Delivered chunks are materialized and retained; drops remove the chunk from residency, from any in-flight tracking, and from the renderer. Both kinds enqueue the affected neighbourhood for re-meshing. fn apply_deliveries( &mut self, deliveries: &mut net::ChunkStream, @@ -138,6 +178,7 @@ impl ChunkManager { } Ok(ChunkMessage::Drop { pos }) => { if self.resident.remove(&pos).is_some() { + self.in_flight.remove(&pos); sink.remove_mesh((pos.x, pos.y, pos.z)); dropped.push(pos); } @@ -153,7 +194,7 @@ impl ChunkManager { /// Evicts every resident chunk outside the load radius around `center`, returning the number removed. /// - /// Each evicted chunk's resident neighbours have a boundary toward it that is now exposed, so they are enqueued for re-meshing. + /// Each evicted chunk is removed from residency, from in-flight tracking, and from the renderer; its resident neighbours have a boundary toward it that is now exposed, so they are enqueued for re-meshing. fn unload_outside(&mut self, center: ChunkPos, sink: &mut impl MeshSink) -> usize { let desired = desired_chunks(center, LOAD_RADIUS); let stale: Vec = self @@ -165,6 +206,7 @@ impl ChunkManager { for pos in &stale { sink.remove_mesh((pos.x, pos.y, pos.z)); self.resident.remove(pos); + self.in_flight.remove(pos); } self.queue_remesh(&[], &stale); stale.len() @@ -176,10 +218,10 @@ impl ChunkManager { self.pending_remesh.extend(targets); } - /// Meshes and uploads up to [`MESHES_PER_UPDATE`] chunks from the pending set, returning the number processed. + /// Dispatches up to [`MESHES_PER_UPDATE`] pending re-mesh jobs to the worker pool, returning the number dispatched. /// - /// Positions no longer resident (dropped after being enqueued) are discarded without meshing. - fn drain_remesh(&mut self, sink: &mut impl MeshSink) -> usize { + /// Each dispatched position snapshots its chunk and current resident neighbours behind [`Arc`]s, is stamped with a fresh generation, and is recorded as in-flight (superseding any previous outstanding job for it). Positions no longer resident (dropped after being enqueued) are skipped without dispatch. + fn dispatch_pending(&mut self) -> usize { // Take a bounded batch out of the set; the remainder stays queued for later frames. let batch: Vec = self .pending_remesh @@ -188,66 +230,41 @@ impl ChunkManager { .copied() .collect(); - let mut meshed = 0; + let mut dispatched = 0; for pos in batch { self.pending_remesh.remove(&pos); - if self.resident.contains_key(&pos) { - self.mesh_and_upload(pos, sink); - meshed += 1; - } + let Some(chunk) = self.resident.get(&pos) else { + continue; + }; + let chunk = Arc::clone(chunk); + let neighbors = self.neighbor_arcs(pos); + let generation = self.bump_gen(); + self.in_flight.insert(pos, generation); + self.pool.dispatch(MeshJob { + pos, + generation, + chunk, + neighbors, + }); + dispatched += 1; } - meshed + dispatched } - /// Meshes the resident chunk at `pos` against its resident neighbours and uploads the result to the sink. - /// - /// A chunk that meshes to no geometry (all air, or fully enclosed by solid neighbours) is removed from the sink rather than uploaded, since a zero-length buffer is invalid; this also clears any mesh a previous state had left there. - fn mesh_and_upload(&mut self, pos: ChunkPos, sink: &mut impl MeshSink) { - let Some(chunk) = self.resident.get(&pos) else { - return; - }; - let neighbors = self.neighbors_of(pos); - let (vertices, indices) = renderer::meshing::generate_mesh(chunk, &neighbors); - - if indices.is_empty() { - sink.remove_mesh((pos.x, pos.y, pos.z)); - return; - } - - // Chunk coordinates and CHUNK_SIZE are small and represent exactly as f32. - #[expect( - clippy::cast_precision_loss, - reason = "chunk coordinates stay well within f32's exact-integer range" - )] - let world_offset = { - let size = CHUNK_SIZE as f32; - [ - pos.x as f32 * size, - pos.y as f32 * size, - pos.z as f32 * size, - ] - }; - - if let Err(e) = sink.insert_mesh((pos.x, pos.y, pos.z), &vertices, &indices, world_offset) { - error!(?pos, "failed to upload chunk mesh: {e}"); - } - } - - /// Gathers the six face-adjacent resident chunks of `pos` into a [`Neighbors`] set for meshing. - fn neighbors_of(&self, pos: ChunkPos) -> Neighbors<'_> { - let get = |dx, dy, dz| { + /// Snapshots the six face-adjacent resident chunks of `pos` as [`Arc`] handles, ordered to match [`NEIGHBOR_OFFSETS`]. Absent neighbours are `None`. + fn neighbor_arcs(&self, pos: ChunkPos) -> [Option>; 6] { + NEIGHBOR_OFFSETS.map(|(dx, dy, dz)| { self.resident .get(&ChunkPos::new(pos.x + dx, pos.y + dy, pos.z + dz)) - .map(|chunk| &**chunk) - }; - Neighbors { - pos_x: get(1, 0, 0), - neg_x: get(-1, 0, 0), - pos_y: get(0, 1, 0), - neg_y: get(0, -1, 0), - pos_z: get(0, 0, 1), - neg_z: get(0, 0, -1), - } + .map(Arc::clone) + }) + } + + /// Returns a fresh, never-before-used generation and advances the counter. + fn bump_gen(&mut self) -> JobGen { + let current = self.next_gen; + self.next_gen = self.next_gen.next(); + current } } @@ -257,6 +274,49 @@ impl Default for ChunkManager { } } +/// Reports whether a finished mesh should be uploaded. +/// +/// A mesh is current, and therefore applied, only when its position is still wanted (resident) and the generation recorded as in-flight for that position still equals the mesh's own generation. A missing in-flight entry (the position was evicted) or a mismatched generation (a newer job superseded this one) both mean the result is stale and must be discarded. +fn should_apply( + pos: ChunkPos, + generation: JobGen, + is_wanted: impl Fn(ChunkPos) -> bool, + in_flight: &HashMap, +) -> bool { + is_wanted(pos) && in_flight.get(&pos) == Some(&generation) +} + +/// Uploads a finished mesh to the sink, or clears the slot when the mesh is empty. +/// +/// A chunk that meshes to no geometry (all air, or fully enclosed by solid neighbours) is removed from the sink rather than uploaded, since a zero-length buffer is invalid; this also clears any mesh a previous state had left there. +fn upload_result(result: &MeshResult, sink: &mut impl MeshSink) { + let pos = result.pos; + let key = (pos.x, pos.y, pos.z); + + if result.indices.is_empty() { + sink.remove_mesh(key); + return; + } + + // Chunk coordinates and CHUNK_SIZE are small and represent exactly as f32. + #[expect( + clippy::cast_precision_loss, + reason = "chunk coordinates stay well within f32's exact-integer range" + )] + let world_offset = { + let size = CHUNK_SIZE as f32; + [ + pos.x as f32 * size, + pos.y as f32 * size, + pos.z as f32 * size, + ] + }; + + if let Err(e) = sink.insert_mesh(key, &result.vertices, &result.indices, world_offset) { + error!(?pos, "failed to upload chunk mesh: {e}"); + } +} + /// Returns the six face-adjacent neighbour positions of `pos`. fn neighbor_positions(pos: ChunkPos) -> [ChunkPos; 6] { NEIGHBOR_OFFSETS.map(|(dx, dy, dz)| ChunkPos::new(pos.x + dx, pos.y + dy, pos.z + dz)) diff --git a/crates/client/src/main.rs b/crates/client/src/main.rs index f5a816e..941fb3d 100644 --- a/crates/client/src/main.rs +++ b/crates/client/src/main.rs @@ -6,6 +6,7 @@ mod camera; mod chunks; +mod mesh_pool; use std::time::Instant; diff --git a/crates/client/src/mesh_pool.rs b/crates/client/src/mesh_pool.rs new file mode 100644 index 0000000..cb8643e --- /dev/null +++ b/crates/client/src/mesh_pool.rs @@ -0,0 +1,150 @@ +// SPDX-License-Identifier: AGPL-3.0-only + +//! Background worker pool that meshes chunks off the winit thread. + +use std::num::NonZero; +use std::sync::Arc; +use std::thread::JoinHandle; + +use crossbeam_channel::{Receiver, Sender}; +use renderer::meshing::{Neighbors, generate_mesh}; +use renderer::vertex::Vertex; +use shared::world::{Chunk, ChunkPos}; + +/// Monotonic staleness token stamped on every dispatched [`MeshJob`]. +/// +/// Between dispatching a job for a position and the worker returning it, that position may have been evicted or re-dispatched with fresher neighbours (a neighbour loaded or dropped). A returned mesh is applied only when its generation still matches the latest generation recorded for the position; older generations are superseded and discarded. +#[derive(Copy, Clone, Debug, PartialEq, Eq, Hash)] +pub(crate) struct JobGen(u64); + +impl JobGen { + /// The generation of the first job ever dispatched. + pub(crate) const FIRST: Self = Self(0); + + /// Returns the next generation after `self`. + /// + /// Wraps on overflow rather than panicking; wrap-around requires 2^64 dispatches in one session, at which point a collision would additionally require the wrapped-to job to still be outstanding, which is unreachable in practice. + pub(crate) fn next(self) -> Self { + Self(self.0.wrapping_add(1)) + } +} + +/// A unit of meshing work handed to a worker: an owned snapshot so the worker borrows nothing from the manager. +/// +/// The chunk and its neighbours are carried as [`Arc`] handles so dispatch is a cheap refcount bump rather than a copy of the 64 KiB voxel volume. Meshing is neighbour-dependent (boundary faces are culled against adjacent chunks), so the six face-adjacent neighbours are snapshotted at dispatch time; a `None` entry means that neighbour is not resident and the boundary is treated as exposed. +pub(crate) struct MeshJob { + /// Chunk-space position of the chunk to mesh. + pub(crate) pos: ChunkPos, + /// Staleness token identifying this dispatch; echoed back on the result. + pub(crate) generation: JobGen, + /// The chunk to mesh. + pub(crate) chunk: Arc, + /// The six face-adjacent neighbours, ordered `[+X, -X, +Y, -Y, +Z, -Z]` to match `chunks::NEIGHBOR_OFFSETS`. `None` marks an absent neighbour. + pub(crate) neighbors: [Option>; 6], +} + +/// A finished mesh returned from a worker to the main thread for upload. +pub(crate) struct MeshResult { + /// Chunk-space position the mesh belongs to. + pub(crate) pos: ChunkPos, + /// Generated vertices; empty when the chunk meshes to no geometry. + pub(crate) vertices: Vec, + /// Generated triangle indices; empty when the chunk meshes to no geometry. + pub(crate) indices: Vec, + /// The generation stamped on the originating [`MeshJob`], used to discard superseded results. + pub(crate) generation: JobGen, +} + +/// A pool of worker threads that mesh chunks and return CPU geometry. +pub(crate) struct MeshPool { + /// Sending end of the job queue; the main thread pushes [`MeshJob`]s. + job_tx: Sender, + /// Receiving end of the result queue; the main thread drains finished meshes. + result_rx: Receiver, + /// Handles to the worker threads, retained for a future graceful-stop path that drops `job_tx` and joins them; the process currently relies on OS teardown at exit. + #[expect( + dead_code, + reason = "retained for a future graceful-shutdown join path, mirroring the server pool" + )] + workers: Vec>, +} + +impl MeshPool { + /// Spawns the worker pool, sizing it to leave one logical core for the main thread. + pub(crate) fn new() -> Self { + let (job_tx, job_rx) = crossbeam_channel::unbounded::(); + let (result_tx, result_rx) = crossbeam_channel::unbounded::(); + + // One worker per logical core, less one to keep the winit thread responsive, but never fewer than one. + let cores = std::thread::available_parallelism().map_or(4, NonZero::get); + let worker_count = cores.saturating_sub(1).max(1); + + let workers = (0..worker_count) + .map(|_| { + // Each worker owns its own clone of the shared job queue and of the sender back into the result queue. + let job_rx = job_rx.clone(); + let result_tx = result_tx.clone(); + + std::thread::spawn(move || { + // Block until a job arrives; a blocking recv is fine off the main thread. + while let Ok(job) = job_rx.recv() { + let (vertices, indices) = mesh_job(&job); + let result = MeshResult { + pos: job.pos, + vertices, + indices, + generation: job.generation, + }; + // A send error means the main thread has gone away; the worker winds down. + if result_tx.send(result).is_err() { + break; + } + } + }) + }) + .collect(); + + // Drop the template ends left over after cloning so the channels close once the real holders are gone: workers observe job-channel shutdown, and the main thread observes result-channel shutdown. + drop(job_rx); + drop(result_tx); + + Self { + job_tx, + result_rx, + workers, + } + } + + /// Enqueues a meshing job for the pool. + /// + /// A send error (the workers have shut down) is ignored: there is nothing useful to do with the job, and shutdown only happens at process teardown. + pub(crate) fn dispatch(&self, job: MeshJob) { + let _ = self.job_tx.send(job); + } + + /// Returns the next finished mesh without blocking, or `None` when none is ready. + pub(crate) fn poll(&self) -> Option { + self.result_rx.try_recv().ok() + } +} + +impl Default for MeshPool { + fn default() -> Self { + Self::new() + } +} + +/// Reconstructs a borrowed [`Neighbors`] view from a job's owned neighbour [`Arc`]s and meshes the chunk. +/// +/// The [`Neighbors`] view borrows `&Chunk` out of the job's `Arc`s, so it is built and consumed here in one scope while those `Arc`s are still alive. +fn mesh_job(job: &MeshJob) -> (Vec, Vec) { + let neighbors = Neighbors { + pos_x: job.neighbors[0].as_deref(), + neg_x: job.neighbors[1].as_deref(), + pos_y: job.neighbors[2].as_deref(), + neg_y: job.neighbors[3].as_deref(), + pos_z: job.neighbors[4].as_deref(), + neg_z: job.neighbors[5].as_deref(), + }; + generate_mesh(&job.chunk, &neighbors) +} diff --git a/crates/client/src/tests/chunks.rs b/crates/client/src/tests/chunks.rs index 5895e96..d65ba8c 100644 --- a/crates/client/src/tests/chunks.rs +++ b/crates/client/src/tests/chunks.rs @@ -2,6 +2,10 @@ //! Unit tests for the chunk streaming logic in [`crate::chunks`]. +use std::time::{Duration, Instant}; + +use shared::world::BlockId; + use super::*; #[test] @@ -74,3 +78,153 @@ fn remesh_targets_are_deduplicated() { let targets = remesh_targets(&[a, b], &[], resident_in(&resident)); assert_eq!(targets, resident.into_iter().collect()); } + +// --- Staleness decision (`should_apply`) --------------------------------- + +#[test] +fn should_apply_accepts_current_result() { + let pos = ChunkPos::new(1, 2, 3); + let generation = JobGen::FIRST; + let mut in_flight = HashMap::new(); + in_flight.insert(pos, generation); + // Resident and generation matches the outstanding job: apply. + assert!(should_apply(pos, generation, |_| true, &in_flight)); +} + +#[test] +fn should_apply_rejects_stale_generation() { + let pos = ChunkPos::new(0, 0, 0); + let mut in_flight = HashMap::new(); + // A newer job (next generation) is outstanding for the position. + in_flight.insert(pos, JobGen::FIRST.next()); + // The result carries the older generation and must be discarded. + assert!(!should_apply(pos, JobGen::FIRST, |_| true, &in_flight)); +} + +#[test] +fn should_apply_rejects_unwanted_position() { + let pos = ChunkPos::new(0, 0, 0); + let generation = JobGen::FIRST; + let mut in_flight = HashMap::new(); + in_flight.insert(pos, generation); + // The position is no longer resident even though a job is tracked. + assert!(!should_apply(pos, generation, |_| false, &in_flight)); +} + +#[test] +fn should_apply_rejects_missing_in_flight() { + let pos = ChunkPos::new(0, 0, 0); + // No job is tracked for the position (it was evicted after dispatch). + let in_flight = HashMap::new(); + assert!(!should_apply(pos, JobGen::FIRST, |_| true, &in_flight)); +} + +// --- Ingest pipeline plumbing -------------------------------------------- + +/// Recording [`MeshSink`] double capturing the keys passed to it, so ingest can be exercised without a GPU. +#[derive(Default)] +struct RecordingSink { + /// Keys uploaded via [`MeshSink::insert_mesh`], in call order. + inserted: Vec, + /// Keys cleared via [`MeshSink::remove_mesh`], in call order. + removed: Vec, +} + +impl MeshSink for RecordingSink { + fn insert_mesh( + &mut self, + key: MeshKey, + _vertices: &[Vertex], + _indices: &[u32], + _world_offset: [f32; 3], + ) -> Result<(), RendererError> { + self.inserted.push(key); + Ok(()) + } + + fn remove_mesh(&mut self, key: MeshKey) { + self.removed.push(key); + } +} + +/// Builds a chunk with a single solid block so it meshes to non-empty geometry. +fn solid_chunk() -> Chunk { + let mut chunk = Chunk::default(); + chunk.set(0, 0, 0, BlockId(1)); + chunk +} + +/// Blocks until the pool yields a finished mesh, panicking if none arrives within a generous deadline. +fn wait_for_result(pool: &MeshPool) -> MeshResult { + let deadline = Instant::now() + Duration::from_secs(5); + loop { + if let Some(result) = pool.poll() { + return result; + } + assert!( + Instant::now() < deadline, + "worker pool did not return a mesh within the deadline" + ); + std::thread::sleep(Duration::from_millis(1)); + } +} + +#[test] +fn finished_mesh_is_uploaded() { + let mut manager = ChunkManager::new(); + let pos = ChunkPos::new(0, 0, 0); + manager.resident.insert(pos, Arc::new(solid_chunk())); + manager.pending_remesh.insert(pos); + + assert_eq!(manager.dispatch_pending(), 1); + let result = wait_for_result(&manager.pool); + + let mut sink = RecordingSink::default(); + assert!(manager.apply_result(&result, &mut sink)); + // A non-empty mesh is uploaded once and the in-flight entry is cleared. + assert_eq!(sink.inserted, vec![(0, 0, 0)]); + assert!(sink.removed.is_empty()); + assert!(!manager.in_flight.contains_key(&pos)); +} + +#[test] +fn superseded_mesh_is_discarded() { + let mut manager = ChunkManager::new(); + let pos = ChunkPos::new(0, 0, 0); + manager.resident.insert(pos, Arc::new(solid_chunk())); + manager.pending_remesh.insert(pos); + + manager.dispatch_pending(); + let stale = wait_for_result(&manager.pool); + + // A newer job supersedes the outstanding one before the first result is applied. + manager.pending_remesh.insert(pos); + manager.dispatch_pending(); + + let mut sink = RecordingSink::default(); + assert!(!manager.apply_result(&stale, &mut sink)); + assert!(sink.inserted.is_empty()); + assert!(sink.removed.is_empty()); + // The newer job remains tracked as outstanding. + assert!(manager.in_flight.contains_key(&pos)); +} + +#[test] +fn evicted_mesh_is_discarded() { + let mut manager = ChunkManager::new(); + let pos = ChunkPos::new(0, 0, 0); + manager.resident.insert(pos, Arc::new(solid_chunk())); + manager.pending_remesh.insert(pos); + + manager.dispatch_pending(); + let result = wait_for_result(&manager.pool); + + // The chunk leaves the load radius before its mesh arrives. + manager.resident.remove(&pos); + manager.in_flight.remove(&pos); + + let mut sink = RecordingSink::default(); + assert!(!manager.apply_result(&result, &mut sink)); + assert!(sink.inserted.is_empty()); + assert!(sink.removed.is_empty()); +}