From f630da6d1c4999b080ab3e959c3500353e7f72e2 Mon Sep 17 00:00:00 2001 From: Patrick Owen Date: Sun, 9 Aug 2026 23:07:43 -0400 Subject: [PATCH 1/6] Create a new AssetLoader to eventually replace Loader --- Cargo.lock | 56 ++++ client/Cargo.toml | 4 +- client/src/config.rs | 20 ++ client/src/graphics/asset_loader.rs | 392 ++++++++++++++++++++++++++++ client/src/graphics/mod.rs | 1 + 5 files changed, 472 insertions(+), 1 deletion(-) create mode 100644 client/src/graphics/asset_loader.rs diff --git a/Cargo.lock b/Cargo.lock index e1ac6afa..5ea6bf17 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -196,6 +196,18 @@ dependencies = [ "syn 2.0.119", ] +[[package]] +name = "async-channel" +version = "2.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "924ed96dd52d1b75e9c1a3e6275715fd320f5f9439fb5a4a11fa51f4221158d2" +dependencies = [ + "concurrent-queue", + "event-listener-strategy", + "futures-core", + "pin-project-lite", +] + [[package]] name = "atomic-waker" version = "1.1.2" @@ -430,12 +442,14 @@ dependencies = [ "metrics", "nalgebra", "png", + "pollster", "quinn", "raw-window-handle", "renderdoc", "save", "serde", "server", + "skid-steer", "tokio", "toml", "tracing", @@ -798,6 +812,26 @@ dependencies = [ "windows-sys 0.61.2", ] +[[package]] +name = "event-listener" +version = "5.4.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5a23add41df1562121a9393cb065eab5146a1242410f23a644851e90cfd669d2" +dependencies = [ + "parking", + "pin-project-lite", +] + +[[package]] +name = "event-listener-strategy" +version = "0.5.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8be9f3dfaaffdae2972880079a491a1a8bb7cbed0b8dd7a347f668b4150a3b93" +dependencies = [ + "event-listener", + "pin-project-lite", +] + [[package]] name = "fast-srgb8" version = "1.0.0" @@ -1889,6 +1923,12 @@ dependencies = [ "winapi", ] +[[package]] +name = "parking" +version = "2.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f38d5652c16fde515bb1ecef450ab0f6a219d619a7274976324d5e377f7dceba" + [[package]] name = "parking_lot" version = "0.12.5" @@ -2022,6 +2062,12 @@ dependencies = [ "windows-sys 0.61.2", ] +[[package]] +name = "pollster" +version = "1.0.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bc6355899e1c9462875b6757c79f3caa011a1fdae12bbb1a2e72dd1f234f8336" + [[package]] name = "portable-atomic" version = "1.14.0" @@ -2760,6 +2806,16 @@ version = "0.1.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e3a9fe34e3e7a50316060351f37187a3f546bce95496156754b601a5fa71b76e" +[[package]] +name = "skid-steer" +version = "0.1.0" +source = "git+https://github.com/Ralith/skid-steer?rev=16feada2408810f78a9f469001ef9277e98d4061#16feada2408810f78a9f469001ef9277e98d4061" +dependencies = [ + "async-channel", + "foldhash 0.1.5", + "tokio", +] + [[package]] name = "slab" version = "0.4.12" diff --git a/client/Cargo.toml b/client/Cargo.toml index b549ea2b..604f1cca 100644 --- a/client/Cargo.toml +++ b/client/Cargo.toml @@ -23,7 +23,8 @@ directories = "6.0.0" vk-shader-macros = "0.2.5" nalgebra = { workspace = true } libm = "0.2.16" -tokio = { version = "1.43.0", features = ["rt-multi-thread", "sync", "macros"] } +tokio = { version = "1.52.3", features = ["rt", "rt-multi-thread", "sync", "macros"] } +pollster = "1.0.1" png = "0.18.0" anyhow = "1.0.26" serde = { version = "1.0.104", features = ["derive", "rc"] } @@ -40,6 +41,7 @@ metrics = "0.24.0" hdrhistogram = { version = "7", default-features = false } save = { path = "../save" } lru-slab = "0.1.2" +skid-steer = { git="https://github.com/Ralith/skid-steer", rev = "16feada2408810f78a9f469001ef9277e98d4061" } [features] default = ["use-repo-assets"] diff --git a/client/src/config.rs b/client/src/config.rs index a13d78df..a48d858c 100644 --- a/client/src/config.rs +++ b/client/src/config.rs @@ -15,6 +15,7 @@ pub struct Config { pub data_dirs: Vec, pub save: PathBuf, pub chunk_load_parallelism: u32, + pub asset_load_parallelism: u32, pub server: Option, pub local_simulation: SimConfig, } @@ -30,6 +31,7 @@ impl Config { save, local_simulation, chunk_load_parallelism, + asset_load_parallelism, server, } = match fs::read(&path) { Ok(data) => { @@ -83,6 +85,10 @@ impl Config { data_dirs, save: save.unwrap_or("default.save".into()), chunk_load_parallelism: chunk_load_parallelism.unwrap_or(256), + asset_load_parallelism: asset_load_parallelism.unwrap_or_else(|| { + std::thread::available_parallelism() + .map_or(1, |threads| threads.get().min(16) as u32) + }), server, local_simulation: SimConfig::from_raw(&local_simulation), } @@ -98,6 +104,19 @@ impl Config { } None } + + #[cfg(test)] + pub fn create_for_test() -> Config { + Config { + name: "test_player".into(), + data_dirs: vec![], + save: "save_file_only_to_be_used_for_unit_test.save".into(), + chunk_load_parallelism: 4, + asset_load_parallelism: 4, + server: None, + local_simulation: SimConfig::from_raw(&SimConfigRaw::default()), + } + } } /// Data as parsed directly out of the config file @@ -108,6 +127,7 @@ struct RawConfig { data_dir: Option, save: Option, chunk_load_parallelism: Option, + asset_load_parallelism: Option, server: Option, #[serde(default)] local_simulation: SimConfigRaw, diff --git a/client/src/graphics/asset_loader.rs b/client/src/graphics/asset_loader.rs new file mode 100644 index 00000000..e657e911 --- /dev/null +++ b/client/src/graphics/asset_loader.rs @@ -0,0 +1,392 @@ +use std::{ + path::{Path, PathBuf}, + rc::Rc, + sync::Arc, + thread::{self, JoinHandle}, +}; + +use ash::vk; +use skid_steer::Context; + +use crate::{Config, graphics::Base}; + +/// Contains all the dependencies necessary to load assets. +pub struct AssetLoadContext { + gfx: Arc, + config: Arc, +} + +// Rather than exposing its fields, we expose helper functions for the kinds of tasks +// one would need the context for. This helps add some level of separation between how global +// data is organized and what asset loading code sees +impl AssetLoadContext { + pub fn device(&self) -> &ash::Device { + self.gfx.device.as_ref() + } + + pub fn memory_properties(&self) -> &vk::PhysicalDeviceMemoryProperties { + &self.gfx.memory_properties + } + + pub fn queue_family(&self) -> u32 { + self.gfx.queue_family + } + + pub fn find_asset(&self, path: &Path) -> Option { + self.config.find_asset(path) + } +} + +pub struct AssetLoader { + loader: skid_steer::Loader, + task_executor_threads: Vec>, +} + +impl AssetLoader { + pub fn new(gfx: Arc, config: Arc) -> Self { + let loader = skid_steer::Loader::new(); + + let mut task_executor_threads = vec![]; + tracing::debug!( + "Using asset load parallelism {}", + config.asset_load_parallelism + ); + for i in 0..(config.asset_load_parallelism) { + let asset_load_context = AssetLoadContext { + gfx: Arc::clone(&gfx), + config: Arc::clone(&config), + }; + let loader = loader.clone(); + + let thread = thread::Builder::new() + .name(format!("task_executor_{}", i).to_owned()) + .spawn(move || run_task_executor_thread(asset_load_context, loader)) + .unwrap(); + task_executor_threads.push(thread); + } + + AssetLoader { + loader, + task_executor_threads, + } + } + + pub fn load( + &self, + source: S, + ) -> skid_steer::Asset<::Output> { + self.loader.load(source) + } +} + +/// Runs one thread of the task executor logic. This executor owns an [`AssetLoadContext`] and +/// concurrently runs tasks from the [`skid_steer::Loader`]. Returns the [`parallel_queue::Handle`] +/// from the [`AssetLoadContext`] for later cleanup. +fn run_task_executor_thread(asset_load_context: AssetLoadContext, loader: skid_steer::Loader) { + let asset_load_context = Rc::new(asset_load_context); + + tokio::runtime::LocalRuntime::new() + .unwrap() + .block_on(async { + let mut join_set = tokio::task::JoinSet::new(); + while let Some(task) = loader.next_task().await { + tracing::trace!( + "Found task on {}", + thread::current().name().unwrap_or("") + ); + let asset_load_context = Rc::clone(&asset_load_context); + join_set.spawn_local(async move { + let mut context = Context::new(); + context.insert::(&asset_load_context); + task.run(&context).await; + tracing::trace!( + "Task complete on {}", + thread::current().name().unwrap_or("") + ); + }); + while join_set.try_join_next().is_some() {} // Drain the join set to avoid memory leaks + } + tracing::trace!( + "Ending task executor {}", + thread::current().name().unwrap_or("") + ); + join_set.join_all().await; // Since a bug can result in a deadlock here, we log before and after this call. + tracing::trace!( + "Task executor ended successfully {}", + thread::current().name().unwrap_or("") + ); + }); +} + +impl Drop for AssetLoader { + fn drop(&mut self) { + tracing::trace!("Shutting down AssetLoader"); + pollster::block_on(async { + self.loader.drain().await; + self.loader.clear_cache(); + self.loader.drain().await; + }); + assert!(self.loader.all_assets_freed()); + self.loader.close(); + for thread in self.task_executor_threads.drain(..) { + thread.join().unwrap(); + } + } +} + +#[cfg(test)] +mod tests { + use std::{collections::HashSet, sync::Mutex, time::Duration}; + + use super::*; + + /// Record-keeping for what has happened so far with the asset loader, useful for assertions + #[derive(Debug, Clone)] + struct EventList { + events: Arc>>, + num_checked_events: usize, + event_received: Arc, + } + + impl EventList { + fn new() -> Self { + EventList { + events: Arc::new(Mutex::new(vec![])), + num_checked_events: 0, + event_received: Arc::new(std::sync::Condvar::new()), + } + } + + fn push_event(&self, event: Event) { + self.events.lock().unwrap().push(event); + self.event_received.notify_all(); + } + + fn get_all_events(&mut self) -> Vec { + let events = self.events.lock().unwrap().clone(); + self.num_checked_events = events.len(); + events + } + + /// Wait until an event has been received since the last call to get_all_events + fn wait_timeout_while( + &self, + timeout: std::time::Duration, + mut condition: impl FnMut(&[Event]) -> bool, + ) -> std::sync::WaitTimeoutResult { + let events = self.events.lock().unwrap(); + self.event_received + .wait_timeout_while(events, timeout, |events| condition(events)) + .unwrap() + .1 + } + + /// Drains all events that were returned by the most recent call to get_all_events + fn drain_queried_events(&mut self) { + self.events + .lock() + .unwrap() + .drain(0..self.num_checked_events); + self.num_checked_events = 0; + } + } + + #[derive(Clone, Debug, PartialEq, Eq, PartialOrd, Ord, Hash)] + enum Event { + Progress { + asset: String, + percent_progress: u32, + }, + Loaded { + asset: String, + }, + Freed { + asset: String, + }, + LoadCanceled { + asset: String, + }, + } + + impl Event { + fn progress(asset: &str, percent_progress: u32) -> Self { + Event::Progress { + asset: asset.to_owned(), + percent_progress, + } + } + + fn loaded(asset: &str) -> Self { + Event::Loaded { + asset: asset.to_owned(), + } + } + + fn freed(asset: &str) -> Self { + Event::Freed { + asset: asset.to_owned(), + } + } + + fn load_cancelled(asset: &str) -> Self { + Event::LoadCanceled { + asset: asset.to_owned(), + } + } + } + + struct TestAsset { + asset: skid_steer::Asset, + progress_sender: tokio::sync::mpsc::UnboundedSender, + } + + impl TestAsset { + fn add_percent_progress(&self, percent_progress: u32) { + self.progress_sender.send(percent_progress).unwrap(); + } + + fn wait_for_completion(&self) { + // Note: This design pattern is somewhat dangerous because it can panic if called from within another runtime. + // Since this is only used in unit tests, this should be an acceptable level of risk. + tokio::runtime::LocalRuntime::new() + .unwrap() + .block_on(async { + tokio::select! { + biased; + _ = self.asset.get() => (), + _ = tokio::time::sleep(Duration::from_secs(5)) => { panic!("Timed out waiting for asset to load") }, + }; + }); + } + } + + struct DummyAsset { + name: String, + events: EventList, + } + + struct DummyAssetSource { + name: String, + events: EventList, + progress_receiver: tokio::sync::mpsc::UnboundedReceiver, + } + + impl skid_steer::Source for DummyAssetSource { + type Output = DummyAsset; + + async fn load(mut self, _context: &Context<'_>) -> Option { + let mut status = DummyAssetStatus { + name: self.name.clone(), + progress: 0, + can_cancel: true, + events: self.events.clone(), + }; + + while status.progress < 100 { + status.progress += self.progress_receiver.recv().await?; + self.events + .push_event(Event::progress(&self.name, status.progress)); + } + + status.can_cancel = false; // Done loading + self.events.push_event(Event::loaded(&self.name)); + Some(DummyAsset { + name: self.name, + events: self.events, + }) + } + + fn free(output: Self::Output, _context: &Context) { + output.events.push_event(Event::freed(&output.name)); + } + } + + struct DummyAssetStatus { + name: String, + progress: u32, + can_cancel: bool, + events: EventList, + } + + impl Drop for DummyAssetStatus { + fn drop(&mut self) { + if self.can_cancel { + self.events.push_event(Event::load_cancelled(&self.name)); + } + } + } + + fn init_asset_loader(asset_load_parallelism: u32) -> AssetLoader { + let gfx = Arc::new(Base::headless()); + let config = Arc::new({ + let mut config = Config::create_for_test(); + config.asset_load_parallelism = asset_load_parallelism; + config + }); + AssetLoader::new(Arc::clone(&gfx), Arc::clone(&config)) + } + + fn load_dummy_asset( + asset_loader: &AssetLoader, + events: &EventList, + name: &str, + ) -> TestAsset { + let (progress_sender, progress_receiver) = tokio::sync::mpsc::unbounded_channel(); + TestAsset { + asset: asset_loader.load(DummyAssetSource { + name: name.to_owned(), + events: events.clone(), + progress_receiver, + }), + progress_sender, + } + } + + #[test] + fn test_load_and_free() { + let mut events = EventList::new(); + let asset_loader = init_asset_loader(2); + let dummy_asset = load_dummy_asset(&asset_loader, &events, "asset"); + dummy_asset.add_percent_progress(50); + dummy_asset.add_percent_progress(50); + dummy_asset.wait_for_completion(); + assert!(dummy_asset.asset.try_get().is_some()); + assert_eq!( + events.get_all_events(), + &[ + Event::progress("asset", 50), + Event::progress("asset", 100), + Event::loaded("asset") + ] + ); + events.drain_queried_events(); + drop(dummy_asset); + drop(asset_loader); + assert_eq!(events.get_all_events(), &[Event::freed("asset")]); + } + + #[test] + fn test_concurrency_and_cancellation() { + let mut events = EventList::new(); + let asset_loader = init_asset_loader(2); + let assets: Vec<_> = (0..4) + .map(|i| load_dummy_asset(&asset_loader, &events, &format!("asset{i}"))) + .collect(); + for asset in &assets { + asset.add_percent_progress(50); + } + let expected_events = [ + Event::progress("asset0", 50), + Event::progress("asset1", 50), + Event::progress("asset2", 50), + Event::progress("asset3", 50), + ] + .into_iter() + .collect::>(); + events.wait_timeout_while(Duration::from_secs(5), |events| { + events.len() < expected_events.len() + }); + let actual_events = events.get_all_events().into_iter().collect::>(); + assert_eq!(actual_events, expected_events); + } +} diff --git a/client/src/graphics/mod.rs b/client/src/graphics/mod.rs index c78a0a5b..2d27751a 100644 --- a/client/src/graphics/mod.rs +++ b/client/src/graphics/mod.rs @@ -1,5 +1,6 @@ #![allow(clippy::missing_safety_doc)] // Vulkan wrangling is categorically unsafe +#[expect(unused)] mod asset_loader; mod base; mod core; mod draw; From 82bcfefc3b41f6a18fbfba7fb6783f26934f5012 Mon Sep 17 00:00:00 2001 From: Patrick Owen Date: Sun, 16 Aug 2026 16:30:10 -0400 Subject: [PATCH 2/6] Add Vulkan functionality to new asset_loader --- Cargo.lock | 22 ++- client/Cargo.toml | 3 +- client/src/graphics/asset_loader.rs | 232 +++++++++++++++++++++++++++- client/src/graphics/base.rs | 3 +- 4 files changed, 253 insertions(+), 7 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 5ea6bf17..3d80633e 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -451,6 +451,7 @@ dependencies = [ "server", "skid-steer", "tokio", + "tokio-util", "toml", "tracing", "vk-shader-macros", @@ -950,6 +951,12 @@ dependencies = [ "syn 2.0.119", ] +[[package]] +name = "futures-sink" +version = "0.3.34" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1944426bf7d03f1d14f708785e4b33efd750b36d48a157b836b3efc15ede8e1d" + [[package]] name = "futures-task" version = "0.3.33" @@ -1314,7 +1321,7 @@ dependencies = [ [[package]] name = "lahar" version = "0.1.0" -source = "git+https://github.com/Ralith/lahar?rev=cb2ceca83aa3a02f20727772f6dc1bcf39fc2462#cb2ceca83aa3a02f20727772f6dc1bcf39fc2462" +source = "git+https://github.com/Ralith/lahar?rev=67c190a000472720f7e9dc93899bcaaa72fe1574#67c190a000472720f7e9dc93899bcaaa72fe1574" dependencies = [ "ash", ] @@ -3105,6 +3112,19 @@ dependencies = [ "syn 3.0.3", ] +[[package]] +name = "tokio-util" +version = "0.7.19" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "494815d09bf52b5548659851081238f0ca39ff638363907596da739561c62c52" +dependencies = [ + "bytes", + "futures-core", + "futures-sink", + "pin-project-lite", + "tokio", +] + [[package]] name = "toml" version = "1.1.4+spec-1.1.0" diff --git a/client/Cargo.toml b/client/Cargo.toml index 604f1cca..4a1f9fc0 100644 --- a/client/Cargo.toml +++ b/client/Cargo.toml @@ -13,7 +13,7 @@ common = { path = "../common" } server = { path = "../server" } tracing = "0.1.10" ash = { version = "0.38.0", default-features = false, features = ["loaded", "debug", "std"] } -lahar = { git = "https://github.com/Ralith/lahar", rev = "cb2ceca83aa3a02f20727772f6dc1bcf39fc2462" } +lahar = { git = "https://github.com/Ralith/lahar", rev = "67c190a000472720f7e9dc93899bcaaa72fe1574" } yakui = "0.3.0" yakui-vulkan = "0.3.0" winit = "0.30.4" @@ -24,6 +24,7 @@ vk-shader-macros = "0.2.5" nalgebra = { workspace = true } libm = "0.2.16" tokio = { version = "1.52.3", features = ["rt", "rt-multi-thread", "sync", "macros"] } +tokio-util = "0.7.19" pollster = "1.0.1" png = "0.18.0" anyhow = "1.0.26" diff --git a/client/src/graphics/asset_loader.rs b/client/src/graphics/asset_loader.rs index e657e911..f331e705 100644 --- a/client/src/graphics/asset_loader.rs +++ b/client/src/graphics/asset_loader.rs @@ -1,12 +1,15 @@ use std::{ path::{Path, PathBuf}, + ptr::NonNull, rc::Rc, sync::Arc, thread::{self, JoinHandle}, }; use ash::vk; +use lahar::{GrowableRing, ParallelQueue, parallel_queue}; use skid_steer::Context; +use tokio_util::sync::CancellationToken; use crate::{Config, graphics::Base}; @@ -14,12 +17,42 @@ use crate::{Config, graphics::Base}; pub struct AssetLoadContext { gfx: Arc, config: Arc, + queue_handle: parallel_queue::Handle, + queue_watch_receiver: tokio::sync::watch::Receiver, + queue_unparker: Arc, + staging: Arc, } // Rather than exposing its fields, we expose helper functions for the kinds of tasks // one would need the context for. This helps add some level of separation between how global // data is organized and what asset loading code sees impl AssetLoadContext { + /// # Safety + /// - [`Work::cmd`] must not be used outside the lifetime of the returned [`Work`] + /// - Any Vulkan resources this work uses must not be destroyed before the [`Work`] + /// is fully executed (or dropped without being sent for submission). + pub unsafe fn begin_work(&self) -> parallel_queue::Work<'_> { + unsafe { self.queue_handle.begin(&self.gfx.device) } + } + + pub fn alloc_staging( + &self, + count: usize, + align: usize, + free_at: u64, + ) -> GrowableRingAllocation { + let (buffer, offset, pointer) = + self.staging + .alloc(&self.gfx.device, None, count, align, free_at); + let size = (count * std::mem::size_of::()) as u64; + GrowableRingAllocation { + buffer, + offset, + size, + pointer, + } + } + pub fn device(&self) -> &ash::Device { self.gfx.device.as_ref() } @@ -28,8 +61,16 @@ impl AssetLoadContext { &self.gfx.memory_properties } - pub fn queue_family(&self) -> u32 { - self.gfx.queue_family + pub async fn wait_for_completion(&self, semaphore_value: u64) { + // To actually get the work to start, we need to unpark the queue. We do it here to avoid getting stuck awaiting something we never kicked off. + unsafe { self.queue_unparker.unpark_queue(&self.gfx.device) }; + + self + .queue_watch_receiver + .clone() + .wait_for(|&value| value >= semaphore_value) + .await + .expect("queue_watch_sender should not be dropped until there are no more AssetLoadContexts"); } pub fn find_asset(&self, path: &Path) -> Option { @@ -37,14 +78,37 @@ impl AssetLoadContext { } } +/// Convenience wrapper around lahar::GrowableRing's allocation tuple +pub struct GrowableRingAllocation { + pub buffer: vk::Buffer, + pub offset: u64, + pub size: u64, + pub pointer: NonNull, +} + pub struct AssetLoader { + gfx: Arc, + queue_shutdown_token: CancellationToken, loader: skid_steer::Loader, + staging: Arc, + queue_unparker: Arc, task_executor_threads: Vec>, + queue_driver_thread: Option>, } impl AssetLoader { pub fn new(gfx: Arc, config: Arc) -> Self { let loader = skid_steer::Loader::new(); + let queue_shutdown_token = CancellationToken::new(); + let queue = unsafe { ParallelQueue::new(&gfx.device, gfx.queue_family, gfx.queue, None) }; + let staging = Arc::new(GrowableRing::new( + &gfx.device, + gfx.memory_properties, + None, + 32 * 1024 * 1024, + )); + let queue_unparker = Arc::new(QueueUnparker::new(&gfx.device)); + let (queue_watch_sender, queue_watch_receiver) = tokio::sync::watch::channel(0); let mut task_executor_threads = vec![]; tracing::debug!( @@ -55,6 +119,10 @@ impl AssetLoader { let asset_load_context = AssetLoadContext { gfx: Arc::clone(&gfx), config: Arc::clone(&config), + queue_handle: unsafe { queue.handle(&gfx.device) }, + queue_watch_receiver: queue_watch_receiver.clone(), + queue_unparker: Arc::clone(&queue_unparker), + staging: Arc::clone(&staging), }; let loader = loader.clone(); @@ -65,9 +133,30 @@ impl AssetLoader { task_executor_threads.push(thread); } + let queue_driver = QueueDriver { + gfx: Arc::clone(&gfx), + queue, + queue_unpark_semaphore: queue_unparker.semaphore(), + queue_shutdown_token: queue_shutdown_token.clone(), + queue_watch_sender: queue_watch_sender.clone(), + staging: Arc::clone(&staging), + }; + + let queue_driver_thread = thread::Builder::new() + .name("queue_driver".to_owned()) + .spawn(move || { + queue_driver.run(); + }) + .unwrap(); + AssetLoader { + gfx, + queue_shutdown_token, loader, + staging, + queue_unparker, task_executor_threads, + queue_driver_thread: Some(queue_driver_thread), } } @@ -79,11 +168,64 @@ impl AssetLoader { } } +struct QueueUnparker { + semaphore_value: std::sync::Mutex, + semaphore: vk::Semaphore, +} + +impl QueueUnparker { + fn new(device: &ash::Device) -> Self { + QueueUnparker { + semaphore_value: std::sync::Mutex::new(0), + semaphore: unsafe { + device + .create_semaphore( + &vk::SemaphoreCreateInfo::default().push_next( + &mut vk::SemaphoreTypeCreateInfo::default() + .semaphore_type(vk::SemaphoreType::TIMELINE) + .initial_value(0), + ), + None, + ) + .unwrap() + }, + } + } + + /// Safety: The device passed in must match the device passed in to `new` + unsafe fn unpark_queue(&self, device: &ash::Device) { + let mut semaphore_value = self.semaphore_value.lock().unwrap(); + *semaphore_value += 1; + + // Safety: The `semaphore_value` lock is held while the semaphore is updated, so that should ensure that it is always + // signaled with a strictly increasing value + unsafe { + device + .signal_semaphore( + &vk::SemaphoreSignalInfo::default() + .semaphore(self.semaphore) + .value(*semaphore_value), + ) + .unwrap() + }; + } + + fn semaphore(&self) -> vk::Semaphore { + self.semaphore + } + + /// Safety: The device passed in must match the device passed in to `new`. Also, the semaphore must + /// no longer be in use. This also means that it is unsound to call `unpark_queue` after calling `destroy` + unsafe fn destroy(&self, device: &ash::Device) { + unsafe { device.destroy_semaphore(self.semaphore, None) }; + } +} + /// Runs one thread of the task executor logic. This executor owns an [`AssetLoadContext`] and /// concurrently runs tasks from the [`skid_steer::Loader`]. Returns the [`parallel_queue::Handle`] /// from the [`AssetLoadContext`] for later cleanup. fn run_task_executor_thread(asset_load_context: AssetLoadContext, loader: skid_steer::Loader) { - let asset_load_context = Rc::new(asset_load_context); + let mut asset_load_context = Rc::new(asset_load_context); tokio::runtime::LocalRuntime::new() .unwrap() @@ -116,6 +258,67 @@ fn run_task_executor_thread(asset_load_context: AssetLoadContext, loader: skid_s thread::current().name().unwrap_or("") ); }); + let asset_load_context = Rc::get_mut(&mut asset_load_context) + .expect("runtime using this context should already be dropped"); + + // Safety: We make sure not to destroy the handle until we drain it. Since there are no other references to the handle, + // no work will be in flight when the handle is destroyed. + unsafe { + // Fail-safe to ensure that the parallel queue is driven at least once after all work has been sent for submission before + // we drain the handle + asset_load_context + .queue_unparker + .unpark_queue(asset_load_context.device()); + + asset_load_context + .queue_handle + .drain(&asset_load_context.gfx.device); + asset_load_context + .queue_handle + .destroy(&asset_load_context.gfx.device); + }; +} + +struct QueueDriver { + gfx: Arc, + queue: ParallelQueue, + queue_unpark_semaphore: vk::Semaphore, + queue_shutdown_token: CancellationToken, + queue_watch_sender: tokio::sync::watch::Sender, + staging: Arc, +} + +impl QueueDriver { + pub fn run(mut self) { + loop { + // Systems increment the `queue_unpark_semaphore` value when they want to guarantee + // that we don't park unless certain things are done. `AssetLoadContext::wait_for_completion` + // wants to ensure that `ParallelQueue::drive` is called, while the cleanup code + // wants to ensure `queue_shutdown_token` is checked. Therefore, we put these two + // operations between reading and waiting on `queue_unpark_semaphore` + let queue_unpark_semaphore_current_value = unsafe { + self.gfx + .device + .get_semaphore_counter_value(self.queue_unpark_semaphore) + .unwrap() + }; + unsafe { self.queue.drive(&self.gfx.device) }; + if self.queue_shutdown_token.is_cancelled() { + break; + } + let semaphore_value = unsafe { + self.queue.park( + &self.gfx.device, + self.queue_unpark_semaphore, + queue_unpark_semaphore_current_value + 1, + ) + }; + let _ = self.queue_watch_sender.send(semaphore_value); + unsafe { self.staging.tick(&self.gfx.device, semaphore_value) }; + } + unsafe { self.queue.drain(&self.gfx.device) }; + unsafe { self.queue.destroy(&self.gfx.device) }; + } } impl Drop for AssetLoader { @@ -131,6 +334,18 @@ impl Drop for AssetLoader { for thread in self.task_executor_threads.drain(..) { thread.join().unwrap(); } + self.queue_shutdown_token.cancel(); + + unsafe { self.queue_unparker.unpark_queue(&self.gfx.device) }; + self.queue_driver_thread.take().unwrap().join().unwrap(); + + unsafe { self.queue_unparker.destroy(&self.gfx.device) }; + + unsafe { + Arc::get_mut(&mut self.staging) + .expect("All threads using staging should now be joined") + .destroy(&self.gfx.device); + } } } @@ -274,7 +489,7 @@ mod tests { impl skid_steer::Source for DummyAssetSource { type Output = DummyAsset; - async fn load(mut self, _context: &Context<'_>) -> Option { + async fn load(mut self, context: &Context<'_>) -> Option { let mut status = DummyAssetStatus { name: self.name.clone(), progress: 0, @@ -282,12 +497,21 @@ mod tests { events: self.events.clone(), }; + // Use the parallel queue and set up an allocation to exercise this functionality + let ctx: &AssetLoadContext = context.get().unwrap(); + let work = unsafe { ctx.begin_work() }; + let finish_time = work.time().get(); + let _alloc = ctx.alloc_staging::(8, 1, finish_time); + while status.progress < 100 { status.progress += self.progress_receiver.recv().await?; self.events .push_event(Event::progress(&self.name, status.progress)); } + work.end(); + ctx.wait_for_completion(finish_time).await; + status.can_cancel = false; // Done loading self.events.push_event(Event::loaded(&self.name)); Some(DummyAsset { diff --git a/client/src/graphics/base.rs b/client/src/graphics/base.rs index ce80478d..c85bdfee 100644 --- a/client/src/graphics/base.rs +++ b/client/src/graphics/base.rs @@ -144,7 +144,8 @@ impl Base { .push_next( &mut vk::PhysicalDeviceVulkan12Features::default() .descriptor_binding_partially_bound(true) - .descriptor_binding_sampled_image_update_after_bind(true), + .descriptor_binding_sampled_image_update_after_bind(true) + .timeline_semaphore(true), ), None, ) From 49846977bb5b390c07c9dd0d45082cf28f727a6f Mon Sep 17 00:00:00 2001 From: Patrick Owen Date: Sun, 16 Aug 2026 16:35:28 -0400 Subject: [PATCH 3/6] Use skid_steer to load png_array --- client/src/graphics/draw.rs | 17 +- client/src/graphics/mod.rs | 2 +- client/src/graphics/png_array.rs | 330 +++++++++++++------------- client/src/graphics/voxels/mod.rs | 8 +- client/src/graphics/voxels/surface.rs | 20 +- 5 files changed, 190 insertions(+), 187 deletions(-) diff --git a/client/src/graphics/draw.rs b/client/src/graphics/draw.rs index b4189159..004155f2 100644 --- a/client/src/graphics/draw.rs +++ b/client/src/graphics/draw.rs @@ -7,6 +7,7 @@ use lahar::Staged; use metrics::histogram; use super::{Base, Fog, Frustum, GltfScene, Meshes, Voxels, fog, voxels}; +use crate::graphics::asset_loader::AssetLoader; use crate::{Asset, Config, Loader, Sim}; use common::SimConfig; use common::proto::{Character, Position}; @@ -53,6 +54,9 @@ pub struct Draw { /// Miscellany character_model: Asset, + + /// Drives async asset loading + asset_loader: AssetLoader, // TODO: Make code more robust by not requiring this to be defined last (due to Drop order) } /// Maximum number of simultaneous frames in flight @@ -128,6 +132,7 @@ impl Draw { .unwrap(); let mut loader = Loader::new(cfg.clone(), gfx.clone()); + let asset_loader = AssetLoader::new(gfx.clone(), cfg.clone()); // Construct the per-frame states let states = cmds @@ -225,6 +230,8 @@ impl Draw { yakui_vulkan, character_model, + + asset_loader, } } } @@ -234,7 +241,7 @@ impl Draw { let voxels = Voxels::new( &self.gfx, self.cfg.clone(), - &mut self.loader, + &self.asset_loader, u32::from(cfg.chunk_size), PIPELINE_DEPTH, ); @@ -473,13 +480,7 @@ impl Draw { // Record the actual rendering commands if let Some(ref mut voxels) = self.voxels { - voxels.draw( - device, - &self.loader, - state.common_ds, - state.voxels.as_ref().unwrap(), - cmd, - ); + voxels.draw(device, state.common_ds, state.voxels.as_ref().unwrap(), cmd); } if let Some(sim) = sim.as_deref() { diff --git a/client/src/graphics/mod.rs b/client/src/graphics/mod.rs index 2d27751a..eb6dda4b 100644 --- a/client/src/graphics/mod.rs +++ b/client/src/graphics/mod.rs @@ -1,6 +1,6 @@ #![allow(clippy::missing_safety_doc)] // Vulkan wrangling is categorically unsafe -#[expect(unused)] mod asset_loader; +mod asset_loader; mod base; mod core; mod draw; diff --git a/client/src/graphics/png_array.rs b/client/src/graphics/png_array.rs index 37bc79f4..bebf9eb3 100644 --- a/client/src/graphics/png_array.rs +++ b/client/src/graphics/png_array.rs @@ -4,182 +4,190 @@ use std::{ path::PathBuf, }; -use anyhow::{Context, anyhow, bail}; +use anyhow::{Context, anyhow, bail, ensure}; use ash::vk; use common::Anonymize; use lahar::DedicatedImage; -use tracing::trace; -use crate::loader::{LoadCtx, LoadFuture, Loadable}; +use crate::graphics::asset_loader::AssetLoadContext; pub struct PngArray { pub path: PathBuf, pub size: usize, } -impl Loadable for PngArray { - type Output = DedicatedImage; - - fn load(self, handle: &LoadCtx) -> LoadFuture<'_, Self::Output> { - Box::pin(async move { - let full_path = handle - .cfg - .find_asset(&self.path) - .ok_or_else(|| anyhow!("{} not found", self.path.anonymize().display()))?; - let mut paths = fs::read_dir(&full_path) - .with_context(|| format!("reading {}", full_path.anonymize().display()))? - .map(|x| x.map(|x| x.path())) - .collect::, _>>() - .with_context(|| format!("reading {}", full_path.anonymize().display()))?; - if paths.is_empty() { - bail!("{} is empty", full_path.anonymize().display()); - } - if paths.len() < self.size { - bail!( - "{}: expected {} textures, found {}", - full_path.anonymize().display(), - self.size, - paths.len() - ); - } - paths.sort(); - paths.truncate(self.size); - let mut dims: Option<(u32, u32)> = None; - let mut mem = None; - for (i, path) in paths.iter().enumerate() { - trace!(layer=i, path=%path.anonymize().display(), "loading"); - let file = File::open(path) - .with_context(|| format!("reading {}", path.anonymize().display()))?; - let decoder = png::Decoder::new(BufReader::new(file)); - let mut reader = decoder - .read_info() - .with_context(|| format!("decoding {}", path.anonymize().display()))?; - let info = reader.info(); - if let Some(dims) = dims { - if dims != (info.width, info.height) { - bail!( - "inconsistent dimensions: expected {}x{}, got {}x{}", - dims.0, - dims.1, - info.width, - info.height - ); - } - } else { - dims = Some((info.width, info.height)); - mem = Some( - handle - .staging - .alloc(info.width as usize * info.height as usize * 4 * self.size) - .await - .ok_or_else(|| { - anyhow!( - "{}: image array too large", - full_path.anonymize().display() - ) - })?, +impl PngArray { + async fn load_inner(self, context: &skid_steer::Context<'_>) -> anyhow::Result { + tracing::trace!("Started loading png array"); + let ctx: &AssetLoadContext = context.get().unwrap(); + let full_path = ctx + .find_asset(&self.path) + .ok_or_else(|| anyhow!("{} not found", self.path.anonymize().display()))?; + let mut paths = fs::read_dir(&full_path) + .with_context(|| format!("reading {}", full_path.anonymize().display()))? + .map(|x| x.map(|x| x.path())) + .collect::, _>>() + .with_context(|| format!("reading {}", full_path.anonymize().display()))?; + if paths.len() < self.size { + bail!( + "{}: expected {} textures, found {}", + full_path.anonymize().display(), + self.size, + paths.len() + ); + } + paths.sort(); + paths.truncate(self.size); + let mut dims: Option<(u32, u32)> = None; + let mut image_data: Vec = Vec::new(); + for (i, path) in paths.iter().enumerate() { + tracing::trace!(layer=i, path=%path.anonymize().display(), "loading"); + let file = File::open(path) + .with_context(|| format!("reading {}", path.anonymize().display()))?; + let decoder = png::Decoder::new(BufReader::new(file)); + let mut reader = decoder + .read_info() + .with_context(|| format!("decoding {}", path.anonymize().display()))?; + let info = reader.info(); + let step_size = info.width as usize * info.height as usize * 4; + ensure!(info.color_type == png::ColorType::Rgba); + ensure!(info.bit_depth == png::BitDepth::Eight); + ensure!(reader.output_buffer_size() == Some(step_size)); + if let Some(dims) = dims { + if dims != (info.width, info.height) { + bail!( + "inconsistent dimensions: expected {}x{}, got {}x{}", + dims.0, + dims.1, + info.width, + info.height ); } - let mem = mem.as_mut().unwrap(); - let step_size = info.width as usize * info.height as usize * 4; - reader - .next_frame(&mut mem[i * step_size..(i + 1) * step_size]) - .with_context(|| format!("decoding {}", path.anonymize().display()))?; + } else { + dims = Some((info.width, info.height)); + image_data.resize(step_size * self.size, 0); } - let (width, height) = dims.unwrap(); - let mem = mem.unwrap(); - unsafe { - let image = DedicatedImage::new( - &handle.gfx.device, - &handle.gfx.memory_properties, - &vk::ImageCreateInfo::default() - .image_type(vk::ImageType::TYPE_2D) - .format(vk::Format::R8G8B8A8_SRGB) - .extent(vk::Extent3D { - width, - height, - depth: 1, - }) - .mip_levels(1) - .array_layers(self.size as u32) - .samples(vk::SampleCountFlags::TYPE_1) - .usage(vk::ImageUsageFlags::SAMPLED | vk::ImageUsageFlags::TRANSFER_DST), - ); + reader + .next_frame(&mut image_data[i * step_size..(i + 1) * step_size]) + .with_context(|| format!("decoding {}", path.anonymize().display()))?; + } + let (width, height) = dims.unwrap(); + unsafe { + let image = DedicatedImage::new( + ctx.device(), + ctx.memory_properties(), + &vk::ImageCreateInfo::default() + .image_type(vk::ImageType::TYPE_2D) + .format(vk::Format::R8G8B8A8_SRGB) + .extent(vk::Extent3D { + width, + height, + depth: 1, + }) + .mip_levels(1) + .array_layers(self.size as u32) + .samples(vk::SampleCountFlags::TYPE_1) + .usage(vk::ImageUsageFlags::SAMPLED | vk::ImageUsageFlags::TRANSFER_DST), + ); - let range = vk::ImageSubresourceRange { - aspect_mask: vk::ImageAspectFlags::COLOR, - base_mip_level: 0, - level_count: 1, - base_array_layer: 0, - layer_count: self.size as u32, - }; - let src = handle.staging.buffer(); - let buffer_offset = mem.offset(); - let dst = image.handle; + let range = vk::ImageSubresourceRange { + aspect_mask: vk::ImageAspectFlags::COLOR, + base_mip_level: 0, + level_count: 1, + base_array_layer: 0, + layer_count: self.size as u32, + }; + let work = ctx.begin_work(); + let finish_time = work.time().get(); + let mem = ctx.alloc_staging(image_data.len(), 4, finish_time); + std::ptr::copy_nonoverlapping( + image_data.as_ptr(), + mem.pointer.as_ptr(), + image_data.len(), + ); + ctx.device().cmd_pipeline_barrier( + work.cmd(), + vk::PipelineStageFlags::TOP_OF_PIPE, + vk::PipelineStageFlags::TRANSFER, + vk::DependencyFlags::default(), + &[], + &[], + &[vk::ImageMemoryBarrier::default() + .dst_access_mask(vk::AccessFlags::TRANSFER_WRITE) + .src_queue_family_index(vk::QUEUE_FAMILY_IGNORED) + .dst_queue_family_index(vk::QUEUE_FAMILY_IGNORED) + .old_layout(vk::ImageLayout::UNDEFINED) + .new_layout(vk::ImageLayout::TRANSFER_DST_OPTIMAL) + .image(image.handle) + .subresource_range(range)], + ); + ctx.device().cmd_copy_buffer_to_image( + work.cmd(), + mem.buffer, + image.handle, + vk::ImageLayout::TRANSFER_DST_OPTIMAL, + &[vk::BufferImageCopy { + buffer_offset: mem.offset, + image_subresource: vk::ImageSubresourceLayers { + aspect_mask: vk::ImageAspectFlags::COLOR, + mip_level: 0, + base_array_layer: 0, + layer_count: range.layer_count, + }, + image_extent: vk::Extent3D { + width, + height, + depth: 1, + }, + ..Default::default() + }], + ); + ctx.device().cmd_pipeline_barrier( + work.cmd(), + vk::PipelineStageFlags::TRANSFER, + vk::PipelineStageFlags::FRAGMENT_SHADER, + vk::DependencyFlags::default(), + &[], + &[], + &[vk::ImageMemoryBarrier::default() + .src_access_mask(vk::AccessFlags::TRANSFER_WRITE) + .dst_access_mask(vk::AccessFlags::SHADER_READ) + .src_queue_family_index(vk::QUEUE_FAMILY_IGNORED) + .dst_queue_family_index(vk::QUEUE_FAMILY_IGNORED) + .old_layout(vk::ImageLayout::TRANSFER_DST_OPTIMAL) + .new_layout(vk::ImageLayout::SHADER_READ_ONLY_OPTIMAL) + .image(image.handle) + .subresource_range(range)], + ); + work.end(); + tracing::trace!("Awaiting parallel queue"); + ctx.wait_for_completion(finish_time).await; + tracing::trace!("Finished awaiting parallel queue"); - handle - .transfer - .run(move |xf, cmd| { - xf.device.cmd_pipeline_barrier( - cmd, - vk::PipelineStageFlags::TOP_OF_PIPE, - vk::PipelineStageFlags::TRANSFER, - vk::DependencyFlags::default(), - &[], - &[], - &[vk::ImageMemoryBarrier::default() - .dst_access_mask(vk::AccessFlags::TRANSFER_WRITE) - .src_queue_family_index(vk::QUEUE_FAMILY_IGNORED) - .dst_queue_family_index(vk::QUEUE_FAMILY_IGNORED) - .old_layout(vk::ImageLayout::UNDEFINED) - .new_layout(vk::ImageLayout::TRANSFER_DST_OPTIMAL) - .image(dst) - .subresource_range(range)], - ); - xf.device.cmd_copy_buffer_to_image( - cmd, - src, - dst, - vk::ImageLayout::TRANSFER_DST_OPTIMAL, - &[vk::BufferImageCopy { - buffer_offset, - image_subresource: vk::ImageSubresourceLayers { - aspect_mask: vk::ImageAspectFlags::COLOR, - mip_level: 0, - base_array_layer: 0, - layer_count: range.layer_count, - }, - image_extent: vk::Extent3D { - width, - height, - depth: 1, - }, - ..Default::default() - }], - ); - xf.stages |= vk::PipelineStageFlags::FRAGMENT_SHADER; - xf.image_barriers.push( - vk::ImageMemoryBarrier::default() - .src_access_mask(vk::AccessFlags::TRANSFER_WRITE) - .dst_access_mask(vk::AccessFlags::SHADER_READ) - .src_queue_family_index(xf.queue_family) - .dst_queue_family_index(xf.dst_queue_family) - .old_layout(vk::ImageLayout::TRANSFER_DST_OPTIMAL) - .new_layout(vk::ImageLayout::SHADER_READ_ONLY_OPTIMAL) - .image(dst) - .subresource_range(range), - ); - }) - .await?; + tracing::trace!( + width = width, + height = height, + path = %full_path.anonymize().display(), + "loaded array" + ); + tracing::trace!("png_array loaded"); + Ok(image) + } + } +} - trace!( - width = width, - height = height, - path = %full_path.anonymize().display(), - "loaded array" - ); - Ok(image) - } - }) +impl skid_steer::Source for PngArray { + type Output = DedicatedImage; + + async fn load(self, context: &skid_steer::Context<'_>) -> Option { + self.load_inner(context) + .await + .inspect_err(|e| tracing::error!("{}", e)) + .ok() + } + + fn free(mut output: Self::Output, context: &skid_steer::Context) { + let ctx: &AssetLoadContext = context.get().unwrap(); + unsafe { output.destroy(ctx.device()) }; } } diff --git a/client/src/graphics/voxels/mod.rs b/client/src/graphics/voxels/mod.rs index 5126c4e1..849bbb6a 100644 --- a/client/src/graphics/voxels/mod.rs +++ b/client/src/graphics/voxels/mod.rs @@ -12,8 +12,8 @@ use metrics::histogram; use tracing::warn; use crate::{ - Config, Loader, Sim, - graphics::{Base, Frustum}, + Config, Sim, + graphics::{Base, Frustum, asset_loader::AssetLoader}, }; use common::{ dodeca::{self, Vertex}, @@ -39,7 +39,7 @@ impl Voxels { pub fn new( gfx: &Base, config: Arc, - loader: &mut Loader, + loader: &AssetLoader, dimension: u32, frames: u32, ) -> Self { @@ -208,7 +208,6 @@ impl Voxels { pub unsafe fn draw( &mut self, device: &Device, - loader: &Loader, common_ds: vk::DescriptorSet, frame: &Frame, cmd: vk::CommandBuffer, @@ -217,7 +216,6 @@ impl Voxels { let started = Instant::now(); if !self.draw.bind( device, - loader, self.surfaces.dimension(), common_ds, &frame.surface, diff --git a/client/src/graphics/voxels/surface.rs b/client/src/graphics/voxels/surface.rs index 54813f65..67db28d7 100644 --- a/client/src/graphics/voxels/surface.rs +++ b/client/src/graphics/voxels/surface.rs @@ -3,7 +3,7 @@ use lahar::{DedicatedImage, DedicatedMapping}; use vk_shader_macros::include_glsl; use super::surface_extraction::DrawBuffer; -use crate::{Asset, Loader, graphics::Base}; +use crate::graphics::{Base, asset_loader::AssetLoader}; use common::{defer, world::Material}; const VERT: &[u32] = include_glsl!("shaders/voxels.vert"); @@ -15,12 +15,12 @@ pub struct Surface { pipeline: vk::Pipeline, descriptor_pool: vk::DescriptorPool, ds: vk::DescriptorSet, - colors: Asset, + colors: skid_steer::Asset, colors_view: vk::ImageView, } impl Surface { - pub fn new(gfx: &Base, loader: &mut Loader, buffer: &DrawBuffer) -> Self { + pub fn new(gfx: &Base, loader: &AssetLoader, buffer: &DrawBuffer) -> Self { let device = &*gfx.device; unsafe { // Construct the shader modules @@ -224,13 +224,10 @@ impl Surface { v_guard.invoke(); f_guard.invoke(); - let colors = loader.load( - "voxel materials", - crate::graphics::PngArray { - path: "materials".into(), - size: common::world::Material::COUNT - 1, - }, - ); + let colors = loader.load(crate::graphics::PngArray { + path: "materials".into(), + size: common::world::Material::COUNT - 1, + }); Self { static_ds_layout, @@ -247,7 +244,6 @@ impl Surface { pub unsafe fn bind( &mut self, device: &Device, - loader: &Loader, dimension: u32, common_ds: vk::DescriptorSet, frame: &Frame, @@ -255,7 +251,7 @@ impl Surface { ) -> bool { unsafe { if self.colors_view == vk::ImageView::null() { - if let Some(colors) = loader.get(self.colors) { + if let Some(colors) = self.colors.try_get() { self.colors_view = device .create_image_view( &vk::ImageViewCreateInfo::default() From 97bcbfbe71d26204de8410f56b0fa79c34a2e4e8 Mon Sep 17 00:00:00 2001 From: Patrick Owen Date: Sun, 16 Aug 2026 22:19:57 -0400 Subject: [PATCH 4/6] Refactor: Gather global data sent to shaders into a single struct --- client/src/graphics/base.rs | 53 ++---------- client/src/graphics/draw.rs | 10 ++- client/src/graphics/fog.rs | 3 +- client/src/graphics/gltf_mesh.rs | 14 ++-- client/src/graphics/meshes.rs | 8 +- client/src/graphics/mod.rs | 1 + client/src/graphics/shader_data.rs | 113 ++++++++++++++++++++++++++ client/src/graphics/voxels/surface.rs | 4 +- client/src/loader.rs | 50 ------------ 9 files changed, 144 insertions(+), 112 deletions(-) create mode 100644 client/src/graphics/shader_data.rs diff --git a/client/src/graphics/base.rs b/client/src/graphics/base.rs index c85bdfee..ff98caa0 100644 --- a/client/src/graphics/base.rs +++ b/client/src/graphics/base.rs @@ -10,6 +10,8 @@ use tracing::{error, info, trace, warn}; use ash::{Device, vk}; +use crate::graphics::shader_data::ShaderData; + use super::Core; /// Vulkan resources shared between many parts of the renderer @@ -29,10 +31,8 @@ pub struct Base { pub pipeline_cache: vk::PipelineCache, /// Context in which the main rendering work occurs pub render_pass: vk::RenderPass, - /// A reasonable general-purpose texture sampler - pub linear_sampler: vk::Sampler, - /// Layout of common shader resources, such as the common uniform buffer - pub common_layout: vk::DescriptorSetLayout, + /// Holds globally-managed memory for data that is sent to shaders + pub shader_data: ShaderData, pub limits: vk::PhysicalDeviceLimits, pub timestamp_bits: u32, pipeline_cache_path: Option, @@ -48,9 +48,7 @@ impl Drop for Base { self.device .destroy_pipeline_cache(self.pipeline_cache, None); self.device.destroy_render_pass(self.render_pass, None); - self.device.destroy_sampler(self.linear_sampler, None); - self.device - .destroy_descriptor_set_layout(self.common_layout, None); + self.shader_data.destroy(&self.device); self.device.destroy_device(None); } } @@ -235,43 +233,7 @@ impl Base { ) .unwrap(); - let linear_sampler = device - .create_sampler( - &vk::SamplerCreateInfo::default() - .min_filter(vk::Filter::LINEAR) - .mag_filter(vk::Filter::LINEAR) - .mipmap_mode(vk::SamplerMipmapMode::NEAREST) - .address_mode_u(vk::SamplerAddressMode::CLAMP_TO_EDGE) - .address_mode_v(vk::SamplerAddressMode::CLAMP_TO_EDGE) - .address_mode_w(vk::SamplerAddressMode::CLAMP_TO_EDGE), - None, - ) - .unwrap(); - - let common_layout = device - .create_descriptor_set_layout( - &vk::DescriptorSetLayoutCreateInfo::default().bindings(&[ - // Uniforms - vk::DescriptorSetLayoutBinding { - binding: 0, - descriptor_type: vk::DescriptorType::UNIFORM_BUFFER, - descriptor_count: 1, - stage_flags: vk::ShaderStageFlags::VERTEX - | vk::ShaderStageFlags::FRAGMENT, - ..Default::default() - }, - // Depth buffer - vk::DescriptorSetLayoutBinding { - binding: 1, - descriptor_type: vk::DescriptorType::INPUT_ATTACHMENT, - descriptor_count: 1, - stage_flags: vk::ShaderStageFlags::FRAGMENT, - ..Default::default() - }, - ]), - None, - ) - .unwrap(); + let shader_data = ShaderData::new(&device, &memory_properties); let debug_utils = core .debug_utils .as_ref() @@ -286,8 +248,7 @@ impl Base { memory_properties, pipeline_cache, render_pass, - linear_sampler, - common_layout, + shader_data, pipeline_cache_path, limits: physical_properties.properties.limits, timestamp_bits: queue_family_properties.timestamp_valid_bits, diff --git a/client/src/graphics/draw.rs b/client/src/graphics/draw.rs index 004155f2..e17d70f9 100644 --- a/client/src/graphics/draw.rs +++ b/client/src/graphics/draw.rs @@ -99,7 +99,8 @@ impl Draw { let common_pipeline_layout = device .create_pipeline_layout( - &vk::PipelineLayoutCreateInfo::default().set_layouts(&[gfx.common_layout]), + &vk::PipelineLayoutCreateInfo::default() + .set_layouts(&[gfx.shader_data.common_layout]), None, ) .unwrap(); @@ -127,7 +128,10 @@ impl Draw { .allocate_descriptor_sets( &vk::DescriptorSetAllocateInfo::default() .descriptor_pool(common_descriptor_pool) - .set_layouts(&vec![gfx.common_layout; PIPELINE_DEPTH as usize]), + .set_layouts(&vec![ + gfx.shader_data.common_layout; + PIPELINE_DEPTH as usize + ]), ) .unwrap(); @@ -183,7 +187,7 @@ impl Draw { }) .collect(); - let meshes = Meshes::new(&gfx, loader.ctx().mesh_ds_layout); + let meshes = Meshes::new(&gfx); let fog = Fog::new(&gfx); diff --git a/client/src/graphics/fog.rs b/client/src/graphics/fog.rs index ea14689c..786fdf25 100644 --- a/client/src/graphics/fog.rs +++ b/client/src/graphics/fog.rs @@ -31,7 +31,8 @@ impl Fog { // Define the outward-facing interface of the shaders, incl. uniforms, samplers, etc. let pipeline_layout = device .create_pipeline_layout( - &vk::PipelineLayoutCreateInfo::default().set_layouts(&[gfx.common_layout]), + &vk::PipelineLayoutCreateInfo::default() + .set_layouts(&[gfx.shader_data.common_layout]), None, ) .unwrap(); diff --git a/client/src/graphics/gltf_mesh.rs b/client/src/graphics/gltf_mesh.rs index cf77878c..14ae0103 100644 --- a/client/src/graphics/gltf_mesh.rs +++ b/client/src/graphics/gltf_mesh.rs @@ -174,7 +174,7 @@ async fn load_primitive( .allocate_descriptor_sets( &vk::DescriptorSetAllocateInfo::default() .descriptor_pool(pool) - .set_layouts(&[ctx.mesh_ds_layout]), + .set_layouts(&[ctx.gfx.shader_data.mesh_ds_layout]), ) .unwrap()[0]; device.update_descriptor_sets( @@ -293,11 +293,11 @@ async fn load_geom( storage.copy_from_slice(&idx.to_ne_bytes()); } - let vert_alloc = - ctx.vertex_alloc - .lock() - .unwrap() - .alloc(&ctx.gfx.device, byte_size as vk::DeviceSize, 4); + let vert_alloc = ctx.gfx.shader_data.vertex_alloc.lock().unwrap().alloc( + &ctx.gfx.device, + byte_size as vk::DeviceSize, + 4, + ); let staging_buffer = ctx.staging.buffer(); let vert_buffer = vert_alloc.buffer; let vert_src_offset = v_staging.offset(); @@ -328,7 +328,7 @@ async fn load_geom( }) }; - let idx_alloc = ctx.index_alloc.lock().unwrap().alloc( + let idx_alloc = ctx.gfx.shader_data.index_alloc.lock().unwrap().alloc( &ctx.gfx.device, index_count as vk::DeviceSize * 4, 4, diff --git a/client/src/graphics/meshes.rs b/client/src/graphics/meshes.rs index 8dd5cc60..8c91c15c 100644 --- a/client/src/graphics/meshes.rs +++ b/client/src/graphics/meshes.rs @@ -17,8 +17,7 @@ pub struct Meshes { } impl Meshes { - #[allow(clippy::unneeded_field_pattern)] // Silence offset_of warnings nonsense - pub fn new(gfx: &Base, ds_layout: vk::DescriptorSetLayout) -> Self { + pub fn new(gfx: &Base) -> Self { let device = &*gfx.device; unsafe { // Construct the shader modules @@ -37,7 +36,10 @@ impl Meshes { let pipeline_layout = device .create_pipeline_layout( &vk::PipelineLayoutCreateInfo::default() - .set_layouts(&[gfx.common_layout, ds_layout]) + .set_layouts(&[ + gfx.shader_data.common_layout, + gfx.shader_data.mesh_ds_layout, + ]) .push_constant_ranges(&[vk::PushConstantRange { stage_flags: vk::ShaderStageFlags::VERTEX, offset: 0, diff --git a/client/src/graphics/mod.rs b/client/src/graphics/mod.rs index eb6dda4b..2d79cb01 100644 --- a/client/src/graphics/mod.rs +++ b/client/src/graphics/mod.rs @@ -10,6 +10,7 @@ mod gltf_mesh; mod gui; mod meshes; mod png_array; +pub mod shader_data; pub mod voxels; mod window; diff --git a/client/src/graphics/shader_data.rs b/client/src/graphics/shader_data.rs new file mode 100644 index 00000000..df0b35f4 --- /dev/null +++ b/client/src/graphics/shader_data.rs @@ -0,0 +1,113 @@ +use std::sync::Mutex; + +use ash::vk; +use lahar::BufferRegion; + +pub struct ShaderData { + pub vertex_alloc: Mutex, + pub index_alloc: Mutex, + /// A reasonable general-purpose texture sampler + pub linear_sampler: vk::Sampler, + pub mesh_ds_layout: vk::DescriptorSetLayout, + /// Layout of common shader resources, such as the common uniform buffer + pub common_layout: vk::DescriptorSetLayout, +} + +impl ShaderData { + pub fn new( + device: &ash::Device, + memory_properties: &vk::PhysicalDeviceMemoryProperties, + ) -> Self { + let vertex_alloc = unsafe { + BufferRegion::new( + device, + memory_properties, + 16 * 1024 * 1024, + vk::BufferUsageFlags::TRANSFER_DST | vk::BufferUsageFlags::VERTEX_BUFFER, + ) + }; + let index_alloc = unsafe { + BufferRegion::new( + device, + memory_properties, + 16 * 1024 * 1024, + vk::BufferUsageFlags::TRANSFER_DST | vk::BufferUsageFlags::INDEX_BUFFER, + ) + }; + let linear_sampler = unsafe { + device + .create_sampler( + &vk::SamplerCreateInfo::default() + .min_filter(vk::Filter::LINEAR) + .mag_filter(vk::Filter::LINEAR) + .mipmap_mode(vk::SamplerMipmapMode::NEAREST) + .address_mode_u(vk::SamplerAddressMode::CLAMP_TO_EDGE) + .address_mode_v(vk::SamplerAddressMode::CLAMP_TO_EDGE) + .address_mode_w(vk::SamplerAddressMode::CLAMP_TO_EDGE), + None, + ) + .unwrap() + }; + let mesh_ds_layout = unsafe { + device + .create_descriptor_set_layout( + &vk::DescriptorSetLayoutCreateInfo::default().bindings(&[ + vk::DescriptorSetLayoutBinding { + binding: 0, + descriptor_type: vk::DescriptorType::COMBINED_IMAGE_SAMPLER, + descriptor_count: 1, + stage_flags: vk::ShaderStageFlags::FRAGMENT, + p_immutable_samplers: &linear_sampler, + ..vk::DescriptorSetLayoutBinding::default() + }, + ]), + None, + ) + .unwrap() + }; + let common_layout = unsafe { + device + .create_descriptor_set_layout( + &vk::DescriptorSetLayoutCreateInfo::default().bindings(&[ + // Uniforms + vk::DescriptorSetLayoutBinding { + binding: 0, + descriptor_type: vk::DescriptorType::UNIFORM_BUFFER, + descriptor_count: 1, + stage_flags: vk::ShaderStageFlags::VERTEX + | vk::ShaderStageFlags::FRAGMENT, + ..Default::default() + }, + // Depth buffer + vk::DescriptorSetLayoutBinding { + binding: 1, + descriptor_type: vk::DescriptorType::INPUT_ATTACHMENT, + descriptor_count: 1, + stage_flags: vk::ShaderStageFlags::FRAGMENT, + ..Default::default() + }, + ]), + None, + ) + .unwrap() + }; + + ShaderData { + vertex_alloc: Mutex::new(vertex_alloc), + index_alloc: Mutex::new(index_alloc), + linear_sampler, + mesh_ds_layout, + common_layout, + } + } + + pub unsafe fn destroy(&mut self, device: &ash::Device) { + unsafe { + device.destroy_descriptor_set_layout(self.common_layout, None); + device.destroy_descriptor_set_layout(self.mesh_ds_layout, None); + device.destroy_sampler(self.linear_sampler, None); + self.index_alloc.lock().unwrap().destroy(device); + self.vertex_alloc.lock().unwrap().destroy(device); + } + } +} diff --git a/client/src/graphics/voxels/surface.rs b/client/src/graphics/voxels/surface.rs index 67db28d7..0efe9266 100644 --- a/client/src/graphics/voxels/surface.rs +++ b/client/src/graphics/voxels/surface.rs @@ -50,7 +50,7 @@ impl Surface { descriptor_type: vk::DescriptorType::COMBINED_IMAGE_SAMPLER, descriptor_count: 1, stage_flags: vk::ShaderStageFlags::FRAGMENT, - p_immutable_samplers: &gfx.linear_sampler, + p_immutable_samplers: &gfx.shader_data.linear_sampler, ..Default::default() }, ]), @@ -99,7 +99,7 @@ impl Surface { let pipeline_layout = device .create_pipeline_layout( &vk::PipelineLayoutCreateInfo::default() - .set_layouts(&[gfx.common_layout, static_ds_layout]) + .set_layouts(&[gfx.shader_data.common_layout, static_ds_layout]) .push_constant_ranges(&[vk::PushConstantRange { stage_flags: vk::ShaderStageFlags::VERTEX, offset: 0, diff --git a/client/src/loader.rs b/client/src/loader.rs index 91ef0762..f56493b7 100644 --- a/client/src/loader.rs +++ b/client/src/loader.rs @@ -63,39 +63,6 @@ impl Loader { let (transfer, reactor) = unsafe { transfer::Reactor::new(gfx.device.clone(), gfx.queue_family, gfx.queue, None) }; - let vertex_alloc = unsafe { - BufferRegion::new( - &gfx.device, - &gfx.memory_properties, - 16 * 1024 * 1024, - vk::BufferUsageFlags::TRANSFER_DST | vk::BufferUsageFlags::VERTEX_BUFFER, - ) - }; - let index_alloc = unsafe { - BufferRegion::new( - &gfx.device, - &gfx.memory_properties, - 16 * 1024 * 1024, - vk::BufferUsageFlags::TRANSFER_DST | vk::BufferUsageFlags::INDEX_BUFFER, - ) - }; - let mesh_ds_layout = unsafe { - gfx.device - .create_descriptor_set_layout( - &vk::DescriptorSetLayoutCreateInfo::default().bindings(&[ - vk::DescriptorSetLayoutBinding { - binding: 0, - descriptor_type: vk::DescriptorType::COMBINED_IMAGE_SAMPLER, - descriptor_count: 1, - stage_flags: vk::ShaderStageFlags::FRAGMENT, - p_immutable_samplers: &gfx.linear_sampler, - ..vk::DescriptorSetLayoutBinding::default() - }, - ]), - None, - ) - .unwrap() - }; let shared = Arc::new(Shared { send, ctx: LoadCtx { @@ -103,9 +70,6 @@ impl Loader { gfx, staging, transfer, - vertex_alloc: Mutex::new(vertex_alloc), - index_alloc: Mutex::new(index_alloc), - mesh_ds_layout, }, }); Self { @@ -236,9 +200,6 @@ pub struct LoadCtx { pub gfx: Arc, pub staging: StagingBuffer, pub transfer: TransferHandle, - pub vertex_alloc: Mutex, - pub index_alloc: Mutex, - pub mesh_ds_layout: vk::DescriptorSetLayout, } impl LoadCtx { @@ -247,17 +208,6 @@ impl LoadCtx { } } -impl Drop for LoadCtx { - fn drop(&mut self) { - let device = &*self.gfx.device; - unsafe { - self.index_alloc.lock().unwrap().destroy(device); - self.vertex_alloc.lock().unwrap().destroy(device); - device.destroy_descriptor_set_layout(self.mesh_ds_layout, None); - } - } -} - trait AnyTable: Downcast { fn finish(&mut self, index: u32, value: Box); fn cleanup(self: Box, gfx: &Base); From f203fd71573e6e91e50b55a11770c189a75b68d1 Mon Sep 17 00:00:00 2001 From: Patrick Owen Date: Sun, 16 Aug 2026 22:25:49 -0400 Subject: [PATCH 5/6] Use skid_steer to load meshes --- Cargo.lock | 7 + client/Cargo.toml | 1 + client/src/graphics/asset_loader.rs | 27 +- client/src/graphics/draw.rs | 15 +- client/src/graphics/gltf_mesh.rs | 449 +++++----------------------- client/src/graphics/meshes.rs | 315 ++++++++++++++++++- 6 files changed, 419 insertions(+), 395 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 3d80633e..873df2ca 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -427,6 +427,7 @@ dependencies = [ "ash", "ash-window", "bencher", + "color", "common", "directories", "downcast-rs 2.0.2", @@ -509,6 +510,12 @@ dependencies = [ "objc", ] +[[package]] +name = "color" +version = "0.3.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2ec7c5eb7a16992b1904d76c517d170ab353b0e0b3d5a0c81a8a0cd1037893cf" + [[package]] name = "combine" version = "4.6.7" diff --git a/client/Cargo.toml b/client/Cargo.toml index 4a1f9fc0..e1f0aef4 100644 --- a/client/Cargo.toml +++ b/client/Cargo.toml @@ -43,6 +43,7 @@ hdrhistogram = { version = "7", default-features = false } save = { path = "../save" } lru-slab = "0.1.2" skid-steer = { git="https://github.com/Ralith/skid-steer", rev = "16feada2408810f78a9f469001ef9277e98d4061" } +color = "0.3.3" [features] default = ["use-repo-assets"] diff --git a/client/src/graphics/asset_loader.rs b/client/src/graphics/asset_loader.rs index f331e705..306f6c4e 100644 --- a/client/src/graphics/asset_loader.rs +++ b/client/src/graphics/asset_loader.rs @@ -7,11 +7,14 @@ use std::{ }; use ash::vk; -use lahar::{GrowableRing, ParallelQueue, parallel_queue}; +use lahar::{BufferRegionAlloc, GrowableRing, ParallelQueue, parallel_queue}; use skid_steer::Context; use tokio_util::sync::CancellationToken; -use crate::{Config, graphics::Base}; +use crate::{ + Config, + graphics::{Base, meshes, shader_data::ShaderData}, +}; /// Contains all the dependencies necessary to load assets. pub struct AssetLoadContext { @@ -53,6 +56,26 @@ impl AssetLoadContext { } } + pub fn alloc_vertices(&self, num_vertices: usize) -> BufferRegionAlloc { + self.gfx.shader_data.vertex_alloc.lock().unwrap().alloc( + &self.gfx.device, + (size_of::() * num_vertices) as vk::DeviceSize, + 4, + ) + } + + pub fn alloc_indices(&self, num_indices: usize) -> BufferRegionAlloc { + self.gfx.shader_data.index_alloc.lock().unwrap().alloc( + &self.gfx.device, + (size_of::() * num_indices) as vk::DeviceSize, + 4, + ) + } + + pub fn shader_data(&self) -> &ShaderData { + &self.gfx.shader_data + } + pub fn device(&self) -> &ash::Device { self.gfx.device.as_ref() } diff --git a/client/src/graphics/draw.rs b/client/src/graphics/draw.rs index e17d70f9..d711baf3 100644 --- a/client/src/graphics/draw.rs +++ b/client/src/graphics/draw.rs @@ -8,7 +8,7 @@ use metrics::histogram; use super::{Base, Fog, Frustum, GltfScene, Meshes, Voxels, fog, voxels}; use crate::graphics::asset_loader::AssetLoader; -use crate::{Asset, Config, Loader, Sim}; +use crate::{Config, Loader, Sim}; use common::SimConfig; use common::proto::{Character, Position}; @@ -53,7 +53,7 @@ pub struct Draw { yakui_vulkan: yakui_vulkan::YakuiVulkan, /// Miscellany - character_model: Asset, + character_model: skid_steer::Asset, /// Drives async asset loading asset_loader: AssetLoader, // TODO: Make code more robust by not requiring this to be defined last (due to Drop order) @@ -204,12 +204,9 @@ impl Draw { yakui_vulkan.transfers_submitted(); } - let character_model = loader.load( - "character model", - super::GlbFile { - path: "character.glb".into(), - }, - ); + let character_model = asset_loader.load(super::GlbFile { + path: "character.glb".into(), + }); Self { gfx, @@ -498,7 +495,7 @@ impl Draw { .world .get::<&Position>(entity) .expect("positionless entity in graph"); - if let Some(character_model) = self.loader.get(self.character_model) + if let Some(character_model) = self.character_model.try_get() && let Ok(ch) = sim.world.get::<&Character>(entity) { let transform = na::Matrix4::from(transform * pos.local) diff --git a/client/src/graphics/gltf_mesh.rs b/client/src/graphics/gltf_mesh.rs index 14ae0103..0fcf93ce 100644 --- a/client/src/graphics/gltf_mesh.rs +++ b/client/src/graphics/gltf_mesh.rs @@ -2,37 +2,47 @@ use std::{ borrow::Cow, fs::{self, File}, io::Cursor, - mem, path::{Path, PathBuf}, - ptr, }; use anyhow::{Context, Result, anyhow, bail}; -use ash::vk; +use color::{AlphaColor, LinearSrgb}; use common::Anonymize; -use futures_util::future::{BoxFuture, FutureExt, try_join_all}; -use lahar::{BufferRegionAlloc, DedicatedImage}; +use futures_util::future::{FutureExt, LocalBoxFuture, try_join_all}; use tracing::{error, trace}; -use super::{Base, Mesh, meshes::Vertex}; -use crate::loader::{Cleanup, LoadCtx, LoadFuture, Loadable}; +use super::{Mesh, meshes::Vertex}; +use crate::graphics::{ + asset_loader::AssetLoadContext, + meshes::{MeshGeometryDefinition, MeshMaterialDefinition}, +}; pub struct GlbFile { pub path: PathBuf, } -impl Loadable for GlbFile { +impl skid_steer::Source for GlbFile { type Output = GltfScene; - fn load(self, ctx: &LoadCtx) -> LoadFuture<'_, Self::Output> { - Box::pin(self.load(ctx)) + async fn load<'a>(self, context: &'a skid_steer::Context<'a>) -> Option { + let ctx: &AssetLoadContext = context.get().unwrap(); + self.load_inner(ctx) + .await + .inspect_err(|e| tracing::error!("{}", e)) + .ok() + } + + fn free(mut output: Self::Output, context: &skid_steer::Context) { + let ctx: &AssetLoadContext = context.get().unwrap(); + unsafe { + output.destroy(ctx.device()); + } } } impl GlbFile { - async fn load(self, ctx: &LoadCtx) -> Result { + async fn load_inner(self, ctx: &AssetLoadContext) -> Result { let path = ctx - .cfg .find_asset(&self.path) .ok_or_else(|| anyhow!("{} not found", self.path.anonymize().display()))?; @@ -68,22 +78,22 @@ impl GlbFile { pub struct GltfScene(pub Vec); -impl Cleanup for GltfScene { - unsafe fn cleanup(self, gfx: &Base) { +impl GltfScene { + unsafe fn destroy(&mut self, device: &ash::Device) { unsafe { - for mesh in self.0 { - mesh.cleanup(gfx); + for mesh in &mut self.0 { + mesh.destroy(device); } } } } fn load_node<'a>( - ctx: &'a LoadCtx, + ctx: &'a AssetLoadContext, buffer: &'a [u8], transform: &'a na::Matrix4, node: gltf::Node<'a>, -) -> BoxFuture<'a, Result>> { +) -> LocalBoxFuture<'a, Result>> { async move { let transform = transform * na::Matrix4::from(node.transform().matrix()); let (mut local, children) = tokio::try_join!( @@ -104,11 +114,11 @@ fn load_node<'a>( Ok(local) } - .boxed() + .boxed_local() } async fn load_mesh( - ctx: &LoadCtx, + ctx: &AssetLoadContext, buffer: &[u8], transform: &na::Matrix4, mesh: &gltf::Mesh<'_>, @@ -121,12 +131,11 @@ async fn load_mesh( } async fn load_primitive( - ctx: &LoadCtx, + ctx: &AssetLoadContext, buffer: &[u8], transform: &na::Matrix4, prim: gltf::Primitive<'_>, ) -> Result { - let device = &*ctx.gfx.device; let texcoord_index = prim .material() .pbr_metallic_roughness() @@ -136,85 +145,20 @@ async fn load_primitive( // Concurrent upload // TODO: Don't leak resources on error let (geom, color) = tokio::join!( - load_geom(ctx, buffer, &prim, transform, texcoord_index), + load_geom(buffer, &prim, transform, texcoord_index), load_material(ctx, buffer, &prim) ); let geom = geom?; let color = color?; - - unsafe { - let color_view = device - .create_image_view( - &vk::ImageViewCreateInfo::default() - .image(color.handle) - .view_type(vk::ImageViewType::TYPE_2D) - .format(vk::Format::R8G8B8A8_SRGB) - .subresource_range(vk::ImageSubresourceRange { - aspect_mask: vk::ImageAspectFlags::COLOR, - base_mip_level: 0, - level_count: 1, - base_array_layer: 0, - layer_count: 1, - }), - None, - ) - .unwrap(); - let pool = device - .create_descriptor_pool( - &vk::DescriptorPoolCreateInfo::default() - .max_sets(1) - .pool_sizes(&[vk::DescriptorPoolSize { - ty: vk::DescriptorType::COMBINED_IMAGE_SAMPLER, - descriptor_count: 1, - }]), - None, - ) - .unwrap(); - let ds = device - .allocate_descriptor_sets( - &vk::DescriptorSetAllocateInfo::default() - .descriptor_pool(pool) - .set_layouts(&[ctx.gfx.shader_data.mesh_ds_layout]), - ) - .unwrap()[0]; - device.update_descriptor_sets( - &[vk::WriteDescriptorSet::default() - .dst_set(ds) - .dst_binding(0) - .descriptor_type(vk::DescriptorType::COMBINED_IMAGE_SAMPLER) - .image_info(&[vk::DescriptorImageInfo { - sampler: vk::Sampler::null(), - image_view: color_view, - image_layout: vk::ImageLayout::SHADER_READ_ONLY_OPTIMAL, - }])], - &[], - ); - - Ok(Mesh { - vertices: geom.vertices, - indices: geom.indices, - index_count: geom.index_count, - pool, - ds, - color, - color_view, - }) - } -} - -struct Geometry { - vertices: BufferRegionAlloc, - indices: BufferRegionAlloc, - index_count: u32, + Ok(Mesh::from_definition(ctx, geom, color).await) } async fn load_geom( - ctx: &LoadCtx, buffer: &[u8], prim: &gltf::Primitive<'_>, transform: &na::Matrix4, texcoord_index: Option, -) -> Result { +) -> Result { let normal_transform = match transform.try_inverse() { None => { error!("non-invertible transform"); @@ -233,7 +177,7 @@ async fn load_geom( let positions = prim .read_positions() .ok_or_else(|| anyhow!("vertex positions missing"))?; - let mut texcoords = texcoord_index + let texcoords = texcoord_index .map(|i| -> Result<_> { Ok(prim .read_tex_coords(i) @@ -249,153 +193,66 @@ async fn load_geom( { bail!("inconsistent vertex attribute counts"); } - let byte_size = vertex_count * mem::size_of::(); - - let mut v_staging = ctx - .staging - .alloc(byte_size) - .await - .ok_or_else(|| anyhow!("too large"))?; - for ((pos, norm), storage) in positions + let vertices: Vec<_> = positions .zip(normals) - .zip(v_staging.as_chunks_mut::<{ mem::size_of::() }>().0) - { - let v = Vertex { + .zip( + texcoords + .into_iter() + .flatten() + .chain(std::iter::repeat([0.0, 0.0])), + ) + .map(|((position, normal), texcoords)| Vertex { position: na::Point3::from_homogeneous( - transform * (na::Point3::from(pos)).to_homogeneous(), + transform * (na::Point3::from(position)).to_homogeneous(), ) .unwrap_or_else(na::Point3::origin), - texcoords: texcoords - .as_mut() - .map_or_else(na::zero, |x| x.next().unwrap().into()), normal: na::Unit::new_normalize( - (normal_transform * na::Vector3::from(norm).to_homogeneous()).xyz(), + (normal_transform * na::Vector3::from(normal).to_homogeneous()).xyz(), ), - }; - // write_unaligned accepts misaligned pointers - #[allow(clippy::cast_ptr_alignment)] - unsafe { - ptr::write_unaligned(storage.as_ptr() as *mut Vertex, v); - } - } - - let indices = prim + texcoords: texcoords.into(), + }) + .collect(); + let indices: Vec<_> = prim .read_indices() .ok_or_else(|| anyhow!("indices missing"))? - .into_u32(); - let index_count = indices.len(); - let mut i_staging = ctx - .staging - .alloc(index_count * 4) - .await - .ok_or_else(|| anyhow!("too large"))?; - for (idx, storage) in indices.zip(i_staging.as_chunks_mut::<4>().0) { - storage.copy_from_slice(&idx.to_ne_bytes()); - } - - let vert_alloc = ctx.gfx.shader_data.vertex_alloc.lock().unwrap().alloc( - &ctx.gfx.device, - byte_size as vk::DeviceSize, - 4, - ); - let staging_buffer = ctx.staging.buffer(); - let vert_buffer = vert_alloc.buffer; - let vert_src_offset = v_staging.offset(); - let vert_dst_offset = vert_alloc.offset; - let vertex_upload = unsafe { - ctx.transfer.run(move |xf, cmd| { - xf.device.cmd_copy_buffer( - cmd, - staging_buffer, - vert_buffer, - &[vk::BufferCopy { - src_offset: vert_src_offset, - dst_offset: vert_dst_offset, - size: byte_size as vk::DeviceSize, - }], - ); - xf.stages |= vk::PipelineStageFlags::VERTEX_INPUT; - xf.buffer_barriers.push( - vk::BufferMemoryBarrier::default() - .src_access_mask(vk::AccessFlags::TRANSFER_WRITE) - .dst_access_mask(vk::AccessFlags::VERTEX_ATTRIBUTE_READ) - .src_queue_family_index(xf.queue_family) - .dst_queue_family_index(xf.dst_queue_family) - .buffer(vert_buffer) - .offset(vert_dst_offset) - .size(byte_size as vk::DeviceSize), - ); - }) - }; - - let idx_alloc = ctx.gfx.shader_data.index_alloc.lock().unwrap().alloc( - &ctx.gfx.device, - index_count as vk::DeviceSize * 4, - 4, - ); - let idx_buffer = idx_alloc.buffer; - let idx_src_offset = i_staging.offset(); - let idx_dst_offset = idx_alloc.offset; - let index_upload = unsafe { - ctx.transfer.run(move |xf, cmd| { - xf.device.cmd_copy_buffer( - cmd, - staging_buffer, - idx_buffer, - &[vk::BufferCopy { - src_offset: idx_src_offset, - dst_offset: idx_dst_offset, - size: index_count as vk::DeviceSize * 4, - }], - ); - xf.stages |= vk::PipelineStageFlags::VERTEX_INPUT; - xf.buffer_barriers.push( - vk::BufferMemoryBarrier::default() - .src_access_mask(vk::AccessFlags::TRANSFER_WRITE) - .dst_access_mask(vk::AccessFlags::INDEX_READ) - .src_queue_family_index(xf.queue_family) - .dst_queue_family_index(xf.dst_queue_family) - .buffer(idx_buffer) - .offset(idx_dst_offset) - .size(index_count as vk::DeviceSize * 4), - ); - }) - }; - // Upload concurrently - let (r1, r2) = tokio::join!(vertex_upload, index_upload); - r1?; - r2?; - Ok(Geometry { - vertices: vert_alloc, - indices: idx_alloc, - index_count: index_count as u32, - }) + .into_u32() + .collect(); + Ok(MeshGeometryDefinition { vertices, indices }) } async fn load_material( - ctx: &LoadCtx, + ctx: &AssetLoadContext, buffer: &[u8], prim: &gltf::Primitive<'_>, -) -> Result { - let device = &*ctx.gfx.device; +) -> Result { let color = match prim .material() .pbr_metallic_roughness() .base_color_texture() { None => { - return load_solid_color( - ctx, - prim.material().pbr_metallic_roughness().base_color_factor(), - ) - .await; + return Ok(MeshMaterialDefinition { + width: 1, + height: 1, + srgb_rgba_color_data: AlphaColor::::new( + prim.material().pbr_metallic_roughness().base_color_factor(), + ) + .to_rgba8() + .to_u8_array() + .to_vec(), + }); } Some(x) => x, }; + if prim.material().pbr_metallic_roughness().base_color_factor() != [1.0, 1.0, 1.0, 1.0] { + tracing::warn!( + "Ignoring base color factor {:?}, as this setting is currently only supported for GLTF materials without color textures.", + prim.material().pbr_metallic_roughness().base_color_factor() + ); + } let color_data = match color.texture().source().source() { gltf::image::Source::Uri { uri, .. } => { let path = ctx - .cfg .find_asset(Path::new(uri)) .ok_or_else(|| anyhow!("texture {} not found", uri))?; trace!(path = %path.anonymize().display(), "reading texture"); @@ -419,165 +276,13 @@ async fn load_material( let info = color_reader.info(); (info.width, info.height) }; - let mut color_staging = ctx - .staging - .alloc(width as usize * height as usize * 4) - .await - .ok_or_else(|| anyhow!("texture too large"))?; + let mut image_data = vec![0; width as usize * height as usize * 4]; color_reader - .next_frame(&mut color_staging) + .next_frame(&mut image_data) .with_context(|| "decoding PNG data")?; - let color = unsafe { - DedicatedImage::new( - device, - &ctx.gfx.memory_properties, - &vk::ImageCreateInfo::default() - .image_type(vk::ImageType::TYPE_2D) - .format(vk::Format::R8G8B8A8_SRGB) - .extent(vk::Extent3D { - width, - height, - depth: 1, - }) - .mip_levels(1) - .array_layers(1) - .samples(vk::SampleCountFlags::TYPE_1) - .usage(vk::ImageUsageFlags::SAMPLED | vk::ImageUsageFlags::TRANSFER_DST), - ) - }; - let staging_buffer = ctx.staging.buffer(); - let color_handle = color.handle; - let color_offset = color_staging.offset(); - unsafe { - ctx.transfer - .run(move |xf, cmd| { - let range = vk::ImageSubresourceRange { - aspect_mask: vk::ImageAspectFlags::COLOR, - base_mip_level: 0, - level_count: 1, - base_array_layer: 0, - layer_count: 1, - }; - xf.device.cmd_pipeline_barrier( - cmd, - vk::PipelineStageFlags::TOP_OF_PIPE, - vk::PipelineStageFlags::TRANSFER, - vk::DependencyFlags::default(), - &[], - &[], - &[vk::ImageMemoryBarrier::default() - .dst_access_mask(vk::AccessFlags::TRANSFER_WRITE) - .src_queue_family_index(vk::QUEUE_FAMILY_IGNORED) - .dst_queue_family_index(vk::QUEUE_FAMILY_IGNORED) - .old_layout(vk::ImageLayout::UNDEFINED) - .new_layout(vk::ImageLayout::TRANSFER_DST_OPTIMAL) - .image(color_handle) - .subresource_range(range)], - ); - xf.device.cmd_copy_buffer_to_image( - cmd, - staging_buffer, - color_handle, - vk::ImageLayout::TRANSFER_DST_OPTIMAL, - &[vk::BufferImageCopy { - buffer_offset: color_offset, - image_subresource: vk::ImageSubresourceLayers { - aspect_mask: vk::ImageAspectFlags::COLOR, - mip_level: 0, - base_array_layer: 0, - layer_count: 1, - }, - image_extent: vk::Extent3D { - width, - height, - depth: 1, - }, - ..Default::default() - }], - ); - xf.stages |= vk::PipelineStageFlags::FRAGMENT_SHADER; - xf.image_barriers.push( - vk::ImageMemoryBarrier::default() - .src_access_mask(vk::AccessFlags::TRANSFER_WRITE) - .dst_access_mask(vk::AccessFlags::SHADER_READ) - .src_queue_family_index(xf.queue_family) - .dst_queue_family_index(xf.dst_queue_family) - .old_layout(vk::ImageLayout::TRANSFER_DST_OPTIMAL) - .new_layout(vk::ImageLayout::SHADER_READ_ONLY_OPTIMAL) - .image(color_handle) - .subresource_range(range), - ); - }) - .await?; - } - Ok(color) -} - -async fn load_solid_color(ctx: &LoadCtx, rgba: [f32; 4]) -> Result { - unsafe { - let image = DedicatedImage::new( - &ctx.gfx.device, - &ctx.gfx.memory_properties, - &vk::ImageCreateInfo::default() - .image_type(vk::ImageType::TYPE_2D) - .format(vk::Format::R8G8B8A8_SRGB) - .extent(vk::Extent3D { - width: 1, - height: 1, - depth: 1, - }) - .mip_levels(1) - .array_layers(1) - .samples(vk::SampleCountFlags::TYPE_1) - .usage(vk::ImageUsageFlags::SAMPLED | vk::ImageUsageFlags::TRANSFER_DST), - ); - let handle = image.handle; - ctx.transfer - .run(move |xf, cmd| { - let range = vk::ImageSubresourceRange { - aspect_mask: vk::ImageAspectFlags::COLOR, - base_mip_level: 0, - level_count: 1, - base_array_layer: 0, - layer_count: 1, - }; - xf.device.cmd_pipeline_barrier( - cmd, - vk::PipelineStageFlags::TOP_OF_PIPE, - vk::PipelineStageFlags::TRANSFER, - vk::DependencyFlags::default(), - &[], - &[], - &[vk::ImageMemoryBarrier::default() - .dst_access_mask(vk::AccessFlags::TRANSFER_WRITE) - .src_queue_family_index(vk::QUEUE_FAMILY_IGNORED) - .dst_queue_family_index(vk::QUEUE_FAMILY_IGNORED) - .old_layout(vk::ImageLayout::UNDEFINED) - .new_layout(vk::ImageLayout::TRANSFER_DST_OPTIMAL) - .image(handle) - .subresource_range(range)], - ); - xf.device.cmd_clear_color_image( - cmd, - handle, - vk::ImageLayout::TRANSFER_DST_OPTIMAL, - &vk::ClearColorValue { float32: rgba }, - &[range], - ); - xf.stages |= vk::PipelineStageFlags::FRAGMENT_SHADER; - xf.image_barriers.push( - vk::ImageMemoryBarrier::default() - .src_access_mask(vk::AccessFlags::TRANSFER_WRITE) - .dst_access_mask(vk::AccessFlags::SHADER_READ) - .src_queue_family_index(xf.queue_family) - .dst_queue_family_index(xf.dst_queue_family) - .old_layout(vk::ImageLayout::TRANSFER_DST_OPTIMAL) - .new_layout(vk::ImageLayout::SHADER_READ_ONLY_OPTIMAL) - .image(handle) - .subresource_range(range), - ); - }) - .await?; - Ok(image) - } + Ok(MeshMaterialDefinition { + width, + height, + srgb_rgba_color_data: image_data, + }) } diff --git a/client/src/graphics/meshes.rs b/client/src/graphics/meshes.rs index 8c91c15c..a5ab44ac 100644 --- a/client/src/graphics/meshes.rs +++ b/client/src/graphics/meshes.rs @@ -5,6 +5,8 @@ use lahar::{BufferRegionAlloc, DedicatedImage}; use memoffset::offset_of; use vk_shader_macros::include_glsl; +use crate::graphics::asset_loader::AssetLoadContext; + use super::Base; use common::defer; @@ -193,16 +195,16 @@ impl Meshes { device.cmd_bind_vertex_buffers( cmd, 0, - &[mesh.vertices.buffer], - &[mesh.vertices.offset], + &[mesh.geometry.vertices.buffer], + &[mesh.geometry.vertices.offset], ); device.cmd_bind_index_buffer( cmd, - mesh.indices.buffer, - mesh.indices.offset, + mesh.geometry.indices.buffer, + mesh.geometry.indices.offset, vk::IndexType::UINT32, ); - device.cmd_draw_indexed(cmd, mesh.index_count, 1, 0, 0, 0); + device.cmd_draw_indexed(cmd, mesh.geometry.index_count, 1, 0, 0, 0); } } @@ -221,23 +223,312 @@ pub struct Vertex { pub normal: na::Unit>, } +pub struct MeshGeometryDefinition { + pub vertices: Vec, + pub indices: Vec, +} + +pub struct MeshMaterialDefinition { + pub width: u32, + pub height: u32, + pub srgb_rgba_color_data: Vec, +} + #[derive(Copy, Clone)] pub struct Mesh { - pub vertices: BufferRegionAlloc, - pub indices: BufferRegionAlloc, - pub index_count: u32, + pub geometry: MeshGeometry, pub pool: vk::DescriptorPool, pub ds: vk::DescriptorSet, // TODO: Make shareable + pub material: MeshMaterial, +} + +impl Mesh { + pub async fn from_definition( + ctx: &AssetLoadContext, + mesh_geometry: MeshGeometryDefinition, + mesh_material: MeshMaterialDefinition, + ) -> Self { + unsafe { + let (geometry, material) = tokio::join!( + MeshGeometry::from_definition(ctx, mesh_geometry), + MeshMaterial::from_definition(ctx, mesh_material) + ); + + let pool = ctx + .device() + .create_descriptor_pool( + &vk::DescriptorPoolCreateInfo::default() + .max_sets(1) + .pool_sizes(&[vk::DescriptorPoolSize { + ty: vk::DescriptorType::COMBINED_IMAGE_SAMPLER, + descriptor_count: 1, + }]), + None, + ) + .unwrap(); + let ds = ctx + .device() + .allocate_descriptor_sets( + &vk::DescriptorSetAllocateInfo::default() + .descriptor_pool(pool) + .set_layouts(&[ctx.shader_data().mesh_ds_layout]), + ) + .unwrap()[0]; + ctx.device().update_descriptor_sets( + &[vk::WriteDescriptorSet::default() + .dst_set(ds) + .dst_binding(0) + .descriptor_type(vk::DescriptorType::COMBINED_IMAGE_SAMPLER) + .image_info(&[vk::DescriptorImageInfo { + sampler: vk::Sampler::null(), + image_view: material.color_view, + image_layout: vk::ImageLayout::SHADER_READ_ONLY_OPTIMAL, + }])], + &[], + ); + + Mesh { + geometry, + pool, + ds, + material, + } + } + } + + pub unsafe fn destroy(&mut self, device: &ash::Device) { + unsafe { + device.destroy_descriptor_pool(self.pool, None); + self.material.destroy(device); + self.geometry.destroy(device); + } + } +} + +#[derive(Copy, Clone)] +pub struct MeshGeometry { + pub vertices: BufferRegionAlloc, + pub indices: BufferRegionAlloc, + pub index_count: u32, +} + +impl MeshGeometry { + pub async fn from_definition( + ctx: &AssetLoadContext, + mesh_geometry: MeshGeometryDefinition, + ) -> Self { + unsafe { + let work = ctx.begin_work(); + let finish_time = work.time().get(); + let vertex_staging = + ctx.alloc_staging::(mesh_geometry.vertices.len(), 1, finish_time); + let index_staging = + ctx.alloc_staging::(mesh_geometry.indices.len(), 1, finish_time); + std::ptr::copy_nonoverlapping( + mesh_geometry.vertices.as_ptr(), + vertex_staging.pointer.as_ptr(), + mesh_geometry.vertices.len(), + ); + std::ptr::copy_nonoverlapping( + mesh_geometry.indices.as_ptr(), + index_staging.pointer.as_ptr(), + mesh_geometry.indices.len(), + ); + let vertex_alloc = ctx.alloc_vertices(mesh_geometry.vertices.len()); + let index_alloc = ctx.alloc_indices(mesh_geometry.indices.len()); + ctx.device().cmd_copy_buffer( + work.cmd(), + vertex_staging.buffer, + vertex_alloc.buffer, + &[vk::BufferCopy { + src_offset: vertex_staging.offset, + dst_offset: vertex_alloc.offset, + size: vertex_staging.size, + }], + ); + ctx.device().cmd_copy_buffer( + work.cmd(), + index_staging.buffer, + index_alloc.buffer, + &[vk::BufferCopy { + src_offset: index_staging.offset, + dst_offset: index_alloc.offset, + size: index_staging.size, + }], + ); + ctx.device().cmd_pipeline_barrier( + work.cmd(), + vk::PipelineStageFlags::TRANSFER, + vk::PipelineStageFlags::VERTEX_INPUT, + vk::DependencyFlags::default(), + &[], + &[ + vk::BufferMemoryBarrier::default() + .src_access_mask(vk::AccessFlags::TRANSFER_WRITE) + .dst_access_mask(vk::AccessFlags::VERTEX_ATTRIBUTE_READ) + .src_queue_family_index(vk::QUEUE_FAMILY_IGNORED) + .dst_queue_family_index(vk::QUEUE_FAMILY_IGNORED) + .buffer(vertex_alloc.buffer) + .offset(vertex_alloc.offset) + .size(vertex_staging.size), + vk::BufferMemoryBarrier::default() + .src_access_mask(vk::AccessFlags::TRANSFER_WRITE) + .dst_access_mask(vk::AccessFlags::INDEX_READ) + .src_queue_family_index(vk::QUEUE_FAMILY_IGNORED) + .dst_queue_family_index(vk::QUEUE_FAMILY_IGNORED) + .buffer(index_alloc.buffer) + .offset(index_alloc.offset) + .size(index_staging.size), + ], + &[], + ); + work.end(); + ctx.wait_for_completion(finish_time).await; + + MeshGeometry { + vertices: vertex_alloc, + indices: index_alloc, + index_count: u32::try_from(mesh_geometry.indices.len()).unwrap(), + } + } + } + + pub unsafe fn destroy(&mut self, _device: &ash::Device) { + // Nothing actually needs to be cleaned up here. + // This implementation is left in so that we can remember to call it, ensuring + // that if this ever changes, we don't forget to clean things up. + } +} + +#[derive(Copy, Clone)] +pub struct MeshMaterial { pub color: DedicatedImage, pub color_view: vk::ImageView, } -impl crate::loader::Cleanup for Mesh { - unsafe fn cleanup(mut self, gfx: &Base) { +impl MeshMaterial { + pub async fn from_definition( + ctx: &AssetLoadContext, + mesh_material: MeshMaterialDefinition, + ) -> Self { + unsafe { + let work = ctx.begin_work(); + let finish_time = work.time().get(); + let color_staging = ctx.alloc_staging::( + mesh_material.width as usize * mesh_material.height as usize * 4, + 4, + finish_time, + ); + std::ptr::copy_nonoverlapping( + mesh_material.srgb_rgba_color_data.as_ptr(), + color_staging.pointer.as_ptr(), + mesh_material.srgb_rgba_color_data.len(), + ); + let color = DedicatedImage::new( + ctx.device(), + ctx.memory_properties(), + &vk::ImageCreateInfo::default() + .image_type(vk::ImageType::TYPE_2D) + .format(vk::Format::R8G8B8A8_SRGB) + .extent(vk::Extent3D { + width: mesh_material.width, + height: mesh_material.height, + depth: 1, + }) + .mip_levels(1) + .array_layers(1) + .samples(vk::SampleCountFlags::TYPE_1) + .usage(vk::ImageUsageFlags::SAMPLED | vk::ImageUsageFlags::TRANSFER_DST), + ); + let range = vk::ImageSubresourceRange { + aspect_mask: vk::ImageAspectFlags::COLOR, + base_mip_level: 0, + level_count: 1, + base_array_layer: 0, + layer_count: 1, + }; + ctx.device().cmd_pipeline_barrier( + work.cmd(), + vk::PipelineStageFlags::TOP_OF_PIPE, + vk::PipelineStageFlags::TRANSFER, + vk::DependencyFlags::default(), + &[], + &[], + &[vk::ImageMemoryBarrier::default() + .dst_access_mask(vk::AccessFlags::TRANSFER_WRITE) + .src_queue_family_index(vk::QUEUE_FAMILY_IGNORED) + .dst_queue_family_index(vk::QUEUE_FAMILY_IGNORED) + .old_layout(vk::ImageLayout::UNDEFINED) + .new_layout(vk::ImageLayout::TRANSFER_DST_OPTIMAL) + .image(color.handle) + .subresource_range(range)], + ); + ctx.device().cmd_copy_buffer_to_image( + work.cmd(), + color_staging.buffer, + color.handle, + vk::ImageLayout::TRANSFER_DST_OPTIMAL, + &[vk::BufferImageCopy { + buffer_offset: color_staging.offset, + image_subresource: vk::ImageSubresourceLayers { + aspect_mask: vk::ImageAspectFlags::COLOR, + mip_level: 0, + base_array_layer: 0, + layer_count: 1, + }, + image_extent: vk::Extent3D { + width: mesh_material.width, + height: mesh_material.height, + depth: 1, + }, + ..Default::default() + }], + ); + ctx.device().cmd_pipeline_barrier( + work.cmd(), + vk::PipelineStageFlags::TRANSFER, + vk::PipelineStageFlags::FRAGMENT_SHADER, + vk::DependencyFlags::default(), + &[], + &[], + &[vk::ImageMemoryBarrier::default() + .src_access_mask(vk::AccessFlags::TRANSFER_WRITE) + .dst_access_mask(vk::AccessFlags::SHADER_READ) + .src_queue_family_index(vk::QUEUE_FAMILY_IGNORED) + .dst_queue_family_index(vk::QUEUE_FAMILY_IGNORED) + .old_layout(vk::ImageLayout::TRANSFER_DST_OPTIMAL) + .new_layout(vk::ImageLayout::SHADER_READ_ONLY_OPTIMAL) + .image(color.handle) + .subresource_range(range)], + ); + work.end(); + ctx.wait_for_completion(finish_time).await; + + let color_view = ctx + .device() + .create_image_view( + &vk::ImageViewCreateInfo::default() + .image(color.handle) + .view_type(vk::ImageViewType::TYPE_2D) + .format(vk::Format::R8G8B8A8_SRGB) + .subresource_range(vk::ImageSubresourceRange { + aspect_mask: vk::ImageAspectFlags::COLOR, + base_mip_level: 0, + level_count: 1, + base_array_layer: 0, + layer_count: 1, + }), + None, + ) + .unwrap(); + + MeshMaterial { color, color_view } + } + } + + pub unsafe fn destroy(&mut self, device: &ash::Device) { unsafe { - let device = &*gfx.device; - device.destroy_descriptor_pool(self.pool, None); device.destroy_image_view(self.color_view, None); self.color.destroy(device); } From 7be06ff32b15ee44863006e95994ee25d58d56c2 Mon Sep 17 00:00:00 2001 From: Patrick Owen Date: Sun, 16 Aug 2026 22:27:56 -0400 Subject: [PATCH 6/6] Delete code related to old Loader --- client/src/graphics/draw.rs | 9 +- client/src/lahar_deprecated/condition.rs | 43 --- client/src/lahar_deprecated/mod.rs | 7 - client/src/lahar_deprecated/ring_alloc.rs | 108 -------- client/src/lahar_deprecated/staging.rs | 142 ---------- client/src/lahar_deprecated/transfer.rs | 289 -------------------- client/src/lib.rs | 4 - client/src/loader.rs | 314 ---------------------- 8 files changed, 1 insertion(+), 915 deletions(-) delete mode 100644 client/src/lahar_deprecated/condition.rs delete mode 100644 client/src/lahar_deprecated/mod.rs delete mode 100644 client/src/lahar_deprecated/ring_alloc.rs delete mode 100644 client/src/lahar_deprecated/staging.rs delete mode 100644 client/src/lahar_deprecated/transfer.rs delete mode 100644 client/src/loader.rs diff --git a/client/src/graphics/draw.rs b/client/src/graphics/draw.rs index d711baf3..3c72d2a5 100644 --- a/client/src/graphics/draw.rs +++ b/client/src/graphics/draw.rs @@ -8,7 +8,7 @@ use metrics::histogram; use super::{Base, Fog, Frustum, GltfScene, Meshes, Voxels, fog, voxels}; use crate::graphics::asset_loader::AssetLoader; -use crate::{Config, Loader, Sim}; +use crate::{Config, Sim}; use common::SimConfig; use common::proto::{Character, Position}; @@ -33,9 +33,6 @@ pub struct Draw { /// Descriptor pool from which descriptor sets shared between many pipelines are allocated common_descriptor_pool: vk::DescriptorPool, - /// Drives async asset loading - loader: Loader, - // // Rendering pipelines // @@ -135,7 +132,6 @@ impl Draw { ) .unwrap(); - let mut loader = Loader::new(cfg.clone(), gfx.clone()); let asset_loader = AssetLoader::new(gfx.clone(), cfg.clone()); // Construct the per-frame states @@ -219,8 +215,6 @@ impl Draw { common_pipeline_layout, common_descriptor_pool, - loader, - voxels: None, meshes, fog, @@ -301,7 +295,6 @@ impl Draw { let view = sim.as_ref().map_or_else(Position::origin, |sim| sim.view()); let projection = frustum.projection(1.0e-4); let view_projection = projection.matrix() * na::Matrix4::from(view.local.inverse()); - self.loader.drive(); let device = &*self.gfx.device; let state_index = self.next_state; diff --git a/client/src/lahar_deprecated/condition.rs b/client/src/lahar_deprecated/condition.rs deleted file mode 100644 index dcf821ef..00000000 --- a/client/src/lahar_deprecated/condition.rs +++ /dev/null @@ -1,43 +0,0 @@ -use std::task::{Context, Waker}; - -/// Manages tasks waiting on a single condition -pub struct Condition { - wakers: Vec, - generation: u64, -} - -impl Condition { - pub fn new() -> Self { - Self { - wakers: Vec::new(), - generation: 0, - } - } - - /// Ensure the next `wake` call will wake the calling task - /// - /// Checks the task-associated generation counter stored in `state`. If it's present and - /// current, we already have this task's `Waker` and no action is necessary. Otherwise, record a - /// `Waker` and store the current generation in `state`. - pub fn register(&mut self, cx: &mut Context, state: &mut State) { - if state.0 == Some(self.generation) { - return; - } - state.0 = Some(self.generation); - self.wakers.push(cx.waker().clone()); - } - - /// Wake all known tasks - pub fn notify(&mut self) { - self.generation = self.generation.wrapping_add(1); - for waker in self.wakers.drain(..) { - waker.wake(); - } - } -} - -/// State maintained by each interested task -/// -/// Stores the generation at which the task previously registered a `Waker`, if any. -#[derive(Default)] -pub struct State(Option); diff --git a/client/src/lahar_deprecated/mod.rs b/client/src/lahar_deprecated/mod.rs deleted file mode 100644 index d3df1d1a..00000000 --- a/client/src/lahar_deprecated/mod.rs +++ /dev/null @@ -1,7 +0,0 @@ -//! This code is directly copied from https://github.com/Ralith/lahar/tree/fbc889a4538e2d3b6b519a6cb7a3538d7b3bfcdf -//! with minor modifications for interoperability with the current versions of ash and lahar. It is intended to be temporary -//! and will be replaced when the code is sufficiently refactored to support the newer lahar structures. -mod condition; -mod ring_alloc; -pub mod staging; -pub mod transfer; diff --git a/client/src/lahar_deprecated/ring_alloc.rs b/client/src/lahar_deprecated/ring_alloc.rs deleted file mode 100644 index 0ee0ee61..00000000 --- a/client/src/lahar_deprecated/ring_alloc.rs +++ /dev/null @@ -1,108 +0,0 @@ -use std::collections::VecDeque; - -/// State tracker for a ring buffer of contiguous variable-sized allocations with random frees -pub struct RingAlloc { - /// List of starting offsets, and whether they've been freed - allocations: VecDeque<(usize, bool)>, - /// Offset at which the next allocation will start - head: usize, - /// Number of allocations which have been freed - /// - /// Tracking this supports random freeing by making it easy to keep track of a single element - /// inside `allocations` even as items are added/removed. - freed: u64, -} - -impl RingAlloc { - pub fn new() -> Self { - RingAlloc { - allocations: VecDeque::new(), - head: 0, - freed: 0, - } - } - - /// Returns the starting offset of a contiguous run of `size` units, or `None` if none exists. - /// - /// `capacity` is the total capacity of the ring. - pub fn alloc(&mut self, capacity: usize, size: usize) -> Option<(usize, Id)> { - let tail = if let Some(&(tail, _)) = self.allocations.front() { - tail - } else { - if size > capacity { - return None; - } - // No allocations, reset to initial state - self.allocations.push_back((0, false)); - self.head = size; - self.freed = 0; - return Some((0, Id(0))); - }; - let id = Id(self.freed.wrapping_add(self.allocations.len() as u64)); - if self.head > tail { - // There's a run from the head to the end of the buffer - let free = capacity - self.head; - if free >= size { - let start = self.head; - self.allocations.push_back((start, false)); - self.head = (start + size) % capacity; - return Some((start, id)); - } - // and from the start of the buffer to the tail - if tail >= size { - self.allocations.push_back((0, false)); - self.head = size; - return Some((0, id)); - } - return None; - } - // Only one run, from head to tail - let free = tail - self.head; - if free >= size { - let start = self.head; - self.allocations.push_back((start, false)); - self.head = start + size; - return Some((start, id)); - } - None - } - - pub fn free(&mut self, id: Id) { - self.allocations[id.0.wrapping_sub(self.freed) as usize].1 = true; - while let Some(&(_, true)) = self.allocations.front() { - self.allocations.pop_front(); - self.freed += 1; - } - } -} - -#[derive(Debug, Copy, Clone)] -pub struct Id(u64); - -#[cfg(test)] -mod tests { - use super::*; - - #[test] - fn sanity() { - let mut r = RingAlloc::new(); - const CAP: usize = 4; - let a = r.alloc(CAP, 3).unwrap(); - assert!(r.alloc(CAP, 2).is_none()); - let b = r.alloc(CAP, 1).unwrap(); - assert_eq!(b.0, 3); - assert!(r.alloc(CAP, 1).is_none()); - r.free(a.1); - let c = r.alloc(CAP, 1).unwrap(); - assert_eq!(c.0, 0); - let d = r.alloc(CAP, 2).unwrap(); - assert_eq!(d.0, 1); - assert!(r.alloc(CAP, 1).is_none()); - r.free(c.1); - r.free(b.1); - let e = r.alloc(CAP, 1).unwrap(); - assert_eq!(e.0, 3); - let f = r.alloc(CAP, 1).unwrap(); - assert_eq!(f.0, 0); - } -} diff --git a/client/src/lahar_deprecated/staging.rs b/client/src/lahar_deprecated/staging.rs deleted file mode 100644 index b7f49419..00000000 --- a/client/src/lahar_deprecated/staging.rs +++ /dev/null @@ -1,142 +0,0 @@ -use std::future::Future; -use std::ops::{Deref, DerefMut}; -use std::sync::{Arc, Mutex}; -use std::task::Poll; - -use ash::{Device, vk}; -use futures_util::future; - -use super::condition::{self, Condition}; -use super::ring_alloc::{self, RingAlloc}; -use lahar::DedicatedMapping; - -/// A host-visible circular buffer for short-lived allocations -/// -/// Best for transient uses like streaming transfers. Retaining an allocation of any size will block -/// future allocations once the buffer wraps back aground. -pub struct StagingBuffer { - device: Arc, - buffer: DedicatedMapping<[u8]>, - state: Mutex, -} - -struct State { - alloc: RingAlloc, - free: Condition, -} - -impl StagingBuffer { - pub fn new( - device: Arc, - props: &vk::PhysicalDeviceMemoryProperties, - capacity: usize, - ) -> Self { - let buffer = unsafe { - DedicatedMapping::zeroed_array( - &device, - props, - vk::BufferUsageFlags::TRANSFER_SRC, - capacity, - ) - }; - Self { - device, - buffer, - state: Mutex::new(State { - alloc: RingAlloc::new(), - free: Condition::new(), - }), - } - } - - pub fn buffer(&self) -> vk::Buffer { - self.buffer.buffer() - } - - /// Largest possible allocation - pub fn capacity(&self) -> usize { - self.buffer.len() - } - - /// Completes when sufficient space is available - /// - /// Yields `None` if `size > self.capacity()`. No fairness guarantees, i.e. small allocations - /// may starve large ones. - pub fn alloc(&self, size: usize) -> impl Future>> { - let mut cond_state = condition::State::default(); - future::poll_fn(move |cx| { - if size > self.capacity() { - return Poll::Ready(None); - } - let mut state = self.state.lock().unwrap(); - match state.alloc.alloc(self.capacity(), size) { - None => { - state.free.register(cx, &mut cond_state); - Poll::Pending - } - Some((offset, id)) => Poll::Ready(Some(Alloc { - buf: self, - bytes: unsafe { - std::slice::from_raw_parts_mut( - (self.buffer.as_ptr() as *const u8).add(offset) as *mut u8, - size, - ) - }, - id, - })), - } - }) - } - - fn free(&self, id: ring_alloc::Id) { - let mut state = self.state.lock().unwrap(); - state.alloc.free(id); - state.free.notify(); - } -} - -impl Drop for StagingBuffer { - fn drop(&mut self) { - unsafe { - self.buffer.destroy(&self.device); - } - } -} - -/// An allocation from a `StagingBuffer` -pub struct Alloc<'a> { - buf: &'a StagingBuffer, - bytes: &'a mut [u8], - id: ring_alloc::Id, -} - -impl Alloc<'_> { - pub fn offset(&self) -> vk::DeviceSize { - self.bytes.as_ptr() as vk::DeviceSize - - self.buf.buffer.as_ptr() as *const u8 as vk::DeviceSize - } - - pub fn size(&self) -> vk::DeviceSize { - self.bytes.len() as _ - } -} - -impl Deref for Alloc<'_> { - type Target = [u8]; - - fn deref(&self) -> &[u8] { - self.bytes - } -} - -impl DerefMut for Alloc<'_> { - fn deref_mut(&mut self) -> &mut [u8] { - self.bytes - } -} - -impl Drop for Alloc<'_> { - fn drop(&mut self) { - self.buf.free(self.id); - } -} diff --git a/client/src/lahar_deprecated/transfer.rs b/client/src/lahar_deprecated/transfer.rs deleted file mode 100644 index 11260cf6..00000000 --- a/client/src/lahar_deprecated/transfer.rs +++ /dev/null @@ -1,289 +0,0 @@ -use std::convert::TryFrom; -use std::fmt; -use std::future::Future; -use std::sync::Arc; -use std::thread; -use std::time::Duration; - -use ash::vk; -use futures_util::FutureExt; -use tokio::sync::{ - mpsc::{self, error::TryRecvError}, - oneshot, -}; - -#[derive(Clone)] -pub struct TransferHandle { - send: mpsc::UnboundedSender, -} - -impl TransferHandle { - pub unsafe fn run( - &self, - f: impl FnOnce(&mut TransferContext, vk::CommandBuffer) + Send + 'static, - ) -> impl Future> { - let (sender, recv) = oneshot::channel(); - let _ = self.send.send(Message { - sender, - op: Box::new(f), - }); - recv.map(|x| x.map_err(|_| ShutDown)) - } -} - -pub struct TransferContext { - pub device: Arc, - pub queue_family: u32, - /// May be equal to queue_family - pub dst_queue_family: u32, - pub stages: vk::PipelineStageFlags, - pub buffer_barriers: Vec>, - pub image_barriers: Vec>, -} - -#[derive(Debug, Copy, Clone)] -pub struct ShutDown; - -impl fmt::Display for ShutDown { - fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - f.pad("transfer reactor shut down") - } -} - -impl std::error::Error for ShutDown {} - -#[allow(clippy::type_complexity)] -struct Message { - sender: oneshot::Sender<()>, - op: Box, -} - -pub struct Reactor { - queue: vk::Queue, - spare_fences: Vec, - spare_cmds: Vec, - in_flight: Vec, - /// Fences for in-flight transfer operations; directly corresponds to in_flight entries - in_flight_fences: Vec, - cmd_pool: vk::CommandPool, - pending: Option, - recv: mpsc::UnboundedReceiver, - ctx: TransferContext, -} - -impl Reactor { - /// Safety: valid use use of queue_family, queue - pub unsafe fn new( - device: Arc, - queue_family: u32, - queue: vk::Queue, - dst_queue_family: Option, - ) -> (TransferHandle, Self) { - unsafe { - let (send, recv) = mpsc::unbounded_channel(); - let cmd_pool = device - .create_command_pool( - &vk::CommandPoolCreateInfo::default() - .queue_family_index(queue_family) - .flags(vk::CommandPoolCreateFlags::RESET_COMMAND_BUFFER), - None, - ) - .unwrap(); - ( - TransferHandle { send }, - Self { - queue, - spare_fences: Vec::new(), - spare_cmds: Vec::new(), - in_flight: Vec::new(), - in_flight_fences: Vec::new(), - cmd_pool, - pending: None, - recv, - ctx: TransferContext { - device, - queue_family, - dst_queue_family: dst_queue_family.unwrap_or(queue_family), - stages: vk::PipelineStageFlags::empty(), - buffer_barriers: Vec::new(), - image_barriers: Vec::new(), - }, - }, - ) - } - } - - pub fn poll(&mut self) -> Result<(), Disconnected> { - self.run_for(Duration::from_secs(0)) - } - - pub fn run_for(&mut self, timeout: Duration) -> Result<(), Disconnected> { - self.queue()?; - self.flush(); - - if self.in_flight.is_empty() { - thread::sleep(timeout); - return Ok(()); - } - - // We could move this to a background thread and continue to submit new work while it's - // waiting, but we want to batch up operations a bit anyway. - let result = unsafe { - self.ctx.device.wait_for_fences( - &self.in_flight_fences, - false, - u64::try_from(timeout.as_nanos()).unwrap_or(u64::MAX), - ) - }; - match result { - Err(vk::Result::TIMEOUT) => return Ok(()), - Err(e) => panic!("{}", e), - Ok(()) => {} - } - for i in (0..self.in_flight.len()).rev() { - unsafe { - if self - .ctx - .device - .get_fence_status(self.in_flight_fences[i]) - .unwrap() - { - let fence = self.in_flight_fences.swap_remove(i); - self.ctx.device.reset_fences(&[fence]).unwrap(); - self.spare_fences.push(fence); - let batch = self.in_flight.swap_remove(i); - for sender in batch.senders { - let _ = sender.send(()); - } - self.spare_cmds.push(batch.cmd); - } - } - } - Ok(()) - } - - fn queue(&mut self) -> Result<(), Disconnected> { - loop { - match self.recv.try_recv() { - Ok(Message { sender, op }) => { - let cmd = self.prepare(sender); - op(&mut self.ctx, cmd); - } - Err(TryRecvError::Disconnected) => return Err(self::Disconnected), - Err(TryRecvError::Empty) => return Ok(()), - } - } - } - - fn prepare(&mut self, send: oneshot::Sender<()>) -> vk::CommandBuffer { - if let Some(ref mut pending) = self.pending { - pending.senders.push(send); - return pending.cmd; - } - let cmd = if let Some(cmd) = self.spare_cmds.pop() { - cmd - } else { - unsafe { - self.ctx - .device - .allocate_command_buffers( - &vk::CommandBufferAllocateInfo::default() - .command_pool(self.cmd_pool) - .command_buffer_count(1), - ) - .unwrap() - .into_iter() - .next() - .unwrap() - } - }; - unsafe { - self.ctx - .device - .begin_command_buffer( - cmd, - &vk::CommandBufferBeginInfo::default() - .flags(vk::CommandBufferUsageFlags::ONE_TIME_SUBMIT), - ) - .unwrap(); - } - self.pending = Some(Batch { - cmd, - senders: vec![send], - }); - cmd - } - - /// Submit queued operations - fn flush(&mut self) { - let pending = match self.pending.take() { - Some(x) => x, - None => return, - }; - let device = &self.ctx.device; - let fence = if let Some(fence) = self.spare_fences.pop() { - fence - } else { - unsafe { - device - .create_fence(&vk::FenceCreateInfo::default(), None) - .unwrap() - } - }; - unsafe { - device.cmd_pipeline_barrier( - pending.cmd, - vk::PipelineStageFlags::TRANSFER, - self.ctx.stages, - vk::DependencyFlags::default(), - &[], - &self.ctx.buffer_barriers, - &self.ctx.image_barriers, - ); - device.end_command_buffer(pending.cmd).unwrap(); - device - .queue_submit( - self.queue, - &[vk::SubmitInfo::default().command_buffers(&[pending.cmd])], - fence, - ) - .unwrap(); - } - self.ctx.stages = vk::PipelineStageFlags::empty(); - self.ctx.buffer_barriers.clear(); - self.ctx.image_barriers.clear(); - self.in_flight.push(pending); - self.in_flight_fences.push(fence); - } -} - -impl Drop for Reactor { - fn drop(&mut self) { - let device = &self.ctx.device; - unsafe { - if !self.in_flight.is_empty() { - device - .wait_for_fences(&self.in_flight_fences, true, u64::MAX) - .unwrap(); - } - device.destroy_command_pool(self.cmd_pool, None); - for fence in self.spare_fences.drain(..) { - device.destroy_fence(fence, None); - } - for fence in self.in_flight_fences.drain(..) { - device.destroy_fence(fence, None); - } - } - } -} - -unsafe impl Send for Reactor {} - -struct Batch { - cmd: vk::CommandBuffer, - // Future work: efficient broadcast future - senders: Vec>, -} - -#[derive(Debug, Copy, Clone)] -pub struct Disconnected; diff --git a/client/src/lib.rs b/client/src/lib.rs index 75d3a940..76f56799 100644 --- a/client/src/lib.rs +++ b/client/src/lib.rs @@ -13,8 +13,6 @@ macro_rules! cstr { extern crate nalgebra as na; mod config; pub mod graphics; -mod lahar_deprecated; -mod loader; mod local_character_controller; pub mod metrics; pub mod net; @@ -24,5 +22,3 @@ mod worldgen_driver; pub use config::Config; pub use sim::Sim; - -use loader::{Asset, Loader}; diff --git a/client/src/loader.rs b/client/src/loader.rs deleted file mode 100644 index f56493b7..00000000 --- a/client/src/loader.rs +++ /dev/null @@ -1,314 +0,0 @@ -use std::{ - any::{Any, TypeId}, - convert::TryFrom, - marker::PhantomData, - sync::{Arc, Mutex}, -}; - -use anyhow::Result; -use ash::vk; -use downcast_rs::{Downcast, impl_downcast}; -use fxhash::FxHashMap; -use lahar::{BufferRegion, DedicatedImage}; -use tokio::sync::mpsc; -use tracing::error; - -use crate::{ - Config, - graphics::Base, - lahar_deprecated::{ - staging::StagingBuffer, - transfer::{self, TransferHandle}, - }, -}; - -pub trait Cleanup { - unsafe fn cleanup(self, gfx: &Base); -} - -impl Cleanup for DedicatedImage { - unsafe fn cleanup(mut self, gfx: &Base) { - unsafe { - self.destroy(&gfx.device); - } - } -} - -pub trait Loadable: Send + 'static { - type Output: Send + 'static + Cleanup; - fn load(self, ctx: &LoadCtx) -> LoadFuture<'_, Self::Output>; -} - -pub type LoadFuture<'a, T> = - std::pin::Pin> + 'a + Send>>; - -pub struct Loader { - runtime: tokio::runtime::Runtime, - recv: mpsc::UnboundedReceiver, - shared: Arc, - reactor: transfer::Reactor, - tables_index: FxHashMap, - tables: Vec>, -} - -impl Loader { - pub fn new(cfg: Arc, gfx: Arc) -> Self { - let runtime = tokio::runtime::Builder::new_multi_thread().build().unwrap(); - let (send, recv) = mpsc::unbounded_channel(); - let staging = - StagingBuffer::new(gfx.device.clone(), &gfx.memory_properties, 32 * 1024 * 1024); - unsafe { - gfx.set_name(staging.buffer(), cstr!("staging")); - } - let (transfer, reactor) = unsafe { - transfer::Reactor::new(gfx.device.clone(), gfx.queue_family, gfx.queue, None) - }; - let shared = Arc::new(Shared { - send, - ctx: LoadCtx { - cfg, - gfx, - staging, - transfer, - }, - }); - Self { - runtime, - recv, - shared, - reactor, - tables_index: FxHashMap::default(), - tables: Vec::new(), - } - } - - pub fn load(&mut self, description: &'static str, x: L) -> Asset { - let tables = &mut self.tables; - let table = *self - .tables_index - .entry(TypeId::of::()) - .or_insert_with(|| { - let n = u32::try_from(tables.len()).unwrap(); - tables.push(Box::new(Table::::new())); - n - }); - let index = self.tables[table as usize] - .downcast_mut::>() - .unwrap() - .alloc(); - let shared = self.shared.clone(); - self.runtime.spawn(async move { - match shared.ctx.load(x).await { - Ok(x) => { - let _ = shared.send.send(Message { - table, - index, - result: Box::new(x), - }); - } - Err(e) => { - error!("{} load failed: {:#}", description, e); - } - } - }); - Asset { - table, - index, - _marker: PhantomData, - } - } - - pub fn make_queue(&mut self, capacity: usize) -> WorkQueue { - let (input_send, mut input_recv) = mpsc::channel::(capacity); - let (output_send, output_recv) = mpsc::channel::(capacity); - let shared = self.shared.clone(); - self.runtime.spawn(async move { - while let Some(x) = input_recv.recv().await { - let shared = shared.clone(); - let out = output_send.clone(); - tokio::spawn(async move { - match shared.ctx.load(x).await { - Ok(x) => { - if let Err(e) = out.send(x).await { - unsafe { - e.0.cleanup(&shared.ctx.gfx); - } - } - } - Err(e) => { - error!( - "streaming {} load failed: {:#}", - std::any::type_name::(), - e - ); - } - } - }); - } - }); - WorkQueue { - shared: self.shared.clone(), - send: input_send, - recv: output_recv, - capacity, - fill: 0, - } - } - - /// Invoke `finish` functions of spawned loading operations - pub fn drive(&mut self) { - self.reactor.poll().unwrap(); - while let Ok(msg) = self.recv.try_recv() { - self.tables[msg.table as usize].finish(msg.index, msg.result); - } - } - - pub fn get(&self, handle: Asset) -> Option<&T> { - self.tables[handle.table as usize] - .downcast_ref::>() - .unwrap() - .data[handle.index as usize] - .as_ref() - } - - pub fn ctx(&self) -> &LoadCtx { - &self.shared.ctx - } -} - -impl Drop for Loader { - fn drop(&mut self) { - for table in self.tables.drain(..) { - table.cleanup(&self.shared.ctx.gfx); - } - } -} - -struct Shared { - send: mpsc::UnboundedSender, - ctx: LoadCtx, -} - -struct Message { - table: u32, - index: u32, - result: Box, -} - -pub struct LoadCtx { - pub cfg: Arc, - pub gfx: Arc, - pub staging: StagingBuffer, - pub transfer: TransferHandle, -} - -impl LoadCtx { - async fn load(&self, x: T) -> Result { - x.load(self).await - } -} - -trait AnyTable: Downcast { - fn finish(&mut self, index: u32, value: Box); - fn cleanup(self: Box, gfx: &Base); -} - -impl_downcast!(AnyTable); - -struct Table { - data: Vec>, -} - -impl Table { - fn new() -> Self { - Self { data: Vec::new() } - } - - fn alloc(&mut self) -> u32 { - let n = u32::try_from(self.data.len()).unwrap(); - self.data.push(None); - n - } -} - -impl AnyTable for Table { - fn finish(&mut self, index: u32, value: Box) { - self.data[index as usize] = Some(*value.downcast().unwrap()); - } - - fn cleanup(self: Box, gfx: &Base) { - for x in self.data.into_iter().flatten() { - unsafe { - x.cleanup(gfx); - } - } - } -} - -#[derive(Debug, Eq, PartialEq)] -pub struct Asset { - table: u32, - index: u32, - _marker: PhantomData T>, -} - -impl Clone for Asset { - fn clone(&self) -> Self { - *self - } -} - -impl Copy for Asset {} - -/// A bounded-capacity queue for streaming specific data (e.g. terrain chunks) -/// -/// Limiting capacity ensures predictable memory usage and helps focus computational resources on -/// recent requests when the total number of requests that could be submitted is large. This is -/// particularly useful for terrain, where recent requests are more likely to be close to the -/// viewpoint. -pub struct WorkQueue { - shared: Arc, - send: mpsc::Sender, - recv: mpsc::Receiver, - capacity: usize, - fill: usize, -} - -impl WorkQueue { - /// Begin loading a single item, if capacity is available - pub fn load(&mut self, x: T) -> Result<(), T> { - use tokio::sync::mpsc::error::TrySendError::*; - if self.fill == self.capacity { - return Err(x); - } - self.fill += 1; - self.send.try_send(x).map_err(|e| { - self.fill -= 1; - match e { - Full(x) => x, - Closed(x) => x, - } - }) - } - - /// Fetch a load result if one is ready, freeing capacity - pub fn poll(&mut self) -> Option { - let result = self.recv.try_recv().ok()?; - self.fill -= 1; - Some(result) - } -} - -impl Drop for WorkQueue { - fn drop(&mut self) { - // Ensure any future completions will be cleaned up by the loader - self.recv.close(); - // Gracefully drain already-completed tasks - while let Ok(x) = self.recv.try_recv() { - self.fill -= 1; - unsafe { - x.cleanup(&self.shared.ctx.gfx); - } - } - } -}