Add a SystemParam primitive for deferred mutations; allow #[derive]ing more types of SystemParam (#6817)

# Objective

One pattern to increase parallelism is deferred mutation: instead of directly mutating the world (and preventing other systems from running at the same time), you queue up operations to be applied to the world at the end of the stage. The most common example of this pattern uses the `Commands` SystemParam.

In order to avoid the overhead associated with commands, some power users may want to add their own deferred mutation behavior. To do this, you must implement the unsafe trait `SystemParam`, which interfaces with engine internals in a way that we'd like users to be able to avoid.

## Solution

Add the `Deferred<T>` primitive `SystemParam`, which encapsulates the deferred mutation pattern.
This can be combined with other types of `SystemParam` to safely and ergonomically create powerful custom types.

Essentially, this is just a variant of `Local<T>` which can run code at the end of the stage.

This type is used in the engine to derive `Commands` and `ParallelCommands`, which removes a bunch of unsafe boilerplate.

### Example

```rust
use bevy_ecs::system::{Deferred, SystemBuffer};

/// Sends events with a delay, but may run in parallel with other event writers.
#[derive(SystemParam)]
pub struct BufferedEventWriter<'s, E: Event> {
    queue: Deferred<'s, EventQueue<E>>,
}

struct EventQueue<E>(Vec<E>);

impl<'s, E: Event> BufferedEventWriter<'s, E> {
    /// Queues up an event to be sent at the end of this stage.
    pub fn send(&mut self, event: E) {
        self.queue.0.push(event);
    }
}

// The `SystemBuffer` trait controls how [`Deferred`] gets applied at the end of the stage.
impl<E: Event> SystemBuffer for EventQueue<E> {
    fn apply(&mut self, world: &mut World) {
        let mut events = world.resource_mut::<Events<E>>();
        for e in self.0.drain(..) {
            events.send(e);
        }
    }
}
```

---

## Changelog

+ Added the `SystemParam` type `Deferred<T>`, which can be used to defer `World` mutations. Powered by the new trait `SystemBuffer`.

Author: joseph-gio

crates/bevy_ecs/src/lib.rs

@@ -43,8 +43,8 @@ pub mod prelude {
         system::{
             adapter as system_adapter,
             adapter::{dbg, error, ignore, info, unwrap, warn},
-            Commands, In, IntoPipeSystem, IntoSystem, Local, NonSend, NonSendMut, ParallelCommands,
-            ParamSet, Query, Res, ResMut, Resource, System, SystemParamFunction,
+            Commands, Deferred, In, IntoPipeSystem, IntoSystem, Local, NonSend, NonSendMut,
+            ParallelCommands, ParamSet, Query, Res, ResMut, Resource, System, SystemParamFunction,
         },
         world::{FromWorld, World},
     };

crates/bevy_ecs/src/system/commands/mod.rs

@@ -2,16 +2,18 @@ mod command_queue;
 mod parallel_scope;
 
 use crate::{
+    self as bevy_ecs,
     bundle::Bundle,
     entity::{Entities, Entity},
     world::{FromWorld, World},
 };
+use bevy_ecs_macros::SystemParam;
 use bevy_utils::tracing::{error, info};
 pub use command_queue::CommandQueue;
 pub use parallel_scope::*;
 use std::marker::PhantomData;
 
-use super::Resource;
+use super::{Deferred, Resource, SystemBuffer, SystemMeta};
 
 /// A [`World`] mutation.
 ///
@@ -97,22 +99,31 @@ pub trait Command: Send + 'static {
 /// [`System::apply_buffers`]: crate::system::System::apply_buffers
 /// [`apply_system_buffers`]: crate::schedule::apply_system_buffers
 /// [`Schedule::apply_system_buffers`]: crate::schedule::Schedule::apply_system_buffers
+#[derive(SystemParam)]
 pub struct Commands<'w, 's> {
-    queue: &'s mut CommandQueue,
+    queue: Deferred<'s, CommandQueue>,
     entities: &'w Entities,
 }
 
