diff --git a/Cargo.toml b/Cargo.toml
index 2bf46c8a7f777..3280002be3532 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -193,10 +193,18 @@ path = "examples/diagnostics/print_diagnostics.rs"
 name = "event"
 path = "examples/ecs/event.rs"
 
+[[example]]
+name = "fixed_timestep"
+path = "examples/ecs/fixed_timestep.rs"
+
 [[example]]
 name = "startup_system"
 path = "examples/ecs/startup_system.rs"
 
+[[example]]
+name = "state"
+path = "examples/ecs/state.rs"
+
 [[example]]
 name = "system_chaining"
 path = "examples/ecs/system_chaining.rs"
diff --git a/crates/bevy_app/src/app.rs b/crates/bevy_app/src/app.rs
index 61d447fd286f2..adf2091ef2932 100644
--- a/crates/bevy_app/src/app.rs
+++ b/crates/bevy_app/src/app.rs
@@ -1,5 +1,5 @@
 use crate::app_builder::AppBuilder;
-use bevy_ecs::{ParallelExecutor, Resources, Schedule, World};
+use bevy_ecs::{Resources, Schedule, World};
 #[cfg(feature = "trace")]
 use bevy_utils::tracing::info_span;
 
@@ -30,9 +30,6 @@ pub struct App {
     pub resources: Resources,
     pub runner: Box<dyn Fn(App)>,
     pub schedule: Schedule,
-    pub executor: ParallelExecutor,
-    pub startup_schedule: Schedule,
-    pub startup_executor: ParallelExecutor,
 }
 
 impl Default for App {
@@ -41,16 +38,12 @@ impl Default for App {
             world: Default::default(),
             resources: Default::default(),
             schedule: Default::default(),
-            executor: Default::default(),
-            startup_schedule: Default::default(),
-            startup_executor: ParallelExecutor::without_tracker_clears(),
             runner: Box::new(run_once),
         }
     }
 }
 
 fn run_once(mut app: App) {
-    app.initialize();
     app.update();
 }
 
@@ -61,34 +54,15 @@ impl App {
 
     pub fn update(&mut self) {
         self.schedule
-            .initialize(&mut self.world, &mut self.resources);
-        self.executor
-            .run(&mut self.schedule, &mut self.world, &mut self.resources);
-    }
-
-    pub fn initialize(&mut self) {
-        #[cfg(feature = "trace")]
-        let startup_schedule_span = info_span!("startup_schedule");
-        #[cfg(feature = "trace")]
-        let _startup_schedule_guard = startup_schedule_span.enter();
-        self.startup_schedule
-            .initialize(&mut self.world, &mut self.resources);
-        self.startup_executor.initialize(&mut self.resources);
-        self.startup_executor.run(
-            &mut self.startup_schedule,
-            &mut self.world,
-            &mut self.resources,
-        );
+            .initialize_and_run(&mut self.world, &mut self.resources);
     }
 
     pub fn run(mut self) {
         #[cfg(feature = "trace")]
-        let bevy_app_run_span = info_span!("bevy_app_run");
+        let bevy_app_run_span = info_span!("bevy_app");
         #[cfg(feature = "trace")]
         let _bevy_app_run_guard = bevy_app_run_span.enter();
 
-        self.executor.initialize(&mut self.resources);
-
         let runner = std::mem::replace(&mut self.runner, Box::new(run_once));
         (runner)(self);
     }
diff --git a/crates/bevy_app/src/app_builder.rs b/crates/bevy_app/src/app_builder.rs
index 68e0063da74c7..a25198bcabdb1 100644
--- a/crates/bevy_app/src/app_builder.rs
+++ b/crates/bevy_app/src/app_builder.rs
@@ -1,10 +1,15 @@
+use std::any::Any;
+
 use crate::{
     app::{App, AppExit},
     event::Events,
     plugin::Plugin,
     stage, startup_stage, PluginGroup, PluginGroupBuilder,
 };
-use bevy_ecs::{FromResources, IntoSystem, Resources, System, World};
+use bevy_ecs::{
+    clear_trackers_system, FromResources, IntoStage, IntoSystem, Resource, Resources, RunOnce,
+    Schedule, Stage, State, StateStage, System, SystemStage, World,
+};
 use bevy_utils::tracing::debug;
 
 /// Configure [App]s using the builder pattern
@@ -18,8 +23,10 @@ impl Default for AppBuilder {
             app: App::default(),
         };
 
-        app_builder.add_default_stages();
-        app_builder.add_event::<AppExit>();
+        app_builder
+            .add_default_stages()
+            .add_event::<AppExit>()
+            .add_system_to_stage(stage::LAST, clear_trackers_system);
         app_builder
     }
 }
@@ -49,55 +56,88 @@ impl AppBuilder {
         self
     }
 
