Wrap the AssetSources in a RwLock.

Author: andriyDev

crates/bevy_asset/src/lib.rs

@@ -225,6 +225,7 @@ use bevy_ecs::{
 use bevy_platform::collections::HashSet;
 use bevy_reflect::{FromReflect, GetTypeRegistration, Reflect, TypePath};
 use core::any::TypeId;
+use std::sync::RwLock;
 use tracing::error;
 
 /// Provides "asset" loading and processing functionality. An [`Asset`] is a "runtime value" that is loaded from an [`AssetSource`],
@@ -375,7 +376,7 @@ impl Plugin for AssetPlugin {
                     let sources = builders.build_sources(watch, false, None);
 
                     app.insert_resource(AssetServer::new_with_meta_check(
-                        Arc::new(sources),
+                        Arc::new(RwLock::new(sources)),
                         AssetServerMode::Unprocessed,
                         self.meta_check.clone(),
                         watch,
@@ -404,7 +405,7 @@ impl Plugin for AssetPlugin {
                         let mut builders = app.world_mut().resource_mut::<AssetSourceBuilders>();
                         let sources = builders.build_sources(false, watch, None);
                         app.insert_resource(AssetServer::new_with_meta_check(
-                            Arc::new(sources),
+                            Arc::new(RwLock::new(sources)),
                             AssetServerMode::Processed,
                             AssetMetaCheck::Always,
                             watch,

crates/bevy_asset/src/processor/mod.rs

@@ -115,7 +115,8 @@ pub struct AssetProcessorData {
     processors: RwLock<HashMap<&'static str, Arc<dyn ErasedProcessor>>>,
     /// Default processors for file extensions
     default_processors: RwLock<HashMap<Box<str>, &'static str>>,
-    sources: Arc<AssetSources>,
+    /// The asset sources for which this processor will process assets.
+    sources: Arc<RwLock<AssetSources>>,
 }
 
 /// The current state of processing, including the overall state and the state of all assets.
@@ -137,10 +138,10 @@ impl AssetProcessor {
     pub fn new(
         sources: &mut AssetSourceBuilders,
         watch_processed: bool,
-    ) -> (Self, Arc<AssetSources>) {
+    ) -> (Self, Arc<RwLock<AssetSources>>) {
         let state = Arc::new(ProcessingState::new());
         let sources = sources.build_sources(true, watch_processed, Some(state.clone()));
-        let sources = Arc::new(sources);
+        let sources = Arc::new(RwLock::new(sources));
 
         let data = Arc::new(AssetProcessorData::new(sources.clone(), state));
         // The asset processor uses its own asset server with its own id space
@@ -176,11 +177,15 @@ impl AssetProcessor {
         &self,
         id: impl Into<AssetSourceId<'a>>,
     ) -> Result<Arc<AssetSource>, MissingAssetSourceError> {
-        self.data.sources.get(id.into())
+        self.data
+            .sources
+            .read()
+            .unwrap_or_else(PoisonError::into_inner)
+            .get(id.into())
     }
 
     #[inline]
-    pub fn sources(&self) -> &AssetSources {
+    pub fn sources(&self) -> &RwLock<AssetSources> {
         &self.data.sources
     }
 
@@ -233,11 +238,19 @@ impl AssetProcessor {
                 let start_time = std::time::Instant::now();
                 debug!("Processing Assets");
 
-                processor.initialize().await.unwrap();
+                let sources = processor
+                    .sources()
+                    .read()
+                    .unwrap_or_else(PoisonError::into_inner)
+                    .iter_processed()
+                    .cloned()
+                    .collect::<Vec<_>>();
+
+                processor.initialize(&sources).await.unwrap();
 
                 let (new_task_sender, new_task_receiver) = async_channel::unbounded();
                 processor
-                    .queue_initial_processing_tasks(&new_task_sender)
+                    .queue_initial_processing_tasks(&sources, &new_task_sender)
                     .await;
 
                 // Once all the tasks are queued for the initial processing, start actually
@@ -260,17 +273,18 @@ impl AssetProcessor {
                 debug!("Processing finished in {:?}", end_time - start_time);
 
                 debug!("Listening for changes to source assets");
-                processor.spawn_source_change_event_listeners(&new_task_sender);
+                processor.spawn_source_change_event_listeners(&sources, &new_task_sender);
             })
             .detach();
     }
 
     /// Sends start task events for all assets in all processed sources into `sender`.
     async fn queue_initial_processing_tasks(
         &self,
+        sources: &[Arc<AssetSource>],
         sender: &async_channel::Sender<(AssetSourceId<'static>, PathBuf)>,
     ) {
-        for source in self.sources().iter_processed() {
+        for source in sources {
             self.queue_processing_tasks_for_folder(source, PathBuf::from(""), sender)
                 .await
                 .unwrap();
@@ -281,9 +295,10 @@ impl AssetProcessor {
     /// response.
     fn spawn_source_change_event_listeners(
         &self,
+        sources: &[Arc<AssetSource>],
         sender: &async_channel::Sender<(AssetSourceId<'static>, PathBuf)>,
     ) {
-        for source in self.data.sources.iter_processed() {
+        for source in sources {
             let Some(receiver) = source.event_receiver().cloned() else {
                 continue;
             };
@@ -753,8 +768,8 @@ impl AssetProcessor {
     /// This info will later be used to determine whether or not to re-process an asset
     ///
     /// This will validate transactions and recover failed transactions when necessary.
-    async fn initialize(&self) -> Result<(), InitializeError> {
-        self.validate_transaction_log_and_recover().await;
+    async fn initialize(&self, sources: &[Arc<AssetSource>]) -> Result<(), InitializeError> {
+        self.validate_transaction_log_and_recover(sources).await;
         let mut asset_infos = self.data.processing_state.asset_infos.write().await;
 
         /// Retrieves asset paths recursively. If `clean_empty_folders_writer` is Some, it will be used to clean up empty
@@ -793,7 +808,7 @@ impl AssetProcessor {
             }
         }
 
-        for source in self.sources().iter_processed() {
+        for source in sources {
             let Some(processed_reader) = source.ungated_processed_reader() else {
                 continue;
             };
@@ -1129,7 +1144,7 @@ impl AssetProcessor {
         Ok(ProcessResult::Processed(new_processed_info))
     }
 
-    async fn validate_transaction_log_and_recover(&self) {
+    async fn validate_transaction_log_and_recover(&self, sources: &[Arc<AssetSource>]) {
         let log_factory = self
             .data
             .log_factory
@@ -1203,7 +1218,7 @@ impl AssetProcessor {
 
             if !state_is_valid {
                 error!("Processed asset transaction log state was invalid and unrecoverable for some reason (see previous logs). Removing processed assets and starting fresh.");
-                for source in self.sources().iter_processed() {
+                for source in sources {
                     let Ok(processed_writer) = source.processed_writer() else {
                         continue;
                     };
@@ -1226,7 +1241,10 @@ impl AssetProcessor {
 
 impl AssetProcessorData {
     /// Initializes a new [`AssetProcessorData`] using the given [`AssetSources`].
-    pub(crate) fn new(sources: Arc<AssetSources>, processing_state: Arc<ProcessingState>) -> Self {
+    pub(crate) fn new(
+        sources: Arc<RwLock<AssetSources>>,
+        processing_state: Arc<ProcessingState>,
+    ) -> Self {
         AssetProcessorData {
             processing_state,
             sources,

crates/bevy_asset/src/server/mod.rs

@@ -69,7 +69,7 @@ pub(crate) struct AssetServerData {
     pub(crate) loaders: Arc<RwLock<AssetLoaders>>,
     asset_event_sender: Sender<InternalAssetEvent>,
     asset_event_receiver: Receiver<InternalAssetEvent>,
-    sources: Arc<AssetSources>,
+    sources: Arc<RwLock<AssetSources>>,
     mode: AssetServerMode,
     meta_check: AssetMetaCheck,
     unapproved_path_mode: UnapprovedPathMode,
@@ -91,7 +91,7 @@ impl AssetServer {
     /// Create a new instance of [`AssetServer`]. If `watch_for_changes` is true, the [`AssetReader`](crate::io::AssetReader) storage will watch for changes to
     /// asset sources and hot-reload them.
     pub fn new(
-        sources: Arc<AssetSources>,
+        sources: Arc<RwLock<AssetSources>>,
         mode: AssetServerMode,
         watching_for_changes: bool,
         unapproved_path_mode: UnapprovedPathMode,
@@ -109,7 +109,7 @@ impl AssetServer {
     /// Create a new instance of [`AssetServer`]. If `watch_for_changes` is true, the [`AssetReader`](crate::io::AssetReader) storage will watch for changes to
     /// asset sources and hot-reload them.
     pub fn new_with_meta_check(
-        sources: Arc<AssetSources>,
+        sources: Arc<RwLock<AssetSources>>,
         mode: AssetServerMode,
         meta_check: AssetMetaCheck,
         watching_for_changes: bool,
@@ -126,7 +126,7 @@ impl AssetServer {
     }
 
     pub(crate) fn new_with_loaders(
-        sources: Arc<AssetSources>,
+        sources: Arc<RwLock<AssetSources>>,
         loaders: Arc<RwLock<AssetLoaders>>,
         mode: AssetServerMode,
         meta_check: AssetMetaCheck,
@@ -183,7 +183,11 @@ impl AssetServer {
         &self,
         source: impl Into<AssetSourceId<'a>>,
     ) -> Result<Arc<AssetSource>, MissingAssetSourceError> {
-        self.data.sources.get(source.into())
+        self.data
+            .sources
+            .read()
+            .unwrap_or_else(PoisonError::into_inner)
+            .get(source.into())
     }
 
     /// Returns true if the [`AssetServer`] watches for changes.
@@ -1849,7 +1853,13 @@ pub fn handle_internal_asset_events(world: &mut World) {
             }
         };
 
-        for source in server.data.sources.iter() {
+        for source in server
+            .data
+            .sources
+            .read()
+            .unwrap_or_else(PoisonError::into_inner)
+            .iter()
+        {
             match server.data.mode {
                 AssetServerMode::Unprocessed => {
                     if let Some(receiver) = source.event_receiver() {