Optimize Event Updates (#12936)

# Objective
Improve performance scalability when adding new event types to a Bevy
app. Currently, just using Bevy in the default configuration, all apps
spend upwards of 100+us in the `First` schedule, every app tick,
evaluating if it should update events or not, even if events are not
being used for that particular frame, and this scales with the number of
Events registered in the app.

## Solution
As `Events::update` is guaranteed `O(1)` by just checking if a
resource's value, swapping two Vecs, and then clearing one of them, the
actual cost of running `event_update_system` is *very* cheap. The
overhead of doing system dependency injection, task scheduling ,and the
multithreaded executor outweighs the cost of running the system by a
large margin.

Create an `EventRegistry` resource that keeps a number of function
pointers that update each event. Replace the per-event type
`event_update_system` with a singular exclusive system uses the
`EventRegistry` to update all events instead. Update `SubApp::add_event`
to use `EventRegistry` instead.

## Performance
This speeds reduces the cost of the `First` schedule in both many_foxes
and many_cubes by over 80%. Note this is with system spans on. The
majority of this is now context-switching costs from launching
`time_system`, which should be mostly eliminated with #12869.

![image](https://github.com/bevyengine/bevy/assets/3137680/037624be-21a2-4dc2-a42f-9d0bfa3e9b4a)

The actual `event_update_system` is usually *very* short, using only a
few microseconds on average.

![image](https://github.com/bevyengine/bevy/assets/3137680/01ff1689-3595-49b6-8f09-5c44bcf903e8)

---

## Changelog
TODO

## Migration Guide
TODO

---------

Co-authored-by: Josh Matthews <[email protected]>
Co-authored-by: Alice Cecile <[email protected]>

Author: 3 people

crates/bevy_app/src/app.rs

@@ -1,8 +1,10 @@
 use crate::{
-    Main, MainSchedulePlugin, PlaceholderPlugin, Plugin, Plugins, PluginsState, SubApp, SubApps,
+    First, Main, MainSchedulePlugin, PlaceholderPlugin, Plugin, Plugins, PluginsState, SubApp,
+    SubApps,
 };
 pub use bevy_derive::AppLabel;
 use bevy_ecs::{
+    event::event_update_system,
     intern::Interned,
     prelude::*,
     schedule::{ScheduleBuildSettings, ScheduleLabel},
@@ -89,7 +91,12 @@ impl Default for App {
         #[cfg(feature = "bevy_reflect")]
         app.init_resource::<AppTypeRegistry>();
         app.add_plugins(MainSchedulePlugin);
-
+        app.add_systems(
+            First,
+            event_update_system
+                .in_set(bevy_ecs::event::EventUpdates)
+                .run_if(bevy_ecs::event::event_update_condition),
+        );
         app.add_event::<AppExit>();
 
         app
@@ -369,8 +376,6 @@ impl App {
     /// #
     /// app.add_event::<MyEvent>();
     /// ```
-    ///
-    /// [`event_update_system`]: bevy_ecs::event::event_update_system
     pub fn add_event<T>(&mut self) -> &mut Self
     where
         T: Event,

crates/bevy_app/src/sub_app.rs

@@ -1,5 +1,6 @@
-use crate::{App, First, InternedAppLabel, Plugin, Plugins, PluginsState, StateTransition};
+use crate::{App, InternedAppLabel, Plugin, Plugins, PluginsState, StateTransition};
 use bevy_ecs::{
+    event::EventRegistry,
     prelude::*,
     schedule::{
         common_conditions::run_once as run_once_condition, run_enter_schedule,
@@ -362,12 +363,7 @@ impl SubApp {
         T: Event,
     {
         if !self.world.contains_resource::<Events<T>>() {
-            self.init_resource::<Events<T>>().add_systems(
-                First,
-                bevy_ecs::event::event_update_system::<T>
-                    .in_set(bevy_ecs::event::EventUpdates)
-                    .run_if(bevy_ecs::event::event_update_condition::<T>),
-            );
+            EventRegistry::register_event::<T>(self.world_mut());
         }
 
         self

crates/bevy_ecs/examples/events.rs

@@ -1,12 +1,14 @@
 //! In this example a system sends a custom event with a 50/50 chance during any frame.
 //! If an event was send, it will be printed by the console in a receiving system.
 
-use bevy_ecs::prelude::*;
+use bevy_ecs::{event::EventRegistry, prelude::*};
 
 fn main() {
     // Create a new empty world and add the event as a resource
     let mut world = World::new();
-    world.insert_resource(Events::<MyEvent>::default());
+    // The event registry is stored as a resource, and allows us to quickly update all events at once.
+    // This call adds both the registry resource and the events resource into the world.
+    EventRegistry::register_event::<MyEvent>(&mut world);
 
     // Create a schedule to store our systems
     let mut schedule = Schedule::default();
@@ -17,7 +19,7 @@ fn main() {
     #[derive(SystemSet, Debug, Clone, PartialEq, Eq, Hash)]
     pub struct FlushEvents;
 
-    schedule.add_systems(bevy_ecs::event::event_update_system::<MyEvent>.in_set(FlushEvents));
+    schedule.add_systems(bevy_ecs::event::event_update_system.in_set(FlushEvents));
 
     // Add systems sending and receiving events after the events are flushed.
     schedule.add_systems((

crates/bevy_ecs/src/change_detection.rs

@@ -833,6 +833,12 @@ impl<'w> MutUntyped<'w> {
         }
     }
 
+    /// Returns `true` if this value was changed or mutably dereferenced
+    /// either since a specific change tick.
+    pub fn has_changed_since(&self, tick: Tick) -> bool {
+        self.ticks.changed.is_newer_than(tick, self.ticks.this_run)
+    }
+
     /// Returns a pointer to the value without taking ownership of this smart pointer, marking it as changed.
     ///
     /// In order to avoid marking the value as changed, you need to call [`bypass_change_detection`](DetectChangesMut::bypass_change_detection).

crates/bevy_ecs/src/component.rs

@@ -765,7 +765,7 @@ impl Components {
 
 /// A value that tracks when a system ran relative to other systems.
 /// This is used to power change detection.
-#[derive(Copy, Clone, Debug, Eq, PartialEq)]
+#[derive(Copy, Clone, Default, Debug, Eq, PartialEq)]
 #[cfg_attr(feature = "bevy_reflect", derive(Reflect), reflect(Debug, PartialEq))]
 pub struct Tick {
     tick: u32,

crates/bevy_ecs/src/event.rs

@@ -1,7 +1,13 @@
 //! Event handling types.
 
 use crate as bevy_ecs;
-use crate::system::{Local, Res, ResMut, Resource, SystemParam};
+use crate::change_detection::MutUntyped;
+use crate::{
+    change_detection::{DetectChangesMut, Mut},
+    component::{ComponentId, Tick},
+    system::{Local, Res, ResMut, Resource, SystemParam},
+    world::World,
+};
 pub use bevy_ecs_macros::Event;
 use bevy_ecs_macros::SystemSet;
 use bevy_utils::detailed_trace;
@@ -253,7 +259,13 @@ impl<E: Event> Events<E> {
     ///
     /// If you need access to the events that were removed, consider using [`Events::update_drain`].
     pub fn update(&mut self) {
-        let _ = self.update_drain();
+        std::mem::swap(&mut self.events_a, &mut self.events_b);
+        self.events_b.clear();
+        self.events_b.start_event_count = self.event_count;
+        debug_assert_eq!(
+            self.events_a.start_event_count + self.events_a.len(),
+            self.events_b.start_event_count
+        );
     }
 
     /// Swaps the event buffers and drains the oldest event buffer, returning an iterator
@@ -798,48 +810,90 @@ impl<'a, E: Event> ExactSizeIterator for EventIteratorWithId<'a, E> {
     }
 }
 
-#[doc(hidden)]
+struct RegisteredEvent {
+    component_id: ComponentId,
+    // Required to flush the secondary buffer and drop events even if left unchanged.
+    previously_updated: bool,
+    // SAFETY: The component ID and the function must be used to fetch the Events<T> resource
+    // of the same type initialized in `register_event`, or improper type casts will occur.
+    update: unsafe fn(MutUntyped),
+}
+
+/// A registry of all of the [`Events`] in the [`World`], used by [`event_update_system`]
+/// to update all of the events.
 #[derive(Resource, Default)]
-pub struct EventUpdateSignal(bool);
+pub struct EventRegistry {
+    needs_update: bool,
+    event_updates: Vec<RegisteredEvent>,
+}
+
+impl EventRegistry {
+    /// Registers an event type to be updated.
+    pub fn register_event<T: Event>(world: &mut World) {
+        // By initializing the resource here, we can be sure that it is present,
+        // and receive the correct, up-to-date `ComponentId` even if it was previously removed.
+        let component_id = world.init_resource::<Events<T>>();
+        let mut registry = world.get_resource_or_insert_with(Self::default);
+        registry.event_updates.push(RegisteredEvent {
+            component_id,
+            previously_updated: false,
+            update: |ptr| {
+                // SAFETY: The resource was initialized with the type Events<T>.
+                unsafe { ptr.with_type::<Events<T>>() }
+                    .bypass_change_detection()
+                    .update();
+            },
+        });
+    }
+
+    /// Updates all of the registered events in the World.
+    pub fn run_updates(&mut self, world: &mut World, last_change_tick: Tick) {
+        for registered_event in &mut self.event_updates {
+            // Bypass the type ID -> Component ID lookup with the cached component ID.
+            if let Some(events) = world.get_resource_mut_by_id(registered_event.component_id) {
+                let has_changed = events.has_changed_since(last_change_tick);
+                if registered_event.previously_updated || has_changed {
+                    // SAFETY: The update function pointer is called with the resource
+                    // fetched from the same component ID.
+                    unsafe { (registered_event.update)(events) };
+                    // Always set to true if the events have changed, otherwise disable running on the second invocation
+                    // to wait for more changes.
+                    registered_event.previously_updated =
+                        has_changed || !registered_event.previously_updated;
+                }
+            }
+        }
+    }
+}
 
 #[doc(hidden)]
 #[derive(SystemSet, Clone, Debug, PartialEq, Eq, Hash)]
 pub struct EventUpdates;
 
 /// Signals the [`event_update_system`] to run after `FixedUpdate` systems.
-pub fn signal_event_update_system(signal: Option<ResMut<EventUpdateSignal>>) {
-    if let Some(mut s) = signal {
-        s.0 = true;
+pub fn signal_event_update_system(signal: Option<ResMut<EventRegistry>>) {
+    if let Some(mut registry) = signal {
+        registry.needs_update = true;
     }
 }
 
-/// Resets the `EventUpdateSignal`
-pub fn reset_event_update_signal_system(signal: Option<ResMut<EventUpdateSignal>>) {
-    if let Some(mut s) = signal {
-        s.0 = false;
-    }
-}
-
-/// A system that calls [`Events::update`].
-pub fn event_update_system<T: Event>(
-    update_signal: Option<Res<EventUpdateSignal>>,
-    mut events: ResMut<Events<T>>,
-) {
-    if let Some(signal) = update_signal {
-        // If we haven't got a signal to update the events, but we *could* get such a signal
-        // return early and update the events later.
-        if !signal.0 {
-            return;
-        }
+/// A system that calls [`Events::update`] on all registered [`Events`] in the world.
+pub fn event_update_system(world: &mut World, mut last_change_tick: Local<Tick>) {
+    if world.contains_resource::<EventRegistry>() {
+        world.resource_scope(|world, mut registry: Mut<EventRegistry>| {
+            registry.run_updates(world, *last_change_tick);
+            // Disable the system until signal_event_update_system runs again.
+            registry.needs_update = false;
+        });
     }
-
-    events.update();
+    *last_change_tick = world.change_tick();
 }
 
-/// A run condition that checks if the event's [`event_update_system`]
-/// needs to run or not.
-pub fn event_update_condition<T: Event>(events: Res<Events<T>>) -> bool {
-    !events.events_a.is_empty() || !events.events_b.is_empty()
+/// A run condition for [`event_update_system`].
+pub fn event_update_condition(signal: Option<Res<EventRegistry>>) -> bool {
+    // If we haven't got a signal to update the events, but we *could* get such a signal
+    // return early and update the events later.
+    signal.map_or(false, |signal| signal.needs_update)
 }
 
 /// [`Iterator`] over sent [`EventIds`](`EventId`) from a batch.

crates/bevy_ecs/src/schedule/condition.rs

@@ -837,7 +837,7 @@ pub mod common_conditions {
     /// # let mut world = World::new();
     /// # world.init_resource::<Counter>();
     /// # world.init_resource::<Events<MyEvent>>();
-    /// # app.add_systems(bevy_ecs::event::event_update_system::<MyEvent>.before(my_system));
+    /// # app.add_systems(bevy_ecs::event::event_update_system.before(my_system));
     ///
     /// app.add_systems(
     ///     my_system.run_if(on_event::<MyEvent>()),

crates/bevy_time/src/lib.rs

@@ -30,7 +30,7 @@ pub mod prelude {
 }
 
 use bevy_app::{prelude::*, RunFixedMainLoop};
-use bevy_ecs::event::{signal_event_update_system, EventUpdateSignal, EventUpdates};
+use bevy_ecs::event::signal_event_update_system;
 use bevy_ecs::prelude::*;
 use bevy_utils::{tracing::warn, Duration, Instant};
 pub use crossbeam_channel::TrySendError;
@@ -57,19 +57,11 @@ impl Plugin for TimePlugin {
             .register_type::<Time<Virtual>>()
             .register_type::<Time<Fixed>>()
             .register_type::<Timer>()
-            .add_systems(
-                First,
-                (time_system, virtual_time_system.after(time_system)).in_set(TimeSystem),
-            )
+            .add_systems(First, time_system.in_set(TimeSystem))
             .add_systems(RunFixedMainLoop, run_fixed_main_schedule);
 
         // ensure the events are not dropped until `FixedMain` systems can observe them
-        app.init_resource::<EventUpdateSignal>()
-            .add_systems(
-                First,
-                bevy_ecs::event::reset_event_update_signal_system.after(EventUpdates),
-            )
-            .add_systems(FixedPostUpdate, signal_event_update_system);
+        app.add_systems(FixedPostUpdate, signal_event_update_system);
     }
 }
 
@@ -111,7 +103,9 @@ pub fn create_time_channels() -> (TimeSender, TimeReceiver) {
 /// The system used to update the [`Time`] used by app logic. If there is a render world the time is
 /// sent from there to this system through channels. Otherwise the time is updated in this system.
 fn time_system(
-    mut time: ResMut<Time<Real>>,
+    mut real_time: ResMut<Time<Real>>,
+    mut virtual_time: ResMut<Time<Virtual>>,
+    mut time: ResMut<Time>,
     update_strategy: Res<TimeUpdateStrategy>,
     time_recv: Option<Res<TimeReceiver>>,
     mut has_received_time: Local<bool>,
@@ -132,10 +126,12 @@ fn time_system(
     };
 
     match update_strategy.as_ref() {
-        TimeUpdateStrategy::Automatic => time.update_with_instant(new_time),
-        TimeUpdateStrategy::ManualInstant(instant) => time.update_with_instant(*instant),
-        TimeUpdateStrategy::ManualDuration(duration) => time.update_with_duration(*duration),
+        TimeUpdateStrategy::Automatic => real_time.update_with_instant(new_time),
+        TimeUpdateStrategy::ManualInstant(instant) => real_time.update_with_instant(*instant),
+        TimeUpdateStrategy::ManualDuration(duration) => real_time.update_with_duration(*duration),
     }
+
+    update_virtual_time(&mut time, &mut virtual_time, &real_time);
 }
 
 #[cfg(test)]

crates/bevy_time/src/virt.rs

@@ -1,4 +1,3 @@
-use bevy_ecs::system::{Res, ResMut};
 use bevy_reflect::Reflect;
 use bevy_utils::{tracing::debug, Duration};
 
@@ -268,11 +267,7 @@ impl Default for Virtual {
 /// Advances [`Time<Virtual>`] and [`Time`] based on the elapsed [`Time<Real>`].
 ///
 /// The virtual time will be advanced up to the provided [`Time::max_delta`].
-pub fn virtual_time_system(
-    mut current: ResMut<Time>,
-    mut virt: ResMut<Time<Virtual>>,
-    real: Res<Time<Real>>,
-) {
+pub fn update_virtual_time(current: &mut Time, virt: &mut Time<Virtual>, real: &Time<Real>) {
     let raw_delta = real.delta();
     virt.advance_with_raw_delta(raw_delta);
     *current = virt.as_generic();