-    pub fn add_stage(&mut self, stage_name: &'static str) -> &mut Self {
-        self.app.schedule.add_stage(stage_name);
+    pub fn add_stage<Params, S: IntoStage<Params>>(
+        &mut self,
+        name: &'static str,
+        stage: S,
+    ) -> &mut Self {
+        self.app.schedule.add_stage(name, stage);
         self
     }
 
-    pub fn add_stage_after(&mut self, target: &'static str, stage_name: &'static str) -> &mut Self {
-        self.app.schedule.add_stage_after(target, stage_name);
+    pub fn add_stage_after<Params, S: IntoStage<Params>>(
+        &mut self,
+        target: &'static str,
+        name: &'static str,
+        stage: S,
+    ) -> &mut Self {
+        self.app.schedule.add_stage_after(target, name, stage);
         self
     }
 
-    pub fn add_stage_before(
+    pub fn add_stage_before<Params, S: IntoStage<Params>>(
         &mut self,
         target: &'static str,
-        stage_name: &'static str,
+        name: &'static str,
+        stage: S,
     ) -> &mut Self {
-        self.app.schedule.add_stage_before(target, stage_name);
+        self.app.schedule.add_stage_before(target, name, stage);
         self
     }
 
-    pub fn add_startup_stage(&mut self, stage_name: &'static str) -> &mut Self {
-        self.app.startup_schedule.add_stage(stage_name);
+    pub fn add_startup_stage<Params, S: IntoStage<Params>>(
+        &mut self,
+        name: &'static str,
+        stage: S,
+    ) -> &mut Self {
+        self.app
+            .schedule
+            .stage(stage::STARTUP, |schedule: &mut Schedule| {
+                schedule.add_stage(name, stage)
+            });
         self
     }
 
-    pub fn add_startup_stage_after(
+    pub fn add_startup_stage_after<Params, S: IntoStage<Params>>(
         &mut self,
         target: &'static str,
-        stage_name: &'static str,
+        name: &'static str,
+        stage: S,
     ) -> &mut Self {
         self.app
-            .startup_schedule
-            .add_stage_after(target, stage_name);
+            .schedule
+            .stage(stage::STARTUP, |schedule: &mut Schedule| {
+                schedule.add_stage_after(target, name, stage)
+            });
         self
     }
 
-    pub fn add_startup_stage_before(
+    pub fn add_startup_stage_before<Params, S: IntoStage<Params>>(
         &mut self,
         target: &'static str,
-        stage_name: &'static str,
+        name: &'static str,
+        stage: S,
     ) -> &mut Self {
         self.app
-            .startup_schedule
-            .add_stage_before(target, stage_name);
+            .schedule
+            .stage(stage::STARTUP, |schedule: &mut Schedule| {
+                schedule.add_stage_before(target, name, stage)
+            });
+        self
+    }
+
+    pub fn stage<T: Stage, F: FnOnce(&mut T) -> &mut T>(
+        &mut self,
+        name: &str,
+        func: F,
+    ) -> &mut Self {
+        self.app.schedule.stage(name, func);
         self
     }
 
     pub fn add_system<S, Params, IntoS>(&mut self, system: IntoS) -> &mut Self
     where
-        S: System<Input = (), Output = ()>,
+        S: System<In = (), Out = ()>,
         IntoS: IntoSystem<Params, S>,
     {
         self.add_system_to_stage(stage::UPDATE, system)
@@ -109,37 +149,41 @@ impl AppBuilder {
         system: IntoS,
     ) -> &mut Self
     where
-        S: System<Input = (), Output = ()>,
+        S: System<In = (), Out = ()>,
         IntoS: IntoSystem<Params, S>,
     {
         self.app
-            .startup_schedule
-            .add_system_to_stage(stage_name, system);
+            .schedule
+            .stage(stage::STARTUP, |schedule: &mut Schedule| {
+                schedule.add_system_to_stage(stage_name, system)
+            });
         self
     }
 
     pub fn add_startup_system<S, Params, IntoS>(&mut self, system: IntoS) -> &mut Self
     where
-        S: System<Input = (), Output = ()>,
+        S: System<In = (), Out = ()>,
         IntoS: IntoSystem<Params, S>,
     {
-        self.app
-            .startup_schedule
-            .add_system_to_stage(startup_stage::STARTUP, system);
-        self
+        self.add_startup_system_to_stage(startup_stage::STARTUP, system)
     }
 
     pub fn add_default_stages(&mut self) -> &mut Self {
-        self.add_startup_stage(startup_stage::PRE_STARTUP)
-            .add_startup_stage(startup_stage::STARTUP)
-            .add_startup_stage(startup_stage::POST_STARTUP)
-            .add_stage(stage::FIRST)
-            .add_stage(stage::PRE_EVENT)
-            .add_stage(stage::EVENT)
-            .add_stage(stage::PRE_UPDATE)
-            .add_stage(stage::UPDATE)
-            .add_stage(stage::POST_UPDATE)
-            .add_stage(stage::LAST)
+        self.add_stage(
+            stage::STARTUP,
+            Schedule::default()
+                .with_run_criteria(RunOnce::default())
+                .with_stage(startup_stage::PRE_STARTUP, SystemStage::parallel())
+                .with_stage(startup_stage::STARTUP, SystemStage::parallel())
+                .with_stage(startup_stage::POST_STARTUP, SystemStage::parallel()),
+        )
+        .add_stage(stage::FIRST, SystemStage::parallel())
+        .add_stage(stage::PRE_EVENT, SystemStage::parallel())
+        .add_stage(stage::EVENT, SystemStage::parallel())
+        .add_stage(stage::PRE_UPDATE, SystemStage::parallel())
+        .add_stage(stage::UPDATE, SystemStage::parallel())
+        .add_stage(stage::POST_UPDATE, SystemStage::parallel())
+        .add_stage(stage::LAST, SystemStage::parallel())
     }
 
     pub fn add_system_to_stage<S, Params, IntoS>(
@@ -148,28 +192,13 @@ impl AppBuilder {
         system: IntoS,
     ) -> &mut Self
     where
-        S: System<Input = (), Output = ()>,
+        S: System<In = (), Out = ()>,
         IntoS: IntoSystem<Params, S>,
     {
         self.app.schedule.add_system_to_stage(stage_name, system);
         self
     }
 
-    pub fn add_system_to_stage_front<S, Params, IntoS>(
-        &mut self,
-        stage_name: &'static str,
-        system: IntoS,
-    ) -> &mut Self
-    where
-        S: System<Input = (), Output = ()>,
-        IntoS: IntoSystem<Params, S>,
-    {
-        self.app
-            .schedule
-            .add_system_to_stage_front(stage_name, system.system());
-        self
-    }
-
     pub fn add_event<T>(&mut self) -> &mut Self
     where
         T: Send + Sync + 'static,
@@ -178,6 +207,53 @@ impl AppBuilder {
             .add_system_to_stage(stage::EVENT, Events::<T>::update_system)
     }
 
+    pub fn state_stage_name<T: Any>() -> String {
+        format!("state({})", std::any::type_name::<T>())
+    }
+
+    pub fn add_state<T: Clone + Resource>(&mut self, initial: T) -> &mut Self {
+        self.add_resource(State::new(initial));
+        self.app.schedule.add_stage_after(
+            stage::UPDATE,
+            &Self::state_stage_name::<T>(),
+            StateStage::<T>::default(),
+        );
+        self
+    }
+
+    pub fn on_state_enter<T: Clone + Resource, Params, S: IntoStage<Params>>(
+        &mut self,
+        value: T,
+        stage: S,
+    ) -> &mut Self {
+        self.stage(
+            &Self::state_stage_name::<T>(),
+            |state_stage: &mut StateStage<T>| state_stage.on_state_enter(value, stage),
+        )
+    }
+
+    pub fn on_state_update<T: Clone + Resource, Params, S: IntoStage<Params>>(
+        &mut self,
+        value: T,
+        stage: S,
+    ) -> &mut Self {
+        self.stage(
+            &Self::state_stage_name::<T>(),
+            |state_stage: &mut StateStage<T>| state_stage.on_state_update(value, stage),
+        )
+    }
+
+    pub fn on_state_exit<T: Clone + Resource, Params, S: IntoStage<Params>>(
+        &mut self,
+        value: T,
+        stage: S,
+    ) -> &mut Self {
+        self.stage(
+            &Self::state_stage_name::<T>(),
+            |state_stage: &mut StateStage<T>| state_stage.on_state_exit(value, stage),
+        )
+    }
+
     /// Adds a resource to the current [App] and overwrites any resource previously added of the same type.
     pub fn add_resource<T>(&mut self, resource: T) -> &mut Self
     where
diff --git a/crates/bevy_app/src/schedule_runner.rs b/crates/bevy_app/src/schedule_runner.rs
index 9074685d898a4..c533720406001 100644
--- a/crates/bevy_app/src/schedule_runner.rs
+++ b/crates/bevy_app/src/schedule_runner.rs
@@ -56,8 +56,6 @@ impl Plugin for ScheduleRunnerPlugin {
             .get_or_insert_with(ScheduleRunnerSettings::default)
             .to_owned();
         app.set_runner(move |mut app: App| {
-            app.initialize();
-
             let mut app_exit_event_reader = EventReader::<AppExit>::default();
             match settings.run_mode {
                 RunMode::Once => {
diff --git a/crates/bevy_app/src/stage.rs b/crates/bevy_app/src/stage.rs
index 1e4fa590528b7..fb8eb416307ee 100644
--- a/crates/bevy_app/src/stage.rs
+++ b/crates/bevy_app/src/stage.rs
@@ -1,11 +1,14 @@
+/// Name of the app stage that runs once at the beginning of the app
+pub const STARTUP: &str = "startup";
+
 /// Name of app stage that runs before all other app stages
 pub const FIRST: &str = "first";
 
 /// Name of app stage that runs before EVENT
-pub const PRE_EVENT: &str = "pre_events";
+pub const PRE_EVENT: &str = "pre_event";
 
 /// Name of app stage that updates events. Runs before UPDATE
-pub const EVENT: &str = "events";
+pub const EVENT: &str = "event";
 
 /// Name of app stage responsible for performing setup before an update. Runs before UPDATE.
 pub const PRE_UPDATE: &str = "pre_update";
diff --git a/crates/bevy_asset/src/lib.rs b/crates/bevy_asset/src/lib.rs
index a7d8e008a000a..ef2dc16dc872a 100644
--- a/crates/bevy_asset/src/lib.rs
+++ b/crates/bevy_asset/src/lib.rs
@@ -13,6 +13,7 @@ mod path;
 
 pub use asset_server::*;
 pub use assets::*;
+use bevy_ecs::SystemStage;
 use bevy_reflect::RegisterTypeBuilder;
 use bevy_tasks::IoTaskPool;
 pub use handle::*;
@@ -73,14 +74,22 @@ impl Plugin for AssetPlugin {
             AssetServer::new(source, task_pool)
         };
 
-        app.add_stage_before(bevy_app::stage::PRE_UPDATE, stage::LOAD_ASSETS)
-            .add_stage_after(bevy_app::stage::POST_UPDATE, stage::ASSET_EVENTS)
-            .add_resource(asset_server)
-            .register_type::<HandleId>()
-            .add_system_to_stage(
-                bevy_app::stage::PRE_UPDATE,
-                asset_server::free_unused_assets_system,
-            );
+        app.add_stage_before(
+            bevy_app::stage::PRE_UPDATE,
+            stage::LOAD_ASSETS,
+            SystemStage::parallel(),
+        )
+        .add_stage_after(
+            bevy_app::stage::POST_UPDATE,
+            stage::ASSET_EVENTS,
+            SystemStage::parallel(),
+        )
+        .add_resource(asset_server)
+        .register_type::<HandleId>()
+        .add_system_to_stage(
+            bevy_app::stage::PRE_UPDATE,
+            asset_server::free_unused_assets_system,
+        );
 
         #[cfg(all(
             feature = "filesystem_watcher",
diff --git a/crates/bevy_core/src/lib.rs b/crates/bevy_core/src/lib.rs
index 01339438be2f2..198def8ee5703 100644
--- a/crates/bevy_core/src/lib.rs
+++ b/crates/bevy_core/src/lib.rs
@@ -33,6 +33,7 @@ impl Plugin for CorePlugin {
 
         app.init_resource::<Time>()
             .init_resource::<EntityLabels>()
+            .init_resource::<FixedTimesteps>()
             .register_type::<Option<String>>()
             .register_type::<Range<f32>>()
             .register_type::<Timer>()
diff --git a/crates/bevy_core/src/time/fixed_timestep.rs b/crates/bevy_core/src/time/fixed_timestep.rs
new file mode 100644
index 0000000000000..e8ddea5303726
--- /dev/null
+++ b/crates/bevy_core/src/time/fixed_timestep.rs
@@ -0,0 +1,168 @@
+use crate::Time;
+use bevy_ecs::{ArchetypeComponent, ShouldRun, System, SystemId, ThreadLocalExecution, TypeAccess};
+use bevy_utils::HashMap;
+use std::{any::TypeId, borrow::Cow};
+
+pub struct FixedTimestepState {
+    pub step: f64,
+    pub accumulator: f64,
+}
+
+impl FixedTimestepState {
+    /// The amount of time each step takes
+    pub fn step(&self) -> f64 {
+        self.step
+    }
+
+    /// The number of steps made in a second
+    pub fn steps_per_second(&self) -> f64 {
+        1.0 / self.step
+    }
+
+    /// The amount of time (in seconds) left over from the last step
+    pub fn accumulator(&self) -> f64 {
+        self.accumulator
+    }
+
+    /// The percentage of "step" stored inside the accumulator. Calculated as accumulator / step
+    pub fn overstep_percentage(&self) -> f64 {
+        self.accumulator / self.step
+    }
+}
+
+#[derive(Default)]
+pub struct FixedTimesteps {
+    fixed_timesteps: HashMap<String, FixedTimestepState>,
+}
+
+impl FixedTimesteps {
+    pub fn get(&self, name: &str) -> Option<&FixedTimestepState> {
+        self.fixed_timesteps.get(name)
+    }
+}
+
+pub struct FixedTimestep {
+    step: f64,
+    accumulator: f64,
+    looping: bool,
+    system_id: SystemId,
+    label: Option<String>, // TODO: consider making this a TypedLabel
+    resource_access: TypeAccess<TypeId>,
+    archetype_access: TypeAccess<ArchetypeComponent>,
+}
+
+impl Default for FixedTimestep {
+    fn default() -> Self {
+        Self {
+            system_id: SystemId::new(),
+            step: 1.0 / 60.0,
+            accumulator: 0.0,
+            looping: false,
+            label: None,
+            resource_access: Default::default(),
+            archetype_access: Default::default(),
+        }
+    }
+}
+
+impl FixedTimestep {
+    pub fn step(step: f64) -> Self {
+        Self {
+            step,
+            ..Default::default()
+        }
+    }
+
+    pub fn steps_per_second(rate: f64) -> Self {
+        Self {
+            step: 1.0 / rate,
+            ..Default::default()
+        }
+    }
+
+    pub fn with_label(mut self, label: &str) -> Self {
+        self.label = Some(label.to_string());
+        self
+    }
+
+    pub fn update(&mut self, time: &Time) -> ShouldRun {
+        if !self.looping {
+            self.accumulator += time.delta_seconds_f64();
+        }
+
+        if self.accumulator >= self.step {
+            self.accumulator -= self.step;
+            self.looping = true;
+            ShouldRun::YesAndLoop
+        } else {
+            self.looping = false;
+            ShouldRun::No
+        }
+    }
+}
+
+impl System for FixedTimestep {
+    type In = ();
+    type Out = ShouldRun;
+
+    fn name(&self) -> Cow<'static, str> {
+        Cow::Borrowed(std::any::type_name::<FixedTimestep>())
+    }
+
+    fn id(&self) -> SystemId {
+        self.system_id
+    }
+
+    fn update(&mut self, _world: &bevy_ecs::World) {}
+
+    fn archetype_component_access(&self) -> &TypeAccess<ArchetypeComponent> {
+        &self.archetype_access
+    }
+
+    fn resource_access(&self) -> &TypeAccess<TypeId> {
+        &self.resource_access
+    }
+
+    fn thread_local_execution(&self) -> ThreadLocalExecution {
+        ThreadLocalExecution::Immediate
+    }
+
+    unsafe fn run_unsafe(
+        &mut self,
+        _input: Self::In,
+        _world: &bevy_ecs::World,
+        resources: &bevy_ecs::Resources,
+    ) -> Option<Self::Out> {
+        let time = resources.get::<Time>().unwrap();
+        let result = self.update(&time);
+        if let Some(ref label) = self.label {
+            let mut fixed_timesteps = resources.get_mut::<FixedTimesteps>().unwrap();
+            let state = fixed_timesteps.fixed_timesteps.get_mut(label).unwrap();
+            state.step = self.step;
+            state.accumulator = self.accumulator;
+        }
+
+        Some(result)
+    }
+
+    fn run_thread_local(
+        &mut self,
+        _world: &mut bevy_ecs::World,
+        _resources: &mut bevy_ecs::Resources,
+    ) {
+    }
+
+    fn initialize(&mut self, _world: &mut bevy_ecs::World, resources: &mut bevy_ecs::Resources) {
+        self.resource_access.add_read(TypeId::of::<Time>());
+        if let Some(ref label) = self.label {
+            let mut fixed_timesteps = resources.get_mut::<FixedTimesteps>().unwrap();
+            fixed_timesteps.fixed_timesteps.insert(
+                label.clone(),
+                FixedTimestepState {
+                    accumulator: 0.0,
+                    step: self.step,
+                },
+            );
+        }
+    }
+}
diff --git a/crates/bevy_core/src/time/mod.rs b/crates/bevy_core/src/time/mod.rs
index ce3863df4062f..8be564301c013 100644
--- a/crates/bevy_core/src/time/mod.rs
+++ b/crates/bevy_core/src/time/mod.rs
@@ -1,6 +1,8 @@
+mod fixed_timestep;
 #[allow(clippy::module_inception)]
 mod time;
 mod timer;
 
+pub use fixed_timestep::*;
 pub use time::*;
 pub use timer::*;
diff --git a/crates/bevy_ecs/src/lib.rs b/crates/bevy_ecs/src/lib.rs
index a8ead603feed2..cdf8dfa5d63c3 100644
--- a/crates/bevy_ecs/src/lib.rs
+++ b/crates/bevy_ecs/src/lib.rs
@@ -14,6 +14,7 @@ pub mod prelude {
     pub use crate::{
         core::WorldBuilderSource,
         resource::{ChangedRes, FromResources, Local, Res, ResMut, Resource, Resources},
+        schedule::{Schedule, State, SystemStage},
         system::{Commands, IntoSystem, Query, System},
         Added, Bundle, Changed, Component, Entity, In, IntoChainSystem, Mut, Mutated, Or, QuerySet,
         Ref, RefMut, With, Without, World,
diff --git a/crates/bevy_ecs/src/schedule/mod.rs b/crates/bevy_ecs/src/schedule/mod.rs
index be45138cb8e8e..6d6239a74eb03 100644
--- a/crates/bevy_ecs/src/schedule/mod.rs
+++ b/crates/bevy_ecs/src/schedule/mod.rs
@@ -1,6 +1,557 @@
-mod parallel_executor;
-#[allow(clippy::module_inception)]
-mod schedule;
+mod stage;
+mod stage_executor;
+mod state;
 
-pub use parallel_executor::*;
-pub use schedule::*;
+pub use stage::*;
+pub use stage_executor::*;
+pub use state::*;
+
+use crate::{IntoSystem, Resources, System, World};
+use bevy_utils::HashMap;
+
+#[derive(Default)]
+pub struct Schedule {
+    stages: HashMap<String, Box<dyn Stage>>,
+    stage_order: Vec<String>,
+    run_criteria: Option<Box<dyn System<In = (), Out = ShouldRun>>>,
+    run_criteria_initialized: bool,
+}
+
+impl Schedule {
+    pub fn with_stage<Params, S: IntoStage<Params>>(mut self, name: &str, stage: S) -> Self {
+        self.add_stage(name, stage.into_stage());
+        self
+    }
+
+    pub fn with_stage_after<Params, S: IntoStage<Params>>(
+        mut self,
+        target: &str,
+        name: &str,
+        stage: S,
+    ) -> Self {
+        self.add_stage_after(target, name, stage);
+        self
+    }
+
+    pub fn with_stage_before<Params, S: IntoStage<Params>>(
+        mut self,
+        target: &str,
+        name: &str,
+        stage: S,
+    ) -> Self {
+        self.add_stage_before(target, name, stage);
+        self
+    }
+
+    pub fn with_run_criteria<S, Params, IntoS>(mut self, system: IntoS) -> Self
+    where
+        S: System<In = (), Out = ShouldRun>,
+        IntoS: IntoSystem<Params, S>,
+    {
+        self.set_run_criteria(system);
+        self
+    }
+
+    pub fn with_system_in_stage<S, Params, IntoS>(
+        mut self,
+        stage_name: &'static str,
+        system: IntoS,
+    ) -> Self
+    where
+        S: System<In = (), Out = ()>,
+        IntoS: IntoSystem<Params, S>,
+    {
+        self.add_system_to_stage(stage_name, system);
+        self
+    }
+
+    pub fn set_run_criteria<S, Params, IntoS>(&mut self, system: IntoS) -> &mut Self
+    where
+        S: System<In = (), Out = ShouldRun>,
+        IntoS: IntoSystem<Params, S>,
+    {
+        self.run_criteria = Some(Box::new(system.system()));
+        self.run_criteria_initialized = false;
+        self
+    }
+
+    pub fn add_stage<Params, S: IntoStage<Params>>(&mut self, name: &str, stage: S) -> &mut Self {
+        self.stage_order.push(name.to_string());
+        self.stages
+            .insert(name.to_string(), Box::new(stage.into_stage()));
+        self
+    }
+
+    pub fn add_stage_after<Params, S: IntoStage<Params>>(
+        &mut self,
+        target: &str,
+        name: &str,
+        stage: S,
+    ) -> &mut Self {
+        if self.stages.get(name).is_some() {
+            panic!("Stage already exists: {}.", name);
+        }
+
+        let target_index = self
+            .stage_order
+            .iter()
+            .enumerate()
+            .find(|(_i, stage_name)| *stage_name == target)
+            .map(|(i, _)| i)
+            .unwrap_or_else(|| panic!("Target stage does not exist: {}.", target));
+
+        self.stages
+            .insert(name.to_string(), Box::new(stage.into_stage()));
+        self.stage_order.insert(target_index + 1, name.to_string());
+        self
+    }
+
+    pub fn add_stage_before<Params, S: IntoStage<Params>>(
+        &mut self,
+        target: &str,
+        name: &str,
+        stage: S,
+    ) -> &mut Self {
+        if self.stages.get(name).is_some() {
+            panic!("Stage already exists: {}.", name);
+        }
+
+        let target_index = self
+            .stage_order
+            .iter()
+            .enumerate()
+            .find(|(_i, stage_name)| *stage_name == target)
+            .map(|(i, _)| i)
+            .unwrap_or_else(|| panic!("Target stage does not exist: {}.", target));
+
+        self.stages
+            .insert(name.to_string(), Box::new(stage.into_stage()));
+        self.stage_order.insert(target_index, name.to_string());
+        self
+    }
+
+    pub fn add_system_to_stage<S, Params, IntoS>(
+        &mut self,
+        stage_name: &'static str,
+        system: IntoS,
+    ) -> &mut Self
+    where
+        S: System<In = (), Out = ()>,
+        IntoS: IntoSystem<Params, S>,
+    {
+        let stage = self
+            .get_stage_mut::<SystemStage>(stage_name)
+            .unwrap_or_else(|| {
+                panic!(
+                    "Stage '{}' does not exist or is not a SystemStage",
+                    stage_name
+                )
+            });
+        stage.add_system(system);
+        self
+    }
+
+    pub fn stage<T: Stage, F: FnOnce(&mut T) -> &mut T>(
+        &mut self,
+        name: &str,
+        func: F,
+    ) -> &mut Self {
+        let stage = self
+            .get_stage_mut::<T>(name)
+            .expect("stage does not exist or is the wrong type");
+        func(stage);
+        self
+    }
+
+    pub fn get_stage<T: Stage>(&self, name: &str) -> Option<&T> {
+        self.stages
+            .get(name)
+            .and_then(|stage| stage.downcast_ref::<T>())
+    }
+
+    pub fn get_stage_mut<T: Stage>(&mut self, name: &str) -> Option<&mut T> {
+        self.stages
+            .get_mut(name)
+            .and_then(|stage| stage.downcast_mut::<T>())
+    }
+
+    pub fn run_once(&mut self, world: &mut World, resources: &mut Resources) {
+        for name in self.stage_order.iter() {
+            #[cfg(feature = "trace")]
+            let stage_span = bevy_utils::tracing::info_span!("stage", name = name.as_str());
+            #[cfg(feature = "trace")]
+            let _stage_guard = stage_span.enter();
+            let stage = self.stages.get_mut(name).unwrap();
+            stage.run(world, resources);
+        }
+    }
+
+    /// Shorthand for [Schedule::initialize] and [Schedule::run]
+    pub fn initialize_and_run(&mut self, world: &mut World, resources: &mut Resources) {
+        self.initialize(world, resources);
+        self.run(world, resources);
+    }
+}
+
+impl Stage for Schedule {
+    fn initialize(&mut self, world: &mut World, resources: &mut Resources) {
+        if let Some(ref mut run_criteria) = self.run_criteria {
+            if !self.run_criteria_initialized {
+                run_criteria.initialize(world, resources);
+                self.run_criteria_initialized = true;
+            }
+        }
+
+        for name in self.stage_order.iter() {
+            let stage = self.stages.get_mut(name).unwrap();
+            stage.initialize(world, resources);
+        }
+    }
+
+    fn run(&mut self, world: &mut World, resources: &mut Resources) {
+        loop {
+            let should_run = if let Some(ref mut run_criteria) = self.run_criteria {
+                let should_run = run_criteria.run((), world, resources);
+                run_criteria.run_thread_local(world, resources);
+                // don't run when no result is returned or false is returned
+                should_run.unwrap_or(ShouldRun::No)
+            } else {
+                ShouldRun::Yes
+            };
+
+            match should_run {
+                ShouldRun::No => return,
+                ShouldRun::Yes => {
+                    self.run_once(world, resources);
+                    return;
+                }
+                ShouldRun::YesAndLoop => {
+                    self.run_once(world, resources);
+                }
+            }
+        }
+    }
+}
+
+pub fn clear_trackers_system(world: &mut World, resources: &mut Resources) {
+    world.clear_trackers();
+    resources.clear_trackers();
+}
+
+#[cfg(test)]
+mod tests {
+    use crate::{
+        resource::{Res, ResMut, Resources},
+        schedule::{ParallelSystemStageExecutor, Schedule, SystemStage},
+        system::Query,
+        Commands, Entity, World,
+    };
+    use bevy_tasks::{ComputeTaskPool, TaskPool};
+    use fixedbitset::FixedBitSet;
+    use parking_lot::Mutex;
+    use std::{collections::HashSet, sync::Arc};
+
+    #[derive(Default)]
+    struct CompletedSystems {
+        completed_systems: Arc<Mutex<HashSet<&'static str>>>,
+    }
+
+    #[test]
+    fn cross_stage_archetype_change_prepare() {
+        let mut world = World::new();
+        let mut resources = Resources::default();
+        resources.insert(ComputeTaskPool(TaskPool::default()));
+
+        fn insert(commands: &mut Commands) {
+            commands.spawn((1u32,));
+        }
+
+        fn read(query: Query<&u32>, entities: Query<Entity>) {
+            for entity in &mut entities.iter() {
+                // query.get() does a "system permission check" that will fail if the entity is from a
+                // new archetype which hasnt been "prepared yet"
+                query.get_component::<u32>(entity).unwrap();
+            }
+
+            assert_eq!(1, entities.iter().count());
+        }
+
+        let mut schedule = Schedule::default();
+        let mut pre_archetype_change = SystemStage::parallel();
+        pre_archetype_change.add_system(insert);
+        schedule.add_stage("PreArchetypeChange", pre_archetype_change);
+        let mut post_archetype_change = SystemStage::parallel();
+        post_archetype_change.add_system(read);
+        schedule.add_stage("PostArchetypeChange", post_archetype_change);
+
+        schedule.initialize_and_run(&mut world, &mut resources);
+    }
+
+    #[test]
+    fn intra_stage_archetype_change_prepare() {
+        let mut world = World::new();
+        let mut resources = Resources::default();
+        resources.insert(ComputeTaskPool(TaskPool::default()));
+
+        fn insert(world: &mut World, _resources: &mut Resources) {
+            world.spawn((1u32,));
+        }
+
+        fn read(query: Query<&u32>, entities: Query<Entity>) {
+            for entity in &mut entities.iter() {
+                // query.get() does a "system permission check" that will fail if the entity is from a
+                // new archetype which hasnt been "prepared yet"
+                query.get_component::<u32>(entity).unwrap();
+            }
+
+            assert_eq!(1, entities.iter().count());
+        }
+
+        let mut update = SystemStage::parallel();
+        update.add_system(insert);
+        update.add_system(read);
+
+        let mut schedule = Schedule::default();
+        schedule.add_stage("update", update);
+
+        schedule.initialize_and_run(&mut world, &mut resources);
+    }
+
+    #[test]
+    fn schedule() {
+        let mut world = World::new();
+        let mut resources = Resources::default();
+        resources.insert(ComputeTaskPool(TaskPool::default()));
+        resources.insert(CompletedSystems::default());
+        resources.insert(1.0f64);
+        resources.insert(2isize);
+
+        world.spawn((1.0f32,));
+        world.spawn((1u32, 1u64));
+        world.spawn((2u32,));
+
+        let mut stage_a = SystemStage::parallel(); // component queries
+        let mut stage_b = SystemStage::parallel(); // thread local
+        let mut stage_c = SystemStage::parallel(); // resources
+
+        // A system names
+        const READ_U32_SYSTEM_NAME: &str = "read_u32";
+        const WRITE_FLOAT_SYSTEM_NAME: &str = "write_float";
+        const READ_U32_WRITE_U64_SYSTEM_NAME: &str = "read_u32_write_u64";
+        const READ_U64_SYSTEM_NAME: &str = "read_u64";
+
+        // B system names
+        const WRITE_U64_SYSTEM_NAME: &str = "write_u64";
+        const THREAD_LOCAL_SYSTEM_SYSTEM_NAME: &str = "thread_local_system";
+        const WRITE_F32_SYSTEM_NAME: &str = "write_f32";
+
+        // C system names
+        const READ_F64_RES_SYSTEM_NAME: &str = "read_f64_res";
+        const READ_ISIZE_RES_SYSTEM_NAME: &str = "read_isize_res";
+        const READ_ISIZE_WRITE_F64_RES_SYSTEM_NAME: &str = "read_isize_write_f64_res";
+        const WRITE_F64_RES_SYSTEM_NAME: &str = "write_f64_res";
+
+        // A systems
+
+        fn read_u32(completed_systems: Res<CompletedSystems>, _query: Query<&u32>) {
+            let mut completed_systems = completed_systems.completed_systems.lock();
+            completed_systems.insert(READ_U32_SYSTEM_NAME);
+        }
+
+        fn write_float(completed_systems: Res<CompletedSystems>, _query: Query<&f32>) {
+            let mut completed_systems = completed_systems.completed_systems.lock();
+            completed_systems.insert(WRITE_FLOAT_SYSTEM_NAME);
+        }
+
+        fn read_u32_write_u64(
+            completed_systems: Res<CompletedSystems>,
+            _query: Query<(&u32, &mut u64)>,
+        ) {
+            let mut completed_systems = completed_systems.completed_systems.lock();
+            assert!(!completed_systems.contains(READ_U64_SYSTEM_NAME));
+            completed_systems.insert(READ_U32_WRITE_U64_SYSTEM_NAME);
+        }
+
+        fn read_u64(completed_systems: Res<CompletedSystems>, _query: Query<&u64>) {
+            let mut completed_systems = completed_systems.completed_systems.lock();
+            assert!(completed_systems.contains(READ_U32_WRITE_U64_SYSTEM_NAME));
+            assert!(!completed_systems.contains(WRITE_U64_SYSTEM_NAME));
+            completed_systems.insert(READ_U64_SYSTEM_NAME);
+        }
+
+        stage_a.add_system(read_u32);
+        stage_a.add_system(write_float);
+        stage_a.add_system(read_u32_write_u64);
+        stage_a.add_system(read_u64);
+
+        // B systems
+
+        fn write_u64(completed_systems: Res<CompletedSystems>, _query: Query<&mut u64>) {
+            let mut completed_systems = completed_systems.completed_systems.lock();
+            assert!(completed_systems.contains(READ_U64_SYSTEM_NAME));
+            assert!(!completed_systems.contains(THREAD_LOCAL_SYSTEM_SYSTEM_NAME));
+            assert!(!completed_systems.contains(WRITE_F32_SYSTEM_NAME));
+            completed_systems.insert(WRITE_U64_SYSTEM_NAME);
+        }
+
+        fn thread_local_system(_world: &mut World, resources: &mut Resources) {
+            let completed_systems = resources.get::<CompletedSystems>().unwrap();
+            let mut completed_systems = completed_systems.completed_systems.lock();
+            assert!(completed_systems.contains(WRITE_U64_SYSTEM_NAME));
+            assert!(!completed_systems.contains(WRITE_F32_SYSTEM_NAME));
+            completed_systems.insert(THREAD_LOCAL_SYSTEM_SYSTEM_NAME);
+        }
+
+        fn write_f32(completed_systems: Res<CompletedSystems>, _query: Query<&mut f32>) {
+            let mut completed_systems = completed_systems.completed_systems.lock();
+            assert!(completed_systems.contains(WRITE_U64_SYSTEM_NAME));
+            assert!(completed_systems.contains(THREAD_LOCAL_SYSTEM_SYSTEM_NAME));
+            assert!(!completed_systems.contains(READ_F64_RES_SYSTEM_NAME));
+            completed_systems.insert(WRITE_F32_SYSTEM_NAME);
+        }
+
+        stage_b.add_system(write_u64);
+        stage_b.add_system(thread_local_system);
+        stage_b.add_system(write_f32);
+
+        // C systems
+
+        fn read_f64_res(completed_systems: Res<CompletedSystems>, _f64_res: Res<f64>) {
+            let mut completed_systems = completed_systems.completed_systems.lock();
+            assert!(completed_systems.contains(WRITE_F32_SYSTEM_NAME));
+            assert!(!completed_systems.contains(READ_ISIZE_WRITE_F64_RES_SYSTEM_NAME));
+            assert!(!completed_systems.contains(WRITE_F64_RES_SYSTEM_NAME));
+            completed_systems.insert(READ_F64_RES_SYSTEM_NAME);
+        }
+
+        fn read_isize_res(completed_systems: Res<CompletedSystems>, _isize_res: Res<isize>) {
+            let mut completed_systems = completed_systems.completed_systems.lock();
+            completed_systems.insert(READ_ISIZE_RES_SYSTEM_NAME);
+        }
+
+        fn read_isize_write_f64_res(
+            completed_systems: Res<CompletedSystems>,
+            _isize_res: Res<isize>,
+            _f64_res: ResMut<f64>,
+        ) {
+            let mut completed_systems = completed_systems.completed_systems.lock();
+            assert!(completed_systems.contains(READ_F64_RES_SYSTEM_NAME));
+            assert!(!completed_systems.contains(WRITE_F64_RES_SYSTEM_NAME));
+            completed_systems.insert(READ_ISIZE_WRITE_F64_RES_SYSTEM_NAME);
+        }
+
+        fn write_f64_res(completed_systems: Res<CompletedSystems>, _f64_res: ResMut<f64>) {
+            let mut completed_systems = completed_systems.completed_systems.lock();
+            assert!(completed_systems.contains(READ_F64_RES_SYSTEM_NAME));
+            assert!(completed_systems.contains(READ_ISIZE_WRITE_F64_RES_SYSTEM_NAME));
+            completed_systems.insert(WRITE_F64_RES_SYSTEM_NAME);
+        }
+
+        stage_c.add_system(read_f64_res);
+        stage_c.add_system(read_isize_res);
+        stage_c.add_system(read_isize_write_f64_res);
+        stage_c.add_system(write_f64_res);
+
+        fn run_and_validate(schedule: &mut Schedule, world: &mut World, resources: &mut Resources) {
+            schedule.initialize_and_run(world, resources);
+
+            let stage_a = schedule.get_stage::<SystemStage>("a").unwrap();
+            let stage_b = schedule.get_stage::<SystemStage>("b").unwrap();
+            let stage_c = schedule.get_stage::<SystemStage>("c").unwrap();
+
+            let a_executor = stage_a
+                .get_executor::<ParallelSystemStageExecutor>()
+                .unwrap();
+            let b_executor = stage_b
+                .get_executor::<ParallelSystemStageExecutor>()
+                .unwrap();
+            let c_executor = stage_c
+                .get_executor::<ParallelSystemStageExecutor>()
+                .unwrap();
+
+            assert_eq!(
+                a_executor.system_dependents(),
+                vec![vec![], vec![], vec![3], vec![]]
+            );
+            assert_eq!(
+                b_executor.system_dependents(),
+                vec![vec![1], vec![2], vec![]]
+            );
+            assert_eq!(
+                c_executor.system_dependents(),
+                vec![vec![2, 3], vec![], vec![3], vec![]]
+            );
+
+            let stage_a_len = a_executor.system_dependencies().len();
+            let mut read_u64_deps = FixedBitSet::with_capacity(stage_a_len);
+            read_u64_deps.insert(2);
+
+            assert_eq!(
+                a_executor.system_dependencies(),
+                vec![
+                    FixedBitSet::with_capacity(stage_a_len),
+                    FixedBitSet::with_capacity(stage_a_len),
+                    FixedBitSet::with_capacity(stage_a_len),
+                    read_u64_deps,
+                ]
+            );
+
+            let stage_b_len = b_executor.system_dependencies().len();
+            let mut thread_local_deps = FixedBitSet::with_capacity(stage_b_len);
+            thread_local_deps.insert(0);
+            let mut write_f64_deps = FixedBitSet::with_capacity(stage_b_len);
+            write_f64_deps.insert(1);
+            assert_eq!(
+                b_executor.system_dependencies(),
+                vec![
+                    FixedBitSet::with_capacity(stage_b_len),
+                    thread_local_deps,
+                    write_f64_deps
+                ]
+            );
+
+            let stage_c_len = c_executor.system_dependencies().len();
+            let mut read_isize_write_f64_res_deps = FixedBitSet::with_capacity(stage_c_len);
+            read_isize_write_f64_res_deps.insert(0);
+            let mut write_f64_res_deps = FixedBitSet::with_capacity(stage_c_len);
+            write_f64_res_deps.insert(0);
+            write_f64_res_deps.insert(2);
+            assert_eq!(
+                c_executor.system_dependencies(),
+                vec![
+                    FixedBitSet::with_capacity(stage_c_len),
+                    FixedBitSet::with_capacity(stage_c_len),
+                    read_isize_write_f64_res_deps,
+                    write_f64_res_deps
+                ]
+            );
+
+            let completed_systems = resources.get::<CompletedSystems>().unwrap();
+            assert_eq!(
+                completed_systems.completed_systems.lock().len(),
+                11,
+                "completed_systems should have been incremented once for each system"
+            );
+        }
+
+        let mut schedule = Schedule::default();
+        schedule.add_stage("a", stage_a);
+        schedule.add_stage("b", stage_b);
+        schedule.add_stage("c", stage_c);
+
+        // Test the "clean start" case
+        run_and_validate(&mut schedule, &mut world, &mut resources);
+
+        // Stress test the "continue running" case
+        for _ in 0..1000 {
+            // run again (with completed_systems reset) to ensure executor works correctly across runs
+            resources
+                .get::<CompletedSystems>()
+                .unwrap()
+                .completed_systems
+                .lock()
+                .clear();
+            run_and_validate(&mut schedule, &mut world, &mut resources);
+        }
+    }
+}
diff --git a/crates/bevy_ecs/src/schedule/schedule.rs b/crates/bevy_ecs/src/schedule/schedule.rs
deleted file mode 100644
index a33ead6a3986d..0000000000000
--- a/crates/bevy_ecs/src/schedule/schedule.rs
+++ /dev/null
@@ -1,242 +0,0 @@
-use crate::{
-    resource::Resources,
-    system::{System, SystemId, ThreadLocalExecution},
-    IntoSystem, World,
-};
-use bevy_utils::{HashMap, HashSet};
-use std::{borrow::Cow, fmt};
-
-/// An ordered collection of stages, which each contain an ordered list of [System]s.
-/// Schedules are essentially the "execution plan" for an App's systems.
-/// They are run on a given [World] and [Resources] reference.
-#[derive(Default)]
-pub struct Schedule {
-    pub(crate) stages: HashMap<Cow<'static, str>, Vec<Box<dyn System<Input = (), Output = ()>>>>,
-    pub(crate) stage_order: Vec<Cow<'static, str>>,
-    pub(crate) system_ids: HashSet<SystemId>,
-    generation: usize,
-    last_initialize_generation: usize,
-}
-
-impl fmt::Debug for Schedule {
-    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
-        writeln!(f, "Schedule {{")?;
-
-        let stages = self
-            .stage_order
-            .iter()
-            .map(|s| (s, self.stages[s].iter().map(|s| (s.name(), s.id()))));
-
-        for (stage, syss) in stages {
-            writeln!(f, "    Stage \"{}\"", stage)?;
-
-            for (name, id) in syss {
-                writeln!(f, "        System {{ name: \"{}\", id: {:?} }}", name, id)?;
-            }
-        }
-
-        writeln!(f, "}}")
-    }
-}
-
-impl Schedule {
-    pub fn add_stage(&mut self, stage: impl Into<Cow<'static, str>>) {
-        let stage: Cow<str> = stage.into();
-        if self.stages.get(&stage).is_some() {
-            panic!("Stage already exists: {}.", stage);
-        } else {
-            self.stages.insert(stage.clone(), Vec::new());
-            self.stage_order.push(stage);
-        }
-    }
-
-    pub fn add_stage_after(
-        &mut self,
-        target: impl Into<Cow<'static, str>>,
-        stage: impl Into<Cow<'static, str>>,
-    ) {
-        let target: Cow<str> = target.into();
-        let stage: Cow<str> = stage.into();
-        if self.stages.get(&stage).is_some() {
-            panic!("Stage already exists: {}.", stage);
-        }
-
-        let target_index = self
-            .stage_order
-            .iter()
-            .enumerate()
-            .find(|(_i, stage)| **stage == target)
-            .map(|(i, _)| i)
-            .unwrap_or_else(|| panic!("Target stage does not exist: {}.", target));
-
-        self.stages.insert(stage.clone(), Vec::new());
-        self.stage_order.insert(target_index + 1, stage);
-    }
-
-    pub fn add_stage_before(
-        &mut self,
-        target: impl Into<Cow<'static, str>>,
-        stage: impl Into<Cow<'static, str>>,
-    ) {
-        let target: Cow<str> = target.into();
-        let stage: Cow<str> = stage.into();
-        if self.stages.get(&stage).is_some() {
-            panic!("Stage already exists: {}.", stage);
-        }
-
-        let target_index = self
-            .stage_order
-            .iter()
-            .enumerate()
-            .find(|(_i, stage)| **stage == target)
-            .map(|(i, _)| i)
-            .unwrap_or_else(|| panic!("Target stage does not exist: {}.", target));
-
-        self.stages.insert(stage.clone(), Vec::new());
-        self.stage_order.insert(target_index, stage);
-    }
-
-    pub fn add_system_to_stage<S, Params, IntoS>(
-        &mut self,
-        stage_name: impl Into<Cow<'static, str>>,
-        system: IntoS,
-    ) -> &mut Self
-    where
-        S: System<Input = (), Output = ()>,
-        IntoS: IntoSystem<Params, S>,
-    {
-        self.add_system_to_stage_internal(stage_name.into(), Box::new(system.system()));
-        self
-    }
-
-    pub fn add_boxed_system_to_stage(
-        &mut self,
-        stage_name: impl Into<Cow<'static, str>>,
-        system: Box<dyn System<Input = (), Output = ()>>,
-    ) {
-        self.add_system_to_stage_internal(stage_name.into(), system);
-    }
-
-    fn add_system_to_stage_internal(
-        &mut self,
-        stage_name: Cow<'static, str>,
-        system: Box<dyn System<Input = (), Output = ()>>,
-    ) {
-        let systems = self
-            .stages
-            .get_mut(&stage_name)
-            .unwrap_or_else(|| panic!("Stage does not exist: {}.", stage_name));
-        if self.system_ids.contains(&system.id()) {
-            panic!(
-                "System with id {:?} ({}) already exists.",
-                system.id(),
-                system.name()
-            );
-        }
-        self.system_ids.insert(system.id());
-        systems.push(system);
-
-        self.generation += 1;
-    }
-
-    pub fn add_system_to_stage_front<S, Params, IntoS>(
-        &mut self,
-        stage_name: impl Into<Cow<'static, str>>,
-        system: IntoS,
-    ) -> &mut Self
-    where
-        S: System<Input = (), Output = ()>,
-        IntoS: IntoSystem<Params, S>,
-    {
-        self.add_system_to_stage_front_internal(stage_name.into(), Box::new(system.system()));
-        self
-    }
-
-    fn add_system_to_stage_front_internal(
-        &mut self,
-        stage_name: Cow<'static, str>,
-        system: Box<dyn System<Input = (), Output = ()>>,
-    ) {
-        let systems = self
-            .stages
-            .get_mut(&stage_name)
-            .unwrap_or_else(|| panic!("Stage does not exist: {}", stage_name));
-        if self.system_ids.contains(&system.id()) {
-            panic!(
-                "System with id {:?} ({}) already exists.",
-                system.id(),
-                system.name()
-            );
-        }
-        self.system_ids.insert(system.id());
-        systems.insert(0, system);
-
-        self.generation += 1;
-    }
-
-    pub fn run(&mut self, world: &mut World, resources: &mut Resources) {
-        for stage_name in self.stage_order.iter() {
-            if let Some(stage_systems) = self.stages.get_mut(stage_name) {
-                for system in stage_systems.iter_mut() {
-                    system.update(world);
-                    match system.thread_local_execution() {
-                        ThreadLocalExecution::NextFlush => {
-                            system.run((), world, resources);
-                        }
-                        ThreadLocalExecution::Immediate => {
-                            system.run((), world, resources);
-                            // NOTE: when this is made parallel a full sync is required here
-                            system.run_thread_local(world, resources);
-                        }
-                    }
-                }
-
-                // "flush"
-                // NOTE: when this is made parallel a full sync is required here
-                for system in stage_systems.iter_mut() {
-                    match system.thread_local_execution() {
-                        ThreadLocalExecution::NextFlush => {
-                            system.run_thread_local(world, resources)
-                        }
-                        ThreadLocalExecution::Immediate => { /* already ran immediate */ }
-                    }
-                }
-            }
-        }
-
-        world.clear_trackers();
-        resources.clear_trackers();
-    }
-
-    // TODO: move this code to ParallelExecutor
-    pub fn initialize(&mut self, world: &mut World, resources: &mut Resources) {
-        if self.last_initialize_generation == self.generation {
-            return;
-        }
-
-        for stage in self.stages.values_mut() {
-            for system in stage.iter_mut() {
-                system.initialize(world, resources);
-            }
-        }
-
-        self.last_initialize_generation = self.generation;
-    }
-
-    pub fn generation(&self) -> usize {
-        self.generation
-    }
-
-    pub fn run_on_systems(
-        &mut self,
-        mut func: impl FnMut(&mut dyn System<Input = (), Output = ()>),
-    ) {
-        for stage_name in self.stage_order.iter() {
-            if let Some(stage_systems) = self.stages.get_mut(stage_name) {
-                for system in stage_systems.iter_mut() {
-                    func(&mut **system);
-                }
-            }
-        }
-    }
-}
diff --git a/crates/bevy_ecs/src/schedule/stage.rs b/crates/bevy_ecs/src/schedule/stage.rs
new file mode 100644
index 0000000000000..f15d60ebee801
--- /dev/null
+++ b/crates/bevy_ecs/src/schedule/stage.rs
@@ -0,0 +1,261 @@
+use std::{any::TypeId, borrow::Cow};
+
+use crate::{
+    ArchetypeComponent, IntoSystem, Resources, System, SystemId, ThreadLocalExecution, TypeAccess,
+    World,
+};
+use bevy_utils::HashSet;
+use downcast_rs::{impl_downcast, Downcast};
+
+use super::{ParallelSystemStageExecutor, SerialSystemStageExecutor, SystemStageExecutor};
+
+pub enum StageError {
+    SystemAlreadyExists(SystemId),
+}
+
+pub trait Stage: Downcast + Send + Sync {
+    /// Stages can perform setup here. Initialize should be called for every stage before calling [Stage::run]. Initialize will
+    /// be called once per update, so internally this should avoid re-doing work where possible.
+    fn initialize(&mut self, world: &mut World, resources: &mut Resources);
+
+    /// Runs the stage. This happens once per update (after [Stage::initialize] is called).
+    fn run(&mut self, world: &mut World, resources: &mut Resources);
+}
+
+impl_downcast!(Stage);
+
+pub struct SystemStage {
+    systems: Vec<Box<dyn System<In = (), Out = ()>>>,
+    system_ids: HashSet<SystemId>,
+    executor: Box<dyn SystemStageExecutor>,
+    run_criteria: Option<Box<dyn System<In = (), Out = ShouldRun>>>,
+    run_criteria_initialized: bool,
+    uninitialized_systems: Vec<usize>,
+    unexecuted_systems: Vec<usize>,
+}
+
+impl SystemStage {
+    pub fn new(executor: Box<dyn SystemStageExecutor>) -> Self {
+        SystemStage {
+            executor,
+            run_criteria: None,
+            run_criteria_initialized: false,
+            systems: Default::default(),
+            system_ids: Default::default(),
+            uninitialized_systems: Default::default(),
+            unexecuted_systems: Default::default(),
+        }
+    }
+
+    pub fn single<Params, S: System<In = (), Out = ()>, Into: IntoSystem<Params, S>>(
+        system: Into,
+    ) -> Self {
+        Self::serial().with_system(system)
+    }
+
+    pub fn serial() -> Self {
+        Self::new(Box::new(SerialSystemStageExecutor::default()))
+    }
+
+    pub fn parallel() -> Self {
+        Self::new(Box::new(ParallelSystemStageExecutor::default()))
+    }
+
+    pub fn with_system<S, Params, IntoS>(mut self, system: IntoS) -> Self
+    where
+        S: System<In = (), Out = ()>,
+        IntoS: IntoSystem<Params, S>,
+    {
+        self.add_system_boxed(Box::new(system.system()));
+        self
+    }
+
+    pub fn with_run_criteria<S, Params, IntoS>(mut self, system: IntoS) -> Self
+    where
+        S: System<In = (), Out = ShouldRun>,
+        IntoS: IntoSystem<Params, S>,
+    {
+        self.run_criteria = Some(Box::new(system.system()));
+        self.run_criteria_initialized = false;
+        self
+    }
+
+    pub fn add_system<S, Params, IntoS>(&mut self, system: IntoS) -> &mut Self
+    where
+        S: System<In = (), Out = ()>,
+        IntoS: IntoSystem<Params, S>,
+    {
+        self.add_system_boxed(Box::new(system.system()));
+        self
+    }
+
+    pub fn add_system_boxed(&mut self, system: Box<dyn System<In = (), Out = ()>>) -> &mut Self {
+        if self.system_ids.contains(&system.id()) {
+            panic!(
+                "System with id {:?} ({}) already exists",
+                system.id(),
+                system.name()
+            );
+        }
+        self.system_ids.insert(system.id());
+        self.unexecuted_systems.push(self.systems.len());
+        self.uninitialized_systems.push(self.systems.len());
+        self.systems.push(system);
+        self
+    }
+
+    pub fn get_executor<T: SystemStageExecutor>(&self) -> Option<&T> {
+        self.executor.downcast_ref()
+    }
+
+    pub fn get_executor_mut<T: SystemStageExecutor>(&mut self) -> Option<&mut T> {
+        self.executor.downcast_mut()
+    }
+
+    pub fn run_once(&mut self, world: &mut World, resources: &mut Resources) {
+        let unexecuted_systems = std::mem::take(&mut self.unexecuted_systems);
+        self.executor
+            .execute_stage(&mut self.systems, &unexecuted_systems, world, resources);
+    }
+}
+
+impl Stage for SystemStage {
+    fn initialize(&mut self, world: &mut World, resources: &mut Resources) {
+        if let Some(ref mut run_criteria) = self.run_criteria {
+            if !self.run_criteria_initialized {
+                run_criteria.initialize(world, resources);
+                self.run_criteria_initialized = true;
+            }
+        }
+
+        let uninitialized_systems = std::mem::take(&mut self.uninitialized_systems);
+        for system_index in uninitialized_systems.iter() {
+            self.systems[*system_index].initialize(world, resources);
+        }
+    }
+
+    fn run(&mut self, world: &mut World, resources: &mut Resources) {
+        loop {
+            let should_run = if let Some(ref mut run_criteria) = self.run_criteria {
+                let should_run = run_criteria.run((), world, resources);
+                run_criteria.run_thread_local(world, resources);
+                // don't run when no result is returned or false is returned
+                should_run.unwrap_or(ShouldRun::No)
+            } else {
+                ShouldRun::Yes
+            };
+
+            match should_run {
+                ShouldRun::No => return,
+                ShouldRun::Yes => {
+                    self.run_once(world, resources);
+                    return;
+                }
+                ShouldRun::YesAndLoop => {
+                    self.run_once(world, resources);
+                }
+            }
+        }
+    }
+}
+
+pub enum ShouldRun {
+    /// No, the system should not run
+    No,
+    /// Yes, the system should run
+    Yes,
+    /// Yes, the system should run and after running, the criteria should be checked again.
+    YesAndLoop,
+}
+
+impl<S: System<In = (), Out = ()>> From<S> for SystemStage {
+    fn from(system: S) -> Self {
+        SystemStage::single(system)
+    }
+}
+
+pub trait IntoStage<Params> {
+    type Stage: Stage;
+    fn into_stage(self) -> Self::Stage;
+}
+
+impl<Params, S: System<In = (), Out = ()>, IntoS: IntoSystem<Params, S>> IntoStage<(Params, S)>
+    for IntoS
+{
+    type Stage = SystemStage;
+
+    fn into_stage(self) -> Self::Stage {
+        SystemStage::single(self)
+    }
+}
+
+impl<S: Stage> IntoStage<()> for S {
+    type Stage = S;
+
+    fn into_stage(self) -> Self::Stage {
+        self
+    }
+}
+
+pub struct RunOnce {
+    ran: bool,
+    system_id: SystemId,
+    resource_access: TypeAccess<TypeId>,
+    archetype_access: TypeAccess<ArchetypeComponent>,
+}
+
+impl Default for RunOnce {
+    fn default() -> Self {
+        Self {
+            ran: false,
+            system_id: SystemId::new(),
+            resource_access: Default::default(),
+            archetype_access: Default::default(),
+        }
+    }
+}
+
+impl System for RunOnce {
+    type In = ();
+    type Out = ShouldRun;
+
+    fn name(&self) -> Cow<'static, str> {
+        Cow::Borrowed(std::any::type_name::<RunOnce>())
+    }
+
+    fn id(&self) -> SystemId {
+        self.system_id
+    }
+
+    fn update(&mut self, _world: &World) {}
+
+    fn archetype_component_access(&self) -> &TypeAccess<ArchetypeComponent> {
+        &self.archetype_access
+    }
+
+    fn resource_access(&self) -> &TypeAccess<TypeId> {
+        &self.resource_access
+    }
+
+    fn thread_local_execution(&self) -> ThreadLocalExecution {
+        ThreadLocalExecution::Immediate
+    }
+
+    unsafe fn run_unsafe(
+        &mut self,
+        _input: Self::In,
+        _world: &World,
+        _resources: &Resources,
+    ) -> Option<Self::Out> {
+        Some(if self.ran {
+            ShouldRun::No
+        } else {
+            self.ran = true;
+            ShouldRun::Yes
+        })
+    }
+
+    fn run_thread_local(&mut self, _world: &mut World, _resources: &mut Resources) {}
+
+    fn initialize(&mut self, _world: &mut World, _resources: &mut Resources) {}
+}
diff --git a/crates/bevy_ecs/src/schedule/parallel_executor.rs b/crates/bevy_ecs/src/schedule/stage_executor.rs
similarity index 54%
rename from crates/bevy_ecs/src/schedule/parallel_executor.rs
rename to crates/bevy_ecs/src/schedule/stage_executor.rs
index 691c31cf59465..b59a58aa457dc 100644
--- a/crates/bevy_ecs/src/schedule/parallel_executor.rs
+++ b/crates/bevy_ecs/src/schedule/stage_executor.rs
@@ -1,112 +1,66 @@
-use super::Schedule;
-use crate::{
-    resource::Resources,
-    system::{System, ThreadLocalExecution},
-    ArchetypesGeneration, TypeAccess, World,
-};
+use std::ops::Range;
+
 use bevy_tasks::{ComputeTaskPool, CountdownEvent, TaskPool};
-#[cfg(feature = "trace")]
-use bevy_utils::tracing::info_span;
 use bevy_utils::tracing::trace;
+use downcast_rs::{impl_downcast, Downcast};
 use fixedbitset::FixedBitSet;
-use std::ops::Range;
-
-/// Executes each schedule stage in parallel by analyzing system dependencies.
-/// System execution order is undefined except under the following conditions:
-/// * systems in earlier stages run before systems in later stages
-/// * in a given stage, systems that mutate [archetype+component] X cannot run before systems registered before them that read/write [archetype+component] X
-/// * in a given stage, systems the read [archetype+component] X cannot run before systems registered before them that write [archetype+component] X
-/// * in a given stage, systems that mutate resource Y cannot run before systems registered before them that read/write resource Y
-/// * in a given stage, systems the read resource Y cannot run before systems registered before them that write resource Y
 
-#[derive(Debug)]
-pub struct ParallelExecutor {
-    stages: Vec<ExecutorStage>,
-    last_schedule_generation: usize,
-    clear_trackers: bool,
-}
+use crate::{ArchetypesGeneration, Resources, System, ThreadLocalExecution, TypeAccess, World};
 
-impl Default for ParallelExecutor {
-    fn default() -> Self {
-        Self {
-            stages: Default::default(),
-            last_schedule_generation: usize::MAX, // MAX forces prepare to run the first time
-            clear_trackers: true,
-        }
-    }
+pub trait SystemStageExecutor: Downcast + Send + Sync {
+    fn execute_stage(
+        &mut self,
+        systems: &mut [Box<dyn System<In = (), Out = ()>>],
+        changed_systems: &[usize],
+        world: &mut World,
+        resources: &mut Resources,
+    );
 }
 
-impl ParallelExecutor {
-    pub fn without_tracker_clears() -> Self {
-        Self {
-            clear_trackers: false,
-            ..Default::default()
-        }
-    }
+impl_downcast!(SystemStageExecutor);
 
-    pub fn initialize(&mut self, resources: &mut Resources) {
-        if resources.get::<ComputeTaskPool>().is_none() {
-            resources.insert(ComputeTaskPool(TaskPool::default()));
-        }
-    }
+#[derive(Default)]
+pub struct SerialSystemStageExecutor;
 
-    pub fn run(&mut self, schedule: &mut Schedule, world: &mut World, resources: &mut Resources) {
-        #[cfg(feature = "trace")]
-        let schedule_span = info_span!("schedule");
-        #[cfg(feature = "trace")]
-        let _schedule_guard = schedule_span.enter();
-
-        let schedule_generation = schedule.generation();
-        let schedule_changed = schedule.generation() != self.last_schedule_generation;
-        if schedule_changed {
-            self.stages.clear();
-            self.stages
-                .resize_with(schedule.stage_order.len(), ExecutorStage::default);
-        }
-        for (stage_name, executor_stage) in schedule.stage_order.iter().zip(self.stages.iter_mut())
-        {
-            #[cfg(feature = "trace")]
-            let stage_span = info_span!("stage", name = stage_name.as_ref());
-            #[cfg(feature = "trace")]
-            let _stage_guard = stage_span.enter();
-            if let Some(stage_systems) = schedule.stages.get_mut(stage_name) {
-                executor_stage.run(world, resources, stage_systems, schedule_changed);
+impl SystemStageExecutor for SerialSystemStageExecutor {
+    fn execute_stage(
+        &mut self,
+        systems: &mut [Box<dyn System<In = (), Out = ()>>],
+        _changed_systems: &[usize],
+        world: &mut World,
+        resources: &mut Resources,
+    ) {
+        for system in systems.iter_mut() {
+            system.update(world);
+            match system.thread_local_execution() {
+                ThreadLocalExecution::NextFlush => {
+                    system.run((), world, resources);
+                }
+                ThreadLocalExecution::Immediate => {
+                    system.run((), world, resources);
+                    system.run_thread_local(world, resources);
+                }
             }
         }
 
-        if self.clear_trackers {
-            world.clear_trackers();
-            resources.clear_trackers();
-        }
-
-        self.last_schedule_generation = schedule_generation;
-    }
-
-    pub fn print_order(&self, schedule: &Schedule) {
-        println!("----------------------------");
-        for (stage_name, executor_stage) in schedule.stage_order.iter().zip(self.stages.iter()) {
-            println!("stage {:?}", stage_name);
-            if let Some(stage_systems) = schedule.stages.get(stage_name) {
-                for (i, system) in stage_systems.iter().enumerate() {
-                    println!("  {}-{}", i, system.name());
-                    println!(
-                        "      dependencies({:?})",
-                        executor_stage.system_dependencies[i]
-                            .ones()
-                            .collect::<Vec<usize>>()
-                    );
-                    println!(
-                        "      dependants({:?})",
-                        executor_stage.system_dependents[i]
-                    );
-                }
+        // "flush"
+        for system in systems.iter_mut() {
+            match system.thread_local_execution() {
+                ThreadLocalExecution::NextFlush => system.run_thread_local(world, resources),
+                ThreadLocalExecution::Immediate => { /* already ran immediate */ }
             }
         }
     }
 }
 
-#[derive(Debug, Clone)]
-pub struct ExecutorStage {
+/// Executes the stage in parallel by analyzing system dependencies.
+/// System execution order is undefined except under the following conditions:
+/// * systems in earlier stages run before systems in later stages
+/// * in a given stage, systems that mutate [archetype+component] X cannot run before systems registered before them that read/write [archetype+component] X
+/// * in a given stage, systems the read [archetype+component] X cannot run before systems registered before them that write [archetype+component] X
+/// * in a given stage, systems that mutate resource Y cannot run before systems registered before them that read/write resource Y
+/// * in a given stage, systems the read resource Y cannot run before systems registered before them that write resource Y
+pub struct ParallelSystemStageExecutor {
     /// each system's set of dependencies
     system_dependencies: Vec<FixedBitSet>,
     /// count of each system's dependencies
@@ -124,7 +78,7 @@ pub struct ExecutorStage {
     last_archetypes_generation: ArchetypesGeneration,
 }
 
-impl Default for ExecutorStage {
+impl Default for ParallelSystemStageExecutor {
     fn default() -> Self {
         Self {
             system_dependents: Default::default(),
@@ -138,7 +92,15 @@ impl Default for ExecutorStage {
     }
 }
 
-impl ExecutorStage {
+impl ParallelSystemStageExecutor {
+    pub fn system_dependents(&self) -> &[Vec<usize>] {
+        &self.system_dependents
+    }
+
+    pub fn system_dependencies(&self) -> &[FixedBitSet] {
+        &self.system_dependencies
+    }
+
     /// Sets up state to run the next "batch" of systems. Each batch contains 0..n systems and
     /// optionally a thread local system at the end. After this function runs, a bunch of state
     /// in self will be populated for systems in this batch. Returns the range of systems
@@ -147,8 +109,8 @@ impl ExecutorStage {
     pub fn prepare_to_next_thread_local(
         &mut self,
         world: &World,
-        systems: &mut [Box<dyn System<Input = (), Output = ()>>],
-        schedule_changed: bool,
+        systems: &mut [Box<dyn System<In = (), Out = ()>>],
+        stage_changed: bool,
         next_thread_local_index: usize,
     ) -> Range<usize> {
         // Find the first system in this batch and (if there is one) the thread local system that
@@ -178,7 +140,7 @@ impl ExecutorStage {
         let archetypes_generation_changed =
             self.last_archetypes_generation != world.archetypes_generation();
 
-        if schedule_changed || archetypes_generation_changed {
+        if stage_changed || archetypes_generation_changed {
             // update each system's [archetype+component] access to latest world archetypes
             for system_index in prepare_system_index_range.clone() {
                 systems[system_index].update(world);
@@ -329,7 +291,7 @@ impl ExecutorStage {
         &self,
         world: &World,
         resources: &Resources,
-        systems: &mut [Box<dyn System<Input = (), Output = ()>>],
+        systems: &mut [Box<dyn System<In = (), Out = ()>>],
         prepared_system_range: Range<usize>,
         compute_pool: &TaskPool,
     ) {
@@ -398,7 +360,10 @@ impl ExecutorStage {
                     // triggering dependents
                     {
                         #[cfg(feature = "trace")]
-                        let system_span = info_span!("system", name = system.name().as_ref());
+                        let system_span = bevy_utils::tracing::info_span!(
+                            "system",
+                            name = system.name().as_ref()
+                        );
                         #[cfg(feature = "trace")]
                         let _system_guard = system_span.enter();
 
@@ -417,21 +382,27 @@ impl ExecutorStage {
             }
         });
     }
+}
 
-    pub fn run(
+impl SystemStageExecutor for ParallelSystemStageExecutor {
+    fn execute_stage(
         &mut self,
+        systems: &mut [Box<dyn System<In = (), Out = ()>>],
+        changed_systems: &[usize],
         world: &mut World,
         resources: &mut Resources,
-        systems: &mut [Box<dyn System<Input = (), Output = ()>>],
-        schedule_changed: bool,
     ) {
         let start_archetypes_generation = world.archetypes_generation();
-        let compute_pool = resources.get_cloned::<ComputeTaskPool>().unwrap();
+        let compute_pool = resources
+            .get_or_insert_with(|| ComputeTaskPool(TaskPool::default()))
+            .clone();
+
+        let stage_changed = !changed_systems.is_empty();
 
         // if the schedule has changed, clear executor state / fill it with new defaults
         // This is mostly zeroing out a bunch of arrays parallel to the systems array. They will get
         // repopulated by prepare_to_next_thread_local() calls
-        if schedule_changed {
+        if stage_changed {
             self.system_dependencies.clear();
             self.system_dependencies
                 .resize_with(systems.len(), || FixedBitSet::with_capacity(systems.len()));
@@ -451,7 +422,8 @@ impl ExecutorStage {
             for (system_index, system) in systems.iter().enumerate() {
                 if system.thread_local_execution() == ThreadLocalExecution::Immediate {
                     #[cfg(feature = "trace")]
-                    let system_span = info_span!("system", name = system.name().as_ref());
+                    let system_span =
+                        bevy_utils::tracing::info_span!("system", name = system.name().as_ref());
                     #[cfg(feature = "trace")]
                     let _system_guard = system_span.enter();
 
@@ -470,7 +442,7 @@ impl ExecutorStage {
             let prepared_system_range = self.prepare_to_next_thread_local(
                 world,
                 systems,
-                schedule_changed,
+                stage_changed,
                 next_thread_local_index,
             );
 
@@ -498,7 +470,10 @@ impl ExecutorStage {
                 let system = systems[thread_local_system_index].as_mut();
 
                 #[cfg(feature = "trace")]
-                let system_span = info_span!("thread_local_system", name = system.name().as_ref());
+                let system_span = bevy_utils::tracing::info_span!(
+                    "thread_local_system",
+                    name = system.name().as_ref()
+                );
                 #[cfg(feature = "trace")]
                 let _system_guard = system_span.enter();
 
@@ -514,7 +489,7 @@ impl ExecutorStage {
             let run_ready_system_index_range = self.prepare_to_next_thread_local(
                 world,
                 systems,
-                schedule_changed,
+                stage_changed,
                 next_thread_local_index,
             );
 
@@ -532,7 +507,8 @@ impl ExecutorStage {
             match system.thread_local_execution() {
                 ThreadLocalExecution::NextFlush => {
                     #[cfg(feature = "trace")]
-                    let system_span = info_span!("system", name = system.name().as_ref());
+                    let system_span =
+                        bevy_utils::tracing::info_span!("system", name = system.name().as_ref());
                     #[cfg(feature = "trace")]
                     let _system_guard = system_span.enter();
                     system.run_thread_local(world, resources);
@@ -548,323 +524,3 @@ impl ExecutorStage {
         }
     }
 }
-
-#[cfg(test)]
-mod tests {
-    use super::ParallelExecutor;
-    use crate::{
-        resource::{Res, ResMut, Resources},
-        schedule::Schedule,
-        system::Query,
-        Commands, Entity, World,
-    };
-    use bevy_tasks::{ComputeTaskPool, TaskPool};
-    use fixedbitset::FixedBitSet;
-    use parking_lot::Mutex;
-    use std::{collections::HashSet, sync::Arc};
-
-    #[derive(Default)]
-    struct CompletedSystems {
-        completed_systems: Arc<Mutex<HashSet<&'static str>>>,
-    }
-
-    #[test]
-    fn cross_stage_archetype_change_prepare() {
-        let mut world = World::new();
-        let mut resources = Resources::default();
-        resources.insert(ComputeTaskPool(TaskPool::default()));
-
-        let mut schedule = Schedule::default();
-        schedule.add_stage("PreArchetypeChange");
-        schedule.add_stage("PostArchetypeChange");
-
-        fn insert(commands: &mut Commands) {
-            commands.spawn((1u32,));
-        }
-
-        fn read(query: Query<&u32>, entities: Query<Entity>) {
-            for entity in &mut entities.iter() {
-                // query.get() does a "system permission check" that will fail if the entity is from a
-                // new archetype which hasnt been "prepared yet"
-                query.get_component::<u32>(entity).unwrap();
-            }
-
-            assert_eq!(1, entities.iter().count());
-        }
-
-        schedule.add_system_to_stage("PreArchetypeChange", insert);
-        schedule.add_system_to_stage("PostArchetypeChange", read);
-
-        let mut executor = ParallelExecutor::default();
-        schedule.initialize(&mut world, &mut resources);
-        executor.run(&mut schedule, &mut world, &mut resources);
-    }
-
-    #[test]
-    fn intra_stage_archetype_change_prepare() {
-        let mut world = World::new();
-        let mut resources = Resources::default();
-        resources.insert(ComputeTaskPool(TaskPool::default()));
-
-        let mut schedule = Schedule::default();
-        schedule.add_stage("update");
-
-        fn insert(world: &mut World, _resources: &mut Resources) {
-            world.spawn((1u32,));
-        }
-
-        fn read(query: Query<&u32>, entities: Query<Entity>) {
-            for entity in &mut entities.iter() {
-                // query.get() does a "system permission check" that will fail if the entity is from a
-                // new archetype which hasnt been "prepared yet"
-                query.get_component::<u32>(entity).unwrap();
-            }
-
-            assert_eq!(1, entities.iter().count());
-        }
-
-        schedule.add_system_to_stage("update", insert);
-        schedule.add_system_to_stage("update", read);
-        schedule.initialize(&mut world, &mut resources);
-
-        let mut executor = ParallelExecutor::default();
-        executor.run(&mut schedule, &mut world, &mut resources);
-    }
-
-    #[test]
-    fn schedule() {
-        let mut world = World::new();
-        let mut resources = Resources::default();
-        resources.insert(ComputeTaskPool(TaskPool::default()));
-        resources.insert(CompletedSystems::default());
-        resources.insert(1.0f64);
-        resources.insert(2isize);
-
-        world.spawn((1.0f32,));
-        world.spawn((1u32, 1u64));
-        world.spawn((2u32,));
-
-        let mut schedule = Schedule::default();
-        schedule.add_stage("A"); // component queries
-        schedule.add_stage("B"); // thread local
-        schedule.add_stage("C"); // resources
-
-        // A system names
-        const READ_U32_SYSTEM_NAME: &str = "read_u32";
-        const WRITE_FLOAT_SYSTEM_NAME: &str = "write_float";
-        const READ_U32_WRITE_U64_SYSTEM_NAME: &str = "read_u32_write_u64";
-        const READ_U64_SYSTEM_NAME: &str = "read_u64";
-
-        // B system names
-        const WRITE_U64_SYSTEM_NAME: &str = "write_u64";
-        const THREAD_LOCAL_SYSTEM_SYSTEM_NAME: &str = "thread_local_system";
-        const WRITE_F32_SYSTEM_NAME: &str = "write_f32";
-
-        // C system names
-        const READ_F64_RES_SYSTEM_NAME: &str = "read_f64_res";
-        const READ_ISIZE_RES_SYSTEM_NAME: &str = "read_isize_res";
-        const READ_ISIZE_WRITE_F64_RES_SYSTEM_NAME: &str = "read_isize_write_f64_res";
-        const WRITE_F64_RES_SYSTEM_NAME: &str = "write_f64_res";
-
-        // A systems
-
-        fn read_u32(completed_systems: Res<CompletedSystems>, _query: Query<&u32>) {
-            let mut completed_systems = completed_systems.completed_systems.lock();
-            completed_systems.insert(READ_U32_SYSTEM_NAME);
-        }
-
-        fn write_float(completed_systems: Res<CompletedSystems>, _query: Query<&f32>) {
-            let mut completed_systems = completed_systems.completed_systems.lock();
-            completed_systems.insert(WRITE_FLOAT_SYSTEM_NAME);
-        }
-
-        fn read_u32_write_u64(
-            completed_systems: Res<CompletedSystems>,
-            _query: Query<(&u32, &mut u64)>,
-        ) {
-            let mut completed_systems = completed_systems.completed_systems.lock();
-            assert!(!completed_systems.contains(READ_U64_SYSTEM_NAME));
-            completed_systems.insert(READ_U32_WRITE_U64_SYSTEM_NAME);
-        }
-
-        fn read_u64(completed_systems: Res<CompletedSystems>, _query: Query<&u64>) {
-            let mut completed_systems = completed_systems.completed_systems.lock();
-            assert!(completed_systems.contains(READ_U32_WRITE_U64_SYSTEM_NAME));
-            assert!(!completed_systems.contains(WRITE_U64_SYSTEM_NAME));
-            completed_systems.insert(READ_U64_SYSTEM_NAME);
-        }
-
-        schedule.add_system_to_stage("A", read_u32);
-        schedule.add_system_to_stage("A", write_float);
-        schedule.add_system_to_stage("A", read_u32_write_u64);
-        schedule.add_system_to_stage("A", read_u64);
-
-        // B systems
-
-        fn write_u64(completed_systems: Res<CompletedSystems>, _query: Query<&mut u64>) {
-            let mut completed_systems = completed_systems.completed_systems.lock();
-            assert!(completed_systems.contains(READ_U64_SYSTEM_NAME));
-            assert!(!completed_systems.contains(THREAD_LOCAL_SYSTEM_SYSTEM_NAME));
-            assert!(!completed_systems.contains(WRITE_F32_SYSTEM_NAME));
-            completed_systems.insert(WRITE_U64_SYSTEM_NAME);
-        }
-
-        fn thread_local_system(_world: &mut World, resources: &mut Resources) {
-            let completed_systems = resources.get::<CompletedSystems>().unwrap();
-            let mut completed_systems = completed_systems.completed_systems.lock();
-            assert!(completed_systems.contains(WRITE_U64_SYSTEM_NAME));
-            assert!(!completed_systems.contains(WRITE_F32_SYSTEM_NAME));
-            completed_systems.insert(THREAD_LOCAL_SYSTEM_SYSTEM_NAME);
-        }
-
-        fn write_f32(completed_systems: Res<CompletedSystems>, _query: Query<&mut f32>) {
-            let mut completed_systems = completed_systems.completed_systems.lock();
-            assert!(completed_systems.contains(WRITE_U64_SYSTEM_NAME));
-            assert!(completed_systems.contains(THREAD_LOCAL_SYSTEM_SYSTEM_NAME));
-            assert!(!completed_systems.contains(READ_F64_RES_SYSTEM_NAME));
-            completed_systems.insert(WRITE_F32_SYSTEM_NAME);
-        }
-
-        schedule.add_system_to_stage("B", write_u64);
-        schedule.add_system_to_stage("B", thread_local_system);
-        schedule.add_system_to_stage("B", write_f32);
-
-        // C systems
-
-        fn read_f64_res(completed_systems: Res<CompletedSystems>, _f64_res: Res<f64>) {
-            let mut completed_systems = completed_systems.completed_systems.lock();
-            assert!(completed_systems.contains(WRITE_F32_SYSTEM_NAME));
-            assert!(!completed_systems.contains(READ_ISIZE_WRITE_F64_RES_SYSTEM_NAME));
-            assert!(!completed_systems.contains(WRITE_F64_RES_SYSTEM_NAME));
-            completed_systems.insert(READ_F64_RES_SYSTEM_NAME);
-        }
-
-        fn read_isize_res(completed_systems: Res<CompletedSystems>, _isize_res: Res<isize>) {
-            let mut completed_systems = completed_systems.completed_systems.lock();
-            completed_systems.insert(READ_ISIZE_RES_SYSTEM_NAME);
-        }
-
-        fn read_isize_write_f64_res(
-            completed_systems: Res<CompletedSystems>,
-            _isize_res: Res<isize>,
-            _f64_res: ResMut<f64>,
-        ) {
-            let mut completed_systems = completed_systems.completed_systems.lock();
-            assert!(completed_systems.contains(READ_F64_RES_SYSTEM_NAME));
-            assert!(!completed_systems.contains(WRITE_F64_RES_SYSTEM_NAME));
-            completed_systems.insert(READ_ISIZE_WRITE_F64_RES_SYSTEM_NAME);
-        }
-
-        fn write_f64_res(completed_systems: Res<CompletedSystems>, _f64_res: ResMut<f64>) {
-            let mut completed_systems = completed_systems.completed_systems.lock();
-            assert!(completed_systems.contains(READ_F64_RES_SYSTEM_NAME));
-            assert!(completed_systems.contains(READ_ISIZE_WRITE_F64_RES_SYSTEM_NAME));
-            completed_systems.insert(WRITE_F64_RES_SYSTEM_NAME);
-        }
-
-        schedule.add_system_to_stage("C", read_f64_res);
-        schedule.add_system_to_stage("C", read_isize_res);
-        schedule.add_system_to_stage("C", read_isize_write_f64_res);
-        schedule.add_system_to_stage("C", write_f64_res);
-        schedule.initialize(&mut world, &mut resources);
-
-        fn run_executor_and_validate(
-            executor: &mut ParallelExecutor,
-            schedule: &mut Schedule,
-            world: &mut World,
-            resources: &mut Resources,
-        ) {
-            executor.run(schedule, world, resources);
-
-            assert_eq!(
-                executor.stages[0].system_dependents,
-                vec![vec![], vec![], vec![3], vec![]]
-            );
-            assert_eq!(
-                executor.stages[1].system_dependents,
-                vec![vec![1], vec![2], vec![]]
-            );
-            assert_eq!(
-                executor.stages[2].system_dependents,
-                vec![vec![2, 3], vec![], vec![3], vec![]]
-            );
-
-            let stage_0_len = executor.stages[0].system_dependencies.len();
-            let mut read_u64_deps = FixedBitSet::with_capacity(stage_0_len);
-            read_u64_deps.insert(2);
-
-            assert_eq!(
-                executor.stages[0].system_dependencies,
-                vec![
-                    FixedBitSet::with_capacity(stage_0_len),
-                    FixedBitSet::with_capacity(stage_0_len),
-                    FixedBitSet::with_capacity(stage_0_len),
-                    read_u64_deps,
-                ]
-            );
-
-            let stage_1_len = executor.stages[1].system_dependencies.len();
-            let mut thread_local_deps = FixedBitSet::with_capacity(stage_1_len);
-            thread_local_deps.insert(0);
-            let mut write_f64_deps = FixedBitSet::with_capacity(stage_1_len);
-            write_f64_deps.insert(1);
-            assert_eq!(
-                executor.stages[1].system_dependencies,
-                vec![
-                    FixedBitSet::with_capacity(stage_1_len),
-                    thread_local_deps,
-                    write_f64_deps
-                ]
-            );
-
-            let stage_2_len = executor.stages[2].system_dependencies.len();
-            let mut read_isize_write_f64_res_deps = FixedBitSet::with_capacity(stage_2_len);
-            read_isize_write_f64_res_deps.insert(0);
-            let mut write_f64_res_deps = FixedBitSet::with_capacity(stage_2_len);
-            write_f64_res_deps.insert(0);
-            write_f64_res_deps.insert(2);
-            assert_eq!(
-                executor.stages[2].system_dependencies,
-                vec![
-                    FixedBitSet::with_capacity(stage_2_len),
-                    FixedBitSet::with_capacity(stage_2_len),
-                    read_isize_write_f64_res_deps,
-                    write_f64_res_deps
-                ]
-            );
-
-            let completed_systems = resources.get::<CompletedSystems>().unwrap();
-            assert_eq!(
-                completed_systems.completed_systems.lock().len(),
-                11,
-                "completed_systems should have been incremented once for each system"
-            );
-        }
-
-        // Stress test the "clean start" case
-        for _ in 0..1000 {
-            let mut executor = ParallelExecutor::default();
-            run_executor_and_validate(&mut executor, &mut schedule, &mut world, &mut resources);
-            resources
-                .get::<CompletedSystems>()
-                .unwrap()
-                .completed_systems
-                .lock()
-                .clear();
-        }
-
-        // Stress test the "continue running" case
-        let mut executor = ParallelExecutor::default();
-        run_executor_and_validate(&mut executor, &mut schedule, &mut world, &mut resources);
-        for _ in 0..1000 {
-            // run again (with completed_systems reset) to ensure executor works correctly across runs
-            resources
-                .get::<CompletedSystems>()
-                .unwrap()
-                .completed_systems
-                .lock()
-                .clear();
-            run_executor_and_validate(&mut executor, &mut schedule, &mut world, &mut resources);
-        }
-    }
-}
diff --git a/crates/bevy_ecs/src/schedule/state.rs b/crates/bevy_ecs/src/schedule/state.rs
new file mode 100644
index 0000000000000..629f926336b28
--- /dev/null
+++ b/crates/bevy_ecs/src/schedule/state.rs
@@ -0,0 +1,224 @@
+use crate::{IntoStage, Resource, Resources, Stage, World};
+use bevy_utils::HashMap;
+use std::{mem::Discriminant, ops::Deref};
+use thiserror::Error;
+
+#[derive(Default)]
+pub(crate) struct StateStages {
+    update: Option<Box<dyn Stage>>,
+    enter: Option<Box<dyn Stage>>,
+    exit: Option<Box<dyn Stage>>,
+}
+
+pub struct StateStage<T> {
+    stages: HashMap<Discriminant<T>, StateStages>,
+}
+
+impl<T> Default for StateStage<T> {
+    fn default() -> Self {
+        Self {
+            stages: Default::default(),
+        }
+    }
+}
+
+#[allow(clippy::mem_discriminant_non_enum)]
+impl<T> StateStage<T> {
+    pub fn with_on_state_enter<Params, S: IntoStage<Params>>(mut self, state: T, stage: S) -> Self {
+        self.on_state_enter(state, stage);
+        self
+    }
+
+    pub fn with_on_state_exit<Params, S: IntoStage<Params>>(mut self, state: T, stage: S) -> Self {
+        self.on_state_exit(state, stage);
+        self
+    }
+
+    pub fn with_on_state_update<Params, S: IntoStage<Params>>(
+        mut self,
+        state: T,
+        stage: S,
+    ) -> Self {
+        self.on_state_update(state, stage);
+        self
+    }
+
+    pub fn on_state_enter<Params, S: IntoStage<Params>>(
+        &mut self,
+        state: T,
+        stage: S,
+    ) -> &mut Self {
+        let stages = self
+            .stages
+            .entry(std::mem::discriminant(&state))
+            .or_default();
+        stages.enter = Some(Box::new(stage.into_stage()));
+        self
+    }
+
+    pub fn on_state_exit<Params, S: IntoStage<Params>>(&mut self, state: T, stage: S) -> &mut Self {
+        let stages = self
+            .stages
+            .entry(std::mem::discriminant(&state))
+            .or_default();
+        stages.exit = Some(Box::new(stage.into_stage()));
+        self
+    }
+
+    pub fn on_state_update<Params, S: IntoStage<Params>>(
+        &mut self,
+        state: T,
+        stage: S,
+    ) -> &mut Self {
+        let stages = self
+            .stages
+            .entry(std::mem::discriminant(&state))
+            .or_default();
+        stages.update = Some(Box::new(stage.into_stage()));
+        self
+    }
+}
+
+#[allow(clippy::mem_discriminant_non_enum)]
+impl<T: Resource + Clone> Stage for StateStage<T> {
+    fn initialize(&mut self, world: &mut World, resources: &mut Resources) {
+        for state_stages in self.stages.values_mut() {
+            if let Some(ref mut enter) = state_stages.enter {
+                enter.initialize(world, resources);
+            }
+
+            if let Some(ref mut update) = state_stages.update {
+                update.initialize(world, resources);
+            }
+
+            if let Some(ref mut exit) = state_stages.exit {
+                exit.initialize(world, resources);
+            }
+        }
+    }
+
+    fn run(&mut self, world: &mut World, resources: &mut Resources) {
+        loop {
+            let (next_stage, current_stage) = {
+                let mut state = resources
+                    .get_mut::<State<T>>()
+                    .expect("Missing state resource");
+                let result = (
+                    state.next.as_ref().map(|next| std::mem::discriminant(next)),
+                    std::mem::discriminant(&state.current),
+                );
+
+                state.apply_next();
+
+                result
+            };
+
+            // if next_stage is Some, we just applied a new state
+            if let Some(next_stage) = next_stage {
+                if next_stage != current_stage {
+                    if let Some(exit_current) = self
+                        .stages
+                        .get_mut(&current_stage)
+                        .and_then(|stage| stage.exit.as_mut())
+                    {
+                        exit_current.run(world, resources);
+                    }
+                }
+
+                if let Some(enter_next) = self
+                    .stages
+                    .get_mut(&next_stage)
+                    .and_then(|stage| stage.enter.as_mut())
+                {
+                    enter_next.run(world, resources);
+                }
+            } else if let Some(update_current) = self
+                .stages
+                .get_mut(&current_stage)
+                .and_then(|stage| stage.update.as_mut())
+            {
+                update_current.run(world, resources);
+                break;
+            }
+        }
+    }
+}
+#[derive(Debug, Error)]
+pub enum StateError {
+    #[error("Attempted to change the state to the current state.")]
+    AlreadyInState,
+    #[error("Attempted to queue a state change, but there was already a state queued.")]
+    StateAlreadyQueued,
+}
+
+#[derive(Debug)]
+pub struct State<T: Clone> {
+    previous: Option<T>,
+    current: T,
+    next: Option<T>,
+}
+
+#[allow(clippy::mem_discriminant_non_enum)]
+impl<T: Clone> State<T> {
+    pub fn new(state: T) -> Self {
+        Self {
+            current: state.clone(),
+            previous: None,
+            // add value to queue so that we "enter" the state
+            next: Some(state),
+        }
+    }
+
+    pub fn current(&self) -> &T {
+        &self.current
+    }
+
+    pub fn previous(&self) -> Option<&T> {
+        self.previous.as_ref()
+    }
+
+    pub fn next(&self) -> Option<&T> {
+        self.next.as_ref()
+    }
+
+    /// Queue a state change. This will fail if there is already a state in the queue, or if the given `state` matches the current state
+    pub fn set_next(&mut self, state: T) -> Result<(), StateError> {
+        if std::mem::discriminant(&self.current) == std::mem::discriminant(&state) {
+            return Err(StateError::AlreadyInState);
+        }
+
+        if self.next.is_some() {
+            return Err(StateError::StateAlreadyQueued);
+        }
+
+        self.next = Some(state);
+        Ok(())
+    }
+
+    /// Same as [Self::queue], but there is already a next state, it will be overwritten instead of failing
+    pub fn overwrite_next(&mut self, state: T) -> Result<(), StateError> {
+        if std::mem::discriminant(&self.current) == std::mem::discriminant(&state) {
+            return Err(StateError::AlreadyInState);
+        }
+
+        self.next = Some(state);
+        Ok(())
+    }
+
+    fn apply_next(&mut self) {
+        if let Some(next) = self.next.take() {
+            let previous = std::mem::replace(&mut self.current, next);
+            if std::mem::discriminant(&previous) != std::mem::discriminant(&self.current) {
+                self.previous = Some(previous)
+            }
+        }
+    }
+}
+
+impl<T: Clone> Deref for State<T> {
+    type Target = T;
+
+    fn deref(&self) -> &Self::Target {
+        &self.current
+    }
+}
diff --git a/crates/bevy_ecs/src/system/into_system.rs b/crates/bevy_ecs/src/system/into_system.rs
index 2d80b75fc1388..967fe9666d3b7 100644
--- a/crates/bevy_ecs/src/system/into_system.rs
+++ b/crates/bevy_ecs/src/system/into_system.rs
@@ -8,7 +8,6 @@ use std::{any::TypeId, borrow::Cow, sync::Arc};
 pub struct SystemState {
     pub(crate) id: SystemId,
     pub(crate) name: Cow<'static, str>,
-    pub(crate) is_initialized: bool,
     pub(crate) archetype_component_access: TypeAccess<ArchetypeComponent>,
     pub(crate) resource_access: TypeAccess<TypeId>,
     pub(crate) local_resource_access: TypeAccess<TypeId>,
@@ -73,12 +72,9 @@ impl SystemState {
     }
 }
 
-pub struct FuncSystem<Input, Return> {
+pub struct FuncSystem<In, Out> {
     func: Box<
-        dyn FnMut(Input, &mut SystemState, &World, &Resources) -> Option<Return>
-            + Send
-            + Sync
-            + 'static,
+        dyn FnMut(In, &mut SystemState, &World, &Resources) -> Option<Out> + Send + Sync + 'static,
     >,
     thread_local_func:
         Box<dyn FnMut(&mut SystemState, &mut World, &mut Resources) + Send + Sync + 'static>,
@@ -86,9 +82,9 @@ pub struct FuncSystem<Input, Return> {
     state: SystemState,
 }
 
-impl<Input: 'static, Output: 'static> System for FuncSystem<Input, Output> {
-    type Input = Input;
-    type Output = Output;
+impl<In: 'static, Out: 'static> System for FuncSystem<In, Out> {
+    type In = In;
+    type Out = Out;
 
     fn name(&self) -> std::borrow::Cow<'static, str> {
         self.state.name.clone()
@@ -116,10 +112,10 @@ impl<Input: 'static, Output: 'static> System for FuncSystem<Input, Output> {
 
     unsafe fn run_unsafe(
         &mut self,
-        input: Input,
+        input: In,
         world: &World,
         resources: &Resources,
-    ) -> Option<Output> {
+    ) -> Option<Out> {
         (self.func)(input, &mut self.state, world, resources)
     }
 
@@ -129,11 +125,6 @@ impl<Input: 'static, Output: 'static> System for FuncSystem<Input, Output> {
 
     fn initialize(&mut self, world: &mut World, resources: &mut Resources) {
         (self.init_func)(&mut self.state, world, resources);
-        self.state.is_initialized = true;
-    }
-
-    fn is_initialized(&self) -> bool {
-        self.state.is_initialized
     }
 }
 
@@ -150,20 +141,19 @@ impl<Sys: System> IntoSystem<(), Sys> for Sys {
 
 macro_rules! impl_into_system {
     ($($param: ident),*) => {
-        impl<Func, Input, Return, $($param: SystemParam<Input>),*> IntoSystem<($($param,)*), FuncSystem<Input, Return>> for Func
-        where Func: FnMut($($param),*) -> Return + Send + Sync + 'static, Return: 'static, Input: 'static
+        impl<Func, In, Out, $($param: SystemParam<In>),*> IntoSystem<($($param,)*), FuncSystem<In, Out>> for Func
+        where Func: FnMut($($param),*) -> Out + Send + Sync + 'static, Out: 'static, In: 'static
         {
             #[allow(unused_variables)]
             #[allow(unused_unsafe)]
             #[allow(non_snake_case)]
-            fn system(mut self) -> FuncSystem<Input, Return> {
+            fn system(mut self) -> FuncSystem<In, Out> {
                 FuncSystem {
                     state: SystemState {
                         name: std::any::type_name::<Self>().into(),
                         archetype_component_access: TypeAccess::default(),
                         resource_access: TypeAccess::default(),
                         local_resource_access: TypeAccess::default(),
-                        is_initialized: false,
                         id: SystemId::new(),
                         commands: Commands::default(),
                         arc_commands: Default::default(),
@@ -221,9 +211,10 @@ impl_into_system!(A, B, C, D, E, F, G, H, I, J, K, L, M, N, O, P);
 mod tests {
     use super::IntoSystem;
     use crate::{
+        clear_trackers_system,
         resource::{Res, ResMut, Resources},
         schedule::Schedule,
-        ChangedRes, Entity, Local, Or, Query, QuerySet, System, With, World,
+        ChangedRes, Entity, Local, Or, Query, QuerySet, System, SystemStage, With, World,
     };
 
     #[derive(Debug, Eq, PartialEq, Default)]
@@ -336,18 +327,19 @@ mod tests {
         let ent = world.spawn((0,));
 
         let mut schedule = Schedule::default();
-        schedule.add_stage("update");
-        schedule.add_system_to_stage("update", incr_e_on_flip);
-        schedule.initialize(&mut world, &mut resources);
+        let mut update = SystemStage::parallel();
+        update.add_system(incr_e_on_flip);
+        schedule.add_stage("update", update);
+        schedule.add_stage("clear_trackers", clear_trackers_system);
 
-        schedule.run(&mut world, &mut resources);
+        schedule.initialize_and_run(&mut world, &mut resources);
         assert_eq!(*(world.get::<i32>(ent).unwrap()), 1);
 
-        schedule.run(&mut world, &mut resources);
+        schedule.initialize_and_run(&mut world, &mut resources);
         assert_eq!(*(world.get::<i32>(ent).unwrap()), 1);
 
         *resources.get_mut::<bool>().unwrap() = true;
-        schedule.run(&mut world, &mut resources);
+        schedule.initialize_and_run(&mut world, &mut resources);
         assert_eq!(*(world.get::<i32>(ent).unwrap()), 2);
     }
 
@@ -369,25 +361,26 @@ mod tests {
         let ent = world.spawn((0,));
 
         let mut schedule = Schedule::default();
-        schedule.add_stage("update");
-        schedule.add_system_to_stage("update", incr_e_on_flip);
-        schedule.initialize(&mut world, &mut resources);
+        let mut update = SystemStage::parallel();
+        update.add_system(incr_e_on_flip);
+        schedule.add_stage("update", update);
+        schedule.add_stage("clear_trackers", clear_trackers_system);
 
-        schedule.run(&mut world, &mut resources);
+        schedule.initialize_and_run(&mut world, &mut resources);
         assert_eq!(*(world.get::<i32>(ent).unwrap()), 1);
 
-        schedule.run(&mut world, &mut resources);
+        schedule.initialize_and_run(&mut world, &mut resources);
         assert_eq!(*(world.get::<i32>(ent).unwrap()), 1);
 
         *resources.get_mut::<bool>().unwrap() = true;
-        schedule.run(&mut world, &mut resources);
+        schedule.initialize_and_run(&mut world, &mut resources);
         assert_eq!(*(world.get::<i32>(ent).unwrap()), 2);
 
-        schedule.run(&mut world, &mut resources);
+        schedule.initialize_and_run(&mut world, &mut resources);
         assert_eq!(*(world.get::<i32>(ent).unwrap()), 2);
 
         *resources.get_mut::<i32>().unwrap() = 20;
-        schedule.run(&mut world, &mut resources);
+        schedule.initialize_and_run(&mut world, &mut resources);
         assert_eq!(*(world.get::<i32>(ent).unwrap()), 3);
     }
 
@@ -451,7 +444,7 @@ mod tests {
 
     fn run_system<
         Params,
-        SystemType: System<Input = (), Output = ()>,
+        SystemType: System<In = (), Out = ()>,
         Sys: IntoSystem<Params, SystemType>,
     >(
         world: &mut World,
@@ -459,11 +452,10 @@ mod tests {
         system: Sys,
     ) {
         let mut schedule = Schedule::default();
-        schedule.add_stage("update");
-        schedule.add_system_to_stage("update", system);
-
-        schedule.initialize(world, resources);
-        schedule.run(world, resources);
+        let mut update = SystemStage::parallel();
+        update.add_system(system);
+        schedule.add_stage("update", update);
+        schedule.initialize_and_run(world, resources);
     }
 
     #[derive(Default)]
@@ -473,7 +465,7 @@ mod tests {
 
     fn test_for_conflicting_resources<
         Params,
-        SystemType: System<Input = (), Output = ()>,
+        SystemType: System<In = (), Out = ()>,
         Sys: IntoSystem<Params, SystemType>,
     >(
         sys: Sys,
diff --git a/crates/bevy_ecs/src/system/into_thread_local.rs b/crates/bevy_ecs/src/system/into_thread_local.rs
index 4fe61b3da74b0..d08f33a200d7a 100644
--- a/crates/bevy_ecs/src/system/into_thread_local.rs
+++ b/crates/bevy_ecs/src/system/into_thread_local.rs
@@ -15,8 +15,8 @@ pub struct ThreadLocalSystemFn {
 }
 
 impl System for ThreadLocalSystemFn {
-    type Input = ();
-    type Output = ();
+    type In = ();
+    type Out = ();
 
     fn name(&self) -> Cow<'static, str> {
         self.name.clone()
@@ -54,10 +54,6 @@ impl System for ThreadLocalSystemFn {
     fn id(&self) -> SystemId {
         self.id
     }
-
-    fn is_initialized(&self) -> bool {
-        true
-    }
 }
 
 impl<F> IntoSystem<(&mut World, &mut Resources), ThreadLocalSystemFn> for F
diff --git a/crates/bevy_ecs/src/system/system.rs b/crates/bevy_ecs/src/system/system.rs
index 85bb0cefd7ea5..fac4eab2e9442 100644
--- a/crates/bevy_ecs/src/system/system.rs
+++ b/crates/bevy_ecs/src/system/system.rs
@@ -20,11 +20,10 @@ impl SystemId {
 
 /// An ECS system that can be added to a [Schedule](crate::Schedule)
 pub trait System: Send + Sync + 'static {
-    type Input;
-    type Output;
+    type In;
+    type Out;
     fn name(&self) -> Cow<'static, str>;
     fn id(&self) -> SystemId;
-    fn is_initialized(&self) -> bool;
     fn update(&mut self, world: &World);
     fn archetype_component_access(&self) -> &TypeAccess<ArchetypeComponent>;
     fn resource_access(&self) -> &TypeAccess<TypeId>;
@@ -35,16 +34,16 @@ pub trait System: Send + Sync + 'static {
     /// 2. This system only runs in parallel with other systems that do not conflict with the `archetype_component_access()` or `resource_access()`
     unsafe fn run_unsafe(
         &mut self,
-        input: Self::Input,
+        input: Self::In,
         world: &World,
         resources: &Resources,
-    ) -> Option<Self::Output>;
+    ) -> Option<Self::Out>;
     fn run(
         &mut self,
-        input: Self::Input,
+        input: Self::In,
         world: &mut World,
         resources: &mut Resources,
-    ) -> Option<Self::Output> {
+    ) -> Option<Self::Out> {
         // SAFE: world and resources are exclusively borrowed
         unsafe { self.run_unsafe(input, world, resources) }
     }
diff --git a/crates/bevy_ecs/src/system/system_chaining.rs b/crates/bevy_ecs/src/system/system_chaining.rs
index 5d5397c80c654..7822b3d85d7db 100644
--- a/crates/bevy_ecs/src/system/system_chaining.rs
+++ b/crates/bevy_ecs/src/system/system_chaining.rs
@@ -13,11 +13,9 @@ pub struct ChainSystem<SystemA, SystemB> {
     pub(crate) resource_access: TypeAccess<TypeId>,
 }
 
-impl<SystemA: System, SystemB: System<Input = SystemA::Output>> System
-    for ChainSystem<SystemA, SystemB>
-{
-    type Input = SystemA::Input;
-    type Output = SystemB::Output;
+impl<SystemA: System, SystemB: System<In = SystemA::Out>> System for ChainSystem<SystemA, SystemB> {
+    type In = SystemA::In;
+    type Out = SystemB::Out;
 
     fn name(&self) -> Cow<'static, str> {
         self.name.clone()
@@ -27,10 +25,6 @@ impl<SystemA: System, SystemB: System<Input = SystemA::Output>> System
         self.id
     }
 
-    fn is_initialized(&self) -> bool {
-        self.system_a.is_initialized() && self.system_b.is_initialized()
-    }
-
     fn update(&mut self, world: &World) {
         self.archetype_component_access.clear();
         self.resource_access.clear();
@@ -56,10 +50,10 @@ impl<SystemA: System, SystemB: System<Input = SystemA::Output>> System
 
     unsafe fn run_unsafe(
         &mut self,
-        input: Self::Input,
+        input: Self::In,
         world: &World,
         resources: &Resources,
-    ) -> Option<Self::Output> {
+    ) -> Option<Self::Out> {
         let out = self.system_a.run_unsafe(input, world, resources).unwrap();
         self.system_b.run_unsafe(out, world, resources)
     }
@@ -80,7 +74,7 @@ pub trait IntoChainSystem<AParams, BParams, IntoB, SystemA, SystemB>:
 where
     IntoB: IntoSystem<BParams, SystemB>,
     SystemA: System,
-    SystemB: System<Input = SystemA::Output>,
+    SystemB: System<In = SystemA::Out>,
 {
     fn chain(self, system: IntoB) -> ChainSystem<SystemA, SystemB>;
 }
@@ -89,7 +83,7 @@ impl<AParams, BParams, IntoA, IntoB, SystemA, SystemB>
     IntoChainSystem<AParams, BParams, IntoB, SystemA, SystemB> for IntoA
 where
     SystemA: System,
-    SystemB: System<Input = SystemA::Output>,
+    SystemB: System<In = SystemA::Out>,
     IntoA: IntoSystem<AParams, SystemA>,
     IntoB: IntoSystem<BParams, SystemB>,
 {
diff --git a/crates/bevy_input/src/lib.rs b/crates/bevy_input/src/lib.rs
index a21fbca1e4b69..4ef0ccaeef52e 100644
--- a/crates/bevy_input/src/lib.rs
+++ b/crates/bevy_input/src/lib.rs
@@ -27,7 +27,6 @@ use keyboard::{keyboard_input_system, KeyCode, KeyboardInput};
 use mouse::{mouse_button_input_system, MouseButton, MouseButtonInput, MouseMotion, MouseWheel};
 use touch::{touch_screen_input_system, TouchInput, Touches};
 
-use bevy_app::startup_stage::STARTUP;
 use gamepad::{
     gamepad_event_system, GamepadAxis, GamepadButton, GamepadEvent, GamepadEventRaw,
     GamepadSettings,
@@ -54,7 +53,7 @@ impl Plugin for InputPlugin {
             .init_resource::<Axis<GamepadAxis>>()
             .init_resource::<Axis<GamepadButton>>()
             .add_system_to_stage(bevy_app::stage::EVENT, gamepad_event_system)
-            .add_startup_system_to_stage(STARTUP, gamepad_event_system)
+            .add_startup_system_to_stage(bevy_app::startup_stage::STARTUP, gamepad_event_system)
             .add_event::<TouchInput>()
             .init_resource::<Touches>()
             .add_system_to_stage(bevy_app::stage::EVENT, touch_screen_input_system);
diff --git a/crates/bevy_pbr/src/render_graph/lights_node.rs b/crates/bevy_pbr/src/render_graph/lights_node.rs
index 12b720a973a1a..a4a5d4bbe1fdb 100644
--- a/crates/bevy_pbr/src/render_graph/lights_node.rs
+++ b/crates/bevy_pbr/src/render_graph/lights_node.rs
@@ -51,7 +51,7 @@ struct LightCount {
 unsafe impl Byteable for LightCount {}
 
 impl SystemNode for LightsNode {
-    fn get_system(&self, commands: &mut Commands) -> Box<dyn System<Input = (), Output = ()>> {
+    fn get_system(&self, commands: &mut Commands) -> Box<dyn System<In = (), Out = ()>> {
         let system = lights_node_system.system();
         commands.insert_local_resource(
             system.id(),
diff --git a/crates/bevy_render/src/lib.rs b/crates/bevy_render/src/lib.rs
index 506b597c67fae..2aa74b8909e60 100644
--- a/crates/bevy_render/src/lib.rs
+++ b/crates/bevy_render/src/lib.rs
@@ -11,6 +11,7 @@ pub mod renderer;
 pub mod shader;
 pub mod texture;
 
+use bevy_ecs::SystemStage;
 use bevy_reflect::RegisterTypeBuilder;
 use draw::Visible;
 pub use once_cell;
@@ -95,59 +96,70 @@ impl Plugin for RenderPlugin {
             app.resources_mut().insert(ClearColor::default());
         }
 
-        app.add_stage_after(bevy_asset::stage::ASSET_EVENTS, stage::RENDER_RESOURCE)
-            .add_stage_after(stage::RENDER_RESOURCE, stage::RENDER_GRAPH_SYSTEMS)
-            .add_stage_after(stage::RENDER_GRAPH_SYSTEMS, stage::DRAW)
-            .add_stage_after(stage::DRAW, stage::RENDER)
-            .add_stage_after(stage::RENDER, stage::POST_RENDER)
-            .add_asset::<Mesh>()
-            .add_asset::<Texture>()
-            .add_asset::<Shader>()
-            .add_asset::<PipelineDescriptor>()
-            .register_type::<Camera>()
-            .register_type::<Draw>()
-            .register_type::<Visible>()
-            .register_type::<RenderPipelines>()
-            .register_type::<OrthographicProjection>()
-            .register_type::<PerspectiveProjection>()
-            .register_type::<MainPass>()
-            .register_type::<VisibleEntities>()
-            .register_type::<Color>()
-            .register_type::<ShaderSpecialization>()
-            .register_type::<PrimitiveTopology>()
-            .register_type::<IndexFormat>()
-            .register_type::<PipelineSpecialization>()
-            .init_resource::<RenderGraph>()
-            .init_resource::<PipelineCompiler>()
-            .init_resource::<RenderResourceBindings>()
-            .init_resource::<TextureResourceSystemState>()
-            .init_resource::<AssetRenderResourceBindings>()
-            .init_resource::<ActiveCameras>()
-            .add_system_to_stage(bevy_app::stage::PRE_UPDATE, draw::clear_draw_system)
-            .add_system_to_stage(bevy_app::stage::POST_UPDATE, camera::active_cameras_system)
-            .add_system_to_stage(
-                bevy_app::stage::POST_UPDATE,
-                camera::camera_system::<OrthographicProjection>,
-            )
-            .add_system_to_stage(
-                bevy_app::stage::POST_UPDATE,
-                camera::camera_system::<PerspectiveProjection>,
-            )
-            // registration order matters here. this must come after all camera_system::<T> systems
-            .add_system_to_stage(
-                bevy_app::stage::POST_UPDATE,
-                camera::visible_entities_system,
-            )
-            // TODO: turn these "resource systems" into graph nodes and remove the RENDER_RESOURCE stage
-            .add_system_to_stage(stage::RENDER_RESOURCE, shader::shader_update_system)
-            .add_system_to_stage(stage::RENDER_RESOURCE, mesh::mesh_resource_provider_system)
-            .add_system_to_stage(stage::RENDER_RESOURCE, Texture::texture_resource_system)
-            .add_system_to_stage(
-                stage::RENDER_GRAPH_SYSTEMS,
-                render_graph::render_graph_schedule_executor_system,
-            )
-            .add_system_to_stage(stage::DRAW, pipeline::draw_render_pipelines_system)
-            .add_system_to_stage(stage::POST_RENDER, shader::clear_shader_defs_system);
+        app.add_stage_after(
+            bevy_asset::stage::ASSET_EVENTS,
+            stage::RENDER_RESOURCE,
+            SystemStage::parallel(),
+        )
+        .add_stage_after(
+            stage::RENDER_RESOURCE,
+            stage::RENDER_GRAPH_SYSTEMS,
+            SystemStage::parallel(),
+        )
+        .add_stage_after(
+            stage::RENDER_GRAPH_SYSTEMS,
+            stage::DRAW,
+            SystemStage::parallel(),
+        )
+        .add_stage_after(stage::DRAW, stage::RENDER, SystemStage::parallel())
+        .add_stage_after(stage::RENDER, stage::POST_RENDER, SystemStage::parallel())
+        .add_asset::<Mesh>()
+        .add_asset::<Texture>()
+        .add_asset::<Shader>()
+        .add_asset::<PipelineDescriptor>()
+        .register_type::<Camera>()
+        .register_type::<Draw>()
+        .register_type::<Visible>()
+        .register_type::<RenderPipelines>()
+        .register_type::<OrthographicProjection>()
+        .register_type::<PerspectiveProjection>()
+        .register_type::<MainPass>()
+        .register_type::<VisibleEntities>()
+        .register_type::<Color>()
+        .register_type::<ShaderSpecialization>()
+        .register_type::<PrimitiveTopology>()
+        .register_type::<IndexFormat>()
+        .register_type::<PipelineSpecialization>()
+        .init_resource::<RenderGraph>()
+        .init_resource::<PipelineCompiler>()
+        .init_resource::<RenderResourceBindings>()
+        .init_resource::<TextureResourceSystemState>()
+        .init_resource::<AssetRenderResourceBindings>()
+        .init_resource::<ActiveCameras>()
+        .add_system_to_stage(bevy_app::stage::PRE_UPDATE, draw::clear_draw_system)
+        .add_system_to_stage(bevy_app::stage::POST_UPDATE, camera::active_cameras_system)
+        .add_system_to_stage(
+            bevy_app::stage::POST_UPDATE,
+            camera::camera_system::<OrthographicProjection>,
+        )
+        .add_system_to_stage(
+            bevy_app::stage::POST_UPDATE,
+            camera::camera_system::<PerspectiveProjection>,
+        )
+        // registration order matters here. this must come after all camera_system::<T> systems
+        .add_system_to_stage(
+            bevy_app::stage::POST_UPDATE,
+            camera::visible_entities_system,
+        )
+        .add_system_to_stage(stage::RENDER_RESOURCE, shader::shader_update_system)
+        .add_system_to_stage(stage::RENDER_RESOURCE, mesh::mesh_resource_provider_system)
+        .add_system_to_stage(stage::RENDER_RESOURCE, Texture::texture_resource_system)
+        .add_system_to_stage(
+            stage::RENDER_GRAPH_SYSTEMS,
+            render_graph::render_graph_schedule_executor_system,
+        )
+        .add_system_to_stage(stage::DRAW, pipeline::draw_render_pipelines_system)
+        .add_system_to_stage(stage::POST_RENDER, shader::clear_shader_defs_system);
 
         if app.resources().get::<Msaa>().is_none() {
             app.init_resource::<Msaa>();
diff --git a/crates/bevy_render/src/render_graph/graph.rs b/crates/bevy_render/src/render_graph/graph.rs
index 65c3319b50a46..22cbb72893dca 100644
--- a/crates/bevy_render/src/render_graph/graph.rs
+++ b/crates/bevy_render/src/render_graph/graph.rs
@@ -1,5 +1,5 @@
 use super::{Edge, Node, NodeId, NodeLabel, NodeState, RenderGraphError, SlotLabel, SystemNode};
-use bevy_ecs::{Commands, Schedule};
+use bevy_ecs::{Commands, Schedule, SystemStage};
 use bevy_utils::HashMap;
 use std::{borrow::Cow, fmt::Debug};
 pub struct RenderGraph {
@@ -12,7 +12,7 @@ pub struct RenderGraph {
 impl Default for RenderGraph {
     fn default() -> Self {
         let mut schedule = Schedule::default();
-        schedule.add_stage("update");
+        schedule.add_stage("update", SystemStage::parallel());
         Self {
             nodes: Default::default(),
             node_names: Default::default(),
@@ -40,10 +40,9 @@ impl RenderGraph {
     where
         T: SystemNode + 'static,
     {
-        self.system_node_schedule
-            .as_mut()
-            .unwrap()
-            .add_boxed_system_to_stage("update", node.get_system(&mut self.commands));
+        let schedule = self.system_node_schedule.as_mut().unwrap();
+        let stage = schedule.get_stage_mut::<SystemStage>("update").unwrap();
+        stage.add_system_boxed(node.get_system(&mut self.commands));
         self.add_node(name, node)
     }
 
diff --git a/crates/bevy_render/src/render_graph/node.rs b/crates/bevy_render/src/render_graph/node.rs
index 66eaa8f6db545..1c2a19443314e 100644
--- a/crates/bevy_render/src/render_graph/node.rs
+++ b/crates/bevy_render/src/render_graph/node.rs
@@ -37,7 +37,7 @@ pub trait Node: Downcast + Send + Sync + 'static {
 impl_downcast!(Node);
 
 pub trait SystemNode: Node {
-    fn get_system(&self, commands: &mut Commands) -> Box<dyn System<Input = (), Output = ()>>;
+    fn get_system(&self, commands: &mut Commands) -> Box<dyn System<In = (), Out = ()>>;
 }
 
 #[derive(Debug)]
diff --git a/crates/bevy_render/src/render_graph/nodes/camera_node.rs b/crates/bevy_render/src/render_graph/nodes/camera_node.rs
index 574535f4f28c4..5e677c75b3019 100644
--- a/crates/bevy_render/src/render_graph/nodes/camera_node.rs
+++ b/crates/bevy_render/src/render_graph/nodes/camera_node.rs
@@ -44,7 +44,7 @@ impl Node for CameraNode {
 }
 
 impl SystemNode for CameraNode {
-    fn get_system(&self, commands: &mut Commands) -> Box<dyn System<Input = (), Output = ()>> {
+    fn get_system(&self, commands: &mut Commands) -> Box<dyn System<In = (), Out = ()>> {
         let system = camera_node_system.system();
         commands.insert_local_resource(
             system.id(),
diff --git a/crates/bevy_render/src/render_graph/nodes/render_resources_node.rs b/crates/bevy_render/src/render_graph/nodes/render_resources_node.rs
index df1a592f3c9e4..e513d1434cccc 100644
--- a/crates/bevy_render/src/render_graph/nodes/render_resources_node.rs
+++ b/crates/bevy_render/src/render_graph/nodes/render_resources_node.rs
@@ -401,7 +401,7 @@ impl<T> SystemNode for RenderResourcesNode<T>
 where
     T: renderer::RenderResources,
 {
-    fn get_system(&self, commands: &mut Commands) -> Box<dyn System<Input = (), Output = ()>> {
+    fn get_system(&self, commands: &mut Commands) -> Box<dyn System<In = (), Out = ()>> {
         let system = render_resources_node_system::<T>.system();
         commands.insert_local_resource(
             system.id(),
@@ -473,7 +473,6 @@ fn render_resources_node_system<T: RenderResources>(
         if !visible.is_visible {
             continue;
         }
-
         uniform_buffer_arrays.prepare_uniform_buffers(entity, uniforms);
         if !setup_uniform_texture_resources::<T>(
             &uniforms,
@@ -585,7 +584,7 @@ impl<T> SystemNode for AssetRenderResourcesNode<T>
 where
     T: renderer::RenderResources + Asset,
 {
-    fn get_system(&self, commands: &mut Commands) -> Box<dyn System<Input = (), Output = ()>> {
+    fn get_system(&self, commands: &mut Commands) -> Box<dyn System<In = (), Out = ()>> {
         let system = asset_render_resources_node_system::<T>.system();
         commands.insert_local_resource(
             system.id(),
diff --git a/crates/bevy_render/src/render_graph/system.rs b/crates/bevy_render/src/render_graph/system.rs
index 875c33298c653..78fa184c2469e 100644
--- a/crates/bevy_render/src/render_graph/system.rs
+++ b/crates/bevy_render/src/render_graph/system.rs
@@ -10,12 +10,7 @@ pub fn render_graph_schedule_executor_system(world: &mut World, resources: &mut
 
     commands.apply(world, resources);
     if let Some(schedule) = system_schedule.as_mut() {
-        schedule.run_on_systems(|system| {
-            if !system.is_initialized() {
-                system.initialize(world, resources);
-            }
-        });
-        schedule.run(world, resources);
+        schedule.initialize_and_run(world, resources);
     }
     let mut render_graph = resources.get_mut::<RenderGraph>().unwrap();
     if let Some(schedule) = system_schedule.take() {
diff --git a/crates/bevy_scene/src/lib.rs b/crates/bevy_scene/src/lib.rs
index 3d0d8d86893f2..8bf00bda2831b 100644
--- a/crates/bevy_scene/src/lib.rs
+++ b/crates/bevy_scene/src/lib.rs
@@ -5,6 +5,7 @@ mod scene_loader;
 mod scene_spawner;
 pub mod serde;
 
+use bevy_ecs::SystemStage;
 pub use command::*;
 pub use dynamic_scene::*;
 pub use scene::*;
@@ -31,7 +32,7 @@ impl Plugin for ScenePlugin {
             .add_asset::<Scene>()
             .init_asset_loader::<SceneLoader>()
             .init_resource::<SceneSpawner>()
-            .add_stage_after(stage::EVENT, SCENE_STAGE)
+            .add_stage_after(stage::EVENT, SCENE_STAGE, SystemStage::parallel())
             .add_system_to_stage(SCENE_STAGE, scene_spawner_system);
     }
 }
diff --git a/crates/bevy_transform/src/hierarchy/hierarchy_maintenance_system.rs b/crates/bevy_transform/src/hierarchy/hierarchy_maintenance_system.rs
index 5eea65682e2a3..7cff4d4de7145 100644
--- a/crates/bevy_transform/src/hierarchy/hierarchy_maintenance_system.rs
+++ b/crates/bevy_transform/src/hierarchy/hierarchy_maintenance_system.rs
@@ -71,7 +71,7 @@ pub fn parent_update_system(
 mod test {
     use super::*;
     use crate::{hierarchy::BuildChildren, transform_propagate_system::transform_propagate_system};
-    use bevy_ecs::{Resources, Schedule, World};
+    use bevy_ecs::{Resources, Schedule, SystemStage, World};
     use bevy_math::Vec3;
 
     #[test]
@@ -79,10 +79,12 @@ mod test {
         let mut world = World::default();
         let mut resources = Resources::default();
 
+        let mut update_stage = SystemStage::parallel();
+        update_stage.add_system(parent_update_system);
+        update_stage.add_system(transform_propagate_system);
+
         let mut schedule = Schedule::default();
-        schedule.add_stage("update");
-        schedule.add_system_to_stage("update", parent_update_system);
-        schedule.add_system_to_stage("update", transform_propagate_system);
+        schedule.add_stage("update", update_stage);
 
         // Add parent entities
         let mut commands = Commands::default();
@@ -101,8 +103,7 @@ mod test {
             });
         let parent = parent.unwrap();
         commands.apply(&mut world, &mut resources);
-        schedule.initialize(&mut world, &mut resources);
-        schedule.run(&mut world, &mut resources);
+        schedule.initialize_and_run(&mut world, &mut resources);
 
         assert_eq!(
             world
@@ -118,7 +119,7 @@ mod test {
         // Parent `e1` to `e2`.
         (*world.get_mut::<Parent>(children[0]).unwrap()).0 = children[1];
 
-        schedule.run(&mut world, &mut resources);
+        schedule.initialize_and_run(&mut world, &mut resources);
 
         assert_eq!(
             world
@@ -142,7 +143,7 @@ mod test {
 
         world.despawn(children[0]).unwrap();
 
-        schedule.run(&mut world, &mut resources);
+        schedule.initialize_and_run(&mut world, &mut resources);
 
         assert_eq!(
             world
diff --git a/crates/bevy_transform/src/transform_propagate_system.rs b/crates/bevy_transform/src/transform_propagate_system.rs
index 27dc9ef2be7ac..aa7c98262a8a3 100644
--- a/crates/bevy_transform/src/transform_propagate_system.rs
+++ b/crates/bevy_transform/src/transform_propagate_system.rs
@@ -71,7 +71,7 @@ fn propagate_recursive(
 mod test {
     use super::*;
     use crate::hierarchy::{parent_update_system, BuildChildren, BuildWorldChildren};
-    use bevy_ecs::{Resources, Schedule, World};
+    use bevy_ecs::{Resources, Schedule, SystemStage, World};
     use bevy_math::Vec3;
 
     #[test]
@@ -79,10 +79,12 @@ mod test {
         let mut world = World::default();
         let mut resources = Resources::default();
 
+        let mut update_stage = SystemStage::parallel();
+        update_stage.add_system(parent_update_system);
+        update_stage.add_system(transform_propagate_system);
+
         let mut schedule = Schedule::default();
-        schedule.add_stage("update");
-        schedule.add_system_to_stage("update", parent_update_system);
-        schedule.add_system_to_stage("update", transform_propagate_system);
+        schedule.add_stage("update", update_stage);
 
         // Root entity
         world.spawn((
@@ -110,11 +112,7 @@ mod test {
                     ))
                     .for_current_entity(|entity| children.push(entity));
             });
-        // we need to run the schedule two times because components need to be filled in
-        // to resolve this problem in code, just add the correct components, or use Commands
-        // which adds all of the components needed with the correct state (see next test)
-        schedule.initialize(&mut world, &mut resources);
-        schedule.run(&mut world, &mut resources);
+        schedule.initialize_and_run(&mut world, &mut resources);
 
         assert_eq!(
             *world.get::<GlobalTransform>(children[0]).unwrap(),
@@ -134,10 +132,12 @@ mod test {
         let mut world = World::default();
         let mut resources = Resources::default();
 
+        let mut update_stage = SystemStage::parallel();
+        update_stage.add_system(parent_update_system);
+        update_stage.add_system(transform_propagate_system);
+
         let mut schedule = Schedule::default();
-        schedule.add_stage("update");
-        schedule.add_system_to_stage("update", parent_update_system);
-        schedule.add_system_to_stage("update", transform_propagate_system);
+        schedule.add_stage("update", update_stage);
 
         // Root entity
         let mut commands = Commands::default();
@@ -162,8 +162,7 @@ mod test {
                     .for_current_entity(|entity| children.push(entity));
             });
         commands.apply(&mut world, &mut resources);
-        schedule.initialize(&mut world, &mut resources);
-        schedule.run(&mut world, &mut resources);
+        schedule.initialize_and_run(&mut world, &mut resources);
 
         assert_eq!(
             *world.get::<GlobalTransform>(children[0]).unwrap(),
diff --git a/crates/bevy_ui/src/lib.rs b/crates/bevy_ui/src/lib.rs
index 9152ac306e91a..46ee8b661fa7b 100644
--- a/crates/bevy_ui/src/lib.rs
+++ b/crates/bevy_ui/src/lib.rs
@@ -25,6 +25,7 @@ pub mod prelude {
 }
 
 use bevy_app::prelude::*;
+use bevy_ecs::SystemStage;
 use bevy_render::render_graph::RenderGraph;
 use update::ui_z_system;
 
@@ -38,7 +39,11 @@ pub mod stage {
 impl Plugin for UiPlugin {
     fn build(&self, app: &mut AppBuilder) {
         app.init_resource::<FlexSurface>()
-            .add_stage_before(bevy_app::stage::POST_UPDATE, stage::UI)
+            .add_stage_before(
+                bevy_app::stage::POST_UPDATE,
+                stage::UI,
+                SystemStage::parallel(),
+            )
             .add_system_to_stage(bevy_app::stage::PRE_UPDATE, ui_focus_system)
             // add these stages to front because these must run before transform update systems
             .add_system_to_stage(stage::UI, widget::text_system)
diff --git a/crates/bevy_ui/src/update.rs b/crates/bevy_ui/src/update.rs
index 498a487596705..fe6bc25237a64 100644
--- a/crates/bevy_ui/src/update.rs
+++ b/crates/bevy_ui/src/update.rs
@@ -48,7 +48,7 @@ fn update_hierarchy(
 }
 #[cfg(test)]
 mod tests {
-    use bevy_ecs::{Commands, IntoSystem, Resources, Schedule, World};
+    use bevy_ecs::{Commands, Resources, Schedule, SystemStage, World};
     use bevy_transform::{components::Transform, hierarchy::BuildChildren};
 
     use crate::Node;
@@ -115,10 +115,10 @@ mod tests {
         commands.apply(&mut world, &mut resources);
 
         let mut schedule = Schedule::default();
-        schedule.add_stage("update");
-        schedule.add_system_to_stage("update", ui_z_system.system());
-        schedule.initialize(&mut world, &mut resources);
-        schedule.run(&mut world, &mut resources);
+        let mut update_stage = SystemStage::parallel();
+        update_stage.add_system(ui_z_system);
+        schedule.add_stage("update", update_stage);
+        schedule.initialize_and_run(&mut world, &mut resources);
 
         let mut actual_result = world
             .query::<(&String, &Transform)>()
diff --git a/crates/bevy_winit/src/lib.rs b/crates/bevy_winit/src/lib.rs
index 3393183993c23..47e4b484ad51d 100644
--- a/crates/bevy_winit/src/lib.rs
+++ b/crates/bevy_winit/src/lib.rs
@@ -165,16 +165,6 @@ pub fn winit_runner(mut app: App) {
 
     app.resources.insert_thread_local(event_loop.create_proxy());
 
-    // Create Windows and WinitWindows resources, so startup systems
-    // in below app.initialize() have access to them.
-    handle_create_window_events(
-        &mut app.resources,
-        &event_loop,
-        &mut create_window_event_reader,
-    );
-
-    app.initialize();
-
     trace!("Entering winit event loop");
 
     let should_return_from_run = app
diff --git a/examples/2d/texture_atlas.rs b/examples/2d/texture_atlas.rs
index fbe6db078d062..08823067d4fa9 100644
--- a/examples/2d/texture_atlas.rs
+++ b/examples/2d/texture_atlas.rs
@@ -5,70 +5,78 @@ fn main() {
     App::build()
         .init_resource::<RpgSpriteHandles>()
         .add_plugins(DefaultPlugins)
-        .add_startup_system(setup)
-        .add_system(load_atlas)
+        .add_state(AppState::Setup)
+        .on_state_enter(AppState::Setup, load_textures)
+        .on_state_update(AppState::Setup, check_textures)
+        .on_state_enter(AppState::Finshed, setup)
         .run();
 }
 
+#[derive(Clone)]
+enum AppState {
+    Setup,
+    Finshed,
+}
+
 #[derive(Default)]
-pub struct RpgSpriteHandles {
+struct RpgSpriteHandles {
     handles: Vec<HandleUntyped>,
-    atlas_loaded: bool,
 }
 
-fn setup(mut rpg_sprite_handles: ResMut<RpgSpriteHandles>, asset_server: Res<AssetServer>) {
+fn load_textures(mut rpg_sprite_handles: ResMut<RpgSpriteHandles>, asset_server: Res<AssetServer>) {
     rpg_sprite_handles.handles = asset_server.load_folder("textures/rpg").unwrap();
 }
 
-fn load_atlas(
+fn check_textures(
+    mut state: ResMut<State<AppState>>,
+    rpg_sprite_handles: ResMut<RpgSpriteHandles>,
+    asset_server: Res<AssetServer>,
+) {
+    if let LoadState::Loaded =
+        asset_server.get_group_load_state(rpg_sprite_handles.handles.iter().map(|handle| handle.id))
+    {
+        state.set_next(AppState::Finshed).unwrap();
+    }
+}
+
+fn setup(
     commands: &mut Commands,
-    mut rpg_sprite_handles: ResMut<RpgSpriteHandles>,
+    rpg_sprite_handles: Res<RpgSpriteHandles>,
     asset_server: Res<AssetServer>,
     mut texture_atlases: ResMut<Assets<TextureAtlas>>,
     mut textures: ResMut<Assets<Texture>>,
     mut materials: ResMut<Assets<ColorMaterial>>,
 ) {
-    if rpg_sprite_handles.atlas_loaded {
-        return;
-    }
-
     let mut texture_atlas_builder = TextureAtlasBuilder::default();
-    if let LoadState::Loaded =
-        asset_server.get_group_load_state(rpg_sprite_handles.handles.iter().map(|handle| handle.id))
-    {
-        for handle in rpg_sprite_handles.handles.iter() {
-            let texture = textures.get(handle).unwrap();
-            texture_atlas_builder.add_texture(handle.clone_weak().typed::<Texture>(), texture);
-        }
+    for handle in rpg_sprite_handles.handles.iter() {
+        let texture = textures.get(handle).unwrap();
+        texture_atlas_builder.add_texture(handle.clone_weak().typed::<Texture>(), texture);
+    }
 
-        let texture_atlas = texture_atlas_builder.finish(&mut textures).unwrap();
-        let texture_atlas_texture = texture_atlas.texture.clone();
-        let vendor_handle =
-            asset_server.get_handle("textures/rpg/chars/vendor/generic-rpg-vendor.png");
-        let vendor_index = texture_atlas.get_texture_index(&vendor_handle).unwrap();
-        let atlas_handle = texture_atlases.add(texture_atlas);
+    let texture_atlas = texture_atlas_builder.finish(&mut textures).unwrap();
+    let texture_atlas_texture = texture_atlas.texture.clone();
+    let vendor_handle = asset_server.get_handle("textures/rpg/chars/vendor/generic-rpg-vendor.png");
+    let vendor_index = texture_atlas.get_texture_index(&vendor_handle).unwrap();
+    let atlas_handle = texture_atlases.add(texture_atlas);
 
-        // set up a scene to display our texture atlas
-        commands
-            .spawn(Camera2dBundle::default())
-            // draw a sprite from the atlas
-            .spawn(SpriteSheetBundle {
-                transform: Transform {
-                    translation: Vec3::new(150.0, 0.0, 0.0),
-                    scale: Vec3::splat(4.0),
-                    ..Default::default()
-                },
-                sprite: TextureAtlasSprite::new(vendor_index as u32),
-                texture_atlas: atlas_handle,
-                ..Default::default()
-            })
-            // draw the atlas itself
-            .spawn(SpriteBundle {
-                material: materials.add(texture_atlas_texture.into()),
-                transform: Transform::from_translation(Vec3::new(-300.0, 0.0, 0.0)),
+    // set up a scene to display our texture atlas
+    commands
+        .spawn(Camera2dBundle::default())
+        // draw a sprite from the atlas
+        .spawn(SpriteSheetBundle {
+            transform: Transform {
+                translation: Vec3::new(150.0, 0.0, 0.0),
+                scale: Vec3::splat(4.0),
                 ..Default::default()
-            });
-
-        rpg_sprite_handles.atlas_loaded = true;
-    }
+            },
+            sprite: TextureAtlasSprite::new(vendor_index as u32),
+            texture_atlas: atlas_handle,
+            ..Default::default()
+        })
+        // draw the atlas itself
+        .spawn(SpriteBundle {
+            material: materials.add(texture_atlas_texture.into()),
+            transform: Transform::from_translation(Vec3::new(-300.0, 0.0, 0.0)),
+            ..Default::default()
+        });
 }
diff --git a/examples/app/custom_loop.rs b/examples/app/custom_loop.rs
index 647f23eaba89b..48337c8a17e82 100644
--- a/examples/app/custom_loop.rs
+++ b/examples/app/custom_loop.rs
@@ -6,8 +6,6 @@ struct Input(String);
 /// This example demonstrates you can create a custom runner (to update an app manually). It reads
 /// lines from stdin and prints them from within the ecs.
 fn my_runner(mut app: App) {
-    app.initialize();
-
     println!("Type stuff into the console");
     for line in io::stdin().lock().lines() {
         {
diff --git a/examples/ecs/ecs_guide.rs b/examples/ecs/ecs_guide.rs
index 7e67abab5b962..cd7c7ffb763ea 100644
--- a/examples/ecs/ecs_guide.rs
+++ b/examples/ecs/ecs_guide.rs
@@ -1,5 +1,6 @@
 use bevy::{
     app::{AppExit, ScheduleRunnerPlugin, ScheduleRunnerSettings},
+    ecs::SystemStage,
     prelude::*,
     utils::Duration,
 };
@@ -282,8 +283,8 @@ fn main() {
         // "before_round": new_player_system, new_round_system
         // "update": print_message_system, score_system
         // "after_round": score_check_system, game_over_system
-        .add_stage_before(stage::UPDATE, "before_round")
-        .add_stage_after(stage::UPDATE, "after_round")
+        .add_stage_before(stage::UPDATE, "before_round", SystemStage::parallel())
+        .add_stage_after(stage::UPDATE, "after_round", SystemStage::parallel())
         .add_system_to_stage("before_round", new_round_system)
         .add_system_to_stage("before_round", new_player_system)
         .add_system_to_stage("after_round", score_check_system)
diff --git a/examples/ecs/fixed_timestep.rs b/examples/ecs/fixed_timestep.rs
new file mode 100644
index 0000000000000..88136badd5494
--- /dev/null
+++ b/examples/ecs/fixed_timestep.rs
@@ -0,0 +1,46 @@
+use bevy::{
+    core::{FixedTimestep, FixedTimesteps},
+    prelude::*,
+};
+
+const LABEL: &str = "my_fixed_timestep";
+
+fn main() {
+    App::build()
+        .add_plugins(DefaultPlugins)
+        // this system will run once every update (it should match your screen's refresh rate)
+        .add_system(update)
+        // add a new stage that runs every two seconds
+        .add_stage_after(
+            stage::UPDATE,
+            "fixed_update",
+            SystemStage::parallel()
+                .with_run_criteria(
+                    FixedTimestep::step(2.0)
+                        // labels are optional. they provide a way to access the current FixedTimestep state from within a system
+                        .with_label(LABEL),
+                )
+                .with_system(fixed_update),
+        )
+        .run();
+}
+
+fn update(mut last_time: Local<f64>, time: Res<Time>) {
+    println!("update: {}", time.seconds_since_startup() - *last_time);
+    *last_time = time.seconds_since_startup();
+}
+
+fn fixed_update(mut last_time: Local<f64>, time: Res<Time>, fixed_timesteps: Res<FixedTimesteps>) {
+    println!(
+        "fixed_update: {}",
+        time.seconds_since_startup() - *last_time,
+    );
+
+    let fixed_timestep = fixed_timesteps.get(LABEL).unwrap();
+    println!(
+        "  overstep_percentage: {}",
+        fixed_timestep.overstep_percentage()
+    );
+
+    *last_time = time.seconds_since_startup();
+}
diff --git a/examples/ecs/state.rs b/examples/ecs/state.rs
new file mode 100644
index 0000000000000..07ae083bb12de
--- /dev/null
+++ b/examples/ecs/state.rs
@@ -0,0 +1,170 @@
+use bevy::prelude::*;
+
+/// This example illustrates how to use States to control transitioning from a Menu state to an InGame state.
+fn main() {
+    App::build()
+        .add_plugins(DefaultPlugins)
+        .init_resource::<ButtonMaterials>()
+        .add_state(AppState::Menu)
+        .on_state_enter(AppState::Menu, setup_menu)
+        .on_state_update(AppState::Menu, menu)
+        .on_state_exit(AppState::Menu, cleanup_menu)
+        .on_state_enter(AppState::InGame, setup_game)
+        .on_state_update(
+            AppState::InGame,
+            SystemStage::parallel()
+                .with_system(movement)
+                .with_system(change_color),
+        )
+        .run();
+}
+
+#[derive(Clone)]
+enum AppState {
+    Menu,
+    InGame,
+}
+
+struct MenuData {
+    button_entity: Entity,
+}
+
+fn setup_menu(
+    commands: &mut Commands,
+    asset_server: Res<AssetServer>,
+    button_materials: Res<ButtonMaterials>,
+) {
+    commands
+        // ui camera
+        .spawn(CameraUiBundle::default())
+        .spawn(ButtonBundle {
+            style: Style {
+                size: Size::new(Val::Px(150.0), Val::Px(65.0)),
+                // center button
+                margin: Rect::all(Val::Auto),
+                // horizontally center child text
+                justify_content: JustifyContent::Center,
+                // vertically center child text
+                align_items: AlignItems::Center,
+                ..Default::default()
+            },
+            material: button_materials.normal.clone(),
+            ..Default::default()
+        })
+        .with_children(|parent| {
+            parent.spawn(TextBundle {
+                text: Text {
+                    value: "Play".to_string(),
+                    font: asset_server.load("fonts/FiraSans-Bold.ttf"),
+                    style: TextStyle {
+                        font_size: 40.0,
+                        color: Color::rgb(0.9, 0.9, 0.9),
+                        ..Default::default()
+                    },
+                },
+                ..Default::default()
+            });
+        });
+    commands.insert_resource(MenuData {
+        button_entity: commands.current_entity().unwrap(),
+    });
+}
+
+fn menu(
+    mut state: ResMut<State<AppState>>,
+    button_materials: Res<ButtonMaterials>,
+    mut interaction_query: Query<
+        (&Interaction, &mut Handle<ColorMaterial>),
+        (Mutated<Interaction>, With<Button>),
+    >,
+) {
+    for (interaction, mut material) in interaction_query.iter_mut() {
+        match *interaction {
+            Interaction::Clicked => {
+                *material = button_materials.pressed.clone();
+                state.set_next(AppState::InGame).unwrap();
+            }
+            Interaction::Hovered => {
+                *material = button_materials.hovered.clone();
+            }
+            Interaction::None => {
+                *material = button_materials.normal.clone();
+            }
+        }
+    }
+}
+
+fn cleanup_menu(commands: &mut Commands, menu_data: Res<MenuData>) {
+    commands.despawn_recursive(menu_data.button_entity);
+}
+
+fn setup_game(
+    commands: &mut Commands,
+    asset_server: Res<AssetServer>,
+    mut materials: ResMut<Assets<ColorMaterial>>,
+) {
+    let texture_handle = asset_server.load("branding/icon.png");
+    commands
+        .spawn(Camera2dBundle::default())
+        .spawn(SpriteBundle {
+            material: materials.add(texture_handle.into()),
+            ..Default::default()
+        });
+}
+
+const SPEED: f32 = 100.0;
+fn movement(
+    time: Res<Time>,
+    input: Res<Input<KeyCode>>,
+    mut query: Query<&mut Transform, With<Sprite>>,
+) {
+    for mut transform in query.iter_mut() {
+        let mut direction = Vec3::default();
+        if input.pressed(KeyCode::Left) {
+            direction.x += 1.0;
+        }
+        if input.pressed(KeyCode::Right) {
+            direction.x -= 1.0;
+        }
+        if input.pressed(KeyCode::Up) {
+            direction.y += 1.0;
+        }
+        if input.pressed(KeyCode::Down) {
+            direction.y -= 1.0;
+        }
+
+        if direction != Vec3::default() {
+            transform.translation += direction.normalize() * SPEED * time.delta_seconds();
+        }
+    }
+}
+
+fn change_color(
+    time: Res<Time>,
+    mut assets: ResMut<Assets<ColorMaterial>>,
+    query: Query<&Handle<ColorMaterial>, With<Sprite>>,
+) {
+    for handle in query.iter() {
+        let material = assets.get_mut(handle).unwrap();
+        material
+            .color
+            .set_b((time.seconds_since_startup() * 5.0).sin() as f32 + 2.0);
+    }
+}
+
+struct ButtonMaterials {
+    normal: Handle<ColorMaterial>,
+    hovered: Handle<ColorMaterial>,
+    pressed: Handle<ColorMaterial>,
+}
+
+impl FromResources for ButtonMaterials {
+    fn from_resources(resources: &Resources) -> Self {
+        let mut materials = resources.get_mut::<Assets<ColorMaterial>>().unwrap();
+        ButtonMaterials {
+            normal: materials.add(Color::rgb(0.15, 0.15, 0.15).into()),
+            hovered: materials.add(Color::rgb(0.25, 0.25, 0.25).into()),
+            pressed: materials.add(Color::rgb(0.35, 0.75, 0.35).into()),
+        }
+    }
+}