Use rayon as the bevy_task thread pool

crates/bevy_asset/src/processor/mod.rs

@@ -166,7 +166,7 @@ impl AssetProcessor {
         let start_time = std::time::Instant::now();
         debug!("Processing Assets");
         IoTaskPool::get().scope(|scope| {
-            scope.spawn(async move {
+            scope.spawn_async(async move {
                 self.initialize().await.unwrap();
                 for source in self.sources().iter_processed() {
                     self.process_assets_internal(scope, source, PathBuf::from(""))
@@ -316,7 +316,7 @@ impl AssetProcessor {
         error!("AddFolder event cannot be handled in single threaded mode (or WASM) yet.");
         #[cfg(all(not(target_arch = "wasm32"), feature = "multi-threaded"))]
         IoTaskPool::get().scope(|scope| {
-            scope.spawn(async move {
+            scope.spawn_async(async move {
                 self.process_assets_internal(scope, source, path)
                     .await
                     .unwrap();
@@ -445,7 +445,7 @@ impl AssetProcessor {
             } else {
                 // Files without extensions are skipped
                 let processor = self.clone();
-                scope.spawn(async move {
+                scope.spawn_async(async move {
                     processor.process_asset(source, path).await;
                 });
             }
@@ -461,7 +461,7 @@ impl AssetProcessor {
                 for path in check_reprocess_queue.drain(..) {
                     let processor = self.clone();
                     let source = self.get_source(path.source()).unwrap();
-                    scope.spawn(async move {
+                    scope.spawn_async(async move {
                         processor.process_asset(source, path.into()).await;
                     });
                 }

crates/bevy_asset/src/server/loaders.rs

@@ -79,7 +79,7 @@ impl AssetLoaders {
                 MaybeAssetLoader::Ready(_) => unreachable!(),
                 MaybeAssetLoader::Pending { sender, .. } => {
                     IoTaskPool::get()
-                        .spawn(async move {
+                        .spawn_async(async move {
                             let _ = sender.broadcast(loader).await;
                         })
                         .detach();

crates/bevy_asset/src/server/mod.rs

@@ -297,7 +297,7 @@ impl AssetServer {
             let owned_handle = Some(handle.clone().untyped());
             let server = self.clone();
             IoTaskPool::get()
-                .spawn(async move {
+                .spawn_async(async move {
                     if let Err(err) = server.load_internal(owned_handle, path, false, None).await {
                         error!("{}", err);
                     }
@@ -367,7 +367,7 @@ impl AssetServer {
 
         let server = self.clone();
         IoTaskPool::get()
-            .spawn(async move {
+            .spawn_async(async move {
                 let path_clone = path.clone();
                 match server.load_untyped_async(path).await {
                     Ok(handle) => server.send_asset_event(InternalAssetEvent::Loaded {
@@ -552,7 +552,7 @@ impl AssetServer {
         let server = self.clone();
         let path = path.into().into_owned();
         IoTaskPool::get()
-            .spawn(async move {
+            .spawn_async(async move {
                 let mut reloaded = false;
 
                 let requests = server
@@ -691,7 +691,7 @@ impl AssetServer {
         let path = path.into_owned();
         let server = self.clone();
         IoTaskPool::get()
-            .spawn(async move {
+            .spawn_async(async move {
                 let Ok(source) = server.get_source(path.source()) else {
                     error!(
                         "Failed to load {path}. AssetSource {:?} does not exist",

crates/bevy_ecs/src/query/state.rs

@@ -1131,7 +1131,7 @@ impl<D: QueryData, F: QueryFilter> QueryState<D, F> {
     ) {
         // NOTE: If you are changing query iteration code, remember to update the following places, where relevant:
         // QueryIter, QueryIterationCursor, QueryManyIter, QueryCombinationIter, QueryState::for_each_unchecked_manual, QueryState::par_for_each_unchecked_manual
-        bevy_tasks::ComputeTaskPool::get().scope(|scope| {
+        let _: Vec<()> = bevy_tasks::ComputeTaskPool::get().scope(|scope| {
             if D::IS_DENSE && F::IS_DENSE {
                 // SAFETY: We only access table data that has been registered in `self.archetype_component_access`.
                 let tables = &world.storages().tables;
@@ -1145,7 +1145,7 @@ impl<D: QueryData, F: QueryFilter> QueryState<D, F> {
                     while offset < table.entity_count() {
                         let mut func = func.clone();
                         let len = batch_size.min(table.entity_count() - offset);
-                        scope.spawn(async move {
+                        scope.spawn(move || {
                             #[cfg(feature = "trace")]
                             let _span = self.par_iter_span.enter();
                             let table = &world
@@ -1172,7 +1172,7 @@ impl<D: QueryData, F: QueryFilter> QueryState<D, F> {
                     while offset < archetype.len() {
                         let mut func = func.clone();
                         let len = batch_size.min(archetype.len() - offset);
-                        scope.spawn(async move {
+                        scope.spawn(move || {
                             #[cfg(feature = "trace")]
                             let _span = self.par_iter_span.enter();
                             let archetype =

crates/bevy_ecs/src/schedule/executor/multi_threaded.rs

@@ -256,7 +256,7 @@ impl SystemExecutor for MultiThreadedExecutor {
                 // and thread wake-up. Only spawn the task if the executor does not immediately
                 // terminate.
                 if block_on(poll_once(&mut executor)).is_none() {
-                    scope.spawn(executor);
+                    scope.spawn_async(executor);
                 }
             },
         );
@@ -525,7 +525,7 @@ impl MultiThreadedExecutor {
         let system = unsafe { &mut *systems[system_index].get() };
         let sender = self.sender.clone();
         let panic_payload = self.panic_payload.clone();
-        let task = async move {
+        let mut task = move || {
             let res = std::panic::catch_unwind(AssertUnwindSafe(|| {
                 // SAFETY:
                 // - The caller ensures that we have permission to
@@ -550,20 +550,21 @@ impl MultiThreadedExecutor {
             }
         };
 
-        #[cfg(feature = "trace")]
-        let task = task.instrument(
-            self.system_task_metadata[system_index]
-                .system_task_span
-                .clone(),
-        );
-
         let system_meta = &self.system_task_metadata[system_index];
         self.active_access
             .extend(&system_meta.archetype_component_access);
 
         if system_meta.is_send {
             scope.spawn(task);
         } else {
+            let task = async move { task() };
+            #[cfg(feature = "trace")]
+            let task = task.instrument(
+                self.system_task_metadata[system_index]
+                    .system_task_span
+                    .clone(),
+            );
+
             self.local_thread_running = true;
             scope.spawn_on_external(task);
         }

crates/bevy_ecs/src/schedule/mod.rs

@@ -1139,6 +1139,7 @@ mod tests {
 
         /// verify the [`SimpleExecutor`] supports stepping
         #[test]
+        #[allow(deprecated)]
         fn simple_executor() {
             assert_executor_supports_stepping!(ExecutorKind::Simple);
         }

crates/bevy_gltf/src/loader.rs

@@ -354,7 +354,7 @@ async fn load_gltf<'a, 'b, 'c>(
                     let parent_path = load_context.path().parent().unwrap();
                     let linear_textures = &linear_textures;
                     let buffer_data = &buffer_data;
-                    scope.spawn(async move {
+                    scope.spawn_async(async move {
                         load_image(
                             gltf_texture,
                             buffer_data,

crates/bevy_render/src/pipelined_rendering.rs

@@ -149,7 +149,7 @@ impl Plugin for PipelinedRenderingPlugin {
                 // run a scope here to allow main world to use this thread while it's waiting for the render app
                 let sent_app = compute_task_pool
                     .scope(|s| {
-                        s.spawn(async { app_to_render_receiver.recv().await });
+                        s.spawn_async(async { app_to_render_receiver.recv().await });
                     })
                     .pop();
                 let Some(Ok(mut render_app)) = sent_app else {
@@ -182,7 +182,7 @@ fn update_rendering(app_world: &mut World, _sub_app: &mut App) {
             // while we wait for the render world to be received.
             let mut render_app = ComputeTaskPool::get()
                 .scope_with_executor(true, Some(&*main_thread_executor.0), |s| {
-                    s.spawn(async { render_channels.recv().await });
+                    s.spawn_async(async { render_channels.recv().await });
                 })
                 .pop()
                 .unwrap();

crates/bevy_render/src/render_resource/pipeline_cache.rs

@@ -941,7 +941,9 @@ fn create_pipeline_task(
     sync: bool,
 ) -> CachedPipelineState {
     if !sync {
-        return CachedPipelineState::Creating(bevy_tasks::AsyncComputeTaskPool::get().spawn(task));
+        return CachedPipelineState::Creating(
+            bevy_tasks::AsyncComputeTaskPool::get().spawn_async(task),
+        );
     }
 
     match futures_lite::future::block_on(task) {

crates/bevy_render/src/renderer/mod.rs

@@ -405,7 +405,7 @@ impl<'w> RenderContext<'w> {
                             command_buffers
                                 .push((i, command_buffer_generation_task(render_device)));
                         } else {
-                            task_pool.spawn(async move {
+                            task_pool.spawn_async(async move {
                                 (i, command_buffer_generation_task(render_device))
                             });
                         }

crates/bevy_render/src/view/window/screenshot.rs

@@ -367,7 +367,7 @@ pub(crate) fn collect_screenshots(world: &mut World) {
                 ));
             };
 
-            AsyncComputeTaskPool::get().spawn(finish).detach();
+            AsyncComputeTaskPool::get().spawn_async(finish).detach();
         }
     }
 }

crates/bevy_tasks/Cargo.toml

@@ -14,10 +14,11 @@ multi-threaded = []
 [dependencies]
 futures-lite = "2.0.1"
 async-executor = "1.7.2"
-async-channel = "2.2.0"
 async-io = { version = "2.0.0", optional = true }
 async-task = "4.2.0"
 concurrent-queue = "2.0.0"
+rayon-core = "1.0"
+parking = "2.2"
 
 [target.'cfg(target_arch = "wasm32")'.dependencies]
 wasm-bindgen-futures = "0.4"

crates/bevy_tasks/src/iter/mod.rs

@@ -36,7 +36,7 @@ where
     fn count(mut self, pool: &TaskPool) -> usize {
         pool.scope(|s| {
             while let Some(batch) = self.next_batch() {
-                s.spawn(async move { batch.count() });
+                s.spawn_async(async move { batch.count() });
             }
         })
         .iter()
@@ -108,7 +108,7 @@ where
         pool.scope(|s| {
             while let Some(batch) = self.next_batch() {
                 let newf = f.clone();
-                s.spawn(async move {
+                s.spawn_async(async move {
                     batch.for_each(newf);
                 });
             }
@@ -197,7 +197,7 @@ where
     {
         pool.scope(|s| {
             while let Some(batch) = self.next_batch() {
-                s.spawn(async move { batch.collect::<Vec<_>>() });
+                s.spawn_async(async move { batch.collect::<Vec<_>>() });
             }
         })
         .into_iter()
@@ -219,7 +219,7 @@ where
         pool.scope(|s| {
             while let Some(batch) = self.next_batch() {
                 let newf = f.clone();
-                s.spawn(async move { batch.partition::<Vec<_>, F>(newf) });
+                s.spawn_async(async move { batch.partition::<Vec<_>, F>(newf) });
             }
         })
         .into_iter()
@@ -246,7 +246,7 @@ where
             while let Some(batch) = self.next_batch() {
                 let newf = f.clone();
                 let newi = init.clone();
-                s.spawn(async move { batch.fold(newi, newf) });
+                s.spawn_async(async move { batch.fold(newi, newf) });
             }
         })
     }
@@ -263,7 +263,7 @@ where
         pool.scope(|s| {
             while let Some(mut batch) = self.next_batch() {
                 let newf = f.clone();
-                s.spawn(async move { batch.all(newf) });
+                s.spawn_async(async move { batch.all(newf) });
             }
         })
         .into_iter()
@@ -282,7 +282,7 @@ where
         pool.scope(|s| {
             while let Some(mut batch) = self.next_batch() {
                 let newf = f.clone();
-                s.spawn(async move { batch.any(newf) });
+                s.spawn_async(async move { batch.any(newf) });
             }
         })
         .into_iter()
@@ -302,7 +302,7 @@ where
         let poses = pool.scope(|s| {
             while let Some(batch) = self.next_batch() {
                 let mut newf = f.clone();
-                s.spawn(async move {
+                s.spawn_async(async move {
                     let mut len = 0;
                     let mut pos = None;
                     for item in batch {
@@ -334,7 +334,7 @@ where
     {
         pool.scope(|s| {
             while let Some(batch) = self.next_batch() {
-                s.spawn(async move { batch.max() });
+                s.spawn_async(async move { batch.max() });
             }
         })
         .into_iter()
@@ -351,7 +351,7 @@ where
     {
         pool.scope(|s| {
             while let Some(batch) = self.next_batch() {
-                s.spawn(async move { batch.min() });
+                s.spawn_async(async move { batch.min() });
             }
         })
         .into_iter()
@@ -371,7 +371,7 @@ where
         pool.scope(|s| {
             while let Some(batch) = self.next_batch() {
                 let newf = f.clone();
-                s.spawn(async move { batch.max_by_key(newf) });
+                s.spawn_async(async move { batch.max_by_key(newf) });
             }
         })
         .into_iter()
@@ -391,7 +391,7 @@ where
         pool.scope(|s| {
             while let Some(batch) = self.next_batch() {
                 let newf = f.clone();
-                s.spawn(async move { batch.max_by(newf) });
+                s.spawn_async(async move { batch.max_by(newf) });
             }
         })
         .into_iter()
@@ -411,7 +411,7 @@ where
         pool.scope(|s| {
             while let Some(batch) = self.next_batch() {
                 let newf = f.clone();
-                s.spawn(async move { batch.min_by_key(newf) });
+                s.spawn_async(async move { batch.min_by_key(newf) });
             }
         })
         .into_iter()
@@ -431,7 +431,7 @@ where
         pool.scope(|s| {
             while let Some(batch) = self.next_batch() {
                 let newf = f.clone();
-                s.spawn(async move { batch.min_by(newf) });
+                s.spawn_async(async move { batch.min_by(newf) });
             }
         })
         .into_iter()
@@ -484,7 +484,7 @@ where
     {
         pool.scope(|s| {
             while let Some(batch) = self.next_batch() {
-                s.spawn(async move { batch.sum() });
+                s.spawn_async(async move { batch.sum() });
             }
         })
         .into_iter()
@@ -501,7 +501,7 @@ where
     {
         pool.scope(|s| {
             while let Some(batch) = self.next_batch() {
-                s.spawn(async move { batch.product() });
+                s.spawn_async(async move { batch.product() });
             }
         })
         .into_iter()