[Merged by Bors] - Pipelined Rendering

crates/bevy_app/src/app.rs

@@ -67,7 +67,7 @@ pub struct App {
     /// the application's event loop and advancing the [`Schedule`].
     /// Typically, it is not configured manually, but set by one of Bevy's built-in plugins.
     /// See `bevy::winit::WinitPlugin` and [`ScheduleRunnerPlugin`](crate::schedule_runner::ScheduleRunnerPlugin).
-    pub runner: Box<dyn Fn(App)>,
+    pub runner: Box<dyn Fn(App) + Send>, // Send bound is required to make App Send
     /// A container of [`Stage`]s set to be run in a linear order.
     pub schedule: Schedule,
     sub_apps: HashMap<AppLabelId, SubApp>,
@@ -87,10 +87,55 @@ impl Debug for App {
     }
 }
 
-/// Each `SubApp` has its own [`Schedule`] and [`World`], enabling a separation of concerns.
-struct SubApp {
-    app: App,
-    extract: Box<dyn Fn(&mut World, &mut App)>,
+/// A [`SubApp`] contains its own [`Schedule`] and [`World`] separate from the main [`App`].
+/// This is useful for situations where data and data processing should be kept completely separate
+/// from the main application. The primary use of this feature in bevy is to enable pipelined rendering.
+///
+/// # Example
+///
+/// ```rust
+/// # use bevy_app::{App, AppLabel};
+/// # use bevy_ecs::prelude::*;
+///
+/// #[derive(Resource, Default)]
+/// struct Val(pub i32);
+///
+/// #[derive(Debug, Clone, Copy, Hash, PartialEq, Eq, AppLabel)]
+/// struct ExampleApp;
+///
+/// #[derive(Debug, Hash, PartialEq, Eq, Clone, StageLabel)]
+/// struct ExampleStage;
+///
+/// let mut app = App::empty();
+/// // initialize the main app with a value of 0;
+/// app.insert_resource(Val(10));
+///
+/// // create a app with a resource and a single stage
+/// let mut sub_app = App::empty();
+/// sub_app.insert_resource(Val(100));
+/// let mut example_stage = SystemStage::single_threaded();
+/// example_stage.add_system(|counter: Res<Val>| {
+///     // since we assigned the value from the main world in extract
+///     // we see that value instead of 100
+///     assert_eq!(counter.0, 10);
+/// });
+/// sub_app.add_stage(ExampleStage, example_stage);
+///
+/// // add the sub_app to the app
+/// app.add_sub_app(ExampleApp, sub_app, |main_world, sub_app| {
+///     sub_app.world.resource_mut::<Val>().0 = main_world.resource::<Val>().0;
+/// });
+///
+/// // This will run the schedules once, since we're using the default runner
+/// app.run();
+/// ```
+pub struct SubApp {
+    /// The [`SubApp`]'s instance of [`App`]
+    pub app: App,
+
+    /// A function that allows access to both the [`SubApp`] [`World`] and the main [`App`]. This is
+    /// useful for moving data between the sub app and the main app.
+    pub extract: Box<dyn Fn(&mut World, &mut App) + Send>,
 }
 
 impl SubApp {
@@ -161,11 +206,14 @@ impl App {
     ///
     /// See [`add_sub_app`](Self::add_sub_app) and [`run_once`](Schedule::run_once) for more details.
     pub fn update(&mut self) {
-        #[cfg(feature = "trace")]
-        let _bevy_frame_update_span = info_span!("frame").entered();
-        self.schedule.run(&mut self.world);
-
-        for sub_app in self.sub_apps.values_mut() {
+        {
+            #[cfg(feature = "trace")]
+            let _bevy_frame_update_span = info_span!("main app").entered();
+            self.schedule.run(&mut self.world);
+        }
+        for (_label, sub_app) in self.sub_apps.iter_mut() {
+            #[cfg(feature = "trace")]
+            let _sub_app_span = info_span!("sub app", name = ?_label).entered();
             sub_app.extract(&mut self.world);
             sub_app.run();
         }
@@ -832,7 +880,7 @@ impl App {
     /// App::new()
     ///     .set_runner(my_runner);
     /// ```
-    pub fn set_runner(&mut self, run_fn: impl Fn(App) + 'static) -> &mut Self {
+    pub fn set_runner(&mut self, run_fn: impl Fn(App) + 'static + Send) -> &mut Self {
         self.runner = Box::new(run_fn);
         self
     }
@@ -1017,14 +1065,15 @@ impl App {
 
     /// Adds an [`App`] as a child of the current one.
     ///
-    /// The provided function `sub_app_runner` is called by the [`update`](Self::update) method. The [`World`]
+    /// The provided function `extract` is normally called by the [`update`](Self::update) method.
+    /// After extract is called, the [`Schedule`] of the sub app is run. The [`World`]
     /// parameter represents the main app world, while the [`App`] parameter is just a mutable
     /// reference to the `SubApp` itself.
     pub fn add_sub_app(
         &mut self,
         label: impl AppLabel,
         app: App,
-        extract: impl Fn(&mut World, &mut App) + 'static,
+        extract: impl Fn(&mut World, &mut App) + 'static + Send,
     ) -> &mut Self {
         self.sub_apps.insert(
             label.as_label(),
@@ -1070,6 +1119,16 @@ impl App {
         }
     }
 
+    /// Inserts an existing sub app into the app
+    pub fn insert_sub_app(&mut self, label: impl AppLabel, sub_app: SubApp) {
+        self.sub_apps.insert(label.as_label(), sub_app);
+    }
+
+    /// Removes a sub app from the app. Returns [`None`] if the label doesn't exist.
+    pub fn remove_sub_app(&mut self, label: impl AppLabel) -> Option<SubApp> {
+        self.sub_apps.remove(&label.as_label())
+    }
+
     /// Retrieves a `SubApp` inside this [`App`] with the given label, if it exists. Otherwise returns
     /// an [`Err`] containing the given label.
     pub fn get_sub_app(&self, label: impl AppLabel) -> Result<&App, impl AppLabel> {

crates/bevy_ecs/src/schedule/executor_parallel.rs

@@ -1,11 +1,15 @@
+use std::sync::Arc;
+
+use crate as bevy_ecs;
 use crate::{
     archetype::ArchetypeComponentId,
     query::Access,
     schedule::{ParallelSystemExecutor, SystemContainer},
+    system::Resource,
     world::World,
 };
 use async_channel::{Receiver, Sender};
-use bevy_tasks::{ComputeTaskPool, Scope, TaskPool};
+use bevy_tasks::{ComputeTaskPool, Scope, TaskPool, ThreadExecutor};
 #[cfg(feature = "trace")]
 use bevy_utils::tracing::Instrument;
 use event_listener::Event;
@@ -14,6 +18,16 @@ use fixedbitset::FixedBitSet;
 #[cfg(test)]
 use scheduling_event::*;
 
+/// New-typed [`ThreadExecutor`] [`Resource`] that is used to run systems on the main thread
+#[derive(Resource, Default, Clone)]
+pub struct MainThreadExecutor(pub Arc<ThreadExecutor<'static>>);
+
+impl MainThreadExecutor {
+    pub fn new() -> Self {
+        MainThreadExecutor(Arc::new(ThreadExecutor::new()))
+    }
+}
+
 struct SystemSchedulingMetadata {
     /// Used to signal the system's task to start the system.
     start: Event,
@@ -124,40 +138,46 @@ impl ParallelSystemExecutor for ParallelExecutor {
             }
         }
 
-        ComputeTaskPool::init(TaskPool::default).scope(|scope| {
-            self.prepare_systems(scope, systems, world);
-            if self.should_run.count_ones(..) == 0 {
-                return;
-            }
-            let parallel_executor = async {
-                // All systems have been ran if there are no queued or running systems.
-                while 0 != self.queued.count_ones(..) + self.running.count_ones(..) {
-                    self.process_queued_systems();
-                    // Avoid deadlocking if no systems were actually started.
-                    if self.running.count_ones(..) != 0 {
-                        // Wait until at least one system has finished.
-                        let index = self
-                            .finish_receiver
-                            .recv()
-                            .await
-                            .unwrap_or_else(|error| unreachable!("{}", error));
-                        self.process_finished_system(index);
-                        // Gather other systems than may have finished.
-                        while let Ok(index) = self.finish_receiver.try_recv() {
+        let thread_executor = world.get_resource::<MainThreadExecutor>().map(|e| &*e.0);
+
+        ComputeTaskPool::init(TaskPool::default).scope_with_executor(
+            false,
+            thread_executor,
+            |scope| {
+                self.prepare_systems(scope, systems, world);
+                if self.should_run.count_ones(..) == 0 {
+                    return;
+                }
+                let parallel_executor = async {
+                    // All systems have been ran if there are no queued or running systems.
+                    while 0 != self.queued.count_ones(..) + self.running.count_ones(..) {
+                        self.process_queued_systems();
+                        // Avoid deadlocking if no systems were actually started.
+                        if self.running.count_ones(..) != 0 {
+                            // Wait until at least one system has finished.
+                            let index = self
+                                .finish_receiver
+                                .recv()
+                                .await
+                                .unwrap_or_else(|error| unreachable!("{}", error));
                             self.process_finished_system(index);
+                            // Gather other systems than may have finished.
+                            while let Ok(index) = self.finish_receiver.try_recv() {
+                                self.process_finished_system(index);
+                            }
+                            // At least one system has finished, so active access is outdated.
+                            self.rebuild_active_access();
                         }
-                        // At least one system has finished, so active access is outdated.
-                        self.rebuild_active_access();
+                        self.update_counters_and_queue_systems();
                     }
-                    self.update_counters_and_queue_systems();
-                }
-            };
-            #[cfg(feature = "trace")]
-            let span = bevy_utils::tracing::info_span!("parallel executor");
-            #[cfg(feature = "trace")]
-            let parallel_executor = parallel_executor.instrument(span);
-            scope.spawn(parallel_executor);
-        });
+                };
+                #[cfg(feature = "trace")]
+                let span = bevy_utils::tracing::info_span!("parallel executor");
+                #[cfg(feature = "trace")]
+                let parallel_executor = parallel_executor.instrument(span);
+                scope.spawn(parallel_executor);
+            },
+        );
     }
 }
 

crates/bevy_internal/src/default_plugins.rs

@@ -68,6 +68,12 @@ impl PluginGroup for DefaultPlugins {
                 // NOTE: Load this after renderer initialization so that it knows about the supported
                 // compressed texture formats
                 .add(bevy_render::texture::ImagePlugin::default());
+
+            #[cfg(not(target_arch = "wasm32"))]
+            {
+                group = group
+                    .add(bevy_render::pipelined_rendering::PipelinedRenderingPlugin::default());
+            }
         }
 
         #[cfg(feature = "bevy_core_pipeline")]

crates/bevy_render/Cargo.toml

@@ -44,6 +44,7 @@ bevy_time = { path = "../bevy_time", version = "0.9.0" }
 bevy_transform = { path = "../bevy_transform", version = "0.9.0" }
 bevy_window = { path = "../bevy_window", version = "0.9.0" }
 bevy_utils = { path = "../bevy_utils", version = "0.9.0" }
+bevy_tasks = { path = "../bevy_tasks", version = "0.9.0" }
 
 # rendering
 image = { version = "0.24", default-features = false }
@@ -76,3 +77,4 @@ basis-universal = { version = "0.2.0", optional = true }
 encase = { version = "0.4", features = ["glam"] }
 # For wgpu profiling using tracing. Use `RUST_LOG=info` to also capture the wgpu spans.
 profiling = { version = "1", features = ["profile-with-tracing"], optional = true }
+async-channel = "1.4"

crates/bevy_render/src/lib.rs

@@ -10,6 +10,7 @@ mod extract_param;
 pub mod extract_resource;
 pub mod globals;
 pub mod mesh;
+pub mod pipelined_rendering;
 pub mod primitives;
 pub mod render_asset;
 pub mod render_graph;
@@ -71,6 +72,9 @@ pub enum RenderStage {
     /// running the next frame while rendering the current frame.
     Extract,
 
+    /// A stage for applying the commands from the [`Extract`] stage
+    ExtractCommands,
+
     /// Prepare render resources from the extracted data for the GPU.
     Prepare,
 
@@ -190,8 +194,14 @@ impl Plugin for RenderPlugin {
             // after access to the main world is removed
             // See also https://github.com/bevyengine/bevy/issues/5082
             extract_stage.set_apply_buffers(false);
+
+            // This stage applies the commands from the extract stage while the render schedule
+            // is running in parallel with the main app.
+            let mut extract_commands_stage = SystemStage::parallel();
+            extract_commands_stage.add_system(apply_extract_commands.at_start());
             render_app
                 .add_stage(RenderStage::Extract, extract_stage)
+                .add_stage(RenderStage::ExtractCommands, extract_commands_stage)
                 .add_stage(RenderStage::Prepare, SystemStage::parallel())
                 .add_stage(RenderStage::Queue, SystemStage::parallel())
                 .add_stage(RenderStage::PhaseSort, SystemStage::parallel())
@@ -222,7 +232,7 @@ impl Plugin for RenderPlugin {
 
             app.add_sub_app(RenderApp, render_app, move |app_world, render_app| {
                 #[cfg(feature = "trace")]
-                let _render_span = bevy_utils::tracing::info_span!("renderer subapp").entered();
+                let _render_span = bevy_utils::tracing::info_span!("extract main app to render subapp").entered();
                 {
                     #[cfg(feature = "trace")]
                     let _stage_span =
@@ -308,10 +318,12 @@ fn extract(app_world: &mut World, render_app: &mut App) {
             let inserted_world = render_world.remove_resource::<MainWorld>().unwrap();
             let scratch_world = std::mem::replace(app_world, inserted_world.0);
             app_world.insert_resource(ScratchMainWorld(scratch_world));
-
-            // Note: We apply buffers (read, Commands) after the `MainWorld` has been removed from the render app's world
-            // so that in future, pipelining will be able to do this too without any code relying on it.
-            // see <https://github.com/bevyengine/bevy/issues/5082>
-            extract_stage.0.apply_buffers(render_world);
         });
 }
+
+// system for render app to apply the extract commands
+fn apply_extract_commands(world: &mut World) {
+    world.resource_scope(|world, mut extract_stage: Mut<ExtractStage>| {
+        extract_stage.0.apply_buffers(world);
+    });
+}