From 00bdc183c316aa62c10ae9149e95116855676c6e Mon Sep 17 00:00:00 2001 From: Patrick Walton Date: Sat, 1 Aug 2026 23:03:46 -0700 Subject: [PATCH 01/12] Implement parallel contiguous iteration. --- crates/bevy_ecs/src/query/fetch.rs | 70 ++++++++++++++++++- crates/bevy_ecs/src/query/par_iter.rs | 90 +++++++++++++++++++++++++ crates/bevy_ecs/src/query/state.rs | 96 ++++++++++++++++++++++++++- crates/bevy_ecs/src/system/query.rs | 39 +++++++++-- 4 files changed, 288 insertions(+), 7 deletions(-) diff --git a/crates/bevy_ecs/src/query/fetch.rs b/crates/bevy_ecs/src/query/fetch.rs index d4397e814af83..385d74b964135 100644 --- a/crates/bevy_ecs/src/query/fetch.rs +++ b/crates/bevy_ecs/src/query/fetch.rs @@ -4247,9 +4247,11 @@ impl Copy for StorageSwitch {} #[cfg(test)] mod tests { + use core::sync::atomic::{AtomicUsize, Ordering}; + use super::*; use crate::change_detection::DetectChanges; - use crate::query::Without; + use crate::query::{QueryNotDenseError, Without}; use crate::system::{assert_is_system, Query}; use bevy_ecs::prelude::Schedule; use bevy_ecs_macros::QueryData; @@ -4661,4 +4663,70 @@ mod tests { assert_eq!(present, [true; 3]); } + + #[test] + fn contiguous_par_iter_success_test() { + #[derive(Component, PartialEq, Eq, Debug)] + pub struct C { + id: i32, + found: bool, + } + + #[derive(Component, PartialEq, Eq, Debug)] + pub struct D(i32); + + bevy_tasks::ComputeTaskPool::get_or_init(bevy_tasks::TaskPool::new); + + // Spawn a world with a couple of tables. + let mut world = World::new(); + for id in 0..100 { + world.spawn(C { id, found: false }); + } + for id in 100..150 { + world.spawn((C { id, found: false }, D(id))); + } + + // Check that the correct number of rows were matched contiguously. + let total_found = AtomicUsize::new(0); + let mut contiguous_query = world.query::<&mut C>(); + contiguous_query + .contiguous_par_iter_mut(&mut world) + .unwrap() + .for_each(|cs| { + for c in cs { + c.found = true; + total_found.fetch_add(1, Ordering::Relaxed); + } + }); + assert_eq!(total_found.load(Ordering::Relaxed), 150); + + // Check that the query updated every row. + let mut check_query = world.query::<&C>(); + assert!(check_query.iter(&world).all(|c| c.found)); + } + + #[test] + fn contiguous_par_iter_failure_test() { + #[derive(Component, Clone, Copy)] + struct C; + + #[derive(Component, Clone, Copy)] + #[component(storage = "SparseSet")] + struct S; + + bevy_tasks::ComputeTaskPool::get_or_init(bevy_tasks::TaskPool::new); + + // Spawn a world with a couple of tables. + let mut world = World::new(); + for id in 0..100 { + world.spawn((C, S)); + } + + // This query should fail, as queries over sparse sets aren't dense. + let mut sparse_query = world.query::<(&C, &S)>(); + assert!(matches!( + sparse_query.contiguous_par_iter(&world), + Err(QueryNotDenseError(_)) + )); + } } diff --git a/crates/bevy_ecs/src/query/par_iter.rs b/crates/bevy_ecs/src/query/par_iter.rs index 2391a06fae0fc..180c2cfbfb29d 100644 --- a/crates/bevy_ecs/src/query/par_iter.rs +++ b/crates/bevy_ecs/src/query/par_iter.rs @@ -2,6 +2,7 @@ use crate::{ batching::BatchingStrategy, change_detection::Tick, entity::{EntityEquivalent, UniqueEntityEquivalentVec}, + query::{ArchetypeFilter, ContiguousQueryData}, world::unsafe_world_cell::UnsafeWorldCell, }; @@ -155,6 +156,95 @@ impl<'w, 's, D: IterQueryData, F: QueryFilter> QueryParIter<'w, 's, D, F> { } } +pub struct QueryContiguousParIter<'w, 's, D, F> +where + D: ContiguousQueryData, + F: ArchetypeFilter, +{ + pub(crate) world: UnsafeWorldCell<'w>, + pub(crate) state: &'s QueryState, + pub(crate) last_run: Tick, + pub(crate) this_run: Tick, + pub(crate) batching_strategy: BatchingStrategy, +} + +impl<'w, 's, D, F> QueryContiguousParIter<'w, 's, D, F> +where + D: ContiguousQueryData, + F: ArchetypeFilter, +{ + /// Returns `None` if `query_state` is not dense, and hence not contiguously iterable. + pub(crate) fn new( + world: UnsafeWorldCell<'w>, + state: &'s QueryState, + last_run: Tick, + this_run: Tick, + ) -> Option { + state.is_dense.then(|| Self { + world, + state, + last_run, + this_run, + batching_strategy: BatchingStrategy::new(), + }) + } + + pub fn batching_strategy(mut self, strategy: BatchingStrategy) -> Self { + self.batching_strategy = strategy; + self + } + + #[inline] + pub fn for_each(self, func: impl Fn(D::Contiguous<'w, 's>) + Send + Sync + Clone) { + self.for_each_init(|| {}, |_, item| func(item)); + } + + pub fn for_each_init( + self, + init: impl Fn() -> T + Sync + Send + Clone, + func: impl Fn(&mut T, D::Contiguous<'w, 's>) + Send + Sync + Clone, + ) { + let func = |mut init, item| { + func(&mut init, item); + init + }; + + // TODO: wasm32/single threaded version + + { + let thread_count = bevy_tasks::ComputeTaskPool::get().thread_num(); + // TODO: thread_count <= 1 + { + let batch_size = self.get_batch_size(thread_count).max(1); + unsafe { + self.state.par_fold_contiguous_init_unchecked_manual( + init, + self.world, + batch_size, + func, + self.last_run, + self.this_run, + ); + } + } + } + } + + fn get_batch_size(&self, thread_count: usize) -> u32 { + let max_items = || { + let id_iter = self.state.matched_storage_ids.iter(); + let tables = unsafe { &self.world.world_metadata().storages().tables }; + id_iter + .map(|id| unsafe { tables[id.table_id].entity_count() }) + .max() + .map(|v| v as usize) + .unwrap_or(0) + }; + self.batching_strategy + .calc_batch_size(max_items, thread_count) as u32 + } +} + /// A parallel iterator over the unique query items generated from an [`Entity`] list. /// /// This struct is created by the [`Query::par_iter_many`] method. diff --git a/crates/bevy_ecs/src/query/state.rs b/crates/bevy_ecs/src/query/state.rs index 6b97387cd980c..9fbabf51ac66c 100644 --- a/crates/bevy_ecs/src/query/state.rs +++ b/crates/bevy_ecs/src/query/state.rs @@ -7,8 +7,8 @@ use crate::{ prelude::FromWorld, query::{ ArchetypeFilter, ContiguousQueryData, FilteredAccess, FilteredAccessSet, IterQueryData, - QueryCombinationIter, QueryContiguousIter, QueryIter, QueryNotDenseError, QueryParIter, - SingleEntityQueryData, WorldQuery, + QueryCombinationIter, QueryContiguousIter, QueryContiguousParIter, QueryIter, + QueryNotDenseError, QueryParIter, SingleEntityQueryData, WorldQuery, }, storage::TableId, system::Query, @@ -19,6 +19,7 @@ use crate::{ use crate::entity::UniqueEntityEquivalentSlice; use alloc::{format, vec::Vec}; +use arrayvec::ArrayVec; use bevy_utils::prelude::DebugName; use core::{fmt, ptr}; use fixedbitset::FixedBitSet; @@ -1503,6 +1504,30 @@ impl QueryState { self.query_mut(world).contiguous_iter_inner() } + #[inline] + pub fn contiguous_par_iter<'w, 's>( + &'s mut self, + world: &'w World, + ) -> Result, QueryNotDenseError> + where + D::ReadOnly: ContiguousQueryData, + F: ArchetypeFilter, + { + self.query(world).contiguous_par_iter_inner() + } + + #[inline] + pub fn contiguous_par_iter_mut<'w, 's>( + &'s mut self, + world: &'w mut World, + ) -> Result, QueryNotDenseError> + where + D: ContiguousQueryData, + F: ArchetypeFilter, + { + self.query_mut(world).contiguous_par_iter_inner() + } + /// Runs `func` on each query result in parallel for the given [`World`], where the last change and /// the current change tick are given. This is faster than the equivalent /// `iter()` method, but cannot be chained like a normal [`Iterator`]. @@ -1679,6 +1704,73 @@ impl QueryState { .fold(accum, &mut func); }); } + + pub(crate) unsafe fn par_fold_contiguous_init_unchecked_manual<'w, 's, T>( + &'s self, + init_accum: impl Fn() -> T + Send + Sync + Clone, + world: UnsafeWorldCell<'w>, + batch_size: u32, + func: impl Fn(T, D::Contiguous<'w, 's>) -> T + Send + Sync + Clone, + last_run: Tick, + this_run: Tick, + ) where + D: ContiguousQueryData, + F: ArchetypeFilter, + { + debug_assert!(self.is_dense); + + bevy_tasks::ComputeTaskPool::get().scope(|scope| { + // SAFETY: We only access table data that has been registered in `self.component_access`. + let tables = unsafe { &world.storages().tables }; + let mut batch_queue = ArrayVec::new(); + let mut queue_entity_count = 0; + + let submit_batch_queue = |queue: ArrayVec| { + let (func, init_accum) = (func.clone(), init_accum.clone()); + scope.spawn(async move { + #[cfg(feature = "trace")] + let _span = self.par_iter_span.enter(); + let tables = unsafe { &world.storages().tables }; + let mut fetch = D::init_fetch(world, &self.fetch_state, last_run, this_run); + let mut accum = init_accum(); + for table_id in queue { + let table = tables.get(table_id).expect("Table must be present"); + D::set_table(&mut fetch, &self.fetch_state, table); + let item = + D::fetch_contiguous(&self.fetch_state, &mut fetch, table.entities()); + accum = func(accum, item); + } + }); + }; + + for storage_id in &self.matched_storage_ids { + let table_id = storage_id.table_id; + let count = tables[table_id].entity_count(); + + // Skip empty tables. + if count == 0 { + continue; + } + // Immediately submit large storage. + if count >= batch_size { + submit_batch_queue(TryFrom::try_from(&[table_id][..]).unwrap()); + continue; + } + // Merge small tables. + batch_queue.push(table_id); + queue_entity_count += count; + + // Submit batch queue. + if queue_entity_count >= batch_size || batch_queue.is_full() { + submit_batch_queue(core::mem::take(&mut batch_queue)); + queue_entity_count = 0; + } + } + if !batch_queue.is_empty() { + submit_batch_queue(batch_queue); + } + }); + } } impl QueryState { diff --git a/crates/bevy_ecs/src/system/query.rs b/crates/bevy_ecs/src/system/query.rs index aaa31167388f4..41ed1207fa00a 100644 --- a/crates/bevy_ecs/src/system/query.rs +++ b/crates/bevy_ecs/src/system/query.rs @@ -6,10 +6,10 @@ use crate::{ entity::{Entity, EntityEquivalent, EntitySet, UniqueEntityArray}, query::{ ArchetypeFilter, ContiguousQueryData, DebugCheckedUnwrap, IterQueryData, NopWorldQuery, - QueryCombinationIter, QueryContiguousIter, QueryData, QueryEntityError, QueryFilter, - QueryIter, QueryManyIter, QueryManyUniqueIter, QueryNotDenseError, QueryParIter, - QueryParManyIter, QueryParManyUniqueIter, QuerySingleError, QueryState, ROQueryItem, - ReadOnlyQueryData, SingleEntityQueryData, + QueryCombinationIter, QueryContiguousIter, QueryContiguousParIter, QueryData, + QueryEntityError, QueryFilter, QueryIter, QueryManyIter, QueryManyUniqueIter, + QueryNotDenseError, QueryParIter, QueryParManyIter, QueryParManyUniqueIter, + QuerySingleError, QueryState, ROQueryItem, ReadOnlyQueryData, SingleEntityQueryData, }, world::unsafe_world_cell::UnsafeWorldCell, }; @@ -1540,6 +1540,37 @@ impl<'w, 's, D: QueryData, F: QueryFilter> Query<'w, 's, D, F> { .ok_or(QueryNotDenseError(DebugName::type_name::())) } + pub fn contiguous_par_iter( + &self, + ) -> Result, QueryNotDenseError> + where + D::ReadOnly: ContiguousQueryData, + F: ArchetypeFilter, + { + self.as_readonly().contiguous_par_iter_inner() + } + + pub fn contiguous_par_iter_mut( + &mut self, + ) -> Result, QueryNotDenseError> + where + D: ContiguousQueryData, + F: ArchetypeFilter, + { + self.reborrow().contiguous_par_iter_inner() + } + + pub fn contiguous_par_iter_inner( + self, + ) -> Result, QueryNotDenseError> + where + D: ContiguousQueryData, + F: ArchetypeFilter, + { + QueryContiguousParIter::new(self.world, self.state, self.last_run, self.this_run) + .ok_or(QueryNotDenseError(DebugName::type_name::())) + } + /// Returns the read-only query item for the given [`Entity`]. /// /// In case of a nonexisting entity or mismatched component, a [`QueryEntityError`] is returned instead. From 3658e0f00f6ad657ed7ccc96d0ee35e86ecfb545 Mon Sep 17 00:00:00 2001 From: Patrick Walton Date: Sun, 2 Aug 2026 11:05:08 -0700 Subject: [PATCH 02/12] Add documentation --- crates/bevy_ecs/src/query/fetch.rs | 2 +- crates/bevy_ecs/src/query/par_iter.rs | 117 ++++++++++++++++++++++---- crates/bevy_ecs/src/query/state.rs | 70 ++++++++++++++- crates/bevy_ecs/src/system/query.rs | 74 ++++++++++++++++ 4 files changed, 244 insertions(+), 19 deletions(-) diff --git a/crates/bevy_ecs/src/query/fetch.rs b/crates/bevy_ecs/src/query/fetch.rs index 385d74b964135..05a07aa580ec4 100644 --- a/crates/bevy_ecs/src/query/fetch.rs +++ b/crates/bevy_ecs/src/query/fetch.rs @@ -4718,7 +4718,7 @@ mod tests { // Spawn a world with a couple of tables. let mut world = World::new(); - for id in 0..100 { + for _ in 0..100 { world.spawn((C, S)); } diff --git a/crates/bevy_ecs/src/query/par_iter.rs b/crates/bevy_ecs/src/query/par_iter.rs index 180c2cfbfb29d..9f90078190d18 100644 --- a/crates/bevy_ecs/src/query/par_iter.rs +++ b/crates/bevy_ecs/src/query/par_iter.rs @@ -2,7 +2,7 @@ use crate::{ batching::BatchingStrategy, change_detection::Tick, entity::{EntityEquivalent, UniqueEntityEquivalentVec}, - query::{ArchetypeFilter, ContiguousQueryData}, + query::{ArchetypeFilter, ContiguousQueryData, QueryContiguousIter}, world::unsafe_world_cell::UnsafeWorldCell, }; @@ -156,15 +156,29 @@ impl<'w, 's, D: IterQueryData, F: QueryFilter> QueryParIter<'w, 's, D, F> { } } +/// A parallel iterator over contiguous query results on a +/// [`Query`](crate::system::Query). +/// +/// The +/// [`Query::contiguous_par_iter`](crate::system::Query::contiguous_par_iter) +/// and +/// [`Query::contiguous_par_iter_mut`](crate::system::Query::contiguous_par_iter_mut) +/// methods create instances of this structure. pub struct QueryContiguousParIter<'w, 's, D, F> where D: ContiguousQueryData, F: ArchetypeFilter, { + /// A reference to the world that contains the components that this query + /// iterates over. pub(crate) world: UnsafeWorldCell<'w>, + /// Scoped access to the world state. pub(crate) state: &'s QueryState, + /// The tick that corresponds to the previous time this query ran. pub(crate) last_run: Tick, + /// The tick that corresponds to the current run of the query. pub(crate) this_run: Tick, + /// How matched rows are to be divided among worker threads. pub(crate) batching_strategy: BatchingStrategy, } @@ -189,16 +203,61 @@ where }) } + /// Changes the batching strategy used when iterating. + /// + /// For more information on how this affects the resultant iteration, see + /// [`BatchingStrategy`]. pub fn batching_strategy(mut self, strategy: BatchingStrategy) -> Self { self.batching_strategy = strategy; self } + /// Runs `func` on each contiguous chunk of query results in parallel. + /// + /// # Panics + /// If the [`ComputeTaskPool`] is not initialized. If using this from a + /// query that is being initialized and run from the ECS scheduler, this + /// should never panic. + /// + /// [`ComputeTaskPool`]: bevy_tasks::ComputeTaskPool #[inline] pub fn for_each(self, func: impl Fn(D::Contiguous<'w, 's>) + Send + Sync + Clone) { self.for_each_init(|| {}, |_, item| func(item)); } + /// Runs `func` on each query result in parallel on a value returned by `init`. + /// + /// `init` may be called multiple times per thread, and the values returned may be discarded between tasks on any given thread. + /// Callers should avoid using this function as if it were a parallel version + /// of [`Iterator::fold`]. + /// + /// # Example + /// + /// ``` + /// use bevy_utils::Parallel; + /// use crate::{bevy_ecs::prelude::Component, bevy_ecs::system::Query}; + /// #[derive(Component)] + /// struct T; + /// fn system(query: Query<&T>){ + /// let mut queue: Parallel = Parallel::default(); + /// // queue.borrow_local_mut() will get or create a thread_local queue for each task/thread; + /// query.contiguous_par_iter().for_each_init(|| queue.borrow_local_mut(),|local_queue, items| { + /// for _ in items { + /// **local_queue += 1; + /// } + /// }); + /// + /// // collect value from every thread + /// let entity_count: usize = queue.iter_mut().map(|v| *v).sum(); + /// } + /// ``` + /// + /// # Panics + /// If the [`ComputeTaskPool`] is not initialized. If using this from a + /// query that is being initialized and run from the ECS scheduler, this + /// should never panic. + /// + /// [`ComputeTaskPool`]: bevy_tasks::ComputeTaskPool pub fn for_each_init( self, init: impl Fn() -> T + Sync + Send + Clone, @@ -209,33 +268,59 @@ where init }; - // TODO: wasm32/single threaded version + #[cfg(any(target_arch = "wasm32", not(feature = "multi_threaded")))] + { + QueryContiguousIter::new(self.world, self.state, self.last_run, self_this_run) + .unwrap() + .fold(init, func) + } + #[cfg(all(not(target_arch = "wasm32"), feature = "multi_threaded"))] { let thread_count = bevy_tasks::ComputeTaskPool::get().thread_num(); - // TODO: thread_count <= 1 - { - let batch_size = self.get_batch_size(thread_count).max(1); - unsafe { - self.state.par_fold_contiguous_init_unchecked_manual( - init, - self.world, - batch_size, - func, - self.last_run, - self.this_run, - ); + // SAFETY: This method can only be called once per instance of + // `QueryContiguousParIter`, which ensures that mutable queries + // cannot be executed multiple times at once. Mutable instances of + // `QueryContiguousParIter` can only be created via an exclusive + // borrow of a `Query` or a `World`, which ensures that multiple + // aliasing `QueryContiguousParIter`s cannot exist at the same time. + unsafe { + if thread_count <= 1 { + // Just run sequentially. + QueryContiguousIter::new(self.world, self.state, self.last_run, self.this_run) + .unwrap() + .fold(init(), func); + return; } + + // Dispatch to `contiguous_par_fold_init_unchecked_manual` for + // parallel iteration. + let batch_size = self.get_batch_size(thread_count).max(1); + self.state.contiguous_par_fold_init_unchecked_manual( + init, + self.world, + batch_size, + func, + self.last_run, + self.this_run, + ); } } } + /// Returns the size of each batch in rows, given a thread count and the + /// current batching strategy. fn get_batch_size(&self, thread_count: usize) -> u32 { let max_items = || { let id_iter = self.state.matched_storage_ids.iter(); - let tables = unsafe { &self.world.world_metadata().storages().tables }; + // SAFETY: We only access table metadata. + let tables = unsafe { &self.world.storages().tables }; id_iter - .map(|id| unsafe { tables[id.table_id].entity_count() }) + .map(|id| { + // SAFETY: Contiguous iteration can only process tables, so + // we must have a table here. + unsafe { tables[id.table_id].entity_count() } + }) .max() .map(|v| v as usize) .unwrap_or(0) diff --git a/crates/bevy_ecs/src/query/state.rs b/crates/bevy_ecs/src/query/state.rs index 9fbabf51ac66c..827038a8c209b 100644 --- a/crates/bevy_ecs/src/query/state.rs +++ b/crates/bevy_ecs/src/query/state.rs @@ -1504,6 +1504,22 @@ impl QueryState { self.query_mut(world).contiguous_iter_inner() } + /// Returns a parallel contiguous iterator over the query results for the + /// given [`World`] or [`Err`] with [`QueryNotDenseError`] if the query is + /// not dense hence not contiguously iterable. + /// + /// This can only be called for read-only queries. See + /// [`Self::contiguous_par_iter_mut`] for queries that may write to the + /// components. + /// + /// Note that you must use the [`QueryContiguousParIter::for_each`] method + /// to iterate over the results. See [`Self::contiguous_par_iter_mut`] for + /// an example. + /// + /// # Panics + /// The [`ComputeTaskPool`] is not initialized. If using this from a query + /// that is being initialized and run from the ECS scheduler, this should + /// never panic. #[inline] pub fn contiguous_par_iter<'w, 's>( &'s mut self, @@ -1516,6 +1532,53 @@ impl QueryState { self.query(world).contiguous_par_iter_inner() } + /// Returns a parallel contiguous iterator over the query results for the + /// given [`World`] or [`Err`] with [`QueryNotDenseError`] if the query is + /// not dense hence not contiguously iterable. + /// + /// This version of the method is for mutable queries. For read-only + /// queries, see [`Self::contiguous_par_iter`]. + /// + /// # Examples + /// + /// ``` + /// use bevy_ecs::prelude::*; + /// use bevy_ecs::query::QueryEntityError; + /// + /// #[derive(Component, PartialEq, Debug)] + /// struct A(usize); + /// + /// # bevy_tasks::ComputeTaskPool::get_or_init(|| bevy_tasks::TaskPool::new()); + /// + /// let mut world = World::new(); + /// + /// # let entities: Vec = (0..3).map(|i| world.spawn(A(i)).id()).collect(); + /// # let entities: [Entity; 3] = entities.try_into().unwrap(); + /// + /// let mut query_state = world.query::<&mut A>(); + /// + /// query_state.contiguous_par_iter_mut(&mut world).for_each(|mut as| { + /// for a in as { + /// a.0 += 5; + /// } + /// }); + /// + /// # let component_values = query_state.get_many(&world, entities).unwrap(); + /// + /// # assert_eq!(component_values, [&A(5), &A(6), &A(7)]); + /// + /// # let wrong_entity = Entity::from_raw_u32(57).unwrap(); + /// # let invalid_entity = world.spawn_empty().id(); + /// + /// # assert_eq!(match query_state.get_many(&mut world, [wrong_entity]).unwrap_err() {QueryEntityError::NotSpawned(error) => error.entity(), _ => panic!()}, wrong_entity); + /// assert_eq!(match query_state.get_many_mut(&mut world, [invalid_entity]).unwrap_err() {QueryEntityError::QueryDoesNotMatch(entity, _) => entity, _ => panic!()}, invalid_entity); + /// # assert_eq!(query_state.get_many_mut(&mut world, [entities[0], entities[0]]).unwrap_err(), QueryEntityError::AliasedMutability(entities[0])); + /// ``` + /// + /// # Panics + /// The [`ComputeTaskPool`] is not initialized. If using this from a query + /// that is being initialized and run from the ECS scheduler, this should + /// never panic. #[inline] pub fn contiguous_par_iter_mut<'w, 's>( &'s mut self, @@ -1705,7 +1768,7 @@ impl QueryState { }); } - pub(crate) unsafe fn par_fold_contiguous_init_unchecked_manual<'w, 's, T>( + pub(crate) unsafe fn contiguous_par_fold_init_unchecked_manual<'w, 's, T>( &'s self, init_accum: impl Fn() -> T + Send + Sync + Clone, world: UnsafeWorldCell<'w>, @@ -1720,7 +1783,8 @@ impl QueryState { debug_assert!(self.is_dense); bevy_tasks::ComputeTaskPool::get().scope(|scope| { - // SAFETY: We only access table data that has been registered in `self.component_access`. + // SAFETY: We only access table data that has been registered in + // `self.component_access`. let tables = unsafe { &world.storages().tables }; let mut batch_queue = ArrayVec::new(); let mut queue_entity_count = 0; @@ -1730,6 +1794,8 @@ impl QueryState { scope.spawn(async move { #[cfg(feature = "trace")] let _span = self.par_iter_span.enter(); + // SAFETY: Contiguous iteration can only process tables, so + // we must have a table here. let tables = unsafe { &world.storages().tables }; let mut fetch = D::init_fetch(world, &self.fetch_state, last_run, this_run); let mut accum = init_accum(); diff --git a/crates/bevy_ecs/src/system/query.rs b/crates/bevy_ecs/src/system/query.rs index 41ed1207fa00a..4e25dc8ec6bbf 100644 --- a/crates/bevy_ecs/src/system/query.rs +++ b/crates/bevy_ecs/src/system/query.rs @@ -1540,6 +1540,27 @@ impl<'w, 's, D: QueryData, F: QueryFilter> Query<'w, 's, D, F> { .ok_or(QueryNotDenseError(DebugName::type_name::())) } + /// Returns a parallel iterator over contiguous query results for the given + /// [`World`]. + /// + /// Contiguous iteration enables getting slices of contiguously laid out + /// components that reside in the same table. These slices may for example + /// be used for SIMD operations. + /// + /// This parallel iterator is always guaranteed to return results from each + /// matching entity once and only once. Iteration order and thread + /// assignment is not guaranteed. + /// + /// If the `multithreaded` feature is disabled, iterating with this operates + /// identically to [`Iterator::for_each`] on [`QueryContiguousIter`]. + /// + /// This can only be called for read-only queries. For queries that may + /// write to the components they query, see [`par_iter_mut`]. + /// + /// Note that you must use the `for_each` method to iterate over the + /// results. See [`Self::contiguous_par_iter_mut`] for an example. + /// + /// [`World`]: crate::world::World pub fn contiguous_par_iter( &self, ) -> Result, QueryNotDenseError> @@ -1550,6 +1571,45 @@ impl<'w, 's, D: QueryData, F: QueryFilter> Query<'w, 's, D, F> { self.as_readonly().contiguous_par_iter_inner() } + /// Returns a parallel iterator over contiguous query results for the given + /// [`World`]. + /// + /// Contiguous iteration enables getting slices of contiguously laid out + /// components that reside in the same table. These slices may for example + /// be used for SIMD operations. + /// + /// This parallel contiguous iterator is always guaranteed to return results + /// from each matching entity once and only once. Iteration order and thread + /// assignment is not guaranteed. + /// + /// If the `multithreaded` feature is disabled, iterating with this operates + /// identically to [`Iterator::for_each`] on [`QueryContiguousIter`]. + /// + /// This can only be called for mutable queries. See [`par_iter`] for + /// read-only queries. + /// + /// # Example + /// + /// Here, the `gravity_system` updates the `Velocity` component of every entity that contains it: + /// + /// ``` + /// # use bevy_ecs::prelude::*; + /// # + /// # #[derive(Component)] + /// # struct Velocity { x: f32, y: f32, z: f32 } + /// fn gravity_system(mut query: Query<&mut Velocity>) { + /// const DELTA: f32 = 1.0 / 60.0; + /// query.contiguous_par_iter_mut().for_each(|mut velocities| { + /// for mut velocity in velocities { + /// velocity.y -= 9.8 * DELTA; + /// } + /// }); + /// } + /// # bevy_ecs::system::assert_is_system(gravity_system); + /// ``` + /// + /// [`par_iter`]: Self::par_iter + /// [`World`]: crate::world::World pub fn contiguous_par_iter_mut( &mut self, ) -> Result, QueryNotDenseError> @@ -1560,6 +1620,20 @@ impl<'w, 's, D: QueryData, F: QueryFilter> Query<'w, 's, D, F> { self.reborrow().contiguous_par_iter_inner() } + /// Returns a parallel iterator over contiguous query results for the given + /// [`World`](crate::world::World). This consumes the [`Query`] to return + /// results with the actual "inner" world lifetime. + /// + /// Contiguous iteration enables getting slices of contiguously laid out + /// components that reside in the same table. These slices may for example + /// be used for SIMD operations. + /// + /// This parallel iterator is always guaranteed to return results from each + /// matching entity once and only once. Iteration order and thread + /// assignment is not guaranteed. + /// + /// If the `multithreaded` feature is disabled, iterating with this operates + /// identically to [`Iterator::for_each`] on [`QueryContiguousIter`]. pub fn contiguous_par_iter_inner( self, ) -> Result, QueryNotDenseError> From 619d32f01afb3854557f10d9ea6ba58fb95c4249 Mon Sep 17 00:00:00 2001 From: Patrick Walton Date: Sun, 2 Aug 2026 11:11:12 -0700 Subject: [PATCH 03/12] Update comments for tests --- crates/bevy_ecs/src/query/fetch.rs | 15 +++++++++++++-- 1 file changed, 13 insertions(+), 2 deletions(-) diff --git a/crates/bevy_ecs/src/query/fetch.rs b/crates/bevy_ecs/src/query/fetch.rs index 05a07aa580ec4..a8d592f0216e5 100644 --- a/crates/bevy_ecs/src/query/fetch.rs +++ b/crates/bevy_ecs/src/query/fetch.rs @@ -4664,8 +4664,12 @@ mod tests { assert_eq!(present, [true; 3]); } + // Tests that contiguous parallel iteration can correctly mutate all + // instances of a component in the world. #[test] fn contiguous_par_iter_success_test() { + // Declare a couple of components. + #[derive(Component, PartialEq, Eq, Debug)] pub struct C { id: i32, @@ -4675,6 +4679,7 @@ mod tests { #[derive(Component, PartialEq, Eq, Debug)] pub struct D(i32); + // Build a task pool. bevy_tasks::ComputeTaskPool::get_or_init(bevy_tasks::TaskPool::new); // Spawn a world with a couple of tables. @@ -4686,7 +4691,8 @@ mod tests { world.spawn((C { id, found: false }, D(id))); } - // Check that the correct number of rows were matched contiguously. + // Update every row, and check that the correct number of rows were + // matched contiguously. let total_found = AtomicUsize::new(0); let mut contiguous_query = world.query::<&mut C>(); contiguous_query @@ -4705,8 +4711,12 @@ mod tests { assert!(check_query.iter(&world).all(|c| c.found)); } + // Tests that attempting to contiguously iterate in parallel over a query + // that contains sparse sets fails (as the query isn't dense). #[test] fn contiguous_par_iter_failure_test() { + // Declare a couple of components, one of which is a sparse set. + #[derive(Component, Clone, Copy)] struct C; @@ -4714,9 +4724,10 @@ mod tests { #[component(storage = "SparseSet")] struct S; + // Build a task pool. bevy_tasks::ComputeTaskPool::get_or_init(bevy_tasks::TaskPool::new); - // Spawn a world with a couple of tables. + // Spawn a world with those two components. let mut world = World::new(); for _ in 0..100 { world.spawn((C, S)); From 4e4d5d0a9acd24bdf418a9fbf8f77ec04a7c9e1f Mon Sep 17 00:00:00 2001 From: Patrick Walton Date: Sun, 2 Aug 2026 20:08:17 -0700 Subject: [PATCH 04/12] Add a benchmark --- .../iteration/contiguous_par_iter_simple.rs | 78 +++++++++++++++++++ benches/benches/bevy_ecs/iteration/mod.rs | 17 ++++ 2 files changed, 95 insertions(+) create mode 100644 benches/benches/bevy_ecs/iteration/contiguous_par_iter_simple.rs diff --git a/benches/benches/bevy_ecs/iteration/contiguous_par_iter_simple.rs b/benches/benches/bevy_ecs/iteration/contiguous_par_iter_simple.rs new file mode 100644 index 0000000000000..02d5dbe0701c4 --- /dev/null +++ b/benches/benches/bevy_ecs/iteration/contiguous_par_iter_simple.rs @@ -0,0 +1,78 @@ +use bevy_ecs::prelude::*; +use bevy_tasks::{ComputeTaskPool, TaskPool}; +use glam::*; + +#[derive(Component, Copy, Clone)] +struct Transform(Mat4); + +#[derive(Component, Copy, Clone)] +struct Position(Vec3); + +#[derive(Component, Copy, Clone)] +struct Rotation(Vec3); + +#[derive(Component, Copy, Clone)] +struct Velocity(Vec3); + +#[derive(Component, Copy, Clone, Default)] +struct Data(f32); +pub struct Benchmark<'w>(World, QueryState<(&'w Velocity, &'w mut Position)>); + +fn insert_if_bit_enabled(entity: &mut EntityWorldMut, i: u16) { + if i & (1 << B) != 0 { + entity.insert(Data::(1.0)); + } +} + +impl<'w> Benchmark<'w> { + pub fn new(fragment: u16) -> Self { + ComputeTaskPool::get_or_init(TaskPool::default); + + let mut world = World::new(); + + let iter = world.spawn_batch(core::iter::repeat_n( + ( + Transform(Mat4::from_scale(Vec3::ONE)), + Position(Vec3::X), + Rotation(Vec3::X), + Velocity(Vec3::X), + ), + 100_000, + )); + let entities = iter.into_iter().collect::>(); + for i in 0..fragment { + let mut e = world.entity_mut(entities[i as usize]); + insert_if_bit_enabled::<0>(&mut e, i); + insert_if_bit_enabled::<1>(&mut e, i); + insert_if_bit_enabled::<2>(&mut e, i); + insert_if_bit_enabled::<3>(&mut e, i); + insert_if_bit_enabled::<4>(&mut e, i); + insert_if_bit_enabled::<5>(&mut e, i); + insert_if_bit_enabled::<6>(&mut e, i); + insert_if_bit_enabled::<7>(&mut e, i); + insert_if_bit_enabled::<8>(&mut e, i); + insert_if_bit_enabled::<9>(&mut e, i); + insert_if_bit_enabled::<10>(&mut e, i); + insert_if_bit_enabled::<11>(&mut e, i); + insert_if_bit_enabled::<12>(&mut e, i); + insert_if_bit_enabled::<13>(&mut e, i); + insert_if_bit_enabled::<14>(&mut e, i); + insert_if_bit_enabled::<15>(&mut e, i); + } + + let query = world.query::<(&Velocity, &mut Position)>(); + Self(world, query) + } + + #[inline(never)] + pub fn run(&mut self) { + self.1 + .contiguous_par_iter_mut(&mut self.0) + .unwrap() + .for_each(|(vs, mut ps)| { + for (v, p) in vs.iter().zip(ps.iter_mut()) { + p.0 += v.0; + } + }); + } +} diff --git a/benches/benches/bevy_ecs/iteration/mod.rs b/benches/benches/bevy_ecs/iteration/mod.rs index 7867507f62f67..d7a99d6f8e1fc 100644 --- a/benches/benches/bevy_ecs/iteration/mod.rs +++ b/benches/benches/bevy_ecs/iteration/mod.rs @@ -1,3 +1,4 @@ +mod contiguous_par_iter_simple; mod heavy_compute; mod iter_frag; mod iter_frag_foreach; @@ -166,3 +167,19 @@ fn par_iter_simple(c: &mut Criterion) { b.iter(move || bench.run()); }); } + +fn contiguous_par_iter_simple(c: &mut Criterion) { + let mut group = c.benchmark_group("contiguous_par_iter_simple"); + group.warm_up_time(core::time::Duration::from_millis(500)); + group.measurement_time(core::time::Duration::from_secs(4)); + for f in [0, 10, 100, 1000] { + group.bench_function(format!("with_{f}_fragment"), |b| { + let mut bench = par_iter_simple::Benchmark::new(f); + b.iter(move || bench.run()); + }); + } + group.bench_function("hybrid".to_string(), |b| { + let mut bench = par_iter_simple_foreach_hybrid::Benchmark::new(); + b.iter(move || bench.run()); + }); +} From 931505fa1fb0c1f4548dcb2582e596d20a62470b Mon Sep 17 00:00:00 2001 From: Patrick Walton Date: Mon, 3 Aug 2026 01:02:20 -0700 Subject: [PATCH 05/12] Fix compile --- crates/bevy_ecs/src/query/par_iter.rs | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/crates/bevy_ecs/src/query/par_iter.rs b/crates/bevy_ecs/src/query/par_iter.rs index 9f90078190d18..f358e88b13fd6 100644 --- a/crates/bevy_ecs/src/query/par_iter.rs +++ b/crates/bevy_ecs/src/query/par_iter.rs @@ -270,9 +270,9 @@ where #[cfg(any(target_arch = "wasm32", not(feature = "multi_threaded")))] { - QueryContiguousIter::new(self.world, self.state, self.last_run, self_this_run) + QueryContiguousIter::new(self.world, self.state, self.last_run, self.this_run) .unwrap() - .fold(init, func) + .fold(init, func); } #[cfg(all(not(target_arch = "wasm32"), feature = "multi_threaded"))] From e3e42ea36bc686e7e2d0bb81ebf75228cae1ddb8 Mon Sep 17 00:00:00 2001 From: Patrick Walton Date: Mon, 3 Aug 2026 01:07:08 -0700 Subject: [PATCH 06/12] Fix single threaded version? --- crates/bevy_ecs/src/query/par_iter.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/crates/bevy_ecs/src/query/par_iter.rs b/crates/bevy_ecs/src/query/par_iter.rs index f358e88b13fd6..55a9bda03ad04 100644 --- a/crates/bevy_ecs/src/query/par_iter.rs +++ b/crates/bevy_ecs/src/query/par_iter.rs @@ -272,7 +272,7 @@ where { QueryContiguousIter::new(self.world, self.state, self.last_run, self.this_run) .unwrap() - .fold(init, func); + .fold(init(), func); } #[cfg(all(not(target_arch = "wasm32"), feature = "multi_threaded"))] From 2dd162f90e5ab29eadd672cfeaeffd7367501e62 Mon Sep 17 00:00:00 2001 From: Patrick Walton Date: Mon, 3 Aug 2026 01:10:30 -0700 Subject: [PATCH 07/12] Fix single threaded version harder? --- crates/bevy_ecs/src/query/par_iter.rs | 8 +++++--- 1 file changed, 5 insertions(+), 3 deletions(-) diff --git a/crates/bevy_ecs/src/query/par_iter.rs b/crates/bevy_ecs/src/query/par_iter.rs index 55a9bda03ad04..059fc4277cc1f 100644 --- a/crates/bevy_ecs/src/query/par_iter.rs +++ b/crates/bevy_ecs/src/query/par_iter.rs @@ -270,9 +270,11 @@ where #[cfg(any(target_arch = "wasm32", not(feature = "multi_threaded")))] { - QueryContiguousIter::new(self.world, self.state, self.last_run, self.this_run) - .unwrap() - .fold(init(), func); + unsafe { + QueryContiguousIter::new(self.world, self.state, self.last_run, self.this_run) + .unwrap() + .fold(init(), func); + } } #[cfg(all(not(target_arch = "wasm32"), feature = "multi_threaded"))] From 4c7cba2adb4b16c891bf555ad5164a225756c3b9 Mon Sep 17 00:00:00 2001 From: Patrick Walton Date: Sat, 8 Aug 2026 02:48:16 -0700 Subject: [PATCH 08/12] Fix silly errors in doctests --- crates/bevy_ecs/src/query/state.rs | 5 ++--- crates/bevy_ecs/src/system/query.rs | 2 +- 2 files changed, 3 insertions(+), 4 deletions(-) diff --git a/crates/bevy_ecs/src/query/state.rs b/crates/bevy_ecs/src/query/state.rs index 71125a977aa9c..bedb15ae9875a 100644 --- a/crates/bevy_ecs/src/query/state.rs +++ b/crates/bevy_ecs/src/query/state.rs @@ -19,7 +19,6 @@ use crate::{ use crate::entity::UniqueEntityEquivalentSlice; use alloc::{format, vec::Vec}; -use arrayvec::ArrayVec; use bevy_utils::prelude::DebugName; use core::{fmt, ptr}; use fixedbitset::FixedBitSet; @@ -1557,8 +1556,8 @@ impl QueryState { /// /// let mut query_state = world.query::<&mut A>(); /// - /// query_state.contiguous_par_iter_mut(&mut world).for_each(|mut as| { - /// for a in as { + /// query_state.contiguous_par_iter_mut(&mut world).unwrap().for_each(|mut batch| { + /// for a in batch { /// a.0 += 5; /// } /// }); diff --git a/crates/bevy_ecs/src/system/query.rs b/crates/bevy_ecs/src/system/query.rs index 48243457c179e..33b046719a3cd 100644 --- a/crates/bevy_ecs/src/system/query.rs +++ b/crates/bevy_ecs/src/system/query.rs @@ -1636,7 +1636,7 @@ impl<'w, 's, D: QueryData, F: QueryFilter> Query<'w, 's, D, F> { /// # struct Velocity { x: f32, y: f32, z: f32 } /// fn gravity_system(mut query: Query<&mut Velocity>) { /// const DELTA: f32 = 1.0 / 60.0; - /// query.contiguous_par_iter_mut().for_each(|mut velocities| { + /// query.contiguous_par_iter_mut().unwrap().for_each(|mut velocities| { /// for mut velocity in velocities { /// velocity.y -= 9.8 * DELTA; /// } From 662c7085335a32753eeaa4da91c2ced2e1e57511 Mon Sep 17 00:00:00 2001 From: Patrick Walton Date: Sat, 8 Aug 2026 02:54:34 -0700 Subject: [PATCH 09/12] Import fix --- crates/bevy_ecs/src/query/state.rs | 2 ++ 1 file changed, 2 insertions(+) diff --git a/crates/bevy_ecs/src/query/state.rs b/crates/bevy_ecs/src/query/state.rs index bedb15ae9875a..34a2d3bd7ff6b 100644 --- a/crates/bevy_ecs/src/query/state.rs +++ b/crates/bevy_ecs/src/query/state.rs @@ -1782,6 +1782,8 @@ impl QueryState { { debug_assert!(self.is_dense); + use arrayvec::ArrayVec; + bevy_tasks::ComputeTaskPool::get().scope(|scope| { // SAFETY: We only access table data that has been registered in // `self.component_access`. From 29299bc729282c0c2efbe79fa94e108a9f555437 Mon Sep 17 00:00:00 2001 From: Patrick Walton Date: Sat, 8 Aug 2026 07:00:39 -0700 Subject: [PATCH 10/12] Fix doctest --- crates/bevy_ecs/src/query/par_iter.rs | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/crates/bevy_ecs/src/query/par_iter.rs b/crates/bevy_ecs/src/query/par_iter.rs index 5c4d1ce3810f7..9b112e7ae62b1 100644 --- a/crates/bevy_ecs/src/query/par_iter.rs +++ b/crates/bevy_ecs/src/query/par_iter.rs @@ -240,8 +240,9 @@ where /// struct T; /// fn system(query: Query<&T>){ /// let mut queue: Parallel = Parallel::default(); - /// // queue.borrow_local_mut() will get or create a thread_local queue for each task/thread; - /// query.contiguous_par_iter().for_each_init(|| queue.borrow_local_mut(),|local_queue, items| { + /// // queue.borrow_local_mut() will get or create a thread_local queue for each task/thread. + /// // We unwrap the call to `contiguous_par_iter()` because we know the query in question is dense. + /// query.contiguous_par_iter().unwrap().for_each_init(|| queue.borrow_local_mut(),|local_queue, items| { /// for _ in items { /// **local_queue += 1; /// } From 457e3815fded15cbc015e0ee7c917052422f0a18 Mon Sep 17 00:00:00 2001 From: Patrick Walton Date: Sat, 8 Aug 2026 07:42:20 -0700 Subject: [PATCH 11/12] Doc check police --- crates/bevy_ecs/src/query/state.rs | 4 ++++ crates/bevy_ecs/src/system/query.rs | 2 +- 2 files changed, 5 insertions(+), 1 deletion(-) diff --git a/crates/bevy_ecs/src/query/state.rs b/crates/bevy_ecs/src/query/state.rs index 34a2d3bd7ff6b..c5d6d93f04322 100644 --- a/crates/bevy_ecs/src/query/state.rs +++ b/crates/bevy_ecs/src/query/state.rs @@ -1519,6 +1519,8 @@ impl QueryState { /// The [`ComputeTaskPool`] is not initialized. If using this from a query /// that is being initialized and run from the ECS scheduler, this should /// never panic. + /// + /// [`ComputeTaskPool`]: bevy_tasks::ComputeTaskPool #[inline] pub fn contiguous_par_iter<'w, 's>( &'s mut self, @@ -1578,6 +1580,8 @@ impl QueryState { /// The [`ComputeTaskPool`] is not initialized. If using this from a query /// that is being initialized and run from the ECS scheduler, this should /// never panic. + /// + /// [`ComputeTaskPool`]: bevy_tasks::ComputeTaskPool #[inline] pub fn contiguous_par_iter_mut<'w, 's>( &'s mut self, diff --git a/crates/bevy_ecs/src/system/query.rs b/crates/bevy_ecs/src/system/query.rs index 33b046719a3cd..8a2943e9c813d 100644 --- a/crates/bevy_ecs/src/system/query.rs +++ b/crates/bevy_ecs/src/system/query.rs @@ -1592,7 +1592,7 @@ impl<'w, 's, D: QueryData, F: QueryFilter> Query<'w, 's, D, F> { /// identically to [`Iterator::for_each`] on [`QueryContiguousIter`]. /// /// This can only be called for read-only queries. For queries that may - /// write to the components they query, see [`par_iter_mut`]. + /// write to the components they query, see [`Self::par_iter_mut`]. /// /// Note that you must use the `for_each` method to iterate over the /// results. See [`Self::contiguous_par_iter_mut`] for an example. From aac75f2481098ce57b76f3933b217ad71fbe320d Mon Sep 17 00:00:00 2001 From: Patrick Walton Date: Sat, 8 Aug 2026 14:19:21 -0700 Subject: [PATCH 12/12] Address review comments --- crates/bevy_ecs/src/query/state.rs | 8 +++++++- crates/bevy_ecs/src/system/query.rs | 3 +++ 2 files changed, 10 insertions(+), 1 deletion(-) diff --git a/crates/bevy_ecs/src/query/state.rs b/crates/bevy_ecs/src/query/state.rs index c5d6d93f04322..96ea0a060dd61 100644 --- a/crates/bevy_ecs/src/query/state.rs +++ b/crates/bevy_ecs/src/query/state.rs @@ -1795,6 +1795,12 @@ impl QueryState { let mut batch_queue = ArrayVec::new(); let mut queue_entity_count = 0; + // Submit a list of tables which are smaller than the batch size as + // a single task. + // + // The 128 table limit is an arbitrary tuning parameter unrelated to + // the batch size. It matches the limit in + // `Self::par_fold_init_unchecked_manual`. let submit_batch_queue = |queue: ArrayVec| { let (func, init_accum) = (func.clone(), init_accum.clone()); scope.spawn(async move { @@ -1806,7 +1812,7 @@ impl QueryState { let mut fetch = D::init_fetch(world, &self.fetch_state, last_run, this_run); let mut accum = init_accum(); for table_id in queue { - let table = tables.get(table_id).expect("Table must be present"); + let table = &tables[table_id]; D::set_table(&mut fetch, &self.fetch_state, table); let item = D::fetch_contiguous(&self.fetch_state, &mut fetch, table.entities()); diff --git a/crates/bevy_ecs/src/system/query.rs b/crates/bevy_ecs/src/system/query.rs index 8a2943e9c813d..b21276eee87ac 100644 --- a/crates/bevy_ecs/src/system/query.rs +++ b/crates/bevy_ecs/src/system/query.rs @@ -1588,6 +1588,9 @@ impl<'w, 's, D: QueryData, F: QueryFilter> Query<'w, 's, D, F> { /// matching entity once and only once. Iteration order and thread /// assignment is not guaranteed. /// + /// If the query isn't contiguously iterable because it isn't dense, this + /// method returns a [`QueryNotDenseError`]. + /// /// If the `multithreaded` feature is disabled, iterating with this operates /// identically to [`Iterator::for_each`] on [`QueryContiguousIter`]. ///