Immediately poll the executor once before spawning it as a task (#11801)

# Objective
At the start of every schedule run, there's currently a guaranteed piece
of overhead as the async executor spawns the MultithreadeExecutor task
onto one of the ComputeTaskPool threads.

## Solution
Poll the executor once to immediately schedule systems without waiting
for the async executor, then spawn the task if and only if the executor
does not immediately terminate.

On a similar note, having the executor task immediately start executing
a system in the same async task might yield similar results over a
broader set of cases. However, this might be more involved, and may need
a solution like #8304.

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

@@ -3,7 +3,7 @@ use std::{
     sync::{Arc, Mutex},
 };
 
-use bevy_tasks::{ComputeTaskPool, Scope, TaskPool, ThreadExecutor};
+use bevy_tasks::{block_on, poll_once, ComputeTaskPool, Scope, TaskPool, ThreadExecutor};
 use bevy_utils::default;
 use bevy_utils::syncunsafecell::SyncUnsafeCell;
 #[cfg(feature = "trace")]
@@ -227,7 +227,8 @@ impl SystemExecutor for MultiThreadedExecutor {
             |scope| {
                 // the executor itself is a `Send` future so that it can run
                 // alongside systems that claim the local thread
-                let executor = async {
+                #[allow(unused_mut)]
+                let mut executor = Box::pin(async {
                     let world_cell = world.as_unsafe_world_cell();
                     while self.num_completed_systems < self.num_systems {
                         // SAFETY:
@@ -252,13 +253,19 @@ impl SystemExecutor for MultiThreadedExecutor {
                             self.rebuild_active_access();
                         }
                     }
-                };
+                });
 
                 #[cfg(feature = "trace")]
                 let executor_span = info_span!("multithreaded executor");
                 #[cfg(feature = "trace")]
-                let executor = executor.instrument(executor_span);
-                scope.spawn(executor);
+                let mut executor = executor.instrument(executor_span);
+
+                // Immediately poll the task once to avoid the overhead of the executor
+                // 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);
+                }
             },
         );
 

crates/bevy_tasks/src/lib.rs

@@ -30,6 +30,7 @@ pub use thread_executor::{ThreadExecutor, ThreadExecutorTicker};
 pub use async_io::block_on;
 #[cfg(not(feature = "async-io"))]
 pub use futures_lite::future::block_on;
+pub use futures_lite::future::poll_once;
 
 mod iter;
 pub use iter::ParallelIterator;