Share AssetSources between the AssetServer, the asset processor, and the asset processor's internal asset server. (#21763)

# Objective

- Previously, we would build the sources up to 3 times with different
options: once for the regular asset server, once for the asset
processor, and once for the asset processor's internal asset server.
- This is a step towards #21758 (since now adding a source to this
shared `AssetSources` will be reflected in all the uses.

## Solution

1) Skip all the hot-reloading polling if `watch_for_changes` is false.
If we don't do this, sharing the sources between the asset server and
the processor-internal asset server will result in two tasks consuming
asset events, so the regular asset server will miss asset events.
2) Move the state of the asset processor into a separate struct that we
then Arc. We need to be able to gate the processed asset reader, but we
can't create an asset processor without the sources. So instead we allow
ourselves to create the state first, so that we can gate on that state,
and then create the processor with that state.
3) Split the processed reader into a gated and an ungated form. The gate
first blocks on the processor being done initialized, followed by gating
on the per-asset lock. However the asset processor needs to iterate
through all the directories in order to finish initializing. So just
unconditionally gating doesn't work - we hit a deadlock. So we provide
access to the ungated processed reader, so that the processor can use
that to initialize its state.
4) Finally do the sharing!

One thing I'm starting in this PR is making things more private. For
example, it's not clear why `ProcessorGatedReader` was `pub`. I've also
made `AssetSources::gate_on_processor` no longer `pub`. This does mean
that users can no longer initialize their own `AssetServer` properly
(e.g., gated on the processor), but I don't think we should really
support this - our focus should be on the `AssetServer` initialized by
`AssetPlugin`, not hypothetical uses where someone wants to insert their
own `AssetServer`.

## Testing

- CI

Author: andriyDev

crates/bevy_asset/src/io/processor_gated.rs

@@ -1,6 +1,6 @@
 use crate::{
     io::{AssetReader, AssetReaderError, AssetSourceId, PathStream, Reader},
-    processor::{AssetProcessorData, ProcessStatus},
+    processor::{ProcessStatus, ProcessingState},
     AssetPath,
 };
 use alloc::{borrow::ToOwned, boxed::Box, sync::Arc, vec::Vec};
@@ -16,46 +16,33 @@ use super::{AsyncSeekForward, ErasedAssetReader};
 /// given path until that path has been processed by [`AssetProcessor`].
 ///
 /// [`AssetProcessor`]: crate::processor::AssetProcessor
