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
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
//! Concurrency tests via loom for the CLI slot semaphore.
//!
//! Models the central invariant of `src/lock.rs`: at most `MAX_SLOTS` threads
//! can hold a slot simultaneously. The tests use `AtomicUsize` as the
//! counter of active slots — loom-visible equivalent of the `flock` semaphore.
//!
//! loom caps the total number of threads (including main) at `loom::MAX_THREADS = 5`.
//! Therefore each model uses at most 4 spawned threads.
//!
//! Run with:
//! ```text
//! RUSTFLAGS="--cfg sqlite_graphrag_loom" cargo nextest run --profile heavy -E 'test(/^loom_/)'
//! ```
//! Or via script: `bash scripts/test-loom.sh`
//!
//! DO NOT run with `cargo test` without `--test-threads=1` — loom in parallel
//! may saturate the CPU and cause thermal livelock (incident 2026-04-19).
#![cfg(sqlite_graphrag_loom)]
use loom::sync::atomic::{AtomicUsize, Ordering};
use loom::sync::Arc;
use serial_test::serial;
/// Non-blocking counting semaphore that mirrors the logic of `try_any_slot`.
///
/// `try_acquire` uses optimistic CAS: reads the current counter and, if below
/// `max`, attempts to increment it atomically. Returns `true` on success and
/// `false` if all slots are occupied — identical to the behavior
/// de `try_lock_exclusive` retornando `WouldBlock`.
struct SlotSemaforo {
contador: Arc<AtomicUsize>,
max: usize,
}
impl SlotSemaforo {
fn novo(max: usize) -> Self {
Self {
contador: Arc::new(AtomicUsize::new(0)),
max,
}
}
fn clonar(&self) -> Self {
Self {
contador: Arc::clone(&self.contador),
max: self.max,
}
}
/// Tries to acquire a slot without blocking. Returns `true` se adquiriu.
fn try_acquire(&self) -> bool {
let mut atual = self.contador.load(Ordering::Acquire);
loop {
if atual >= self.max {
return false;
}
match self.contador.compare_exchange_weak(
atual,
atual + 1,
Ordering::AcqRel,
Ordering::Acquire,
) {
Ok(_) => return true,
Err(novo) => atual = novo,
}
}
}
/// Releases a previously acquired slot.
fn release(&self) {
let anterior = self.contador.fetch_sub(1, Ordering::AcqRel);
assert!(
anterior > 0,
"release sem acquire correspondente — double-free detectado"
);
}
/// Reads the current number of occupied slots.
fn ocupados(&self) -> usize {
self.contador.load(Ordering::Acquire)
}
}
/// Test 1 — Maximum capacity invariant: 4 threads compete for 3 slots.
///
/// With more threads than slots, at least 1 thread always fails acquisition.
/// Checks that the occupied-slot counter NEVER exceeds `max_slots`,
/// regardless of the concurrent scheduling explored by loom.
///
/// loom::MAX_THREADS = 5 (main + 4 spawned), therefore maximum of 4 spawns.
#[serial(loom_model)]
#[test]
fn quatro_threads_invariante_maximo_tres_slots() {
const NUM_THREADS: usize = 4;
const MAX_SLOTS: usize = 3;
let mut builder = loom::model::Builder::new();
builder.preemption_bound = Some(1);
builder.max_branches = 100;
builder.check(|| {
let sem = Arc::new(SlotSemaforo::novo(MAX_SLOTS));
let contador_holds = Arc::new(AtomicUsize::new(0));
let mut handles = Vec::new();
for _ in 0..NUM_THREADS {
let sem_t = Arc::new(sem.clonar());
let holds_t = Arc::clone(&contador_holds);
let h = loom::thread::spawn(move || {
if sem_t.try_acquire() {
// Record that this thread holds a slot.
let agora = holds_t.fetch_add(1, Ordering::AcqRel) + 1;
// Core invariant: never more than MAX_SLOTS simultaneous holds.
assert!(
agora <= MAX_SLOTS,
"violação: {agora} holds simultâneos ultrapassam o limite {MAX_SLOTS}"
);
loom::thread::yield_now();
holds_t.fetch_sub(1, Ordering::AcqRel);
sem_t.release();
}
// Thread that did not acquire a slot simply returns — no panic.
});
handles.push(h);
}
for h in handles {
h.join().expect("thread terminou com pânico");
}
// Ao final, todos os slots devem ter sido liberados.
assert_eq!(
sem.ocupados(),
0,
"slots ainda ocupados após todas as threads terminarem"
);
});
}
/// Test 2 — Release frees a slot and lets another thread acquire it.
///
/// Thread A acquires the single available slot. Thread B tries to acquire and fails.
/// After A releases, B successfully acquires on the next attempt.
/// Modela o comportamento de polling de `acquire_cli_slot` com `wait_seconds > 0`.
#[serial(loom_model)]
#[test]
fn release_frees_slot_for_next_thread() {
let mut builder = loom::model::Builder::new();
builder.preemption_bound = Some(1);
builder.max_branches = 100;
builder.check(|| {
// Semaphore with 1 slot to force deterministic contention.
let sem = Arc::new(SlotSemaforo::novo(1));
assert!(
sem.try_acquire(),
"main deve ocupar o único slot antes de iniciar as threads"
);
let sem_a = Arc::new(sem.clonar());
let sem_b = Arc::new(sem.clonar());
// Signal that A released the slot.
let liberado = Arc::new(AtomicUsize::new(0));
let liberado_b = Arc::clone(&liberado);
let b_adquiriu = Arc::new(AtomicUsize::new(0));
let b_adquiriu_t = Arc::clone(&b_adquiriu);
let ha = loom::thread::spawn(move || {
loom::thread::yield_now();
sem_a.release();
// Sinaliza que o slot foi liberado.
liberado.store(1, Ordering::Release);
});
let hb = loom::thread::spawn(move || {
// B retries in a loop until the slot is free — models wait_seconds polling.
loop {
if sem_b.try_acquire() {
b_adquiriu_t.store(1, Ordering::Release);
sem_b.release();
break;
}
// No slot held: check if A already released before retrying.
if liberado_b.load(Ordering::Acquire) == 1 {
// Try one last time after the release was confirmed.
if sem_b.try_acquire() {
b_adquiriu_t.store(1, Ordering::Release);
sem_b.release();
}
break;
}
loom::thread::yield_now();
}
});
ha.join().expect("thread A terminou com pânico");
hb.join().expect("thread B terminou com pânico");
assert_eq!(
sem.ocupados(),
0,
"slot deve estar livre após ambas as threads terminarem"
);
assert_eq!(
b_adquiriu.load(Ordering::Acquire),
1,
"thread B deve adquirir o slot após a liberação"
);
});
}
/// Test 3 — Clean shutdown: all slots released after termination.
///
/// 4 threads acquire and release slots in parallel. After all finish,
/// the counter must be zero — shutdown invariant of the `flock` semaphore.
///
/// loom::MAX_THREADS = 5 (main + 4 spawned). Aqui usamos exatamente 4.
#[serial(loom_model)]
#[test]
fn clean_shutdown_releases_all_slots() {
const NUM_THREADS: usize = 4;
const MAX_SLOTS: usize = 4;
let mut builder = loom::model::Builder::new();
builder.preemption_bound = Some(1);
builder.max_branches = 100;
builder.check(|| {
let sem = Arc::new(SlotSemaforo::novo(MAX_SLOTS));
let adquiridos_total = Arc::new(AtomicUsize::new(0));
let mut handles = Vec::new();
for _ in 0..NUM_THREADS {
let sem_t = Arc::new(sem.clonar());
let total_t = Arc::clone(&adquiridos_total);
let h = loom::thread::spawn(move || {
if sem_t.try_acquire() {
total_t.fetch_add(1, Ordering::AcqRel);
loom::thread::yield_now();
sem_t.release();
}
});
handles.push(h);
}
for h in handles {
h.join()
.expect("thread terminou com pânico durante shutdown");
}
// Invariante de shutdown: contador retorna a zero.
assert_eq!(
sem.ocupados(),
0,
"shutdown sujo: {n} slots ainda ocupados",
n = sem.ocupados()
);
// At least one thread acquired a slot — the system was not stuck.
let total = adquiridos_total.load(Ordering::Acquire);
assert!(
total > 0,
"nenhuma thread adquiriu slot — possível deadlock no modelo"
);
});
}