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 996401b00..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; @@ -27,6 +28,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 +52,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 +76,9 @@ 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; + + 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 @@ -179,38 +184,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 +366,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 @@ -461,7 +417,7 @@ where self, zalsa, self.database_key_index(id), - C::CYCLE_STRATEGY, + C::CYCLE_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..d54633430 --- /dev/null +++ b/src/function/cycle_strategy.rs @@ -0,0 +1,92 @@ +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 type ExecuteResult<'db, C> = 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 type FetchCycleResult<'db, C> = &'db Memo; + +pub trait CycleStrategy: 'static { + const RECOVERY_STRATEGY: CycleRecoveryStrategy = CycleRecoveryStrategy::Panic; + + fn execute(context: ExecuteContext<'_, C>) -> ExecuteResult<'_, C>; + + fn fetch_cold_cycle(context: FetchCycleContext<'_, C>) -> FetchCycleResult<'_, C>; +} + +impl CycleStrategy for Panic { + fn execute(context: ExecuteContext<'_, C>) -> ExecuteResult<'_, C> { + IngredientImpl::execute_panic(context) + } + + fn fetch_cold_cycle(context: FetchCycleContext<'_, C>) -> FetchCycleResult<'_, C> { + fetch_cold_cycle_panic(context.zalsa_local, context.database_key_index) + } +} + +impl CycleStrategy for FallbackImmediate { + const RECOVERY_STRATEGY: CycleRecoveryStrategy = CycleRecoveryStrategy::FallbackImmediate; + + fn execute(context: ExecuteContext<'_, C>) -> ExecuteResult<'_, C> { + IngredientImpl::execute_cycle(context, CyclePolicy::FallbackImmediate) + } + + fn fetch_cold_cycle(context: FetchCycleContext<'_, C>) -> FetchCycleResult<'_, C> { + fetch_cold_cycle_recoverable(context) + } +} + +impl CycleStrategy for Fixpoint { + const RECOVERY_STRATEGY: CycleRecoveryStrategy = CycleRecoveryStrategy::Fixpoint; + + fn execute(context: ExecuteContext<'_, C>) -> ExecuteResult<'_, C> { + IngredientImpl::execute_cycle(context, CyclePolicy::Fixpoint) + } + + fn fetch_cold_cycle(context: FetchCycleContext<'_, C>) -> FetchCycleResult<'_, C> { + fetch_cold_cycle_recoverable(context) + } +} + +fn fetch_cold_cycle_recoverable( + context: FetchCycleContext<'_, C>, +) -> FetchCycleResult<'_, C> { + let mut state = CycleStateImpl::new( + context.ingredient, + context.db, + context.memo_ingredient_index, + ); + 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 cea05487e..f7dd8eae8 100644 --- a/src/function/execute.rs +++ b/src/function/execute.rs @@ -1,84 +1,135 @@ use smallvec::SmallVec; use crate::active_query::CompletedQuery; -use crate::cycle::{CycleHeads, CycleRecoveryStrategy, IterationStamp, ProvisionalStatus}; -use crate::function::memo::{Memo, MemoHeader}; +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::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. +impl IngredientImpl { + /// Executes the query function and stores a new memo with the result, backdated if possible. /// - /// # 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)] + /// 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, - mut claim_guard: ClaimGuard<'db>, + claim_guard: ClaimGuard<'db>, opt_old_memo: Option<&'db Memo>, + memo_ingredient_index: MemoIngredientIndex, ) -> Option<&'db Memo> { + report_will_execute(&claim_guard); + + >::execute(ExecuteContext { + ingredient: self, + db, + claim_guard, + opt_old_memo, + memo_ingredient_index, + }) + } + + pub(super) fn execute_panic(context: ExecuteContext<'_, C>) -> Option<&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 memo_ingredient_index = self.memo_ingredient_index(zalsa, id); - crate::tracing::info!("{:?}: executing query", 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)); - zalsa.event(&|| { - Event::new(EventKind::WillExecute { - database_key: database_key_index, - }) - }); + // 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 (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), - ); + let memo = ingredient.finish_memo( + zalsa, + database_key_index, + opt_old_memo, + new_value, + completed_query, + memo_ingredient_index, + ); - // 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, - ) - } + if claim_guard.drop() { None } else { Some(memo) } + } + + pub(super) fn execute_cycle( + context: ExecuteContext<'_, C>, + policy: CyclePolicy, + ) -> Option<&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, + zalsa, + id, + memo_ingredient_index, }; + let opt_old_memo_erased = opt_old_memo.map(Memo::erase); + 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, + policy, + ); + let value = state + .value + .take() + .expect("query execution must produce a value"); + let memo = ingredient.finish_memo( + zalsa, + database_key_index, + opt_old_memo, + 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 { // 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 @@ -88,7 +139,7 @@ where 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 @@ -101,243 +152,326 @@ where #[cfg(not(feature = "persistence"))] completed_query.revisions.discard_edges_if_never_change(); - let memo = self.insert_memo( + self.insert_memo( zalsa, - id, + database_key_index.key_index(), Memo::new( - Some(new_value), + Some(value), zalsa.current_revision(), completed_query.revisions, ), memo_ingredient_index, - ); - - if claim_guard.drop() { None } else { Some(memo) } + ) } +} - 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); - - let database_key_index = claim_guard.database_key_index(); - let zalsa = claim_guard.zalsa(); - - 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: Option<&Memo> = None; +fn report_will_execute(claim_guard: &ClaimGuard<'_>) { + let database_key_index = claim_guard.database_key_index(); - 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); + crate::tracing::info!("{:?}: executing query", database_key_index); + claim_guard.zalsa().event(&|| { + Event::new(EventKind::WillExecute { + database_key: database_key_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; +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>, + 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::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); } - None => opt_old_memo = None, + + iteration = previous_iteration.iteration; } + None => opt_old_memo = None, } } + } - let _poison_guard = - PoisonProvisionalIfPanicking::new(self, zalsa, id, memo_ingredient_index); - - let (new_value, completed_query) = loop { - let active_query = claim_guard.zalsa_local().push_query(database_key_index); + let completed_query = loop { + let active_query = claim_guard.zalsa_local().push_query(database_key_index); - // 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); + // 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); - 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), - ); + if let Some(old_memo) = last_provisional_memo_opt.or(opt_old_memo) { + old_memo.header().seed_active_query(zalsa, &active_query); + } - 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 { + 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, + } => { + policy.complete_participant(state, zalsa, id); + + break complete_cycle_participant( active_query, + claim_guard, 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); - } - 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" - ) - }); + iteration, + ); + } + QueryExecutionOutcome::CycleHead { + active_query, + cycle_heads, + outer_cycle, + cycle_iteration, + } => (active_query, cycle_heads, outer_cycle, cycle_iteration), + }; - debug_assert!(memo.header.may_be_provisional()); - memo + // 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 = ingredient.memo(zalsa, id).unwrap_or_else(|| { + unreachable!( + "{database_key_index:#?} is a cycle head, \ + but no provisional memo found" + ) }); - let last_provisional_value = last_provisional_memo.value(); - - 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: \ + debug_assert!(memo.header().may_be_provisional()); + memo + }); + 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 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), - ); + 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); - C::values_equal(&new_value, last_provisional_value) - }; + let new_cycle_heads = active_query.take_cycle_heads(); + assert_no_new_cycle_heads(&cycle_heads, new_cycle_heads, database_key_index); - 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 - } - }; + 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 + } + }; - let new_memo = self.insert_memo( - zalsa, - id, - Memo::new(Some(new_value), 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); + last_provisional_memo_opt = Some(new_memo); - last_stale_tracked_ids = completed_query.stale_tracked_structs; + last_stale_tracked_ids = completed_query.stale_tracked_structs; + }; - continue; - }; + crate::tracing::debug!( + "{database_key_index:?}: execute_maybe_iterate: result.revisions = {revisions:#?}", + revisions = &completed_query.revisions + ); - tracing::debug!( - "{database_key_index:?}: execute_maybe_iterate: result.revisions = {revisions:#?}", - revisions = &completed_query.revisions - ); + completed_query +} - (new_value, 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 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); + } } - #[inline] - fn execute_query<'db>( - db: &'db C::DbView, + /// 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, - 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); + 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), } + } +} - // Query was not previously executed, or value is potentially - // stale, or value is absent. Let's execute! - let new_value = C::execute( +/// 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 insert_provisional_memo( + &mut self, + zalsa: &'db Zalsa, + id: Id, + revision: Revision, + revisions: QueryRevisions, + ) -> 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, + memo_ingredient_index: MemoIngredientIndex, + ) -> Self { + Self { + ingredient, db, - C::id_to_input(zalsa, active_query.database_key_index.key_index()), + memo_ingredient_index, + value: None, + } + } +} + +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 + } - (new_value, active_query) + fn insert_provisional_memo( + &mut self, + zalsa: &'db Zalsa, + id: Id, + revision: Revision, + revisions: QueryRevisions, + ) -> 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), + self.memo_ingredient_index, + ) + .erase() } } @@ -530,22 +664,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() { @@ -656,7 +774,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 { @@ -765,7 +884,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, diff --git a/src/function/fetch.rs b/src/function/fetch.rs index 0a4037adc..a2146f0ef 100644 --- a/src/function/fetch.rs +++ b/src/function/fetch.rs @@ -1,8 +1,10 @@ -use crate::cycle::{CycleRecoveryStrategy, IterationStamp}; +use crate::cycle::IterationStamp; +use crate::function::cycle_strategy::{CycleStrategy, FetchCycleContext}; use crate::function::eviction::EvictionPolicy; -use crate::function::memo::Memo; +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}; @@ -125,17 +127,16 @@ where } ClaimResult::Cycle { .. } => { return Some(self.fetch_cold_cycle( + db, 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. + // 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 { @@ -143,111 +144,119 @@ where && old_memo.header.verify_memo( db.into(), &claim_guard, - C::CYCLE_STRATEGY, + C::CYCLE_RECOVERY_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. + // 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)) }; } } - self.execute(db, claim_guard, opt_old_memo) + self.execute(db, claim_guard, opt_old_memo, memo_ingredient_index) } #[cold] - #[inline(never)] fn fetch_cold_cycle<'db>( &'db self, + db: &'db C::DbView, 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:#?}, \ - 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) }; - } - } - - crate::tracing::debug!( - "hit cycle at {database_key_index:#?}, \ - inserting and returning fixpoint initial value" - ); - - 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, - ) - } + >::fetch_cold_cycle(FetchCycleContext { + ingredient: self, + db, + zalsa, + zalsa_local, + database_key_index, + memo_ingredient_index, + }) + } +} + +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| { + panic!( + "dependency graph cycle when querying {database_key_index:#?}, \ + set cycle_fn/cycle_initial to fixpoint iterate.\n\ + Query stack:\n{stack:#?}", + ); + }) + } +} + +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> { + 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 = ingredient.memo(zalsa, id); + + 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:#?}, \ + 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_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 3d7e8e1a6..703641f5a 100644 --- a/src/function/maybe_changed_after.rs +++ b/src/function/maybe_changed_after.rs @@ -234,7 +234,7 @@ where memo_slot, database_key_index, revision, - C::CYCLE_STRATEGY, + C::CYCLE_RECOVERY_STRATEGY, ) { ColdResult::Retry => None, ColdResult::Verified(result) => Some(result), @@ -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. @@ -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..c6f44887b 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}; @@ -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 @@ -234,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() @@ -323,12 +338,27 @@ 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. + 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 @@ -547,7 +577,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 { @@ -592,7 +623,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; @@ -647,13 +677,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; }