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
108
109
110
|
//! Thread-ID based reentrant session lock.
//!
//! Guarantees at most one logical RPC "call session" is in flight against the shared worker at
//! any time, while allowing the owning thread to recurse into it freely (a handler may itself
//! call back into the other side). A different thread attempting to start a session blocks until
//! the entire outer session (including all of its nested calls) completes — it is never rejected
//! or panicked on, only serialized. This keeps the "exactly one child process" invariant intact
//! even when multiple OS threads use this crate concurrently, which happens today only under
//! `cargo test`'s parallel test harness but is not assumed to be forbidden in the future.
use std::sync::{Condvar, LazyLock, Mutex};
struct SessionLock {
owner: Mutex<Option<(std::thread::ThreadId, u32)>>,
cvar: Condvar,
}
impl SessionLock {
fn acquire(&self) {
let mut owner = self.owner.lock().unwrap();
let me = std::thread::current().id();
loop {
match *owner {
None => {
*owner = Some((me, 1));
return;
}
Some((tid, depth)) if tid == me => {
*owner = Some((me, depth + 1));
return;
}
Some(_) => {
owner = self.cvar.wait(owner).unwrap();
}
}
}
}
fn release(&self) {
let mut owner = self.owner.lock().unwrap();
let me = std::thread::current().id();
match *owner {
Some((tid, depth)) if tid == me => {
if depth == 1 {
*owner = None;
self.cvar.notify_all();
} else {
*owner = Some((me, depth - 1));
}
}
_ => unreachable!("SessionLock::release without a matching acquire on this thread"),
}
}
}
static SESSION: LazyLock<SessionLock> = LazyLock::new(|| SessionLock {
owner: Mutex::new(None),
cvar: Condvar::new(),
});
/// RAII guard; acquired once at the outermost `rpc_call`, re-entered (depth += 1, no blocking)
/// by nested calls from the same thread.
pub struct SessionGuard;
impl SessionGuard {
pub fn enter() -> Self {
SESSION.acquire();
SessionGuard
}
}
impl Drop for SessionGuard {
fn drop(&mut self) {
SESSION.release();
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn same_thread_reenters_without_blocking() {
let _outer = SessionGuard::enter();
let _inner = SessionGuard::enter();
}
#[test]
fn other_threads_are_serialized() {
use std::sync::Arc;
use std::sync::atomic::{AtomicU32, Ordering};
let concurrent = Arc::new(AtomicU32::new(0));
let mut handles = Vec::new();
for _ in 0..4 {
let concurrent = Arc::clone(&concurrent);
handles.push(std::thread::spawn(move || {
for _ in 0..50 {
let _guard = SessionGuard::enter();
let now = concurrent.fetch_add(1, Ordering::SeqCst);
assert_eq!(now, 0, "two sessions were in flight at once");
concurrent.fetch_sub(1, Ordering::SeqCst);
}
}));
}
for handle in handles {
handle.join().unwrap();
}
}
}
|
