From c3d0f844b9998dc31f77653d86b8d5949bc03ac1 Mon Sep 17 00:00:00 2001 From: Micha Reiser Date: Sun, 28 Jun 2026 18:18:40 +0000 Subject: [PATCH 1/8] perf: reduce function ingredient monomorphization --- src/function.rs | 49 ----------------------------- src/function/execute.rs | 3 +- src/function/maybe_changed_after.rs | 5 ++- src/function/memo.rs | 20 ++++++++++-- 4 files changed, 22 insertions(+), 55 deletions(-) diff --git a/src/function.rs b/src/function.rs index 996401b00..2c576aa72 100644 --- a/src/function.rs +++ b/src/function.rs @@ -179,38 +179,12 @@ impl<'db> FunctionIngredientRef<'db> { pub(crate) fn sync_table(&self) -> &'db SyncTable { self.ingredient.sync_table() } - - /// Returns information about the current provisional status of `input`. - /// - /// Is it a provisional value, a poisoned provisional memo, or has it been finalized and in - /// which iteration. - /// - /// Returns `None` if `input` doesn't exist. - pub(crate) fn provisional_status( - &self, - zalsa: &'db Zalsa, - input: Id, - ) -> Option> { - self.ingredient.provisional_status(zalsa, input) - } } pub(crate) trait FunctionIngredient: Send + Sync { fn memo<'db>(&'db self, zalsa: &'db Zalsa, input: Id) -> Option>; fn sync_table(&self) -> &SyncTable; - - /// Returns information about the current provisional status of `input`. - /// - /// Is it a provisional value, a poisoned provisional memo, or has it been finalized and in - /// which iteration. - /// - /// Returns `None` if `input` doesn't exist. - fn provisional_status<'db>( - &'db self, - zalsa: &'db Zalsa, - input: Id, - ) -> Option>; } /// Function ingredients are the "workhorse" of salsa. @@ -387,29 +361,6 @@ where fn sync_table(&self) -> &SyncTable { &self.sync_table } - - /// Returns `final` if the memo has the `verified_final` flag set. - /// - /// Otherwise, the value is still provisional or the provisional memo has been poisoned. It - /// also returns the iteration in which this memo was created (always 0 except for cycle - /// heads). - fn provisional_status<'db>( - &'db self, - zalsa: &'db Zalsa, - input: Id, - ) -> Option> { - let memo = - self.get_memo_from_table_for(zalsa, input, self.memo_ingredient_index(zalsa, input))?; - - if memo.value.is_none() && memo.header.may_be_provisional() { - return Some(ProvisionalStatus::Poisoned { - iteration: memo.header.revisions.iteration(), - verified_at: memo.header.verified_at.load(), - }); - } - - Some(memo.header.provisional_status()) - } } impl Ingredient for IngredientImpl diff --git a/src/function/execute.rs b/src/function/execute.rs index cea05487e..09f51c0d5 100644 --- a/src/function/execute.rs +++ b/src/function/execute.rs @@ -656,7 +656,8 @@ fn collect_all_cycle_heads( .expect("cycle heads must be function ingredients"); let provisional_status = function - .provisional_status(zalsa, current_head.key_index()) + .memo(zalsa, current_head.key_index()) + .map(|memo| memo.provisional_status()) .expect("cycle head memo must have been created during the execution"); if let ProvisionalStatus::Poisoned { diff --git a/src/function/maybe_changed_after.rs b/src/function/maybe_changed_after.rs index 3d7e8e1a6..15c0be5ee 100644 --- a/src/function/maybe_changed_after.rs +++ b/src/function/maybe_changed_after.rs @@ -663,9 +663,8 @@ fn validate_provisional( let Some(provisional_status) = zalsa .lookup_ingredient(cycle_head.database_key_index.ingredient_index()) .as_function() - .and_then(|function| { - function.provisional_status(zalsa, cycle_head.database_key_index.key_index()) - }) + .and_then(|function| function.memo(zalsa, cycle_head.database_key_index.key_index())) + .map(|memo| memo.provisional_status()) else { return false; }; diff --git a/src/function/memo.rs b/src/function/memo.rs index 7869dd9a1..b11c0cd54 100644 --- a/src/function/memo.rs +++ b/src/function/memo.rs @@ -105,7 +105,7 @@ pub struct Memo { /// covering the entire allocation, which remains valid for shared access for `'db`, even after /// replacement. `to_dyn_fn` and `type_id` describe the same `C`. #[derive(Clone, Copy)] -pub(crate) struct ErasedMemo<'db> { +pub struct ErasedMemo<'db> { /// A pointer to the base address of the [`Memo`] allocation, with spatial provenance covering /// the entire allocation. data: NonNull, @@ -159,6 +159,21 @@ impl<'memo> ErasedMemo<'memo> { unsafe { (self.to_dyn_fn)(self.data).as_ref() }.has_value() } + /// Returns whether this memo is provisional, finalized, or poisoned. + #[inline] + pub(super) fn provisional_status(self) -> ProvisionalStatus<'memo> { + let header = self.header(); + + if !self.has_value() && header.may_be_provisional() { + return ProvisionalStatus::Poisoned { + iteration: header.revisions.iteration(), + verified_at: header.verified_at.load(), + }; + } + + header.provisional_status() + } + /// Returns the concrete memo after asserting that it uses configuration `C`. /// /// # Panics @@ -547,7 +562,8 @@ impl Iterator for TryClaimCycleHeadsIter<'_> { crate::tracing::trace!("Waiting for {head_database_key:?} results in a cycle"); let provisional_status = function - .provisional_status(self.zalsa, head_key_index) + .memo(self.zalsa, head_key_index) + .map(|memo| memo.provisional_status()) .expect("cycle head memo to exist"); let (current_iteration, verified_at) = match provisional_status { ProvisionalStatus::Provisional { From 31f81b1d797dfd733c132d52ae9cdcc127570707 Mon Sep 17 00:00:00 2001 From: Micha Reiser Date: Wed, 17 Jun 2026 13:33:15 +0000 Subject: [PATCH 2/8] refactor: erase tracked query lifecycle --- src/function/execute.rs | 689 ++++++++++++++++++++++++---------------- src/function/fetch.rs | 276 ++++++++-------- src/function/memo.rs | 2 +- 3 files changed, 561 insertions(+), 406 deletions(-) diff --git a/src/function/execute.rs b/src/function/execute.rs index 09f51c0d5..09b8af2a5 100644 --- a/src/function/execute.rs +++ b/src/function/execute.rs @@ -2,95 +2,219 @@ use smallvec::SmallVec; use crate::active_query::CompletedQuery; use crate::cycle::{CycleHeads, CycleRecoveryStrategy, IterationStamp, ProvisionalStatus}; -use crate::function::memo::{Memo, MemoHeader}; +use crate::function::memo::{ErasedMemo, Memo, MemoHeader}; use crate::function::sync::ReleaseMode; use crate::function::{ClaimGuard, ClaimResult, Configuration, IngredientImpl, Reentrancy}; use crate::hash::{FxHashSet, FxIndexSet}; use crate::plumbing::ZalsaLocal; use crate::sync::thread; +use crate::table::memo::MemoSlot; use crate::tracked_struct::Identity; use crate::zalsa::{MemoIngredientIndex, Zalsa}; use crate::zalsa_local::{ActiveQueryGuard, QueryEdge, QueryEdgeKind, QueryRevisions}; -use crate::{Cancelled, Cycle, tracing}; +use crate::{Cancelled, Cycle, Revision, tracing}; use crate::{DatabaseKeyIndex, Event, EventKind, Id}; -impl IngredientImpl -where - C: Configuration, -{ - /// Executes the query function for the given `active_query`. Creates and stores - /// a new memo with the result, backdated if possible. Once this completes, - /// the query will have been popped off the active query stack. - /// - /// # Parameters - /// - /// * `db`, the database. - /// * `active_query`, the active stack frame for the query to execute. - /// * `opt_old_memo`, the older memo, if any existed. Used for backdating. - /// - /// # Returns - /// The newly computed memo or `None` if this query is part of a larger cycle - /// and `execute` blocked on a cycle head running on another thread. In this case, - /// the memo is potentially outdated and needs to be refetched. - #[inline(never)] - pub(super) fn execute<'db>( - &'db self, - db: &'db C::DbView, - mut claim_guard: ClaimGuard<'db>, - opt_old_memo: Option<&'db Memo>, - ) -> Option<&'db Memo> { - let database_key_index = claim_guard.database_key_index(); - let zalsa = claim_guard.zalsa(); +/// Type-specific operations needed by the shared cold query lifecycle. +/// +/// Every [`ErasedMemo`] passed to a state must come from that state's ingredient. +pub(super) trait QueryState<'db> { + fn execute_query(&mut self, zalsa: &'db Zalsa, id: Id); - let id = database_key_index.key_index(); - let memo_ingredient_index = self.memo_ingredient_index(zalsa, id); + fn use_fallback(&mut self, zalsa: &'db Zalsa, id: Id); - crate::tracing::info!("{:?}: executing query", database_key_index); + fn recover_from_cycle( + &mut self, + zalsa: &'db Zalsa, + cycle: &Cycle, + last_provisional_memo: ErasedMemo<'db>, + ) -> bool; - zalsa.event(&|| { - Event::new(EventKind::WillExecute { - database_key: database_key_index, - }) - }); + fn get_memo( + &self, + zalsa: &'db Zalsa, + id: Id, + memo_ingredient_index: MemoIngredientIndex, + ) -> Option>; - let (new_value, mut completed_query) = match C::CYCLE_STRATEGY { - CycleRecoveryStrategy::Panic => { - let (new_value, active_query) = Self::execute_query( - db, - zalsa, - claim_guard.zalsa_local().push_query(database_key_index), - opt_old_memo.map(|memo| &memo.header), - ); + fn insert_memo( + &mut self, + zalsa: &'db Zalsa, + id: Id, + revision: Revision, + revisions: QueryRevisions, + memo_ingredient_index: MemoIngredientIndex, + ) -> ErasedMemo<'db>; - // Ordinary queries don't need a cycle iteration stamp. Keeping the default avoids - // allocating `QueryRevisionsExtra` after a revision-preserving cancellation. - (new_value, active_query.pop(IterationStamp::default())) - } - CycleRecoveryStrategy::FallbackImmediate | CycleRecoveryStrategy::Fixpoint => { - let _cancellation_guard = - DisableLocalCancellationGuard::new(claim_guard.zalsa_local()); - - self.execute_maybe_iterate( - db, - opt_old_memo, - &mut claim_guard, - memo_ingredient_index, - ) - } + fn execute_iterated( + &mut self, + zalsa: &'db Zalsa, + opt_old_memo: Option>, + claim_guard: &mut ClaimGuard<'db>, + memo_ingredient_index: MemoIngredientIndex, + strategy: CycleRecoveryStrategy, + ) -> CompletedQuery; + + fn finish_memo( + &mut self, + zalsa: &'db Zalsa, + database_key_index: DatabaseKeyIndex, + opt_old_memo: Option>, + completed_query: CompletedQuery, + memo_ingredient_index: MemoIngredientIndex, + ) -> ErasedMemo<'db>; +} + +pub(super) struct QueryStateImpl<'db, C: Configuration> { + ingredient: &'db IngredientImpl, + db: &'db C::DbView, + value: Option>, +} + +impl<'db, C: Configuration> QueryStateImpl<'db, C> { + pub(super) fn new(ingredient: &'db IngredientImpl, db: &'db C::DbView) -> Self { + Self { + ingredient, + db, + value: None, + } + } + + fn memo_slot( + &self, + zalsa: &'db Zalsa, + id: Id, + memo_ingredient_index: MemoIngredientIndex, + ) -> MemoSlot<'db> { + // SAFETY: Replaced memo allocations remain in deleted_entries until the next revision. + // The database is borrowed for 'db, so a new revision cannot begin while this state can + // still observe an allocation. + unsafe { + MemoSlot::new( + zalsa.memo_table_for::>(id), + memo_ingredient_index, + ) + } + } +} + +impl<'db, C: Configuration> QueryState<'db> for QueryStateImpl<'db, C> { + fn execute_query(&mut self, zalsa: &'db Zalsa, id: Id) { + self.value = Some(C::execute(self.db, C::id_to_input(zalsa, id))); + } + + fn use_fallback(&mut self, zalsa: &'db Zalsa, id: Id) { + self.value = Some(C::cycle_initial(self.db, id, C::id_to_input(zalsa, id))); + } + + fn recover_from_cycle( + &mut self, + zalsa: &'db Zalsa, + cycle: &Cycle, + last_provisional_memo: ErasedMemo<'db>, + ) -> bool { + let last_provisional_memo = last_provisional_memo.downcast::(); + let last_provisional_value = last_provisional_memo.value().expect( + "`fetch_cold_cycle` should have inserted a provisional memo with Cycle::initial", + ); + let value = self + .value + .take() + .expect("query state must contain the value from the latest execution"); + let value = C::recover_from_cycle( + self.db, + cycle, + last_provisional_value, + value, + C::id_to_input(zalsa, cycle.id), + ); + let converged = C::values_equal(&value, last_provisional_value); + self.value = Some(value); + converged + } + + fn get_memo( + &self, + zalsa: &'db Zalsa, + id: Id, + memo_ingredient_index: MemoIngredientIndex, + ) -> Option> { + self.memo_slot(zalsa, id, memo_ingredient_index) + .get_erased() + } + + fn insert_memo( + &mut self, + zalsa: &'db Zalsa, + id: Id, + revision: Revision, + revisions: QueryRevisions, + memo_ingredient_index: MemoIngredientIndex, + ) -> ErasedMemo<'db> { + let value = self + .value + .take() + .expect("query state must contain a value before memo insertion"); + self.ingredient.insert_memo( + zalsa, + id, + Memo::new(Some(value), revision, revisions), + memo_ingredient_index, + ); + self.memo_slot(zalsa, id, memo_ingredient_index) + .get_erased() + .expect("memo was just inserted") + } + + fn execute_iterated( + &mut self, + zalsa: &'db Zalsa, + opt_old_memo: Option>, + claim_guard: &mut ClaimGuard<'db>, + memo_ingredient_index: MemoIngredientIndex, + strategy: CycleRecoveryStrategy, + ) -> CompletedQuery { + let id = claim_guard.database_key_index().key_index(); + let _poison_guard = PoisonProvisionalIfPanicking { + ingredient: self.ingredient, + zalsa, + id, + memo_ingredient_index, }; + execute_maybe_iterate_erased( + self, + zalsa, + opt_old_memo, + claim_guard, + memo_ingredient_index, + strategy, + ) + } + + fn finish_memo( + &mut self, + zalsa: &'db Zalsa, + database_key_index: DatabaseKeyIndex, + opt_old_memo: Option>, + mut completed_query: CompletedQuery, + memo_ingredient_index: MemoIngredientIndex, + ) -> ErasedMemo<'db> { + let id = database_key_index.key_index(); + let value = self + .value + .take() + .expect("query execution must produce a value"); if let Some(old_memo) = opt_old_memo { - // If the new value is equal to the old one, then it didn't - // really change, even if some of its inputs have. So we can - // "backdate" its `changed_at` revision to be the same as the - // old value. - self.backdate_if_appropriate( + let old_memo = old_memo.downcast::(); + // An equal output did not logically change even if an input did, so preserve its + // old `changed_at` revision. + self.ingredient.backdate_if_appropriate( old_memo, database_key_index, &mut completed_query.revisions, - &new_value, + &value, ); - // Diff the new outputs with the old, to discard any no-longer-emitted // outputs and update the tracked struct IDs for seeding the next revision. old_memo @@ -101,244 +225,265 @@ where #[cfg(not(feature = "persistence"))] completed_query.revisions.discard_edges_if_never_change(); - let memo = self.insert_memo( + self.ingredient.insert_memo( zalsa, id, Memo::new( - Some(new_value), + Some(value), zalsa.current_revision(), completed_query.revisions, ), memo_ingredient_index, ); - - if claim_guard.drop() { None } else { Some(memo) } + self.memo_slot(zalsa, id, memo_ingredient_index) + .get_erased() + .expect("memo was just inserted") } +} - fn execute_maybe_iterate<'db>( - &'db self, - db: &'db C::DbView, - opt_old_memo: Option<&'db Memo>, - claim_guard: &mut ClaimGuard<'db>, - memo_ingredient_index: MemoIngredientIndex, - ) -> (C::Output<'db>, CompletedQuery) { - claim_guard.set_release_mode(ReleaseMode::Default); +pub(super) fn execute_erased<'db>( + state: &mut dyn QueryState<'db>, + mut claim_guard: ClaimGuard<'db>, + opt_old_memo: Option>, + memo_ingredient_index: MemoIngredientIndex, + strategy: CycleRecoveryStrategy, +) -> Option> { + let database_key_index = claim_guard.database_key_index(); + let zalsa = claim_guard.zalsa(); - let database_key_index = claim_guard.database_key_index(); - let zalsa = claim_guard.zalsa(); + crate::tracing::info!("{:?}: executing query", database_key_index); + zalsa.event(&|| { + Event::new(EventKind::WillExecute { + database_key: database_key_index, + }) + }); - let id = database_key_index.key_index(); + let completed_query = match strategy { + CycleRecoveryStrategy::Panic => { + let active_query = claim_guard.zalsa_local().push_query(database_key_index); + seed_query_from_old_memo(zalsa, &active_query, opt_old_memo); + state.execute_query(zalsa, database_key_index.key_index()); - // Our provisional value from the previous iteration, when doing fixpoint iteration. - // This is different from `opt_old_memo` which might be from a different revision. - let mut last_provisional_memo_opt: Option<&Memo> = None; + // Ordinary queries don't need an epoch stamp. Keeping the default avoids allocating + // `QueryRevisionsExtra` after a revision-preserving cancellation. + active_query.pop(IterationStamp::default()) + } + CycleRecoveryStrategy::FallbackImmediate | CycleRecoveryStrategy::Fixpoint => { + let _cancellation_guard = DisableLocalCancellationGuard::new(claim_guard.zalsa_local()); + state.execute_iterated( + zalsa, + opt_old_memo, + &mut claim_guard, + memo_ingredient_index, + strategy, + ) + } + }; - let mut last_stale_tracked_ids: Vec<(Identity, Id)> = Vec::new(); - let current_revision = zalsa.current_revision(); - let cancellation_count = zalsa.runtime().cancellation_count(); - let mut opt_old_memo = opt_old_memo; - let mut iteration = IterationStamp::initial(cancellation_count); + let memo = state.finish_memo( + zalsa, + database_key_index, + opt_old_memo, + completed_query, + memo_ingredient_index, + ); - // An ordinary query doesn't memoize a cancelled execution. Match that behavior for - // fixpoint queries: a memo from an abandoned cancellation epoch in this revision doesn't - // seed the retry, while a memo from an older revision remains useful for backdating and - // output bookkeeping. Cancellation counts are only comparable within a revision. - if let Some(old_memo) = opt_old_memo { - if old_memo.header.verified_at.load() == current_revision { - match old_memo.header.previous_iteration( - database_key_index, - cancellation_count, - old_memo.value.is_some(), - ) { - Some(previous_iteration) => { - if previous_iteration.reuse_as_provisional { - last_provisional_memo_opt = Some(old_memo); - } - - iteration = previous_iteration.iteration; - } - None => opt_old_memo = None, - } - } - } + if claim_guard.drop() { None } else { Some(memo) } +} - let _poison_guard = - PoisonProvisionalIfPanicking::new(self, zalsa, id, memo_ingredient_index); +fn seed_query_from_old_memo( + zalsa: &Zalsa, + active_query: &ActiveQueryGuard<'_>, + old_memo: Option>, +) { + let Some(old_memo) = old_memo else { + return; + }; - let (new_value, completed_query) = loop { - let active_query = claim_guard.zalsa_local().push_query(database_key_index); + old_memo.header().seed_active_query(zalsa, active_query); +} - // Tracked struct ids that existed in the previous revision - // but weren't recreated in the last iteration. It's important that we seed the next - // query with these ids because the query might re-create them as part of the next iteration. - // This is not only important to ensure that the re-created tracked structs have the same ids, - // it's also important to ensure that these tracked structs get removed - // if they aren't recreated when reaching the final iteration. - active_query.seed_tracked_struct_ids(&last_stale_tracked_ids); +fn execute_maybe_iterate_erased<'db>( + state: &mut dyn QueryState<'db>, + zalsa: &'db Zalsa, + opt_old_memo: Option>, + claim_guard: &mut ClaimGuard<'db>, + memo_ingredient_index: MemoIngredientIndex, + strategy: CycleRecoveryStrategy, +) -> CompletedQuery { + claim_guard.set_release_mode(ReleaseMode::Default); - let (mut new_value, active_query) = Self::execute_query( - db, - zalsa, - active_query, - last_provisional_memo_opt - .or(opt_old_memo) - .map(|memo| &memo.header), - ); + let database_key_index = claim_guard.database_key_index(); - let (mut active_query, cycle_heads, outer_cycle, cycle_iteration) = - match try_complete_query(zalsa, active_query, claim_guard, iteration) { - QueryExecutionOutcome::Completed(completed_query) => { - break (new_value, completed_query); - } - QueryExecutionOutcome::Participant { - active_query, - cycle_heads, - outer_cycle, - } => { - // For FallbackImmediate, use the fallback value instead of the computed value - // for all cycle participants. This ensures that the results don't depend on the query call order, see - // https://github.com/salsa-rs/salsa/pull/798#issuecomment-2812855285. - if C::CYCLE_STRATEGY == CycleRecoveryStrategy::FallbackImmediate { - new_value = C::cycle_initial(db, id, C::id_to_input(zalsa, id)); - } - - let completed_query = complete_cycle_participant( - active_query, - claim_guard, - cycle_heads, - outer_cycle, - iteration, - ); - - break (new_value, completed_query); + let id = database_key_index.key_index(); + + // Our provisional value from the previous iteration, when doing fixpoint iteration. + // This is different from `opt_old_memo` which might be from a different revision. + let mut last_provisional_memo_opt = None; + + let mut last_stale_tracked_ids: Vec<(Identity, Id)> = Vec::new(); + let current_revision = zalsa.current_revision(); + let cancellation_count = zalsa.runtime().cancellation_count(); + let mut opt_old_memo = opt_old_memo; + let mut iteration = IterationStamp::initial(cancellation_count); + + // An ordinary query doesn't memoize a cancelled execution. Match that behavior for + // fixpoint queries: a memo from an abandoned cancellation epoch in this revision doesn't + // seed the retry, while a memo from an older revision remains useful for backdating and + // output bookkeeping. Cancellation counts are only comparable within a revision. + if let Some(old_memo) = opt_old_memo { + if old_memo.header().verified_at.load() == current_revision { + match old_memo.header().previous_iteration( + database_key_index, + cancellation_count, + old_memo.has_value(), + ) { + Some(previous_iteration) => { + if previous_iteration.reuse_as_provisional { + last_provisional_memo_opt = Some(old_memo); } - QueryExecutionOutcome::CycleHead { - active_query, - cycle_heads, - outer_cycle, - cycle_iteration, - } => (active_query, cycle_heads, outer_cycle, cycle_iteration), - }; - - // Get the last provisional value for this query so that we can compare it with the new value - // to test if the cycle converged. - let last_provisional_memo = last_provisional_memo_opt.unwrap_or_else(|| { - // This is our first time around the loop; a provisional value must have been - // inserted into the memo table when the cycle was hit, so let's pull our - // initial provisional value from there. - let memo = self - .get_memo_from_table_for(zalsa, id, memo_ingredient_index) - .unwrap_or_else(|| { - unreachable!( - "{database_key_index:#?} is a cycle head, \ - but no provisional memo found" - ) - }); - debug_assert!(memo.header.may_be_provisional()); - memo - }); + iteration = previous_iteration.iteration; + } + None => opt_old_memo = None, + } + } + } - let last_provisional_value = last_provisional_memo.value(); + let completed_query = loop { + let active_query = claim_guard.zalsa_local().push_query(database_key_index); - let last_provisional_value = last_provisional_value.expect( - "`fetch_cold_cycle` should have inserted a provisional memo with Cycle::initial", - ); - tracing::debug!( - "{database_key_index:?}: execute: \ - I am a cycle head, comparing last provisional value with new value" - ); + // Tracked struct ids that existed in the previous revision + // but weren't recreated in the last iteration. It's important that we seed the next + // query with these ids because the query might re-create them as part of the next iteration. + // This is not only important to ensure that the re-created tracked structs have the same ids, + // it's also important to ensure that these tracked structs get removed + // if they aren't recreated when reaching the final iteration. + active_query.seed_tracked_struct_ids(&last_stale_tracked_ids); - // For FallbackImmediate, the value always converges immediately (we use the - // fallback directly). We still iterate if metadata hasn't converged. - // For Fixpoint, ask the recovery function what value to use and check convergence. - let value_converged = if C::CYCLE_STRATEGY == CycleRecoveryStrategy::FallbackImmediate { - // Use the fallback value instead of the computed value. - new_value = C::cycle_initial(db, id, C::id_to_input(zalsa, id)); - true - } else { - let cycle = Cycle { - head_ids: cycle_heads.ids(), - id, - iteration: cycle_iteration.iteration_as_u32(), - }; - // We are in a cycle that hasn't converged; ask the user's - // cycle-recovery function what to do (it may return the same value or a different one): - new_value = C::recover_from_cycle( - db, - &cycle, - last_provisional_value, - new_value, - C::id_to_input(zalsa, id), - ); + seed_query_from_old_memo( + zalsa, + &active_query, + last_provisional_memo_opt.or(opt_old_memo), + ); - C::values_equal(&new_value, last_provisional_value) - }; + state.execute_query(zalsa, id); + let (mut active_query, cycle_heads, outer_cycle, cycle_iteration) = + match try_complete_query(zalsa, active_query, claim_guard, iteration) { + QueryExecutionOutcome::Completed(completed_query) => break completed_query, + QueryExecutionOutcome::Participant { + active_query, + cycle_heads, + outer_cycle, + } => { + // FallbackImmediate uses the fallback for every participant so the result + // does not depend on query call order. + if strategy == CycleRecoveryStrategy::FallbackImmediate { + state.use_fallback(zalsa, id); + } - let new_cycle_heads = active_query.take_cycle_heads(); - assert_no_new_cycle_heads(&cycle_heads, new_cycle_heads, database_key_index); - - let completed_query = match try_complete_cycle_head( - active_query, - claim_guard, - cycle_heads, - &last_provisional_memo.header.revisions, - outer_cycle, - iteration, - cycle_iteration, - value_converged, - ) { - Ok(completed_query) => { - break (new_value, completed_query); - } - Err((completed_query, new_iteration)) => { - iteration = new_iteration; - completed_query + break complete_cycle_participant( + active_query, + claim_guard, + cycle_heads, + outer_cycle, + iteration, + ); } + QueryExecutionOutcome::CycleHead { + active_query, + cycle_heads, + outer_cycle, + cycle_iteration, + } => (active_query, cycle_heads, outer_cycle, cycle_iteration), }; - let new_memo = self.insert_memo( - zalsa, - id, - Memo::new(Some(new_value), current_revision, completed_query.revisions), - memo_ingredient_index, + // Get the last provisional value for this query so that we can compare it with the new value + // to test if the cycle converged. + let last_provisional_memo = last_provisional_memo_opt.unwrap_or_else(|| { + // This is our first time around the loop; a provisional value must have been + // inserted into the memo table when the cycle was hit, so let's pull our + // initial provisional value from there. + let memo = state + .get_memo(zalsa, id, memo_ingredient_index) + .unwrap_or_else(|| { + unreachable!( + "{database_key_index:#?} is a cycle head, \ + but no provisional memo found" + ) + }); + + debug_assert!( + !memo + .header() + .revisions + .verified_final + .load(std::sync::atomic::Ordering::Relaxed) ); + memo + }); + tracing::debug!( + "{database_key_index:?}: execute: \ + I am a cycle head, comparing last provisional value with new value" + ); - last_provisional_memo_opt = Some(new_memo); + // For FallbackImmediate, the value always converges immediately (we use the + // fallback directly). We still iterate if metadata hasn't converged. + // For Fixpoint, ask the recovery function what value to use and check convergence. + let value_converged = if strategy == CycleRecoveryStrategy::FallbackImmediate { + // Use the fallback value instead of the computed value. + state.use_fallback(zalsa, id); + true + } else { + let cycle = Cycle { + head_ids: cycle_heads.ids(), + id, + iteration: cycle_iteration.iteration_as_u32(), + }; + state.recover_from_cycle(zalsa, &cycle, last_provisional_memo) + }; - last_stale_tracked_ids = completed_query.stale_tracked_structs; + let new_cycle_heads = active_query.take_cycle_heads(); + assert_no_new_cycle_heads(&cycle_heads, new_cycle_heads, database_key_index); - continue; + let completed_query = match try_complete_cycle_head( + active_query, + claim_guard, + cycle_heads, + &last_provisional_memo.header().revisions, + outer_cycle, + iteration, + cycle_iteration, + value_converged, + ) { + Ok(completed_query) => { + break completed_query; + } + Err((completed_query, new_iteration)) => { + iteration = new_iteration; + completed_query + } }; - tracing::debug!( - "{database_key_index:?}: execute_maybe_iterate: result.revisions = {revisions:#?}", - revisions = &completed_query.revisions + let new_memo = state.insert_memo( + zalsa, + id, + current_revision, + completed_query.revisions, + memo_ingredient_index, ); - (new_value, completed_query) - } + last_provisional_memo_opt = Some(new_memo); - #[inline] - fn execute_query<'db>( - db: &'db C::DbView, - zalsa: &'db Zalsa, - active_query: ActiveQueryGuard<'db>, - opt_old_header: Option<&MemoHeader>, - ) -> (C::Output<'db>, ActiveQueryGuard<'db>) { - if let Some(old_header) = opt_old_header { - old_header.seed_active_query(zalsa, &active_query); - } + last_stale_tracked_ids = completed_query.stale_tracked_structs; + }; - // Query was not previously executed, or value is potentially - // stale, or value is absent. Let's execute! - let new_value = C::execute( - db, - C::id_to_input(zalsa, active_query.database_key_index.key_index()), - ); + tracing::debug!( + "{database_key_index:?}: execute_maybe_iterate: result.revisions = {revisions:#?}", + revisions = &completed_query.revisions + ); - (new_value, active_query) - } + completed_query } struct PreviousIteration { @@ -530,22 +675,6 @@ struct PoisonProvisionalIfPanicking<'a, C: Configuration> { memo_ingredient_index: MemoIngredientIndex, } -impl<'a, C: Configuration> PoisonProvisionalIfPanicking<'a, C> { - fn new( - ingredient: &'a IngredientImpl, - zalsa: &'a Zalsa, - id: Id, - memo_ingredient_index: MemoIngredientIndex, - ) -> Self { - Self { - ingredient, - zalsa, - id, - memo_ingredient_index, - } - } -} - impl Drop for PoisonProvisionalIfPanicking<'_, C> { fn drop(&mut self) { if thread::panicking() { diff --git a/src/function/fetch.rs b/src/function/fetch.rs index 0a4037adc..ba259cfe4 100644 --- a/src/function/fetch.rs +++ b/src/function/fetch.rs @@ -1,7 +1,9 @@ use crate::cycle::{CycleRecoveryStrategy, IterationStamp}; +use crate::database::RawDatabase; use crate::function::eviction::EvictionPolicy; -use crate::function::memo::Memo; -use crate::function::sync::ClaimResult; +use crate::function::execute::{QueryState, QueryStateImpl, execute_erased}; +use crate::function::memo::{ErasedMemo, Memo}; +use crate::function::sync::{ClaimResult, SyncTable}; use crate::function::{Configuration, IngredientImpl, Reentrancy}; use crate::zalsa::{MemoIngredientIndex, Zalsa}; use crate::zalsa_local::{QueryRevisions, ZalsaLocal}; @@ -113,141 +115,165 @@ where memo_ingredient_index: MemoIngredientIndex, ) -> Option<&'db Memo> { let database_key_index = self.database_key_index(id); - // Try to claim this query: if someone else has claimed it already, go back and start again. - let claim_guard = match self - .sync_table - .try_claim(zalsa, zalsa_local, id, Reentrancy::Allow) - { - ClaimResult::Claimed(guard) => guard, - ClaimResult::Running(blocked_on) => { - let _ = blocked_on.block_on(zalsa); - return None; - } - ClaimResult::Cycle { .. } => { - return Some(self.fetch_cold_cycle( - zalsa, - zalsa_local, - db, - id, - database_key_index, - memo_ingredient_index, - )); - } - }; - - // Now that we've claimed the item, check again to see if there's a "hot" value. - let opt_old_memo = self.get_memo_from_table_for(zalsa, id, memo_ingredient_index); - - if let Some(old_memo) = opt_old_memo { - if old_memo.value.is_some() - && old_memo.header.verify_memo( - db.into(), - &claim_guard, - C::CYCLE_STRATEGY, - #[cfg(feature = "detailed-trace")] - true, - ) - { - // SAFETY: memo is present in memo_map and we have verified that it is - // still valid for the current revision. - return unsafe { Some(self.extend_memo_lifetime(old_memo)) }; - } + let mut state = QueryStateImpl::new(self, db); + let memo = fetch_cold_erased( + &mut state, + &self.sync_table, + zalsa, + zalsa_local, + db.into(), + database_key_index, + memo_ingredient_index, + C::CYCLE_STRATEGY, + )?; + Some(memo.downcast::()) + } +} + +#[allow(clippy::too_many_arguments)] +fn fetch_cold_erased<'db>( + state: &mut dyn QueryState<'db>, + sync_table: &'db SyncTable, + zalsa: &'db Zalsa, + zalsa_local: &'db ZalsaLocal, + db: RawDatabase<'db>, + database_key_index: DatabaseKeyIndex, + memo_ingredient_index: MemoIngredientIndex, + strategy: CycleRecoveryStrategy, +) -> Option> { + let id = database_key_index.key_index(); + + // Try to claim this query: if someone else has claimed it already, go back and start again. + let claim_guard = match sync_table.try_claim(zalsa, zalsa_local, id, Reentrancy::Allow) { + ClaimResult::Claimed(guard) => guard, + ClaimResult::Running(blocked_on) => { + let _ = blocked_on.block_on(zalsa); + return None; + } + ClaimResult::Cycle { .. } => { + return Some(fetch_cold_cycle_erased( + state, + zalsa, + zalsa_local, + database_key_index, + memo_ingredient_index, + strategy, + )); } + }; - self.execute(db, claim_guard, opt_old_memo) + // Now that we've claimed the item, check again to see if there's a hot value. + let opt_old_memo = state.get_memo(zalsa, id, memo_ingredient_index); + + if let Some(old_memo) = opt_old_memo { + if old_memo.has_value() + && old_memo.header().verify_memo( + db, + &claim_guard, + strategy, + #[cfg(feature = "detailed-trace")] + true, + ) + { + return Some(old_memo); + } } - #[cold] - #[inline(never)] - fn fetch_cold_cycle<'db>( - &'db self, - zalsa: &'db Zalsa, - zalsa_local: &'db ZalsaLocal, - db: &'db C::DbView, - id: Id, - database_key_index: DatabaseKeyIndex, - memo_ingredient_index: MemoIngredientIndex, - ) -> &'db Memo { - // no provisional value; create/insert/return initial provisional value - match C::CYCLE_STRATEGY { - // SAFETY: We do not access the query stack reentrantly. - CycleRecoveryStrategy::Panic => unsafe { - zalsa_local.with_query_stack_unchecked(|stack| { - panic!( - "dependency graph cycle when querying {database_key_index:#?}, \ + execute_erased( + state, + claim_guard, + opt_old_memo, + memo_ingredient_index, + strategy, + ) +} + +#[cold] +fn fetch_cold_cycle_erased<'db>( + state: &mut dyn QueryState<'db>, + zalsa: &'db Zalsa, + zalsa_local: &'db ZalsaLocal, + database_key_index: DatabaseKeyIndex, + memo_ingredient_index: MemoIngredientIndex, + strategy: CycleRecoveryStrategy, +) -> ErasedMemo<'db> { + let id = database_key_index.key_index(); + + match strategy { + // SAFETY: We do not access the query stack reentrantly. + CycleRecoveryStrategy::Panic => unsafe { + zalsa_local.with_query_stack_unchecked(|stack| { + panic!( + "dependency graph cycle when querying {database_key_index:#?}, \ set cycle_fn/cycle_initial to fixpoint iterate.\n\ Query stack:\n{stack:#?}", - ); - }) - }, - CycleRecoveryStrategy::Fixpoint | CycleRecoveryStrategy::FallbackImmediate => { - let cancellation_count = zalsa.runtime().cancellation_count(); - // check if there's a provisional value for this query - // Note we don't `validate_may_be_provisional` the memo here as we want to reuse an - // existing provisional memo if it exists - let memo_guard = self.get_memo_from_table_for(zalsa, id, memo_ingredient_index); - if let Some(memo) = &memo_guard { - let revisions = &memo.header.revisions; - // Don't replace a poisoned memo from this execution with a new initial value. - if memo.value.is_none() - && memo.header.may_be_provisional() - && memo.header.verified_at.load() == zalsa.current_revision() - && revisions.iteration().cancellation_count() == cancellation_count - { - Cancelled::PropagatedPanic.throw(); - } - - // Ideally, we'd use the last provisional memo even if it wasn't a cycle head in the last iteration - // but that would require inserting itself as a cycle head, which either requires clone - // on the value OR a concurrent `Vec` for cycle heads. - if memo.header.verified_at.load() == zalsa.current_revision() - && memo.value.is_some() - && revisions.iteration().cancellation_count() == cancellation_count - && revisions.cycle_heads().contains(&database_key_index) - { - revisions - .cycle_heads() - .remove_all_except(database_key_index); - - crate::tracing::debug!( - "hit cycle at {database_key_index:#?}, \ - returning last provisional value: {:#?}", - revisions - ); - - // SAFETY: memo is present in memo_map. - return unsafe { self.extend_memo_lifetime(memo) }; - } + ); + }) + }, + CycleRecoveryStrategy::Fixpoint | CycleRecoveryStrategy::FallbackImmediate => { + let cancellation_count = zalsa.runtime().cancellation_count(); + // Don't validate provisional memos here: an existing value should be reused. + let current_memo = state.get_memo(zalsa, id, memo_ingredient_index); + + if let Some(memo) = current_memo { + let header = memo.header(); + + // Don't replace a poisoned memo from this execution with a new initial value. + if !memo.has_value() + && header.may_be_provisional() + && header.verified_at.load() == zalsa.current_revision() + && header.revisions.iteration().cancellation_count() == cancellation_count + { + Cancelled::PropagatedPanic.throw(); } + } + + let current_memo = current_memo.filter(|memo| { + let header = memo.header(); + header.verified_at.load() == zalsa.current_revision() + && memo.has_value() + && header.revisions.iteration().cancellation_count() == cancellation_count + }); + + // Ideally, any current provisional value could be reused. Reusing a value that was + // not a cycle head in the last iteration would require inserting itself as a head, + // which in turn requires cloning the value or making the cycle-head list concurrent. + if let Some(memo) = current_memo.filter(|memo| { + memo.header() + .revisions + .cycle_heads() + .contains(&database_key_index) + }) { + memo.header() + .revisions + .cycle_heads() + .remove_all_except(database_key_index); crate::tracing::debug!( "hit cycle at {database_key_index:#?}, \ - inserting and returning fixpoint initial value" + returning last provisional value: {:#?}", + memo.header().revisions ); - - let iteration = memo_guard - .and_then(|old_memo| { - let revisions = &old_memo.header.revisions; - if old_memo.header.verified_at.load() == zalsa.current_revision() - && old_memo.value.is_some() - && revisions.iteration().cancellation_count() == cancellation_count - { - Some(revisions.iteration()) - } else { - None - } - }) - .unwrap_or_else(|| IterationStamp::initial(cancellation_count)); - let revisions = QueryRevisions::fixpoint_initial(database_key_index, iteration); - - let initial_value = C::cycle_initial(db, id, C::id_to_input(zalsa, id)); - self.insert_memo( - zalsa, - id, - Memo::new(Some(initial_value), zalsa.current_revision(), revisions), - memo_ingredient_index, - ) + return memo; } + + crate::tracing::debug!( + "hit cycle at {database_key_index:#?}, \ + inserting and returning fixpoint initial value" + ); + + let iteration = current_memo + .map(|memo| memo.header().revisions.iteration()) + .unwrap_or_else(|| IterationStamp::initial(cancellation_count)); + let revisions = QueryRevisions::fixpoint_initial(database_key_index, iteration); + state.use_fallback(zalsa, id); + state.insert_memo( + zalsa, + id, + zalsa.current_revision(), + revisions, + memo_ingredient_index, + ) } } } diff --git a/src/function/memo.rs b/src/function/memo.rs index b11c0cd54..f8d353406 100644 --- a/src/function/memo.rs +++ b/src/function/memo.rs @@ -249,7 +249,7 @@ impl MemoHeader { } } - /// Returns `true` if this memo was part of a cycle in it's last iteration. + /// Returns `true` if this memo was part of a cycle in its last iteration. #[inline(always)] pub(super) fn was_cycle_participant(&self) -> bool { !self.revisions.cycle_heads().is_empty() From 5ed36e47051cf308e16127596c39cedd9cfaa00c Mon Sep 17 00:00:00 2001 From: Micha Reiser Date: Sat, 27 Jun 2026 14:37:31 +0000 Subject: [PATCH 3/8] perf: keep the fetch path generic --- src/function/execute.rs | 31 ++- src/function/fetch.rs | 283 ++++++++++++++-------------- src/function/maybe_changed_after.rs | 2 +- 3 files changed, 168 insertions(+), 148 deletions(-) diff --git a/src/function/execute.rs b/src/function/execute.rs index 09b8af2a5..6842778fa 100644 --- a/src/function/execute.rs +++ b/src/function/execute.rs @@ -86,8 +86,8 @@ impl<'db, C: Configuration> QueryStateImpl<'db, C> { id: Id, memo_ingredient_index: MemoIngredientIndex, ) -> MemoSlot<'db> { - // SAFETY: Replaced memo allocations remain in deleted_entries until the next revision. - // The database is borrowed for 'db, so a new revision cannot begin while this state can + // SAFETY: Replaced memo allocations remain in `deleted_entries` until the next revision. + // The database is borrowed for `'db`, so a new revision cannot begin while this state can // still observe an allocation. unsafe { MemoSlot::new( @@ -98,6 +98,33 @@ impl<'db, C: Configuration> QueryStateImpl<'db, C> { } } +impl IngredientImpl { + /// Executes this query through the shared query lifecycle and restores its typed memo. + pub(super) fn execute<'db>( + &'db self, + db: &'db C::DbView, + claim_guard: ClaimGuard<'db>, + opt_old_memo: Option<&'db Memo>, + memo_ingredient_index: MemoIngredientIndex, + ) -> Option<&'db Memo> { + let id = claim_guard.database_key_index().key_index(); + let mut state = QueryStateImpl::new(self, db); + let opt_old_memo = opt_old_memo.map(|_| { + state + .get_memo(claim_guard.zalsa(), id, memo_ingredient_index) + .expect("typed old memo came from this memo table") + }); + let memo = execute_erased( + &mut state, + claim_guard, + opt_old_memo, + memo_ingredient_index, + C::CYCLE_STRATEGY, + )?; + Some(memo.downcast::()) + } +} + impl<'db, C: Configuration> QueryState<'db> for QueryStateImpl<'db, C> { fn execute_query(&mut self, zalsa: &'db Zalsa, id: Id) { self.value = Some(C::execute(self.db, C::id_to_input(zalsa, id))); diff --git a/src/function/fetch.rs b/src/function/fetch.rs index ba259cfe4..1f7eb5637 100644 --- a/src/function/fetch.rs +++ b/src/function/fetch.rs @@ -1,9 +1,8 @@ use crate::cycle::{CycleRecoveryStrategy, IterationStamp}; -use crate::database::RawDatabase; use crate::function::eviction::EvictionPolicy; -use crate::function::execute::{QueryState, QueryStateImpl, execute_erased}; +use crate::function::execute::{QueryState, QueryStateImpl}; use crate::function::memo::{ErasedMemo, Memo}; -use crate::function::sync::{ClaimResult, SyncTable}; +use crate::function::sync::ClaimResult; use crate::function::{Configuration, IngredientImpl, Reentrancy}; use crate::zalsa::{MemoIngredientIndex, Zalsa}; use crate::zalsa_local::{QueryRevisions, ZalsaLocal}; @@ -115,165 +114,159 @@ where memo_ingredient_index: MemoIngredientIndex, ) -> Option<&'db Memo> { let database_key_index = self.database_key_index(id); - let mut state = QueryStateImpl::new(self, db); - let memo = fetch_cold_erased( - &mut state, - &self.sync_table, - zalsa, - zalsa_local, - db.into(), - database_key_index, - memo_ingredient_index, - C::CYCLE_STRATEGY, - )?; - Some(memo.downcast::()) - } -} + // Try to claim this query: if someone else has claimed it already, go back and start again. + let claim_guard = match self + .sync_table + .try_claim(zalsa, zalsa_local, id, Reentrancy::Allow) + { + ClaimResult::Claimed(guard) => guard, + ClaimResult::Running(blocked_on) => { + let _ = blocked_on.block_on(zalsa); + return None; + } + ClaimResult::Cycle { .. } => { + return Some(self.fetch_cold_cycle( + db, + zalsa, + zalsa_local, + database_key_index, + memo_ingredient_index, + )); + } + }; + + // Now that we've claimed the item, check again to see if there's a hot value. + let opt_old_memo = self.get_memo_from_table_for(zalsa, id, memo_ingredient_index); + + if let Some(old_memo) = opt_old_memo { + if old_memo.value.is_some() + && old_memo.header.verify_memo( + db.into(), + &claim_guard, + C::CYCLE_STRATEGY, + #[cfg(feature = "detailed-trace")] + true, + ) + { + // SAFETY: The memo is present in the memo table, and we verified that it is valid + // for the current revision. + return unsafe { Some(self.extend_memo_lifetime(old_memo)) }; + } + } -#[allow(clippy::too_many_arguments)] -fn fetch_cold_erased<'db>( - state: &mut dyn QueryState<'db>, - sync_table: &'db SyncTable, - zalsa: &'db Zalsa, - zalsa_local: &'db ZalsaLocal, - db: RawDatabase<'db>, - database_key_index: DatabaseKeyIndex, - memo_ingredient_index: MemoIngredientIndex, - strategy: CycleRecoveryStrategy, -) -> Option> { - let id = database_key_index.key_index(); + self.execute(db, claim_guard, opt_old_memo, memo_ingredient_index) + } - // Try to claim this query: if someone else has claimed it already, go back and start again. - let claim_guard = match sync_table.try_claim(zalsa, zalsa_local, id, Reentrancy::Allow) { - ClaimResult::Claimed(guard) => guard, - ClaimResult::Running(blocked_on) => { - let _ = blocked_on.block_on(zalsa); - return None; - } - ClaimResult::Cycle { .. } => { - return Some(fetch_cold_cycle_erased( - state, - zalsa, - zalsa_local, - database_key_index, - memo_ingredient_index, - strategy, - )); - } - }; - - // Now that we've claimed the item, check again to see if there's a hot value. - let opt_old_memo = state.get_memo(zalsa, id, memo_ingredient_index); - - if let Some(old_memo) = opt_old_memo { - if old_memo.has_value() - && old_memo.header().verify_memo( - db, - &claim_guard, - strategy, - #[cfg(feature = "detailed-trace")] - true, - ) - { - return Some(old_memo); + #[cold] + #[inline(never)] + fn fetch_cold_cycle<'db>( + &'db self, + db: &'db C::DbView, + zalsa: &'db Zalsa, + zalsa_local: &'db ZalsaLocal, + database_key_index: DatabaseKeyIndex, + memo_ingredient_index: MemoIngredientIndex, + ) -> &'db Memo { + match C::CYCLE_STRATEGY { + CycleRecoveryStrategy::Panic => fetch_cold_cycle_panic(zalsa_local, database_key_index), + CycleRecoveryStrategy::FallbackImmediate | CycleRecoveryStrategy::Fixpoint => { + let mut state = QueryStateImpl::new(self, db); + let memo = fetch_cold_cycle_recoverable_erased( + &mut state, + zalsa, + database_key_index, + memo_ingredient_index, + ); + memo.downcast::() + } } } +} - execute_erased( - state, - claim_guard, - opt_old_memo, - memo_ingredient_index, - strategy, - ) +#[cold] +fn fetch_cold_cycle_panic(zalsa_local: &ZalsaLocal, database_key_index: DatabaseKeyIndex) -> ! { + // SAFETY: We do not access the query stack reentrantly. + unsafe { + zalsa_local.with_query_stack_unchecked(|stack| { + panic!( + "dependency graph cycle when querying {database_key_index:#?}, \ + set cycle_fn/cycle_initial to fixpoint iterate.\n\ + Query stack:\n{stack:#?}", + ); + }) + } } #[cold] -fn fetch_cold_cycle_erased<'db>( +fn fetch_cold_cycle_recoverable_erased<'db>( state: &mut dyn QueryState<'db>, zalsa: &'db Zalsa, - zalsa_local: &'db ZalsaLocal, database_key_index: DatabaseKeyIndex, memo_ingredient_index: MemoIngredientIndex, - strategy: CycleRecoveryStrategy, ) -> ErasedMemo<'db> { let id = database_key_index.key_index(); - match strategy { - // SAFETY: We do not access the query stack reentrantly. - CycleRecoveryStrategy::Panic => unsafe { - zalsa_local.with_query_stack_unchecked(|stack| { - panic!( - "dependency graph cycle when querying {database_key_index:#?}, \ - set cycle_fn/cycle_initial to fixpoint iterate.\n\ - Query stack:\n{stack:#?}", - ); - }) - }, - CycleRecoveryStrategy::Fixpoint | CycleRecoveryStrategy::FallbackImmediate => { - let cancellation_count = zalsa.runtime().cancellation_count(); - // Don't validate provisional memos here: an existing value should be reused. - let current_memo = state.get_memo(zalsa, id, memo_ingredient_index); - - if let Some(memo) = current_memo { - let header = memo.header(); - - // Don't replace a poisoned memo from this execution with a new initial value. - if !memo.has_value() - && header.may_be_provisional() - && header.verified_at.load() == zalsa.current_revision() - && header.revisions.iteration().cancellation_count() == cancellation_count - { - Cancelled::PropagatedPanic.throw(); - } - } + let cancellation_count = zalsa.runtime().cancellation_count(); + // Don't validate provisional memos here: an existing value should be reused. + let current_memo = state.get_memo(zalsa, id, memo_ingredient_index); - let current_memo = current_memo.filter(|memo| { - let header = memo.header(); - header.verified_at.load() == zalsa.current_revision() - && memo.has_value() - && header.revisions.iteration().cancellation_count() == cancellation_count - }); - - // Ideally, any current provisional value could be reused. Reusing a value that was - // not a cycle head in the last iteration would require inserting itself as a head, - // which in turn requires cloning the value or making the cycle-head list concurrent. - if let Some(memo) = current_memo.filter(|memo| { - memo.header() - .revisions - .cycle_heads() - .contains(&database_key_index) - }) { - memo.header() - .revisions - .cycle_heads() - .remove_all_except(database_key_index); - - crate::tracing::debug!( - "hit cycle at {database_key_index:#?}, \ - returning last provisional value: {:#?}", - memo.header().revisions - ); - return memo; - } - - crate::tracing::debug!( - "hit cycle at {database_key_index:#?}, \ - inserting and returning fixpoint initial value" - ); + if let Some(memo) = current_memo { + let header = memo.header(); - let iteration = current_memo - .map(|memo| memo.header().revisions.iteration()) - .unwrap_or_else(|| IterationStamp::initial(cancellation_count)); - let revisions = QueryRevisions::fixpoint_initial(database_key_index, iteration); - state.use_fallback(zalsa, id); - state.insert_memo( - zalsa, - id, - zalsa.current_revision(), - revisions, - memo_ingredient_index, - ) + // Don't replace a poisoned memo from this execution with a new initial value. + if !memo.has_value() + && header.may_be_provisional() + && header.verified_at.load() == zalsa.current_revision() + && header.revisions.iteration().cancellation_count() == cancellation_count + { + Cancelled::PropagatedPanic.throw(); } } + + let current_memo = current_memo.filter(|memo| { + let header = memo.header(); + header.verified_at.load() == zalsa.current_revision() + && memo.has_value() + && header.revisions.iteration().cancellation_count() == cancellation_count + }); + + // Ideally, any current provisional value could be reused. Reusing a value that was not a + // cycle head in the last iteration would require inserting itself as a head, which in turn + // requires cloning the value or making the cycle-head list concurrent. + if let Some(memo) = current_memo.filter(|memo| { + memo.header() + .revisions + .cycle_heads() + .contains(&database_key_index) + }) { + memo.header() + .revisions + .cycle_heads() + .remove_all_except(database_key_index); + + crate::tracing::debug!( + "hit cycle at {database_key_index:#?}, \ + returning last provisional value: {:#?}", + memo.header().revisions + ); + return memo; + } + + crate::tracing::debug!( + "hit cycle at {database_key_index:#?}, \ + inserting and returning fixpoint initial value" + ); + + let iteration = current_memo + .map(|memo| memo.header().revisions.iteration()) + .unwrap_or_else(|| IterationStamp::initial(cancellation_count)); + let revisions = QueryRevisions::fixpoint_initial(database_key_index, iteration); + state.use_fallback(zalsa, id); + state.insert_memo( + zalsa, + id, + zalsa.current_revision(), + revisions, + memo_ingredient_index, + ) } diff --git a/src/function/maybe_changed_after.rs b/src/function/maybe_changed_after.rs index 15c0be5ee..592ea7fa2 100644 --- a/src/function/maybe_changed_after.rs +++ b/src/function/maybe_changed_after.rs @@ -248,7 +248,7 @@ where return Some(VerifyResult::changed()); } - let memo = self.execute(db, claim_guard, Some(old_memo))?; + let memo = self.execute(db, claim_guard, Some(old_memo), memo_ingredient_index)?; let changed_at = memo.header.revisions.changed_at; // Always assume that a provisional value has changed. From b7287613be669ec46b3cfc6ff5922152bc7e2c6f Mon Sep 17 00:00:00 2001 From: Micha Reiser Date: Sat, 27 Jun 2026 19:00:03 +0000 Subject: [PATCH 4/8] refactor: keep hot tracked query paths typed --- src/function/execute.rs | 357 +++++++++++++++++++++------------------- 1 file changed, 188 insertions(+), 169 deletions(-) diff --git a/src/function/execute.rs b/src/function/execute.rs index 6842778fa..f903189f4 100644 --- a/src/function/execute.rs +++ b/src/function/execute.rs @@ -45,24 +45,45 @@ pub(super) trait QueryState<'db> { revisions: QueryRevisions, memo_ingredient_index: MemoIngredientIndex, ) -> ErasedMemo<'db>; +} - fn execute_iterated( - &mut self, - zalsa: &'db Zalsa, - opt_old_memo: Option>, - claim_guard: &mut ClaimGuard<'db>, - memo_ingredient_index: MemoIngredientIndex, - strategy: CycleRecoveryStrategy, - ) -> CompletedQuery; +#[derive(Copy, Clone)] +enum CyclePolicy { + FallbackImmediate, + Fixpoint, +} - fn finish_memo( - &mut self, +impl CyclePolicy { + /// Adjusts the query value before completing a non-head cycle participant. + /// + /// Fallback recovery replaces the computed value with the fallback. Fixpoint recovery keeps + /// the computed value unchanged. + fn complete_participant<'db>(self, state: &mut dyn QueryState<'db>, zalsa: &'db Zalsa, id: Id) { + if matches!(self, Self::FallbackImmediate) { + state.use_fallback(zalsa, id); + } + } + + /// Recovers the value produced by the latest cycle-head iteration. + /// + /// Returns whether the query value converged. A `true` result does not mean the entire cycle + /// converged: cycle-head metadata is compared separately by `try_complete_cycle_head`. + fn recover_cycle_head<'db>( + self, + state: &mut dyn QueryState<'db>, zalsa: &'db Zalsa, - database_key_index: DatabaseKeyIndex, - opt_old_memo: Option>, - completed_query: CompletedQuery, - memo_ingredient_index: MemoIngredientIndex, - ) -> ErasedMemo<'db>; + id: Id, + cycle: &Cycle, + last_provisional_memo: ErasedMemo<'db>, + ) -> bool { + match self { + Self::FallbackImmediate => { + state.use_fallback(zalsa, id); + true + } + Self::Fixpoint => state.recover_from_cycle(zalsa, cycle, last_provisional_memo), + } + } } pub(super) struct QueryStateImpl<'db, C: Configuration> { @@ -107,21 +128,155 @@ impl IngredientImpl { opt_old_memo: Option<&'db Memo>, memo_ingredient_index: MemoIngredientIndex, ) -> Option<&'db Memo> { - let id = claim_guard.database_key_index().key_index(); + match C::CYCLE_STRATEGY { + CycleRecoveryStrategy::Panic => { + self.execute_panic(db, claim_guard, opt_old_memo, memo_ingredient_index) + } + CycleRecoveryStrategy::FallbackImmediate => self.execute_cycle( + db, + claim_guard, + opt_old_memo, + memo_ingredient_index, + CyclePolicy::FallbackImmediate, + ), + CycleRecoveryStrategy::Fixpoint => self.execute_cycle( + db, + claim_guard, + opt_old_memo, + memo_ingredient_index, + CyclePolicy::Fixpoint, + ), + } + } + + fn execute_cycle<'db>( + &'db self, + db: &'db C::DbView, + mut claim_guard: ClaimGuard<'db>, + opt_old_memo: Option<&'db Memo>, + memo_ingredient_index: MemoIngredientIndex, + policy: CyclePolicy, + ) -> Option<&'db Memo> { + let database_key_index = claim_guard.database_key_index(); + let zalsa = claim_guard.zalsa(); + let id = database_key_index.key_index(); + + crate::tracing::info!("{:?}: executing query", database_key_index); + zalsa.event(&|| { + Event::new(EventKind::WillExecute { + database_key: database_key_index, + }) + }); + + let _cancellation_guard = DisableLocalCancellationGuard::new(claim_guard.zalsa_local()); + let _poison_guard = PoisonProvisionalIfPanicking { + ingredient: self, + zalsa, + id, + memo_ingredient_index, + }; let mut state = QueryStateImpl::new(self, db); - let opt_old_memo = opt_old_memo.map(|_| { + let opt_old_memo_erased = opt_old_memo.map(|_| { state - .get_memo(claim_guard.zalsa(), id, memo_ingredient_index) + .get_memo(zalsa, id, memo_ingredient_index) .expect("typed old memo came from this memo table") }); - let memo = execute_erased( + let completed_query = execute_maybe_iterate_erased( &mut state, - claim_guard, + zalsa, + opt_old_memo_erased, + &mut claim_guard, + memo_ingredient_index, + policy, + ); + let value = state + .value + .take() + .expect("query execution must produce a value"); + let memo = self.finish_memo( + zalsa, + database_key_index, + opt_old_memo, + value, + completed_query, + memo_ingredient_index, + ); + + if claim_guard.drop() { None } else { Some(memo) } + } + + #[inline(never)] + fn execute_panic<'db>( + &'db self, + db: &'db C::DbView, + claim_guard: ClaimGuard<'db>, + opt_old_memo: Option<&'db Memo>, + memo_ingredient_index: MemoIngredientIndex, + ) -> Option<&'db Memo> { + let database_key_index = claim_guard.database_key_index(); + let zalsa = claim_guard.zalsa(); + let id = database_key_index.key_index(); + + crate::tracing::info!("{:?}: executing query", database_key_index); + zalsa.event(&|| { + Event::new(EventKind::WillExecute { + database_key: database_key_index, + }) + }); + + let active_query = claim_guard.zalsa_local().push_query(database_key_index); + if let Some(old_memo) = opt_old_memo { + old_memo.header.seed_active_query(zalsa, &active_query); + } + let new_value = C::execute(db, C::id_to_input(zalsa, id)); + let completed_query = active_query.pop(IterationStamp::default()); + + let memo = self.finish_memo( + zalsa, + database_key_index, opt_old_memo, + new_value, + completed_query, + memo_ingredient_index, + ); + + if claim_guard.drop() { None } else { Some(memo) } + } + + fn finish_memo<'db>( + &'db self, + zalsa: &'db Zalsa, + database_key_index: DatabaseKeyIndex, + opt_old_memo: Option<&'db Memo>, + value: C::Output<'db>, + mut completed_query: CompletedQuery, + memo_ingredient_index: MemoIngredientIndex, + ) -> &'db Memo { + if let Some(old_memo) = opt_old_memo { + self.backdate_if_appropriate( + old_memo, + database_key_index, + &mut completed_query.revisions, + &value, + ); + old_memo + .header + .diff_outputs(zalsa, database_key_index, &completed_query); + } + + #[cfg(not(feature = "persistence"))] + completed_query.revisions.discard_edges_if_never_change(); + + self.insert_memo( + zalsa, + database_key_index.key_index(), + Memo::new( + Some(value), + zalsa.current_revision(), + completed_query.revisions, + ), memo_ingredient_index, - C::CYCLE_STRATEGY, - )?; - Some(memo.downcast::()) + ) } } @@ -192,130 +347,6 @@ impl<'db, C: Configuration> QueryState<'db> for QueryStateImpl<'db, C> { .get_erased() .expect("memo was just inserted") } - - fn execute_iterated( - &mut self, - zalsa: &'db Zalsa, - opt_old_memo: Option>, - claim_guard: &mut ClaimGuard<'db>, - memo_ingredient_index: MemoIngredientIndex, - strategy: CycleRecoveryStrategy, - ) -> CompletedQuery { - let id = claim_guard.database_key_index().key_index(); - let _poison_guard = PoisonProvisionalIfPanicking { - ingredient: self.ingredient, - zalsa, - id, - memo_ingredient_index, - }; - execute_maybe_iterate_erased( - self, - zalsa, - opt_old_memo, - claim_guard, - memo_ingredient_index, - strategy, - ) - } - - fn finish_memo( - &mut self, - zalsa: &'db Zalsa, - database_key_index: DatabaseKeyIndex, - opt_old_memo: Option>, - mut completed_query: CompletedQuery, - memo_ingredient_index: MemoIngredientIndex, - ) -> ErasedMemo<'db> { - let id = database_key_index.key_index(); - let value = self - .value - .take() - .expect("query execution must produce a value"); - - if let Some(old_memo) = opt_old_memo { - let old_memo = old_memo.downcast::(); - // An equal output did not logically change even if an input did, so preserve its - // old `changed_at` revision. - self.ingredient.backdate_if_appropriate( - old_memo, - database_key_index, - &mut completed_query.revisions, - &value, - ); - // Diff the new outputs with the old, to discard any no-longer-emitted - // outputs and update the tracked struct IDs for seeding the next revision. - old_memo - .header - .diff_outputs(zalsa, database_key_index, &completed_query); - } - - #[cfg(not(feature = "persistence"))] - completed_query.revisions.discard_edges_if_never_change(); - - self.ingredient.insert_memo( - zalsa, - id, - Memo::new( - Some(value), - zalsa.current_revision(), - completed_query.revisions, - ), - memo_ingredient_index, - ); - self.memo_slot(zalsa, id, memo_ingredient_index) - .get_erased() - .expect("memo was just inserted") - } -} - -pub(super) fn execute_erased<'db>( - state: &mut dyn QueryState<'db>, - mut claim_guard: ClaimGuard<'db>, - opt_old_memo: Option>, - memo_ingredient_index: MemoIngredientIndex, - strategy: CycleRecoveryStrategy, -) -> Option> { - let database_key_index = claim_guard.database_key_index(); - let zalsa = claim_guard.zalsa(); - - crate::tracing::info!("{:?}: executing query", database_key_index); - zalsa.event(&|| { - Event::new(EventKind::WillExecute { - database_key: database_key_index, - }) - }); - - let completed_query = match strategy { - CycleRecoveryStrategy::Panic => { - let active_query = claim_guard.zalsa_local().push_query(database_key_index); - seed_query_from_old_memo(zalsa, &active_query, opt_old_memo); - state.execute_query(zalsa, database_key_index.key_index()); - - // Ordinary queries don't need an epoch stamp. Keeping the default avoids allocating - // `QueryRevisionsExtra` after a revision-preserving cancellation. - active_query.pop(IterationStamp::default()) - } - CycleRecoveryStrategy::FallbackImmediate | CycleRecoveryStrategy::Fixpoint => { - let _cancellation_guard = DisableLocalCancellationGuard::new(claim_guard.zalsa_local()); - state.execute_iterated( - zalsa, - opt_old_memo, - &mut claim_guard, - memo_ingredient_index, - strategy, - ) - } - }; - - let memo = state.finish_memo( - zalsa, - database_key_index, - opt_old_memo, - completed_query, - memo_ingredient_index, - ); - - if claim_guard.drop() { None } else { Some(memo) } } fn seed_query_from_old_memo( @@ -336,7 +367,7 @@ fn execute_maybe_iterate_erased<'db>( opt_old_memo: Option>, claim_guard: &mut ClaimGuard<'db>, memo_ingredient_index: MemoIngredientIndex, - strategy: CycleRecoveryStrategy, + policy: CyclePolicy, ) -> CompletedQuery { claim_guard.set_release_mode(ReleaseMode::Default); @@ -403,11 +434,7 @@ fn execute_maybe_iterate_erased<'db>( cycle_heads, outer_cycle, } => { - // FallbackImmediate uses the fallback for every participant so the result - // does not depend on query call order. - if strategy == CycleRecoveryStrategy::FallbackImmediate { - state.use_fallback(zalsa, id); - } + policy.complete_participant(state, zalsa, id); break complete_cycle_participant( active_query, @@ -449,26 +476,18 @@ fn execute_maybe_iterate_erased<'db>( ); memo }); - tracing::debug!( + crate::tracing::debug!( "{database_key_index:?}: execute: \ I am a cycle head, comparing last provisional value with new value" ); - // For FallbackImmediate, the value always converges immediately (we use the - // fallback directly). We still iterate if metadata hasn't converged. - // For Fixpoint, ask the recovery function what value to use and check convergence. - let value_converged = if strategy == CycleRecoveryStrategy::FallbackImmediate { - // Use the fallback value instead of the computed value. - state.use_fallback(zalsa, id); - true - } else { - let cycle = Cycle { - head_ids: cycle_heads.ids(), - id, - iteration: cycle_iteration.iteration_as_u32(), - }; - state.recover_from_cycle(zalsa, &cycle, last_provisional_memo) + let cycle = Cycle { + head_ids: cycle_heads.ids(), + id, + iteration: cycle_iteration.iteration_as_u32(), }; + let value_converged = + policy.recover_cycle_head(state, zalsa, id, &cycle, last_provisional_memo); let new_cycle_heads = active_query.take_cycle_heads(); assert_no_new_cycle_heads(&cycle_heads, new_cycle_heads, database_key_index); @@ -505,7 +524,7 @@ fn execute_maybe_iterate_erased<'db>( last_stale_tracked_ids = completed_query.stale_tracked_structs; }; - tracing::debug!( + crate::tracing::debug!( "{database_key_index:?}: execute_maybe_iterate: result.revisions = {revisions:#?}", revisions = &completed_query.revisions ); From f49f4ef7b57a723dd58be8b707850ae48d20209e Mon Sep 17 00:00:00 2001 From: Micha Reiser Date: Sun, 28 Jun 2026 16:11:12 +0000 Subject: [PATCH 5/8] refactor: specialize query execution by cycle strategy --- .../salsa-macro-rules/src/setup_tracked_fn.rs | 2 +- src/function.rs | 8 +- src/function/cycle_strategy.rs | 140 ++++++ src/function/execute.rs | 447 +++++++++--------- src/function/fetch.rs | 43 +- src/function/maybe_changed_after.rs | 4 +- src/function/memo.rs | 25 +- src/lib.rs | 2 +- 8 files changed, 410 insertions(+), 261 deletions(-) create mode 100644 src/function/cycle_strategy.rs diff --git a/components/salsa-macro-rules/src/setup_tracked_fn.rs b/components/salsa-macro-rules/src/setup_tracked_fn.rs index 562bfa3a2..90f03cd8e 100644 --- a/components/salsa-macro-rules/src/setup_tracked_fn.rs +++ b/components/salsa-macro-rules/src/setup_tracked_fn.rs @@ -334,7 +334,7 @@ macro_rules! setup_tracked_fn { type Eviction = $Eviction; - const CYCLE_STRATEGY: $zalsa::CycleRecoveryStrategy = $zalsa::CycleRecoveryStrategy::$cycle_recovery_strategy; + type CycleStrategy = $zalsa::function::cycle_strategy::$cycle_recovery_strategy; $($values_equal)+ diff --git a/src/function.rs b/src/function.rs index 2c576aa72..b752b481f 100644 --- a/src/function.rs +++ b/src/function.rs @@ -27,6 +27,8 @@ use crate::{Cycle, Id, Revision}; #[cfg(feature = "accumulator")] mod accumulated; mod backdate; +#[doc(hidden)] +pub mod cycle_strategy; mod delete; mod diff_outputs; mod eviction; @@ -49,7 +51,7 @@ pub type Memo = memo::Memo; /// after erasing `'db` and to use after rebranding it with a later database /// lifetime. This is guaranteed when the output implements [`crate::SalsaValue`] /// or when it is the same `'static` type for every `'db`. -pub unsafe trait Configuration: Any { +pub unsafe trait Configuration: Any + Sized { const DEBUG_NAME: &'static str; const LOCATION: crate::ingredient::Location; const PERSIST: bool; @@ -73,7 +75,7 @@ pub unsafe trait Configuration: Any { /// Determines whether this function can recover from being a participant in a cycle /// (and, if so, how). - const CYCLE_STRATEGY: CycleRecoveryStrategy; + type CycleStrategy: cycle_strategy::CycleStrategy; /// Invokes after a new result `new_value` has been computed for which an older memoized value /// existed `old_value`, or in fixpoint iteration. Returns true if the new value is equal to @@ -412,7 +414,7 @@ where self, zalsa, self.database_key_index(id), - C::CYCLE_STRATEGY, + cycle_strategy::recovery_strategy::(), flattened_input_outputs, seen, ); diff --git a/src/function/cycle_strategy.rs b/src/function/cycle_strategy.rs new file mode 100644 index 000000000..7412aefec --- /dev/null +++ b/src/function/cycle_strategy.rs @@ -0,0 +1,140 @@ +use super::execute::{CyclePolicy, CycleStateImpl}; +use super::fetch::{fetch_cold_cycle_panic, fetch_cold_cycle_recoverable_erased}; +use super::memo::Memo; +use super::{ClaimGuard, Configuration, IngredientImpl}; +use crate::DatabaseKeyIndex; +use crate::cycle::CycleRecoveryStrategy; +use crate::zalsa::{MemoIngredientIndex, Zalsa}; +use crate::zalsa_local::ZalsaLocal; + +pub struct Panic; +pub struct FallbackImmediate; +pub struct Fixpoint; + +pub struct ExecuteContext<'db, C: Configuration> { + pub(super) ingredient: &'db IngredientImpl, + pub(super) db: &'db C::DbView, + pub(super) claim_guard: ClaimGuard<'db>, + pub(super) opt_old_memo: Option<&'db Memo>, + pub(super) memo_ingredient_index: MemoIngredientIndex, +} + +pub struct ExecuteResult<'db, C: Configuration>(pub(super) Option<&'db Memo>); + +pub struct FetchCycleContext<'db, C: Configuration> { + pub(super) ingredient: &'db IngredientImpl, + pub(super) db: &'db C::DbView, + pub(super) zalsa: &'db Zalsa, + pub(super) zalsa_local: &'db ZalsaLocal, + pub(super) database_key_index: DatabaseKeyIndex, + pub(super) memo_ingredient_index: MemoIngredientIndex, +} + +pub struct FetchCycleResult<'db, C: Configuration>(pub(super) &'db Memo); + +pub trait CycleStrategy: 'static { + const RECOVERY_STRATEGY: CycleRecoveryStrategy; + + fn execute<'db>(context: ExecuteContext<'db, C>) -> ExecuteResult<'db, C>; + + fn fetch_cold_cycle<'db>(context: FetchCycleContext<'db, C>) -> FetchCycleResult<'db, C>; +} + +#[inline] +pub(super) fn recovery_strategy() -> CycleRecoveryStrategy { + >::RECOVERY_STRATEGY +} + +impl CycleStrategy for Panic { + const RECOVERY_STRATEGY: CycleRecoveryStrategy = CycleRecoveryStrategy::Panic; + + fn execute<'db>(context: ExecuteContext<'db, C>) -> ExecuteResult<'db, C> { + let ExecuteContext { + ingredient, + db, + claim_guard, + opt_old_memo, + memo_ingredient_index, + } = context; + ExecuteResult(ingredient.execute_panic( + db, + claim_guard, + opt_old_memo, + memo_ingredient_index, + )) + } + + fn fetch_cold_cycle<'db>(context: FetchCycleContext<'db, C>) -> FetchCycleResult<'db, C> { + let FetchCycleContext { + zalsa_local, + database_key_index, + .. + } = context; + fetch_cold_cycle_panic(zalsa_local, database_key_index) + } +} + +impl CycleStrategy for FallbackImmediate { + const RECOVERY_STRATEGY: CycleRecoveryStrategy = CycleRecoveryStrategy::FallbackImmediate; + + fn execute<'db>(context: ExecuteContext<'db, C>) -> ExecuteResult<'db, C> { + execute_recoverable(context, CyclePolicy::FallbackImmediate) + } + + fn fetch_cold_cycle<'db>(context: FetchCycleContext<'db, C>) -> FetchCycleResult<'db, C> { + fetch_cold_cycle_recoverable(context) + } +} + +impl CycleStrategy for Fixpoint { + const RECOVERY_STRATEGY: CycleRecoveryStrategy = CycleRecoveryStrategy::Fixpoint; + + fn execute<'db>(context: ExecuteContext<'db, C>) -> ExecuteResult<'db, C> { + execute_recoverable(context, CyclePolicy::Fixpoint) + } + + fn fetch_cold_cycle<'db>(context: FetchCycleContext<'db, C>) -> FetchCycleResult<'db, C> { + fetch_cold_cycle_recoverable(context) + } +} + +fn execute_recoverable<'db, C: Configuration>( + context: ExecuteContext<'db, C>, + policy: CyclePolicy, +) -> ExecuteResult<'db, C> { + let ExecuteContext { + ingredient, + db, + claim_guard, + opt_old_memo, + memo_ingredient_index, + } = context; + ExecuteResult(ingredient.execute_cycle( + db, + claim_guard, + opt_old_memo, + memo_ingredient_index, + policy, + )) +} + +fn fetch_cold_cycle_recoverable<'db, C: Configuration>( + context: FetchCycleContext<'db, C>, +) -> FetchCycleResult<'db, C> { + let FetchCycleContext { + ingredient, + db, + zalsa, + database_key_index, + memo_ingredient_index, + .. + } = context; + let mut state = CycleStateImpl::new(ingredient, db); + let memo = fetch_cold_cycle_recoverable_erased( + &mut state, + zalsa, + database_key_index, + memo_ingredient_index, + ); + FetchCycleResult(memo.downcast::()) +} diff --git a/src/function/execute.rs b/src/function/execute.rs index f903189f4..f46829285 100644 --- a/src/function/execute.rs +++ b/src/function/execute.rs @@ -1,7 +1,8 @@ use smallvec::SmallVec; use crate::active_query::CompletedQuery; -use crate::cycle::{CycleHeads, CycleRecoveryStrategy, IterationStamp, ProvisionalStatus}; +use crate::cycle::{CycleHeads, IterationStamp, ProvisionalStatus}; +use crate::function::cycle_strategy::{CycleStrategy, ExecuteContext}; use crate::function::memo::{ErasedMemo, Memo, MemoHeader}; use crate::function::sync::ReleaseMode; use crate::function::{ClaimGuard, ClaimResult, Configuration, IngredientImpl, Reentrancy}; @@ -15,110 +16,6 @@ use crate::zalsa_local::{ActiveQueryGuard, QueryEdge, QueryEdgeKind, QueryRevisi use crate::{Cancelled, Cycle, Revision, tracing}; use crate::{DatabaseKeyIndex, Event, EventKind, Id}; -/// Type-specific operations needed by the shared cold query lifecycle. -/// -/// Every [`ErasedMemo`] passed to a state must come from that state's ingredient. -pub(super) trait QueryState<'db> { - fn execute_query(&mut self, zalsa: &'db Zalsa, id: Id); - - fn use_fallback(&mut self, zalsa: &'db Zalsa, id: Id); - - fn recover_from_cycle( - &mut self, - zalsa: &'db Zalsa, - cycle: &Cycle, - last_provisional_memo: ErasedMemo<'db>, - ) -> bool; - - fn get_memo( - &self, - zalsa: &'db Zalsa, - id: Id, - memo_ingredient_index: MemoIngredientIndex, - ) -> Option>; - - fn insert_memo( - &mut self, - zalsa: &'db Zalsa, - id: Id, - revision: Revision, - revisions: QueryRevisions, - memo_ingredient_index: MemoIngredientIndex, - ) -> ErasedMemo<'db>; -} - -#[derive(Copy, Clone)] -enum CyclePolicy { - FallbackImmediate, - Fixpoint, -} - -impl CyclePolicy { - /// Adjusts the query value before completing a non-head cycle participant. - /// - /// Fallback recovery replaces the computed value with the fallback. Fixpoint recovery keeps - /// the computed value unchanged. - fn complete_participant<'db>(self, state: &mut dyn QueryState<'db>, zalsa: &'db Zalsa, id: Id) { - if matches!(self, Self::FallbackImmediate) { - state.use_fallback(zalsa, id); - } - } - - /// Recovers the value produced by the latest cycle-head iteration. - /// - /// Returns whether the query value converged. A `true` result does not mean the entire cycle - /// converged: cycle-head metadata is compared separately by `try_complete_cycle_head`. - fn recover_cycle_head<'db>( - self, - state: &mut dyn QueryState<'db>, - zalsa: &'db Zalsa, - id: Id, - cycle: &Cycle, - last_provisional_memo: ErasedMemo<'db>, - ) -> bool { - match self { - Self::FallbackImmediate => { - state.use_fallback(zalsa, id); - true - } - Self::Fixpoint => state.recover_from_cycle(zalsa, cycle, last_provisional_memo), - } - } -} - -pub(super) struct QueryStateImpl<'db, C: Configuration> { - ingredient: &'db IngredientImpl, - db: &'db C::DbView, - value: Option>, -} - -impl<'db, C: Configuration> QueryStateImpl<'db, C> { - pub(super) fn new(ingredient: &'db IngredientImpl, db: &'db C::DbView) -> Self { - Self { - ingredient, - db, - value: None, - } - } - - fn memo_slot( - &self, - zalsa: &'db Zalsa, - id: Id, - memo_ingredient_index: MemoIngredientIndex, - ) -> MemoSlot<'db> { - // SAFETY: Replaced memo allocations remain in `deleted_entries` until the next revision. - // The database is borrowed for `'db`, so a new revision cannot begin while this state can - // still observe an allocation. - unsafe { - MemoSlot::new( - zalsa.memo_table_for::>(id), - memo_ingredient_index, - ) - } - } -} - impl IngredientImpl { /// Executes this query through the shared query lifecycle and restores its typed memo. pub(super) fn execute<'db>( @@ -128,28 +25,19 @@ impl IngredientImpl { opt_old_memo: Option<&'db Memo>, memo_ingredient_index: MemoIngredientIndex, ) -> Option<&'db Memo> { - match C::CYCLE_STRATEGY { - CycleRecoveryStrategy::Panic => { - self.execute_panic(db, claim_guard, opt_old_memo, memo_ingredient_index) - } - CycleRecoveryStrategy::FallbackImmediate => self.execute_cycle( - db, - claim_guard, - opt_old_memo, - memo_ingredient_index, - CyclePolicy::FallbackImmediate, - ), - CycleRecoveryStrategy::Fixpoint => self.execute_cycle( - db, - claim_guard, - opt_old_memo, - memo_ingredient_index, - CyclePolicy::Fixpoint, - ), - } + report_will_execute(&claim_guard); + + >::execute(ExecuteContext { + ingredient: self, + db, + claim_guard, + opt_old_memo, + memo_ingredient_index, + }) + .0 } - fn execute_cycle<'db>( + pub(super) fn execute_cycle<'db>( &'db self, db: &'db C::DbView, mut claim_guard: ClaimGuard<'db>, @@ -161,13 +49,6 @@ impl IngredientImpl { let zalsa = claim_guard.zalsa(); let id = database_key_index.key_index(); - crate::tracing::info!("{:?}: executing query", database_key_index); - zalsa.event(&|| { - Event::new(EventKind::WillExecute { - database_key: database_key_index, - }) - }); - let _cancellation_guard = DisableLocalCancellationGuard::new(claim_guard.zalsa_local()); let _poison_guard = PoisonProvisionalIfPanicking { ingredient: self, @@ -175,12 +56,8 @@ impl IngredientImpl { id, memo_ingredient_index, }; - let mut state = QueryStateImpl::new(self, db); - let opt_old_memo_erased = opt_old_memo.map(|_| { - state - .get_memo(zalsa, id, memo_ingredient_index) - .expect("typed old memo came from this memo table") - }); + let opt_old_memo_erased = opt_old_memo.map(Memo::erase); + let mut state = CycleStateImpl::new(self, db); let completed_query = execute_maybe_iterate_erased( &mut state, zalsa, @@ -205,8 +82,7 @@ impl IngredientImpl { if claim_guard.drop() { None } else { Some(memo) } } - #[inline(never)] - fn execute_panic<'db>( + pub(super) fn execute_panic<'db>( &'db self, db: &'db C::DbView, claim_guard: ClaimGuard<'db>, @@ -217,13 +93,6 @@ impl IngredientImpl { let zalsa = claim_guard.zalsa(); let id = database_key_index.key_index(); - crate::tracing::info!("{:?}: executing query", database_key_index); - zalsa.event(&|| { - Event::new(EventKind::WillExecute { - database_key: database_key_index, - }) - }); - let active_query = claim_guard.zalsa_local().push_query(database_key_index); if let Some(old_memo) = opt_old_memo { old_memo.header.seed_active_query(zalsa, &active_query); @@ -253,12 +122,19 @@ impl IngredientImpl { memo_ingredient_index: MemoIngredientIndex, ) -> &'db Memo { if let Some(old_memo) = opt_old_memo { + // If the new value is equal to the old one, then it didn't + // really change, even if some of its inputs have. So we can + // "backdate" its `changed_at` revision to be the same as the + // old value. self.backdate_if_appropriate( old_memo, database_key_index, &mut completed_query.revisions, &value, ); + + // Diff the new outputs with the old, to discard any no-longer-emitted + // outputs and update the tracked struct IDs for seeding the next revision. old_memo .header .diff_outputs(zalsa, database_key_index, &completed_query); @@ -280,89 +156,19 @@ impl IngredientImpl { } } -impl<'db, C: Configuration> QueryState<'db> for QueryStateImpl<'db, C> { - fn execute_query(&mut self, zalsa: &'db Zalsa, id: Id) { - self.value = Some(C::execute(self.db, C::id_to_input(zalsa, id))); - } - - fn use_fallback(&mut self, zalsa: &'db Zalsa, id: Id) { - self.value = Some(C::cycle_initial(self.db, id, C::id_to_input(zalsa, id))); - } - - fn recover_from_cycle( - &mut self, - zalsa: &'db Zalsa, - cycle: &Cycle, - last_provisional_memo: ErasedMemo<'db>, - ) -> bool { - let last_provisional_memo = last_provisional_memo.downcast::(); - let last_provisional_value = last_provisional_memo.value().expect( - "`fetch_cold_cycle` should have inserted a provisional memo with Cycle::initial", - ); - let value = self - .value - .take() - .expect("query state must contain the value from the latest execution"); - let value = C::recover_from_cycle( - self.db, - cycle, - last_provisional_value, - value, - C::id_to_input(zalsa, cycle.id), - ); - let converged = C::values_equal(&value, last_provisional_value); - self.value = Some(value); - converged - } - - fn get_memo( - &self, - zalsa: &'db Zalsa, - id: Id, - memo_ingredient_index: MemoIngredientIndex, - ) -> Option> { - self.memo_slot(zalsa, id, memo_ingredient_index) - .get_erased() - } - - fn insert_memo( - &mut self, - zalsa: &'db Zalsa, - id: Id, - revision: Revision, - revisions: QueryRevisions, - memo_ingredient_index: MemoIngredientIndex, - ) -> ErasedMemo<'db> { - let value = self - .value - .take() - .expect("query state must contain a value before memo insertion"); - self.ingredient.insert_memo( - zalsa, - id, - Memo::new(Some(value), revision, revisions), - memo_ingredient_index, - ); - self.memo_slot(zalsa, id, memo_ingredient_index) - .get_erased() - .expect("memo was just inserted") - } -} - -fn seed_query_from_old_memo( - zalsa: &Zalsa, - active_query: &ActiveQueryGuard<'_>, - old_memo: Option>, -) { - let Some(old_memo) = old_memo else { - return; - }; +fn report_will_execute(claim_guard: &ClaimGuard<'_>) { + let database_key_index = claim_guard.database_key_index(); - old_memo.header().seed_active_query(zalsa, active_query); + crate::tracing::info!("{:?}: executing query", database_key_index); + claim_guard.zalsa().event(&|| { + Event::new(EventKind::WillExecute { + database_key: database_key_index, + }) + }); } fn execute_maybe_iterate_erased<'db>( - state: &mut dyn QueryState<'db>, + state: &mut dyn CycleState<'db>, zalsa: &'db Zalsa, opt_old_memo: Option>, claim_guard: &mut ClaimGuard<'db>, @@ -459,7 +265,7 @@ fn execute_maybe_iterate_erased<'db>( // inserted into the memo table when the cycle was hit, so let's pull our // initial provisional value from there. let memo = state - .get_memo(zalsa, id, memo_ingredient_index) + .provisional_memo(zalsa, id, memo_ingredient_index) .unwrap_or_else(|| { unreachable!( "{database_key_index:#?} is a cycle head, \ @@ -511,7 +317,7 @@ fn execute_maybe_iterate_erased<'db>( } }; - let new_memo = state.insert_memo( + let new_memo = state.insert_provisional_memo( zalsa, id, current_revision, @@ -532,6 +338,191 @@ fn execute_maybe_iterate_erased<'db>( completed_query } +#[derive(Copy, Clone)] +pub(super) enum CyclePolicy { + FallbackImmediate, + Fixpoint, +} + +impl CyclePolicy { + /// Adjusts the query value before completing a non-head cycle participant. + /// + /// Fallback recovery replaces the computed value with the fallback. Fixpoint recovery keeps + /// the computed value unchanged. + fn complete_participant<'db>(self, state: &mut dyn CycleState<'db>, zalsa: &'db Zalsa, id: Id) { + if matches!(self, Self::FallbackImmediate) { + state.use_fallback(zalsa, id); + } + } + + /// Recovers the value produced by the latest cycle-head iteration. + /// + /// Returns whether the query value converged. A `true` result does not mean the entire cycle + /// converged: cycle-head metadata is compared separately by `try_complete_cycle_head`. + fn recover_cycle_head<'db>( + self, + state: &mut dyn CycleState<'db>, + zalsa: &'db Zalsa, + id: Id, + cycle: &Cycle, + last_provisional_memo: ErasedMemo<'db>, + ) -> bool { + match self { + Self::FallbackImmediate => { + state.use_fallback(zalsa, id); + true + } + Self::Fixpoint => state.recover_from_cycle(zalsa, cycle, last_provisional_memo), + } + } +} + +/// Type-specific operations needed by recoverable cycle handling. +/// +/// This erased bridge is used only by fallback and fixpoint cycle strategies. Every +/// [`ErasedMemo`] passed to a state must come from that state's ingredient. +pub(super) trait CycleState<'db> { + fn execute_query(&mut self, zalsa: &'db Zalsa, id: Id); + + fn use_fallback(&mut self, zalsa: &'db Zalsa, id: Id); + + fn recover_from_cycle( + &mut self, + zalsa: &'db Zalsa, + cycle: &Cycle, + last_provisional_memo: ErasedMemo<'db>, + ) -> bool; + + fn provisional_memo( + &self, + zalsa: &'db Zalsa, + id: Id, + memo_ingredient_index: MemoIngredientIndex, + ) -> Option>; + + fn insert_provisional_memo( + &mut self, + zalsa: &'db Zalsa, + id: Id, + revision: Revision, + revisions: QueryRevisions, + memo_ingredient_index: MemoIngredientIndex, + ) -> ErasedMemo<'db>; +} + +pub(super) struct CycleStateImpl<'db, C: Configuration> { + ingredient: &'db IngredientImpl, + db: &'db C::DbView, + value: Option>, +} + +impl<'db, C: Configuration> CycleStateImpl<'db, C> { + pub(super) fn new(ingredient: &'db IngredientImpl, db: &'db C::DbView) -> Self { + Self { + ingredient, + db, + value: None, + } + } + + fn memo_slot( + &self, + zalsa: &'db Zalsa, + id: Id, + memo_ingredient_index: MemoIngredientIndex, + ) -> MemoSlot<'db> { + // SAFETY: Replaced memo allocations remain in `deleted_entries` until the next revision. + // The database is borrowed for `'db`, so a new revision cannot begin while this state can + // still observe an allocation. + unsafe { + MemoSlot::new( + zalsa.memo_table_for::>(id), + memo_ingredient_index, + ) + } + } +} + +impl<'db, C: Configuration> CycleState<'db> for CycleStateImpl<'db, C> { + fn execute_query(&mut self, zalsa: &'db Zalsa, id: Id) { + self.value = Some(C::execute(self.db, C::id_to_input(zalsa, id))); + } + + fn use_fallback(&mut self, zalsa: &'db Zalsa, id: Id) { + self.value = Some(C::cycle_initial(self.db, id, C::id_to_input(zalsa, id))); + } + + fn recover_from_cycle( + &mut self, + zalsa: &'db Zalsa, + cycle: &Cycle, + last_provisional_memo: ErasedMemo<'db>, + ) -> bool { + let last_provisional_memo = last_provisional_memo.downcast::(); + let last_provisional_value = last_provisional_memo.value().expect( + "`fetch_cold_cycle` should have inserted a provisional memo with Cycle::initial", + ); + let value = self + .value + .take() + .expect("cycle state must contain the value from the latest execution"); + let value = C::recover_from_cycle( + self.db, + cycle, + last_provisional_value, + value, + C::id_to_input(zalsa, cycle.id), + ); + let converged = C::values_equal(&value, last_provisional_value); + self.value = Some(value); + converged + } + + fn provisional_memo( + &self, + zalsa: &'db Zalsa, + id: Id, + memo_ingredient_index: MemoIngredientIndex, + ) -> Option> { + self.memo_slot(zalsa, id, memo_ingredient_index) + .get_erased() + } + + fn insert_provisional_memo( + &mut self, + zalsa: &'db Zalsa, + id: Id, + revision: Revision, + revisions: QueryRevisions, + memo_ingredient_index: MemoIngredientIndex, + ) -> ErasedMemo<'db> { + let value = self + .value + .take() + .expect("cycle state must contain a value before memo insertion"); + self.ingredient + .insert_memo( + zalsa, + id, + Memo::new(Some(value), revision, revisions), + memo_ingredient_index, + ) + .erase() + } +} + +fn seed_query_from_old_memo( + zalsa: &Zalsa, + active_query: &ActiveQueryGuard<'_>, + old_memo: Option>, +) { + let Some(old_memo) = old_memo else { + return; + }; + + old_memo.header().seed_active_query(zalsa, active_query); +} + struct PreviousIteration { iteration: IterationStamp, reuse_as_provisional: bool, diff --git a/src/function/fetch.rs b/src/function/fetch.rs index 1f7eb5637..7c6fc0d6f 100644 --- a/src/function/fetch.rs +++ b/src/function/fetch.rs @@ -1,6 +1,7 @@ -use crate::cycle::{CycleRecoveryStrategy, IterationStamp}; +use crate::cycle::IterationStamp; +use crate::function::cycle_strategy::{CycleStrategy, FetchCycleContext}; use crate::function::eviction::EvictionPolicy; -use crate::function::execute::{QueryState, QueryStateImpl}; +use crate::function::execute::CycleState; use crate::function::memo::{ErasedMemo, Memo}; use crate::function::sync::ClaimResult; use crate::function::{Configuration, IngredientImpl, Reentrancy}; @@ -143,7 +144,7 @@ where && old_memo.header.verify_memo( db.into(), &claim_guard, - C::CYCLE_STRATEGY, + crate::function::cycle_strategy::recovery_strategy::(), #[cfg(feature = "detailed-trace")] true, ) @@ -158,7 +159,6 @@ where } #[cold] - #[inline(never)] fn fetch_cold_cycle<'db>( &'db self, db: &'db C::DbView, @@ -167,24 +167,23 @@ where database_key_index: DatabaseKeyIndex, memo_ingredient_index: MemoIngredientIndex, ) -> &'db Memo { - match C::CYCLE_STRATEGY { - CycleRecoveryStrategy::Panic => fetch_cold_cycle_panic(zalsa_local, database_key_index), - CycleRecoveryStrategy::FallbackImmediate | CycleRecoveryStrategy::Fixpoint => { - let mut state = QueryStateImpl::new(self, db); - let memo = fetch_cold_cycle_recoverable_erased( - &mut state, - zalsa, - database_key_index, - memo_ingredient_index, - ); - memo.downcast::() - } - } + >::fetch_cold_cycle(FetchCycleContext { + ingredient: self, + db, + zalsa, + zalsa_local, + database_key_index, + memo_ingredient_index, + }) + .0 } } #[cold] -fn fetch_cold_cycle_panic(zalsa_local: &ZalsaLocal, database_key_index: DatabaseKeyIndex) -> ! { +pub(super) fn fetch_cold_cycle_panic( + zalsa_local: &ZalsaLocal, + database_key_index: DatabaseKeyIndex, +) -> ! { // SAFETY: We do not access the query stack reentrantly. unsafe { zalsa_local.with_query_stack_unchecked(|stack| { @@ -198,8 +197,8 @@ fn fetch_cold_cycle_panic(zalsa_local: &ZalsaLocal, database_key_index: Database } #[cold] -fn fetch_cold_cycle_recoverable_erased<'db>( - state: &mut dyn QueryState<'db>, +pub(super) fn fetch_cold_cycle_recoverable_erased<'db>( + state: &mut dyn CycleState<'db>, zalsa: &'db Zalsa, database_key_index: DatabaseKeyIndex, memo_ingredient_index: MemoIngredientIndex, @@ -208,7 +207,7 @@ fn fetch_cold_cycle_recoverable_erased<'db>( let cancellation_count = zalsa.runtime().cancellation_count(); // Don't validate provisional memos here: an existing value should be reused. - let current_memo = state.get_memo(zalsa, id, memo_ingredient_index); + let current_memo = state.provisional_memo(zalsa, id, memo_ingredient_index); if let Some(memo) = current_memo { let header = memo.header(); @@ -262,7 +261,7 @@ fn fetch_cold_cycle_recoverable_erased<'db>( .unwrap_or_else(|| IterationStamp::initial(cancellation_count)); let revisions = QueryRevisions::fixpoint_initial(database_key_index, iteration); state.use_fallback(zalsa, id); - state.insert_memo( + state.insert_provisional_memo( zalsa, id, zalsa.current_revision(), diff --git a/src/function/maybe_changed_after.rs b/src/function/maybe_changed_after.rs index 592ea7fa2..2e29201f9 100644 --- a/src/function/maybe_changed_after.rs +++ b/src/function/maybe_changed_after.rs @@ -133,6 +133,7 @@ where } } + #[inline] fn maybe_changed_after_cold( &self, zalsa: &Zalsa, @@ -152,6 +153,7 @@ where } #[allow(clippy::too_many_arguments)] + #[inline(never)] fn inner<'db>( sync_table: &'db SyncTable, zalsa: &'db Zalsa, @@ -234,7 +236,7 @@ where memo_slot, database_key_index, revision, - C::CYCLE_STRATEGY, + crate::function::cycle_strategy::recovery_strategy::(), ) { ColdResult::Retry => None, ColdResult::Verified(result) => Some(result), diff --git a/src/function/memo.rs b/src/function/memo.rs index f8d353406..b670d2a20 100644 --- a/src/function/memo.rs +++ b/src/function/memo.rs @@ -10,7 +10,7 @@ use crate::function::{ClaimResult, Configuration, IngredientImpl, Reentrancy}; use crate::key::DatabaseKeyIndex; use crate::revision::AtomicRevision; use crate::sync::atomic::Ordering; -use crate::table::memo::{DummyMemo, MemoSlot, MemoTableWithTypesMut, ToDynMemo}; +use crate::table::memo::{DummyMemo, MemoEntryType, MemoSlot, MemoTableWithTypesMut, ToDynMemo}; use crate::zalsa::{MemoIngredientIndex, Zalsa}; use crate::zalsa_local::{QueryOriginRef, QueryRevisions}; use crate::{Cancelled, Event, EventKind, Id, Revision}; @@ -338,12 +338,28 @@ impl Memo { pub(super) fn value(&self) -> Option<&C::Output<'_>> { self.value.as_ref().map(|value| { - // SAFETY: Guaranteed by `Configuration`; the restored lifetime is - // bounded by the borrow of this memo. + // SAFETY: Guaranteed by Configuration; the restored lifetime is bounded by the + // borrow of this memo. unsafe { std::mem::transmute::<&C::Output<'static>, &C::Output<'_>>(value) } }) } + /// Returns a type-erased handle to this memo. + #[inline] + pub(super) fn erase(&self) -> ErasedMemo<'_> { + let data = NonNull::from(self).cast::(); + + // SAFETY: `data` retains the provenance of this complete memo allocation and remains + // valid for the lifetime of the shared borrow. Both metadata values describe `C`. + unsafe { + ErasedMemo::from_raw_parts( + data, + MemoEntryType::to_dyn_fn::>(), + TypeId::of::>(), + ) + } + } + /// Returns `true` if this memo should be serialized. pub(super) fn should_serialize(&self) -> bool { // TODO: Serialization is a good opportunity to prune old query results based on @@ -608,7 +624,6 @@ impl Iterator for TryClaimCycleHeadsIter<'_> { #[cfg(all(not(feature = "shuttle"), target_pointer_width = "64"))] mod _memory_usage { - use crate::cycle::CycleRecoveryStrategy; use crate::ingredient::Location; use crate::plumbing::{self, IngredientIndices, MemoIngredientSingletonIndex, SalsaStructInDb}; use crate::table::memo::MemoTableWithTypes; @@ -663,13 +678,13 @@ mod _memory_usage { const DEBUG_NAME: &'static str = ""; const LOCATION: Location = Location { file: "", line: 0 }; const PERSIST: bool = false; - const CYCLE_STRATEGY: CycleRecoveryStrategy = CycleRecoveryStrategy::Panic; type DbView = dyn Database; type SalsaStruct<'db> = DummyStruct; type Input<'db> = (); type Output<'db> = NonZeroUsize; type Eviction = crate::function::eviction::NoopEviction; + type CycleStrategy = crate::function::cycle_strategy::Panic; fn values_equal<'db>(_: &Self::Output<'db>, _: &Self::Output<'db>) -> bool { unimplemented!() diff --git a/src/lib.rs b/src/lib.rs index 3c5140698..0ab78c2c3 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -412,7 +412,7 @@ pub mod plumbing { } pub mod function { - pub use crate::function::{Configuration, IngredientImpl, Memo}; + pub use crate::function::{Configuration, IngredientImpl, Memo, cycle_strategy}; pub use crate::function::{EvictionPolicy, HasCapacity, Lru, NoopEviction}; pub use crate::table::memo::MemoEntryType; } From 590b3d532067652d648b176ac62fe31d0f83006b Mon Sep 17 00:00:00 2001 From: Micha Reiser Date: Sun, 28 Jun 2026 17:30:57 +0000 Subject: [PATCH 6/8] refactor: simplify cycle strategy dispatch --- src/function.rs | 5 +- src/function/cycle_strategy.rs | 79 ++---------- src/function/execute.rs | 189 ++++++++++++---------------- src/function/fetch.rs | 16 +-- src/function/maybe_changed_after.rs | 4 +- src/function/memo.rs | 1 - 6 files changed, 99 insertions(+), 195 deletions(-) diff --git a/src/function.rs b/src/function.rs index b752b481f..e094825da 100644 --- a/src/function.rs +++ b/src/function.rs @@ -8,6 +8,7 @@ use std::ptr::NonNull; use std::sync::OnceLock; use std::sync::atomic::Ordering; +use self::cycle_strategy::CycleStrategy as _; use crate::cycle::{CycleRecoveryStrategy, IterationStamp, ProvisionalStatus}; use crate::database::RawDatabase; use crate::function::delete::DeletedEntries; @@ -77,6 +78,8 @@ pub unsafe trait Configuration: Any + Sized { /// (and, if so, how). type CycleStrategy: cycle_strategy::CycleStrategy; + const CYCLE_RECOVERY_STRATEGY: CycleRecoveryStrategy = Self::CycleStrategy::RECOVERY_STRATEGY; + /// Invokes after a new result `new_value` has been computed for which an older memoized value /// existed `old_value`, or in fixpoint iteration. Returns true if the new value is equal to /// the older one. @@ -414,7 +417,7 @@ where self, zalsa, self.database_key_index(id), - cycle_strategy::recovery_strategy::(), + C::CYCLE_RECOVERY_STRATEGY, flattened_input_outputs, seen, ); diff --git a/src/function/cycle_strategy.rs b/src/function/cycle_strategy.rs index 7412aefec..77752d9e7 100644 --- a/src/function/cycle_strategy.rs +++ b/src/function/cycle_strategy.rs @@ -19,7 +19,7 @@ pub struct ExecuteContext<'db, C: Configuration> { pub(super) memo_ingredient_index: MemoIngredientIndex, } -pub struct ExecuteResult<'db, C: Configuration>(pub(super) Option<&'db Memo>); +pub type ExecuteResult<'db, C> = Option<&'db Memo>; pub struct FetchCycleContext<'db, C: Configuration> { pub(super) ingredient: &'db IngredientImpl, @@ -30,47 +30,23 @@ pub struct FetchCycleContext<'db, C: Configuration> { pub(super) memo_ingredient_index: MemoIngredientIndex, } -pub struct FetchCycleResult<'db, C: Configuration>(pub(super) &'db Memo); +pub type FetchCycleResult<'db, C> = &'db Memo; pub trait CycleStrategy: 'static { - const RECOVERY_STRATEGY: CycleRecoveryStrategy; + const RECOVERY_STRATEGY: CycleRecoveryStrategy = CycleRecoveryStrategy::Panic; fn execute<'db>(context: ExecuteContext<'db, C>) -> ExecuteResult<'db, C>; fn fetch_cold_cycle<'db>(context: FetchCycleContext<'db, C>) -> FetchCycleResult<'db, C>; } -#[inline] -pub(super) fn recovery_strategy() -> CycleRecoveryStrategy { - >::RECOVERY_STRATEGY -} - impl CycleStrategy for Panic { - const RECOVERY_STRATEGY: CycleRecoveryStrategy = CycleRecoveryStrategy::Panic; - fn execute<'db>(context: ExecuteContext<'db, C>) -> ExecuteResult<'db, C> { - let ExecuteContext { - ingredient, - db, - claim_guard, - opt_old_memo, - memo_ingredient_index, - } = context; - ExecuteResult(ingredient.execute_panic( - db, - claim_guard, - opt_old_memo, - memo_ingredient_index, - )) + IngredientImpl::execute_panic(context) } fn fetch_cold_cycle<'db>(context: FetchCycleContext<'db, C>) -> FetchCycleResult<'db, C> { - let FetchCycleContext { - zalsa_local, - database_key_index, - .. - } = context; - fetch_cold_cycle_panic(zalsa_local, database_key_index) + fetch_cold_cycle_panic(context.zalsa_local, context.database_key_index) } } @@ -78,7 +54,7 @@ impl CycleStrategy for FallbackImmediate { const RECOVERY_STRATEGY: CycleRecoveryStrategy = CycleRecoveryStrategy::FallbackImmediate; fn execute<'db>(context: ExecuteContext<'db, C>) -> ExecuteResult<'db, C> { - execute_recoverable(context, CyclePolicy::FallbackImmediate) + IngredientImpl::execute_cycle(context, CyclePolicy::FallbackImmediate) } fn fetch_cold_cycle<'db>(context: FetchCycleContext<'db, C>) -> FetchCycleResult<'db, C> { @@ -90,7 +66,7 @@ impl CycleStrategy for Fixpoint { const RECOVERY_STRATEGY: CycleRecoveryStrategy = CycleRecoveryStrategy::Fixpoint; fn execute<'db>(context: ExecuteContext<'db, C>) -> ExecuteResult<'db, C> { - execute_recoverable(context, CyclePolicy::Fixpoint) + IngredientImpl::execute_cycle(context, CyclePolicy::Fixpoint) } fn fetch_cold_cycle<'db>(context: FetchCycleContext<'db, C>) -> FetchCycleResult<'db, C> { @@ -98,43 +74,14 @@ impl CycleStrategy for Fixpoint { } } -fn execute_recoverable<'db, C: Configuration>( - context: ExecuteContext<'db, C>, - policy: CyclePolicy, -) -> ExecuteResult<'db, C> { - let ExecuteContext { - ingredient, - db, - claim_guard, - opt_old_memo, - memo_ingredient_index, - } = context; - ExecuteResult(ingredient.execute_cycle( - db, - claim_guard, - opt_old_memo, - memo_ingredient_index, - policy, - )) -} - fn fetch_cold_cycle_recoverable<'db, C: Configuration>( context: FetchCycleContext<'db, C>, ) -> FetchCycleResult<'db, C> { - let FetchCycleContext { - ingredient, - db, - zalsa, - database_key_index, - memo_ingredient_index, - .. - } = context; - let mut state = CycleStateImpl::new(ingredient, db); - let memo = fetch_cold_cycle_recoverable_erased( - &mut state, - zalsa, - database_key_index, - memo_ingredient_index, + let mut state = CycleStateImpl::new( + context.ingredient, + context.db, + context.memo_ingredient_index, ); - FetchCycleResult(memo.downcast::()) + fetch_cold_cycle_recoverable_erased(&mut state, context.zalsa, context.database_key_index) + .downcast::() } diff --git a/src/function/execute.rs b/src/function/execute.rs index f46829285..0df0b0f20 100644 --- a/src/function/execute.rs +++ b/src/function/execute.rs @@ -10,14 +10,16 @@ use crate::hash::{FxHashSet, FxIndexSet}; use crate::plumbing::ZalsaLocal; use crate::sync::thread; use crate::table::memo::MemoSlot; -use crate::tracked_struct::Identity; use crate::zalsa::{MemoIngredientIndex, Zalsa}; use crate::zalsa_local::{ActiveQueryGuard, QueryEdge, QueryEdgeKind, QueryRevisions}; use crate::{Cancelled, Cycle, Revision, tracing}; use crate::{DatabaseKeyIndex, Event, EventKind, Id}; impl IngredientImpl { - /// Executes this query through the shared query lifecycle and restores its typed memo. + /// Executes the query function and stores a new memo with the result, backdated if possible. + /// + /// Returns `None` if this query is part of a larger cycle and blocked on a cycle head running + /// on another thread. The caller must refetch the potentially outdated memo in that case. pub(super) fn execute<'db>( &'db self, db: &'db C::DbView, @@ -34,43 +36,78 @@ impl IngredientImpl { opt_old_memo, memo_ingredient_index, }) - .0 + } + + pub(super) fn execute_panic<'db>(context: ExecuteContext<'db, C>) -> Option<&'db Memo> { + let ExecuteContext { + ingredient, + db, + claim_guard, + opt_old_memo, + memo_ingredient_index, + } = context; + let database_key_index = claim_guard.database_key_index(); + let zalsa = claim_guard.zalsa(); + let id = database_key_index.key_index(); + + let active_query = claim_guard.zalsa_local().push_query(database_key_index); + if let Some(old_memo) = opt_old_memo { + old_memo.header.seed_active_query(zalsa, &active_query); + } + let new_value = C::execute(db, C::id_to_input(zalsa, id)); + + // Ordinary queries don't need a cycle iteration stamp. Keeping the default avoids + // allocating `QueryRevisionsExtra` after a revision-preserving cancellation. + let completed_query = active_query.pop(IterationStamp::default()); + + let memo = ingredient.finish_memo( + zalsa, + database_key_index, + opt_old_memo, + new_value, + completed_query, + memo_ingredient_index, + ); + + if claim_guard.drop() { None } else { Some(memo) } } pub(super) fn execute_cycle<'db>( - &'db self, - db: &'db C::DbView, - mut claim_guard: ClaimGuard<'db>, - opt_old_memo: Option<&'db Memo>, - memo_ingredient_index: MemoIngredientIndex, + context: ExecuteContext<'db, C>, policy: CyclePolicy, ) -> Option<&'db Memo> { + let ExecuteContext { + ingredient, + db, + mut claim_guard, + opt_old_memo, + memo_ingredient_index, + } = context; let database_key_index = claim_guard.database_key_index(); let zalsa = claim_guard.zalsa(); let id = database_key_index.key_index(); let _cancellation_guard = DisableLocalCancellationGuard::new(claim_guard.zalsa_local()); let _poison_guard = PoisonProvisionalIfPanicking { - ingredient: self, + ingredient, zalsa, id, memo_ingredient_index, }; let opt_old_memo_erased = opt_old_memo.map(Memo::erase); - let mut state = CycleStateImpl::new(self, db); + let mut state = CycleStateImpl::new(ingredient, db, memo_ingredient_index); let completed_query = execute_maybe_iterate_erased( &mut state, zalsa, opt_old_memo_erased, &mut claim_guard, - memo_ingredient_index, policy, ); let value = state .value .take() .expect("query execution must produce a value"); - let memo = self.finish_memo( + let memo = ingredient.finish_memo( zalsa, database_key_index, opt_old_memo, @@ -82,36 +119,6 @@ impl IngredientImpl { if claim_guard.drop() { None } else { Some(memo) } } - pub(super) fn execute_panic<'db>( - &'db self, - db: &'db C::DbView, - claim_guard: ClaimGuard<'db>, - opt_old_memo: Option<&'db Memo>, - memo_ingredient_index: MemoIngredientIndex, - ) -> Option<&'db Memo> { - let database_key_index = claim_guard.database_key_index(); - let zalsa = claim_guard.zalsa(); - let id = database_key_index.key_index(); - - let active_query = claim_guard.zalsa_local().push_query(database_key_index); - if let Some(old_memo) = opt_old_memo { - old_memo.header.seed_active_query(zalsa, &active_query); - } - let new_value = C::execute(db, C::id_to_input(zalsa, id)); - let completed_query = active_query.pop(IterationStamp::default()); - - let memo = self.finish_memo( - zalsa, - database_key_index, - opt_old_memo, - new_value, - completed_query, - memo_ingredient_index, - ); - - if claim_guard.drop() { None } else { Some(memo) } - } - fn finish_memo<'db>( &'db self, zalsa: &'db Zalsa, @@ -172,20 +179,18 @@ fn execute_maybe_iterate_erased<'db>( zalsa: &'db Zalsa, opt_old_memo: Option>, claim_guard: &mut ClaimGuard<'db>, - memo_ingredient_index: MemoIngredientIndex, policy: CyclePolicy, ) -> CompletedQuery { claim_guard.set_release_mode(ReleaseMode::Default); let database_key_index = claim_guard.database_key_index(); - let id = database_key_index.key_index(); // Our provisional value from the previous iteration, when doing fixpoint iteration. // This is different from `opt_old_memo` which might be from a different revision. let mut last_provisional_memo_opt = None; - let mut last_stale_tracked_ids: Vec<(Identity, Id)> = Vec::new(); + let mut last_stale_tracked_ids = Vec::new(); let current_revision = zalsa.current_revision(); let cancellation_count = zalsa.runtime().cancellation_count(); let mut opt_old_memo = opt_old_memo; @@ -225,11 +230,9 @@ fn execute_maybe_iterate_erased<'db>( // if they aren't recreated when reaching the final iteration. active_query.seed_tracked_struct_ids(&last_stale_tracked_ids); - seed_query_from_old_memo( - zalsa, - &active_query, - last_provisional_memo_opt.or(opt_old_memo), - ); + if let Some(old_memo) = last_provisional_memo_opt.or(opt_old_memo) { + old_memo.header().seed_active_query(zalsa, &active_query); + } state.execute_query(zalsa, id); let (mut active_query, cycle_heads, outer_cycle, cycle_iteration) = @@ -264,22 +267,14 @@ fn execute_maybe_iterate_erased<'db>( // This is our first time around the loop; a provisional value must have been // inserted into the memo table when the cycle was hit, so let's pull our // initial provisional value from there. - let memo = state - .provisional_memo(zalsa, id, memo_ingredient_index) - .unwrap_or_else(|| { - unreachable!( - "{database_key_index:#?} is a cycle head, \ + let memo = state.provisional_memo(zalsa, id).unwrap_or_else(|| { + unreachable!( + "{database_key_index:#?} is a cycle head, \ but no provisional memo found" - ) - }); - - debug_assert!( - !memo - .header() - .revisions - .verified_final - .load(std::sync::atomic::Ordering::Relaxed) - ); + ) + }); + + debug_assert!(memo.header().may_be_provisional()); memo }); crate::tracing::debug!( @@ -317,13 +312,8 @@ fn execute_maybe_iterate_erased<'db>( } }; - let new_memo = state.insert_provisional_memo( - zalsa, - id, - current_revision, - completed_query.revisions, - memo_ingredient_index, - ); + let new_memo = + state.insert_provisional_memo(zalsa, id, current_revision, completed_query.revisions); last_provisional_memo_opt = Some(new_memo); @@ -347,8 +337,9 @@ pub(super) enum CyclePolicy { impl CyclePolicy { /// Adjusts the query value before completing a non-head cycle participant. /// - /// Fallback recovery replaces the computed value with the fallback. Fixpoint recovery keeps - /// the computed value unchanged. + /// Fallback recovery replaces the computed value with the fallback so that results don't + /// depend on query call order. Fixpoint recovery keeps the computed value unchanged. See + /// . fn complete_participant<'db>(self, state: &mut dyn CycleState<'db>, zalsa: &'db Zalsa, id: Id) { if matches!(self, Self::FallbackImmediate) { state.use_fallback(zalsa, id); @@ -393,12 +384,7 @@ pub(super) trait CycleState<'db> { last_provisional_memo: ErasedMemo<'db>, ) -> bool; - fn provisional_memo( - &self, - zalsa: &'db Zalsa, - id: Id, - memo_ingredient_index: MemoIngredientIndex, - ) -> Option>; + fn provisional_memo(&self, zalsa: &'db Zalsa, id: Id) -> Option>; fn insert_provisional_memo( &mut self, @@ -406,38 +392,38 @@ pub(super) trait CycleState<'db> { id: Id, revision: Revision, revisions: QueryRevisions, - memo_ingredient_index: MemoIngredientIndex, ) -> ErasedMemo<'db>; } pub(super) struct CycleStateImpl<'db, C: Configuration> { ingredient: &'db IngredientImpl, db: &'db C::DbView, + memo_ingredient_index: MemoIngredientIndex, value: Option>, } impl<'db, C: Configuration> CycleStateImpl<'db, C> { - pub(super) fn new(ingredient: &'db IngredientImpl, db: &'db C::DbView) -> Self { + pub(super) fn new( + ingredient: &'db IngredientImpl, + db: &'db C::DbView, + memo_ingredient_index: MemoIngredientIndex, + ) -> Self { Self { ingredient, db, + memo_ingredient_index, value: None, } } - fn memo_slot( - &self, - zalsa: &'db Zalsa, - id: Id, - memo_ingredient_index: MemoIngredientIndex, - ) -> MemoSlot<'db> { + fn memo_slot(&self, zalsa: &'db Zalsa, id: Id) -> MemoSlot<'db> { // SAFETY: Replaced memo allocations remain in `deleted_entries` until the next revision. // The database is borrowed for `'db`, so a new revision cannot begin while this state can // still observe an allocation. unsafe { MemoSlot::new( zalsa.memo_table_for::>(id), - memo_ingredient_index, + self.memo_ingredient_index, ) } } @@ -478,14 +464,8 @@ impl<'db, C: Configuration> CycleState<'db> for CycleStateImpl<'db, C> { converged } - fn provisional_memo( - &self, - zalsa: &'db Zalsa, - id: Id, - memo_ingredient_index: MemoIngredientIndex, - ) -> Option> { - self.memo_slot(zalsa, id, memo_ingredient_index) - .get_erased() + fn provisional_memo(&self, zalsa: &'db Zalsa, id: Id) -> Option> { + self.memo_slot(zalsa, id).get_erased() } fn insert_provisional_memo( @@ -494,7 +474,6 @@ impl<'db, C: Configuration> CycleState<'db> for CycleStateImpl<'db, C> { id: Id, revision: Revision, revisions: QueryRevisions, - memo_ingredient_index: MemoIngredientIndex, ) -> ErasedMemo<'db> { let value = self .value @@ -505,24 +484,12 @@ impl<'db, C: Configuration> CycleState<'db> for CycleStateImpl<'db, C> { zalsa, id, Memo::new(Some(value), revision, revisions), - memo_ingredient_index, + self.memo_ingredient_index, ) .erase() } } -fn seed_query_from_old_memo( - zalsa: &Zalsa, - active_query: &ActiveQueryGuard<'_>, - old_memo: Option>, -) { - let Some(old_memo) = old_memo else { - return; - }; - - old_memo.header().seed_active_query(zalsa, active_query); -} - struct PreviousIteration { iteration: IterationStamp, reuse_as_provisional: bool, diff --git a/src/function/fetch.rs b/src/function/fetch.rs index 7c6fc0d6f..a38cbd41b 100644 --- a/src/function/fetch.rs +++ b/src/function/fetch.rs @@ -144,7 +144,7 @@ where && old_memo.header.verify_memo( db.into(), &claim_guard, - crate::function::cycle_strategy::recovery_strategy::(), + C::CYCLE_RECOVERY_STRATEGY, #[cfg(feature = "detailed-trace")] true, ) @@ -175,11 +175,9 @@ where database_key_index, memo_ingredient_index, }) - .0 } } -#[cold] pub(super) fn fetch_cold_cycle_panic( zalsa_local: &ZalsaLocal, database_key_index: DatabaseKeyIndex, @@ -196,18 +194,16 @@ pub(super) fn fetch_cold_cycle_panic( } } -#[cold] pub(super) fn fetch_cold_cycle_recoverable_erased<'db>( state: &mut dyn CycleState<'db>, zalsa: &'db Zalsa, database_key_index: DatabaseKeyIndex, - memo_ingredient_index: MemoIngredientIndex, ) -> ErasedMemo<'db> { let id = database_key_index.key_index(); let cancellation_count = zalsa.runtime().cancellation_count(); // Don't validate provisional memos here: an existing value should be reused. - let current_memo = state.provisional_memo(zalsa, id, memo_ingredient_index); + let current_memo = state.provisional_memo(zalsa, id); if let Some(memo) = current_memo { let header = memo.header(); @@ -261,11 +257,5 @@ pub(super) fn fetch_cold_cycle_recoverable_erased<'db>( .unwrap_or_else(|| IterationStamp::initial(cancellation_count)); let revisions = QueryRevisions::fixpoint_initial(database_key_index, iteration); state.use_fallback(zalsa, id); - state.insert_provisional_memo( - zalsa, - id, - zalsa.current_revision(), - revisions, - memo_ingredient_index, - ) + state.insert_provisional_memo(zalsa, id, zalsa.current_revision(), revisions) } diff --git a/src/function/maybe_changed_after.rs b/src/function/maybe_changed_after.rs index 2e29201f9..703641f5a 100644 --- a/src/function/maybe_changed_after.rs +++ b/src/function/maybe_changed_after.rs @@ -133,7 +133,6 @@ where } } - #[inline] fn maybe_changed_after_cold( &self, zalsa: &Zalsa, @@ -153,7 +152,6 @@ where } #[allow(clippy::too_many_arguments)] - #[inline(never)] fn inner<'db>( sync_table: &'db SyncTable, zalsa: &'db Zalsa, @@ -236,7 +234,7 @@ where memo_slot, database_key_index, revision, - crate::function::cycle_strategy::recovery_strategy::(), + C::CYCLE_RECOVERY_STRATEGY, ) { ColdResult::Retry => None, ColdResult::Verified(result) => Some(result), diff --git a/src/function/memo.rs b/src/function/memo.rs index b670d2a20..c6f44887b 100644 --- a/src/function/memo.rs +++ b/src/function/memo.rs @@ -345,7 +345,6 @@ impl Memo { } /// Returns a type-erased handle to this memo. - #[inline] pub(super) fn erase(&self) -> ErasedMemo<'_> { let data = NonNull::from(self).cast::(); From 109f5ca812bd6b9e142c527c7fc8772b728274e6 Mon Sep 17 00:00:00 2001 From: Micha Reiser Date: Sun, 28 Jun 2026 18:17:40 +0000 Subject: [PATCH 7/8] chore: satisfy clippy on Rust 1.85 --- src/function/cycle_strategy.rs | 22 +++++++++++----------- src/function/execute.rs | 12 +++++++----- 2 files changed, 18 insertions(+), 16 deletions(-) diff --git a/src/function/cycle_strategy.rs b/src/function/cycle_strategy.rs index 77752d9e7..3db01a762 100644 --- a/src/function/cycle_strategy.rs +++ b/src/function/cycle_strategy.rs @@ -35,17 +35,17 @@ pub type FetchCycleResult<'db, C> = &'db Memo; pub trait CycleStrategy: 'static { const RECOVERY_STRATEGY: CycleRecoveryStrategy = CycleRecoveryStrategy::Panic; - fn execute<'db>(context: ExecuteContext<'db, C>) -> ExecuteResult<'db, C>; + fn execute(context: ExecuteContext<'_, C>) -> ExecuteResult<'_, C>; - fn fetch_cold_cycle<'db>(context: FetchCycleContext<'db, C>) -> FetchCycleResult<'db, C>; + fn fetch_cold_cycle(context: FetchCycleContext<'_, C>) -> FetchCycleResult<'_, C>; } impl CycleStrategy for Panic { - fn execute<'db>(context: ExecuteContext<'db, C>) -> ExecuteResult<'db, C> { + fn execute(context: ExecuteContext<'_, C>) -> ExecuteResult<'_, C> { IngredientImpl::execute_panic(context) } - fn fetch_cold_cycle<'db>(context: FetchCycleContext<'db, C>) -> FetchCycleResult<'db, C> { + fn fetch_cold_cycle(context: FetchCycleContext<'_, C>) -> FetchCycleResult<'_, C> { fetch_cold_cycle_panic(context.zalsa_local, context.database_key_index) } } @@ -53,11 +53,11 @@ impl CycleStrategy for Panic { impl CycleStrategy for FallbackImmediate { const RECOVERY_STRATEGY: CycleRecoveryStrategy = CycleRecoveryStrategy::FallbackImmediate; - fn execute<'db>(context: ExecuteContext<'db, C>) -> ExecuteResult<'db, C> { + fn execute(context: ExecuteContext<'_, C>) -> ExecuteResult<'_, C> { IngredientImpl::execute_cycle(context, CyclePolicy::FallbackImmediate) } - fn fetch_cold_cycle<'db>(context: FetchCycleContext<'db, C>) -> FetchCycleResult<'db, C> { + fn fetch_cold_cycle(context: FetchCycleContext<'_, C>) -> FetchCycleResult<'_, C> { fetch_cold_cycle_recoverable(context) } } @@ -65,18 +65,18 @@ impl CycleStrategy for FallbackImmediate { impl CycleStrategy for Fixpoint { const RECOVERY_STRATEGY: CycleRecoveryStrategy = CycleRecoveryStrategy::Fixpoint; - fn execute<'db>(context: ExecuteContext<'db, C>) -> ExecuteResult<'db, C> { + fn execute(context: ExecuteContext<'_, C>) -> ExecuteResult<'_, C> { IngredientImpl::execute_cycle(context, CyclePolicy::Fixpoint) } - fn fetch_cold_cycle<'db>(context: FetchCycleContext<'db, C>) -> FetchCycleResult<'db, C> { + fn fetch_cold_cycle(context: FetchCycleContext<'_, C>) -> FetchCycleResult<'_, C> { fetch_cold_cycle_recoverable(context) } } -fn fetch_cold_cycle_recoverable<'db, C: Configuration>( - context: FetchCycleContext<'db, C>, -) -> FetchCycleResult<'db, C> { +fn fetch_cold_cycle_recoverable( + context: FetchCycleContext<'_, C>, +) -> FetchCycleResult<'_, C> { let mut state = CycleStateImpl::new( context.ingredient, context.db, diff --git a/src/function/execute.rs b/src/function/execute.rs index 0df0b0f20..b86c38ad6 100644 --- a/src/function/execute.rs +++ b/src/function/execute.rs @@ -38,7 +38,7 @@ impl IngredientImpl { }) } - pub(super) fn execute_panic<'db>(context: ExecuteContext<'db, C>) -> Option<&'db Memo> { + pub(super) fn execute_panic(context: ExecuteContext<'_, C>) -> Option<&Memo> { let ExecuteContext { ingredient, db, @@ -72,10 +72,10 @@ impl IngredientImpl { if claim_guard.drop() { None } else { Some(memo) } } - pub(super) fn execute_cycle<'db>( - context: ExecuteContext<'db, C>, + pub(super) fn execute_cycle( + context: ExecuteContext<'_, C>, policy: CyclePolicy, - ) -> Option<&'db Memo> { + ) -> Option<&Memo> { let ExecuteContext { ingredient, db, @@ -899,7 +899,9 @@ fn complete_cycle_participant( /// /// Returns `Ok` if the cycle head has converged or if it is part of an outer cycle. /// Returns `Err` if the cycle head needs to keep iterating. -#[allow(clippy::too_many_arguments)] +// Both variants carry `CompletedQuery`; boxing only the error would add an allocation without +// reducing the caller's stack requirements. +#[allow(clippy::result_large_err, clippy::too_many_arguments)] fn try_complete_cycle_head( active_query: ActiveQueryGuard, claim_guard: &mut ClaimGuard, From 34bf6e25c0890cafecef7834251d97dfe200a0c2 Mon Sep 17 00:00:00 2001 From: Micha Reiser Date: Sun, 28 Jun 2026 18:30:28 +0000 Subject: [PATCH 8/8] refactor: reuse function ingredient for cycle memos --- src/function/cycle_strategy.rs | 9 +++++++-- src/function/execute.rs | 27 ++++++--------------------- src/function/fetch.rs | 5 +++-- 3 files changed, 16 insertions(+), 25 deletions(-) diff --git a/src/function/cycle_strategy.rs b/src/function/cycle_strategy.rs index 3db01a762..d54633430 100644 --- a/src/function/cycle_strategy.rs +++ b/src/function/cycle_strategy.rs @@ -82,6 +82,11 @@ fn fetch_cold_cycle_recoverable( context.db, context.memo_ingredient_index, ); - fetch_cold_cycle_recoverable_erased(&mut state, context.zalsa, context.database_key_index) - .downcast::() + fetch_cold_cycle_recoverable_erased( + &mut state, + context.ingredient, + context.zalsa, + context.database_key_index, + ) + .downcast::() } diff --git a/src/function/execute.rs b/src/function/execute.rs index b86c38ad6..f7dd8eae8 100644 --- a/src/function/execute.rs +++ b/src/function/execute.rs @@ -5,11 +5,12 @@ use crate::cycle::{CycleHeads, IterationStamp, ProvisionalStatus}; use crate::function::cycle_strategy::{CycleStrategy, ExecuteContext}; use crate::function::memo::{ErasedMemo, Memo, MemoHeader}; use crate::function::sync::ReleaseMode; -use crate::function::{ClaimGuard, ClaimResult, Configuration, IngredientImpl, Reentrancy}; +use crate::function::{ + ClaimGuard, ClaimResult, Configuration, FunctionIngredient, IngredientImpl, Reentrancy, +}; use crate::hash::{FxHashSet, FxIndexSet}; use crate::plumbing::ZalsaLocal; use crate::sync::thread; -use crate::table::memo::MemoSlot; use crate::zalsa::{MemoIngredientIndex, Zalsa}; use crate::zalsa_local::{ActiveQueryGuard, QueryEdge, QueryEdgeKind, QueryRevisions}; use crate::{Cancelled, Cycle, Revision, tracing}; @@ -98,6 +99,7 @@ impl IngredientImpl { let mut state = CycleStateImpl::new(ingredient, db, memo_ingredient_index); let completed_query = execute_maybe_iterate_erased( &mut state, + ingredient, zalsa, opt_old_memo_erased, &mut claim_guard, @@ -176,6 +178,7 @@ fn report_will_execute(claim_guard: &ClaimGuard<'_>) { fn execute_maybe_iterate_erased<'db>( state: &mut dyn CycleState<'db>, + ingredient: &'db dyn FunctionIngredient, zalsa: &'db Zalsa, opt_old_memo: Option>, claim_guard: &mut ClaimGuard<'db>, @@ -267,7 +270,7 @@ fn execute_maybe_iterate_erased<'db>( // This is our first time around the loop; a provisional value must have been // inserted into the memo table when the cycle was hit, so let's pull our // initial provisional value from there. - let memo = state.provisional_memo(zalsa, id).unwrap_or_else(|| { + let memo = ingredient.memo(zalsa, id).unwrap_or_else(|| { unreachable!( "{database_key_index:#?} is a cycle head, \ but no provisional memo found" @@ -384,8 +387,6 @@ pub(super) trait CycleState<'db> { last_provisional_memo: ErasedMemo<'db>, ) -> bool; - fn provisional_memo(&self, zalsa: &'db Zalsa, id: Id) -> Option>; - fn insert_provisional_memo( &mut self, zalsa: &'db Zalsa, @@ -415,18 +416,6 @@ impl<'db, C: Configuration> CycleStateImpl<'db, C> { value: None, } } - - fn memo_slot(&self, zalsa: &'db Zalsa, id: Id) -> MemoSlot<'db> { - // SAFETY: Replaced memo allocations remain in `deleted_entries` until the next revision. - // The database is borrowed for `'db`, so a new revision cannot begin while this state can - // still observe an allocation. - unsafe { - MemoSlot::new( - zalsa.memo_table_for::>(id), - self.memo_ingredient_index, - ) - } - } } impl<'db, C: Configuration> CycleState<'db> for CycleStateImpl<'db, C> { @@ -464,10 +453,6 @@ impl<'db, C: Configuration> CycleState<'db> for CycleStateImpl<'db, C> { converged } - fn provisional_memo(&self, zalsa: &'db Zalsa, id: Id) -> Option> { - self.memo_slot(zalsa, id).get_erased() - } - fn insert_provisional_memo( &mut self, zalsa: &'db Zalsa, diff --git a/src/function/fetch.rs b/src/function/fetch.rs index a38cbd41b..a2146f0ef 100644 --- a/src/function/fetch.rs +++ b/src/function/fetch.rs @@ -4,7 +4,7 @@ use crate::function::eviction::EvictionPolicy; use crate::function::execute::CycleState; use crate::function::memo::{ErasedMemo, Memo}; use crate::function::sync::ClaimResult; -use crate::function::{Configuration, IngredientImpl, Reentrancy}; +use crate::function::{Configuration, FunctionIngredient, IngredientImpl, Reentrancy}; use crate::zalsa::{MemoIngredientIndex, Zalsa}; use crate::zalsa_local::{QueryRevisions, ZalsaLocal}; use crate::{Cancelled, DatabaseKeyIndex, Id}; @@ -196,6 +196,7 @@ pub(super) fn fetch_cold_cycle_panic( pub(super) fn fetch_cold_cycle_recoverable_erased<'db>( state: &mut dyn CycleState<'db>, + ingredient: &'db dyn FunctionIngredient, zalsa: &'db Zalsa, database_key_index: DatabaseKeyIndex, ) -> ErasedMemo<'db> { @@ -203,7 +204,7 @@ pub(super) fn fetch_cold_cycle_recoverable_erased<'db>( let cancellation_count = zalsa.runtime().cancellation_count(); // Don't validate provisional memos here: an existing value should be reused. - let current_memo = state.provisional_memo(zalsa, id); + let current_memo = ingredient.memo(zalsa, id); if let Some(memo) = current_memo { let header = memo.header();