#![allow(deprecated)]
#![cfg(feature = "experimental")]
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::{Arc, Mutex};
use std::thread::scope;
use proptest::prelude::*;
use sefer_alloc::{ShardedHandle, ShardedRegion};
struct Payload {
id: u64,
drops: Arc<AtomicUsize>,
}
impl PartialEq for Payload {
fn eq(&self, other: &Self) -> bool {
self.id == other.id
}
}
impl Drop for Payload {
fn drop(&mut self) {
self.drops.fetch_add(1, Ordering::Relaxed);
}
}
proptest! {
#![proptest_config(ProptestConfig::with_cases(64))]
#[test]
fn cross_thread_remove_honours_invariants(
n_inserts in 1usize..=16,
remove_pattern in prop::collection::vec(any::<bool>(), 1..=32),
) {
const CAP: usize = 32;
let region = Arc::new(ShardedRegion::<Payload>::with_shards(2, CAP));
let drops = Arc::new(AtomicUsize::new(0));
let handles: Vec<ShardedHandle<Payload>> = scope(|s| {
let region = Arc::clone(®ion);
let drops = Arc::clone(&drops);
s.spawn(move || {
let mut hs = Vec::new();
for id in 0..n_inserts as u64 {
let p = Payload { id, drops: Arc::clone(&drops) };
match region.insert(p) {
Ok(h) => hs.push(h),
Err(_) => break, }
}
hs
}).join().expect("inserter panicked")
});
for (i, h) in handles.iter().enumerate() {
prop_assert_eq!(
region.get_with(*h, |p| p.id),
Some(i as u64),
"a live handle must resolve cross-thread (I1, reads are ownership-free)"
);
}
prop_assert_eq!(region.len(), handles.len(), "I4: len tracks inserts");
let mut to_remove: Vec<usize> = Vec::new();
for (i, &rm) in remove_pattern.iter().enumerate() {
if rm {
to_remove.push(i % handles.len());
}
}
to_remove.sort_unstable();
to_remove.dedup();
let removed_set: std::collections::HashSet<usize> = to_remove.iter().copied().collect();
let to_remove_clone = to_remove.clone();
let remove_outcomes: Vec<(usize, bool, bool)> = scope(|s| {
let region = Arc::clone(®ion);
let handles = handles.clone();
s.spawn(move || {
to_remove_clone
.iter()
.map(|&idx| {
let h = handles[idx];
let was_live = region.get_with(h, |p| p.id) == Some(idx as u64);
let removed = region.remove(h);
let second = region.remove(h);
(idx, was_live, removed && !second)
})
.collect::<Vec<_>>()
})
.join()
.expect("remover thread panicked")
});
let mut removed_count = 0usize;
for (idx, was_live, removed_once) in remove_outcomes {
let h = handles[idx];
if was_live {
prop_assert!(
removed_once,
"a live handle must remove cross-thread exactly once; \
second remove must be a no-op (I2, no double-free)"
);
removed_count += 1;
}
prop_assert_eq!(
region.get_with(h, |p| p.id),
None,
"removed handle is None forever (I2)"
);
}
prop_assert_eq!(
region.len(),
handles.len() - removed_count,
"I4: len tracks cross-thread removes"
);
let survivors_resolve = scope(|s| {
let region = Arc::clone(®ion);
let handles = handles.clone();
s.spawn(move || {
handles
.iter()
.enumerate()
.filter(|(i, _)| !removed_set.contains(i))
.all(|(i, h)| region.get_with(*h, |p| p.id) == Some(i as u64))
})
.join()
.expect("survivor-check thread panicked")
});
prop_assert!(
survivors_resolve,
"survivors still resolve cross-thread after remote removes (I1)"
);
}
}
#[test]
fn dead_thread_live_slots_stay_resolvable_and_removable_cross_thread() {
const CAP: usize = 8;
let region = Arc::new(ShardedRegion::<u64>::with_shards(2, CAP));
let handles: Vec<ShardedHandle<u64>> = scope(|s| {
let region = Arc::clone(®ion);
s.spawn(move || {
let mut hs = Vec::new();
for v in [100_u64, 200, 300] {
hs.push(region.insert(v).expect("insert into fresh shard"));
}
for (h, v) in hs.iter().zip([100_u64, 200, 300]) {
assert_eq!(region.get_cloned(*h), Some(v), "owner reads its own values");
}
hs
})
.join()
.expect("owner thread panicked")
});
for (h, v) in handles.iter().zip([100_u64, 200, 300]) {
assert_eq!(
region.get_cloned(*h),
Some(v),
"a dead thread's live slots must stay resolvable (reads are ownership-free)"
);
}
assert_eq!(region.len(), 3, "len holds after the owner thread died");
for h in &handles {
assert!(
region.remove(*h),
"cross-thread remove of a dead owner's handle must succeed"
);
assert_eq!(region.get_cloned(*h), None, "removed handle is None (I2)");
}
assert_eq!(
region.len(),
0,
"all handles removed cross-thread after owner death"
);
for h in &handles {
assert!(!region.remove(*h), "second remove is a no-op (I2)");
}
let new_handle: ShardedHandle<u64> = scope(|s| {
let region = Arc::clone(®ion);
s.spawn(move || region.insert(999_u64).expect("new thread claims a shard"))
.join()
.expect("adopter thread panicked")
});
assert_eq!(
region.get_cloned(new_handle),
Some(999),
"new thread's insert resolves"
);
}
struct Counted {
v: u64,
drops: Arc<AtomicUsize>,
}
impl Counted {
fn id(&self) -> u64 {
self.v
}
}
impl Drop for Counted {
fn drop(&mut self) {
self.drops.fetch_add(1, Ordering::Relaxed);
}
}
type HandoffQueue = Arc<Mutex<Vec<(ShardedHandle<Counted>, u64)>>>;
#[test]
fn concurrent_cross_thread_remove_never_double_frees_and_accounts() {
const N_INSERTERS: usize = 2;
const N_REMOVERS: usize = 2;
const VALS_PER_INSERTER: usize = 64;
const N_SHARDS: usize = 4;
const CAP: usize = VALS_PER_INSERTER * N_INSERTERS;
let region = Arc::new(ShardedRegion::<Counted>::with_shards(N_SHARDS, CAP));
let drops = Arc::new(AtomicUsize::new(0));
let queue: HandoffQueue = Arc::new(Mutex::new(Vec::new()));
let total_inserted = Arc::new(AtomicUsize::new(0));
scope(|s| {
for inserter_tid in 0..N_INSERTERS {
let region = Arc::clone(®ion);
let queue = Arc::clone(&queue);
let drops = Arc::clone(&drops);
let total_inserted = Arc::clone(&total_inserted);
s.spawn(move || {
for seq in 0..VALS_PER_INSERTER as u64 {
let id = u64::try_from(inserter_tid).unwrap() * 1_000_000 + seq;
let v = Counted {
v: id,
drops: Arc::clone(&drops),
};
if let Ok(h) = region.insert(v) {
queue.lock().unwrap().push((h, id));
total_inserted.fetch_add(1, Ordering::Relaxed);
}
}
});
}
for _ in 0..N_REMOVERS {
let region = Arc::clone(®ion);
let queue = Arc::clone(&queue);
s.spawn(move || loop {
let next = queue.lock().unwrap().pop();
let Some((h, _id)) = next else { break };
let removed = region.remove(h);
if removed {
assert!(
!region.remove(h),
"second cross-thread remove must be a no-op (no double-free)"
);
}
});
}
});
let leftover: Vec<_> = queue.lock().unwrap().drain(..).collect();
let survivors = leftover.len();
for (h, id) in &leftover {
assert_eq!(
region.get_with(*h, Counted::id),
Some(*id),
"survivor must resolve after concurrent cross-thread churn (I1)"
);
}
let inserted = total_inserted.load(Ordering::Relaxed);
assert_eq!(
region.len(),
survivors,
"I4: len == survivors after concurrent cross-thread churn"
);
drop(region);
drop(leftover);
let observed_drops = drops.load(Ordering::Relaxed);
assert!(
observed_drops >= survivors,
"at least every survivor dropped once (survivors={survivors}, drops={observed_drops})"
);
assert!(
observed_drops <= inserted,
"no double-free: drops ({observed_drops}) <= inserted ({inserted})"
);
}