mirrors/salsa
1
0
Fork 0
mirror of https://github.com/salsa-rs/salsa.git synced 2025-01-27 15:07:03 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 2

report unexpected cycles using Cancelled

Browse source 
The `Cancelled` struct now reflects multiple potential reasons
for a query execution to be cancelled, including unexpected cycles.
It also gives information about the participants that lacked
recovery information.
This commit is contained in:
Niko Matsakis 2021-10-31 06:15:37 -04:00
parent 4a1dffe7bc
commit 61599cc81a
9 changed files with 149 additions and 99 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

11
src/derived.rs
View file

@ -1,7 +1,6 @@
use crate::debug::TableEntry;
use crate::durability::Durability;
use crate::lru::Lru;
use crate::plumbing::CycleError;
use crate::plumbing::DerivedQueryStorageOps;
use crate::plumbing::LruQueryStorageOps;
use crate::plumbing::QueryFunction;
@ -160,11 +159,7 @@ where
slot.maybe_changed_since(db, revision)
}
fn try_fetch(
&self,
db: &<Q as QueryDb<'_>>::DynDb,
key: &Q::Key,
) -> Result<Q::Value, CycleError> {
fn fetch(&self, db: &<Q as QueryDb<'_>>::DynDb, key: &Q::Key) -> Q::Value {
db.unwind_if_cancelled();
let slot = self.slot(key);
@ -172,7 +167,7 @@ where
value,
durability,
changed_at,
} = slot.read(db)?;
} = slot.read(db);
if let Some(evicted) = self.lru_list.record_use(&slot) {
evicted.evict();
@ -181,7 +176,7 @@ where
db.salsa_runtime()
.report_query_read(slot.database_key_index(), durability, changed_at);
Ok(value)
value
}
fn durability(&self, db: &<Q as QueryDb<'_>>::DynDb, key: &Q::Key) -> Durability {

33
src/derived/slot.rs
View file

@ -3,7 +3,6 @@ use crate::derived::MemoizationPolicy;
use crate::durability::Durability;
use crate::lru::LruIndex;
use crate::lru::LruNode;
use crate::plumbing::CycleError;
use crate::plumbing::{CycleDetected, CycleRecoveryStrategy};
use crate::plumbing::{DatabaseOps, QueryFunction};
use crate::revision::Revision;
@ -115,7 +114,7 @@ enum ProbeState<V, G> {
/// There is an entry which has been verified,
/// and it has the following value-- or, we blocked
/// on another thread, and that resulted in a cycle.
UpToDate(Result<V, CycleError>),
UpToDate(V),
}
/// Return value of `maybe_changed_since_probe` helper.
@ -151,10 +150,7 @@ where
self.database_key_index
}
pub(super) fn read(
&self,
db: &<Q as QueryDb<'_>>::DynDb,
) -> Result<StampedValue<Q::Value>, CycleError> {
pub(super) fn read(&self, db: &<Q as QueryDb<'_>>::DynDb) -> StampedValue<Q::Value> {
let runtime = db.salsa_runtime();
// NB: We don't need to worry about people modifying the
@ -188,7 +184,7 @@ where
&self,
db: &<Q as QueryDb<'_>>::DynDb,
revision_now: Revision,
) -> Result<StampedValue<Q::Value>, CycleError> {
) -> StampedValue<Q::Value> {
let runtime = db.salsa_runtime();
debug!("{:?}: read_upgrade(revision_now={:?})", self, revision_now,);
@ -236,11 +232,11 @@ where
panic_guard.proceed();
return Ok(value);
return value;
}
}
Ok(self.execute(db, runtime, revision_now, active_query, panic_guard))
self.execute(db, runtime, revision_now, active_query, panic_guard)
}
fn execute(
@ -376,18 +372,20 @@ where
anyone_waiting.store(true, Ordering::Relaxed);
return match self.block_on_in_progress_thread(db, runtime, other_id, state) {
Ok(WaitResult::Panicked) => Cancelled::throw(),
Ok(WaitResult::Panicked) => Cancelled::PropagatedPanic.throw(),
Ok(WaitResult::Completed) => ProbeState::Retry,
Err(CycleDetected {
recovery_strategy,
cycle_error,
}) => ProbeState::UpToDate(match recovery_strategy {
CycleRecoveryStrategy::Panic => Err(cycle_error),
CycleRecoveryStrategy::Fallback => Ok(StampedValue {
CycleRecoveryStrategy::Panic => {
Cancelled::UnexpectedCycle(cycle_error.into_unexpected_cycle()).throw()
}
CycleRecoveryStrategy::Fallback => StampedValue {
value: Q::cycle_fallback(db, &cycle_error.cycle, &self.key),
changed_at: cycle_error.changed_at,
durability: cycle_error.durability,
}),
},
}),
};
}
@ -414,7 +412,7 @@ where
self, value.changed_at
);
ProbeState::UpToDate(Ok(value))
ProbeState::UpToDate(value)
} else {
let changed_at = memo.revisions.changed_at;
ProbeState::NoValue(state, changed_at)
@ -524,17 +522,14 @@ where
// If we know when value last changed, we can return right away.
// Note that we don't need the actual value to be available.
ProbeState::NoValue(_, changed_at)
| ProbeState::UpToDate(Ok(StampedValue {
| ProbeState::UpToDate(StampedValue {
value: _,
durability: _,
changed_at,
})) => MaybeChangedSinceProbeState::ChangedAt(changed_at),
}) => MaybeChangedSinceProbeState::ChangedAt(changed_at),
// If we have nothing cached, then value may have changed.
ProbeState::NotComputed(_) => MaybeChangedSinceProbeState::ChangedAt(revision_now),
// Consider cycles as potentially having changed.
ProbeState::UpToDate(Err(_)) => MaybeChangedSinceProbeState::ChangedAt(revision_now),
}
}

