forestia/bevy
2
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

System sets and parallel executor v2 (#1144)

Browse Source 
System sets and parallel executor v2
This commit is contained in:
Alexander Sepity 2021-02-09 23:14:10 +03:00 committed by GitHub
parent b71ada0346
commit d5a7330431
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
45 changed files with 3228 additions and 1417 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
crates/bevy_app/src/app.rs
View File

@ -1,5 +1,5 @@
use crate::app_builder::AppBuilder;
use bevy_ecs::{Resources, Schedule, World};
use bevy_ecs::{Resources, Schedule, Stage, World};
#[cfg(feature = "trace")]
use bevy_utils::tracing::info_span;
@ -53,8 +53,7 @@ impl App {
}
pub fn update(&mut self) {
self.schedule
.initialize_and_run(&mut self.world, &mut self.resources);
self.schedule.run(&mut self.world, &mut self.resources);
}
pub fn run(mut self) {

112
crates/bevy_app/src/app_builder.rs
View File

@ -5,8 +5,8 @@ use crate::{
stage, startup_stage, PluginGroup, PluginGroupBuilder,
};
use bevy_ecs::{
clear_trackers_system, FromResources, IntoSystem, Resource, Resources, RunOnce, Schedule,
Stage, StateStage, System, SystemStage, World,
clear_trackers_system, FromResources, IntoExclusiveSystem, IntoSystem, Resource, Resources,
RunOnce, Schedule, Stage, StateStage, SystemDescriptor, SystemStage, World,
};
use bevy_utils::tracing::debug;
@ -24,7 +24,7 @@ impl Default for AppBuilder {
app_builder
.add_default_stages()
.add_event::<AppExit>()
.add_system_to_stage(stage::LAST, clear_trackers_system.system());
.add_system_to_stage(stage::LAST, clear_trackers_system.exclusive_system());
app_builder
}
}
@ -125,47 +125,27 @@ impl AppBuilder {
self
}
pub fn add_system<S: System<In = (), Out = ()>>(&mut self, system: S) -> &mut Self {
pub fn add_system(&mut self, system: impl Into<SystemDescriptor>) -> &mut Self {
self.add_system_to_stage(stage::UPDATE, system)
}
pub fn on_state_enter<T: Clone + Resource, S: System<In = (), Out = ()>>(
&mut self,
stage: &str,
state: T,
system: S,
) -> &mut Self {
self.stage(stage, |stage: &mut StateStage<T>| {
stage.on_state_enter(state, system)
})
}
pub fn on_state_update<T: Clone + Resource, S: System<In = (), Out = ()>>(
&mut self,
stage: &str,
state: T,
system: S,
) -> &mut Self {
self.stage(stage, |stage: &mut StateStage<T>| {
stage.on_state_update(state, system)
})
}
pub fn on_state_exit<T: Clone + Resource, S: System<In = (), Out = ()>>(
&mut self,
stage: &str,
state: T,
system: S,
) -> &mut Self {
self.stage(stage, |stage: &mut StateStage<T>| {
stage.on_state_exit(state, system)
})
}
pub fn add_startup_system_to_stage<S: System<In = (), Out = ()>>(
pub fn add_system_to_stage(
&mut self,
stage_name: &'static str,
system: S,
system: impl Into<SystemDescriptor>,
) -> &mut Self {
self.app.schedule.add_system_to_stage(stage_name, system);
self
}
pub fn add_startup_system(&mut self, system: impl Into<SystemDescriptor>) -> &mut Self {
self.add_startup_system_to_stage(startup_stage::STARTUP, system)
}
pub fn add_startup_system_to_stage(
&mut self,
stage_name: &'static str,
system: impl Into<SystemDescriptor>,
) -> &mut Self {
self.app
.schedule
@ -175,8 +155,37 @@ impl AppBuilder {
self
}
pub fn add_startup_system<S: System<In = (), Out = ()>>(&mut self, system: S) -> &mut Self {
self.add_startup_system_to_stage(startup_stage::STARTUP, system)
pub fn on_state_enter<T: Clone + Resource>(
&mut self,
stage: &str,
state: T,
system: impl Into<SystemDescriptor>,
) -> &mut Self {
self.stage(stage, |stage: &mut StateStage<T>| {
stage.on_state_enter(state, system)
})
}
pub fn on_state_update<T: Clone + Resource>(
&mut self,
stage: &str,
state: T,
system: impl Into<SystemDescriptor>,
) -> &mut Self {
self.stage(stage, |stage: &mut StateStage<T>| {
stage.on_state_update(state, system)
})
}
pub fn on_state_exit<T: Clone + Resource>(
&mut self,
stage: &str,
state: T,
system: impl Into<SystemDescriptor>,
) -> &mut Self {
self.stage(stage, |stage: &mut StateStage<T>| {
stage.on_state_exit(state, system)
})
}
pub fn add_default_stages(&mut self) -> &mut Self {
@ -197,15 +206,6 @@ impl AppBuilder {
.add_stage(stage::LAST, SystemStage::parallel())
}
pub fn add_system_to_stage<S: System<In = (), Out = ()>>(
&mut self,
stage_name: &'static str,
system: S,
) -> &mut Self {
self.app.schedule.add_system_to_stage(stage_name, system);
self
}
pub fn add_event<T>(&mut self) -> &mut Self
where
T: Send + Sync + 'static,
@ -223,11 +223,11 @@ impl AppBuilder {
self
}
pub fn insert_thread_local_resource<T>(&mut self, resource: T) -> &mut Self
pub fn insert_non_send_resource<T>(&mut self, resource: T) -> &mut Self
where
T: 'static,
{
self.app.resources.insert_thread_local(resource);
self.app.resources.insert_non_send(resource);
self
}
@ -242,20 +242,18 @@ impl AppBuilder {
let resource = R::from_resources(&self.resources());
self.insert_resource(resource);
}
self
}
pub fn init_thread_local_resource<R>(&mut self) -> &mut Self
pub fn init_non_send_resource<R>(&mut self) -> &mut Self
where
R: FromResources + 'static,
{
// See perf comment in init_resource
if self.app.resources.get_thread_local::<R>().is_none() {
if self.app.resources.get_non_send::<R>().is_none() {
let resource = R::from_resources(&self.app.resources);
self.app.resources.insert_thread_local(resource);
self.app.resources.insert_non_send(resource);
}
self
}

2
crates/bevy_audio/src/audio_output.rs
View File

@ -65,7 +65,7 @@ where
<P as Decodable>::Decoder: rodio::Source + Send + Sync,
<<P as Decodable>::Decoder as Iterator>::Item: rodio::Sample + Send + Sync,
{
let audio_output = resources.get_thread_local::<AudioOutput<P>>().unwrap();
let audio_output = resources.get_non_send::<AudioOutput<P>>().unwrap();
let mut audio = resources.get_mut::<Audio<P>>().unwrap();
if let Some(audio_sources) = resources.get::<Assets<P>>() {

6
crates/bevy_audio/src/lib.rs
View File

@ -12,7 +12,7 @@ pub mod prelude {
use bevy_app::prelude::*;
use bevy_asset::AddAsset;
use bevy_ecs::IntoSystem;
use bevy_ecs::IntoExclusiveSystem;
/// Adds support for audio playback to an App
#[derive(Default)]
@ -20,13 +20,13 @@ pub struct AudioPlugin;
impl Plugin for AudioPlugin {
fn build(&self, app: &mut AppBuilder) {
app.init_thread_local_resource::<AudioOutput<AudioSource>>()
app.init_non_send_resource::<AudioOutput<AudioSource>>()
.add_asset::<AudioSource>()
.init_asset_loader::<Mp3Loader>()
.init_resource::<Audio<AudioSource>>()
.add_system_to_stage(
stage::POST_UPDATE,
play_queued_audio_system::<AudioSource>.system(),
play_queued_audio_system::<AudioSource>.exclusive_system(),
);
}
}

31
crates/bevy_core/src/label.rs
View File

@ -115,13 +115,14 @@ pub(crate) fn entity_labels_system(
#[cfg(test)]
mod tests {
use super::*;
use bevy_ecs::Stage;
fn setup() -> (World, Resources, bevy_ecs::Schedule) {
let world = World::new();
let mut resources = Resources::default();
resources.insert(EntityLabels::default());
let mut schedule = bevy_ecs::Schedule::default();
schedule.add_stage("test", SystemStage::serial());
schedule.add_stage("test", SystemStage::single_threaded());
schedule.add_system_to_stage("test", entity_labels_system.system());
(world, resources, schedule)
}
@ -139,7 +140,7 @@ mod tests {
let (mut world, mut resources, mut schedule) = setup();
let e1 = world.spawn((holy_cow(),));
schedule.initialize_and_run(&mut world, &mut resources);
schedule.run(&mut world, &mut resources);
let entity_labels = resources.get::<EntityLabels>().unwrap();
assert_eq!(entity_labels.get("holy"), &[e1], "holy");
@ -151,10 +152,10 @@ mod tests {
fn add_labels() {
let (mut world, mut resources, mut schedule) = setup();
let e1 = world.spawn((holy_cow(),));
schedule.initialize_and_run(&mut world, &mut resources);
schedule.run(&mut world, &mut resources);
world.get_mut::<Labels>(e1).unwrap().insert("shalau");
schedule.initialize_and_run(&mut world, &mut resources);
schedule.run(&mut world, &mut resources);
let entity_labels = resources.get::<EntityLabels>().unwrap();
assert_eq!(entity_labels.get("holy"), &[e1], "holy");
@ -166,10 +167,10 @@ mod tests {
fn remove_labels() {
let (mut world, mut resources, mut schedule) = setup();
let e1 = world.spawn((holy_cow(),));
schedule.initialize_and_run(&mut world, &mut resources);
schedule.run(&mut world, &mut resources);
world.get_mut::<Labels>(e1).unwrap().remove("holy");
schedule.initialize_and_run(&mut world, &mut resources);
schedule.run(&mut world, &mut resources);
let entity_labels = resources.get::<EntityLabels>().unwrap();
assert_eq!(entity_labels.get("holy"), &[], "holy");
@ -181,10 +182,10 @@ mod tests {
fn removes_despawned_entity() {
let (mut world, mut resources, mut schedule) = setup();
let e1 = world.spawn((holy_cow(),));
schedule.initialize_and_run(&mut world, &mut resources);
schedule.run(&mut world, &mut resources);
world.despawn(e1).unwrap();
schedule.initialize_and_run(&mut world, &mut resources);
schedule.run(&mut world, &mut resources);
let entity_labels = resources.get::<EntityLabels>().unwrap();
assert_eq!(entity_labels.get("holy"), &[], "holy");
@ -196,10 +197,10 @@ mod tests {
fn removes_labels_when_component_removed() {
let (mut world, mut resources, mut schedule) = setup();
let e1 = world.spawn((holy_cow(),));
schedule.initialize_and_run(&mut world, &mut resources);
schedule.run(&mut world, &mut resources);
world.remove_one::<Labels>(e1).unwrap();
schedule.initialize_and_run(&mut world, &mut resources);
schedule.run(&mut world, &mut resources);
let entity_labels = resources.get::<EntityLabels>().unwrap();
assert_eq!(entity_labels.get("holy"), &[], "holy");
@ -211,10 +212,10 @@ mod tests {
fn adds_another_spawned_entity() {
let (mut world, mut resources, mut schedule) = setup();
let e1 = world.spawn((holy_cow(),));
schedule.initialize_and_run(&mut world, &mut resources);
schedule.run(&mut world, &mut resources);
let e2 = world.spawn((holy_shamoni(),));
schedule.initialize_and_run(&mut world, &mut resources);
schedule.run(&mut world, &mut resources);
let entity_labels = resources.get::<EntityLabels>().unwrap();
assert_eq!(entity_labels.get("holy"), &[e1, e2], "holy");
@ -227,13 +228,13 @@ mod tests {
fn removes_despawned_entity_but_leaves_other() {
let (mut world, mut resources, mut schedule) = setup();
let e1 = world.spawn((holy_cow(),));
schedule.initialize_and_run(&mut world, &mut resources);
schedule.run(&mut world, &mut resources);
let e2 = world.spawn((holy_shamoni(),));
schedule.initialize_and_run(&mut world, &mut resources);
schedule.run(&mut world, &mut resources);
world.despawn(e1).unwrap();
schedule.initialize_and_run(&mut world, &mut resources);
schedule.run(&mut world, &mut resources);
let entity_labels = resources.get::<EntityLabels>().unwrap();
assert_eq!(entity_labels.get("holy"), &[e2], "holy");

22
crates/bevy_core/src/time/fixed_timestep.rs
View File

@ -1,5 +1,5 @@
use crate::Time;
use bevy_ecs::{ArchetypeComponent, ShouldRun, System, SystemId, ThreadLocalExecution, TypeAccess};
use bevy_ecs::{ArchetypeComponent, ShouldRun, System, SystemId, TypeAccess};
use bevy_utils::HashMap;
use std::{any::TypeId, borrow::Cow};
@ -47,8 +47,9 @@ pub struct FixedTimestep {
looping: bool,
system_id: SystemId,
label: Option<String>, // TODO: consider making this a TypedLabel
resource_access: TypeAccess<TypeId>,
archetype_access: TypeAccess<ArchetypeComponent>,
component_access: TypeAccess<TypeId>,
resource_access: TypeAccess<TypeId>,
}
impl Default for FixedTimestep {
@ -59,8 +60,9 @@ impl Default for FixedTimestep {
accumulator: 0.0,
looping: false,
label: None,
resource_access: Default::default(),
component_access: Default::default(),
archetype_access: Default::default(),
resource_access: Default::default(),
}
}
}
@ -93,7 +95,7 @@ impl FixedTimestep {
if self.accumulator >= self.step {
self.accumulator -= self.step;
self.looping = true;
ShouldRun::YesAndLoop
ShouldRun::YesAndCheckAgain
} else {
self.looping = false;
ShouldRun::No
@ -113,18 +115,22 @@ impl System for FixedTimestep {
self.system_id
}
fn update(&mut self, _world: &bevy_ecs::World) {}
fn update_access(&mut self, _world: &bevy_ecs::World) {}
fn archetype_component_access(&self) -> &TypeAccess<ArchetypeComponent> {
&self.archetype_access
}
fn component_access(&self) -> &TypeAccess<TypeId> {
&self.component_access
}
fn resource_access(&self) -> &TypeAccess<TypeId> {
&self.resource_access
}
fn thread_local_execution(&self) -> ThreadLocalExecution {
ThreadLocalExecution::Immediate
fn is_non_send(&self) -> bool {
false
}
unsafe fn run_unsafe(
@ -145,7 +151,7 @@ impl System for FixedTimestep {
Some(result)
}
fn run_thread_local(
fn apply_buffers(
&mut self,
_world: &mut bevy_ecs::World,
_resources: &mut bevy_ecs::Resources,

4
crates/bevy_diagnostic/src/entity_count_diagnostics_plugin.rs
View File

@ -1,6 +1,6 @@
use crate::{Diagnostic, DiagnosticId, Diagnostics};
use bevy_app::prelude::*;
use bevy_ecs::{IntoSystem, ResMut, Resources, World};
use bevy_ecs::{IntoExclusiveSystem, IntoSystem, ResMut, Resources, World};
/// Adds "entity count" diagnostic to an App
#[derive(Default)]
@ -9,7 +9,7 @@ pub struct EntityCountDiagnosticsPlugin;
impl Plugin for EntityCountDiagnosticsPlugin {
fn build(&self, app: &mut AppBuilder) {
app.add_startup_system(Self::setup_system.system())
.add_system(Self::diagnostic_system.system());
.add_system(Self::diagnostic_system.exclusive_system());
}
}

6
crates/bevy_ecs/Cargo.toml
View File

@ -21,6 +21,7 @@ bevy_tasks = { path = "../bevy_tasks", version = "0.4.0" }
bevy_utils = { path = "../bevy_utils", version = "0.4.0" }
bevy_ecs_macros = { path = "macros", version = "0.4.0" }
fxhash = "0.2"
async-channel = "1.4.2"
rand = "0.8.0"
serde = "1.0"
thiserror = "1.0"
@ -32,8 +33,13 @@ lazy_static = { version = "1.4.0" }
[dev-dependencies]
bencher = "0.1.5"
criterion = "0.3"
[[bench]]
name = "bench"
harness = false
required-features = ["macros"]
[[bench]]
name = "system_stage"
harness = false

172
crates/bevy_ecs/benches/system_stage.rs Normal file
View File

@ -0,0 +1,172 @@
use bevy_ecs::{prelude::*, Stage};
use criterion::{criterion_group, criterion_main, Criterion};
criterion_group!(benches, empty_systems, busy_systems, contrived);
criterion_main!(benches);
fn run_stage(stage: &mut SystemStage, world: &mut World, resources: &mut Resources) {
// !!NB!! Uncomment next line when running with old executor.
//stage.initialize(world, resources);
stage.run(world, resources);
}
struct A(f32);
struct B(f32);
struct C(f32);
struct D(f32);
struct E(f32);
const ENTITY_BUNCH: usize = 5000;
fn empty_systems(criterion: &mut Criterion) {
let mut world = World::new();
let mut resources = Resources::default();
let mut group = criterion.benchmark_group("empty_systems");
group.warm_up_time(std::time::Duration::from_millis(500));
group.measurement_time(std::time::Duration::from_secs(3));
fn empty() {}
for amount in 0..5 {
let mut stage = SystemStage::parallel();
for _ in 0..amount {
stage.add_system(empty.system());
}
run_stage(&mut stage, &mut world, &mut resources);
group.bench_function(&format!("{:03}_systems", amount), |bencher| {
bencher.iter(|| {
run_stage(&mut stage, &mut world, &mut resources);
});
});
}
for amount in 1..21 {
let mut stage = SystemStage::parallel();
for _ in 0..amount {
stage
.add_system(empty.system())
.add_system(empty.system())
.add_system(empty.system())
.add_system(empty.system())
.add_system(empty.system());
}
run_stage(&mut stage, &mut world, &mut resources);
group.bench_function(&format!("{:03}_systems", 5 * amount), |bencher| {
bencher.iter(|| {
run_stage(&mut stage, &mut world, &mut resources);
});
});
}
group.finish()
}
fn busy_systems(criterion: &mut Criterion) {
fn ab(mut q: Query<(&mut A, &mut B)>) {
for (mut a, mut b) in q.iter_mut() {
std::mem::swap(&mut a.0, &mut b.0);
}
}
fn cd(mut q: Query<(&mut C, &mut D)>) {
for (mut c, mut d) in q.iter_mut() {
std::mem::swap(&mut c.0, &mut d.0);
}
}
fn ce(mut q: Query<(&mut C, &mut E)>) {
for (mut c, mut e) in q.iter_mut() {
std::mem::swap(&mut c.0, &mut e.0);
}
}
let mut world = World::new();
let mut resources = Resources::default();
let mut group = criterion.benchmark_group("busy_systems");
group.warm_up_time(std::time::Duration::from_millis(500));
group.measurement_time(std::time::Duration::from_secs(3));
for entity_bunches in 1..6 {
world.spawn_batch((0..4 * ENTITY_BUNCH).map(|_| (A(0.0), B(0.0))));
world.spawn_batch((0..4 * ENTITY_BUNCH).map(|_| (A(0.0), B(0.0), C(0.0))));
world.spawn_batch((0..ENTITY_BUNCH).map(|_| (A(0.0), B(0.0), C(0.0), D(0.0))));
world.spawn_batch((0..ENTITY_BUNCH).map(|_| (A(0.0), B(0.0), C(0.0), E(0.0))));
for system_amount in 0..5 {
let mut stage = SystemStage::parallel();
stage
.add_system(ab.system())
.add_system(cd.system())
.add_system(ce.system());
for _ in 0..system_amount {
stage
.add_system(ab.system())
.add_system(cd.system())
.add_system(ce.system());
}
run_stage(&mut stage, &mut world, &mut resources);
group.bench_function(
&format!(
"{:02}x_entities_{:02}_systems",
entity_bunches,
3 * system_amount + 3
),
|bencher| {
bencher.iter(|| {
run_stage(&mut stage, &mut world, &mut resources);
});
},
);
}
}
group.finish()
}
fn contrived(criterion: &mut Criterion) {
fn s_0(mut q_0: Query<(&mut A, &mut B)>) {
for (mut c_0, mut c_1) in q_0.iter_mut() {
std::mem::swap(&mut c_0.0, &mut c_1.0);
}
}
fn s_1(mut q_0: Query<(&mut A, &mut C)>, mut q_1: Query<(&mut B, &mut D)>) {
for (mut c_0, mut c_1) in q_0.iter_mut() {
std::mem::swap(&mut c_0.0, &mut c_1.0);
}
for (mut c_0, mut c_1) in q_1.iter_mut() {
std::mem::swap(&mut c_0.0, &mut c_1.0);
}
}
fn s_2(mut q_0: Query<(&mut C, &mut D)>) {
for (mut c_0, mut c_1) in q_0.iter_mut() {
std::mem::swap(&mut c_0.0, &mut c_1.0);
}
}
let mut world = World::new();
let mut resources = Resources::default();
let mut group = criterion.benchmark_group("contrived");
group.warm_up_time(std::time::Duration::from_millis(500));
group.measurement_time(std::time::Duration::from_secs(3));
for entity_bunches in 1..6 {
world.spawn_batch((0..ENTITY_BUNCH).map(|_| (A(0.0), B(0.0), C(0.0), D(0.0))));
world.spawn_batch((0..ENTITY_BUNCH).map(|_| (A(0.0), B(0.0))));
world.spawn_batch((0..ENTITY_BUNCH).map(|_| (C(0.0), D(0.0))));
for system_amount in 0..5 {
let mut stage = SystemStage::parallel();
stage
.add_system(s_0.system())
.add_system(s_1.system())
.add_system(s_2.system());
for _ in 0..system_amount {
stage
.add_system(s_0.system())
.add_system(s_1.system())
.add_system(s_2.system());
}
run_stage(&mut stage, &mut world, &mut resources);
group.bench_function(
&format!(
"{:02}x_entities_{:02}_systems",
entity_bunches,
3 * system_amount + 3
),
|bencher| {
bencher.iter(|| {
run_stage(&mut stage, &mut world, &mut resources);
});
},
);
}
}
group.finish()
}

161
crates/bevy_ecs/src/core/access.rs
View File

@ -1,10 +1,11 @@
use bevy_utils::HashSet;
use fixedbitset::FixedBitSet;
use std::{any::TypeId, boxed::Box, hash::Hash, vec::Vec};
use super::{Archetype, World};
#[derive(Debug, Copy, Clone, Ord, PartialOrd, Eq, PartialEq)]
pub enum Access {
enum ArchetypeAccess {
None,
Read,
Write,
@ -81,6 +82,22 @@ impl QueryAccess {
}
}
pub fn get_component_access(&self, type_access: &mut TypeAccess<TypeId>) {
match self {
QueryAccess::None => {}
QueryAccess::Read(ty, _) => type_access.add_read(*ty),
QueryAccess::Write(ty, _) => type_access.add_write(*ty),
QueryAccess::Optional(access) => access.get_component_access(type_access),
QueryAccess::With(_, access) => access.get_component_access(type_access),
QueryAccess::Without(_, access) => access.get_component_access(type_access),
QueryAccess::Union(accesses) => {
for access in accesses {
access.get_component_access(type_access);
}
}
}
}
pub fn get_type_name(&self, type_id: TypeId) -> Option<&'static str> {
match self {
QueryAccess::None => None,
@ -115,20 +132,20 @@ impl QueryAccess {
/// Returns how this [QueryAccess] accesses the given `archetype`.
/// If `type_access` is set, it will populate type access with the types this query reads/writes
pub fn get_access(
fn get_access(
&self,
archetype: &Archetype,
archetype_index: u32,
type_access: Option<&mut TypeAccess<ArchetypeComponent>>,
) -> Option<Access> {
) -> Option<ArchetypeAccess> {
match self {
QueryAccess::None => Some(Access::None),
QueryAccess::None => Some(ArchetypeAccess::None),
QueryAccess::Read(ty, _) => {
if archetype.has_type(*ty) {
if let Some(type_access) = type_access {
type_access.add_read(ArchetypeComponent::new_ty(archetype_index, *ty));
}
Some(Access::Read)
Some(ArchetypeAccess::Read)
} else {
None
}
@ -138,7 +155,7 @@ impl QueryAccess {
if let Some(type_access) = type_access {
type_access.add_write(ArchetypeComponent::new_ty(archetype_index, *ty));
}
Some(Access::Write)
Some(ArchetypeAccess::Write)
} else {
None
}
@ -152,7 +169,7 @@ impl QueryAccess {
Some(access)
}
} else {
Some(Access::Read)
Some(ArchetypeAccess::Read)
}
}
QueryAccess::With(ty, query_access) => {
@ -174,7 +191,7 @@ impl QueryAccess {
for query_access in query_accesses {
if let Some(access) = query_access.get_access(archetype, archetype_index, None)
{
result = Some(result.unwrap_or(Access::Read).max(access));
result = Some(result.unwrap_or(ArchetypeAccess::Read).max(access));
} else {
return None;
}
@ -198,17 +215,17 @@ impl QueryAccess {
/// Provides information about the types a [System] reads and writes
#[derive(Debug, Eq, PartialEq, Clone)]
pub struct TypeAccess<T: Hash + Eq + PartialEq> {
reads_all: bool,
reads_and_writes: HashSet<T>,
writes: HashSet<T>,
reads: HashSet<T>,
}
impl<T: Hash + Eq + PartialEq> Default for TypeAccess<T> {
fn default() -> Self {
Self {
reads_all: false,
reads_and_writes: Default::default(),
writes: Default::default(),
reads: Default::default(),
}
}
}
@ -219,36 +236,44 @@ impl<T: Hash + Eq + PartialEq + Copy> TypeAccess<T> {
for write in writes {
type_access.add_write(write);
}
for read in reads {
type_access.add_read(read);
}
type_access
}
pub fn is_compatible(&self, other: &TypeAccess<T>) -> bool {
self.writes.is_disjoint(&other.reads_and_writes)
&& self.reads_and_writes.is_disjoint(&other.writes)
if self.reads_all {
other.writes.is_empty()
} else if other.reads_all {
self.writes.is_empty()
} else {
self.writes.is_disjoint(&other.reads_and_writes)
&& self.reads_and_writes.is_disjoint(&other.writes)
}
}
pub fn get_conflict<'a>(&'a self, other: &'a TypeAccess<T>) -> Option<&'a T> {
let conflict = self.writes.intersection(&other.reads_and_writes).next();
if conflict.is_some() {
return conflict;
if self.reads_all {
other.writes.iter().next()
} else if other.reads_all {
self.writes.iter().next()
} else {
match self.writes.intersection(&other.reads_and_writes).next() {
Some(element) => Some(element),
None => other.writes.intersection(&self.reads_and_writes).next(),
}
}
self.reads_and_writes.intersection(&other.writes).next()
}
pub fn union(&mut self, other: &TypeAccess<T>) {
pub fn extend(&mut self, other: &TypeAccess<T>) {
self.reads_all = self.reads_all || other.reads_all;
self.writes.extend(&other.writes);
self.reads.extend(&other.reads);
self.reads_and_writes.extend(&other.reads_and_writes);
}
pub fn add_read(&mut self, ty: T) {
self.reads_and_writes.insert(ty);
self.reads.insert(ty);
}
pub fn add_write(&mut self, ty: T) {
@ -256,26 +281,108 @@ impl<T: Hash + Eq + PartialEq + Copy> TypeAccess<T> {
self.writes.insert(ty);
}
pub fn read_all(&mut self) {
self.reads_all = true;
}
pub fn clear(&mut self) {
self.reads_all = false;
self.reads_and_writes.clear();
self.reads.clear();
self.writes.clear();
}
pub fn is_read_or_write(&self, ty: &T) -> bool {
self.reads_and_writes.contains(ty)
self.reads_all || self.reads_and_writes.contains(ty)
}
pub fn is_write(&self, ty: &T) -> bool {
self.writes.contains(ty)
}
pub fn iter_reads(&self) -> impl Iterator<Item = &T> {
self.reads.iter()
pub fn reads_all(&self) -> bool {
self.reads_all
}
pub fn iter_writes(&self) -> impl Iterator<Item = &T> {
self.writes.iter()
/// Returns an iterator of distinct accessed types if only some types are accessed.
pub fn all_distinct_types(&self) -> Option<impl Iterator<Item = &T>> {
if !self.reads_all {
return Some(self.reads_and_writes.iter());
}
None
}
pub fn condense(&self, all_types: &[T]) -> CondensedTypeAccess {
if self.reads_all {
let mut writes = FixedBitSet::with_capacity(all_types.len());
for (index, access_type) in all_types.iter().enumerate() {
if self.writes.contains(access_type) {
writes.insert(index);
}
}
CondensedTypeAccess {
reads_all: true,
reads_and_writes: Default::default(),
writes,
}
} else {
let mut reads_and_writes = FixedBitSet::with_capacity(all_types.len());
let mut writes = FixedBitSet::with_capacity(all_types.len());
for (index, access_type) in all_types.iter().enumerate() {
if self.writes.contains(access_type) {
reads_and_writes.insert(index);
writes.insert(index);
} else if self.reads_and_writes.contains(access_type) {
reads_and_writes.insert(index);
}
}
CondensedTypeAccess {
reads_all: false,
reads_and_writes,
writes,
}
}
}
}
// TODO: consider making it typed, to enable compiler helping with bug hunting?
#[derive(Default, Debug, Eq, PartialEq, Clone)]
pub struct CondensedTypeAccess {
reads_all: bool,
reads_and_writes: FixedBitSet,
writes: FixedBitSet,
}
impl CondensedTypeAccess {
pub fn grow(&mut self, bits: usize) {
self.reads_and_writes.grow(bits);
self.writes.grow(bits);
}
pub fn reads_all(&self) -> bool {
self.reads_all
}
pub fn clear(&mut self) {
self.reads_all = false;
self.reads_and_writes.clear();
self.writes.clear();
}
pub fn extend(&mut self, other: &CondensedTypeAccess) {
self.reads_all = self.reads_all || other.reads_all;
self.reads_and_writes.union_with(&other.reads_and_writes);
self.writes.union_with(&other.writes);
}
pub fn is_compatible(&self, other: &CondensedTypeAccess) -> bool {
if self.reads_all {
0 == other.writes.count_ones(..)
} else if other.reads_all {
0 == self.writes.count_ones(..)
} else {
self.writes.is_disjoint(&other.reads_and_writes)
&& self.reads_and_writes.is_disjoint(&other.writes)
}
}
}

2
crates/bevy_ecs/src/core/mod.rs
View File

@ -43,7 +43,7 @@ mod serde;
mod world;
mod world_builder;
pub use access::{ArchetypeComponent, QueryAccess, TypeAccess};
pub use access::{ArchetypeComponent, CondensedTypeAccess, QueryAccess, TypeAccess};
pub use archetype::{Archetype, ComponentFlags, TypeState};
pub use borrow::{AtomicBorrow, Ref, RefMut};
pub use bundle::{Bundle, DynamicBundle, MissingComponent};

12
crates/bevy_ecs/src/lib.rs
View File

@ -13,10 +13,14 @@ pub use system::{Query, *};
pub mod prelude {
pub use crate::{
core::WorldBuilderSource,
resource::{ChangedRes, FromResources, Local, Res, ResMut, Resource, Resources},
schedule::{Schedule, State, StateStage, SystemStage},
system::{Commands, IntoSystem, Query, System},
resource::{ChangedRes, FromResources, Local, NonSend, Res, ResMut, Resource, Resources},
schedule::{
ExclusiveSystemDescriptorCoercion, ParallelSystemDescriptorCoercion,
ReportExecutionOrderAmbiguities, RunOnce, Schedule, Stage, State, StateStage,
SystemSet, SystemStage,
},
system::{Commands, ExclusiveSystem, IntoExclusiveSystem, IntoSystem, Query, System},
Added, Bundle, Changed, Component, Entity, Flags, In, IntoChainSystem, Mut, Mutated, Or,
QuerySet, Ref, RefMut, With, Without, World,
QuerySet, Ref, RefMut, ShouldRun, With, Without, World,
};
}

31
crates/bevy_ecs/src/resource/resource_query.rs
View File

@ -133,6 +133,37 @@ impl<'a, T: Resource + FromResources> DerefMut for Local<'a, T> {
}
}
/// `NonSend<T>` resources cannot leave the main thread, so any system that wants access to
/// a non-send resource will run on the main thread. See `Resources::insert_non_send()` and friends.
#[derive(Debug)]
pub struct NonSend<'a, T: Resource> {
value: *mut T,
_marker: PhantomData<&'a T>,
}
impl<'a, T: Resource> NonSend<'a, T> {
pub(crate) unsafe fn new(resources: &Resources) -> Self {
NonSend {
value: resources.get_unsafe_non_send_ref::<T>().as_ptr(),
_marker: Default::default(),
}
}
}
impl<'a, T: Resource> Deref for NonSend<'a, T> {
type Target = T;
fn deref(&self) -> &T {
unsafe { &*self.value }
}
}
impl<'a, T: Resource> DerefMut for NonSend<'a, T> {
fn deref_mut(&mut self) -> &mut T {
unsafe { &mut *self.value }
}
}
// #[cfg(test)]
// mod tests {
// use super::*;

76
crates/bevy_ecs/src/resource/resources.rs
View File

@ -99,7 +99,7 @@ impl<T: 'static> ResourceStorage for VecResourceStorage<T> {
/// A collection of resource instances identified by their type.
pub struct Resources {
pub(crate) resource_data: HashMap<TypeId, ResourceData>,
thread_local_data: HashMap<TypeId, Box<dyn ResourceStorage>>,
non_send_data: HashMap<TypeId, Box<dyn ResourceStorage>>,
main_thread_id: ThreadId,
}
@ -107,7 +107,7 @@ impl Default for Resources {
fn default() -> Self {
Resources {
resource_data: Default::default(),
thread_local_data: Default::default(),
non_send_data: Default::default(),
main_thread_id: std::thread::current().id(),
}
}
@ -118,10 +118,10 @@ impl Resources {
self.insert_resource(resource, ResourceIndex::Global);
}
pub fn insert_thread_local<T: 'static>(&mut self, resource: T) {
self.check_thread_local();
pub fn insert_non_send<T: 'static>(&mut self, resource: T) {
self.check_if_main_thread();
let entry = self
.thread_local_data
.non_send_data
.entry(TypeId::of::<T>())
.or_insert_with(|| Box::new(VecResourceStorage::<T>::default()));
let resources = entry.downcast_mut::<VecResourceStorage<T>>().unwrap();
@ -132,9 +132,9 @@ impl Resources {
}
}
fn check_thread_local(&self) {
fn check_if_main_thread(&self) {
if std::thread::current().id() != self.main_thread_id {
panic!("Attempted to access a thread local resource off of the main thread.")
panic!("Attempted to access a non-send resource off of the main thread.")
}
}
@ -150,9 +150,9 @@ impl Resources {
self.get_resource_mut(ResourceIndex::Global)
}
pub fn get_thread_local<T: 'static>(&self) -> Option<ResourceRef<'_, T>> {
self.check_thread_local();
self.thread_local_data
pub fn get_non_send<T: 'static>(&self) -> Option<ResourceRef<'_, T>> {
self.check_if_main_thread();
self.non_send_data
.get(&TypeId::of::<T>())
.and_then(|storage| {
let resources = storage.downcast_ref::<VecResourceStorage<T>>().unwrap();
@ -160,9 +160,9 @@ impl Resources {
})
}
pub fn get_thread_local_mut<T: 'static>(&self) -> Option<ResourceRefMut<'_, T>> {
self.check_thread_local();
self.thread_local_data
pub fn get_non_send_mut<T: 'static>(&self) -> Option<ResourceRefMut<'_, T>> {
self.check_if_main_thread();
self.non_send_data
.get(&TypeId::of::<T>())
.and_then(|storage| {
let resources = storage.downcast_ref::<VecResourceStorage<T>>().unwrap();
@ -282,6 +282,24 @@ impl Resources {
.unwrap_or_else(|| panic!("Resource does not exist {}.", std::any::type_name::<T>()))
}
#[inline]
#[allow(clippy::missing_safety_doc)]
pub unsafe fn get_unsafe_non_send_ref<T: 'static>(&self) -> NonNull<T> {
self.check_if_main_thread();
self.non_send_data
.get(&TypeId::of::<T>())
.map(|storage| {
let resources = storage.downcast_ref::<VecResourceStorage<T>>().unwrap();
resources.get_unsafe_ref(0)
})
.unwrap_or_else(|| {
panic!(
"Non-send resource does not exist {}.",
std::any::type_name::<T>()
)
})
}
#[inline]
#[allow(clippy::missing_safety_doc)]
pub unsafe fn get_unsafe_ref_with_added_and_mutated<T: Resource>(
@ -528,40 +546,40 @@ mod tests {
}
#[test]
fn thread_local_resource() {
fn non_send_resource() {
let mut resources = Resources::default();
resources.insert_thread_local(123i32);
resources.insert_thread_local(456i64);
assert_eq!(*resources.get_thread_local::<i32>().unwrap(), 123);
assert_eq!(*resources.get_thread_local_mut::<i64>().unwrap(), 456);
resources.insert_non_send(123i32);
resources.insert_non_send(456i64);
assert_eq!(*resources.get_non_send::<i32>().unwrap(), 123);
assert_eq!(*resources.get_non_send_mut::<i64>().unwrap(), 456);
}
#[test]
fn thread_local_resource_ref_aliasing() {
fn non_send_resource_ref_aliasing() {
let mut resources = Resources::default();
resources.insert_thread_local(123i32);
let a = resources.get_thread_local::<i32>().unwrap();
let b = resources.get_thread_local::<i32>().unwrap();
resources.insert_non_send(123i32);
let a = resources.get_non_send::<i32>().unwrap();
let b = resources.get_non_send::<i32>().unwrap();
assert_eq!(*a, 123);
assert_eq!(*b, 123);
}
#[test]
#[should_panic]
fn thread_local_resource_mut_ref_aliasing() {
fn non_send_resource_mut_ref_aliasing() {
let mut resources = Resources::default();
resources.insert_thread_local(123i32);
let _a = resources.get_thread_local::<i32>().unwrap();
let _b = resources.get_thread_local_mut::<i32>().unwrap();
resources.insert_non_send(123i32);
let _a = resources.get_non_send::<i32>().unwrap();
let _b = resources.get_non_send_mut::<i32>().unwrap();
}
#[test]
#[should_panic]
fn thread_local_resource_panic() {
fn non_send_resource_panic() {
let mut resources = Resources::default();
resources.insert_thread_local(0i32);
resources.insert_non_send(0i32);
std::thread::spawn(move || {
let _ = resources.get_thread_local_mut::<i32>();
let _ = resources.get_non_send_mut::<i32>();
})
.join()
.unwrap();

37
crates/bevy_ecs/src/schedule/executor.rs Normal file
View File

@ -0,0 +1,37 @@
use downcast_rs::{impl_downcast, Downcast};
use crate::{ParallelSystemContainer, Resources, World};
pub trait ParallelSystemExecutor: Downcast + Send + Sync {
/// Called by `SystemStage` whenever `systems` have been changed.
fn rebuild_cached_data(&mut self, systems: &mut [ParallelSystemContainer], world: &World);
fn run_systems(
&mut self,
systems: &mut [ParallelSystemContainer],
world: &mut World,
resources: &mut Resources,
);
}
impl_downcast!(ParallelSystemExecutor);
#[derive(Default)]
pub struct SingleThreadedExecutor;
impl ParallelSystemExecutor for SingleThreadedExecutor {
fn rebuild_cached_data(&mut self, _: &mut [ParallelSystemContainer], _: &World) {}
fn run_systems(
&mut self,
systems: &mut [ParallelSystemContainer],
world: &mut World,
resources: &mut Resources,
) {
for system in systems {
if system.should_run() {
system.system_mut().run((), world, resources);
}
}
}
}

508
crates/bevy_ecs/src/schedule/executor_parallel.rs Normal file
View File

@ -0,0 +1,508 @@
use async_channel::{Receiver, Sender};
use bevy_tasks::{ComputeTaskPool, Scope, TaskPool};
use bevy_utils::HashSet;
use fixedbitset::FixedBitSet;
use crate::{
ArchetypesGeneration, CondensedTypeAccess, ParallelSystemContainer, ParallelSystemExecutor,
Resources, System, World,
};
#[cfg(test)]
use SchedulingEvent::*;
struct SystemSchedulingMetadata {
/// Used to signal the system's task to start the system.
start_sender: Sender<()>,
/// Receives the signal to start the system.
start_receiver: Receiver<()>,
/// Indices of systems that depend on this one, used to decrement their
/// dependency counters when this system finishes.
dependants: Vec<usize>,
/// Total amount of dependencies this system has.
dependencies_total: usize,
/// Amount of unsatisfied dependencies, when it reaches 0 the system is queued to be started.
dependencies_now: usize,
/// Archetype-component access information condensed into executor-specific bitsets.
archetype_component_access: CondensedTypeAccess,
/// Resource access information condensed into executor-specific bitsets.
resource_access: CondensedTypeAccess,
}
pub struct ParallelExecutor {
/// Last archetypes generation observed by parallel systems.
last_archetypes_generation: ArchetypesGeneration,
/// Cached metadata of every system.
system_metadata: Vec<SystemSchedulingMetadata>,
/// Used by systems to notify the executor that they have finished.
finish_sender: Sender<usize>,
/// Receives finish events from systems.
finish_receiver: Receiver<usize>,
/// Systems that must run on the main thread.
non_send: FixedBitSet,
/// Systems that should be started at next opportunity.
queued: FixedBitSet,
/// Systems that are currently running.
running: FixedBitSet,
/// Whether a non-send system is currently running.
non_send_running: bool,
/// Systems that should run this iteration.
should_run: FixedBitSet,
/// Compound archetype-component access information of currently running systems.
active_archetype_component_access: CondensedTypeAccess,
/// Compound resource access information of currently running systems.
active_resource_access: CondensedTypeAccess,
/// Scratch space to avoid reallocating a vector when updating dependency counters.
dependants_scratch: Vec<usize>,
#[cfg(test)]
events_sender: Option<Sender<SchedulingEvent>>,
}
impl Default for ParallelExecutor {
fn default() -> Self {
let (finish_sender, finish_receiver) = async_channel::unbounded();
Self {
// MAX ensures access information will be initialized on first run.
last_archetypes_generation: ArchetypesGeneration(u64::MAX),
system_metadata: Default::default(),
finish_sender,
finish_receiver,
non_send: Default::default(),
queued: Default::default(),
running: Default::default(),
non_send_running: false,
should_run: Default::default(),
active_archetype_component_access: Default::default(),
active_resource_access: Default::default(),
dependants_scratch: Default::default(),
#[cfg(test)]
events_sender: None,
}
}
}
impl ParallelSystemExecutor for ParallelExecutor {
fn rebuild_cached_data(&mut self, systems: &mut [ParallelSystemContainer], world: &World) {
self.system_metadata.clear();
self.non_send.clear();
self.non_send.grow(systems.len());
self.queued.grow(systems.len());
self.running.grow(systems.len());
self.should_run.grow(systems.len());
// Collect all distinct types accessed by systems in order to condense their
// access sets into bitsets.
let mut all_archetype_components = HashSet::default();
let mut all_resource_types = HashSet::default();
let mut gather_distinct_access_types = |system: &dyn System<In = (), Out = ()>| {
if let Some(archetype_components) =
system.archetype_component_access().all_distinct_types()
{
all_archetype_components.extend(archetype_components);
}
if let Some(resources) = system.resource_access().all_distinct_types() {
all_resource_types.extend(resources);
}
};
// If the archetypes were changed too, system access should be updated
// before gathering the types.
if self.last_archetypes_generation != world.archetypes_generation() {
for container in systems.iter_mut() {
let system = container.system_mut();
system.update_access(world);
gather_distinct_access_types(system);
}
self.last_archetypes_generation = world.archetypes_generation();
} else {
for container in systems.iter() {
gather_distinct_access_types(container.system());
}
}
let all_archetype_components = all_archetype_components.drain().collect::<Vec<_>>();
let all_resource_types = all_resource_types.drain().collect::<Vec<_>>();
// Construct scheduling data for systems.
for container in systems.iter() {
let dependencies_total = container.dependencies().len();
let system = container.system();
if system.is_non_send() {
self.non_send.insert(self.system_metadata.len());
}
let (start_sender, start_receiver) = async_channel::bounded(1);
self.system_metadata.push(SystemSchedulingMetadata {
start_sender,
start_receiver,
dependants: vec![],
dependencies_total,
dependencies_now: 0,
archetype_component_access: system
.archetype_component_access()
.condense(&all_archetype_components),
resource_access: system.resource_access().condense(&all_resource_types),
});
}
// Populate the dependants lists in the scheduling metadata.
for (dependant, container) in systems.iter().enumerate() {
for dependency in container.dependencies() {
self.system_metadata[*dependency].dependants.push(dependant);
}
}
}
fn run_systems(
&mut self,
systems: &mut [ParallelSystemContainer],
world: &mut World,
resources: &mut Resources,
) {
#[cfg(test)]
if self.events_sender.is_none() {
let (sender, receiver) = async_channel::unbounded::<SchedulingEvent>();
resources.insert(receiver);
self.events_sender = Some(sender);
}
if self.last_archetypes_generation != world.archetypes_generation() {
self.update_access(systems, world);
self.last_archetypes_generation = world.archetypes_generation();
}
let compute_pool = resources
.get_or_insert_with(|| ComputeTaskPool(TaskPool::default()))
.clone();
compute_pool.scope(|scope| {
self.prepare_systems(scope, systems, world, resources);
scope.spawn(async {
// All systems have been ran if there are no queued or running systems.
while 0 != self.queued.count_ones(..) + self.running.count_ones(..) {
self.process_queued_systems().await;
// Avoid deadlocking if no systems were actually started.
if self.running.count_ones(..) != 0 {
// Wait until at least one system has finished.
let index = self
.finish_receiver
.recv()
.await
.unwrap_or_else(|error| unreachable!(error));
self.process_finished_system(index);
// Gather other systems than may have finished.
while let Ok(index) = self.finish_receiver.try_recv() {
self.process_finished_system(index);
}
// At least one system has finished, so active access is outdated.
self.rebuild_active_access();
}
self.update_counters_and_queue_systems();
}
});
});
}
}
impl ParallelExecutor {
/// Updates access and recondenses the archetype component bitsets of systems.
fn update_access(&mut self, systems: &mut [ParallelSystemContainer], world: &mut World) {
let mut all_archetype_components = HashSet::default();
for container in systems.iter_mut() {
let system = container.system_mut();
system.update_access(world);
if let Some(archetype_components) =
system.archetype_component_access().all_distinct_types()
{
all_archetype_components.extend(archetype_components);
}
}
let all_archetype_components = all_archetype_components.drain().collect::<Vec<_>>();
for (index, container) in systems.iter().enumerate() {
let system = container.system();
if !system.archetype_component_access().reads_all() {
self.system_metadata[index].archetype_component_access = system
.archetype_component_access()
.condense(&all_archetype_components);
}
}
}
/// Populates `should_run` bitset, spawns tasks for systems that should run this iteration,
/// queues systems with no dependencies to run (or skip) at next opportunity.
fn prepare_systems<'scope>(
&mut self,
scope: &mut Scope<'scope, ()>,
systems: &'scope [ParallelSystemContainer],
world: &'scope World,
resources: &'scope Resources,
) {
self.should_run.clear();
for (index, system_data) in self.system_metadata.iter_mut().enumerate() {
// Spawn the system task.
if systems[index].should_run() {
self.should_run.set(index, true);
let start_receiver = system_data.start_receiver.clone();
let finish_sender = self.finish_sender.clone();
let system = unsafe { systems[index].system_mut_unsafe() };
let task = async move {
start_receiver
.recv()
.await
.unwrap_or_else(|error| unreachable!(error));
unsafe { system.run_unsafe((), world, resources) };
finish_sender
.send(index)
.await
.unwrap_or_else(|error| unreachable!(error));
};
if self.non_send[index] {
scope.spawn_local(task);
} else {
scope.spawn(task);
}
}
// Queue the system if it has no dependencies, otherwise reset its dependency counter.
if system_data.dependencies_total == 0 {
self.queued.insert(index);
} else {
system_data.dependencies_now = system_data.dependencies_total;
}
}
}
/// Determines if the system with given index has no conflicts with already running systems.
fn can_start_now(&self, index: usize) -> bool {
let system_data = &self.system_metadata[index];
// Non-send systems are considered conflicting with each other.
!(self.non_send[index] && self.non_send_running)
&& system_data
.resource_access
.is_compatible(&self.active_resource_access)
&& system_data
.archetype_component_access
.is_compatible(&self.active_archetype_component_access)
}
/// Starts all non-conflicting queued systems, moves them from `queued` to `running`,
/// adds their access information to active access information;
/// processes queued systems that shouldn't run this iteration as completed immediately.
async fn process_queued_systems(&mut self) {
#[cfg(test)]
let mut started_systems = 0;
for index in self.queued.ones() {
// If the system shouldn't actually run this iteration, process it as completed
// immediately; otherwise, check for conflicts and signal its task to start.
if !self.should_run[index] {
self.dependants_scratch
.extend(&self.system_metadata[index].dependants);
} else if self.can_start_now(index) {
#[cfg(test)]
{
started_systems += 1;
}
let system_data = &self.system_metadata[index];
system_data
.start_sender
.send(())
.await
.unwrap_or_else(|error| unreachable!(error));
self.running.set(index, true);
if self.non_send[index] {
self.non_send_running = true;
}
// Add this system's access information to the active access information.
self.active_archetype_component_access
.extend(&system_data.archetype_component_access);
self.active_resource_access
.extend(&system_data.resource_access);
}
}
#[cfg(test)]
if started_systems != 0 {
self.emit_event(StartedSystems(started_systems));
}
// Remove now running systems from the queue.
self.queued.difference_with(&self.running);
// Remove immediately processed systems from the queue.
self.queued.intersect_with(&self.should_run);
}
/// Unmarks the system give index as running, caches indices of its dependants
/// in the `dependants_scratch`.
fn process_finished_system(&mut self, index: usize) {
if self.non_send[index] {
self.non_send_running = false;
}
self.running.set(index, false);
self.dependants_scratch
.extend(&self.system_metadata[index].dependants);
}
/// Discards active access information and builds it again using currently
/// running systems' access information.
fn rebuild_active_access(&mut self) {
self.active_archetype_component_access.clear();
self.active_resource_access.clear();
for index in self.running.ones() {
self.active_archetype_component_access
.extend(&self.system_metadata[index].archetype_component_access);
self.active_resource_access
.extend(&self.system_metadata[index].resource_access);
}
}
/// Drains `dependants_scratch`, decrementing dependency counters and enqueueing any
/// systems that become able to run.
fn update_counters_and_queue_systems(&mut self) {
for index in self.dependants_scratch.drain(..) {
let dependant_data = &mut self.system_metadata[index];
dependant_data.dependencies_now -= 1;
if dependant_data.dependencies_now == 0 {
self.queued.insert(index);
}
}
}
#[cfg(test)]
fn emit_event(&self, event: SchedulingEvent) {
self.events_sender
.as_ref()
.unwrap()
.try_send(event)
.unwrap();
}
}
#[cfg(test)]
#[derive(Debug, PartialEq, Eq)]
enum SchedulingEvent {
StartedSystems(usize),
}
#[cfg(test)]
mod tests {
use super::SchedulingEvent::{self, *};
use crate::{prelude::*, SingleThreadedExecutor};
use async_channel::Receiver;
use std::thread::{self, ThreadId};
fn receive_events(resources: &Resources) -> Vec<SchedulingEvent> {
let mut events = Vec::new();
while let Ok(event) = resources
.get::<Receiver<SchedulingEvent>>()
.unwrap()
.try_recv()
{
events.push(event);
}
events
}
#[test]
fn trivial() {
let mut world = World::new();
let mut resources = Resources::default();
fn wants_for_nothing() {}
let mut stage = SystemStage::parallel()
.with_system(wants_for_nothing.system())
.with_system(wants_for_nothing.system())
.with_system(wants_for_nothing.system());
stage.run(&mut world, &mut resources);
stage.run(&mut world, &mut resources);
assert_eq!(
receive_events(&resources),
vec![StartedSystems(3), StartedSystems(3),]
)
}
#[test]
fn resources() {
let mut world = World::new();
let mut resources = Resources::default();
resources.insert(0usize);
fn wants_mut(_: ResMut<usize>) {}
fn wants_ref(_: Res<usize>) {}
let mut stage = SystemStage::parallel()
.with_system(wants_mut.system())
.with_system(wants_mut.system());
stage.run(&mut world, &mut resources);
assert_eq!(
receive_events(&resources),
vec![StartedSystems(1), StartedSystems(1),]
);
let mut stage = SystemStage::parallel()
.with_system(wants_mut.system())
.with_system(wants_ref.system());
stage.run(&mut world, &mut resources);
assert_eq!(
receive_events(&resources),
vec![StartedSystems(1), StartedSystems(1),]
);
let mut stage = SystemStage::parallel()
.with_system(wants_ref.system())
.with_system(wants_ref.system());
stage.run(&mut world, &mut resources);
assert_eq!(receive_events(&resources), vec![StartedSystems(2),]);
}
#[test]
fn queries() {
let mut world = World::new();
let mut resources = Resources::default();
world.spawn((0usize,));
fn wants_mut(_: Query<&mut usize>) {}
fn wants_ref(_: Query<&usize>) {}
let mut stage = SystemStage::parallel()
.with_system(wants_mut.system())
.with_system(wants_mut.system());
stage.run(&mut world, &mut resources);
assert_eq!(
receive_events(&resources),
vec![StartedSystems(1), StartedSystems(1),]
);
let mut stage = SystemStage::parallel()
.with_system(wants_mut.system())
.with_system(wants_ref.system());
stage.run(&mut world, &mut resources);
assert_eq!(
receive_events(&resources),
vec![StartedSystems(1), StartedSystems(1),]
);
let mut stage = SystemStage::parallel()
.with_system(wants_ref.system())
.with_system(wants_ref.system());
stage.run(&mut world, &mut resources);
assert_eq!(receive_events(&resources), vec![StartedSystems(2),]);
let mut world = World::new();
world.spawn((0usize, 0u32, 0f32));
fn wants_mut_usize(_: Query<(&mut usize, &f32)>) {}
fn wants_mut_u32(_: Query<(&mut u32, &f32)>) {}
let mut stage = SystemStage::parallel()
.with_system(wants_mut_usize.system())
.with_system(wants_mut_u32.system());
stage.run(&mut world, &mut resources);
assert_eq!(receive_events(&resources), vec![StartedSystems(2),]);
}
#[test]
fn non_send_resource() {
let mut world = World::new();
let mut resources = Resources::default();
resources.insert_non_send(thread::current().id());
fn non_send(thread_id: NonSend<ThreadId>) {
assert_eq!(thread::current().id(), *thread_id);
}
fn empty() {}
let mut stage = SystemStage::parallel()
.with_system(non_send.system())
.with_system(non_send.system())
.with_system(empty.system())
.with_system(empty.system())
.with_system(non_send.system())
.with_system(non_send.system());
stage.run(&mut world, &mut resources);
assert_eq!(
receive_events(&resources),
vec![
StartedSystems(3),
StartedSystems(1),
StartedSystems(1),
StartedSystems(1),
]
);
stage.set_executor(Box::new(SingleThreadedExecutor::default()));
stage.run(&mut world, &mut resources);
}
}

492
crates/bevy_ecs/src/schedule/mod.rs
View File

@ -1,20 +1,30 @@
mod executor;
mod executor_parallel;
mod stage;
mod stage_executor;
mod state;
mod system_container;
mod system_descriptor;
mod system_set;
pub use executor::*;
pub use executor_parallel::*;
pub use stage::*;
pub use stage_executor::*;
pub use state::*;
pub use system_container::*;
pub use system_descriptor::*;
pub use system_set::*;
use crate::{BoxedSystem, IntoSystem, Resources, System, World};
use crate::{
ArchetypeComponent, BoxedSystem, IntoSystem, Resources, System, SystemId, TypeAccess, World,
};
use bevy_utils::HashMap;
use std::{any::TypeId, borrow::Cow};
#[derive(Default)]
pub struct Schedule {
stages: HashMap<String, Box<dyn Stage>>,
stage_order: Vec<String>,
run_criteria: Option<BoxedSystem<(), ShouldRun>>,
run_criteria_initialized: bool,
run_criteria: RunCriteria,
}
impl Schedule {
@ -38,10 +48,10 @@ impl Schedule {
self
}
pub fn with_system_in_stage<S: System<In = (), Out = ()>>(
pub fn with_system_in_stage(
mut self,
stage_name: &'static str,
system: S,
system: impl Into<SystemDescriptor>,
) -> Self {
self.add_system_to_stage(stage_name, system);
self
@ -51,8 +61,7 @@ impl Schedule {
&mut self,
system: S,
) -> &mut Self {
self.run_criteria = Some(Box::new(system.system()));
self.run_criteria_initialized = false;
self.run_criteria.set(Box::new(system.system()));
self
}
@ -99,10 +108,10 @@ impl Schedule {
self
}
pub fn add_system_to_stage<S: System<In = (), Out = ()>>(
pub fn add_system_to_stage(
&mut self,
stage_name: &'static str,
system: S,
system: impl Into<SystemDescriptor>,
) -> &mut Self {
let stage = self
.get_stage_mut::<SystemStage>(stage_name)
@ -112,7 +121,7 @@ impl Schedule {
stage_name
)
});
stage.add_system(system.system());
stage.add_system(system);
self
}
@ -150,49 +159,23 @@ impl Schedule {
stage.run(world, resources);
}
}
/// Shorthand for [Schedule::initialize] and [Schedule::run]
pub fn initialize_and_run(&mut self, world: &mut World, resources: &mut Resources) {
self.initialize(world, resources);
self.run(world, resources);
}
}
impl Stage for Schedule {
fn initialize(&mut self, world: &mut World, resources: &mut Resources) {
if let Some(ref mut run_criteria) = self.run_criteria {
if !self.run_criteria_initialized {
run_criteria.initialize(world, resources);
self.run_criteria_initialized = true;
}
}
for name in self.stage_order.iter() {
let stage = self.stages.get_mut(name).unwrap();
stage.initialize(world, resources);
}
}
fn run(&mut self, world: &mut World, resources: &mut Resources) {
loop {
let should_run = if let Some(ref mut run_criteria) = self.run_criteria {
let should_run = run_criteria.run((), world, resources);
run_criteria.run_thread_local(world, resources);
// don't run when no result is returned or false is returned
should_run.unwrap_or(ShouldRun::No)
} else {
ShouldRun::Yes
};
match should_run {
match self.run_criteria.should_run(world, resources) {
ShouldRun::No => return,
ShouldRun::Yes => {
self.run_once(world, resources);
return;
}
ShouldRun::YesAndLoop => {
ShouldRun::YesAndCheckAgain => {
self.run_once(world, resources);
}
ShouldRun::NoAndCheckAgain => {
panic!("`NoAndCheckAgain` would loop infinitely in this situation.")
}
}
}
}
@ -203,320 +186,119 @@ pub fn clear_trackers_system(world: &mut World, resources: &mut Resources) {
resources.clear_trackers();
}
#[cfg(test)]
mod tests {
use crate::{
resource::{Res, ResMut, Resources},
schedule::{ParallelSystemStageExecutor, Schedule, SystemStage},
system::Query,
Commands, Entity, IntoSystem, World,
};
use bevy_tasks::{ComputeTaskPool, TaskPool};
use fixedbitset::FixedBitSet;
use parking_lot::Mutex;
use std::{collections::HashSet, sync::Arc};
#[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,
}
#[derive(Default)]
struct CompletedSystems {
completed_systems: Arc<Mutex<HashSet<&'static str>>>,
}
pub(crate) struct RunCriteria {
criteria_system: Option<BoxedSystem<(), ShouldRun>>,
initialized: bool,
}
#[test]
fn cross_stage_archetype_change_prepare() {
let mut world = World::new();
let mut resources = Resources::default();
resources.insert(ComputeTaskPool(TaskPool::default()));
fn insert(commands: &mut Commands) {
commands.spawn((1u32,));
}
fn read(query: Query<&u32>, entities: Query<Entity>) {
for entity in &mut entities.iter() {
// query.get() does a "system permission check" that will fail if the entity is from a
// new archetype which hasnt been "prepared yet"
query.get_component::<u32>(entity).unwrap();
}
assert_eq!(1, entities.iter().count());
}
let mut schedule = Schedule::default();
let mut pre_archetype_change = SystemStage::parallel();
pre_archetype_change.add_system(insert.system());
schedule.add_stage("PreArchetypeChange", pre_archetype_change);
let mut post_archetype_change = SystemStage::parallel();
post_archetype_change.add_system(read.system());
schedule.add_stage("PostArchetypeChange", post_archetype_change);
schedule.initialize_and_run(&mut world, &mut resources);
}
#[test]
fn intra_stage_archetype_change_prepare() {
let mut world = World::new();
let mut resources = Resources::default();
resources.insert(ComputeTaskPool(TaskPool::default()));
fn insert(world: &mut World, _resources: &mut Resources) {
world.spawn((1u32,));
}
fn read(query: Query<&u32>, entities: Query<Entity>) {
for entity in &mut entities.iter() {
// query.get() does a "system permission check" that will fail if the entity is from a
// new archetype which hasnt been "prepared yet"
query.get_component::<u32>(entity).unwrap();
}
assert_eq!(1, entities.iter().count());
}
let mut update = SystemStage::parallel();
update.add_system(insert.system());
update.add_system(read.system());
let mut schedule = Schedule::default();
schedule.add_stage("update", update);
schedule.initialize_and_run(&mut world, &mut resources);
}
#[test]
fn schedule() {
let mut world = World::new();
let mut resources = Resources::default();
resources.insert(ComputeTaskPool(TaskPool::default()));
resources.insert(CompletedSystems::default());
resources.insert(1.0f64);
resources.insert(2isize);
world.spawn((1.0f32,));
world.spawn((1u32, 1u64));
world.spawn((2u32,));
let mut stage_a = SystemStage::parallel(); // component queries
let mut stage_b = SystemStage::parallel(); // thread local
let mut stage_c = SystemStage::parallel(); // resources
// A system names
const READ_U32_SYSTEM_NAME: &str = "read_u32";
const WRITE_FLOAT_SYSTEM_NAME: &str = "write_float";
const READ_U32_WRITE_U64_SYSTEM_NAME: &str = "read_u32_write_u64";
const READ_U64_SYSTEM_NAME: &str = "read_u64";
// B system names
const WRITE_U64_SYSTEM_NAME: &str = "write_u64";
const THREAD_LOCAL_SYSTEM_SYSTEM_NAME: &str = "thread_local_system";
const WRITE_F32_SYSTEM_NAME: &str = "write_f32";
// C system names
const READ_F64_RES_SYSTEM_NAME: &str = "read_f64_res";
const READ_ISIZE_RES_SYSTEM_NAME: &str = "read_isize_res";
const READ_ISIZE_WRITE_F64_RES_SYSTEM_NAME: &str = "read_isize_write_f64_res";
const WRITE_F64_RES_SYSTEM_NAME: &str = "write_f64_res";
// A systems
fn read_u32(completed_systems: Res<CompletedSystems>, _query: Query<&u32>) {
let mut completed_systems = completed_systems.completed_systems.lock();
completed_systems.insert(READ_U32_SYSTEM_NAME);
}
fn write_float(completed_systems: Res<CompletedSystems>, _query: Query<&f32>) {
let mut completed_systems = completed_systems.completed_systems.lock();
completed_systems.insert(WRITE_FLOAT_SYSTEM_NAME);
}
fn read_u32_write_u64(
completed_systems: Res<CompletedSystems>,
_query: Query<(&u32, &mut u64)>,
) {
let mut completed_systems = completed_systems.completed_systems.lock();
assert!(!completed_systems.contains(READ_U64_SYSTEM_NAME));
completed_systems.insert(READ_U32_WRITE_U64_SYSTEM_NAME);
}
fn read_u64(completed_systems: Res<CompletedSystems>, _query: Query<&u64>) {
let mut completed_systems = completed_systems.completed_systems.lock();
assert!(completed_systems.contains(READ_U32_WRITE_U64_SYSTEM_NAME));
assert!(!completed_systems.contains(WRITE_U64_SYSTEM_NAME));
completed_systems.insert(READ_U64_SYSTEM_NAME);
}
stage_a.add_system(read_u32.system());
stage_a.add_system(write_float.system());
stage_a.add_system(read_u32_write_u64.system());
stage_a.add_system(read_u64.system());
// B systems
fn write_u64(completed_systems: Res<CompletedSystems>, _query: Query<&mut u64>) {
let mut completed_systems = completed_systems.completed_systems.lock();
assert!(completed_systems.contains(READ_U64_SYSTEM_NAME));
assert!(!completed_systems.contains(THREAD_LOCAL_SYSTEM_SYSTEM_NAME));
assert!(!completed_systems.contains(WRITE_F32_SYSTEM_NAME));
completed_systems.insert(WRITE_U64_SYSTEM_NAME);
}
fn thread_local_system(_world: &mut World, resources: &mut Resources) {
let completed_systems = resources.get::<CompletedSystems>().unwrap();
let mut completed_systems = completed_systems.completed_systems.lock();
assert!(completed_systems.contains(WRITE_U64_SYSTEM_NAME));
assert!(!completed_systems.contains(WRITE_F32_SYSTEM_NAME));
completed_systems.insert(THREAD_LOCAL_SYSTEM_SYSTEM_NAME);
}
fn write_f32(completed_systems: Res<CompletedSystems>, _query: Query<&mut f32>) {
let mut completed_systems = completed_systems.completed_systems.lock();
assert!(completed_systems.contains(WRITE_U64_SYSTEM_NAME));
assert!(completed_systems.contains(THREAD_LOCAL_SYSTEM_SYSTEM_NAME));
assert!(!completed_systems.contains(READ_F64_RES_SYSTEM_NAME));
completed_systems.insert(WRITE_F32_SYSTEM_NAME);
}
stage_b.add_system(write_u64.system());
stage_b.add_system(thread_local_system.system());
stage_b.add_system(write_f32.system());
// C systems
fn read_f64_res(completed_systems: Res<CompletedSystems>, _f64_res: Res<f64>) {
let mut completed_systems = completed_systems.completed_systems.lock();
assert!(completed_systems.contains(WRITE_F32_SYSTEM_NAME));
assert!(!completed_systems.contains(READ_ISIZE_WRITE_F64_RES_SYSTEM_NAME));
assert!(!completed_systems.contains(WRITE_F64_RES_SYSTEM_NAME));
completed_systems.insert(READ_F64_RES_SYSTEM_NAME);
}
fn read_isize_res(completed_systems: Res<CompletedSystems>, _isize_res: Res<isize>) {
let mut completed_systems = completed_systems.completed_systems.lock();
completed_systems.insert(READ_ISIZE_RES_SYSTEM_NAME);
}
fn read_isize_write_f64_res(
completed_systems: Res<CompletedSystems>,
_isize_res: Res<isize>,
_f64_res: ResMut<f64>,
) {
let mut completed_systems = completed_systems.completed_systems.lock();
assert!(completed_systems.contains(READ_F64_RES_SYSTEM_NAME));
assert!(!completed_systems.contains(WRITE_F64_RES_SYSTEM_NAME));
completed_systems.insert(READ_ISIZE_WRITE_F64_RES_SYSTEM_NAME);
}
fn write_f64_res(completed_systems: Res<CompletedSystems>, _f64_res: ResMut<f64>) {
let mut completed_systems = completed_systems.completed_systems.lock();
assert!(completed_systems.contains(READ_F64_RES_SYSTEM_NAME));
assert!(completed_systems.contains(READ_ISIZE_WRITE_F64_RES_SYSTEM_NAME));
completed_systems.insert(WRITE_F64_RES_SYSTEM_NAME);
}
stage_c.add_system(read_f64_res.system());
stage_c.add_system(read_isize_res.system());
stage_c.add_system(read_isize_write_f64_res.system());
stage_c.add_system(write_f64_res.system());
fn run_and_validate(schedule: &mut Schedule, world: &mut World, resources: &mut Resources) {
schedule.initialize_and_run(world, resources);
let stage_a = schedule.get_stage::<SystemStage>("a").unwrap();
let stage_b = schedule.get_stage::<SystemStage>("b").unwrap();
let stage_c = schedule.get_stage::<SystemStage>("c").unwrap();
let a_executor = stage_a
.get_executor::<ParallelSystemStageExecutor>()
.unwrap();
let b_executor = stage_b
.get_executor::<ParallelSystemStageExecutor>()
.unwrap();
let c_executor = stage_c
.get_executor::<ParallelSystemStageExecutor>()
.unwrap();
assert_eq!(
a_executor.system_dependents(),
vec![vec![], vec![], vec![3], vec![]]
);
assert_eq!(
b_executor.system_dependents(),
vec![vec![1], vec![2], vec![]]
);
assert_eq!(
c_executor.system_dependents(),
vec![vec![2, 3], vec![], vec![3], vec![]]
);
let stage_a_len = a_executor.system_dependencies().len();
let mut read_u64_deps = FixedBitSet::with_capacity(stage_a_len);
read_u64_deps.insert(2);
assert_eq!(
a_executor.system_dependencies(),
vec![
FixedBitSet::with_capacity(stage_a_len),
FixedBitSet::with_capacity(stage_a_len),
FixedBitSet::with_capacity(stage_a_len),
read_u64_deps,
]
);
let stage_b_len = b_executor.system_dependencies().len();
let mut thread_local_deps = FixedBitSet::with_capacity(stage_b_len);
thread_local_deps.insert(0);
let mut write_f64_deps = FixedBitSet::with_capacity(stage_b_len);
write_f64_deps.insert(1);
assert_eq!(
b_executor.system_dependencies(),
vec![
FixedBitSet::with_capacity(stage_b_len),
thread_local_deps,
write_f64_deps
]
);
let stage_c_len = c_executor.system_dependencies().len();
let mut read_isize_write_f64_res_deps = FixedBitSet::with_capacity(stage_c_len);
read_isize_write_f64_res_deps.insert(0);
let mut write_f64_res_deps = FixedBitSet::with_capacity(stage_c_len);
write_f64_res_deps.insert(0);
write_f64_res_deps.insert(2);
assert_eq!(
c_executor.system_dependencies(),
vec![
FixedBitSet::with_capacity(stage_c_len),
FixedBitSet::with_capacity(stage_c_len),
read_isize_write_f64_res_deps,
write_f64_res_deps
]
);
let completed_systems = resources.get::<CompletedSystems>().unwrap();
assert_eq!(
completed_systems.completed_systems.lock().len(),
11,
"completed_systems should have been incremented once for each system"
);
}
let mut schedule = Schedule::default();
schedule.add_stage("a", stage_a);
schedule.add_stage("b", stage_b);
schedule.add_stage("c", stage_c);
// Test the "clean start" case
run_and_validate(&mut schedule, &mut world, &mut resources);
// Stress test the "continue running" case
for _ in 0..1000 {
// run again (with completed_systems reset) to ensure executor works correctly across runs
resources
.get::<CompletedSystems>()
.unwrap()
.completed_systems
.lock()
.clear();
run_and_validate(&mut schedule, &mut world, &mut resources);
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, resources: &mut Resources) -> ShouldRun {
if let Some(ref mut run_criteria) = self.criteria_system {
if !self.initialized {
run_criteria.initialize(world, resources);
self.initialized = true;
}
let should_run = run_criteria.run((), world, resources);
run_criteria.apply_buffers(world, resources);
// don't run when no result is returned or false is returned
should_run.unwrap_or(ShouldRun::No)
} else {
ShouldRun::Yes
}
}
}
pub struct RunOnce {
ran: bool,
system_id: SystemId,
archetype_component_access: TypeAccess<ArchetypeComponent>,
component_access: TypeAccess<TypeId>,
resource_access: TypeAccess<TypeId>,
}
impl Default for RunOnce {
fn default() -> Self {
Self {
ran: false,
system_id: SystemId::new(),
archetype_component_access: Default::default(),
component_access: Default::default(),
resource_access: Default::default(),
}
}
}
impl System for RunOnce {
type In = ();
type Out = ShouldRun;
fn name(&self) -> Cow<'static, str> {
Cow::Borrowed(std::any::type_name::<RunOnce>())
}
fn id(&self) -> SystemId {
self.system_id
}
fn update_access(&mut self, _world: &World) {}
fn archetype_component_access(&self) -> &TypeAccess<ArchetypeComponent> {
&self.archetype_component_access
}
fn component_access(&self) -> &TypeAccess<TypeId> {
&self.component_access
}
fn resource_access(&self) -> &TypeAccess<TypeId> {
&self.resource_access
}
fn is_non_send(&self) -> bool {
false
}
unsafe fn run_unsafe(
&mut self,
_input: Self::In,
_world: &World,
_resources: &Resources,
) -> Option<Self::Out> {
Some(if self.ran {
ShouldRun::No
} else {
self.ran = true;
ShouldRun::Yes
})
}
fn apply_buffers(&mut self, _world: &mut World, _resources: &mut Resources) {}
fn initialize(&mut self, _world: &mut World, _resources: &mut Resources) {}
}

1452
crates/bevy_ecs/src/schedule/stage.rs
View File

File diff suppressed because it is too large Load Diff

528
crates/bevy_ecs/src/schedule/stage_executor.rs
View File

@ -1,528 +0,0 @@
use std::ops::Range;
use bevy_tasks::{ComputeTaskPool, CountdownEvent, TaskPool};
use bevy_utils::tracing::trace;
use downcast_rs::{impl_downcast, Downcast};
use fixedbitset::FixedBitSet;
use crate::{
ArchetypesGeneration, BoxedSystem, Resources, ThreadLocalExecution, TypeAccess, World,
};
pub trait SystemStageExecutor: Downcast + Send + Sync {
fn execute_stage(
&mut self,
systems: &mut [BoxedSystem],
changed_systems: &[usize],
world: &mut World,
resources: &mut Resources,
);
}
impl_downcast!(SystemStageExecutor);
#[derive(Default)]
pub struct SerialSystemStageExecutor;
impl SystemStageExecutor for SerialSystemStageExecutor {
fn execute_stage(
&mut self,
systems: &mut [BoxedSystem],
_changed_systems: &[usize],
world: &mut World,
resources: &mut Resources,
) {
for system in systems.iter_mut() {
system.update(world);
match system.thread_local_execution() {
ThreadLocalExecution::NextFlush => {
system.run((), world, resources);
}
ThreadLocalExecution::Immediate => {
system.run((), world, resources);
system.run_thread_local(world, resources);
}
}
}
// "flush"
for system in systems.iter_mut() {
match system.thread_local_execution() {
ThreadLocalExecution::NextFlush => system.run_thread_local(world, resources),
ThreadLocalExecution::Immediate => { /* already ran immediate */ }
}
}
}
}
/// Executes the stage in parallel by analyzing system dependencies.
/// System execution order is undefined except under the following conditions:
/// * systems in earlier stages run before systems in later stages
/// * in a given stage, systems that mutate [archetype+component] X cannot run before systems registered before them that read/write [archetype+component] X
/// * in a given stage, systems the read [archetype+component] X cannot run before systems registered before them that write [archetype+component] X
/// * in a given stage, systems that mutate resource Y cannot run before systems registered before them that read/write resource Y
/// * in a given stage, systems the read resource Y cannot run before systems registered before them that write resource Y
pub struct ParallelSystemStageExecutor {
/// each system's set of dependencies
system_dependencies: Vec<FixedBitSet>,
/// count of each system's dependencies
system_dependency_count: Vec<usize>,
/// Countdown of finished dependencies, used to trigger the next system
ready_events: Vec<Option<CountdownEvent>>,
/// When a system finishes, it will decrement the countdown events of all dependents
ready_events_of_dependents: Vec<Vec<CountdownEvent>>,
/// each system's dependents (the systems that can't run until this system has run)
system_dependents: Vec<Vec<usize>>,
/// stores the indices of thread local systems in this stage, which are used during stage.prepare()
thread_local_system_indices: Vec<usize>,
/// When archetypes change a counter is bumped - we cache the state of that counter when it was
/// last read here so that we can detect when archetypes are changed
last_archetypes_generation: ArchetypesGeneration,
}
impl Default for ParallelSystemStageExecutor {
fn default() -> Self {
Self {
system_dependents: Default::default(),
system_dependency_count: Default::default(),
ready_events: Default::default(),
ready_events_of_dependents: Default::default(),
system_dependencies: Default::default(),
thread_local_system_indices: Default::default(),
last_archetypes_generation: ArchetypesGeneration(u64::MAX), // MAX forces prepare to run the first time
}
}
}
impl ParallelSystemStageExecutor {
pub fn system_dependents(&self) -> &[Vec<usize>] {
&self.system_dependents
}
pub fn system_dependencies(&self) -> &[FixedBitSet] {
&self.system_dependencies
}
/// Sets up state to run the next "batch" of systems. Each batch contains 0..n systems and
/// optionally a thread local system at the end. After this function runs, a bunch of state
/// in self will be populated for systems in this batch. Returns the range of systems
/// that we prepared, up to but NOT including the thread local system that MIGHT be at the end
/// of the range
pub fn prepare_to_next_thread_local(
&mut self,
world: &World,
systems: &mut [BoxedSystem],
stage_changed: bool,
next_thread_local_index: usize,
) -> Range<usize> {
// Find the first system in this batch and (if there is one) the thread local system that
// ends it.
let (prepare_system_start_index, last_thread_local_index) = if next_thread_local_index == 0
{
(0, None)
} else {
// start right after the last thread local system
(
self.thread_local_system_indices[next_thread_local_index - 1] + 1,
Some(self.thread_local_system_indices[next_thread_local_index - 1]),
)
};
let prepare_system_index_range = if let Some(index) = self
.thread_local_system_indices
.get(next_thread_local_index)
{
// if there is an upcoming thread local system, prepare up to (and including) it
prepare_system_start_index..(*index + 1)
} else {
// if there are no upcoming thread local systems, prepare everything right now
prepare_system_start_index..systems.len()
};
let archetypes_generation_changed =
self.last_archetypes_generation != world.archetypes_generation();
if stage_changed || archetypes_generation_changed {
// update each system's [archetype+component] access to latest world archetypes
for system_index in prepare_system_index_range.clone() {
systems[system_index].update(world);
// Clear this so that the next block of code that populates it doesn't insert
// duplicates
self.system_dependents[system_index].clear();
self.system_dependencies[system_index].clear();
}
// calculate dependencies between systems and build execution order
let mut current_archetype_access = TypeAccess::default();
let mut current_resource_access = TypeAccess::default();
for system_index in prepare_system_index_range.clone() {
let system = &systems[system_index];
let archetype_access = system.archetype_component_access();
match system.thread_local_execution() {
ThreadLocalExecution::NextFlush => {
let resource_access = system.resource_access();
// if any system before this one conflicts, check all systems that came before for compatibility
if !current_archetype_access.is_compatible(archetype_access)
|| !current_resource_access.is_compatible(resource_access)
{
#[allow(clippy::needless_range_loop)]
for earlier_system_index in
prepare_system_index_range.start..system_index
{
let earlier_system = &systems[earlier_system_index];
// due to how prepare ranges work, previous systems should all be "NextFlush"
debug_assert_eq!(
earlier_system.thread_local_execution(),
ThreadLocalExecution::NextFlush
);
// if earlier system is incompatible, make the current system dependent
if !earlier_system
.archetype_component_access()
.is_compatible(archetype_access)
|| !earlier_system
.resource_access()
.is_compatible(resource_access)
{
self.system_dependents[earlier_system_index].push(system_index);
self.system_dependencies[system_index]
.insert(earlier_system_index);
}
}
}
current_archetype_access.union(archetype_access);
current_resource_access.union(resource_access);
if let Some(last_thread_local_index) = last_thread_local_index {
self.system_dependents[last_thread_local_index].push(system_index);
self.system_dependencies[system_index].insert(last_thread_local_index);
}
}
ThreadLocalExecution::Immediate => {
for earlier_system_index in prepare_system_index_range.start..system_index {
// treat all earlier systems as "incompatible" to ensure we run this thread local system exclusively
self.system_dependents[earlier_system_index].push(system_index);
self.system_dependencies[system_index].insert(earlier_system_index);
}
}
}
}
// Verify that dependents are not duplicated
#[cfg(debug_assertions)]
for system_index in prepare_system_index_range.clone() {
let mut system_dependents_set = std::collections::HashSet::new();
for dependent_system in &self.system_dependents[system_index] {
let inserted = system_dependents_set.insert(*dependent_system);
// This means duplicate values are in the system_dependents list
// This is reproducing when archetypes change. When we fix this, we can remove
// the hack below and make this a debug-only assert or remove it
debug_assert!(inserted);
}
}
// Clear the ready events lists associated with each system so we can rebuild them
for ready_events_of_dependents in
&mut self.ready_events_of_dependents[prepare_system_index_range.clone()]
{
ready_events_of_dependents.clear();
}
// Now that system_dependents and system_dependencies is populated, update
// system_dependency_count and ready_events
for system_index in prepare_system_index_range.clone() {
// Count all dependencies to update system_dependency_count
assert!(!self.system_dependencies[system_index].contains(system_index));
let dependency_count = self.system_dependencies[system_index].count_ones(..);
self.system_dependency_count[system_index] = dependency_count;
// If dependency count > 0, allocate a ready_event
self.ready_events[system_index] = match self.system_dependency_count[system_index] {
0 => None,
dependency_count => Some(CountdownEvent::new(dependency_count as isize)),
}
}
// Now that ready_events are created, we can build ready_events_of_dependents
for system_index in prepare_system_index_range.clone() {
for dependent_system in &self.system_dependents[system_index] {
self.ready_events_of_dependents[system_index].push(
self.ready_events[*dependent_system]
.as_ref()
.expect("A dependent task should have a non-None ready event.")
.clone(),
);
}
}
} else {
// Reset the countdown events for this range of systems. Resetting is required even if the
// schedule didn't change
self.reset_system_ready_events(prepare_system_index_range);
}
if let Some(index) = self
.thread_local_system_indices
.get(next_thread_local_index)
{
// if there is an upcoming thread local system, prepare up to (and NOT including) it
prepare_system_start_index..(*index)
} else {
// if there are no upcoming thread local systems, prepare everything right now
prepare_system_start_index..systems.len()
}
}
fn reset_system_ready_events(&mut self, prepare_system_index_range: Range<usize>) {
for system_index in prepare_system_index_range {
let dependency_count = self.system_dependency_count[system_index];
if dependency_count > 0 {
self.ready_events[system_index]
.as_ref()
.expect("A system with >0 dependency count should have a non-None ready event.")
.reset(dependency_count as isize)
}
}
}
/// Runs the non-thread-local systems in the given prepared_system_range range
pub fn run_systems(
&self,
world: &World,
resources: &Resources,
systems: &mut [BoxedSystem],
prepared_system_range: Range<usize>,
compute_pool: &TaskPool,
) {
// Generate tasks for systems in the given range and block until they are complete
trace!("running systems {:?}", prepared_system_range);
compute_pool.scope(|scope| {
let start_system_index = prepared_system_range.start;
let mut system_index = start_system_index;
for system in &mut systems[prepared_system_range] {
trace!(
"prepare {} {} with {} dependents and {} dependencies",
system_index,
system.name(),
self.system_dependents[system_index].len(),
self.system_dependencies[system_index].count_ones(..)
);
// This event will be awaited, preventing the task from starting until all
// our dependencies finish running
let ready_event = &self.ready_events[system_index];
// Clear any dependencies on systems before this range of systems. We know at this
// point everything before start_system_index is finished, and our ready_event did
// not exist to be decremented until we started processing this range
if start_system_index != 0 {
if let Some(ready_event) = ready_event.as_ref() {
for dependency in self.system_dependencies[system_index].ones() {
if dependency < start_system_index {
ready_event.decrement();
}
}
}
}
let world_ref = &*world;
let resources_ref = &*resources;
let trigger_events = &self.ready_events_of_dependents[system_index];
// Verify that any dependent task has a > 0 count. If a dependent task has > 0
// count, then the current system we are starting now isn't blocking it from running
// as it should be. Failure here implies the sync primitives are not matching the
// intended schedule. This likely compiles out if trace/asserts are disabled but
// make it explicitly debug-only anyways
#[cfg(debug_assertions)]
{
let dependent_systems = &self.system_dependents[system_index];
debug_assert_eq!(trigger_events.len(), dependent_systems.len());
for (trigger_event, dependent_system_index) in
trigger_events.iter().zip(dependent_systems)
{
debug_assert!(
*dependent_system_index < start_system_index || trigger_event.get() > 0
);
}
}
// Spawn the task
scope.spawn(async move {
// Wait until our dependencies are done
if let Some(ready_event) = ready_event {
ready_event.listen().await;
}
// Execute the system - in a scope to ensure the system lock is dropped before
// triggering dependents
{
#[cfg(feature = "trace")]
let system_span = bevy_utils::tracing::info_span!(
"system",
name = system.name().as_ref()
);
#[cfg(feature = "trace")]
let _system_guard = system_span.enter();
// SAFETY: scheduler ensures safe world / resource access
unsafe {
system.run_unsafe((), world_ref, resources_ref);
}
}
// Notify dependents that this task is done
for trigger_event in trigger_events {
trigger_event.decrement();
}
});
system_index += 1;
}
});
}
}
impl SystemStageExecutor for ParallelSystemStageExecutor {
fn execute_stage(
&mut self,
systems: &mut [BoxedSystem],
changed_systems: &[usize],
world: &mut World,
resources: &mut Resources,
) {
let start_archetypes_generation = world.archetypes_generation();
let compute_pool = resources
.get_or_insert_with(|| ComputeTaskPool(TaskPool::default()))
.clone();
let stage_changed = !changed_systems.is_empty();
// if the schedule has changed, clear executor state / fill it with new defaults
// This is mostly zeroing out a bunch of arrays parallel to the systems array. They will get
// repopulated by prepare_to_next_thread_local() calls
if stage_changed {
self.system_dependencies.clear();
self.system_dependencies
.resize_with(systems.len(), || FixedBitSet::with_capacity(systems.len()));
self.system_dependency_count.clear();
self.system_dependency_count.resize(systems.len(), 0);
self.thread_local_system_indices = Vec::new();
self.system_dependents.clear();
self.system_dependents.resize(systems.len(), Vec::new());
self.ready_events.resize(systems.len(), None);
self.ready_events_of_dependents
.resize(systems.len(), Vec::new());
for (system_index, system) in systems.iter().enumerate() {
if system.thread_local_execution() == ThreadLocalExecution::Immediate {
#[cfg(feature = "trace")]
let system_span =
bevy_utils::tracing::info_span!("system", name = system.name().as_ref());
#[cfg(feature = "trace")]
let _system_guard = system_span.enter();
self.thread_local_system_indices.push(system_index);
}
}
}
// index of next thread local system in thread_local_system_indices. (always incremented by one
// when prepare_to_next_thread_local is called. (We prepared up to index 0 above)
let mut next_thread_local_index = 0;
{
// Prepare all system up to and including the first thread local system. This will return
// the range of systems to run, up to but NOT including the next thread local
let prepared_system_range = self.prepare_to_next_thread_local(
world,
systems,
stage_changed,
next_thread_local_index,
);
// Run everything up to the thread local system
self.run_systems(
world,
resources,
systems,
prepared_system_range,
&*compute_pool,
);
}
loop {
// Bail if we have no more thread local systems
if next_thread_local_index >= self.thread_local_system_indices.len() {
break;
}
// Run the thread local system at the end of the range of systems we just processed
let thread_local_system_index =
self.thread_local_system_indices[next_thread_local_index];
{
// if a thread local system is ready to run, run it exclusively on the main thread
let system = systems[thread_local_system_index].as_mut();
#[cfg(feature = "trace")]
let system_span = bevy_utils::tracing::info_span!(
"thread_local_system",
name = system.name().as_ref()
);
#[cfg(feature = "trace")]
let _system_guard = system_span.enter();
system.run((), world, resources);
system.run_thread_local(world, resources);
}
// Now that the previous thread local system has run, time to advance to the next one
next_thread_local_index += 1;
// Prepare all systems up to and including the next thread local system. This will
// return the range of systems to run, up to but NOT including the next thread local
let run_ready_system_index_range = self.prepare_to_next_thread_local(
world,
systems,
stage_changed,
next_thread_local_index,
);
self.run_systems(
world,
resources,
systems,
run_ready_system_index_range,
&*compute_pool,
);
}
// "flush"
for system in systems.iter_mut() {
match system.thread_local_execution() {
ThreadLocalExecution::NextFlush => {
#[cfg(feature = "trace")]
let system_span =
bevy_utils::tracing::info_span!("system", name = system.name().as_ref());
#[cfg(feature = "trace")]
let _system_guard = system_span.enter();
system.run_thread_local(world, resources);
}
ThreadLocalExecution::Immediate => { /* already ran */ }
}
}
// If world's archetypes_generation is the same as it was before running any systems then
// we can assume that all systems have correct archetype accesses.
if start_archetypes_generation == world.archetypes_generation() {
self.last_archetypes_generation = world.archetypes_generation();
}
}
}

28
crates/bevy_ecs/src/schedule/state.rs
View File

@ -1,4 +1,4 @@
use crate::{Resource, Resources, Stage, System, SystemStage, World};
use crate::{Resource, Resources, Stage, SystemDescriptor, SystemStage, World};
use bevy_utils::HashMap;
use std::{mem::Discriminant, ops::Deref};
use thiserror::Error;
@ -66,31 +66,19 @@ impl<T> StateStage<T> {
self
}
pub fn on_state_enter<S: System<In = (), Out = ()>>(
&mut self,
state: T,
system: S,
) -> &mut Self {
pub fn on_state_enter(&mut self, state: T, system: impl Into<SystemDescriptor>) -> &mut Self {
self.enter_stage(state, |system_stage: &mut SystemStage| {
system_stage.add_system(system)
})
}
pub fn on_state_exit<S: System<In = (), Out = ()>>(
&mut self,
state: T,
system: S,
) -> &mut Self {
pub fn on_state_exit(&mut self, state: T, system: impl Into<SystemDescriptor>) -> &mut Self {
self.exit_stage(state, |system_stage: &mut SystemStage| {
system_stage.add_system(system)
})
}
pub fn on_state_update<S: System<In = (), Out = ()>>(
&mut self,
state: T,
system: S,
) -> &mut Self {
pub fn on_state_update(&mut self, state: T, system: impl Into<SystemDescriptor>) -> &mut Self {
self.update_stage(state, |system_stage: &mut SystemStage| {
system_stage.add_system(system)
})
@ -150,14 +138,6 @@ impl<T> StateStage<T> {
#[allow(clippy::mem_discriminant_non_enum)]
impl<T: Resource + Clone> Stage for StateStage<T> {
fn initialize(&mut self, world: &mut World, resources: &mut Resources) {
for state_stages in self.stages.values_mut() {
state_stages.enter.initialize(world, resources);
state_stages.update.initialize(world, resources);
state_stages.exit.initialize(world, resources);
}
}
fn run(&mut self, world: &mut World, resources: &mut Resources) {
let current_stage = loop {
let (next_stage, current_stage) = {

178
crates/bevy_ecs/src/schedule/system_container.rs Normal file
View File

@ -0,0 +1,178 @@
use std::{borrow::Cow, ptr::NonNull};
use crate::{ExclusiveSystem, ExclusiveSystemDescriptor, ParallelSystemDescriptor, System};
pub(super) trait SystemContainer {
fn display_name(&self) -> Cow<'static, str>;
fn dependencies(&self) -> &[usize];
fn set_dependencies(&mut self, dependencies: impl IntoIterator<Item = usize>);
fn system_set(&self) -> usize;
fn label(&self) -> &Option<Cow<'static, str>>;
fn before(&self) -> &[Cow<'static, str>];
fn after(&self) -> &[Cow<'static, str>];
fn is_compatible(&self, other: &Self) -> bool;
}
pub(super) struct ExclusiveSystemContainer {
system: Box<dyn ExclusiveSystem>,
dependencies: Vec<usize>,
set: usize,
label: Option<Cow<'static, str>>,
before: Vec<Cow<'static, str>>,
after: Vec<Cow<'static, str>>,
}
impl ExclusiveSystemContainer {
pub fn from_descriptor(descriptor: ExclusiveSystemDescriptor, set: usize) -> Self {
ExclusiveSystemContainer {
system: descriptor.system,
dependencies: Vec::new(),
set,
label: descriptor.label,
before: descriptor.before,
after: descriptor.after,
}
}
pub fn system_mut(&mut self) -> &mut Box<dyn ExclusiveSystem> {
&mut self.system
}
}
impl SystemContainer for ExclusiveSystemContainer {
fn display_name(&self) -> Cow<'static, str> {
self.label
.as_ref()
.cloned()
.unwrap_or_else(|| self.system.name())
}
fn dependencies(&self) -> &[usize] {
&self.dependencies
}
fn set_dependencies(&mut self, dependencies: impl IntoIterator<Item = usize>) {
self.dependencies.clear();
self.dependencies.extend(dependencies);
}
fn system_set(&self) -> usize {
self.set
}
fn label(&self) -> &Option<Cow<'static, str>> {
&self.label
}
fn before(&self) -> &[Cow<'static, str>] {
&self.before
}
fn after(&self) -> &[Cow<'static, str>] {
&self.after
}
fn is_compatible(&self, _: &Self) -> bool {
false
}
}
pub struct ParallelSystemContainer {
system: NonNull<dyn System<In = (), Out = ()>>,
pub(crate) should_run: bool,
dependencies: Vec<usize>,
set: usize,
label: Option<Cow<'static, str>>,
before: Vec<Cow<'static, str>>,
after: Vec<Cow<'static, str>>,
}
impl SystemContainer for ParallelSystemContainer {
fn display_name(&self) -> Cow<'static, str> {
self.label
.as_ref()
.cloned()
.unwrap_or_else(|| self.system().name())
}
fn dependencies(&self) -> &[usize] {
&self.dependencies
}
fn set_dependencies(&mut self, dependencies: impl IntoIterator<Item = usize>) {
self.dependencies.clear();
self.dependencies.extend(dependencies);
}
fn system_set(&self) -> usize {
self.set
}
fn label(&self) -> &Option<Cow<'static, str>> {
&self.label
}
fn before(&self) -> &[Cow<'static, str>] {
&self.before
}
fn after(&self) -> &[Cow<'static, str>] {
&self.after
}
fn is_compatible(&self, other: &Self) -> bool {
self.system()
.component_access()
.is_compatible(other.system().component_access())
&& self
.system()
.resource_access()
.is_compatible(other.system().resource_access())
}
}
unsafe impl Send for ParallelSystemContainer {}
unsafe impl Sync for ParallelSystemContainer {}
impl ParallelSystemContainer {
pub(crate) fn from_descriptor(descriptor: ParallelSystemDescriptor, set: usize) -> Self {
ParallelSystemContainer {
system: unsafe { NonNull::new_unchecked(Box::into_raw(descriptor.system)) },
should_run: false,
set,
dependencies: Vec::new(),
label: descriptor.label,
before: descriptor.before,
after: descriptor.after,
}
}
pub fn display_name(&self) -> Cow<'static, str> {
SystemContainer::display_name(self)
}
pub fn system(&self) -> &dyn System<In = (), Out = ()> {
// SAFE: statically enforced shared access.
unsafe { self.system.as_ref() }
}
pub fn system_mut(&mut self) -> &mut dyn System<In = (), Out = ()> {
// SAFE: statically enforced exclusive access.
unsafe { self.system.as_mut() }
}
/// # Safety
/// Ensure no other borrows exist along with this one.
#[allow(clippy::mut_from_ref)]
pub unsafe fn system_mut_unsafe(&self) -> &mut dyn System<In = (), Out = ()> {
&mut *self.system.as_ptr()
}
pub fn should_run(&self) -> bool {
self.should_run
}
pub fn dependencies(&self) -> &[usize] {
&self.dependencies
}
}

254
crates/bevy_ecs/src/schedule/system_descriptor.rs Normal file
View File

@ -0,0 +1,254 @@
use crate::{BoxedSystem, ExclusiveSystem, ExclusiveSystemCoerced, ExclusiveSystemFn, System};
use std::borrow::Cow;
/// Encapsulates a system and information on when it run in a `SystemStage`.
///
/// Systems can be inserted into 4 different groups within the stage:
/// * Parallel, accepts non-exclusive systems.
/// * At start, accepts exclusive systems; runs before parallel systems.
/// * Before commands, accepts exclusive systems; runs after parallel systems, but before their
/// command buffers are applied.
/// * At end, accepts exclusive systems; runs after parallel systems' command buffers have
/// been applied.
///
/// All systems can have a label attached to them; other systems in the same group can then specify
/// that they have to run before or after the system with that label.
///
/// # Example
/// ```rust
/// # use bevy_ecs::prelude::*;
/// # fn do_something() {}
/// # fn do_the_other_thing() {}
/// # fn do_something_else() {}
/// SystemStage::parallel()
/// .with_system(do_something.system().label("something"))
/// .with_system(do_the_other_thing.system().after("something"))
/// .with_system(do_something_else.exclusive_system().at_end());
/// ```
pub enum SystemDescriptor {
Parallel(ParallelSystemDescriptor),
Exclusive(ExclusiveSystemDescriptor),
}
impl From<ParallelSystemDescriptor> for SystemDescriptor {
fn from(descriptor: ParallelSystemDescriptor) -> Self {
SystemDescriptor::Parallel(descriptor)
}
}
impl<S> From<S> for SystemDescriptor
where
S: System<In = (), Out = ()>,
{
fn from(system: S) -> Self {
new_parallel_descriptor(Box::new(system)).into()
}
}
impl From<BoxedSystem<(), ()>> for SystemDescriptor {
fn from(system: BoxedSystem<(), ()>) -> Self {
new_parallel_descriptor(system).into()
}
}
impl From<ExclusiveSystemDescriptor> for SystemDescriptor {
fn from(descriptor: ExclusiveSystemDescriptor) -> Self {
SystemDescriptor::Exclusive(descriptor)
}
}
impl From<ExclusiveSystemFn> for SystemDescriptor {
fn from(system: ExclusiveSystemFn) -> Self {
new_exclusive_descriptor(Box::new(system)).into()
}
}
impl From<ExclusiveSystemCoerced> for SystemDescriptor {
fn from(system: ExclusiveSystemCoerced) -> Self {
new_exclusive_descriptor(Box::new(system)).into()
}
}
/// Encapsulates a parallel system and information on when it run in a `SystemStage`.
pub struct ParallelSystemDescriptor {
pub(crate) system: BoxedSystem<(), ()>,
pub(crate) label: Option<Cow<'static, str>>,
pub(crate) before: Vec<Cow<'static, str>>,
pub(crate) after: Vec<Cow<'static, str>>,
}
fn new_parallel_descriptor(system: BoxedSystem<(), ()>) -> ParallelSystemDescriptor {
ParallelSystemDescriptor {
system,
label: None,
before: Vec::new(),
after: Vec::new(),
}
}
pub trait ParallelSystemDescriptorCoercion {
/// Assigns a label to the system.
fn label(self, label: impl Into<Cow<'static, str>>) -> ParallelSystemDescriptor;
/// Specifies that the system should run before the system with given label.
fn before(self, label: impl Into<Cow<'static, str>>) -> ParallelSystemDescriptor;
/// Specifies that the system should run after the system with given label.
fn after(self, label: impl Into<Cow<'static, str>>) -> ParallelSystemDescriptor;
}
impl ParallelSystemDescriptorCoercion for ParallelSystemDescriptor {
fn label(mut self, label: impl Into<Cow<'static, str>>) -> ParallelSystemDescriptor {
self.label = Some(label.into());
self
}
fn before(mut self, label: impl Into<Cow<'static, str>>) -> ParallelSystemDescriptor {
self.before.push(label.into());
self
}
fn after(mut self, label: impl Into<Cow<'static, str>>) -> ParallelSystemDescriptor {
self.after.push(label.into());
self
}
}
impl<S> ParallelSystemDescriptorCoercion for S
where
S: System<In = (), Out = ()>,
{
fn label(self, label: impl Into<Cow<'static, str>>) -> ParallelSystemDescriptor {
new_parallel_descriptor(Box::new(self)).label(label)
}
fn before(self, label: impl Into<Cow<'static, str>>) -> ParallelSystemDescriptor {
new_parallel_descriptor(Box::new(self)).before(label)
}
fn after(self, label: impl Into<Cow<'static, str>>) -> ParallelSystemDescriptor {
new_parallel_descriptor(Box::new(self)).after(label)
}
}
impl ParallelSystemDescriptorCoercion for BoxedSystem<(), ()> {
fn label(self, label: impl Into<Cow<'static, str>>) -> ParallelSystemDescriptor {
new_parallel_descriptor(self).label(label)
}
fn before(self, label: impl Into<Cow<'static, str>>) -> ParallelSystemDescriptor {
new_parallel_descriptor(self).before(label)
}
fn after(self, label: impl Into<Cow<'static, str>>) -> ParallelSystemDescriptor {
new_parallel_descriptor(self).after(label)
}
}
#[derive(Debug, Clone, Copy)]
pub(crate) enum InsertionPoint {
AtStart,
BeforeCommands,
AtEnd,
}
/// Encapsulates an exclusive system and information on when it run in a `SystemStage`.
pub struct ExclusiveSystemDescriptor {
pub(crate) system: Box<dyn ExclusiveSystem>,
pub(crate) label: Option<Cow<'static, str>>,
pub(crate) before: Vec<Cow<'static, str>>,
pub(crate) after: Vec<Cow<'static, str>>,
pub(crate) insertion_point: InsertionPoint,
}
fn new_exclusive_descriptor(system: Box<dyn ExclusiveSystem>) -> ExclusiveSystemDescriptor {
ExclusiveSystemDescriptor {
system,
label: None,
before: Vec::new(),
after: Vec::new(),
insertion_point: InsertionPoint::AtStart,
}
}
pub trait ExclusiveSystemDescriptorCoercion {
/// Assigns a label to the system.
fn label(self, label: impl Into<Cow<'static, str>>) -> ExclusiveSystemDescriptor;
/// Specifies that the system should run before the system with given label.
fn before(self, label: impl Into<Cow<'static, str>>) -> ExclusiveSystemDescriptor;
/// Specifies that the system should run after the system with given label.
fn after(self, label: impl Into<Cow<'static, str>>) -> ExclusiveSystemDescriptor;
/// Specifies that the system should run with other exclusive systems at the start of stage.
fn at_start(self) -> ExclusiveSystemDescriptor;
/// Specifies that the system should run with other exclusive systems after the parallel
/// systems and before command buffer application.
fn before_commands(self) -> ExclusiveSystemDescriptor;
/// Specifies that the system should run with other exclusive systems at the end of stage.
fn at_end(self) -> ExclusiveSystemDescriptor;
}
impl ExclusiveSystemDescriptorCoercion for ExclusiveSystemDescriptor {
fn label(mut self, label: impl Into<Cow<'static, str>>) -> ExclusiveSystemDescriptor {
self.label = Some(label.into());
self
}
fn before(mut self, label: impl Into<Cow<'static, str>>) -> ExclusiveSystemDescriptor {
self.before.push(label.into());
self
}
fn after(mut self, label: impl Into<Cow<'static, str>>) -> ExclusiveSystemDescriptor {
self.after.push(label.into());
self
}
fn at_start(mut self) -> ExclusiveSystemDescriptor {
self.insertion_point = InsertionPoint::AtStart;
self
}
fn before_commands(mut self) -> ExclusiveSystemDescriptor {
self.insertion_point = InsertionPoint::BeforeCommands;
self
}
fn at_end(mut self) -> ExclusiveSystemDescriptor {
self.insertion_point = InsertionPoint::AtEnd;
self
}
}
impl<T> ExclusiveSystemDescriptorCoercion for T
where
T: ExclusiveSystem + 'static,
{
fn label(self, label: impl Into<Cow<'static, str>>) -> ExclusiveSystemDescriptor {
new_exclusive_descriptor(Box::new(self)).label(label)
}
fn before(self, label: impl Into<Cow<'static, str>>) -> ExclusiveSystemDescriptor {
new_exclusive_descriptor(Box::new(self)).before(label)
}
fn after(self, label: impl Into<Cow<'static, str>>) -> ExclusiveSystemDescriptor {
new_exclusive_descriptor(Box::new(self)).after(label)
}
fn at_start(self) -> ExclusiveSystemDescriptor {
new_exclusive_descriptor(Box::new(self)).at_start()
}
fn before_commands(self) -> ExclusiveSystemDescriptor {
new_exclusive_descriptor(Box::new(self)).before_commands()
}
fn at_end(self) -> ExclusiveSystemDescriptor {
new_exclusive_descriptor(Box::new(self)).at_end()
}
}

45
crates/bevy_ecs/src/schedule/system_set.rs Normal file
View File

@ -0,0 +1,45 @@
use crate::{RunCriteria, ShouldRun, System, SystemDescriptor};
/// Describes a group of systems sharing one run criterion.
pub struct SystemSet {
pub(crate) run_criteria: RunCriteria,
pub(crate) descriptors: Vec<SystemDescriptor>,
}
impl Default for SystemSet {
fn default() -> SystemSet {
SystemSet {
run_criteria: Default::default(),
descriptors: vec![],
}
}
}
impl SystemSet {
pub fn new() -> Self {
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 add_run_criteria<S: System<In = (), Out = ShouldRun>>(
&mut self,
system: S,
) -> &mut Self {
self.run_criteria.set(Box::new(system));
self
}
pub fn with_system(mut self, system: impl Into<SystemDescriptor>) -> Self {
self.add_system(system);
self
}
pub fn add_system(&mut self, system: impl Into<SystemDescriptor>) -> &mut Self {
self.descriptors.push(system.into());
self
}
}

125
crates/bevy_ecs/src/system/exclusive_system.rs Normal file
View File

@ -0,0 +1,125 @@
pub use super::Query;
use crate::{resource::Resources, system::SystemId, BoxedSystem, IntoSystem, System, World};
use std::borrow::Cow;
pub trait ExclusiveSystem: Send + Sync + 'static {
fn name(&self) -> Cow<'static, str>;
fn id(&self) -> SystemId;
fn run(&mut self, world: &mut World, resources: &mut Resources);
fn initialize(&mut self, world: &mut World, resources: &mut Resources);
}
pub struct ExclusiveSystemFn {
func: Box<dyn FnMut(&mut World, &mut Resources) + Send + Sync + 'static>,
name: Cow<'static, str>,
id: SystemId,
}
impl ExclusiveSystem for ExclusiveSystemFn {
fn name(&self) -> Cow<'static, str> {
self.name.clone()
}
fn id(&self) -> SystemId {
self.id
}
fn run(&mut self, world: &mut World, resources: &mut Resources) {
(self.func)(world, resources);
}
fn initialize(&mut self, _: &mut World, _: &mut Resources) {}
}
pub trait IntoExclusiveSystem<Params, SystemType> {
fn exclusive_system(self) -> SystemType;
}
impl<F> IntoExclusiveSystem<(&mut World, &mut Resources), ExclusiveSystemFn> for F
where
F: FnMut(&mut World, &mut Resources) + Send + Sync + 'static,
{
fn exclusive_system(self) -> ExclusiveSystemFn {
ExclusiveSystemFn {
func: Box::new(self),
name: core::any::type_name::<F>().into(),
id: SystemId::new(),
}
}
}
impl<F> IntoExclusiveSystem<(&mut Resources, &mut World), ExclusiveSystemFn> for F
where
F: FnMut(&mut Resources, &mut World) + Send + Sync + 'static,
{
fn exclusive_system(mut self) -> ExclusiveSystemFn {
ExclusiveSystemFn {
func: Box::new(move |world, resources| self(resources, world)),
name: core::any::type_name::<F>().into(),
id: SystemId::new(),
}
}
}
impl<F> IntoExclusiveSystem<&mut World, ExclusiveSystemFn> for F
where
F: FnMut(&mut World) + Send + Sync + 'static,
{
fn exclusive_system(mut self) -> ExclusiveSystemFn {
ExclusiveSystemFn {
func: Box::new(move |world, _| self(world)),
name: core::any::type_name::<F>().into(),
id: SystemId::new(),
}
}
}
impl<F> IntoExclusiveSystem<&mut Resources, ExclusiveSystemFn> for F
where
F: FnMut(&mut Resources) + Send + Sync + 'static,
{
fn exclusive_system(mut self) -> ExclusiveSystemFn {
ExclusiveSystemFn {
func: Box::new(move |_, resources| self(resources)),
name: core::any::type_name::<F>().into(),
id: SystemId::new(),
}
}
}
pub struct ExclusiveSystemCoerced {
system: BoxedSystem<(), ()>,
}
impl ExclusiveSystem for ExclusiveSystemCoerced {
fn name(&self) -> Cow<'static, str> {
self.system.name()
}
fn id(&self) -> SystemId {
self.system.id()
}
fn run(&mut self, world: &mut World, resources: &mut Resources) {
self.system.run((), world, resources);
}
fn initialize(&mut self, world: &mut World, resources: &mut Resources) {
self.system.initialize(world, resources);
}
}
impl<S, Params, SystemType> IntoExclusiveSystem<(Params, SystemType), ExclusiveSystemCoerced> for S
where
S: IntoSystem<Params, SystemType>,
SystemType: System<In = (), Out = ()>,
{
fn exclusive_system(self) -> ExclusiveSystemCoerced {
ExclusiveSystemCoerced {
system: Box::new(self.system()),
}
}
}

105
crates/bevy_ecs/src/system/into_system.rs
View File

@ -1,7 +1,7 @@
use super::system_param::FetchSystemParam;
use crate::{
ArchetypeComponent, Commands, QueryAccess, Resources, System, SystemId, SystemParam,
ThreadLocalExecution, TypeAccess, World,
TypeAccess, World,
};
use parking_lot::Mutex;
use std::{any::TypeId, borrow::Cow, cell::UnsafeCell, sync::Arc};
@ -10,7 +10,9 @@ pub struct SystemState {
pub(crate) id: SystemId,
pub(crate) name: Cow<'static, str>,
pub(crate) archetype_component_access: TypeAccess<ArchetypeComponent>,
pub(crate) component_access: TypeAccess<TypeId>,
pub(crate) resource_access: TypeAccess<TypeId>,
pub(crate) is_non_send: bool,
pub(crate) local_resource_access: TypeAccess<TypeId>,
pub(crate) query_archetype_component_accesses: Vec<TypeAccess<ArchetypeComponent>>,
pub(crate) query_accesses: Vec<Vec<QueryAccess>>,
@ -59,7 +61,7 @@ impl SystemState {
});
break;
}
self.archetype_component_access.union(component_access);
self.archetype_component_access.extend(component_access);
}
if let Some(conflict_index) = conflict_index {
let mut conflicts_with_index = None;
@ -82,8 +84,6 @@ impl SystemState {
pub struct FuncSystem<Out> {
func:
Box<dyn FnMut(&mut SystemState, &World, &Resources) -> Option<Out> + Send + Sync + 'static>,
thread_local_func:
Box<dyn FnMut(&mut SystemState, &mut World, &mut Resources) + Send + Sync + 'static>,
init_func: Box<dyn FnMut(&mut SystemState, &World, &mut Resources) + Send + Sync + 'static>,
state: SystemState,
}
@ -100,7 +100,7 @@ impl<Out: 'static> System for FuncSystem<Out> {
self.state.id
}
fn update(&mut self, world: &World) {
fn update_access(&mut self, world: &World) {
self.state.update(world);
}
@ -108,12 +108,16 @@ impl<Out: 'static> System for FuncSystem<Out> {
&self.state.archetype_component_access
}
fn component_access(&self) -> &TypeAccess<TypeId> {
&self.state.component_access
}
fn resource_access(&self) -> &TypeAccess<std::any::TypeId> {
&self.state.resource_access
}
fn thread_local_execution(&self) -> ThreadLocalExecution {
ThreadLocalExecution::NextFlush
fn is_non_send(&self) -> bool {
self.state.is_non_send
}
unsafe fn run_unsafe(
@ -125,8 +129,15 @@ impl<Out: 'static> System for FuncSystem<Out> {
(self.func)(&mut self.state, world, resources)
}
fn run_thread_local(&mut self, world: &mut World, resources: &mut Resources) {
(self.thread_local_func)(&mut self.state, world, resources)
fn apply_buffers(&mut self, world: &mut World, resources: &mut Resources) {
// SAFE: this is called with unique access to SystemState
unsafe {
(&mut *self.state.commands.get()).apply(world, resources);
}
if let Some(ref commands) = self.state.arc_commands {
let mut commands = commands.lock();
commands.apply(world, resources);
}
}
fn initialize(&mut self, world: &mut World, resources: &mut Resources) {
@ -138,8 +149,6 @@ pub struct InputFuncSystem<In, Out> {
func: Box<
dyn FnMut(In, &mut SystemState, &World, &Resources) -> Option<Out> + Send + Sync + 'static,
>,
thread_local_func:
Box<dyn FnMut(&mut SystemState, &mut World, &mut Resources) + Send + Sync + 'static>,
init_func: Box<dyn FnMut(&mut SystemState, &World, &mut Resources) + Send + Sync + 'static>,
state: SystemState,
}
@ -156,7 +165,7 @@ impl<In: 'static, Out: 'static> System for InputFuncSystem<In, Out> {
self.state.id
}
fn update(&mut self, world: &World) {
fn update_access(&mut self, world: &World) {
self.state.update(world);
}
@ -164,12 +173,16 @@ impl<In: 'static, Out: 'static> System for InputFuncSystem<In, Out> {
&self.state.archetype_component_access
}
fn component_access(&self) -> &TypeAccess<TypeId> {
&self.state.component_access
}
fn resource_access(&self) -> &TypeAccess<std::any::TypeId> {
&self.state.resource_access
}
fn thread_local_execution(&self) -> ThreadLocalExecution {
ThreadLocalExecution::NextFlush
fn is_non_send(&self) -> bool {
self.state.is_non_send
}
unsafe fn run_unsafe(
@ -181,8 +194,15 @@ impl<In: 'static, Out: 'static> System for InputFuncSystem<In, Out> {
(self.func)(input, &mut self.state, world, resources)
}
fn run_thread_local(&mut self, world: &mut World, resources: &mut Resources) {
(self.thread_local_func)(&mut self.state, world, resources)
fn apply_buffers(&mut self, world: &mut World, resources: &mut Resources) {
// SAFE: this is called with unique access to SystemState
unsafe {
(&mut *self.state.commands.get()).apply(world, resources);
}
if let Some(ref commands) = self.state.arc_commands {
let mut commands = commands.lock();
commands.apply(world, resources);
}
}
fn initialize(&mut self, world: &mut World, resources: &mut Resources) {
@ -190,7 +210,7 @@ impl<In: 'static, Out: 'static> System for InputFuncSystem<In, Out> {
}
}
pub trait IntoSystem<Params, SystemType: System> {
pub trait IntoSystem<Params, SystemType> {
fn system(self) -> SystemType;
}
@ -219,7 +239,9 @@ macro_rules! impl_into_system {
state: SystemState {
name: std::any::type_name::<Self>().into(),
archetype_component_access: TypeAccess::default(),
component_access: TypeAccess::default(),
resource_access: TypeAccess::default(),
is_non_send: false,
local_resource_access: TypeAccess::default(),
id: SystemId::new(),
commands: Default::default(),
@ -240,16 +262,6 @@ macro_rules! impl_into_system {
}
}
}),
thread_local_func: Box::new(|state, world, resources| {
// SAFE: this is called with unique access to SystemState
unsafe {
(&mut *state.commands.get()).apply(world, resources);
}
if let Some(ref commands) = state.arc_commands {
let mut commands = commands.lock();
commands.apply(world, resources);
}
}),
init_func: Box::new(|state, world, resources| {
<<($($param,)*) as SystemParam>::Fetch as FetchSystemParam>::init(state, world, resources)
}),
@ -271,7 +283,9 @@ macro_rules! impl_into_system {
state: SystemState {
name: std::any::type_name::<Self>().into(),
archetype_component_access: TypeAccess::default(),
component_access: TypeAccess::default(),
resource_access: TypeAccess::default(),
is_non_send: false,
local_resource_access: TypeAccess::default(),
id: SystemId::new(),
commands: Default::default(),
@ -292,16 +306,6 @@ macro_rules! impl_into_system {
}
}
}),
thread_local_func: Box::new(|state, world, resources| {
// SAFE: this is called with unique access to SystemState
unsafe {
(&mut *state.commands.get()).apply(world, resources);
}
if let Some(ref commands) = state.arc_commands {
let mut commands = commands.lock();
commands.apply(world, resources);
}
}),
init_func: Box::new(|state, world, resources| {
<<($($param,)*) as SystemParam>::Fetch as FetchSystemParam>::init(state, world, resources)
}),
@ -336,7 +340,8 @@ mod tests {
clear_trackers_system,
resource::{Res, ResMut, Resources},
schedule::Schedule,
ChangedRes, Entity, Local, Or, Query, QuerySet, System, SystemStage, With, World,
ChangedRes, Entity, IntoExclusiveSystem, Local, Or, Query, QuerySet, Stage, System,
SystemStage, With, World,
};
#[derive(Debug, Eq, PartialEq, Default)]
@ -456,17 +461,17 @@ mod tests {
schedule.add_stage("update", update);
schedule.add_stage(
"clear_trackers",
SystemStage::single(clear_trackers_system.system()),
SystemStage::single(clear_trackers_system.exclusive_system()),
);
schedule.initialize_and_run(&mut world, &mut resources);
schedule.run(&mut world, &mut resources);
assert_eq!(*(world.get::<i32>(ent).unwrap()), 1);
schedule.initialize_and_run(&mut world, &mut resources);
schedule.run(&mut world, &mut resources);
assert_eq!(*(world.get::<i32>(ent).unwrap()), 1);
*resources.get_mut::<bool>().unwrap() = true;
schedule.initialize_and_run(&mut world, &mut resources);
schedule.run(&mut world, &mut resources);
assert_eq!(*(world.get::<i32>(ent).unwrap()), 2);
}
@ -493,24 +498,24 @@ mod tests {
schedule.add_stage("update", update);
schedule.add_stage(
"clear_trackers",
SystemStage::single(clear_trackers_system.system()),
SystemStage::single(clear_trackers_system.exclusive_system()),
);
schedule.initialize_and_run(&mut world, &mut resources);
schedule.run(&mut world, &mut resources);
assert_eq!(*(world.get::<i32>(ent).unwrap()), 1);
schedule.initialize_and_run(&mut world, &mut resources);
schedule.run(&mut world, &mut resources);
assert_eq!(*(world.get::<i32>(ent).unwrap()), 1);
*resources.get_mut::<bool>().unwrap() = true;
schedule.initialize_and_run(&mut world, &mut resources);
schedule.run(&mut world, &mut resources);
assert_eq!(*(world.get::<i32>(ent).unwrap()), 2);
schedule.initialize_and_run(&mut world, &mut resources);
schedule.run(&mut world, &mut resources);
assert_eq!(*(world.get::<i32>(ent).unwrap()), 2);
*resources.get_mut::<i32>().unwrap() = 20;
schedule.initialize_and_run(&mut world, &mut resources);
schedule.run(&mut world, &mut resources);
assert_eq!(*(world.get::<i32>(ent).unwrap()), 3);
}
@ -581,7 +586,7 @@ mod tests {
let mut update = SystemStage::parallel();
update.add_system(system);
schedule.add_stage("update", update);
schedule.initialize_and_run(world, resources);
schedule.run(world, resources);
}
#[derive(Default)]
@ -595,7 +600,7 @@ mod tests {
resources.insert(BufferRes::default());
resources.insert(A);
resources.insert(B);
run_system(&mut world, &mut resources, sys.system());
run_system(&mut world, &mut resources, sys);
}
#[test]

72
crates/bevy_ecs/src/system/into_thread_local.rs
View File

@ -1,72 +0,0 @@
pub use super::Query;
use crate::{
resource::Resources,
system::{System, SystemId, ThreadLocalExecution},
ArchetypeComponent, IntoSystem, TypeAccess, World,
};
use std::{any::TypeId, borrow::Cow};
pub struct ThreadLocalSystemFn {
pub func: Box<dyn FnMut(&mut World, &mut Resources) + Send + Sync + 'static>,
pub resource_access: TypeAccess<TypeId>,
pub archetype_component_access: TypeAccess<ArchetypeComponent>,
pub name: Cow<'static, str>,
pub id: SystemId,
}
impl System for ThreadLocalSystemFn {
type In = ();
type Out = ();
fn name(&self) -> Cow<'static, str> {
self.name.clone()
}
fn update(&mut self, _world: &World) {}
fn archetype_component_access(&self) -> &TypeAccess<ArchetypeComponent> {
&self.archetype_component_access
}
fn resource_access(&self) -> &TypeAccess<TypeId> {
&self.resource_access
}
fn thread_local_execution(&self) -> ThreadLocalExecution {
ThreadLocalExecution::Immediate
}
unsafe fn run_unsafe(
&mut self,
_input: (),
_world: &World,
_resources: &Resources,
) -> Option<()> {
Some(())
}
fn run_thread_local(&mut self, world: &mut World, resources: &mut Resources) {
(self.func)(world, resources);
}
fn initialize(&mut self, _world: &mut World, _resources: &mut Resources) {}
fn id(&self) -> SystemId {
self.id
}
}
impl<F> IntoSystem<(&mut World, &mut Resources), ThreadLocalSystemFn> for F
where
F: FnMut(&mut World, &mut Resources) + Send + Sync + 'static,
{
fn system(mut self) -> ThreadLocalSystemFn {
ThreadLocalSystemFn {
func: Box::new(move |world, resources| (self)(world, resources)),
name: core::any::type_name::<F>().into(),
id: SystemId::new(),
resource_access: TypeAccess::default(),
archetype_component_access: TypeAccess::default(),
}
}
}

4
crates/bevy_ecs/src/system/mod.rs
View File

@ -1,6 +1,6 @@
mod commands;
mod exclusive_system;
mod into_system;
mod into_thread_local;
mod query;
#[allow(clippy::module_inception)]
mod system;
@ -8,8 +8,8 @@ mod system_chaining;
mod system_param;
pub use commands::*;
pub use exclusive_system::*;
pub use into_system::*;
pub use into_thread_local::*;
pub use query::*;
pub use system::*;
pub use system_chaining::*;

16
crates/bevy_ecs/src/system/system.rs
View File

@ -1,13 +1,6 @@
use crate::{ArchetypeComponent, Resources, TypeAccess, World};
use std::{any::TypeId, borrow::Cow};
/// Determines the strategy used to run the `run_thread_local` function in a [System]
#[derive(Copy, Clone, Eq, PartialEq, Debug)]
pub enum ThreadLocalExecution {
Immediate,
NextFlush,
}
#[derive(Debug, Copy, Clone, Eq, PartialEq, Hash)]
pub struct SystemId(pub usize);
@ -24,10 +17,11 @@ pub trait System: Send + Sync + 'static {
type Out;
fn name(&self) -> Cow<'static, str>;
fn id(&self) -> SystemId;
fn update(&mut self, world: &World);
fn update_access(&mut self, world: &World);
fn archetype_component_access(&self) -> &TypeAccess<ArchetypeComponent>;
fn component_access(&self) -> &TypeAccess<TypeId>;
fn resource_access(&self) -> &TypeAccess<TypeId>;
fn thread_local_execution(&self) -> ThreadLocalExecution;
fn is_non_send(&self) -> bool;
/// # Safety
/// This might access World and Resources in an unsafe manner. This should only be called in one of the following contexts:
/// 1. This system is the only system running on the given World and Resources across all threads
@ -47,8 +41,8 @@ pub trait System: Send + Sync + 'static {
// SAFE: world and resources are exclusively borrowed
unsafe { self.run_unsafe(input, world, resources) }
}
fn run_thread_local(&mut self, world: &mut World, resources: &mut Resources);
fn initialize(&mut self, _world: &mut World, _resources: &mut Resources);
fn apply_buffers(&mut self, world: &mut World, resources: &mut Resources);
fn initialize(&mut self, world: &mut World, resources: &mut Resources);
}
pub type BoxedSystem<In = (), Out = ()> = Box<dyn System<In = In, Out = Out>>;

42
crates/bevy_ecs/src/system/system_chaining.rs
View File

@ -1,6 +1,4 @@
use crate::{
ArchetypeComponent, Resources, System, SystemId, ThreadLocalExecution, TypeAccess, World,
};
use crate::{ArchetypeComponent, Resources, System, SystemId, TypeAccess, World};
use std::{any::TypeId, borrow::Cow};
pub struct ChainSystem<SystemA, SystemB> {
@ -8,8 +6,9 @@ pub struct ChainSystem<SystemA, SystemB> {
system_b: SystemB,
name: Cow<'static, str>,
id: SystemId,
pub(crate) archetype_component_access: TypeAccess<ArchetypeComponent>,
pub(crate) resource_access: TypeAccess<TypeId>,
archetype_component_access: TypeAccess<ArchetypeComponent>,
component_access: TypeAccess<TypeId>,
resource_access: TypeAccess<TypeId>,
}
impl<SystemA: System, SystemB: System<In = SystemA::Out>> System for ChainSystem<SystemA, SystemB> {
@ -24,27 +23,39 @@ impl<SystemA: System, SystemB: System<In = SystemA::Out>> System for ChainSystem
self.id
}
fn update(&mut self, world: &World) {
fn update_access(&mut self, world: &World) {
self.archetype_component_access.clear();
self.component_access.clear();
self.resource_access.clear();
self.system_a.update(world);
self.system_b.update(world);
self.system_a.update_access(world);
self.system_b.update_access(world);
self.archetype_component_access
.union(self.system_a.archetype_component_access());
self.resource_access.union(self.system_b.resource_access());
.extend(self.system_a.archetype_component_access());
self.archetype_component_access
.extend(self.system_b.archetype_component_access());
self.component_access
.extend(self.system_a.component_access());
self.component_access
.extend(self.system_b.component_access());
self.resource_access.extend(self.system_a.resource_access());
self.resource_access.extend(self.system_b.resource_access());
}
fn archetype_component_access(&self) -> &TypeAccess<ArchetypeComponent> {
&self.archetype_component_access
}
fn component_access(&self) -> &TypeAccess<TypeId> {
&self.component_access
}
fn resource_access(&self) -> &TypeAccess<TypeId> {
&self.resource_access
}
fn thread_local_execution(&self) -> ThreadLocalExecution {
ThreadLocalExecution::NextFlush
fn is_non_send(&self) -> bool {
self.system_a.is_non_send() || self.system_b.is_non_send()
}
unsafe fn run_unsafe(
@ -57,9 +68,9 @@ impl<SystemA: System, SystemB: System<In = SystemA::Out>> System for ChainSystem
self.system_b.run_unsafe(out, world, resources)
}
fn run_thread_local(&mut self, world: &mut World, resources: &mut Resources) {
self.system_a.run_thread_local(world, resources);
self.system_b.run_thread_local(world, resources);
fn apply_buffers(&mut self, world: &mut World, resources: &mut Resources) {
self.system_a.apply_buffers(world, resources);
self.system_b.apply_buffers(world, resources);
}
fn initialize(&mut self, world: &mut World, resources: &mut Resources) {
@ -86,6 +97,7 @@ where
system_a: self,
system_b: system,
archetype_component_access: Default::default(),
component_access: Default::default(),
resource_access: Default::default(),
id: SystemId::new(),
}

38
crates/bevy_ecs/src/system/system_param.rs
View File

@ -1,7 +1,7 @@
use crate::{
ArchetypeComponent, ChangedRes, Commands, Fetch, FromResources, Local, Or, Query, QueryAccess,
QueryFilter, QuerySet, QueryTuple, Res, ResMut, Resource, ResourceIndex, Resources,
SystemState, TypeAccess, World, WorldQuery,
ArchetypeComponent, ChangedRes, Commands, Fetch, FromResources, Local, NonSend, Or, Query,
QueryAccess, QueryFilter, QuerySet, QueryTuple, Res, ResMut, Resource, ResourceIndex,
Resources, SystemState, TypeAccess, World, WorldQuery,
};
use parking_lot::Mutex;
use std::{any::TypeId, marker::PhantomData, sync::Arc};
@ -49,6 +49,7 @@ impl<'a, Q: WorldQuery, F: QueryFilter> FetchSystemParam<'a> for FetchQuery<Q, F
.query_archetype_component_accesses
.push(TypeAccess::default());
let access = QueryAccess::union(vec![Q::Fetch::access(), F::access()]);
access.get_component_access(&mut system_state.component_access);
system_state.query_accesses.push(vec![access]);
system_state
.query_type_names
@ -83,7 +84,11 @@ impl<'a, T: QueryTuple> FetchSystemParam<'a> for FetchQuerySet<T> {
system_state
.query_archetype_component_accesses
.push(TypeAccess::default());
system_state.query_accesses.push(T::get_accesses());
let accesses = T::get_accesses();
for access in &accesses {
access.get_component_access(&mut system_state.component_access);
}
system_state.query_accesses.push(accesses);
system_state
.query_type_names
.push(std::any::type_name::<T>());
@ -291,6 +296,31 @@ impl<'a, T: Resource + FromResources> FetchSystemParam<'a> for FetchLocal<T> {
}
}
pub struct FetchNonSend<T>(PhantomData<T>);
impl<'a, T: Resource> SystemParam for NonSend<'a, T> {
type Fetch = FetchNonSend<T>;
}
impl<'a, T: Resource> FetchSystemParam<'a> for FetchNonSend<T> {
type Item = NonSend<'a, T>;
fn init(system_state: &mut SystemState, _world: &World, _resources: &mut Resources) {
// !Send systems run only on the main thread, so only one system
// at a time will ever access any thread-local resource.
system_state.is_non_send = true;
}
#[inline]
unsafe fn get_param(
_system_state: &'a SystemState,
_world: &'a World,
resources: &'a Resources,
) -> Option<Self::Item> {
Some(NonSend::new(resources))
}
}
pub struct FetchParamTuple<T>(PhantomData<T>);
pub struct FetchOr<T>(PhantomData<T>);

4
crates/bevy_gilrs/src/gilrs_system.rs
View File

@ -5,7 +5,7 @@ use bevy_input::{gamepad::GamepadEventRaw, prelude::*};
use gilrs::{EventType, Gilrs};
pub fn gilrs_event_startup_system(_world: &mut World, resources: &mut Resources) {
let gilrs = resources.get_thread_local::<Gilrs>().unwrap();
let gilrs = resources.get_non_send::<Gilrs>().unwrap();
let mut event = resources.get_mut::<Events<GamepadEventRaw>>().unwrap();
for (id, _) in gilrs.gamepads() {
event.send(GamepadEventRaw(
@ -16,7 +16,7 @@ pub fn gilrs_event_startup_system(_world: &mut World, resources: &mut Resources)
}
pub fn gilrs_event_system(_world: &mut World, resources: &mut Resources) {
let mut gilrs = resources.get_thread_local_mut::<Gilrs>().unwrap();
let mut gilrs = resources.get_non_send_mut::<Gilrs>().unwrap();
let mut event = resources.get_mut::<Events<GamepadEventRaw>>().unwrap();
event.update();
while let Some(gilrs_event) = gilrs.next_event() {

11
crates/bevy_gilrs/src/lib.rs
View File

@ -2,7 +2,7 @@ mod converter;
mod gilrs_system;
use bevy_app::{prelude::*, startup_stage::PRE_STARTUP};
use bevy_ecs::IntoSystem;
use bevy_ecs::IntoExclusiveSystem;
use bevy_utils::tracing::error;
use gilrs::GilrsBuilder;
use gilrs_system::{gilrs_event_startup_system, gilrs_event_system};
@ -18,9 +18,12 @@ impl Plugin for GilrsPlugin {
.build()
{
Ok(gilrs) => {
app.insert_thread_local_resource(gilrs)
.add_startup_system_to_stage(PRE_STARTUP, gilrs_event_startup_system.system())
.add_system_to_stage(stage::PRE_EVENT, gilrs_event_system.system());
app.insert_non_send_resource(gilrs)
.add_startup_system_to_stage(
PRE_STARTUP,
gilrs_event_startup_system.exclusive_system(),
)
.add_system_to_stage(stage::PRE_EVENT, gilrs_event_system.exclusive_system());
}
Err(err) => error!("Failed to start Gilrs. {}", err),
}

2
crates/bevy_log/src/lib.rs
View File

@ -71,7 +71,7 @@ impl Plugin for LogPlugin {
}
}))
.build();
app.resources_mut().insert_thread_local(guard);
app.resources_mut().insert_non_send(guard);
let subscriber = subscriber.with(chrome_layer);
bevy_utils::tracing::subscriber::set_global_default(subscriber)
.expect("Could not set global default tracing subscriber. If you've already set up a tracing subscriber, please disable LogPlugin from Bevy's DefaultPlugins");

4
crates/bevy_render/src/lib.rs
View File

@ -11,7 +11,7 @@ pub mod renderer;
pub mod shader;
pub mod texture;
use bevy_ecs::{IntoSystem, SystemStage};
use bevy_ecs::{IntoExclusiveSystem, IntoSystem, SystemStage};
use bevy_reflect::RegisterTypeBuilder;
use draw::Visible;
pub use once_cell;
@ -169,7 +169,7 @@ impl Plugin for RenderPlugin {
)
.add_system_to_stage(
stage::RENDER_GRAPH_SYSTEMS,
render_graph::render_graph_schedule_executor_system.system(),
render_graph::render_graph_schedule_executor_system.exclusive_system(),
)
.add_system_to_stage(stage::DRAW, pipeline::draw_render_pipelines_system.system())
.add_system_to_stage(

2
crates/bevy_render/src/render_graph/graph.rs
View File

@ -42,7 +42,7 @@ impl RenderGraph {
{
let schedule = self.system_node_schedule.as_mut().unwrap();
let stage = schedule.get_stage_mut::<SystemStage>("update").unwrap();
stage.add_system_boxed(node.get_system(&mut self.commands));
stage.add_system(node.get_system(&mut self.commands));
self.add_node(name, node)
}

4
crates/bevy_render/src/render_graph/system.rs
View File

@ -1,5 +1,5 @@
use super::RenderGraph;
use bevy_ecs::{Resources, World};
use bevy_ecs::{Resources, Stage, World};
pub fn render_graph_schedule_executor_system(world: &mut World, resources: &mut Resources) {
// run render graph systems
@ -10,7 +10,7 @@ pub fn render_graph_schedule_executor_system(world: &mut World, resources: &mut
commands.apply(world, resources);
if let Some(schedule) = system_schedule.as_mut() {
schedule.initialize_and_run(world, resources);
schedule.run(world, resources);
}
let mut render_graph = resources.get_mut::<RenderGraph>().unwrap();
if let Some(schedule) = system_schedule.take() {

4
crates/bevy_scene/src/lib.rs
View File

@ -5,7 +5,7 @@ mod scene_loader;
mod scene_spawner;
pub mod serde;
use bevy_ecs::{IntoSystem, SystemStage};
use bevy_ecs::{IntoExclusiveSystem, SystemStage};
pub use command::*;
pub use dynamic_scene::*;
pub use scene::*;
@ -33,6 +33,6 @@ impl Plugin for ScenePlugin {
.init_asset_loader::<SceneLoader>()
.init_resource::<SceneSpawner>()
.add_stage_after(stage::EVENT, SCENE_STAGE, SystemStage::parallel())
.add_system_to_stage(SCENE_STAGE, scene_spawner_system.system());
.add_system_to_stage(SCENE_STAGE, scene_spawner_system.exclusive_system());
}
}

3
crates/bevy_text/src/draw.rs
View File

@ -8,6 +8,7 @@ use bevy_render::{
renderer::{BindGroup, RenderResourceBindings, RenderResourceId},
};
use bevy_sprite::TextureAtlasSprite;
use bevy_utils::tracing::error;
use crate::{PositionedGlyph, TextSection};
use bevy_render::pipeline::IndexFormat;
@ -44,7 +45,7 @@ impl<'a> Drawable for DrawableText<'a> {
{
draw.set_vertex_buffer(0, vertex_attribute_buffer_id, 0);
} else {
println!("Could not find vertex buffer for `bevy_sprite::QUAD_HANDLE`.")
error!("Could not find vertex buffer for `bevy_sprite::QUAD_HANDLE`.")
}
let mut indices = 0..0;

8
crates/bevy_transform/src/hierarchy/hierarchy_maintenance_system.rs
View File

@ -71,7 +71,7 @@ pub fn parent_update_system(
mod test {
use super::*;
use crate::{hierarchy::BuildChildren, transform_propagate_system::transform_propagate_system};
use bevy_ecs::{IntoSystem, Resources, Schedule, SystemStage, World};
use bevy_ecs::{IntoSystem, Resources, Schedule, Stage, SystemStage, World};
#[test]
fn correct_children() {
@ -102,7 +102,7 @@ mod test {
});
let parent = parent.unwrap();
commands.apply(&mut world, &mut resources);
schedule.initialize_and_run(&mut world, &mut resources);
schedule.run(&mut world, &mut resources);
assert_eq!(
world
@ -118,7 +118,7 @@ mod test {
// Parent `e1` to `e2`.
(*world.get_mut::<Parent>(children[0]).unwrap()).0 = children[1];
schedule.initialize_and_run(&mut world, &mut resources);
schedule.run(&mut world, &mut resources);
assert_eq!(
world
@ -142,7 +142,7 @@ mod test {
world.despawn(children[0]).unwrap();
schedule.initialize_and_run(&mut world, &mut resources);
schedule.run(&mut world, &mut resources);
assert_eq!(
world

6
crates/bevy_transform/src/transform_propagate_system.rs
View File

@ -71,7 +71,7 @@ fn propagate_recursive(
mod test {
use super::*;
use crate::hierarchy::{parent_update_system, BuildChildren, BuildWorldChildren};
use bevy_ecs::{Resources, Schedule, SystemStage, World};
use bevy_ecs::{Resources, Schedule, Stage, SystemStage, World};
#[test]
fn did_propagate() {
@ -111,7 +111,7 @@ mod test {
))
.for_current_entity(|entity| children.push(entity));
});
schedule.initialize_and_run(&mut world, &mut resources);
schedule.run(&mut world, &mut resources);
assert_eq!(
*world.get::<GlobalTransform>(children[0]).unwrap(),
@ -159,7 +159,7 @@ mod test {
.for_current_entity(|entity| children.push(entity));
});
commands.apply(&mut world, &mut resources);
schedule.initialize_and_run(&mut world, &mut resources);
schedule.run(&mut world, &mut resources);
assert_eq!(
*world.get::<GlobalTransform>(children[0]).unwrap(),

15
crates/bevy_ui/src/lib.rs
View File

@ -20,7 +20,7 @@ pub mod prelude {
}
use bevy_app::prelude::*;
use bevy_ecs::{IntoSystem, SystemStage};
use bevy_ecs::{IntoSystem, ParallelSystemDescriptorCoercion, SystemStage};
use bevy_render::render_graph::RenderGraph;
use update::ui_z_system;
@ -31,6 +31,10 @@ pub mod stage {
pub const UI: &str = "ui";
}
pub mod system {
pub const FLEX: &str = "flex";
}
impl Plugin for UiPlugin {
fn build(&self, app: &mut AppBuilder) {
app.init_resource::<FlexSurface>()
@ -41,10 +45,13 @@ impl Plugin for UiPlugin {
)
.add_system_to_stage(bevy_app::stage::PRE_UPDATE, ui_focus_system.system())
// add these stages to front because these must run before transform update systems
.add_system_to_stage(stage::UI, widget::text_system.system())
.add_system_to_stage(stage::UI, widget::image_node_system.system())
.add_system_to_stage(stage::UI, widget::text_system.system().before(system::FLEX))
.add_system_to_stage(
stage::UI,
widget::image_node_system.system().before(system::FLEX),
)
.add_system_to_stage(stage::UI, flex_node_system.system().label(system::FLEX))
.add_system_to_stage(stage::UI, ui_z_system.system())
.add_system_to_stage(stage::UI, flex_node_system.system())
.add_system_to_stage(bevy_render::stage::DRAW, widget::draw_text_system.system());
let resources = app.resources();

4
crates/bevy_ui/src/update.rs
View File

@ -48,7 +48,7 @@ fn update_hierarchy(
}
#[cfg(test)]
mod tests {
use bevy_ecs::{Commands, IntoSystem, Resources, Schedule, SystemStage, World};
use bevy_ecs::{Commands, IntoSystem, Resources, Schedule, Stage, SystemStage, World};
use bevy_transform::{components::Transform, hierarchy::BuildChildren};
use crate::Node;
@ -118,7 +118,7 @@ mod tests {
let mut update_stage = SystemStage::parallel();
update_stage.add_system(ui_z_system.system());
schedule.add_stage("update", update_stage);
schedule.initialize_and_run(&mut world, &mut resources);
schedule.run(&mut world, &mut resources);
let mut actual_result = world
.query::<(&String, &Transform)>()

4
crates/bevy_wgpu/src/lib.rs
View File

@ -11,7 +11,7 @@ pub use wgpu_renderer::*;
pub use wgpu_resources::*;
use bevy_app::prelude::*;
use bevy_ecs::{IntoSystem, Resources, World};
use bevy_ecs::{IntoExclusiveSystem, IntoSystem, Resources, World};
use bevy_render::renderer::{shared_buffers_update_system, RenderResourceContext, SharedBuffers};
use renderer::WgpuRenderResourceContext;
@ -21,7 +21,7 @@ pub struct WgpuPlugin;
impl Plugin for WgpuPlugin {
fn build(&self, app: &mut AppBuilder) {
let render_system = get_wgpu_render_system(app.resources_mut());
app.add_system_to_stage(bevy_render::stage::RENDER, render_system.system())
app.add_system_to_stage(bevy_render::stage::RENDER, render_system.exclusive_system())
.add_system_to_stage(
bevy_render::stage::POST_RENDER,
shared_buffers_update_system.system(),

6
crates/bevy_winit/src/lib.rs
View File

@ -11,7 +11,7 @@ pub use winit_config::*;
pub use winit_windows::*;
use bevy_app::{prelude::*, AppExit, ManualEventReader};
use bevy_ecs::{IntoSystem, Resources, World};
use bevy_ecs::{IntoExclusiveSystem, Resources, World};
use bevy_math::{ivec2, Vec2};
use bevy_utils::tracing::{error, trace, warn};
use bevy_window::{
@ -41,7 +41,7 @@ impl Plugin for WinitPlugin {
fn build(&self, app: &mut AppBuilder) {
app.init_resource::<WinitWindows>()
.set_runner(winit_runner)
.add_system(change_window.system());
.add_system(change_window.exclusive_system());
}
}
@ -206,7 +206,7 @@ pub fn winit_runner_with(mut app: App, mut event_loop: EventLoop<()>) {
let mut create_window_event_reader = ManualEventReader::<CreateWindow>::default();
let mut app_exit_event_reader = ManualEventReader::<AppExit>::default();
app.resources.insert_thread_local(event_loop.create_proxy());
app.resources.insert_non_send(event_loop.create_proxy());
trace!("Entering winit event loop");

2
examples/scene/scene.rs
View File

@ -6,7 +6,7 @@ fn main() {
.add_plugins(DefaultPlugins)
.register_type::<ComponentA>()
.register_type::<ComponentB>()
.add_startup_system(save_scene_system.system())
.add_startup_system(save_scene_system.exclusive_system())
.add_startup_system(load_scene_system.system())
.add_startup_system(infotext_system.system())
.add_system(print_system.system())