diff --git a/README.md b/README.md index 6280dee..b86c312 100644 --- a/README.md +++ b/README.md @@ -59,7 +59,16 @@ docker compose up `api` on :3000, `frontend` on :3001, both behind Caddy on :80. `api` applies its Postgres migrations on startup (see `api/src/db/migrate.ts`); `worker` consumes the `mcmapper:dirty-chunks` -Redis stream via a consumer group (`mcmapper-workers`) so multiple instances split work safely. +Redis stream via a consumer group (`mcmapper-workers`) so multiple instances split work safely — +scale locally with `docker compose up --scale worker=N`, or run a standalone `worker` container +on separate hardware pointed at the same Postgres/Redis/MinIO over a private network (VPN/ +Tailscale/LAN — never expose those ports publicly). + +`worker`'s `RENDER_THREADS`/`RENDER_PROFILE` (see `worker/.env.example`) size the rayon pool that +renders a batch of dirty chunks in parallel and how many chunks are pulled off the stream per +batch: `RENDER_THREADS=auto` uses all available cores, or set a fixed count; `RENDER_PROFILE` +is `server` (default — many small batches, tuned for a many-core box) or `consumer` (fewer, +larger batches, less scheduling overhead on fewer/faster cores). ### Connecting a mod instance diff --git a/worker/.env.example b/worker/.env.example index cce73a1..80561c6 100644 --- a/worker/.env.example +++ b/worker/.env.example @@ -11,3 +11,17 @@ MINIO_SECRET_KEY=changeme-set-in-untracked-env # cpu | gpu | hybrid — see render::backend. Only `cpu` exists so far (Phase 8 adds gpu/hybrid). RENDER_BACKEND=cpu + +# Thread count for the rayon pool that renders a batch of dirty chunks in parallel (see +# config.rs/main.rs — a single 16x16 tile is too small for rayon to help within itself, so the +# parallelism target is many chunks in flight at once). `auto` (default) uses +# std::thread::available_parallelism(); set a positive integer to override (0/garbage also falls +# back to auto). +RENDER_THREADS=auto + +# server | consumer — tunes how many dirty-chunk entries are pulled off the Redis stream per +# xread cycle before being rendered in parallel. `server` (default) batches more aggressively — +# many small batches so every core on a many-core Xeon-style box stays fed; `consumer` uses +# fewer, larger batches, less scheduling overhead per chunk on fewer/faster cores. Unrecognized +# values fall back to `server`, matching this project's own primary deployment target. +RENDER_PROFILE=server diff --git a/worker/src/config.rs b/worker/src/config.rs new file mode 100644 index 0000000..2202819 --- /dev/null +++ b/worker/src/config.rs @@ -0,0 +1,105 @@ +/// Resolves `RENDER_THREADS`/`RENDER_PROFILE` into a concrete thread count + per-xread batch +/// size for the rayon pool that parallelizes tile/mesh rendering across a batch of dirty chunks +/// (see main.rs — a single 16x16 tile is too small for rayon to help within itself, so the +/// parallelism target is "many chunks in flight at once", tuned by these two knobs). Pure/testable +/// on purpose — env var reading and rayon pool construction stay in main.rs. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum Profile { + /// Many-core Xeon-style deployment (the user's own production box): many small batches so + /// every core stays fed without one slow chunk blocking a large batch's completion. + Server, + /// Fewer/faster consumer cores: fewer, larger batches — less scheduling overhead per chunk. + Consumer, +} + +impl Profile { + pub fn parse(raw: &str) -> Self { + match raw { + "consumer" => Profile::Consumer, + // Unrecognized falls back to `server`, not `consumer` — matches this project's own + // primary deployment target (RTX 3090 + Xeons), see the plan's worker section. + _ => Profile::Server, + } + } +} + +pub struct WorkerConfig { + pub threads: usize, + pub batch_size: usize, +} + +/// `raw` is `RENDER_THREADS`'s value: `None`/`"auto"` uses `available`, anything else must parse +/// to a positive integer or it also falls back to `available` (never zero threads). +pub fn resolve_thread_count(raw: Option<&str>, available: usize) -> usize { + match raw { + None => available, + Some(v) => match v.trim() { + "auto" | "" => available, + n => match n.parse::() { + Ok(0) | Err(_) => available, + Ok(n) => n, + }, + }, + } +} + +/// How many dirty-chunk stream entries to pull (and then render in parallel) per xread cycle. +pub fn batch_size_for(profile: Profile) -> usize { + match profile { + Profile::Server => 32, + Profile::Consumer => 4, + } +} + +pub fn resolve(threads_raw: Option<&str>, profile_raw: Option<&str>, available: usize) -> WorkerConfig { + let profile = Profile::parse(profile_raw.unwrap_or("server")); + WorkerConfig { threads: resolve_thread_count(threads_raw, available), batch_size: batch_size_for(profile) } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn auto_or_unset_uses_available_parallelism() { + assert_eq!(resolve_thread_count(None, 8), 8); + assert_eq!(resolve_thread_count(Some("auto"), 8), 8); + assert_eq!(resolve_thread_count(Some(""), 8), 8); + } + + #[test] + fn a_manual_positive_integer_overrides_available_parallelism() { + assert_eq!(resolve_thread_count(Some("4"), 32), 4); + assert_eq!(resolve_thread_count(Some("64"), 4), 64); + } + + #[test] + fn zero_or_garbage_falls_back_to_available_parallelism_rather_than_hanging_with_no_threads() { + assert_eq!(resolve_thread_count(Some("0"), 8), 8); + assert_eq!(resolve_thread_count(Some("not-a-number"), 8), 8); + assert_eq!(resolve_thread_count(Some("-1"), 8), 8); + } + + #[test] + fn server_profile_batches_more_aggressively_than_consumer() { + assert!(batch_size_for(Profile::Server) > batch_size_for(Profile::Consumer)); + } + + #[test] + fn unrecognized_profile_falls_back_to_server_matching_the_primary_deployment_target() { + assert_eq!(Profile::parse("laptop"), Profile::Server); + assert_eq!(Profile::parse(""), Profile::Server); + } + + #[test] + fn consumer_profile_is_recognized_case_sensitively() { + assert_eq!(Profile::parse("consumer"), Profile::Consumer); + } + + #[test] + fn resolve_combines_both_knobs() { + let cfg = resolve(Some("6"), Some("consumer"), 32); + assert_eq!(cfg.threads, 6); + assert_eq!(cfg.batch_size, batch_size_for(Profile::Consumer)); + } +} diff --git a/worker/src/main.rs b/worker/src/main.rs index 5dc01b7..3091fda 100644 --- a/worker/src/main.rs +++ b/worker/src/main.rs @@ -1,3 +1,4 @@ +mod config; mod db; mod mesh; mod palette; @@ -8,6 +9,7 @@ use std::collections::hash_map::DefaultHasher; use std::hash::{Hash, Hasher}; use base64::{engine::general_purpose::STANDARD, Engine as _}; +use rayon::prelude::*; use redis::streams::{StreamReadOptions, StreamReadReply}; use redis::AsyncCommands; use render::{ColumnPixel, CpuRenderBackend, RenderBackend}; @@ -57,13 +59,30 @@ async fn main() -> anyhow::Result<()> { ); } let backend: Box = Box::new(CpuRenderBackend::new()); + + // A single 16x16 tile is too small for rayon to help within itself, so the parallelism + // target is "many chunks in flight at once" — see config.rs's doc comment. RENDER_THREADS/ + // RENDER_PROFILE (server=many small batches for many-core Xeons, consumer=fewer/larger + // batches) tune the pool size and how many stream entries are pulled per xread cycle. + let available = std::thread::available_parallelism().map(|n| n.get()).unwrap_or(4); + let worker_cfg = config::resolve( + std::env::var("RENDER_THREADS").ok().as_deref(), + std::env::var("RENDER_PROFILE").ok().as_deref(), + available, + ); + let pool = rayon::ThreadPoolBuilder::new().num_threads(worker_cfg.threads).build()?; + println!("[worker] render backend: {}", backend.name()); + println!( + "[worker] threads={} batch_size={} (available_parallelism={available})", + worker_cfg.threads, worker_cfg.batch_size + ); println!("[worker] consuming stream '{DIRTY_CHUNK_STREAM}' as '{consumer_name}'"); loop { let opts = StreamReadOptions::default() .group(CONSUMER_GROUP, &consumer_name) - .count(10) + .count(worker_cfg.batch_size) .block(5000); let reply: StreamReadReply = conn @@ -71,15 +90,51 @@ async fn main() -> anyhow::Result<()> { .await .unwrap_or_default(); + let mut jobs = Vec::new(); for key in reply.keys { for entry in key.ids { - if let Err(err) = process_entry(&pg_pool, &s3_client, backend.as_ref(), &entry).await { - eprintln!("[worker] failed to process {}: {err:#}", entry.id); + match parse_job(&entry) { + Ok(job) => jobs.push(job), + Err(err) => eprintln!("[worker] failed to parse {}: {err:#}", entry.id), + } + } + } + if jobs.is_empty() { + continue; + } + + // Stage 1 (async): fetch each job's source data. Sequential — the pg pool is capped at + // 5 connections (see db::connect) so unbounded concurrency here wouldn't help, and + // keeping this simple leaves the actual parallelism budget for the CPU-bound stage below. + let mut fetched = Vec::with_capacity(jobs.len()); + for job in jobs { + match fetch_chunk_data(&pg_pool, &job).await { + Ok(data) => fetched.push(data), + Err(err) => eprintln!("[worker] failed to fetch {}: {err:#}", job.entry_id), + } + } + + // Stage 2 (CPU-bound, parallel): rasterize + mesh every chunk in the batch across the + // sized rayon pool. + let backend_ref = backend.as_ref(); + let rendered: Vec> = + pool.install(|| fetched.par_iter().map(|data| render_chunk(data, backend_ref)).collect()); + + // Stage 3 (async): write results back and ack, sequentially. + for result in rendered { + let chunk = match result { + Ok(chunk) => chunk, + Err(err) => { + eprintln!("[worker] failed to render: {err:#}"); continue; } - let _: redis::RedisResult<()> = - conn.xack(DIRTY_CHUNK_STREAM, CONSUMER_GROUP, &[&entry.id]).await; + }; + let entry_id = chunk.job.entry_id.clone(); + if let Err(err) = store_rendered_chunk(&pg_pool, &s3_client, chunk).await { + eprintln!("[worker] failed to store {entry_id}: {err:#}"); + continue; } + let _: redis::RedisResult<()> = conn.xack(DIRTY_CHUNK_STREAM, CONSUMER_GROUP, &[&entry_id]).await; } } } @@ -92,52 +147,57 @@ fn field(entry: &redis::streams::StreamId, name: &str) -> anyhow::Result } } -async fn process_entry( - pool: &sqlx::PgPool, - s3_client: &aws_sdk_s3::Client, - backend: &dyn RenderBackend, - entry: &redis::streams::StreamId, -) -> anyhow::Result<()> { - let server_id: Uuid = field(entry, "serverId")?.parse()?; - let dimension: i32 = field(entry, "dimension")?.parse()?; - let chunk_x: i32 = field(entry, "chunkX")?.parse()?; - let chunk_z: i32 = field(entry, "chunkZ")?.parse()?; +#[derive(Clone)] +struct ChunkJob { + entry_id: String, + server_id: Uuid, + dimension: i32, + chunk_x: i32, + chunk_z: i32, +} - let columns = db::fetch_chunk_columns(pool, server_id, dimension, chunk_x, chunk_z).await?; - let pixels: Vec = columns - .iter() - .map(|c| ColumnPixel { - local_x: (c.x - chunk_x * 16) as u8, - local_z: (c.z - chunk_z * 16) as u8, - block_id: c.block_id as u16, - block_meta: c.block_meta as u8, - }) - .collect(); +fn parse_job(entry: &redis::streams::StreamId) -> anyhow::Result { + Ok(ChunkJob { + entry_id: entry.id.clone(), + server_id: field(entry, "serverId")?.parse()?, + dimension: field(entry, "dimension")?.parse()?, + chunk_x: field(entry, "chunkX")?.parse()?, + chunk_z: field(entry, "chunkZ")?.parse()?, + }) +} - let png_bytes = backend.rasterize_tile(&pixels)?; +struct ChunkData { + job: ChunkJob, + columns: Vec, + sections: Vec, +} +async fn fetch_chunk_data(pool: &sqlx::PgPool, job: &ChunkJob) -> anyhow::Result { + let columns = + db::fetch_chunk_columns(pool, job.server_id, job.dimension, job.chunk_x, job.chunk_z).await?; + let sections = + db::fetch_chunk_sections(pool, job.server_id, job.dimension, job.chunk_x, job.chunk_z).await?; + Ok(ChunkData { job: job.clone(), columns, sections }) +} + +struct RenderedMesh { + section_y: i32, + bytes: Vec, + content_hash: String, +} + +struct RenderedChunk { + job: ChunkJob, + png_bytes: Vec, + tile_content_hash: String, + column_count: usize, + meshes: Vec, +} + +fn content_hash(bytes: &[u8]) -> String { let mut hasher = DefaultHasher::new(); - png_bytes.hash(&mut hasher); - let content_hash = format!("{:x}", hasher.finish()); - - let storage_key = format!("{server_id}/{dimension}/0/{chunk_x}/{chunk_z}.png"); - storage::put_object(s3_client, &storage_key, "image/png", png_bytes).await?; - db::upsert_tile_pointer( - pool, - server_id, - dimension, - 0, - chunk_x, - chunk_z, - &storage_key, - &content_hash, - ) - .await?; - - println!("[worker] rendered tile {storage_key} ({} columns)", pixels.len()); - - mesh_chunk(pool, s3_client, server_id, dimension, chunk_x, chunk_z).await?; - Ok(()) + bytes.hash(&mut hasher); + format!("{:x}", hasher.finish()) } fn decode_blocks(base64_blocks: &str) -> anyhow::Result<[u16; 4096]> { @@ -152,56 +212,92 @@ fn decode_blocks(base64_blocks: &str) -> anyhow::Result<[u16; 4096]> { Ok(blocks) } -async fn mesh_chunk( - pool: &sqlx::PgPool, - s3_client: &aws_sdk_s3::Client, - server_id: Uuid, - dimension: i32, - chunk_x: i32, - chunk_z: i32, -) -> anyhow::Result<()> { - let sections = db::fetch_chunk_sections(pool, server_id, dimension, chunk_x, chunk_z).await?; - if sections.is_empty() { - return Ok(()); // this server hasn't sent 3D data yet (Phase 2 mod support) — fine, no-op - } +// Pure CPU-bound work — called from inside the rayon pool, one call per chunk in the batch. +fn render_chunk(data: &ChunkData, backend: &dyn RenderBackend) -> anyhow::Result { + let job = &data.job; + let pixels: Vec = data + .columns + .iter() + .map(|c| ColumnPixel { + local_x: (c.x - job.chunk_x * 16) as u8, + local_z: (c.z - job.chunk_z * 16) as u8, + block_id: c.block_id as u16, + block_meta: c.block_meta as u8, + }) + .collect(); + let png_bytes = backend.rasterize_tile(&pixels)?; + let tile_content_hash = content_hash(&png_bytes); - for section in sections { + let mut meshes = Vec::new(); + for section in &data.sections { let blocks = match decode_blocks(§ion.blocks) { Ok(b) => b, Err(err) => { eprintln!( - "[worker] skipping malformed section ({chunk_x},{chunk_z},{}): {err:#}", - section.section_y + "[worker] skipping malformed section ({},{},{}): {err:#}", + job.chunk_x, job.chunk_z, section.section_y ); continue; } }; - let mesh_buf = mesh::mesh_section(&blocks); if mesh_buf.is_empty() { continue; } - let mesh_bytes = mesh_buf.encode(); + let bytes = mesh_buf.encode(); + let hash = content_hash(&bytes); + meshes.push(RenderedMesh { section_y: section.section_y, bytes, content_hash: hash }); + } - let mut hasher = DefaultHasher::new(); - mesh_bytes.hash(&mut hasher); - let content_hash = format!("{:x}", hasher.finish()); + Ok(RenderedChunk { + job: job.clone(), + png_bytes, + tile_content_hash, + column_count: pixels.len(), + meshes, + }) +} - let storage_key = - format!("{server_id}/{dimension}/mesh/{chunk_x}/{chunk_z}/{}.bin", section.section_y); - storage::put_object(s3_client, &storage_key, "application/octet-stream", mesh_bytes).await?; +async fn store_rendered_chunk( + pool: &sqlx::PgPool, + s3_client: &aws_sdk_s3::Client, + chunk: RenderedChunk, +) -> anyhow::Result<()> { + let job = &chunk.job; + let storage_key = format!("{}/{}/0/{}/{}.png", job.server_id, job.dimension, job.chunk_x, job.chunk_z); + storage::put_object(s3_client, &storage_key, "image/png", chunk.png_bytes).await?; + db::upsert_tile_pointer( + pool, + job.server_id, + job.dimension, + 0, + job.chunk_x, + job.chunk_z, + &storage_key, + &chunk.tile_content_hash, + ) + .await?; + println!("[worker] rendered tile {storage_key} ({} columns)", chunk.column_count); + + for mesh in chunk.meshes { + let storage_key = format!( + "{}/{}/mesh/{}/{}/{}.bin", + job.server_id, job.dimension, job.chunk_x, job.chunk_z, mesh.section_y + ); + storage::put_object(s3_client, &storage_key, "application/octet-stream", mesh.bytes).await?; db::upsert_mesh_pointer( pool, - server_id, - dimension, - chunk_x, - chunk_z, - section.section_y, + job.server_id, + job.dimension, + job.chunk_x, + job.chunk_z, + mesh.section_y, &storage_key, - &content_hash, + &mesh.content_hash, ) .await?; println!("[worker] rendered mesh {storage_key}"); } + Ok(()) }