9
src/input.rs
View file

@ -1,6 +1,5 @@
use crate::debug::TableEntry;
use crate::durability::Durability;
use crate::plumbing::CycleError;
use crate::plumbing::CycleRecoveryStrategy;
use crate::plumbing::InputQueryStorageOps;
use crate::plumbing::QueryStorageMassOps;
@ -97,11 +96,7 @@ where
slot.maybe_changed_since(db, revision)
}
fn try_fetch(
&self,
db: &<Q as QueryDb<'_>>::DynDb,
key: &Q::Key,
) -> Result<Q::Value, CycleError> {
fn fetch(&self, db: &<Q as QueryDb<'_>>::DynDb, key: &Q::Key) -> Q::Value {
db.unwind_if_cancelled();
let slot = self
@ -117,7 +112,7 @@ where
db.salsa_runtime()
.report_query_read(slot.database_key_index, durability, changed_at);
Ok(value)
value
}
fn durability(&self, _db: &<Q as QueryDb<'_>>::DynDb, key: &Q::Key) -> Durability {

18
src/interned.rs
View file

@ -7,7 +7,7 @@ use crate::plumbing::QueryStorageMassOps;
use crate::plumbing::QueryStorageOps;
use crate::revision::Revision;
use crate::Query;
use crate::{CycleError, Database, DatabaseKeyIndex, QueryDb};
use crate::{Database, DatabaseKeyIndex, QueryDb};
use parking_lot::RwLock;
use rustc_hash::FxHashMap;
use std::collections::hash_map::Entry;
@ -227,11 +227,7 @@ where
slot.maybe_changed_since(revision)
}
fn try_fetch(
&self,
db: &<Q as QueryDb<'_>>::DynDb,
key: &Q::Key,
) -> Result<Q::Value, CycleError> {
fn fetch(&self, db: &<Q as QueryDb<'_>>::DynDb, key: &Q::Key) -> Q::Value {
db.unwind_if_cancelled();
let slot = self.intern_index(db, key);
let changed_at = slot.interned_at;
@ -241,7 +237,7 @@ where
INTERN_DURABILITY,
changed_at,
);
Ok(<Q::Value>::from_intern_id(index))
<Q::Value>::from_intern_id(index)
}
fn durability(&self, _db: &<Q as QueryDb<'_>>::DynDb, _key: &Q::Key) -> Durability {
@ -346,11 +342,7 @@ where
interned_storage.maybe_changed_since(Q::convert_db(db), input, revision)
}
fn try_fetch(
&self,
db: &<Q as QueryDb<'_>>::DynDb,
key: &Q::Key,
) -> Result<Q::Value, CycleError> {
fn fetch(&self, db: &<Q as QueryDb<'_>>::DynDb, key: &Q::Key) -> Q::Value {
let index = key.as_intern_id();
let group_storage =
<<Q as QueryDb<'_>>::DynDb as HasQueryGroup<Q::Group>>::group_storage(db);
@ -363,7 +355,7 @@ where
INTERN_DURABILITY,
interned_at,
);
Ok(value)
value
}
fn durability(&self, _db: &<Q as QueryDb<'_>>::DynDb, _key: &Q::Key) -> Durability {

93
src/lib.rs
View file

@ -25,7 +25,7 @@ pub mod debug;
#[doc(hidden)]
pub mod plumbing;
use crate::plumbing::CycleError;
use crate::plumbing::CycleRecoveryStrategy;
use crate::plumbing::DerivedQueryStorageOps;
use crate::plumbing::InputQueryStorageOps;
use crate::plumbing::LruQueryStorageOps;
@ -494,12 +494,7 @@ where
/// queries (those with no inputs, or those with more than one
/// input) the key will be a tuple.
pub fn get(&self, key: Q::Key) -> Q::Value {
self.try_get(key)
.unwrap_or_else(|err| panic!("{:?}", err.debug(self.db)))
}
fn try_get(&self, key: Q::Key) -> Result<Q::Value, CycleError> {
self.storage.try_fetch(self.db, &key)
self.storage.fetch(self.db, &key)
}
/// Completely clears the storage for this query.
@ -601,16 +596,33 @@ where
}
}
/// A panic payload indicating that a salsa revision was cancelled.
/// A panic payload indicating that execution of a salsa query was cancelled.
///
/// This can occur for a few reasons:
/// *
/// *
/// *
#[derive(Debug)]
#[non_exhaustive]
pub struct Cancelled;
pub enum Cancelled {
/// The query was operating on revision R, but there is a pending write to move to revision R+1.
#[non_exhaustive]
PendingWrite,
/// The query was blocked on another thread, and that thread panicked.
#[non_exhaustive]
PropagatedPanic,
/// The query encountered an "unexpected" cycle, meaning one in which some
/// participants lacked cycle recovery annotations.
UnexpectedCycle(UnexpectedCycle),
}
impl Cancelled {
fn throw() -> ! {
fn throw(self) -> ! {
// We use resume and not panic here to avoid running the panic
// hook (that is, to avoid collecting and printing backtrace).
std::panic::resume_unwind(Box::new(Self));
std::panic::resume_unwind(Box::new(self));
}
/// Runs `f`, and catches any salsa cancellation.
@ -630,12 +642,69 @@ impl Cancelled {
impl std::fmt::Display for Cancelled {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.write_str("cancelled")
let why = match self {
Cancelled::PendingWrite => "pending write",
Cancelled::PropagatedPanic => "propagated panic",
Cancelled::UnexpectedCycle(_) => "unexpected cycle",
};
f.write_str("cancelled because of ")?;
f.write_str(why)
}
}
impl std::error::Error for Cancelled {}
/// Information about an "unexpected" cycle, meaning one where some of the
/// participants lacked cycle recovery annotations.
#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
pub struct UnexpectedCycle {
participants: plumbing::CycleParticipants,
}
impl UnexpectedCycle {
/// Returns a vector with the debug information for
/// all the participants in the cycle.
pub fn all_participants(&self, db: &dyn Database) -> Vec<String> {
self.participants
.iter()
.map(|d| format!("{:?}", d.debug(db)))
.collect()
}
/// Returns a vector with the debug information for
/// those participants in the cycle that lacked recovery
/// information.
pub fn unexpected_participants(&self, db: &dyn Database) -> Vec<String> {
self.participants
.iter()
.filter(|&&d| db.cycle_recovery_strategy(d) == CycleRecoveryStrategy::Panic)
.map(|d| format!("{:?}", d.debug(db)))
.collect()
}
/// Returns a "debug" view onto this strict that can be used to print out information.
pub fn debug<'me>(&'me self, db: &'me dyn Database) -> impl std::fmt::Debug + 'me {
struct UnexpectedCycleDebug<'me> {
c: &'me UnexpectedCycle,
db: &'me dyn Database,
}
impl<'me> std::fmt::Debug for UnexpectedCycleDebug<'me> {
fn fmt(&self, fmt: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
fmt.debug_struct("UnexpectedCycle")
.field("all_participants", &self.c.all_participants(self.db))
.field(
"unexpected_participants",
&self.c.unexpected_participants(self.db),
)
.finish()
}
}
UnexpectedCycleDebug { c: self, db }
}
}
// Re-export the procedural macros.
#[allow(unused_imports)]
#[macro_use]

31
src/plumbing.rs
View file

@ -6,6 +6,7 @@ use crate::Database;
use crate::Query;
use crate::QueryTable;
use crate::QueryTableMut;
use crate::UnexpectedCycle;
use std::borrow::Borrow;
use std::fmt::Debug;
use std::hash::Hash;
@ -21,7 +22,7 @@ pub use crate::{revision::Revision, DatabaseKeyIndex, QueryDb, Runtime};
#[derive(Clone, Debug)]
pub struct CycleDetected {
pub(crate) recovery_strategy: CycleRecoveryStrategy,
pub(crate) cycle_error: crate::CycleError,
pub(crate) cycle_error: CycleError,
}
/// Defines various associated types. An impl of this
@ -189,11 +190,7 @@ where
/// Returns `Err` in the event of a cycle, meaning that computing
/// the value for this `key` is recursively attempting to fetch
/// itself.
fn try_fetch(
&self,
db: &<Q as QueryDb<'_>>::DynDb,
key: &Q::Key,
) -> Result<Q::Value, CycleError>;
fn fetch(&self, db: &<Q as QueryDb<'_>>::DynDb, key: &Q::Key) -> Q::Value;
/// Returns the durability associated with a given key.
fn durability(&self, db: &<Q as QueryDb<'_>>::DynDb, key: &Q::Key) -> Durability;
@ -250,26 +247,10 @@ pub struct CycleError {
}
impl CycleError {
pub(crate) fn debug<'a, D: ?Sized>(&'a self, db: &'a D) -> impl Debug + 'a
where
D: DatabaseOps,
{
struct CycleErrorDebug<'a, D: ?Sized> {
db: &'a D,
error: &'a CycleError,
pub(crate) fn into_unexpected_cycle(self) -> UnexpectedCycle {
UnexpectedCycle {
participants: self.cycle,
}
impl<'a, D: ?Sized + DatabaseOps> Debug for CycleErrorDebug<'a, D> {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
writeln!(f, "Internal error, cycle detected:\n")?;
for i in &*self.error.cycle {
writeln!(f, "{:?}", i.debug(self.db))?;
}
Ok(())
}
}
CycleErrorDebug { db, error: self }
}
}

2
src/runtime.rs
View file

@ -150,7 +150,7 @@ impl Runtime {
#[cold]
pub(crate) fn unwind_cancelled(&self) {
self.report_untracked_read();
Cancelled::throw();
Cancelled::PendingWrite.throw();
}
/// Acquires the **global query write lock** (ensuring that no queries are

36
tests/cycles.rs
View file

@ -1,4 +1,4 @@
use salsa::{ParallelDatabase, Snapshot};
use salsa::{Cancelled, ParallelDatabase, Snapshot};
use test_env_log::test;
// Axes:
@ -136,17 +136,39 @@ fn cycle_c(db: &dyn Database) -> Result<(), Error> {
}
#[test]
#[should_panic(expected = "cycle detected")]
fn cycle_memoized() {
let query = DatabaseImpl::default();
query.memoized_a();
let db = DatabaseImpl::default();
match Cancelled::catch(|| {
db.memoized_a();
}) {
Err(Cancelled::UnexpectedCycle(c)) => {
insta::assert_debug_snapshot!(c.unexpected_participants(&db), @r###"
[
"memoized_a(())",
"memoized_b(())",
]
"###);
}
v => panic!("unexpected result: {:#?}", v),
}
}
#[test]
#[should_panic(expected = "cycle detected")]
fn cycle_volatile() {
let query = DatabaseImpl::default();
query.volatile_a();
let db = DatabaseImpl::default();
match Cancelled::catch(|| {
db.volatile_a();
}) {
Err(Cancelled::UnexpectedCycle(c)) => {
insta::assert_debug_snapshot!(c.unexpected_participants(&db), @r###"
[
"volatile_a(())",
"volatile_b(())",
]
"###);
}
v => panic!("unexpected result: {:#?}", v),
}
}
#[test]

15
tests/parallel/cycles.rs
View file

@ -132,14 +132,15 @@ fn panic_parallel_cycle() {
// We expect B to panic because it detects a cycle (it is the one that calls A, ultimately).
// Right now, it panics with a string.
let err_b = thread_b.join().unwrap_err();
if let Some(str_b) = err_b.downcast_ref::<String>() {
assert!(
str_b.contains("cycle detected"),
"unexpeced string: {:?}",
str_b
);
if let Some(Cancelled::UnexpectedCycle(c)) = err_b.downcast_ref::<Cancelled>() {
insta::assert_debug_snapshot!(c.unexpected_participants(&db), @r###"
[
"panic_cycle_a(-1)",
"panic_cycle_b(-1)",
]
"###);
} else {
panic!("b failed in an unexpected way");
panic!("b failed in an unexpected way: {:?}", err_b);
}
// We expect A to propagate a panic, which causes us to use the sentinel