-pub struct ProcessorGatedReader {
-    reader: Box<dyn ErasedAssetReader>,
+pub(crate) struct ProcessorGatedReader {
+    reader: Arc<dyn ErasedAssetReader>,
     source: AssetSourceId<'static>,
-    processor_data: Arc<AssetProcessorData>,
+    processing_state: Arc<ProcessingState>,
 }
 
 impl ProcessorGatedReader {
     /// Creates a new [`ProcessorGatedReader`].
-    pub fn new(
+    pub(crate) fn new(
         source: AssetSourceId<'static>,
-        reader: Box<dyn ErasedAssetReader>,
-        processor_data: Arc<AssetProcessorData>,
+        reader: Arc<dyn ErasedAssetReader>,
+        processing_state: Arc<ProcessingState>,
     ) -> Self {
         Self {
             source,
-            processor_data,
             reader,
+            processing_state,
         }
     }
-
-    /// Gets a "transaction lock" that can be used to ensure no writes to asset or asset meta occur
-    /// while it is held.
-    async fn get_transaction_lock(
-        &self,
-        path: &AssetPath<'static>,
-    ) -> Result<RwLockReadGuardArc<()>, AssetReaderError> {
-        let infos = self.processor_data.asset_infos.read().await;
-        let info = infos
-            .get(path)
-            .ok_or_else(|| AssetReaderError::NotFound(path.path().to_owned()))?;
-        Ok(info.file_transaction_lock.read_arc().await)
-    }
 }
 
 impl AssetReader for ProcessorGatedReader {
     async fn read<'a>(&'a self, path: &'a Path) -> Result<impl Reader + 'a, AssetReaderError> {
         let asset_path = AssetPath::from(path.to_path_buf()).with_source(self.source.clone());
         trace!("Waiting for processing to finish before reading {asset_path}");
         let process_result = self
-            .processor_data
+            .processing_state
             .wait_until_processed(asset_path.clone())
             .await;
         match process_result {
@@ -65,7 +52,10 @@ impl AssetReader for ProcessorGatedReader {
             }
         }
         trace!("Processing finished with {asset_path}, reading {process_result:?}",);
-        let lock = self.get_transaction_lock(&asset_path).await?;
+        let lock = self
+            .processing_state
+            .get_transaction_lock(&asset_path)
+            .await?;
         let asset_reader = self.reader.read(path).await?;
         let reader = TransactionLockedReader::new(asset_reader, lock);
         Ok(reader)
@@ -75,7 +65,7 @@ impl AssetReader for ProcessorGatedReader {
         let asset_path = AssetPath::from(path.to_path_buf()).with_source(self.source.clone());
         trace!("Waiting for processing to finish before reading meta for {asset_path}",);
         let process_result = self
-            .processor_data
+            .processing_state
             .wait_until_processed(asset_path.clone())
             .await;
         match process_result {
@@ -85,7 +75,10 @@ impl AssetReader for ProcessorGatedReader {
             }
         }
         trace!("Processing finished with {process_result:?}, reading meta for {asset_path}",);
-        let lock = self.get_transaction_lock(&asset_path).await?;
+        let lock = self
+            .processing_state
+            .get_transaction_lock(&asset_path)
+            .await?;
         let meta_reader = self.reader.read_meta(path).await?;
         let reader = TransactionLockedReader::new(meta_reader, lock);
         Ok(reader)
@@ -99,7 +92,7 @@ impl AssetReader for ProcessorGatedReader {
             "Waiting for processing to finish before reading directory {:?}",
             path
         );
-        self.processor_data.wait_until_finished().await;
+        self.processing_state.wait_until_finished().await;
         trace!("Processing finished, reading directory {:?}", path);
         let result = self.reader.read_directory(path).await?;
         Ok(result)
@@ -110,7 +103,7 @@ impl AssetReader for ProcessorGatedReader {
             "Waiting for processing to finish before reading directory {:?}",
             path
         );
-        self.processor_data.wait_until_finished().await;
+        self.processing_state.wait_until_finished().await;
         trace!("Processing finished, getting directory status {:?}", path);
         let result = self.reader.is_directory(path).await?;
         Ok(result)

crates/bevy_asset/src/io/source.rs

@@ -1,6 +1,6 @@
 use crate::{
     io::{processor_gated::ProcessorGatedReader, AssetSourceEvent, AssetWatcher},
-    processor::AssetProcessorData,
+    processor::ProcessingState,
 };
 use alloc::{
     boxed::Box,
@@ -180,7 +180,12 @@ impl AssetSourceBuilder {
             id: id.clone(),
             reader,
             writer,
-            processed_reader: self.processed_reader.as_mut().map(|r| r()),
+            processed_reader: self
+                .processed_reader
+                .as_mut()
+                .map(|r| r())
+                .map(Into::<Arc<_>>::into),
+            ungated_processed_reader: None,
             processed_writer,
             event_receiver: None,
             watcher: None,
@@ -386,7 +391,12 @@ pub struct AssetSource {
     id: AssetSourceId<'static>,
     reader: Box<dyn ErasedAssetReader>,
     writer: Option<Box<dyn ErasedAssetWriter>>,
-    processed_reader: Option<Box<dyn ErasedAssetReader>>,
+    processed_reader: Option<Arc<dyn ErasedAssetReader>>,
+    /// The ungated version of `processed_reader`.
+    ///
+    /// This allows the processor to read all the processed assets to initialize itself without
+    /// being gated on itself (causing a deadlock).
+    ungated_processed_reader: Option<Arc<dyn ErasedAssetReader>>,
     processed_writer: Option<Box<dyn ErasedAssetWriter>>,
     watcher: Option<Box<dyn AssetWatcher>>,
     processed_watcher: Option<Box<dyn AssetWatcher>>,
@@ -425,6 +435,13 @@ impl AssetSource {
             .ok_or_else(|| MissingProcessedAssetReaderError(self.id.clone_owned()))
     }
 
+    /// Return's this source's ungated processed [`AssetReader`](crate::io::AssetReader), if it
+    /// exists.
+    #[inline]
+    pub(crate) fn ungated_processed_reader(&self) -> Option<&dyn ErasedAssetReader> {
+        self.ungated_processed_reader.as_deref()
+    }
+
     /// Return's this source's processed [`AssetWriter`](crate::io::AssetWriter), if it exists.
     #[inline]
     pub fn processed_writer(
@@ -560,12 +577,13 @@ impl AssetSource {
 
     /// This will cause processed [`AssetReader`](crate::io::AssetReader) futures (such as [`AssetReader::read`](crate::io::AssetReader::read)) to wait until
     /// the [`AssetProcessor`](crate::AssetProcessor) has finished processing the requested asset.
-    pub fn gate_on_processor(&mut self, processor_data: Arc<AssetProcessorData>) {
+    pub(crate) fn gate_on_processor(&mut self, processing_state: Arc<ProcessingState>) {
         if let Some(reader) = self.processed_reader.take() {
-            self.processed_reader = Some(Box::new(ProcessorGatedReader::new(
+            self.ungated_processed_reader = Some(reader.clone());
+            self.processed_reader = Some(Arc::new(ProcessorGatedReader::new(
                 self.id(),
                 reader,
-                processor_data,
+                processing_state,
             )));
         }
     }
@@ -622,9 +640,9 @@ impl AssetSources {
 
     /// This will cause processed [`AssetReader`](crate::io::AssetReader) futures (such as [`AssetReader::read`](crate::io::AssetReader::read)) to wait until
     /// the [`AssetProcessor`](crate::AssetProcessor) has finished processing the requested asset.
-    pub fn gate_on_processor(&mut self, processor_data: Arc<AssetProcessorData>) {
+    pub(crate) fn gate_on_processor(&mut self, processing_state: Arc<ProcessingState>) {
         for source in self.iter_processed_mut() {
-            source.gate_on_processor(processor_data.clone());
+            source.gate_on_processor(processing_state.clone());
         }
     }
 }

crates/bevy_asset/src/lib.rs

@@ -375,7 +375,7 @@ impl Plugin for AssetPlugin {
                     let sources = builders.build_sources(watch, false);
 
                     app.insert_resource(AssetServer::new_with_meta_check(
-                        sources,
+                        Arc::new(sources),
                         AssetServerMode::Unprocessed,
                         self.meta_check.clone(),
                         watch,
@@ -388,9 +388,7 @@ impl Plugin for AssetPlugin {
                         .unwrap_or(cfg!(feature = "asset_processor"));
                     if use_asset_processor {
                         let mut builders = app.world_mut().resource_mut::<AssetSourceBuilders>();
-                        let processor = AssetProcessor::new(&mut builders);
-                        let mut sources = builders.build_sources(false, watch);
-                        sources.gate_on_processor(processor.data.clone());
+                        let (processor, sources) = AssetProcessor::new(&mut builders, watch);
                         // the main asset server shares loaders with the processor asset server
                         app.insert_resource(AssetServer::new_with_loaders(
                             sources,
@@ -406,7 +404,7 @@ impl Plugin for AssetPlugin {
                         let mut builders = app.world_mut().resource_mut::<AssetSourceBuilders>();
                         let sources = builders.build_sources(false, watch);
                         app.insert_resource(AssetServer::new_with_meta_check(
-                            sources,
+                            Arc::new(sources),
                             AssetServerMode::Processed,
                             AssetMetaCheck::Always,
                             watch,