From d3e020a1e7ed035b59b2f91833d86e252ab22ada Mon Sep 17 00:00:00 2001
From: Alexander Sepity <alexander.sepity@gmail.com>
Date: Wed, 24 Mar 2021 20:11:55 +0000
Subject: [PATCH] System sets and run criteria v2 (#1675)

I'm opening this prematurely; consider this an RFC that predates RFCs and therefore not super-RFC-like.

This PR does two "big" things: decouple run criteria from system sets, reimagine system sets as weapons of mass system description.

### What it lets us do:

* Reuse run criteria within a stage.
* Pipe output of one run criteria as input to another.
* Assign labels, dependencies, run criteria, and ambiguity sets to many systems at the same time.

### Things already done:
* Decoupled run criteria from system sets.
* Mass system description superpowers to `SystemSet`.
* Implemented `RunCriteriaDescriptor`.
* Removed `VirtualSystemSet`.
* Centralized all run criteria of `SystemStage`.
* Extended system descriptors with per-system run criteria.
* `.before()` and `.after()` for run criteria.
* Explicit order between state driver and related run criteria. Fixes #1672.
* Opt-in run criteria deduplication; default behavior is to panic.
* Labels (not exposed) for state run criteria; state run criteria are deduplicated.

### API issues that need discussion:

* [`FixedTimestep::step(1.0).label("my label")`](https://github.com/Ratysz/bevy/blob/eaccf857cdaeb5a5632b6e75feab5c1ad6267d1d/crates/bevy_ecs/src/schedule/run_criteria.rs#L120-L122) and [`FixedTimestep::step(1.0).with_label("my label")`](https://github.com/Ratysz/bevy/blob/eaccf857cdaeb5a5632b6e75feab5c1ad6267d1d/crates/bevy_core/src/time/fixed_timestep.rs#L86-L89) are both valid but do very different things.

---

I will try to maintain this post up-to-date as things change. Do check the diffs in "edited" thingy from time to time.

Co-authored-by: Carter Anderson <mcanders1@gmail.com>
---
 crates/bevy_app/src/app_builder.rs            |  15 +-
 crates/bevy_ecs/macros/src/lib.rs             |   6 +
 crates/bevy_ecs/src/lib.rs                    |   1 +
 crates/bevy_ecs/src/schedule/graph_utils.rs   | 125 +++
 crates/bevy_ecs/src/schedule/label.rs         |   9 +-
 crates/bevy_ecs/src/schedule/mod.rs           | 119 +--
 crates/bevy_ecs/src/schedule/run_criteria.rs  | 397 +++++++++
 crates/bevy_ecs/src/schedule/stage.rs         | 832 ++++++++++++------
 crates/bevy_ecs/src/schedule/state.rs         | 111 ++-
 .../bevy_ecs/src/schedule/system_container.rs |  98 ++-
 .../src/schedule/system_descriptor.rs         |  66 +-
 crates/bevy_ecs/src/schedule/system_set.rs    | 134 ++-
 examples/2d/texture_atlas.rs                  |   2 +-
 examples/ecs/state.rs                         |   2 +-
 examples/game/alien_cake_addict.rs            |   2 +-
 examples/window/multiple_windows.rs           |   2 +-
 16 files changed, 1429 insertions(+), 492 deletions(-)
 create mode 100644 crates/bevy_ecs/src/schedule/graph_utils.rs
 create mode 100644 crates/bevy_ecs/src/schedule/run_criteria.rs

diff --git a/crates/bevy_app/src/app_builder.rs b/crates/bevy_app/src/app_builder.rs
index fe7b5fb3d70fd..18f12f036596a 100644
--- a/crates/bevy_app/src/app_builder.rs
+++ b/crates/bevy_app/src/app_builder.rs
@@ -13,6 +13,7 @@ use bevy_ecs::{
     world::{FromWorld, World},
 };
 use bevy_utils::tracing::debug;
+use std::{fmt::Debug, hash::Hash};
 
 /// Configure [App]s using the builder pattern
 pub struct AppBuilder {
@@ -177,16 +178,18 @@ impl AppBuilder {
         self
     }
 
-    pub fn add_state<T: Component + Clone + Eq>(&mut self, initial: T) -> &mut Self {
+    pub fn add_state<T>(&mut self, initial: T) -> &mut Self
+    where
+        T: Component + Debug + Clone + Eq + Hash,
+    {
         self.insert_resource(State::new(initial))
             .add_system_set(State::<T>::make_driver())
     }
 
-    pub fn add_state_to_stage<T: Component + Clone + Eq>(
-        &mut self,
-        stage: impl StageLabel,
-        initial: T,
-    ) -> &mut Self {
+    pub fn add_state_to_stage<T>(&mut self, stage: impl StageLabel, initial: T) -> &mut Self
+    where
+        T: Component + Debug + Clone + Eq + Hash,
+    {
         self.insert_resource(State::new(initial))
             .add_system_set_to_stage(stage, State::<T>::make_driver())
     }
diff --git a/crates/bevy_ecs/macros/src/lib.rs b/crates/bevy_ecs/macros/src/lib.rs
index b35d8c054711c..07108b59de432 100644
--- a/crates/bevy_ecs/macros/src/lib.rs
+++ b/crates/bevy_ecs/macros/src/lib.rs
@@ -433,6 +433,12 @@ pub fn derive_ambiguity_set_label(input: TokenStream) -> TokenStream {
     derive_label(input, Ident::new("AmbiguitySetLabel", Span::call_site())).into()
 }
 
+#[proc_macro_derive(RunCriteriaLabel)]
+pub fn derive_run_criteria_label(input: TokenStream) -> TokenStream {
+    let input = parse_macro_input!(input as DeriveInput);
+    derive_label(input, Ident::new("RunCriteriaLabel", Span::call_site())).into()
+}
+
 fn derive_label(input: DeriveInput, label_type: Ident) -> TokenStream2 {
     let ident = input.ident;
     let ecs_path: Path = bevy_ecs_path();
diff --git a/crates/bevy_ecs/src/lib.rs b/crates/bevy_ecs/src/lib.rs
index aa1be8d483bb5..c3061d1580e4a 100644
--- a/crates/bevy_ecs/src/lib.rs
+++ b/crates/bevy_ecs/src/lib.rs
@@ -19,6 +19,7 @@ pub mod prelude {
         query::{Added, ChangeTrackers, Changed, Or, QueryState, With, WithBundle, Without},
         schedule::{
             AmbiguitySetLabel, ExclusiveSystemDescriptorCoercion, ParallelSystemDescriptorCoercion,
+            RunCriteria, RunCriteriaDescriptorCoercion, RunCriteriaLabel, RunCriteriaPiping,
             Schedule, Stage, StageLabel, State, SystemLabel, SystemSet, SystemStage,
         },
         system::{
diff --git a/crates/bevy_ecs/src/schedule/graph_utils.rs b/crates/bevy_ecs/src/schedule/graph_utils.rs
new file mode 100644
index 0000000000000..ea5e2a2a58362
--- /dev/null
+++ b/crates/bevy_ecs/src/schedule/graph_utils.rs
@@ -0,0 +1,125 @@
+use bevy_utils::{tracing::warn, HashMap, HashSet};
+use fixedbitset::FixedBitSet;
+use std::{borrow::Cow, fmt::Debug, hash::Hash};
+
+pub enum DependencyGraphError<Labels> {
+    GraphCycles(Vec<(usize, Labels)>),
+}
+
+pub trait GraphNode<Label> {
+    fn name(&self) -> Cow<'static, str>;
+    fn labels(&self) -> &[Label];
+    fn before(&self) -> &[Label];
+    fn after(&self) -> &[Label];
+}
+
+/// Constructs a dependency graph of given nodes.
+pub fn build_dependency_graph<Node, Label>(
+    nodes: &[Node],
+) -> HashMap<usize, HashMap<usize, HashSet<Label>>>
+where
+    Node: GraphNode<Label>,
+    Label: Debug + Clone + Eq + Hash,
+{
+    let mut labels = HashMap::<Label, FixedBitSet>::default();
+    for (label, index) in nodes.iter().enumerate().flat_map(|(index, container)| {
+        container
+            .labels()
+            .iter()
+            .cloned()
+            .map(move |label| (label, index))
+    }) {
+        labels
+            .entry(label)
+            .or_insert_with(|| FixedBitSet::with_capacity(nodes.len()))
+            .insert(index);
+    }
+    let mut graph = HashMap::with_capacity_and_hasher(nodes.len(), Default::default());
+    for (index, node) in nodes.iter().enumerate() {
+        let dependencies = graph.entry(index).or_insert_with(HashMap::default);
+        for label in node.after() {
+            match labels.get(label) {
+                Some(new_dependencies) => {
+                    for dependency in new_dependencies.ones() {
+                        dependencies
+                            .entry(dependency)
+                            .or_insert_with(HashSet::default)
+                            .insert(label.clone());
+                    }
+                }
+                None => warn!(
+                    // TODO: plumb this as proper output?
+                    "{} wants to be after unknown label: {:?}",
+                    nodes[index].name(),
+                    label
+                ),
+            }
+        }
+        for label in node.before() {
+            match labels.get(label) {
+                Some(dependants) => {
+                    for dependant in dependants.ones() {
+                        graph
+                            .entry(dependant)
+                            .or_insert_with(HashMap::default)
+                            .entry(index)
+                            .or_insert_with(HashSet::default)
+                            .insert(label.clone());
+                    }
+                }
+                None => warn!(
+                    "{} wants to be before unknown label: {:?}",
+                    nodes[index].name(),
+                    label
+                ),
+            }
+        }
+    }
+    graph
+}
+
+/// Generates a topological order for the given graph.
+pub fn topological_order<Labels: Clone>(
+    graph: &HashMap<usize, HashMap<usize, Labels>>,
+) -> Result<Vec<usize>, DependencyGraphError<Labels>> {
+    fn check_if_cycles_and_visit<L>(
+        node: &usize,
+        graph: &HashMap<usize, HashMap<usize, L>>,
+        sorted: &mut Vec<usize>,
+        unvisited: &mut HashSet<usize>,
+        current: &mut Vec<usize>,
+    ) -> bool {
+        if current.contains(node) {
+            return true;
+        } else if !unvisited.remove(node) {
+            return false;
+        }
+        current.push(*node);
+        for dependency in graph.get(node).unwrap().keys() {
+            if check_if_cycles_and_visit(dependency, &graph, sorted, unvisited, current) {
+                return true;
+            }
+        }
+        sorted.push(*node);
+        current.pop();
+        false
+    }
+    let mut sorted = Vec::with_capacity(graph.len());
+    let mut current = Vec::with_capacity(graph.len());
+    let mut unvisited = HashSet::with_capacity_and_hasher(graph.len(), Default::default());
+    unvisited.extend(graph.keys().cloned());
+    while let Some(node) = unvisited.iter().next().cloned() {
+        if check_if_cycles_and_visit(&node, graph, &mut sorted, &mut unvisited, &mut current) {
+            let mut cycle = Vec::new();
+            let last_window = [*current.last().unwrap(), current[0]];
+            let mut windows = current
+                .windows(2)
+                .chain(std::iter::once(&last_window as &[usize]));
+            while let Some(&[dependant, dependency]) = windows.next() {
+                cycle.push((dependant, graph[&dependant][&dependency].clone()));
+            }
+            return Err(DependencyGraphError::GraphCycles(cycle));
+        }
+    }
+    Ok(sorted)
+}
diff --git a/crates/bevy_ecs/src/schedule/label.rs b/crates/bevy_ecs/src/schedule/label.rs
index 3223f5614772f..a7f3538741242 100644
--- a/crates/bevy_ecs/src/schedule/label.rs
+++ b/crates/bevy_ecs/src/schedule/label.rs
@@ -1,4 +1,4 @@
-pub use bevy_ecs_macros::{AmbiguitySetLabel, StageLabel, SystemLabel};
+pub use bevy_ecs_macros::{AmbiguitySetLabel, RunCriteriaLabel, StageLabel, SystemLabel};
 
 use std::{
     any::Any,
@@ -67,6 +67,12 @@ pub trait AmbiguitySetLabel: DynHash + Debug + Send + Sync + 'static {
 }
 pub(crate) type BoxedAmbiguitySetLabel = Box<dyn AmbiguitySetLabel>;
 
+pub trait RunCriteriaLabel: DynHash + Debug + Send + Sync + 'static {
+    #[doc(hidden)]
+    fn dyn_clone(&self) -> Box<dyn RunCriteriaLabel>;
+}
+pub(crate) type BoxedRunCriteriaLabel = Box<dyn RunCriteriaLabel>;
+
 macro_rules! impl_label {
     ($trait_name:ident) => {
         impl PartialEq for dyn $trait_name {
@@ -106,3 +112,4 @@ macro_rules! impl_label {
 impl_label!(StageLabel);
 impl_label!(SystemLabel);
 impl_label!(AmbiguitySetLabel);
+impl_label!(RunCriteriaLabel);
diff --git a/crates/bevy_ecs/src/schedule/mod.rs b/crates/bevy_ecs/src/schedule/mod.rs
index faee3ccb27045..7f05afd22171d 100644
--- a/crates/bevy_ecs/src/schedule/mod.rs
+++ b/crates/bevy_ecs/src/schedule/mod.rs
@@ -1,6 +1,8 @@
 mod executor;
 mod executor_parallel;
+pub mod graph_utils;
 mod label;
+mod run_criteria;
 mod stage;
 mod state;
 mod system_container;
@@ -9,7 +11,9 @@ mod system_set;
 
 pub use executor::*;
 pub use executor_parallel::*;
+pub use graph_utils::GraphNode;
 pub use label::*;
+pub use run_criteria::*;
 pub use stage::*;
 pub use state::*;
 pub use system_container::*;
@@ -17,20 +21,16 @@ pub use system_descriptor::*;
 pub use system_set::*;
 
 use crate::{
-    archetype::{Archetype, ArchetypeComponentId},
-    component::ComponentId,
-    query::Access,
-    system::{BoxedSystem, IntoSystem, System, SystemId},
+    system::{IntoSystem, System},
     world::World,
 };
 use bevy_utils::HashMap;
-use std::borrow::Cow;
 
 #[derive(Default)]
 pub struct Schedule {
     stages: HashMap<BoxedStageLabel, Box<dyn Stage>>,
     stage_order: Vec<BoxedStageLabel>,
-    run_criteria: RunCriteria,
+    run_criteria: BoxedRunCriteria,
 }
 
 impl Schedule {
@@ -229,110 +229,3 @@ impl Stage for Schedule {
         }
     }
 }
-
-#[derive(Debug, Clone, Copy, PartialEq, Eq)]
-pub enum ShouldRun {
-    /// Yes, the system should run.
-    Yes,
-    /// No, the system should not run.
-    No,
-    /// Yes, the system should run, and afterwards the criteria should be checked again.
-    YesAndCheckAgain,
-    /// No, the system should not run right now, but the criteria should be checked again later.
-    NoAndCheckAgain,
-}
-
-pub(crate) struct RunCriteria {
-    criteria_system: Option<BoxedSystem<(), ShouldRun>>,
-    initialized: bool,
-}
-
-impl Default for RunCriteria {
-    fn default() -> Self {
-        Self {
-            criteria_system: None,
-            initialized: false,
-        }
-    }
-}
-
-impl RunCriteria {
-    pub fn set(&mut self, criteria_system: BoxedSystem<(), ShouldRun>) {
-        self.criteria_system = Some(criteria_system);
-        self.initialized = false;
-    }
-
-    pub fn should_run(&mut self, world: &mut World) -> ShouldRun {
-        if let Some(ref mut run_criteria) = self.criteria_system {
-            if !self.initialized {
-                run_criteria.initialize(world);
-                self.initialized = true;
-            }
-            let should_run = run_criteria.run((), world);
-            run_criteria.apply_buffers(world);
-            should_run
-        } else {
-            ShouldRun::Yes
-        }
-    }
-}
-
-pub struct RunOnce {
-    ran: bool,
-    system_id: SystemId,
-    archetype_component_access: Access<ArchetypeComponentId>,
-    component_access: Access<ComponentId>,
-}
-
-impl Default for RunOnce {
-    fn default() -> Self {
-        Self {
-            ran: false,
-            system_id: SystemId::new(),
-            archetype_component_access: Default::default(),
-            component_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 new_archetype(&mut self, _archetype: &Archetype) {}
-
-    fn archetype_component_access(&self) -> &Access<ArchetypeComponentId> {
-        &self.archetype_component_access
-    }
-
-    fn component_access(&self) -> &Access<ComponentId> {
-        &self.component_access
-    }
-
-    fn is_send(&self) -> bool {
-        true
-    }
-
-    unsafe fn run_unsafe(&mut self, _input: Self::In, _world: &World) -> Self::Out {
-        if self.ran {
-            ShouldRun::No
-        } else {
-            self.ran = true;
-            ShouldRun::Yes
-        }
-    }
-
-    fn apply_buffers(&mut self, _world: &mut World) {}
-
-    fn initialize(&mut self, _world: &mut World) {}
-
-    fn check_change_tick(&mut self, _change_tick: u32) {}
-}
diff --git a/crates/bevy_ecs/src/schedule/run_criteria.rs b/crates/bevy_ecs/src/schedule/run_criteria.rs
new file mode 100644
index 0000000000000..925179fc52c55
--- /dev/null
+++ b/crates/bevy_ecs/src/schedule/run_criteria.rs
@@ -0,0 +1,397 @@
+use crate::{
+    archetype::{Archetype, ArchetypeComponentId},
+    component::ComponentId,
+    query::Access,
+    schedule::{BoxedRunCriteriaLabel, GraphNode, RunCriteriaLabel},
+    system::{BoxedSystem, System, SystemId},
+    world::World,
+};
+use std::borrow::Cow;
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum ShouldRun {
+    /// Yes, the system should run.
+    Yes,
+    /// No, the system should not run.
+    No,
+    /// Yes, the system should run, and afterwards the criteria should be checked again.
+    YesAndCheckAgain,
+    /// No, the system should not run right now, but the criteria should be checked again later.
+    NoAndCheckAgain,
+}
+
+pub(crate) struct BoxedRunCriteria {
+    criteria_system: Option<BoxedSystem<(), ShouldRun>>,
+    initialized: bool,
+}
+
+impl Default for BoxedRunCriteria {
+    fn default() -> Self {
+        Self {
+            criteria_system: None,
+            initialized: false,
+        }
+    }
+}
+
+impl BoxedRunCriteria {
+    pub fn set(&mut self, criteria_system: BoxedSystem<(), ShouldRun>) {
+        self.criteria_system = Some(criteria_system);
+        self.initialized = false;
+    }
+
+    pub fn should_run(&mut self, world: &mut World) -> ShouldRun {
+        if let Some(ref mut run_criteria) = self.criteria_system {
+            if !self.initialized {
+                run_criteria.initialize(world);
+                self.initialized = true;
+            }
+            let should_run = run_criteria.run((), world);
+            run_criteria.apply_buffers(world);
+            should_run
+        } else {
+            ShouldRun::Yes
+        }
+    }
+}
+
+pub(crate) enum RunCriteriaInner {
+    Single(BoxedSystem<(), ShouldRun>),
+    Piped {
+        input: usize,
+        system: BoxedSystem<ShouldRun, ShouldRun>,
+    },
+}
+
+pub(crate) struct RunCriteriaContainer {
+    pub should_run: ShouldRun,
+    pub inner: RunCriteriaInner,
+    pub label: Option<BoxedRunCriteriaLabel>,
+    pub before: Vec<BoxedRunCriteriaLabel>,
+    pub after: Vec<BoxedRunCriteriaLabel>,
+}
+
+impl RunCriteriaContainer {
+    pub fn from_descriptor(descriptor: RunCriteriaDescriptor) -> Self {
+        Self {
+            should_run: ShouldRun::Yes,
+            inner: match descriptor.system {
+                RunCriteriaSystem::Single(system) => RunCriteriaInner::Single(system),
+                RunCriteriaSystem::Piped(system) => RunCriteriaInner::Piped { input: 0, system },
+            },
+            label: descriptor.label,
+            before: descriptor.before,
+            after: descriptor.after,
+        }
+    }
+
+    pub fn name(&self) -> Cow<'static, str> {
+        match &self.inner {
+            RunCriteriaInner::Single(system) => system.name(),
+            RunCriteriaInner::Piped { system, .. } => system.name(),
+        }
+    }
+
+    pub fn initialize(&mut self, world: &mut World) {
+        match &mut self.inner {
+            RunCriteriaInner::Single(system) => system.initialize(world),
+            RunCriteriaInner::Piped { system, .. } => system.initialize(world),
+        }
+    }
+}
+
+impl GraphNode<BoxedRunCriteriaLabel> for RunCriteriaContainer {
+    fn name(&self) -> Cow<'static, str> {
+        match &self.inner {
+            RunCriteriaInner::Single(system) => system.name(),
+            RunCriteriaInner::Piped { system, .. } => system.name(),
+        }
+    }
+
+    fn labels(&self) -> &[BoxedRunCriteriaLabel] {
+        if let Some(ref label) = self.label {
+            std::slice::from_ref(label)
+        } else {
+            &[]
+        }
+    }
+
+    fn before(&self) -> &[BoxedRunCriteriaLabel] {
+        &self.before
+    }
+
+    fn after(&self) -> &[BoxedRunCriteriaLabel] {
+        &self.after
+    }
+}
+
+pub enum RunCriteriaDescriptorOrLabel {
+    Descriptor(RunCriteriaDescriptor),
+    Label(BoxedRunCriteriaLabel),
+}
+
+#[derive(Clone, Copy)]
+pub(crate) enum DuplicateLabelStrategy {
+    Panic,
+    Discard,
+}
+
+pub struct RunCriteriaDescriptor {
+    pub(crate) system: RunCriteriaSystem,
+    pub(crate) label: Option<BoxedRunCriteriaLabel>,
+    pub(crate) duplicate_label_strategy: DuplicateLabelStrategy,
+    pub(crate) before: Vec<BoxedRunCriteriaLabel>,
+    pub(crate) after: Vec<BoxedRunCriteriaLabel>,
+}
+
+pub(crate) enum RunCriteriaSystem {
+    Single(BoxedSystem<(), ShouldRun>),
+    Piped(BoxedSystem<ShouldRun, ShouldRun>),
+}
+
+pub trait IntoRunCriteria<Marker> {
+    fn into(self) -> RunCriteriaDescriptorOrLabel;
+}
+
+impl IntoRunCriteria<RunCriteriaDescriptor> for RunCriteriaDescriptorOrLabel {
+    fn into(self) -> RunCriteriaDescriptorOrLabel {
+        self
+    }
+}
+
+impl IntoRunCriteria<RunCriteriaDescriptorOrLabel> for RunCriteriaDescriptor {
+    fn into(self) -> RunCriteriaDescriptorOrLabel {
+        RunCriteriaDescriptorOrLabel::Descriptor(self)
+    }
+}
+
+impl IntoRunCriteria<BoxedSystem<(), ShouldRun>> for BoxedSystem<(), ShouldRun> {
+    fn into(self) -> RunCriteriaDescriptorOrLabel {
+        RunCriteriaDescriptorOrLabel::Descriptor(new_run_criteria_descriptor(self))
+    }
+}
+
+impl<S> IntoRunCriteria<BoxedSystem<(), ShouldRun>> for S
+where
+    S: System<In = (), Out = ShouldRun>,
+{
+    fn into(self) -> RunCriteriaDescriptorOrLabel {
+        RunCriteriaDescriptorOrLabel::Descriptor(new_run_criteria_descriptor(Box::new(self)))
+    }
+}
+
+impl IntoRunCriteria<BoxedRunCriteriaLabel> for BoxedRunCriteriaLabel {
+    fn into(self) -> RunCriteriaDescriptorOrLabel {
+        RunCriteriaDescriptorOrLabel::Label(self)
+    }
+}
+
+impl<L> IntoRunCriteria<BoxedRunCriteriaLabel> for L
+where
+    L: RunCriteriaLabel,
+{
+    fn into(self) -> RunCriteriaDescriptorOrLabel {
+        RunCriteriaDescriptorOrLabel::Label(Box::new(self))
+    }
+}
+
+impl IntoRunCriteria<RunCriteria> for RunCriteria {
+    fn into(self) -> RunCriteriaDescriptorOrLabel {
+        RunCriteriaDescriptorOrLabel::Label(self.label)
+    }
+}
+
+pub trait RunCriteriaDescriptorCoercion {
+    /// Assigns a label to the criteria. Must be unique.
+    fn label(self, label: impl RunCriteriaLabel) -> RunCriteriaDescriptor;
+
+    /// Assigns a label to the criteria. If the given label is already in use,
+    /// this criteria will be discarded before initialization.
+    fn label_discard_if_duplicate(self, label: impl RunCriteriaLabel) -> RunCriteriaDescriptor;
+
+    /// Specifies that this criteria must be evaluated before a criteria with the given label.
+    fn before(self, label: impl RunCriteriaLabel) -> RunCriteriaDescriptor;
+
+    /// Specifies that this criteria must be evaluated after a criteria with the given label.
+    fn after(self, label: impl RunCriteriaLabel) -> RunCriteriaDescriptor;
+}
+
+impl RunCriteriaDescriptorCoercion for RunCriteriaDescriptor {
+    fn label(mut self, label: impl RunCriteriaLabel) -> RunCriteriaDescriptor {
+        self.label = Some(Box::new(label));
+        self.duplicate_label_strategy = DuplicateLabelStrategy::Panic;
+        self
+    }
+
+    fn label_discard_if_duplicate(mut self, label: impl RunCriteriaLabel) -> RunCriteriaDescriptor {
+        self.label = Some(Box::new(label));
+        self.duplicate_label_strategy = DuplicateLabelStrategy::Discard;
+        self
+    }
+
+    fn before(mut self, label: impl RunCriteriaLabel) -> RunCriteriaDescriptor {
+        self.before.push(Box::new(label));
+        self
+    }
+
+    fn after(mut self, label: impl RunCriteriaLabel) -> RunCriteriaDescriptor {
+        self.after.push(Box::new(label));
+        self
+    }
+}
+
+fn new_run_criteria_descriptor(system: BoxedSystem<(), ShouldRun>) -> RunCriteriaDescriptor {
+    RunCriteriaDescriptor {
+        system: RunCriteriaSystem::Single(system),
+        label: None,
+        duplicate_label_strategy: DuplicateLabelStrategy::Panic,
+        before: vec![],
+        after: vec![],
+    }
+}
+
+impl RunCriteriaDescriptorCoercion for BoxedSystem<(), ShouldRun> {
+    fn label(self, label: impl RunCriteriaLabel) -> RunCriteriaDescriptor {
+        new_run_criteria_descriptor(self).label(label)
+    }
+
+    fn label_discard_if_duplicate(self, label: impl RunCriteriaLabel) -> RunCriteriaDescriptor {
+        new_run_criteria_descriptor(self).label_discard_if_duplicate(label)
+    }
+
+    fn before(self, label: impl RunCriteriaLabel) -> RunCriteriaDescriptor {
+        new_run_criteria_descriptor(self).before(label)
+    }
+
+    fn after(self, label: impl RunCriteriaLabel) -> RunCriteriaDescriptor {
+        new_run_criteria_descriptor(self).after(label)
+    }
+}
+
+impl<S> RunCriteriaDescriptorCoercion for S
+where
+    S: System<In = (), Out = ShouldRun>,
+{
+    fn label(self, label: impl RunCriteriaLabel) -> RunCriteriaDescriptor {
+        new_run_criteria_descriptor(Box::new(self)).label(label)
+    }
+
+    fn label_discard_if_duplicate(self, label: impl RunCriteriaLabel) -> RunCriteriaDescriptor {
+        new_run_criteria_descriptor(Box::new(self)).label_discard_if_duplicate(label)
+    }
+
+    fn before(self, label: impl RunCriteriaLabel) -> RunCriteriaDescriptor {
+        new_run_criteria_descriptor(Box::new(self)).before(label)
+    }
+
+    fn after(self, label: impl RunCriteriaLabel) -> RunCriteriaDescriptor {
+        new_run_criteria_descriptor(Box::new(self)).after(label)
+    }
+}
+
+pub struct RunCriteria {
+    label: BoxedRunCriteriaLabel,
+}
+
+impl RunCriteria {
+    /// Constructs a new run criteria that will retrieve the result of the criteria `label`
+    /// and pipe it as input to `system`.
+    pub fn pipe(
+        label: impl RunCriteriaLabel,
+        system: impl System<In = ShouldRun, Out = ShouldRun>,
+    ) -> RunCriteriaDescriptor {
+        label.pipe(system)
+    }
+}
+
+pub trait RunCriteriaPiping {
+    /// See [`RunCriteria::pipe()`].
+    fn pipe(self, system: impl System<In = ShouldRun, Out = ShouldRun>) -> RunCriteriaDescriptor;
+}
+
+impl RunCriteriaPiping for BoxedRunCriteriaLabel {
+    fn pipe(self, system: impl System<In = ShouldRun, Out = ShouldRun>) -> RunCriteriaDescriptor {
+        RunCriteriaDescriptor {
+            system: RunCriteriaSystem::Piped(Box::new(system)),
+            label: None,
+            duplicate_label_strategy: DuplicateLabelStrategy::Panic,
+            before: vec![],
+            after: vec![self],
+        }
+    }
+}
+
+impl<L> RunCriteriaPiping for L
+where
+    L: RunCriteriaLabel,
+{
+    fn pipe(self, system: impl System<In = ShouldRun, Out = ShouldRun>) -> RunCriteriaDescriptor {
+        RunCriteriaDescriptor {
+            system: RunCriteriaSystem::Piped(Box::new(system)),
+            label: None,
+            duplicate_label_strategy: DuplicateLabelStrategy::Panic,
+            before: vec![],
+            after: vec![Box::new(self)],
+        }
+    }
+}
+
+pub struct RunOnce {
+    ran: bool,
+    system_id: SystemId,
+    archetype_component_access: Access<ArchetypeComponentId>,
+    component_access: Access<ComponentId>,
+}
+
+impl Default for RunOnce {
+    fn default() -> Self {
+        Self {
+            ran: false,
+            system_id: SystemId::new(),
+            archetype_component_access: Default::default(),
+            component_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 new_archetype(&mut self, _archetype: &Archetype) {}
+
+    fn component_access(&self) -> &Access<ComponentId> {
+        &self.component_access
+    }
+
+    fn archetype_component_access(&self) -> &Access<ArchetypeComponentId> {
+        &self.archetype_component_access
+    }
+
+    fn is_send(&self) -> bool {
+        true
+    }
+
+    unsafe fn run_unsafe(&mut self, _input: Self::In, _world: &World) -> Self::Out {
+        if self.ran {
+            ShouldRun::No
+        } else {
+            self.ran = true;
+            ShouldRun::Yes
+        }
+    }
+
+    fn apply_buffers(&mut self, _world: &mut World) {}
+
+    fn initialize(&mut self, _world: &mut World) {}
+
+    fn check_change_tick(&mut self, _change_tick: u32) {}
+}
diff --git a/crates/bevy_ecs/src/schedule/stage.rs b/crates/bevy_ecs/src/schedule/stage.rs
index e2a83b80813f0..48bae19a64870 100644
--- a/crates/bevy_ecs/src/schedule/stage.rs
+++ b/crates/bevy_ecs/src/schedule/stage.rs
@@ -1,20 +1,20 @@
 use crate::{
     component::ComponentId,
     schedule::{
-        BoxedSystemLabel, ExclusiveSystemContainer, InsertionPoint, ParallelExecutor,
-        ParallelSystemContainer, ParallelSystemExecutor, RunCriteria, ShouldRun,
+        graph_utils::{self, DependencyGraphError},
+        BoxedRunCriteria, BoxedRunCriteriaLabel, BoxedSystemLabel, DuplicateLabelStrategy,
+        ExclusiveSystemContainer, GraphNode, InsertionPoint, ParallelExecutor,
+        ParallelSystemContainer, ParallelSystemExecutor, RunCriteriaContainer,
+        RunCriteriaDescriptor, RunCriteriaDescriptorOrLabel, RunCriteriaInner, ShouldRun,
         SingleThreadedExecutor, SystemContainer, SystemDescriptor, SystemSet,
     },
     system::System,
     world::{World, WorldId},
 };
-use bevy_utils::{
-    tracing::{info, warn},
-    HashMap, HashSet,
-};
+use bevy_utils::{tracing::info, HashMap, HashSet};
 use downcast_rs::{impl_downcast, Downcast};
 use fixedbitset::FixedBitSet;
-use std::borrow::Cow;
+use std::fmt::Debug;
 
 pub trait Stage: Downcast + Send + Sync {
     /// Runs the stage; this happens once per update.
@@ -45,11 +45,6 @@ impl_downcast!(Stage);
 /// to have unambiguous order with regards to a group of already-constrained systems.
 pub struct ReportExecutionOrderAmbiguities;
 
-struct VirtualSystemSet {
-    run_criteria: RunCriteria,
-    should_run: ShouldRun,
-}
-
 /// Stores and executes systems. Execution order is not defined unless explicitly specified;
 /// see `SystemDescriptor` documentation.
 pub struct SystemStage {
@@ -57,8 +52,10 @@ pub struct SystemStage {
     world_id: Option<WorldId>,
     /// Instance of a scheduling algorithm for running the systems.
     executor: Box<dyn ParallelSystemExecutor>,
-    /// Groups of systems; each set has its own run criterion.
-    system_sets: Vec<VirtualSystemSet>,
+    /// Determines whether the stage should run.
+    stage_run_criteria: BoxedRunCriteria,
+    /// Topologically sorted run criteria of systems.
+    run_criteria: Vec<RunCriteriaContainer>,
     /// Topologically sorted exclusive systems that want to be run at the start of the stage.
     exclusive_at_start: Vec<ExclusiveSystemContainer>,
     /// Topologically sorted exclusive systems that want to be run after parallel systems but
@@ -72,6 +69,8 @@ pub struct SystemStage {
     systems_modified: bool,
     /// Determines if the stage's executor was changed.
     executor_modified: bool,
+    /// Newly inserted run criteria that will be initialized at the next opportunity.
+    uninitialized_run_criteria: Vec<(usize, DuplicateLabelStrategy)>,
     /// Newly inserted systems that will be initialized at the next opportunity.
     uninitialized_at_start: Vec<usize>,
     /// Newly inserted systems that will be initialized at the next opportunity.
@@ -86,14 +85,12 @@ pub struct SystemStage {
 
 impl SystemStage {
     pub fn new(executor: Box<dyn ParallelSystemExecutor>) -> Self {
-        let set = VirtualSystemSet {
-            run_criteria: Default::default(),
-            should_run: ShouldRun::Yes,
-        };
         SystemStage {
             world_id: None,
             executor,
-            system_sets: vec![set],
+            stage_run_criteria: Default::default(),
+            run_criteria: vec![],
+            uninitialized_run_criteria: vec![],
             exclusive_at_start: Default::default(),
             exclusive_before_commands: Default::default(),
             exclusive_at_end: Default::default(),
@@ -139,35 +136,73 @@ impl SystemStage {
         self
     }
 
-    pub fn with_system_set(mut self, system_set: SystemSet) -> Self {
-        self.add_system_set(system_set);
-        self
-    }
-
-    pub fn with_run_criteria<S: System<In = (), Out = ShouldRun>>(mut self, system: S) -> Self {
-        self.system_sets[0].run_criteria.set(Box::new(system));
+    pub fn add_system(&mut self, system: impl Into<SystemDescriptor>) -> &mut Self {
+        self.add_system_inner(system, None);
         self
     }
 
-    pub fn add_system_set(&mut self, system_set: SystemSet) -> &mut Self {
+    fn add_system_inner(
+        &mut self,
+        system: impl Into<SystemDescriptor>,
+        default_run_criteria: Option<usize>,
+    ) {
         self.systems_modified = true;
-        let SystemSet {
-            run_criteria,
-            mut descriptors,
-        } = system_set;
-        let set = self.system_sets.len();
-        self.system_sets.push(VirtualSystemSet {
-            run_criteria,
-            should_run: ShouldRun::No,
-        });
-        for system in descriptors.drain(..) {
-            self.add_system_to_set(system, set);
+        match system.into() {
+            SystemDescriptor::Exclusive(mut descriptor) => {
+                let insertion_point = descriptor.insertion_point;
+                let criteria = descriptor.run_criteria.take();
+                let mut container = ExclusiveSystemContainer::from_descriptor(descriptor);
+                match criteria {
+                    Some(RunCriteriaDescriptorOrLabel::Label(label)) => {
+                        container.run_criteria_label = Some(label);
+                    }
+                    Some(RunCriteriaDescriptorOrLabel::Descriptor(criteria_descriptor)) => {
+                        container.run_criteria_label = criteria_descriptor.label.clone();
+                        container.run_criteria_index =
+                            Some(self.add_run_criteria_internal(criteria_descriptor));
+                    }
+                    None => {
+                        container.run_criteria_index = default_run_criteria;
+                    }
+                }
+                match insertion_point {
+                    InsertionPoint::AtStart => {
+                        let index = self.exclusive_at_start.len();
+                        self.uninitialized_at_start.push(index);
+                        self.exclusive_at_start.push(container);
+                    }
+                    InsertionPoint::BeforeCommands => {
+                        let index = self.exclusive_before_commands.len();
+                        self.uninitialized_before_commands.push(index);
+                        self.exclusive_before_commands.push(container);
+                    }
+                    InsertionPoint::AtEnd => {
+                        let index = self.exclusive_at_end.len();
+                        self.uninitialized_at_end.push(index);
+                        self.exclusive_at_end.push(container);
+                    }
+                }
+            }
+            SystemDescriptor::Parallel(mut descriptor) => {
+                let criteria = descriptor.run_criteria.take();
+                let mut container = ParallelSystemContainer::from_descriptor(descriptor);
+                match criteria {
+                    Some(RunCriteriaDescriptorOrLabel::Label(label)) => {
+                        container.run_criteria_label = Some(label);
+                    }
+                    Some(RunCriteriaDescriptorOrLabel::Descriptor(criteria_descriptor)) => {
+                        container.run_criteria_label = criteria_descriptor.label.clone();
+                        container.run_criteria_index =
+                            Some(self.add_run_criteria_internal(criteria_descriptor));
+                    }
+                    None => {
+                        container.run_criteria_index = default_run_criteria;
+                    }
+                }
+                self.uninitialized_parallel.push(self.parallel.len());
+                self.parallel.push(container);
+            }
         }
-        self
-    }
-
-    pub fn add_system(&mut self, system: impl Into<SystemDescriptor>) -> &mut Self {
-        self.add_system_to_set(system, 0)
     }
 
     /// Topologically sorted parallel systems.
@@ -199,109 +234,236 @@ impl SystemStage {
         &self.exclusive_before_commands
     }
 
-    // TODO: consider exposing
-    fn add_system_to_set(&mut self, system: impl Into<SystemDescriptor>, set: usize) -> &mut Self {
-        // This assertion is there to document that a maximum of `u32::MAX / 8` systems should be
-        // added to a stage to guarantee that change detection has no false positive, but it
-        // can be circumvented using exclusive or chained systems
-        assert!(
-            self.exclusive_at_start.len()
-                + self.exclusive_before_commands.len()
-                + self.exclusive_at_end.len()
-                + self.parallel.len()
-                < (u32::MAX / 8) as usize
-        );
+    pub fn with_system_set(mut self, system_set: SystemSet) -> Self {
+        self.add_system_set(system_set);
+        self
+    }
+
+    pub fn add_system_set(&mut self, system_set: SystemSet) -> &mut Self {
         self.systems_modified = true;
-        match system.into() {
-            SystemDescriptor::Exclusive(descriptor) => {
-                let insertion_point = descriptor.insertion_point;
-                let container = ExclusiveSystemContainer::from_descriptor(descriptor, set);
-                match insertion_point {
-                    InsertionPoint::AtStart => {
-                        let index = self.exclusive_at_start.len();
-                        self.uninitialized_at_start.push(index);
-                        self.exclusive_at_start.push(container);
-                    }
-                    InsertionPoint::BeforeCommands => {
-                        let index = self.exclusive_before_commands.len();
-                        self.uninitialized_before_commands.push(index);
-                        self.exclusive_before_commands.push(container);
-                    }
-                    InsertionPoint::AtEnd => {
-                        let index = self.exclusive_at_end.len();
-                        self.uninitialized_at_end.push(index);
-                        self.exclusive_at_end.push(container);
-                    }
+        let (run_criteria, mut systems) = system_set.bake();
+        let set_run_criteria_index = run_criteria.and_then(|criteria| {
+            // validate that no systems have criteria
+            for system in systems.iter_mut() {
+                if let Some(name) = match system {
+                    SystemDescriptor::Exclusive(descriptor) => descriptor
+                        .run_criteria
+                        .is_some()
+                        .then(|| descriptor.system.name()),
+                    SystemDescriptor::Parallel(descriptor) => descriptor
+                        .run_criteria
+                        .is_some()
+                        .then(|| descriptor.system.name()),
+                } {
+                    panic!(
+                        "The system {} has a run criteria, but its `SystemSet` also has a run \
+                        criteria. This is not supported. Consider moving the system into a \
+                        different `SystemSet` or calling `add_system()` instead.",
+                        name
+                    )
                 }
             }
-            SystemDescriptor::Parallel(descriptor) => {
-                self.uninitialized_parallel.push(self.parallel.len());
-                self.parallel
-                    .push(ParallelSystemContainer::from_descriptor(descriptor, set));
+            match criteria {
+                RunCriteriaDescriptorOrLabel::Descriptor(descriptor) => {
+                    Some(self.add_run_criteria_internal(descriptor))
+                }
+                RunCriteriaDescriptorOrLabel::Label(label) => {
+                    for system in systems.iter_mut() {
+                        match system {
+                            SystemDescriptor::Exclusive(descriptor) => {
+                                descriptor.run_criteria =
+                                    Some(RunCriteriaDescriptorOrLabel::Label(label.clone()))
+                            }
+                            SystemDescriptor::Parallel(descriptor) => {
+                                descriptor.run_criteria =
+                                    Some(RunCriteriaDescriptorOrLabel::Label(label.clone()))
+                            }
+                        }
+                    }
+
+                    None
+                }
             }
+        });
+        for system in systems.drain(..) {
+            self.add_system_inner(system, set_run_criteria_index);
         }
         self
     }
 
+    pub fn with_run_criteria<S: System<In = (), Out = ShouldRun>>(mut self, system: S) -> Self {
+        self.set_run_criteria(system);
+        self
+    }
+
+    pub fn set_run_criteria<S: System<In = (), Out = ShouldRun>>(
+        &mut self,
+        system: S,
+    ) -> &mut Self {
+        self.stage_run_criteria.set(Box::new(system));
+        self
+    }
+
+    pub fn with_system_run_criteria(mut self, run_criteria: RunCriteriaDescriptor) -> Self {
+        self.add_system_run_criteria(run_criteria);
+        self
+    }
+
+    pub fn add_system_run_criteria(&mut self, run_criteria: RunCriteriaDescriptor) -> &mut Self {
+        self.add_run_criteria_internal(run_criteria);
+        self
+    }
+
+    pub(crate) fn add_run_criteria_internal(&mut self, descriptor: RunCriteriaDescriptor) -> usize {
+        let index = self.run_criteria.len();
+        self.uninitialized_run_criteria
+            .push((index, descriptor.duplicate_label_strategy));
+
+        self.run_criteria
+            .push(RunCriteriaContainer::from_descriptor(descriptor));
+        index
+    }
+
     fn initialize_systems(&mut self, world: &mut World) {
+        let mut criteria_labels = HashMap::default();
+        let uninitialized_criteria: HashMap<_, _> =
+            self.uninitialized_run_criteria.drain(..).collect();
+        // track the number of filtered criteria to correct run criteria indices
+        let mut filtered_criteria = 0;
+        let mut new_indices = Vec::new();
+        self.run_criteria = self
+            .run_criteria
+            .drain(..)
+            .enumerate()
+            .filter_map(|(index, mut container)| {
+                let new_index = index - filtered_criteria;
+                let label = container.label.clone();
+                if let Some(strategy) = uninitialized_criteria.get(&index) {
+                    if let Some(ref label) = label {
+                        if let Some(duplicate_index) = criteria_labels.get(label) {
+                            match strategy {
+                                DuplicateLabelStrategy::Panic => panic!(
+                                    "Run criteria {} is labelled with {:?}, which \
+                            is already in use. Consider using \
+                            `RunCriteriaDescriptorCoercion::label_discard_if_duplicate().",
+                                    container.name(),
+                                    container.label
+                                ),
+                                DuplicateLabelStrategy::Discard => {
+                                    new_indices.push(*duplicate_index);
+                                    filtered_criteria += 1;
+                                    return None;
+                                }
+                            }
+                        }
+                    }
+                    container.initialize(world);
+                }
+                if let Some(label) = label {
+                    criteria_labels.insert(label, new_index);
+                }
+                new_indices.push(new_index);
+                Some(container)
+            })
+            .collect();
+
         for index in self.uninitialized_at_start.drain(..) {
-            self.exclusive_at_start[index]
-                .system_mut()
-                .initialize(world);
+            let container = &mut self.exclusive_at_start[index];
+            if let Some(index) = container.run_criteria() {
+                container.set_run_criteria(new_indices[index]);
+            }
+            container.system_mut().initialize(world);
         }
         for index in self.uninitialized_before_commands.drain(..) {
-            self.exclusive_before_commands[index]
-                .system_mut()
-                .initialize(world);
+            let container = &mut self.exclusive_before_commands[index];
+            if let Some(index) = container.run_criteria() {
+                container.set_run_criteria(new_indices[index]);
+            }
+            container.system_mut().initialize(world);
         }
         for index in self.uninitialized_at_end.drain(..) {
-            self.exclusive_at_end[index].system_mut().initialize(world);
+            let container = &mut self.exclusive_at_end[index];
+            if let Some(index) = container.run_criteria() {
+                container.set_run_criteria(new_indices[index]);
+            }
+            container.system_mut().initialize(world);
         }
         for index in self.uninitialized_parallel.drain(..) {
-            self.parallel[index].system_mut().initialize(world);
+            let container = &mut self.parallel[index];
+            if let Some(index) = container.run_criteria() {
+                container.set_run_criteria(new_indices[index]);
+            }
+            container.system_mut().initialize(world);
         }
     }
 
     /// Rearranges all systems in topological orders. Systems must be initialized.
     fn rebuild_orders_and_dependencies(&mut self) {
+        // This assertion is there to document that a maximum of `u32::MAX / 8` systems should be
+        // added to a stage to guarantee that change detection has no false positive, but it
+        // can be circumvented using exclusive or chained systems
+        assert!(
+            self.exclusive_at_start.len()
+                + self.exclusive_before_commands.len()
+                + self.exclusive_at_end.len()
+                + self.parallel.len()
+                < (u32::MAX / 8) as usize
+        );
         debug_assert!(
-            self.uninitialized_parallel.is_empty()
+            self.uninitialized_run_criteria.is_empty()
+                && self.uninitialized_parallel.is_empty()
                 && self.uninitialized_at_start.is_empty()
                 && self.uninitialized_before_commands.is_empty()
                 && self.uninitialized_at_end.is_empty()
         );
-        fn sort_systems_unwrap(
-            systems: &mut Vec<impl SystemContainer>,
-            systems_description: &'static str,
-        ) {
-            if let Err(DependencyGraphError::GraphCycles(cycle)) = sort_systems(systems) {
-                use std::fmt::Write;
-                let mut message = format!("Found a dependency cycle in {}:", systems_description);
-                writeln!(message).unwrap();
-                for (name, labels) in &cycle {
-                    writeln!(message, " - {}", name).unwrap();
-                    writeln!(
-                        message,
-                        "    wants to be after (because of labels {:?})",
-                        labels
-                    )
-                    .unwrap();
+        fn unwrap_dependency_cycle_error<Output, Label, Labels: Debug>(
+            result: Result<Output, DependencyGraphError<Labels>>,
+            nodes: &[impl GraphNode<Label>],
+            nodes_description: &'static str,
+        ) -> Output {
+            match result {
+                Ok(output) => output,
+                Err(DependencyGraphError::GraphCycles(cycle)) => {
+                    use std::fmt::Write;
+                    let mut message = format!("Found a dependency cycle in {}:", nodes_description);
+                    writeln!(message).unwrap();
+                    for (index, labels) in &cycle {
+                        writeln!(message, " - {}", nodes[*index].name()).unwrap();
+                        writeln!(
+                            message,
+                            "    wants to be after (because of labels: {:?})",
+                            labels,
+                        )
+                        .unwrap();
+                    }
+                    writeln!(message, " - {}", cycle[0].0).unwrap();
+                    panic!("{}", message);
                 }
-                writeln!(message, " - {}", cycle[0].0).unwrap();
-                panic!("{}", message);
             }
         }
-        sort_systems_unwrap(&mut self.parallel, "parallel systems");
-        sort_systems_unwrap(
-            &mut self.exclusive_at_start,
+        let run_criteria_labels = unwrap_dependency_cycle_error(
+            self.process_run_criteria(),
+            &self.run_criteria,
+            "run criteria",
+        );
+        unwrap_dependency_cycle_error(
+            process_systems(&mut self.parallel, &run_criteria_labels),
+            &self.parallel,
+            "parallel systems",
+        );
+        unwrap_dependency_cycle_error(
+            process_systems(&mut self.exclusive_at_start, &run_criteria_labels),
+            &self.exclusive_at_start,
             "exclusive systems at start of stage",
         );
-        sort_systems_unwrap(
-            &mut self.exclusive_before_commands,
+        unwrap_dependency_cycle_error(
+            process_systems(&mut self.exclusive_before_commands, &run_criteria_labels),
+            &self.exclusive_before_commands,
             "exclusive systems before commands of stage",
         );
-        sort_systems_unwrap(
-            &mut self.exclusive_at_end,
+        unwrap_dependency_cycle_error(
+            process_systems(&mut self.exclusive_at_end, &run_criteria_labels),
+            &self.exclusive_at_end,
             "exclusive systems at end of stage",
         );
     }
@@ -405,19 +567,84 @@ impl SystemStage {
             self.last_tick_check = change_tick;
         }
     }
-}
 
-enum DependencyGraphError {
-    GraphCycles(Vec<(Cow<'static, str>, Vec<BoxedSystemLabel>)>),
+    /// Sorts run criteria and populates resolved input-criteria for piping.
+    /// Returns a map of run criteria labels to their indices.
+    fn process_run_criteria(
+        &mut self,
+    ) -> Result<
+        HashMap<BoxedRunCriteriaLabel, usize>,
+        DependencyGraphError<HashSet<BoxedRunCriteriaLabel>>,
+    > {
+        let graph = graph_utils::build_dependency_graph(&self.run_criteria);
+        let order = graph_utils::topological_order(&graph)?;
+        let mut order_inverted = order.iter().enumerate().collect::<Vec<_>>();
+        order_inverted.sort_unstable_by_key(|(_, &key)| key);
+        let labels: HashMap<_, _> = self
+            .run_criteria
+            .iter()
+            .enumerate()
+            .filter_map(|(index, criteria)| {
+                criteria
+                    .label
+                    .as_ref()
+                    .map(|label| (label.clone(), order_inverted[index].0))
+            })
+            .collect();
+        for criteria in self.run_criteria.iter_mut() {
+            if let RunCriteriaInner::Piped { input: parent, .. } = &mut criteria.inner {
+                let label = &criteria.after[0];
+                *parent = *labels.get(label).unwrap_or_else(|| {
+                    panic!(
+                        "Couldn't find run criteria labelled {:?} to pipe from.",
+                        label
+                    )
+                });
+            }
+        }
+
+        fn update_run_criteria_indices<T: SystemContainer>(
+            systems: &mut [T],
+            order_inverted: &[(usize, &usize)],
+        ) {
+            for system in systems {
+                if let Some(index) = system.run_criteria() {
+                    system.set_run_criteria(order_inverted[index].0);
+                }
+            }
+        }
+
+        update_run_criteria_indices(&mut self.exclusive_at_end, &order_inverted);
+        update_run_criteria_indices(&mut self.exclusive_at_start, &order_inverted);
+        update_run_criteria_indices(&mut self.exclusive_before_commands, &order_inverted);
+        update_run_criteria_indices(&mut self.parallel, &order_inverted);
+
+        let mut temp = self.run_criteria.drain(..).map(Some).collect::<Vec<_>>();
+        for index in order {
+            self.run_criteria.push(temp[index].take().unwrap());
+        }
+        Ok(labels)
+    }
 }
 
-/// Sorts given system containers topologically and populates their resolved dependencies.
-fn sort_systems(systems: &mut Vec<impl SystemContainer>) -> Result<(), DependencyGraphError> {
-    let mut graph = build_dependency_graph(systems);
-    let order = topological_order(systems, &graph)?;
+/// Sorts given system containers topologically, populates their resolved dependencies
+/// and run criteria.
+fn process_systems(
+    systems: &mut Vec<impl SystemContainer>,
+    run_criteria_labels: &HashMap<BoxedRunCriteriaLabel, usize>,
+) -> Result<(), DependencyGraphError<HashSet<BoxedSystemLabel>>> {
+    let mut graph = graph_utils::build_dependency_graph(systems);
+    let order = graph_utils::topological_order(&graph)?;
     let mut order_inverted = order.iter().enumerate().collect::<Vec<_>>();
     order_inverted.sort_unstable_by_key(|(_, &key)| key);
     for (index, container) in systems.iter_mut().enumerate() {
+        if let Some(index) = container.run_criteria_label().map(|label| {
+            *run_criteria_labels
+                .get(label)
+                .unwrap_or_else(|| panic!("No run criteria with label {:?} found.", label))
+        }) {
+            container.set_run_criteria(index);
+        }
         container.set_dependencies(
             graph
                 .get_mut(&index)
@@ -433,116 +660,6 @@ fn sort_systems(systems: &mut Vec<impl SystemContainer>) -> Result<(), Dependenc
     Ok(())
 }
 
-/// Constructs a dependency graph of given system containers.
-fn build_dependency_graph(
-    systems: &[impl SystemContainer],
-) -> HashMap<usize, HashMap<usize, HashSet<BoxedSystemLabel>>> {
-    let mut labelled_systems = HashMap::<BoxedSystemLabel, FixedBitSet>::default();
-    for (label, index) in systems.iter().enumerate().flat_map(|(index, container)| {
-        container
-            .labels()
-            .iter()
-            .cloned()
-            .map(move |label| (label, index))
-    }) {
-        labelled_systems
-            .entry(label)
-            .or_insert_with(|| FixedBitSet::with_capacity(systems.len()))
-            .insert(index);
-    }
-    let mut graph = HashMap::with_capacity_and_hasher(systems.len(), Default::default());
-    for (system_index, container) in systems.iter().enumerate() {
-        let dependencies = graph.entry(system_index).or_insert_with(HashMap::default);
-        for label in container.after() {
-            match labelled_systems.get(label) {
-                Some(new_dependencies) => {
-                    for dependency in new_dependencies.ones() {
-                        dependencies
-                            .entry(dependency)
-                            .or_insert_with(HashSet::default)
-                            .insert(label.clone());
-                    }
-                }
-                None => warn!(
-                    "System {} wants to be after unknown system label: {:?}",
-                    systems[system_index].name(),
-                    label
-                ),
-            }
-        }
-        for label in container.before() {
-            match labelled_systems.get(label) {
-                Some(dependants) => {
-                    for dependant in dependants.ones() {
-                        graph
-                            .entry(dependant)
-                            .or_insert_with(HashMap::default)
-                            .entry(system_index)
-                            .or_insert_with(HashSet::default)
-                            .insert(label.clone());
-                    }
-                }
-                None => warn!(
-                    "System {} wants to be before unknown system label: {:?}",
-                    systems[system_index].name(),
-                    label
-                ),
-            }
-        }
-    }
-    graph
-}
-
-/// Generates a topological order for the given graph.
-fn topological_order(
-    systems: &[impl SystemContainer],
-    graph: &HashMap<usize, HashMap<usize, HashSet<BoxedSystemLabel>>>,
-) -> Result<Vec<usize>, DependencyGraphError> {
-    fn check_if_cycles_and_visit(
-        node: &usize,
-        graph: &HashMap<usize, HashMap<usize, HashSet<BoxedSystemLabel>>>,
-        sorted: &mut Vec<usize>,
-        unvisited: &mut HashSet<usize>,
-        current: &mut Vec<usize>,
-    ) -> bool {
-        if current.contains(node) {
-            return true;
-        } else if !unvisited.remove(node) {
-            return false;
-        }
-        current.push(*node);
-        for dependency in graph.get(node).unwrap().keys() {
-            if check_if_cycles_and_visit(dependency, &graph, sorted, unvisited, current) {
-                return true;
-            }
-        }
-        sorted.push(*node);
-        current.pop();
-        false
-    }
-    let mut sorted = Vec::with_capacity(graph.len());
-    let mut current = Vec::with_capacity(graph.len());
-    let mut unvisited = HashSet::with_capacity_and_hasher(graph.len(), Default::default());
-    unvisited.extend(graph.keys().cloned());
-    while let Some(node) = unvisited.iter().next().cloned() {
-        if check_if_cycles_and_visit(&node, graph, &mut sorted, &mut unvisited, &mut current) {
-            let mut cycle = Vec::new();
-            let last_window = [*current.last().unwrap(), current[0]];
-            let mut windows = current
-                .windows(2)
-                .chain(std::iter::once(&last_window as &[usize]));
-            while let Some(&[dependant, dependency]) = windows.next() {
-                cycle.push((
-                    systems[dependant].name(),
-                    graph[&dependant][&dependency].iter().cloned().collect(),
-                ));
-            }
-            return Err(DependencyGraphError::GraphCycles(cycle));
-        }
-    }
-    Ok(sorted)
-}
-
 /// Returns vector containing all pairs of indices of systems with ambiguous execution order,
 /// along with specific components that have triggered the warning.
 /// Systems must be topologically sorted beforehand.
@@ -629,15 +746,11 @@ impl Stage for SystemStage {
         } else {
             self.world_id = Some(world.id());
         }
-        // Evaluate sets' run criteria, initialize sets as needed, detect if any sets were changed.
-        let mut has_work = false;
-        for system_set in self.system_sets.iter_mut() {
-            let result = system_set.run_criteria.should_run(world);
-            match result {
-                ShouldRun::Yes | ShouldRun::YesAndCheckAgain => has_work = true,
-                ShouldRun::No | ShouldRun::NoAndCheckAgain => (),
-            }
-            system_set.should_run = result;
+
+        if let ShouldRun::No | ShouldRun::NoAndCheckAgain =
+            self.stage_run_criteria.should_run(world)
+        {
+            return;
         }
 
         if self.systems_modified {
@@ -654,12 +767,37 @@ impl Stage for SystemStage {
             self.executor_modified = false;
         }
 
+        // Evaluate run criteria.
+        for index in 0..self.run_criteria.len() {
+            let (run_criteria, tail) = self.run_criteria.split_at_mut(index);
+            let mut criteria = &mut tail[0];
+            match &mut criteria.inner {
+                RunCriteriaInner::Single(system) => criteria.should_run = system.run((), world),
+                RunCriteriaInner::Piped {
+                    input: parent,
+                    system,
+                    ..
+                } => criteria.should_run = system.run(run_criteria[*parent].should_run, world),
+            }
+        }
+
+        let mut has_work = true;
         while has_work {
+            fn should_run(
+                container: &impl SystemContainer,
+                run_criteria: &[RunCriteriaContainer],
+            ) -> bool {
+                matches!(
+                    container
+                        .run_criteria()
+                        .map(|index| run_criteria[index].should_run),
+                    None | Some(ShouldRun::Yes) | Some(ShouldRun::YesAndCheckAgain)
+                )
+            }
+
             // Run systems that want to be at the start of stage.
             for container in &mut self.exclusive_at_start {
-                if let ShouldRun::Yes | ShouldRun::YesAndCheckAgain =
-                    self.system_sets[container.system_set()].should_run
-                {
+                if should_run(container, &self.run_criteria) {
                     container.system_mut().run(world);
                 }
             }
@@ -667,18 +805,13 @@ impl Stage for SystemStage {
             // Run parallel systems using the executor.
             // TODO: hard dependencies, nested sets, whatever... should be evaluated here.
             for container in &mut self.parallel {
-                match self.system_sets[container.system_set()].should_run {
-                    ShouldRun::Yes | ShouldRun::YesAndCheckAgain => container.should_run = true,
-                    ShouldRun::No | ShouldRun::NoAndCheckAgain => container.should_run = false,
-                }
+                container.should_run = should_run(container, &self.run_criteria);
             }
             self.executor.run_systems(&mut self.parallel, world);
 
             // Run systems that want to be between parallel systems and their command buffers.
             for container in &mut self.exclusive_before_commands {
-                if let ShouldRun::Yes | ShouldRun::YesAndCheckAgain =
-                    self.system_sets[container.system_set()].should_run
-                {
+                if should_run(container, &self.run_criteria) {
                     container.system_mut().run(world);
                 }
             }
@@ -692,9 +825,7 @@ impl Stage for SystemStage {
 
             // Run systems that want to be at the end of stage.
             for container in &mut self.exclusive_at_end {
-                if let ShouldRun::Yes | ShouldRun::YesAndCheckAgain =
-                    self.system_sets[container.system_set()].should_run
-                {
+                if should_run(container, &self.run_criteria) {
                     container.system_mut().run(world);
                 }
             }
@@ -702,19 +833,33 @@ impl Stage for SystemStage {
             // Check for old component and system change ticks
             self.check_change_ticks(world);
 
-            // Reevaluate system sets' run criteria.
+            // Reevaluate run criteria.
             has_work = false;
-            for system_set in self.system_sets.iter_mut() {
-                match system_set.should_run {
+            let run_criteria = &mut self.run_criteria;
+            for index in 0..run_criteria.len() {
+                let (run_criteria, tail) = run_criteria.split_at_mut(index);
+                let criteria = &mut tail[0];
+                match criteria.should_run {
                     ShouldRun::No => (),
-                    ShouldRun::Yes => system_set.should_run = ShouldRun::No,
+                    ShouldRun::Yes => criteria.should_run = ShouldRun::No,
                     ShouldRun::YesAndCheckAgain | ShouldRun::NoAndCheckAgain => {
-                        let new_result = system_set.run_criteria.should_run(world);
-                        match new_result {
+                        match &mut criteria.inner {
+                            RunCriteriaInner::Single(system) => {
+                                criteria.should_run = system.run((), world)
+                            }
+                            RunCriteriaInner::Piped {
+                                input: parent,
+                                system,
+                                ..
+                            } => {
+                                criteria.should_run =
+                                    system.run(run_criteria[*parent].should_run, world)
+                            }
+                        }
+                        match criteria.should_run {
                             ShouldRun::Yes | ShouldRun::YesAndCheckAgain => has_work = true,
                             ShouldRun::No | ShouldRun::NoAndCheckAgain => (),
                         }
-                        system_set.should_run = new_result;
                     }
                 }
             }
@@ -730,9 +875,10 @@ mod tests {
         query::Changed,
         schedule::{
             BoxedSystemLabel, ExclusiveSystemDescriptorCoercion, ParallelSystemDescriptorCoercion,
-            ShouldRun, SingleThreadedExecutor, Stage, SystemSet, SystemStage,
+            RunCriteria, RunCriteriaDescriptorCoercion, RunCriteriaPiping, ShouldRun,
+            SingleThreadedExecutor, Stage, SystemSet, SystemStage,
         },
-        system::{IntoExclusiveSystem, IntoSystem, Query, ResMut},
+        system::{In, IntoExclusiveSystem, IntoSystem, Local, Query, ResMut},
         world::World,
     };
 
@@ -750,12 +896,13 @@ mod tests {
         }};
     }
 
-    fn resettable_run_once(mut has_ran: ResMut<bool>) -> ShouldRun {
-        if !*has_ran {
-            *has_ran = true;
-            return ShouldRun::Yes;
+    fn every_other_time(mut has_ran: Local<bool>) -> ShouldRun {
+        *has_ran = !*has_ran;
+        if *has_ran {
+            ShouldRun::Yes
+        } else {
+            ShouldRun::No
         }
-        ShouldRun::No
     }
 
     #[test]
@@ -1002,18 +1149,16 @@ mod tests {
     fn exclusive_run_criteria() {
         let mut world = World::new();
         world.insert_resource(Vec::<usize>::new());
-        world.insert_resource(false);
         let mut stage = SystemStage::parallel()
             .with_system(make_exclusive(0).exclusive_system().before("1"))
             .with_system_set(
                 SystemSet::new()
-                    .with_run_criteria(resettable_run_once.system())
+                    .with_run_criteria(every_other_time.system())
                     .with_system(make_exclusive(1).exclusive_system().label("1")),
             )
             .with_system(make_exclusive(2).exclusive_system().after("1"));
         stage.run(&mut world);
         stage.run(&mut world);
-        *world.get_resource_mut::<bool>().unwrap() = false;
         stage.set_executor(Box::new(SingleThreadedExecutor::default()));
         stage.run(&mut world);
         stage.run(&mut world);
@@ -1223,19 +1368,37 @@ mod tests {
     #[test]
     fn parallel_run_criteria() {
         let mut world = World::new();
+
         world.insert_resource(Vec::<usize>::new());
-        world.insert_resource(false);
+        let mut stage = SystemStage::parallel()
+            .with_system(
+                make_parallel!(0)
+                    .system()
+                    .label("0")
+                    .with_run_criteria(every_other_time.system()),
+            )
+            .with_system(make_parallel!(1).system().after("0"));
+        stage.run(&mut world);
+        stage.run(&mut world);
+        stage.set_executor(Box::new(SingleThreadedExecutor::default()));
+        stage.run(&mut world);
+        stage.run(&mut world);
+        assert_eq!(
+            *world.get_resource::<Vec<usize>>().unwrap(),
+            vec![0, 1, 1, 0, 1, 1]
+        );
+
+        world.get_resource_mut::<Vec<usize>>().unwrap().clear();
         let mut stage = SystemStage::parallel()
             .with_system(make_parallel!(0).system().before("1"))
             .with_system_set(
                 SystemSet::new()
-                    .with_run_criteria(resettable_run_once.system())
+                    .with_run_criteria(every_other_time.system())
                     .with_system(make_parallel!(1).system().label("1")),
             )
             .with_system(make_parallel!(2).system().after("1"));
         stage.run(&mut world);
         stage.run(&mut world);
-        *world.get_resource_mut::<bool>().unwrap() = false;
         stage.set_executor(Box::new(SingleThreadedExecutor::default()));
         stage.run(&mut world);
         stage.run(&mut world);
@@ -1243,6 +1406,131 @@ mod tests {
             *world.get_resource::<Vec<usize>>().unwrap(),
             vec![0, 1, 2, 0, 2, 0, 1, 2, 0, 2]
         );
+
+        // Reusing criteria.
+        world.get_resource_mut::<Vec<usize>>().unwrap().clear();
+        let mut stage = SystemStage::parallel()
+            .with_system_run_criteria(every_other_time.system().label("every other time"))
+            .with_system(make_parallel!(0).system().before("1"))
+            .with_system(
+                make_parallel!(1)
+                    .system()
+                    .label("1")
+                    .with_run_criteria("every other time"),
+            )
+            .with_system(
+                make_parallel!(2)
+                    .system()
+                    .label("2")
+                    .after("1")
+                    .with_run_criteria("every other time"),
+            )
+            .with_system(make_parallel!(3).system().after("2"));
+        stage.run(&mut world);
+        stage.run(&mut world);
+        stage.set_executor(Box::new(SingleThreadedExecutor::default()));
+        stage.run(&mut world);
+        stage.run(&mut world);
+        assert_eq!(
+            *world.get_resource::<Vec<usize>>().unwrap(),
+            vec![0, 1, 2, 3, 0, 3, 0, 1, 2, 3, 0, 3]
+        );
+        assert_eq!(stage.run_criteria.len(), 1);
+
+        // Piping criteria.
+        world.get_resource_mut::<Vec<usize>>().unwrap().clear();
+        fn eot_piped(input: In<ShouldRun>, has_ran: Local<bool>) -> ShouldRun {
+            if let ShouldRun::Yes | ShouldRun::YesAndCheckAgain = input.0 {
+                every_other_time(has_ran)
+            } else {
+                ShouldRun::No
+            }
+        }
+        let mut stage = SystemStage::parallel()
+            .with_system(make_parallel!(0).system().label("0"))
+            .with_system(
+                make_parallel!(1)
+                    .system()
+                    .label("1")
+                    .after("0")
+                    .with_run_criteria(every_other_time.system().label("every other time")),
+            )
+            .with_system(
+                make_parallel!(2)
+                    .system()
+                    .label("2")
+                    .after("1")
+                    .with_run_criteria(RunCriteria::pipe("every other time", eot_piped.system())),
+            )
+            .with_system(
+                make_parallel!(3)
+                    .system()
+                    .label("3")
+                    .after("2")
+                    .with_run_criteria("every other time".pipe(eot_piped.system()).label("piped")),
+            )
+            .with_system(
+                make_parallel!(4)
+                    .system()
+                    .after("3")
+                    .with_run_criteria("piped"),
+            );
+        for _ in 0..4 {
+            stage.run(&mut world);
+        }
+        stage.set_executor(Box::new(SingleThreadedExecutor::default()));
+        for _ in 0..5 {
+            stage.run(&mut world);
+        }
+        assert_eq!(
+            *world.get_resource::<Vec<usize>>().unwrap(),
+            vec![0, 1, 2, 3, 4, 0, 0, 1, 0, 0, 1, 2, 3, 4, 0, 0, 1, 0, 0, 1, 2, 3, 4]
+        );
+        assert_eq!(stage.run_criteria.len(), 3);
+
+        // Discarding extra criteria with matching labels.
+        world.get_resource_mut::<Vec<usize>>().unwrap().clear();
+        let mut stage = SystemStage::parallel()
+            .with_system(make_parallel!(0).system().before("1"))
+            .with_system(
+                make_parallel!(1).system().label("1").with_run_criteria(
+                    every_other_time
+                        .system()
+                        .label_discard_if_duplicate("every other time"),
+                ),
+            )
+            .with_system(
+                make_parallel!(2)
+                    .system()
+                    .label("2")
+                    .after("1")
+                    .with_run_criteria(
+                        every_other_time
+                            .system()
+                            .label_discard_if_duplicate("every other time"),
+                    ),
+            )
+            .with_system(make_parallel!(3).system().after("2"));
+        stage.run(&mut world);
+        stage.run(&mut world);
+        stage.set_executor(Box::new(SingleThreadedExecutor::default()));
+        stage.run(&mut world);
+        stage.run(&mut world);
+        assert_eq!(
+            *world.get_resource::<Vec<usize>>().unwrap(),
+            vec![0, 1, 2, 3, 0, 3, 0, 1, 2, 3, 0, 3]
+        );
+        assert_eq!(stage.run_criteria.len(), 1);
+    }
+
+    #[test]
+    #[should_panic]
+    fn duplicate_run_criteria_label_panic() {
+        let mut world = World::new();
+        let mut stage = SystemStage::parallel()
+            .with_system_run_criteria(every_other_time.system().label("every other time"))
+            .with_system_run_criteria(every_other_time.system().label("every other time"));
+        stage.run(&mut world);
     }
 
     #[test]
diff --git a/crates/bevy_ecs/src/schedule/state.rs b/crates/bevy_ecs/src/schedule/state.rs
index 4d959c4fb2dc7..7b1cb5d2fe8aa 100644
--- a/crates/bevy_ecs/src/schedule/state.rs
+++ b/crates/bevy_ecs/src/schedule/state.rs
@@ -1,8 +1,12 @@
 use crate::{
     component::Component,
-    schedule::{ShouldRun, SystemSet},
-    system::{In, IntoChainSystem, IntoSystem, Local, Res, ResMut, System},
+    schedule::{
+        RunCriteriaDescriptor, RunCriteriaDescriptorCoercion, RunCriteriaLabel, ShouldRun,
+        SystemSet,
+    },
+    system::{In, IntoChainSystem, IntoSystem, Local, Res, ResMut},
 };
+use std::{any::TypeId, fmt::Debug, hash::Hash};
 use thiserror::Error;
 
 /// ### Stack based state machine
@@ -12,7 +16,7 @@ use thiserror::Error;
 /// * Pop removes the current state, and unpauses the last paused state.
 /// * Next unwinds the state stack, and replaces the entire stack with a single new state
 #[derive(Debug)]
-pub struct State<T: Component + Clone + Eq> {
+pub struct State<T> {
     transition: Option<StateTransition<T>>,
     stack: Vec<T>,
     scheduled: Option<ScheduledOperation<T>>,
@@ -20,7 +24,7 @@ pub struct State<T: Component + Clone + Eq> {
 }
 
 #[derive(Debug)]
-enum StateTransition<T: Component + Clone + Eq> {
+enum StateTransition<T> {
     PreStartup,
     Startup,
     // The parameter order is always (leaving, entering)
@@ -32,23 +36,73 @@ enum StateTransition<T: Component + Clone + Eq> {
 }
 
 #[derive(Debug)]
-enum ScheduledOperation<T: Component + Clone + Eq> {
+enum ScheduledOperation<T> {
     Next(T),
     Pop,
     Push(T),
 }
 
-impl<T: Component + Clone + Eq> State<T> {
-    pub fn on_update(s: T) -> impl System<In = (), Out = ShouldRun> {
+#[derive(Debug, PartialEq, Eq, Clone, Hash)]
+enum StateCallback {
+    Update,
+    InactiveUpdate,
+    InStackUpdate,
+    Enter,
+    Exit,
+    Pause,
+    Resume,
+}
+
+impl StateCallback {
+    fn into_label<T>(self, state: T) -> StateRunCriteriaLabel<T>
+    where
+        T: Component + Debug + Clone + Eq + Hash,
+    {
+        StateRunCriteriaLabel(state, self)
+    }
+}
+
+#[derive(Debug, PartialEq, Eq, Clone, Hash)]
+struct StateRunCriteriaLabel<T>(T, StateCallback);
+impl<T> RunCriteriaLabel for StateRunCriteriaLabel<T>
+where
+    T: Component + Debug + Clone + Eq + Hash,
+{
+    fn dyn_clone(&self) -> Box<dyn RunCriteriaLabel> {
+        Box::new(self.clone())
+    }
+}
+
+#[derive(Debug, PartialEq, Eq, Clone, Hash)]
+struct DriverLabel(TypeId);
+impl RunCriteriaLabel for DriverLabel {
+    fn dyn_clone(&self) -> Box<dyn RunCriteriaLabel> {
+        Box::new(self.clone())
+    }
+}
+
+impl DriverLabel {
+    fn of<T: 'static>() -> Self {
+        Self(TypeId::of::<T>())
+    }
+}
+
+impl<T> State<T>
+where
+    T: Component + Debug + Clone + Eq + Hash,
+{
+    pub fn on_update(s: T) -> RunCriteriaDescriptor {
         (|state: Res<State<T>>, pred: Local<Option<T>>| {
             state.stack.last().unwrap() == pred.as_ref().unwrap() && state.transition.is_none()
         })
         .system()
-        .config(|(_, pred)| *pred = Some(Some(s)))
+        .config(|(_, pred)| *pred = Some(Some(s.clone())))
         .chain(should_run_adapter::<T>.system())
+        .after(DriverLabel::of::<T>())
+        .label_discard_if_duplicate(StateCallback::Update.into_label(s))
     }
 
-    pub fn on_inactive_update(s: T) -> impl System<In = (), Out = ShouldRun> {
+    pub fn on_inactive_update(s: T) -> RunCriteriaDescriptor {
         (|state: Res<State<T>>, mut is_inactive: Local<bool>, pred: Local<Option<T>>| match &state
             .transition
         {
@@ -63,11 +117,13 @@ impl<T: Component + Clone + Eq> State<T> {
             None => *is_inactive,
         })
         .system()
-        .config(|(_, _, pred)| *pred = Some(Some(s)))
+        .config(|(_, _, pred)| *pred = Some(Some(s.clone())))
         .chain(should_run_adapter::<T>.system())
+        .after(DriverLabel::of::<T>())
+        .label_discard_if_duplicate(StateCallback::InactiveUpdate.into_label(s))
     }
 
-    pub fn on_in_stack_update(s: T) -> impl System<In = (), Out = ShouldRun> {
+    pub fn on_in_stack_update(s: T) -> RunCriteriaDescriptor {
         (|state: Res<State<T>>, mut is_in_stack: Local<bool>, pred: Local<Option<T>>| match &state
             .transition
         {
@@ -94,11 +150,13 @@ impl<T: Component + Clone + Eq> State<T> {
             None => *is_in_stack,
         })
         .system()
-        .config(|(_, _, pred)| *pred = Some(Some(s)))
+        .config(|(_, _, pred)| *pred = Some(Some(s.clone())))
         .chain(should_run_adapter::<T>.system())
+        .after(DriverLabel::of::<T>())
+        .label_discard_if_duplicate(StateCallback::InStackUpdate.into_label(s))
     }
 
-    pub fn on_enter(s: T) -> impl System<In = (), Out = ShouldRun> {
+    pub fn on_enter(s: T) -> RunCriteriaDescriptor {
         (|state: Res<State<T>>, pred: Local<Option<T>>| {
             state
                 .transition
@@ -112,11 +170,13 @@ impl<T: Component + Clone + Eq> State<T> {
                 })
         })
         .system()
-        .config(|(_, pred)| *pred = Some(Some(s)))
+        .config(|(_, pred)| *pred = Some(Some(s.clone())))
         .chain(should_run_adapter::<T>.system())
+        .after(DriverLabel::of::<T>())
+        .label_discard_if_duplicate(StateCallback::Enter.into_label(s))
     }
 
-    pub fn on_exit(s: T) -> impl System<In = (), Out = ShouldRun> {
+    pub fn on_exit(s: T) -> RunCriteriaDescriptor {
         (|state: Res<State<T>>, pred: Local<Option<T>>| {
             state
                 .transition
@@ -128,11 +188,13 @@ impl<T: Component + Clone + Eq> State<T> {
                 })
         })
         .system()
-        .config(|(_, pred)| *pred = Some(Some(s)))
+        .config(|(_, pred)| *pred = Some(Some(s.clone())))
         .chain(should_run_adapter::<T>.system())
+        .after(DriverLabel::of::<T>())
+        .label_discard_if_duplicate(StateCallback::Exit.into_label(s))
     }
 
-    pub fn on_pause(s: T) -> impl System<In = (), Out = ShouldRun> {
+    pub fn on_pause(s: T) -> RunCriteriaDescriptor {
         (|state: Res<State<T>>, pred: Local<Option<T>>| {
             state
                 .transition
@@ -143,11 +205,13 @@ impl<T: Component + Clone + Eq> State<T> {
                 })
         })
         .system()
-        .config(|(_, pred)| *pred = Some(Some(s)))
+        .config(|(_, pred)| *pred = Some(Some(s.clone())))
         .chain(should_run_adapter::<T>.system())
+        .after(DriverLabel::of::<T>())
+        .label_discard_if_duplicate(StateCallback::Pause.into_label(s))
     }
 
-    pub fn on_resume(s: T) -> impl System<In = (), Out = ShouldRun> {
+    pub fn on_resume(s: T) -> RunCriteriaDescriptor {
         (|state: Res<State<T>>, pred: Local<Option<T>>| {
             state
                 .transition
@@ -158,8 +222,10 @@ impl<T: Component + Clone + Eq> State<T> {
                 })
         })
         .system()
-        .config(|(_, pred)| *pred = Some(Some(s)))
+        .config(|(_, pred)| *pred = Some(Some(s.clone())))
         .chain(should_run_adapter::<T>.system())
+        .after(DriverLabel::of::<T>())
+        .label_discard_if_duplicate(StateCallback::Resume.into_label(s))
     }
 
     pub fn on_update_set(s: T) -> SystemSet {
@@ -191,7 +257,8 @@ impl<T: Component + Clone + Eq> State<T> {
     /// Important note: this set must be inserted **before** all other state-dependant sets to work
     /// properly!
     pub fn make_driver() -> SystemSet {
-        SystemSet::default().with_run_criteria(state_cleaner::<T>.system())
+        SystemSet::default()
+            .with_run_criteria(state_cleaner::<T>.system().label(DriverLabel::of::<T>()))
     }
 
     pub fn new(initial: T) -> Self {
@@ -390,7 +457,7 @@ mod test {
     use super::*;
     use crate::prelude::*;
 
-    #[derive(Copy, Clone, Eq, PartialEq, Debug)]
+    #[derive(Debug, Clone, Copy, Eq, PartialEq, Hash)]
     enum MyState {
         S1,
         S2,
diff --git a/crates/bevy_ecs/src/schedule/system_container.rs b/crates/bevy_ecs/src/schedule/system_container.rs
index f3d5be354abb7..2eef7f2a593ce 100644
--- a/crates/bevy_ecs/src/schedule/system_container.rs
+++ b/crates/bevy_ecs/src/schedule/system_container.rs
@@ -2,32 +2,33 @@ use crate::{
     component::ComponentId,
     query::Access,
     schedule::{
-        BoxedAmbiguitySetLabel, BoxedSystemLabel, ExclusiveSystemDescriptor,
-        ParallelSystemDescriptor,
+        BoxedAmbiguitySetLabel, BoxedRunCriteriaLabel, BoxedSystemLabel, ExclusiveSystemDescriptor,
+        GraphNode, ParallelSystemDescriptor,
     },
     system::{ExclusiveSystem, System},
 };
 use std::{borrow::Cow, ptr::NonNull};
 
 /// System metadata like its name, labels, order requirements and component access.
-pub trait SystemContainer {
-    fn name(&self) -> Cow<'static, str>;
+pub trait SystemContainer: GraphNode<BoxedSystemLabel> {
     #[doc(hidden)]
     fn dependencies(&self) -> &[usize];
     #[doc(hidden)]
     fn set_dependencies(&mut self, dependencies: impl IntoIterator<Item = usize>);
-    fn system_set(&self) -> usize;
-    fn labels(&self) -> &[BoxedSystemLabel];
-    fn before(&self) -> &[BoxedSystemLabel];
-    fn after(&self) -> &[BoxedSystemLabel];
+    #[doc(hidden)]
+    fn run_criteria(&self) -> Option<usize>;
+    #[doc(hidden)]
+    fn set_run_criteria(&mut self, index: usize);
+    fn run_criteria_label(&self) -> Option<&BoxedRunCriteriaLabel>;
     fn ambiguity_sets(&self) -> &[BoxedAmbiguitySetLabel];
     fn component_access(&self) -> Option<&Access<ComponentId>>;
 }
 
 pub(super) struct ExclusiveSystemContainer {
     system: Box<dyn ExclusiveSystem>,
+    pub(super) run_criteria_index: Option<usize>,
+    pub(super) run_criteria_label: Option<BoxedRunCriteriaLabel>,
     dependencies: Vec<usize>,
-    set: usize,
     labels: Vec<BoxedSystemLabel>,
     before: Vec<BoxedSystemLabel>,
     after: Vec<BoxedSystemLabel>,
@@ -35,11 +36,12 @@ pub(super) struct ExclusiveSystemContainer {
 }
 
 impl ExclusiveSystemContainer {
-    pub fn from_descriptor(descriptor: ExclusiveSystemDescriptor, set: usize) -> Self {
+    pub fn from_descriptor(descriptor: ExclusiveSystemDescriptor) -> Self {
         ExclusiveSystemContainer {
             system: descriptor.system,
+            run_criteria_index: None,
+            run_criteria_label: None,
             dependencies: Vec::new(),
-            set,
             labels: descriptor.labels,
             before: descriptor.before,
             after: descriptor.after,
@@ -52,11 +54,25 @@ impl ExclusiveSystemContainer {
     }
 }
 
-impl SystemContainer for ExclusiveSystemContainer {
+impl GraphNode<BoxedSystemLabel> for ExclusiveSystemContainer {
     fn name(&self) -> Cow<'static, str> {
         self.system.name()
     }
 
+    fn labels(&self) -> &[BoxedSystemLabel] {
+        &self.labels
+    }
+
+    fn before(&self) -> &[BoxedSystemLabel] {
+        &self.before
+    }
+
+    fn after(&self) -> &[BoxedSystemLabel] {
+        &self.after
+    }
+}
+
+impl SystemContainer for ExclusiveSystemContainer {
     fn dependencies(&self) -> &[usize] {
         &self.dependencies
     }
@@ -66,20 +82,16 @@ impl SystemContainer for ExclusiveSystemContainer {
         self.dependencies.extend(dependencies);
     }
 
-    fn system_set(&self) -> usize {
-        self.set
-    }
-
-    fn labels(&self) -> &[BoxedSystemLabel] {
-        &self.labels
+    fn run_criteria(&self) -> Option<usize> {
+        self.run_criteria_index
     }
 
-    fn before(&self) -> &[BoxedSystemLabel] {
-        &self.before
+    fn set_run_criteria(&mut self, index: usize) {
+        self.run_criteria_index = Some(index);
     }
 
-    fn after(&self) -> &[BoxedSystemLabel] {
-        &self.after
+    fn run_criteria_label(&self) -> Option<&BoxedRunCriteriaLabel> {
+        self.run_criteria_label.as_ref()
     }
 
     fn ambiguity_sets(&self) -> &[BoxedAmbiguitySetLabel] {
@@ -93,9 +105,10 @@ impl SystemContainer for ExclusiveSystemContainer {
 
 pub struct ParallelSystemContainer {
     system: NonNull<dyn System<In = (), Out = ()>>,
+    pub(crate) run_criteria_index: Option<usize>,
+    pub(crate) run_criteria_label: Option<BoxedRunCriteriaLabel>,
     pub(crate) should_run: bool,
     dependencies: Vec<usize>,
-    set: usize,
     labels: Vec<BoxedSystemLabel>,
     before: Vec<BoxedSystemLabel>,
     after: Vec<BoxedSystemLabel>,
@@ -106,11 +119,12 @@ unsafe impl Send for ParallelSystemContainer {}
 unsafe impl Sync for ParallelSystemContainer {}
 
 impl ParallelSystemContainer {
-    pub(crate) fn from_descriptor(descriptor: ParallelSystemDescriptor, set: usize) -> Self {
+    pub(crate) fn from_descriptor(descriptor: ParallelSystemDescriptor) -> Self {
         ParallelSystemContainer {
             system: unsafe { NonNull::new_unchecked(Box::into_raw(descriptor.system)) },
             should_run: false,
-            set,
+            run_criteria_index: None,
+            run_criteria_label: None,
             dependencies: Vec::new(),
             labels: descriptor.labels,
             before: descriptor.before,
@@ -120,7 +134,7 @@ impl ParallelSystemContainer {
     }
 
     pub fn name(&self) -> Cow<'static, str> {
-        SystemContainer::name(self)
+        GraphNode::name(self)
     }
 
     pub fn system(&self) -> &dyn System<In = (), Out = ()> {
@@ -149,11 +163,25 @@ impl ParallelSystemContainer {
     }
 }
 
-impl SystemContainer for ParallelSystemContainer {
+impl GraphNode<BoxedSystemLabel> for ParallelSystemContainer {
     fn name(&self) -> Cow<'static, str> {
         self.system().name()
     }
 
+    fn labels(&self) -> &[BoxedSystemLabel] {
+        &self.labels
+    }
+
+    fn before(&self) -> &[BoxedSystemLabel] {
+        &self.before
+    }
+
+    fn after(&self) -> &[BoxedSystemLabel] {
+        &self.after
+    }
+}
+
+impl SystemContainer for ParallelSystemContainer {
     fn dependencies(&self) -> &[usize] {
         &self.dependencies
     }
@@ -163,20 +191,16 @@ impl SystemContainer for ParallelSystemContainer {
         self.dependencies.extend(dependencies);
     }
 
-    fn system_set(&self) -> usize {
-        self.set
-    }
-
-    fn labels(&self) -> &[BoxedSystemLabel] {
-        &self.labels
+    fn run_criteria(&self) -> Option<usize> {
+        self.run_criteria_index
     }
 
-    fn before(&self) -> &[BoxedSystemLabel] {
-        &self.before
+    fn set_run_criteria(&mut self, index: usize) {
+        self.run_criteria_index = Some(index);
     }
 
-    fn after(&self) -> &[BoxedSystemLabel] {
-        &self.after
+    fn run_criteria_label(&self) -> Option<&BoxedRunCriteriaLabel> {
+        self.run_criteria_label.as_ref()
     }
 
     fn ambiguity_sets(&self) -> &[BoxedAmbiguitySetLabel] {
diff --git a/crates/bevy_ecs/src/schedule/system_descriptor.rs b/crates/bevy_ecs/src/schedule/system_descriptor.rs
index ce2adeda4512a..961826ffcf577 100644
--- a/crates/bevy_ecs/src/schedule/system_descriptor.rs
+++ b/crates/bevy_ecs/src/schedule/system_descriptor.rs
@@ -1,5 +1,8 @@
 use crate::{
-    schedule::{AmbiguitySetLabel, BoxedAmbiguitySetLabel, BoxedSystemLabel, SystemLabel},
+    schedule::{
+        AmbiguitySetLabel, BoxedAmbiguitySetLabel, BoxedSystemLabel, IntoRunCriteria,
+        RunCriteriaDescriptorOrLabel, SystemLabel,
+    },
     system::{BoxedSystem, ExclusiveSystem, ExclusiveSystemCoerced, ExclusiveSystemFn, System},
 };
 
@@ -77,9 +80,10 @@ impl From<ExclusiveSystemCoerced> for SystemDescriptor {
     }
 }
 
-/// Encapsulates a parallel system and information on when it run in a `SystemStage`.
+/// Encapsulates a parallel system and information on when it runs in a `SystemStage`.
 pub struct ParallelSystemDescriptor {
     pub(crate) system: BoxedSystem<(), ()>,
+    pub(crate) run_criteria: Option<RunCriteriaDescriptorOrLabel>,
     pub(crate) labels: Vec<BoxedSystemLabel>,
     pub(crate) before: Vec<BoxedSystemLabel>,
     pub(crate) after: Vec<BoxedSystemLabel>,
@@ -89,6 +93,7 @@ pub struct ParallelSystemDescriptor {
 fn new_parallel_descriptor(system: BoxedSystem<(), ()>) -> ParallelSystemDescriptor {
     ParallelSystemDescriptor {
         system,
+        run_criteria: None,
         labels: Vec::new(),
         before: Vec::new(),
         after: Vec::new(),
@@ -97,10 +102,17 @@ fn new_parallel_descriptor(system: BoxedSystem<(), ()>) -> ParallelSystemDescrip
 }
 
 pub trait ParallelSystemDescriptorCoercion {
+    /// Assigns a run criteria to the system. Can be a new descriptor or a label of a
+    /// run criteria defined elsewhere.
+    fn with_run_criteria<Marker>(
+        self,
+        run_criteria: impl IntoRunCriteria<Marker>,
+    ) -> ParallelSystemDescriptor;
+
     /// Assigns a label to the system; there can be more than one, and it doesn't have to be unique.
     fn label(self, label: impl SystemLabel) -> ParallelSystemDescriptor;
 
-    /// Specifies that the system should run before systems with the  given label.
+    /// Specifies that the system should run before systems with the given label.
     fn before(self, label: impl SystemLabel) -> ParallelSystemDescriptor;
 
     /// Specifies that the system should run after systems with the given label.
@@ -112,6 +124,14 @@ pub trait ParallelSystemDescriptorCoercion {
 }
 
 impl ParallelSystemDescriptorCoercion for ParallelSystemDescriptor {
+    fn with_run_criteria<Marker>(
+        mut self,
+        run_criteria: impl IntoRunCriteria<Marker>,
+    ) -> ParallelSystemDescriptor {
+        self.run_criteria = Some(run_criteria.into());
+        self
+    }
+
     fn label(mut self, label: impl SystemLabel) -> ParallelSystemDescriptor {
         self.labels.push(Box::new(label));
         self
@@ -137,6 +157,13 @@ impl<S> ParallelSystemDescriptorCoercion for S
 where
     S: System<In = (), Out = ()>,
 {
+    fn with_run_criteria<Marker>(
+        self,
+        run_criteria: impl IntoRunCriteria<Marker>,
+    ) -> ParallelSystemDescriptor {
+        new_parallel_descriptor(Box::new(self)).with_run_criteria(run_criteria)
+    }
+
     fn label(self, label: impl SystemLabel) -> ParallelSystemDescriptor {
         new_parallel_descriptor(Box::new(self)).label(label)
     }
@@ -155,6 +182,13 @@ where
 }
 
 impl ParallelSystemDescriptorCoercion for BoxedSystem<(), ()> {
+    fn with_run_criteria<Marker>(
+        self,
+        run_criteria: impl IntoRunCriteria<Marker>,
+    ) -> ParallelSystemDescriptor {
+        new_parallel_descriptor(self).with_run_criteria(run_criteria)
+    }
+
     fn label(self, label: impl SystemLabel) -> ParallelSystemDescriptor {
         new_parallel_descriptor(self).label(label)
     }
@@ -179,9 +213,10 @@ pub(crate) enum InsertionPoint {
     AtEnd,
 }
 
-/// Encapsulates an exclusive system and information on when it run in a `SystemStage`.
+/// Encapsulates an exclusive system and information on when it runs in a `SystemStage`.
 pub struct ExclusiveSystemDescriptor {
     pub(crate) system: Box<dyn ExclusiveSystem>,
+    pub(crate) run_criteria: Option<RunCriteriaDescriptorOrLabel>,
     pub(crate) labels: Vec<BoxedSystemLabel>,
     pub(crate) before: Vec<BoxedSystemLabel>,
     pub(crate) after: Vec<BoxedSystemLabel>,
@@ -192,6 +227,7 @@ pub struct ExclusiveSystemDescriptor {
 fn new_exclusive_descriptor(system: Box<dyn ExclusiveSystem>) -> ExclusiveSystemDescriptor {
     ExclusiveSystemDescriptor {
         system,
+        run_criteria: None,
         labels: Vec::new(),
         before: Vec::new(),
         after: Vec::new(),
@@ -201,6 +237,13 @@ fn new_exclusive_descriptor(system: Box<dyn ExclusiveSystem>) -> ExclusiveSystem
 }
 
 pub trait ExclusiveSystemDescriptorCoercion {
+    /// Assigns a run criteria to the system. Can be a new descriptor or a label of a
+    /// run criteria defined elsewhere.
+    fn with_run_criteria<Marker>(
+        self,
+        run_criteria: impl IntoRunCriteria<Marker>,
+    ) -> ExclusiveSystemDescriptor;
+
     /// Assigns a label to the system; there can be more than one, and it doesn't have to be unique.
     fn label(self, label: impl SystemLabel) -> ExclusiveSystemDescriptor;
 
@@ -226,6 +269,14 @@ pub trait ExclusiveSystemDescriptorCoercion {
 }
 
 impl ExclusiveSystemDescriptorCoercion for ExclusiveSystemDescriptor {
+    fn with_run_criteria<Marker>(
+        mut self,
+        run_criteria: impl IntoRunCriteria<Marker>,
+    ) -> ExclusiveSystemDescriptor {
+        self.run_criteria = Some(run_criteria.into());
+        self
+    }
+
     fn label(mut self, label: impl SystemLabel) -> ExclusiveSystemDescriptor {
         self.labels.push(Box::new(label));
         self
@@ -266,6 +317,13 @@ impl<T> ExclusiveSystemDescriptorCoercion for T
 where
     T: ExclusiveSystem + 'static,
 {
+    fn with_run_criteria<Marker>(
+        self,
+        run_criteria: impl IntoRunCriteria<Marker>,
+    ) -> ExclusiveSystemDescriptor {
+        new_exclusive_descriptor(Box::new(self)).with_run_criteria(run_criteria)
+    }
+
     fn label(self, label: impl SystemLabel) -> ExclusiveSystemDescriptor {
         new_exclusive_descriptor(Box::new(self)).label(label)
     }
diff --git a/crates/bevy_ecs/src/schedule/system_set.rs b/crates/bevy_ecs/src/schedule/system_set.rs
index cd46bfe01d5d2..b8084ca545846 100644
--- a/crates/bevy_ecs/src/schedule/system_set.rs
+++ b/crates/bevy_ecs/src/schedule/system_set.rs
@@ -1,20 +1,31 @@
 use crate::{
     component::Component,
-    schedule::{RunCriteria, ShouldRun, State, SystemDescriptor},
-    system::System,
+    schedule::{
+        AmbiguitySetLabel, BoxedAmbiguitySetLabel, BoxedSystemLabel, IntoRunCriteria,
+        RunCriteriaDescriptorOrLabel, State, SystemDescriptor, SystemLabel,
+    },
 };
+use std::{fmt::Debug, hash::Hash};
 
-/// Describes a group of systems sharing one run criterion.
+/// A builder for describing several systems at the same time.
 pub struct SystemSet {
-    pub(crate) run_criteria: RunCriteria,
-    pub(crate) descriptors: Vec<SystemDescriptor>,
+    pub(crate) systems: Vec<SystemDescriptor>,
+    pub(crate) run_criteria: Option<RunCriteriaDescriptorOrLabel>,
+    pub(crate) labels: Vec<BoxedSystemLabel>,
+    pub(crate) before: Vec<BoxedSystemLabel>,
+    pub(crate) after: Vec<BoxedSystemLabel>,
+    pub(crate) ambiguity_sets: Vec<BoxedAmbiguitySetLabel>,
 }
 
 impl Default for SystemSet {
     fn default() -> SystemSet {
         SystemSet {
-            run_criteria: Default::default(),
-            descriptors: vec![],
+            systems: Vec::new(),
+            run_criteria: None,
+            labels: Vec::new(),
+            before: Vec::new(),
+            after: Vec::new(),
+            ambiguity_sets: Vec::new(),
         }
     }
 }
@@ -24,50 +35,107 @@ impl SystemSet {
         Default::default()
     }
 
-    pub fn with_run_criteria<S: System<In = (), Out = ShouldRun>>(mut self, system: S) -> Self {
-        self.add_run_criteria(system);
-        self
+    pub fn on_update<T>(s: T) -> SystemSet
+    where
+        T: Component + Debug + Clone + Eq + Hash,
+    {
+        Self::new().with_run_criteria(State::<T>::on_update(s))
     }
 
-    pub fn add_run_criteria<S: System<In = (), Out = ShouldRun>>(
-        &mut self,
-        system: S,
-    ) -> &mut Self {
-        self.run_criteria.set(Box::new(system));
-        self
+    pub fn on_inactive_update<T>(s: T) -> SystemSet
+    where
+        T: Component + Debug + Clone + Eq + Hash,
+    {
+        Self::new().with_run_criteria(State::<T>::on_inactive_update(s))
     }
 
-    pub fn with_system(mut self, system: impl Into<SystemDescriptor>) -> Self {
-        self.add_system(system);
-        self
+    pub fn on_enter<T>(s: T) -> SystemSet
+    where
+        T: Component + Debug + Clone + Eq + Hash,
+    {
+        Self::new().with_run_criteria(State::<T>::on_enter(s))
     }
 
-    pub fn add_system(&mut self, system: impl Into<SystemDescriptor>) -> &mut Self {
-        self.descriptors.push(system.into());
+    pub fn on_exit<T>(s: T) -> SystemSet
+    where
+        T: Component + Debug + Clone + Eq + Hash,
+    {
+        Self::new().with_run_criteria(State::<T>::on_exit(s))
+    }
+
+    pub fn on_pause<T>(s: T) -> SystemSet
+    where
+        T: Component + Debug + Clone + Eq + Hash,
+    {
+        Self::new().with_run_criteria(State::<T>::on_pause(s))
+    }
+
+    pub fn on_resume<T>(s: T) -> SystemSet
+    where
+        T: Component + Debug + Clone + Eq + Hash,
+    {
+        Self::new().with_run_criteria(State::<T>::on_resume(s))
+    }
+
+    pub fn in_ambiguity_set(mut self, set: impl AmbiguitySetLabel) -> Self {
+        self.ambiguity_sets.push(Box::new(set));
         self
     }
 
-    pub fn on_update<T: Component + Clone + Eq>(s: T) -> SystemSet {
-        Self::new().with_run_criteria(State::<T>::on_update(s))
+    pub fn with_system(mut self, system: impl Into<SystemDescriptor>) -> Self {
+        self.systems.push(system.into());
+        self
     }
 
-    pub fn on_inactive_update<T: Component + Clone + Eq>(s: T) -> SystemSet {
-        Self::new().with_run_criteria(State::<T>::on_inactive_update(s))
+    pub fn with_run_criteria<Marker>(mut self, run_criteria: impl IntoRunCriteria<Marker>) -> Self {
+        self.run_criteria = Some(run_criteria.into());
+        self
     }
 
-    pub fn on_enter<T: Component + Clone + Eq>(s: T) -> SystemSet {
-        Self::new().with_run_criteria(State::<T>::on_enter(s))
+    pub fn label(mut self, label: impl SystemLabel) -> Self {
+        self.labels.push(Box::new(label));
+        self
     }
 
-    pub fn on_exit<T: Component + Clone + Eq>(s: T) -> SystemSet {
-        Self::new().with_run_criteria(State::<T>::on_exit(s))
+    pub fn before(mut self, label: impl SystemLabel) -> Self {
+        self.before.push(Box::new(label));
+        self
     }
 
-    pub fn on_pause<T: Component + Clone + Eq>(s: T) -> SystemSet {
-        Self::new().with_run_criteria(State::<T>::on_pause(s))
+    pub fn after(mut self, label: impl SystemLabel) -> Self {
+        self.after.push(Box::new(label));
+        self
     }
 
-    pub fn on_resume<T: Component + Clone + Eq>(s: T) -> SystemSet {
-        Self::new().with_run_criteria(State::<T>::on_resume(s))
+    pub(crate) fn bake(self) -> (Option<RunCriteriaDescriptorOrLabel>, Vec<SystemDescriptor>) {
+        let SystemSet {
+            mut systems,
+            run_criteria,
+            labels,
+            before,
+            after,
+            ambiguity_sets,
+        } = self;
+        for descriptor in &mut systems {
+            match descriptor {
+                SystemDescriptor::Parallel(descriptor) => {
+                    descriptor.labels.extend(labels.iter().cloned());
+                    descriptor.before.extend(before.iter().cloned());
+                    descriptor.after.extend(after.iter().cloned());
+                    descriptor
+                        .ambiguity_sets
+                        .extend(ambiguity_sets.iter().cloned());
+                }
+                SystemDescriptor::Exclusive(descriptor) => {
+                    descriptor.labels.extend(labels.iter().cloned());
+                    descriptor.before.extend(before.iter().cloned());
+                    descriptor.after.extend(after.iter().cloned());
+                    descriptor
+                        .ambiguity_sets
+                        .extend(ambiguity_sets.iter().cloned());
+                }
+            }
+        }
+        (run_criteria, systems)
     }
 }
diff --git a/examples/2d/texture_atlas.rs b/examples/2d/texture_atlas.rs
index d1682ae30e398..17951afcc4ee7 100644
--- a/examples/2d/texture_atlas.rs
+++ b/examples/2d/texture_atlas.rs
@@ -13,7 +13,7 @@ fn main() {
         .run();
 }
 
-#[derive(Clone, PartialEq, Eq)]
+#[derive(Debug, Clone, PartialEq, Eq, Hash)]
 enum AppState {
     Setup,
     Finished,
diff --git a/examples/ecs/state.rs b/examples/ecs/state.rs
index 5cfdbdd96ab0a..8f69028cac183 100644
--- a/examples/ecs/state.rs
+++ b/examples/ecs/state.rs
@@ -19,7 +19,7 @@ fn main() {
         .run();
 }
 
-#[derive(Clone, Eq, PartialEq)]
+#[derive(Debug, Clone, Eq, PartialEq, Hash)]
 enum AppState {
     Menu,
     InGame,
diff --git a/examples/game/alien_cake_addict.rs b/examples/game/alien_cake_addict.rs
index 35b71fc60e29b..9d0a20993f229 100644
--- a/examples/game/alien_cake_addict.rs
+++ b/examples/game/alien_cake_addict.rs
@@ -6,7 +6,7 @@ use bevy::{
 };
 use rand::Rng;
 
-#[derive(Clone, Eq, PartialEq, Debug)]
+#[derive(Clone, Eq, PartialEq, Debug, Hash)]
 enum GameState {
     Playing,
     GameOver,
diff --git a/examples/window/multiple_windows.rs b/examples/window/multiple_windows.rs
index e6d1c124ea7a5..dd35696ed7ddb 100644
--- a/examples/window/multiple_windows.rs
+++ b/examples/window/multiple_windows.rs
@@ -27,7 +27,7 @@ fn main() {
 
 // NOTE: this "state based" approach to multiple windows is a short term workaround.
 // Future Bevy releases shouldn't require such a strict order of operations.
-#[derive(Clone, Eq, PartialEq)]
+#[derive(Debug, Clone, Eq, PartialEq, Hash)]
 enum AppState {
     CreateWindow,
     Setup,