1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
//! Thread-local depth counters for collection `atomic()` calls and for the
//! locked section of a transaction commit.
//!
//! Lives outside [`crate::txn`] so Bitcask/Typed facades can instrument
//! `atomic()` even when the `armour` feature (and thus `db.tx()`) is off.
//!
//! The two counters close the deadlock in **both** directions:
//!
//! * `atomic()` → `commit()` — the transaction would queue behind a shard lock
//! the caller already holds. Rejected by [`crate::txn::Tx::commit`].
//! * `commit()` → `atomic()` — `upsert`/`guard` closures run *under* the
//! transaction's shard locks, so reaching back into a collection from one
//! parks on a mutex this thread owns. Rejected by [`AtomicDepthGuard::try_enter`].
use crate::error::{DbError, DbResult};
thread_local! {
pub(crate) static ATOMIC_DEPTH: std::cell::Cell<u32> = const { std::cell::Cell::new(0) };
/// Non-zero while this thread is inside a transaction's locked section:
/// from the first shard lock to the last release. Hook replay happens after
/// that window, so hooks may touch collections as documented.
static TX_COMMIT_DEPTH: std::cell::Cell<u32> = const { std::cell::Cell::new(0) };
static STREAM_APPLY_DEPTH: std::cell::Cell<u32> = const { std::cell::Cell::new(0) };
}
/// Is this thread inside a transaction's locked section?
#[inline]
pub(crate) fn in_tx_commit() -> bool {
TX_COMMIT_DEPTH.with(|d| d.get()) > 0
}
#[inline]
pub(crate) fn in_stream_apply() -> bool {
STREAM_APPLY_DEPTH.with(|d| d.get() > 0)
}
/// Bypasses stream-source facade refusal only while the logged row is applied
/// (including recovery replay). Stream projection itself remains outside this
/// guard, so user callbacks cannot re-enter a direct write.
pub(crate) struct StreamApplyGuard;
impl StreamApplyGuard {
#[inline]
pub(crate) fn enter() -> Self {
STREAM_APPLY_DEPTH.with(|d| d.set(d.get().saturating_add(1)));
Self
}
}
impl Drop for StreamApplyGuard {
#[inline]
fn drop(&mut self) {
STREAM_APPLY_DEPTH.with(|d| d.set(d.get().saturating_sub(1)));
}
}
/// Held by `commit()` across evaluate + apply, so anything reaching back into a
/// collection from a user closure fails loudly instead of parking forever on a
/// lock this thread already holds.
pub(crate) struct TxCommitGuard;
impl TxCommitGuard {
#[inline]
pub(crate) fn enter() -> Self {
TX_COMMIT_DEPTH.with(|d| d.set(d.get().saturating_add(1)));
Self
}
}
impl Drop for TxCommitGuard {
#[inline]
fn drop(&mut self) {
TX_COMMIT_DEPTH.with(|d| {
let v = d.get();
debug_assert!(v > 0, "tx commit depth underflow");
d.set(v.saturating_sub(1));
});
}
}
/// Incremented for the duration of every collection `atomic()` call so
/// [`crate::txn::Tx::commit`] can reject reentrant use instead of deadlocking
/// (when the `armour` / txn stack is compiled in).
pub(crate) struct AtomicDepthGuard;
impl AtomicDepthGuard {
/// Enter an `atomic()` section, refusing to do so from inside a
/// transaction's locked window.
#[inline]
pub(crate) fn try_enter() -> DbResult<Self> {
if in_tx_commit() {
return Err(DbError::ReentrantTx);
}
ATOMIC_DEPTH.with(|d| d.set(d.get().saturating_add(1)));
Ok(Self)
}
}
impl Drop for AtomicDepthGuard {
#[inline]
fn drop(&mut self) {
ATOMIC_DEPTH.with(|d| {
let v = d.get();
debug_assert!(v > 0, "atomic depth underflow");
d.set(v - 1);
});
}
}