diff options
| -rw-r--r-- | Cargo.lock | 40 | ||||
| -rw-r--r-- | engine/Cargo.toml | 1 | ||||
| -rw-r--r-- | engine/src/asset.rs | 115 | ||||
| -rw-r--r-- | engine/src/lib.rs | 1 | ||||
| -rw-r--r-- | engine/src/work_queue.rs | 77 |
5 files changed, 104 insertions, 130 deletions
@@ -480,6 +480,25 @@ dependencies = [ ] [[package]] +name = "crossbeam-deque" +version = "0.8.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "622f3fc73690be383c7214310406f28a90e6edeadc3cea882f9d71e495b9711a" +dependencies = [ + "crossbeam-epoch", + "crossbeam-utils", +] + +[[package]] +name = "crossbeam-epoch" +version = "0.9.21" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "dc74980687109a3b14c72fd458107bf0baa1da1a1a805e178d15501ba9b86d9d" +dependencies = [ + "crossbeam-utils", +] + +[[package]] name = "crossbeam-queue" version = "0.3.12" source = "registry+https://github.com/rust-lang/crates.io-index" @@ -628,6 +647,7 @@ dependencies = [ "parking_lot", "paste", "raw-window-handle", + "rayon", "safer-ffi", "seq-macro", "serde", @@ -1835,6 +1855,26 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "20675572f6f24e9e76ef639bc5552774ed45f1c30e2951e1e99c59888861c539" [[package]] +name = "rayon" +version = "1.12.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fb39b166781f92d482534ef4b4b1b2568f42613b53e5b6c160e24cfbfa30926d" +dependencies = [ + "either", + "rayon-core", +] + +[[package]] +name = "rayon-core" +version = "1.13.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "22e18b0f0062d30d4230b2e85ff77fdfe4326feb054b9783a3460d8435c8ab91" +dependencies = [ + "crossbeam-deque", + "crossbeam-utils", +] + +[[package]] name = "redox_syscall" version = "0.4.1" source = "registry+https://github.com/rust-lang/crates.io-index" diff --git a/engine/Cargo.toml b/engine/Cargo.toml index b64d190..3fe0680 100644 --- a/engine/Cargo.toml +++ b/engine/Cargo.toml @@ -17,6 +17,7 @@ parking_lot = "0.12.3" circular-buffer = "2.0.0" intmap = "3.1.3" hashbrown = "0.15.2" +rayon = "1.12.0" engine-macros = { workspace = true } engine-ecs = { workspace = true } engine-reflection = { workspace = true } diff --git a/engine/src/asset.rs b/engine/src/asset.rs index 8f7829c..66efe35 100644 --- a/engine/src/asset.rs +++ b/engine/src/asset.rs @@ -4,22 +4,24 @@ use std::cell::RefCell; use std::convert::Infallible; use std::fmt::{Debug, Display}; use std::hash::Hash; -use std::hint::cold_path; use std::marker::PhantomData; +use std::panic::{catch_unwind, RefUnwindSafe, UnwindSafe}; use std::path::{Path, PathBuf}; use std::sync::mpsc::{ channel as mpsc_channel, Receiver as MpscReceiver, Sender as MpscSender, }; +use std::thread::available_parallelism; -use ecs::actions::Actions; +use rayon::{ThreadPool, ThreadPoolBuilder}; use crate::ecs::pair::ChildOf; use crate::ecs::phase::{Phase, PRE_UPDATE as PRE_UPDATE_PHASE}; use crate::ecs::sole::Single; use crate::ecs::{declare_entity, pair, Sole}; -use crate::work_queue::{Work, WorkQueue}; + +const BACKUP_IMPORT_WORK_THREAD_CNT: usize = 2; declare_entity! { pub HANDLE_ASSETS_PHASE: (Phase, pair!(ChildOf, { *PRE_UPDATE_PHASE })); @@ -106,7 +108,7 @@ pub struct Assets store: store::Store, metadata_lut: RefCell<hashbrown::HashMap<LabelOwned, Metadata>>, importers: hashbrown::HashMap<AssetTypeId, WrappedImporterFn>, - import_work_queue: WorkQueue<ImportWorkUserData>, + import_work_thread_pool: ThreadPool, import_work_msg_receiver: MpscReceiver<ImportWorkMessage>, import_work_msg_sender: MpscSender<ImportWorkMessage>, curr_tick_events: Vec<Event>, @@ -115,16 +117,23 @@ pub struct Assets impl Assets { #[must_use] - pub fn with_capacity(capacity: usize) -> Self + #[tracing::instrument(skip_all)] + pub(crate) fn with_capacity(capacity: usize) -> Self { let (import_work_msg_sender, import_work_msg_receiver) = mpsc_channel::<ImportWorkMessage>(); + let import_work_thread_cnt = calc_import_work_thread_cnt(); + Self { store: store::Store::with_capacity(capacity), metadata_lut: RefCell::new(hashbrown::HashMap::with_capacity(capacity)), importers: hashbrown::HashMap::new(), - import_work_queue: WorkQueue::new("asset_importing_work_queue"), + import_work_thread_pool: ThreadPoolBuilder::new() + .thread_name(|thread_index| format!("asset_import_worker_{thread_index}")) + .num_threads(import_work_thread_cnt) + .build() + .expect("Unable to create thread pool"), import_work_msg_receiver, import_work_msg_sender, curr_tick_events: Vec::with_capacity(capacity), @@ -136,7 +145,7 @@ impl Assets func: impl Fn(&mut Submitter<'_>, &Path, Option<&AssetSettings>) -> Result<(), Err>, ) where AssetT: Asset, - AssetSettings: 'static, + AssetSettings: UnwindSafe + RefUnwindSafe + 'static, Err: std::error::Error + Send + Sync + 'static, { self.importers @@ -284,7 +293,7 @@ impl Assets ) -> Handle<AssetT> where AssetT: Asset, - AssetSettings: Send + Sync + Debug + 'static, + AssetSettings: UnwindSafe + RefUnwindSafe + Send + Sync + Debug + 'static, { let label = label.into(); @@ -517,7 +526,7 @@ impl Assets asset_settings: Option<AssetSettings>, ) -> IdValid where - AssetSettings: Send + Sync + Debug + 'static, + AssetSettings: UnwindSafe + RefUnwindSafe + Send + Sync + Debug + 'static, { let id = match self.get_asset_by_label(&label) { Some((Some(_), id)) => { @@ -565,24 +574,25 @@ impl Assets label: &Label<'_>, asset_settings: Option<AssetSettings>, ) where - AssetSettings: Any + Send + Sync, + AssetSettings: UnwindSafe + RefUnwindSafe + Any + Send + Sync, { - let Some(importer) = self.importers.get(&asset_ty_id) else { + let Some(importer) = self.importers.get(&asset_ty_id).cloned() else { tracing::error!("No importer exists for asset"); return; }; - self.import_work_queue.add_work(Work { - func: |ImportWorkUserData { - import_work_msg_sender, - asset_path, - asset_settings, - importer, - }| { + let import_work_msg_sender = self.import_work_msg_sender.clone(); + + let asset_path = label.path.to_path_buf(); + + self.import_work_thread_pool.spawn(move || { + let result = catch_unwind(|| { if let Err(err) = importer.call( import_work_msg_sender, asset_path.as_path(), - asset_settings.as_deref(), + asset_settings + .as_ref() + .map(|asset_settings| asset_settings as &(dyn Any + Send + Sync)), ) { tracing::error!( "Failed to import asset {}: {:#}", @@ -590,15 +600,22 @@ impl Assets crate::Error::new(err) ); } - }, - user_data: ImportWorkUserData { - import_work_msg_sender: self.import_work_msg_sender.clone(), - asset_path: label.path.to_path_buf(), - asset_settings: asset_settings.map(|asset_settings| { - Box::new(asset_settings) as Box<dyn Any + Send + Sync> - }), - importer: importer.clone(), - }, + }); + + if let Err(panic_msg) = result { + let panic_msg = if let Some(msg) = panic_msg.downcast_ref::<&str>() { + msg + } else if let Some(msg) = panic_msg.downcast_ref::<String>() { + msg.as_str() + } else { + "(unknown panic payload type)" + }; + + tracing::error!( + "Failed to import asset {}: Thread panicked: {panic_msg}", + asset_path.display(), + ); + } }); } @@ -827,7 +844,7 @@ impl WrappedImporterFn where InnerFunc: Fn(&mut Submitter<'_>, &Path, Option<&AssetSettings>) -> Result<(), Err>, - AssetSettings: 'static, + AssetSettings: UnwindSafe + RefUnwindSafe + 'static, Err: std::error::Error + Send + Sync + 'static, { assert_eq!(size_of::<InnerFunc>(), 0); @@ -892,7 +909,6 @@ impl crate::ecs::extension::Extension for Extension collector.spawn_declared_entity(&HANDLE_ASSETS_PHASE); collector.add_system(*HANDLE_ASSETS_PHASE, add_received_assets); - collector.add_system(*HANDLE_ASSETS_PHASE, check_import_wq_thread_not_panicked); } } @@ -919,28 +935,6 @@ fn add_received_assets(mut assets: Single<Assets>, mut events: Single<Events>) } } -fn check_import_wq_thread_not_panicked(assets: Single<Assets>, mut actions: Actions<'_>) -{ - let Ok(assets) = assets.get() else { - unreachable!(); - }; - - if assets.import_work_queue.get_thread_panic().is_some() { - cold_path(); - - actions.stop(); - } -} - -#[derive(Debug)] -struct ImportWorkUserData -{ - import_work_msg_sender: MpscSender<ImportWorkMessage>, - asset_path: PathBuf, - asset_settings: Option<Box<dyn Any + Send + Sync>>, - importer: WrappedImporterFn, -} - #[derive(Debug)] enum ImportWorkMessage { @@ -1055,6 +1049,23 @@ impl hashbrown::Equivalent<LabelOwned> for Label<'_> } } +fn calc_import_work_thread_cnt() -> usize +{ + available_parallelism() + .map(|avail_parallelism| avail_parallelism.get().div_ceil(2)) + .inspect_err(|err| { + tracing::warn!( + concat!( + "Failed to get available parallelism for calculating import work ", + "thread count. {} threads will be used. Error: {}" + ), + BACKUP_IMPORT_WORK_THREAD_CNT, + err + ); + }) + .unwrap_or(BACKUP_IMPORT_WORK_THREAD_CNT) +} + mod store { use std::num::NonZero; diff --git a/engine/src/lib.rs b/engine/src/lib.rs index fd0c43a..0b8409d 100644 --- a/engine/src/lib.rs +++ b/engine/src/lib.rs @@ -12,7 +12,6 @@ use crate::ecs::extension::Extension; use crate::ecs::World; mod util; -mod work_queue; pub mod asset; pub mod camera; diff --git a/engine/src/work_queue.rs b/engine/src/work_queue.rs deleted file mode 100644 index 765aec3..0000000 --- a/engine/src/work_queue.rs +++ /dev/null @@ -1,77 +0,0 @@ -use std::borrow::Cow; -use std::marker::PhantomData; -use std::panic::catch_unwind; -use std::sync::mpsc::{channel as mpsc_channel, Sender as MpscSender}; -use std::sync::{Arc, OnceLock}; -use std::thread::{Builder as ThreadBuilder, JoinHandle as ThreadJoinHandle}; - -pub struct Work<UserData: Send + Sync + 'static> -{ - pub func: fn(UserData), - pub user_data: UserData, -} - -#[derive(Debug)] -pub struct WorkQueue<UserData: Send + Sync + 'static> -{ - work_sender: MpscSender<Work<UserData>>, - thread_panic: Arc<OnceLock<Box<str>>>, - _thread: ThreadJoinHandle<()>, - _pd: PhantomData<UserData>, -} - -impl<UserData: Send + Sync + 'static> WorkQueue<UserData> -{ - pub fn new(name: &str) -> Self - { - let (work_sender, work_receiver) = mpsc_channel::<Work<UserData>>(); - - let thread_panic = Arc::new(OnceLock::new()); - - let thread_panic_b = thread_panic.clone(); - - Self { - work_sender, - thread_panic, - _thread: ThreadBuilder::new() - .name(name.to_string()) - .spawn(move || { - if let Err(panic_err) = catch_unwind(|| { - while let Ok(work) = work_receiver.recv() { - (work.func)(work.user_data); - } - }) { - let panic_message: Cow<'static, str> = - if let Some(panic_message) = - panic_err.downcast_ref::<&'static str>() - { - (*panic_message).into() - } else if let Some(panic_message) = - panic_err.downcast_ref::<String>() - { - panic_message.clone().into() - } else { - "(unknown panic payload type)".into() - }; - - let _ = thread_panic_b - .set(panic_message.into_owned().into_boxed_str()); - } - }) - .expect("Failed to create work queue thread"), - _pd: PhantomData, - } - } - - pub fn get_thread_panic(&self) -> Option<&str> - { - self.thread_panic.get().map(|thread_panic| &**thread_panic) - } - - pub fn add_work(&self, work: Work<UserData>) - { - if self.work_sender.send(work).is_err() { - tracing::error!("Cannot add work to work queue. Work queue thread is dead"); - } - } -} |