+impl SystemBuffer for CommandQueue {
+    #[inline]
+    fn apply(&mut self, _system_meta: &SystemMeta, world: &mut World) {
+        #[cfg(feature = "trace")]
+        let _system_span =
+            bevy_utils::tracing::info_span!("system_commands", name = _system_meta.name())
+                .entered();
+        self.apply(world);
+    }
+}
+
 impl<'w, 's> Commands<'w, 's> {
     /// Returns a new `Commands` instance from a [`CommandQueue`] and a [`World`].
     ///
     /// It is not required to call this constructor when using `Commands` as a [system parameter].
     ///
     /// [system parameter]: crate::system::SystemParam
     pub fn new(queue: &'s mut CommandQueue, world: &'w World) -> Self {
-        Self {
-            queue,
-            entities: world.entities(),
-        }
+        Self::new_from_entities(queue, world.entities())
     }
 
     /// Returns a new `Commands` instance from a [`CommandQueue`] and an [`Entities`] reference.
@@ -121,7 +132,10 @@ impl<'w, 's> Commands<'w, 's> {
     ///
     /// [system parameter]: crate::system::SystemParam
     pub fn new_from_entities(queue: &'s mut CommandQueue, entities: &'w Entities) -> Self {
-        Self { queue, entities }
+        Self {
+            queue: Deferred(queue),
+            entities,
+        }
     }
 
     /// Pushes a [`Command`] to the queue for creating a new empty [`Entity`],

crates/bevy_ecs/src/system/commands/parallel_scope.rs

@@ -3,17 +3,16 @@ use std::cell::Cell;
 use thread_local::ThreadLocal;
 
 use crate::{
+    self as bevy_ecs,
     entity::Entities,
     prelude::World,
-    system::{SystemMeta, SystemParam},
+    system::{Deferred, SystemBuffer, SystemMeta, SystemParam},
 };
 
 use super::{CommandQueue, Commands};
 
-/// The internal [`SystemParam`] state of the [`ParallelCommands`] type
-#[doc(hidden)]
 #[derive(Default)]
-pub struct ParallelCommandsState {
+struct ParallelCommandQueue {
     thread_local_storage: ThreadLocal<Cell<CommandQueue>>,
 }
 
@@ -43,41 +42,23 @@ pub struct ParallelCommandsState {
 /// }
 /// # bevy_ecs::system::assert_is_system(parallel_command_system);
 ///```
+#[derive(SystemParam)]
 pub struct ParallelCommands<'w, 's> {
-    state: &'s mut ParallelCommandsState,
+    state: Deferred<'s, ParallelCommandQueue>,
     entities: &'w Entities,
 }
 
-// SAFETY: no component or resource access to report
-unsafe impl SystemParam for ParallelCommands<'_, '_> {
-    type State = ParallelCommandsState;
-    type Item<'w, 's> = ParallelCommands<'w, 's>;
-
-    fn init_state(_: &mut World, _: &mut crate::system::SystemMeta) -> Self::State {
-        ParallelCommandsState::default()
-    }
-
-    fn apply(state: &mut Self::State, _system_meta: &SystemMeta, world: &mut World) {
+impl SystemBuffer for ParallelCommandQueue {
+    #[inline]
+    fn apply(&mut self, _system_meta: &SystemMeta, world: &mut World) {
         #[cfg(feature = "trace")]
         let _system_span =
             bevy_utils::tracing::info_span!("system_commands", name = _system_meta.name())
                 .entered();
-        for cq in &mut state.thread_local_storage {
+        for cq in &mut self.thread_local_storage {
             cq.get_mut().apply(world);
         }
     }
-
-    unsafe fn get_param<'w, 's>(
-        state: &'s mut Self::State,
-        _: &crate::system::SystemMeta,
-        world: &'w World,
-        _: u32,
-    ) -> Self::Item<'w, 's> {
-        ParallelCommands {
-            state,
-            entities: world.entities(),
-        }
-    }
 }
 
 impl<'w, 's> ParallelCommands<'w, 's> {

crates/bevy_ecs/src/system/system_param.rs

@@ -8,7 +8,7 @@ use crate::{
     query::{
         Access, FilteredAccess, FilteredAccessSet, QueryState, ReadOnlyWorldQuery, WorldQuery,
     },
-    system::{CommandQueue, Commands, Query, SystemMeta},
+    system::{Query, SystemMeta},
     world::{FromWorld, World},
 };
 pub use bevy_ecs_macros::Resource;
@@ -153,6 +153,8 @@ pub unsafe trait SystemParam: Sized {
 
     /// Applies any deferred mutations stored in this [`SystemParam`]'s state.
     /// This is used to apply [`Commands`] during [`apply_system_buffers`](crate::prelude::apply_system_buffers).
+    ///
+    /// [`Commands`]: crate::prelude::Commands
     #[inline]
     #[allow(unused_variables)]
     fn apply(state: &mut Self::State, system_meta: &SystemMeta, world: &mut World) {}
@@ -593,37 +595,6 @@ unsafe impl<'a, T: Resource> SystemParam for Option<ResMut<'a, T>> {
     }
 }
 
-// SAFETY: Commands only accesses internal state
-unsafe impl<'w, 's> ReadOnlySystemParam for Commands<'w, 's> {}
-
-// SAFETY: `Commands::get_param` does not access the world.
-unsafe impl SystemParam for Commands<'_, '_> {
-    type State = CommandQueue;
-    type Item<'w, 's> = Commands<'w, 's>;
-
-    fn init_state(_world: &mut World, _system_meta: &mut SystemMeta) -> Self::State {
-        Default::default()
-    }
-
-    fn apply(state: &mut Self::State, _system_meta: &SystemMeta, world: &mut World) {
-        #[cfg(feature = "trace")]
-        let _system_span =
-            bevy_utils::tracing::info_span!("system_commands", name = _system_meta.name())
-                .entered();
-        state.apply(world);
-    }
-
-    #[inline]
-    unsafe fn get_param<'w, 's>(
-        state: &'s mut Self::State,
-        _system_meta: &SystemMeta,
-        world: &'w World,
-        _change_tick: u32,
-    ) -> Self::Item<'w, 's> {
-        Commands::new(state, world)
-    }
-}
-
 /// SAFETY: only reads world
 unsafe impl<'w> ReadOnlySystemParam for &'w World {}
 
@@ -787,6 +758,181 @@ unsafe impl<'a, T: FromWorld + Send + 'static> SystemParam for Local<'a, T> {
     }
 }
 
+/// Types that can be used with [`Deferred<T>`] in systems.
+/// This allows storing system-local data which is used to defer [`World`] mutations.
+///
+/// Types that implement `SystemBuffer` should take care to perform as many
+/// computations up-front as possible. Buffers cannot be applied in parallel,
+/// so you should try to minimize the time spent in [`SystemBuffer::apply`].
+pub trait SystemBuffer: FromWorld + Send + 'static {
+    /// Applies any deferred mutations to the [`World`].
+    fn apply(&mut self, system_meta: &SystemMeta, world: &mut World);
+}
+
+/// A [`SystemParam`] that stores a buffer which gets applied to the [`World`] at the end of a stage.
+/// This is used internally by [`Commands`] to defer `World` mutations.
+///
+/// [`Commands`]: crate::system::Commands
+///
+/// # Examples
+///
+/// By using this type to defer mutations, you can avoid mutable `World` access within
+/// a system, which allows it to run in parallel with more systems.
+///
+/// Note that deferring mutations is *not* free, and should only be used if
+/// the gains in parallelization outweigh the time it takes to apply deferred mutations.
+/// In general, [`Deferred`] should only be used for mutations that are infrequent,
+/// or which otherwise take up a small portion of a system's run-time.
+///
+/// ```
+/// # use bevy_ecs::prelude::*;
+/// // Tracks whether or not there is a threat the player should be aware of.
+/// #[derive(Resource, Default)]
+/// pub struct Alarm(bool);
+///
+/// #[derive(Component)]
+/// pub struct Settlement {
+///     // ...
+/// }
+///
+/// // A threat from inside the settlement.
+/// #[derive(Component)]
+/// pub struct Criminal;
+///
+/// // A threat from outside the settlement.
+/// #[derive(Component)]
+/// pub struct Monster;
+///
+/// # impl Criminal { pub fn is_threat(&self, _: &Settlement) -> bool { true } }
+///
+/// use bevy_ecs::system::{Deferred, SystemBuffer, SystemMeta};
+///
+/// // Uses deferred mutations to allow signalling the alarm from multiple systems in parallel.
+/// #[derive(Resource, Default)]
+/// struct AlarmFlag(bool);
+///
+/// impl AlarmFlag {
+///     /// Sounds the alarm at the end of the current stage.
+///     pub fn flag(&mut self) {
+///         self.0 = true;
+///     }
+/// }
+///
+/// impl SystemBuffer for AlarmFlag {
+///     // When `AlarmFlag` is used in a system, this function will get
+///     // called at the end of the system's stage.
+///     fn apply(&mut self, system_meta: &SystemMeta, world: &mut World) {
+///         if self.0 {
+///             world.resource_mut::<Alarm>().0 = true;
+///             self.0 = false;
+///         }
+///     }
+/// }
+///
+/// // Sound the alarm if there are any criminals who pose a threat.
+/// fn alert_criminal(
+///     settlements: Query<&Settlement>,
+///     criminals: Query<&Criminal>,
+///     mut alarm: Deferred<AlarmFlag>
+/// ) {
+///     let settlement = settlements.single();
+///     for criminal in &criminals {
+///         // Only sound the alarm if the criminal is a threat.
+///         // For this example, assume that this check is expensive to run.
+///         // Since the majority of this system's run-time is dominated
+///         // by calling `is_threat()`, we defer sounding the alarm to
+///         // allow this system to run in parallel with other alarm systems.
+///         if criminal.is_threat(settlement) {
+///             alarm.flag();
+///         }
+///     }
+/// }
+///
+/// // Sound the alarm if there is a monster.
+/// fn alert_monster(
+///     monsters: Query<&Monster>,
+///     mut alarm: ResMut<Alarm>
+/// ) {
+///     if monsters.iter().next().is_some() {
+///         // Since this system does nothing except for sounding the alarm,
+///         // it would be pointless to defer it, so we sound the alarm directly.
+///         alarm.0 = true;
+///     }
+/// }
+///
+/// let mut world = World::new();
+/// world.init_resource::<Alarm>();
+/// world.spawn(Settlement {
+///     // ...
+/// });
+///
+/// let mut schedule = Schedule::new();
+/// schedule
+///     // These two systems have no conflicts and will run in parallel.
+///     .add_system(alert_criminal)
+///     .add_system(alert_monster);
+///
+/// // There are no criminals or monsters, so the alarm is not sounded.
+/// schedule.run(&mut world);
+/// assert_eq!(world.resource::<Alarm>().0, false);
+///
+/// // Spawn a monster, which will cause the alarm to be sounded.
+/// let m_id = world.spawn(Monster).id();
+/// schedule.run(&mut world);
+/// assert_eq!(world.resource::<Alarm>().0, true);
+///
+/// // Remove the monster and reset the alarm.
+/// world.entity_mut(m_id).despawn();
+/// world.resource_mut::<Alarm>().0 = false;
+///
+/// // Spawn a criminal, which will cause the alarm to be sounded.
+/// world.spawn(Criminal);
+/// schedule.run(&mut world);
+/// assert_eq!(world.resource::<Alarm>().0, true);
+/// ```
+pub struct Deferred<'a, T: SystemBuffer>(pub(crate) &'a mut T);
+
+impl<'a, T: SystemBuffer> Deref for Deferred<'a, T> {
+    type Target = T;
+    #[inline]
+    fn deref(&self) -> &Self::Target {
+        self.0
+    }
+}
+
+impl<'a, T: SystemBuffer> DerefMut for Deferred<'a, T> {
+    #[inline]
+    fn deref_mut(&mut self) -> &mut Self::Target {
+        self.0
+    }
+}
+
+// SAFETY: Only local state is accessed.
+unsafe impl<T: SystemBuffer> ReadOnlySystemParam for Deferred<'_, T> {}
+
+// SAFETY: Only local state is accessed.
+unsafe impl<T: SystemBuffer> SystemParam for Deferred<'_, T> {
+    type State = SyncCell<T>;
+    type Item<'w, 's> = Deferred<'s, T>;
+
+    fn init_state(world: &mut World, _system_meta: &mut SystemMeta) -> Self::State {
+        SyncCell::new(T::from_world(world))
+    }
+
+    fn apply(state: &mut Self::State, system_meta: &SystemMeta, world: &mut World) {
+        state.get().apply(system_meta, world);
+    }
+
+    unsafe fn get_param<'w, 's>(
+        state: &'s mut Self::State,
+        _system_meta: &SystemMeta,
+        _world: &'w World,
+        _change_tick: u32,
+    ) -> Self::Item<'w, 's> {
+        Deferred(state.get())
+    }
+}
+
 /// Shared borrow of a non-[`Send`] resource.
 ///
 /// Only `Send` resources may be accessed with the [`Res`] [`SystemParam`]. In case that the