veilid_tools/tracked_mutex.rs
1//! `TrackedMutex<T>`: a drop-in `parking_lot::Mutex<T>` wrapper that, under the
2//! `debug-locks` feature, records the owning thread plus an acquisition backtrace
3//! and reports the holder when an acquisition cannot complete — covering the case
4//! the async `parking_lot` deadlock detector cannot: a sync lock whose holder is a
5//! suspended task or a re-entrant acquire on a runtime so starved that the
6//! detector task itself can never be scheduled to run.
7//!
8//! - Re-entrancy (a thread locking a `Mutex` it already holds) is reported
9//! immediately, without waiting out the timeout.
10//! - A holder still stuck after `DEBUG_LOCKS_DURATION_MS` is reported with both
11//! the blocked-acquire and holder-acquire backtraces, then the process exits.
12//!
13//! Without `debug-locks` this is exactly `parking_lot::Mutex<T>` via type alias —
14//! zero overhead, identical semantics.
15
16use cfg_if::cfg_if;
17
18cfg_if! {
19 if #[cfg(feature = "debug-locks")] {
20 use std::sync::atomic::{AtomicBool, Ordering};
21 use std::sync::Mutex as StdMutex;
22 use std::thread::ThreadId;
23
24 // One report is enough; the other blocked threads would print the same standoff.
25 static REPORTED: AtomicBool = AtomicBool::new(false);
26
27 #[derive(Debug)]
28 struct HolderInfo {
29 thread_id: ThreadId,
30 thread_name: String,
31 backtrace: backtrace::Backtrace,
32 }
33
34 /// See module docs. Holder state lives in a `std` mutex so it never enters
35 /// `parking_lot`'s own deadlock bookkeeping.
36 pub struct TrackedMutex<T> {
37 holder: StdMutex<Option<HolderInfo>>,
38 inner: parking_lot::Mutex<T>,
39 }
40
41 impl<T> TrackedMutex<T> {
42 /// Wrap `val` in a tracked mutex.
43 pub fn new(val: T) -> Self {
44 Self {
45 holder: StdMutex::new(None),
46 inner: parking_lot::Mutex::new(val),
47 }
48 }
49
50 /// Lock, recording the holder. A re-entrant acquire reports the standoff
51 /// and exits the process immediately; off-wasm a non-re-entrant contended
52 /// acquire waits up to `DEBUG_LOCKS_DURATION_MS` before reporting, on
53 /// single-threaded wasm it reports at once. The lock is held until the
54 /// returned guard drops.
55 pub fn lock(&self) -> TrackedMutexGuard<'_, T> {
56 // Fast path: uncontended.
57 if let Some(guard) = self.inner.try_lock() {
58 self.record_holder();
59 return TrackedMutexGuard { mutex: self, guard };
60 }
61 // Contended: a re-entrant acquire would otherwise wait the full timeout for
62 // a lock this very thread holds, so catch it now.
63 if self.held_by_current_thread() {
64 self.report("re-entrant lock (the blocked thread already holds it)");
65 }
66 // Otherwise hand off to the per-target wait strategy.
67 self.wait_or_report()
68 }
69
70 // Has a timer and (potentially) other threads: contention may be transient, so
71 // wait — only a holder still stuck past the timeout is a deadlock.
72 #[cfg(not(all(target_arch = "wasm32", target_os = "unknown")))]
73 fn wait_or_report(&self) -> TrackedMutexGuard<'_, T> {
74 let timeout =
75 std::time::Duration::from_millis(super::DEBUG_LOCKS_DURATION_MS as u64);
76 match self.inner.try_lock_for(timeout) {
77 Some(guard) => {
78 self.record_holder();
79 TrackedMutexGuard { mutex: self, guard }
80 }
81 None => self.report("acquisition timed out"),
82 }
83 }
84
85 // Single-threaded wasm: there is no timer (try_lock_for would panic on
86 // Instant::now), and the holder is a suspended task that cannot run while we
87 // block the only thread — so any contention here is already a deadlock.
88 #[cfg(all(target_arch = "wasm32", target_os = "unknown"))]
89 fn wait_or_report(&self) -> TrackedMutexGuard<'_, T> {
90 self.report("contended on single-threaded executor (holder is a suspended task)")
91 }
92
93 /// Lock without blocking, recording the holder on success.
94 /// Returns `None` immediately if the lock is held; the lock is held
95 /// until the returned guard drops.
96 pub fn try_lock(&self) -> Option<TrackedMutexGuard<'_, T>> {
97 let guard = self.inner.try_lock()?;
98 self.record_holder();
99 Some(TrackedMutexGuard { mutex: self, guard })
100 }
101
102 fn record_holder(&self) {
103 let cur = std::thread::current();
104 *self.holder.lock().unwrap() = Some(HolderInfo {
105 thread_id: cur.id(),
106 thread_name: cur.name().unwrap_or("<unnamed>").to_string(),
107 backtrace: backtrace::Backtrace::new_unresolved(),
108 });
109 }
110
111 fn clear_holder(&self) {
112 *self.holder.lock().unwrap() = None;
113 }
114
115 fn held_by_current_thread(&self) -> bool {
116 let cur = std::thread::current().id();
117 self.holder
118 .lock()
119 .unwrap()
120 .as_ref()
121 .is_some_and(|h| h.thread_id == cur)
122 }
123
124 fn report(&self, reason: &str) -> ! {
125 use core::fmt::Write as _;
126 // Only the first blocked thread prints; wedge the rest so output isn't interleaved.
127 if REPORTED.swap(true, Ordering::SeqCst) {
128 loop {
129 std::thread::park();
130 }
131 }
132 let cur = std::thread::current();
133 let mut r = String::new();
134 let _ = writeln!(r, "\n===== TrackedMutex DEADLOCK: {reason} =====");
135 let _ = writeln!(
136 r,
137 "blocked thread: {:?} ({})",
138 cur.id(),
139 cur.name().unwrap_or("<unnamed>")
140 );
141 let mut blocked_bt = backtrace::Backtrace::new_unresolved();
142 blocked_bt.resolve();
143 let _ = writeln!(
144 r,
145 "blocked-acquire backtrace:\n{}",
146 indent::indent_all_by(4, format!("{blocked_bt:?}"))
147 );
148 match self.holder.lock().unwrap().as_ref() {
149 Some(h) => {
150 let reentrant = h.thread_id == cur.id();
151 let _ = writeln!(
152 r,
153 "current holder: {:?} ({}){}",
154 h.thread_id,
155 h.thread_name,
156 if reentrant {
157 " <<< SAME THREAD — RE-ENTRANT DEADLOCK"
158 } else {
159 ""
160 }
161 );
162 let mut hbt = h.backtrace.clone();
163 hbt.resolve();
164 let _ = writeln!(
165 r,
166 "holder-acquire backtrace:\n{}",
167 indent::indent_all_by(4, format!("{hbt:?}"))
168 );
169 }
170 None => {
171 let _ = writeln!(
172 r,
173 "current holder: <none recorded> — contended but unowned (lost wakeup?)"
174 );
175 }
176 }
177 let _ = writeln!(r, "===== end TrackedMutex deadlock report =====\n");
178 // Belt-and-suspenders: native stderr may not be captured by the test
179 // harness, so also drop the report to a well-known file.
180 eprintln!("{r}");
181 // Native stderr isn't always captured by the harness (e.g. flutter test),
182 // so also drop the report to a well-known file where a filesystem exists.
183 #[cfg(not(all(target_arch = "wasm32", target_os = "unknown")))]
184 {
185 let _ = std::fs::write(
186 std::env::temp_dir().join("veilid_tracked_mutex_deadlock.txt"),
187 r.as_bytes(),
188 );
189 }
190 std::process::exit(1);
191 }
192 }
193
194 impl<T: core::fmt::Debug> core::fmt::Debug for TrackedMutex<T> {
195 fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
196 // Mirror parking_lot::Mutex's Debug (prints data, or <locked>).
197 core::fmt::Debug::fmt(&self.inner, f)
198 }
199 }
200
201 /// Guard for a `TrackedMutex`; clears the recorded holder on drop.
202 pub struct TrackedMutexGuard<'a, T> {
203 mutex: &'a TrackedMutex<T>,
204 guard: parking_lot::MutexGuard<'a, T>,
205 }
206
207 impl<T> Drop for TrackedMutexGuard<'_, T> {
208 fn drop(&mut self) {
209 self.mutex.clear_holder();
210 }
211 }
212
213 impl<T> core::ops::Deref for TrackedMutexGuard<'_, T> {
214 type Target = T;
215 fn deref(&self) -> &T {
216 &self.guard
217 }
218 }
219
220 impl<T> core::ops::DerefMut for TrackedMutexGuard<'_, T> {
221 fn deref_mut(&mut self) -> &mut T {
222 &mut self.guard
223 }
224 }
225 } else {
226 /// Without `debug-locks`, a plain `parking_lot::Mutex<T>`.
227 pub type TrackedMutex<T> = parking_lot::Mutex<T>;
228 /// Without `debug-locks`, a plain `parking_lot::MutexGuard<T>`.
229 pub type TrackedMutexGuard<'a, T> = parking_lot::MutexGuard<'a, T>;
230 }
231}