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
#![allow(clippy::pedantic)]
use tenshift_core::error::Result;
use tenshift_core::pipeline::ErrorPolicy;
use tenshift_core::sample::Sample;
use tenshift_core::source::{Source, SourceIterator};
use tenshift_core::Pipeline;
struct RapidBurstSource {
target: usize,
}
impl Source for RapidBurstSource {
fn open(&self) -> Result<Box<dyn SourceIterator>> {
Ok(Box::new(RapidBurstIterator {
target: self.target,
idx: 0,
}))
}
fn name(&self) -> &str {
"burst"
}
}
struct RapidBurstIterator {
target: usize,
idx: usize,
}
impl SourceIterator for RapidBurstIterator {
fn next_sample(&mut self) -> Option<Result<Sample>> {
if self.idx >= self.target {
return None;
}
self.idx += 1;
Some(Ok(Sample::new()))
}
}
#[test]
fn test_sqlite_massive_oversubscription_and_cancellation() {
// Inject massive thread contention:
// Hundreds of worker threads competing for tiny chunks and tiny prefetch buffers.
// Tests lock-free channel queue wrapping around and ABA anomalies in Crossbeam.
let pipeline = Pipeline::from_source(RapidBurstSource { target: 10_000 })
.workers(128) // Extreme OS thread oversubscription (standard CPUs have 8-32 cores)
.prefetch(1) // Force immediate backpressure blocking across 128 workers simultaneously
.chunk_size(1) // Force an allocation/channel switch every single input
.on_error(ErrorPolicy::Fail)
.map(Ok)
.batch(1);
let mut iter = pipeline
.start()
.expect("Pipeline should not fail 128 thread init");
// Let the first 50 items squeeze through the massive 128-thread pipeline
for _ in 0..50 {
let _ = iter.next();
}
// Now trigger cancellation while 128 hungry threads are blocking on bounded(1)
drop(iter);
// If Tenshift survives this, the channels successfully unblocked 128 threads via
// proper `AtomicBool` propagation without losing tracking or hitting a kernel deadlock.
}