#![allow(deprecated)]
#![cfg(feature = "pinning")]
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::{Arc, Mutex};
use sefer_alloc::{PinnedRunner, ShardedHandle, ShardedRegion};
#[test]
fn bind_routes_inserts_to_bound_shard() {
let region = ShardedRegion::<u64>::with_shards(4, 16);
let handle = std::thread::scope(|s| {
s.spawn(|| {
assert!(
region.bind_current_thread_to_shard(2),
"in-range bind must succeed"
);
region.insert(0xCAFE_u64).expect("shard has capacity")
})
.join()
.expect("worker must not panic")
});
assert_eq!(handle.shard(), 2, "bind must route the insert to shard 2");
assert_eq!(
region.get_cloned(handle),
Some(0xCAFE),
"the bound-shard value must resolve (I1)"
);
}
#[test]
fn bind_routes_per_thread() {
let region = Arc::new(ShardedRegion::<u64>::with_shards(4, 16));
let handles = Arc::new(Mutex::new(Vec::<ShardedHandle<u64>>::new()));
std::thread::scope(|s| {
for shard in 0u16..4 {
let region = Arc::clone(®ion);
let handles = Arc::clone(&handles);
s.spawn(move || {
assert!(region.bind_current_thread_to_shard(shard));
let h = region.insert(u64::from(shard)).expect("capacity");
assert_eq!(
h.shard(),
shard,
"worker {shard} must route to its bound shard"
);
handles.lock().unwrap().push(h);
});
}
});
let handles = handles.lock().unwrap();
assert_eq!(handles.len(), 4, "one handle per worker");
let mut shards: Vec<u16> = handles.iter().map(|h| h.shard()).collect();
shards.sort_unstable();
assert_eq!(
shards,
vec![0, 1, 2, 3],
"each worker bound a distinct shard"
);
for h in handles.iter() {
assert_eq!(region.get_cloned(*h), Some(u64::from(h.shard())));
}
assert_eq!(region.len(), 4, "accounting: 4 live values");
}
#[test]
fn out_of_range_bind_rejected_no_panic() {
let region = ShardedRegion::<u64>::with_shards(3, 16);
let n = region.shard_count();
std::thread::scope(|s| {
s.spawn(|| {
assert!(!region.bind_current_thread_to_shard(u16::try_from(n).unwrap()));
assert!(!region.bind_current_thread_to_shard(u16::MAX));
let h = region.insert(1_u64).expect("capacity");
assert!(
usize::from(h.shard()) < n,
"rejected bind must not leave an out-of-range binding"
);
})
.join()
.expect("must not panic");
});
}
#[test]
fn pinned_runner_inserts_resolve_and_accounting_holds() {
let region = Arc::new(ShardedRegion::<u64>::with_shards(4, 64));
let runner = match PinnedRunner::with_workers(®ion, 4) {
Some(r) => r,
None => {
eprintln!("skip: core_affinity::get_core_ids returned None on this host");
return;
}
};
let workers = runner.worker_count();
assert!((1..=4).contains(&workers));
let inserted = Arc::new(AtomicUsize::new(0));
let per_worker_values = Arc::new(Mutex::new(Vec::<(u16, u64, ShardedHandle<u64>)>::new()));
let cap = 8usize; runner.run_arc(®ion, |shard_id, region| {
let mut local = Vec::new();
for v in 0..u64::try_from(cap).unwrap() {
let h = region.insert(v).expect("shard has capacity");
assert_eq!(
h.shard(),
shard_id,
"runner worker must route to its bound shard"
);
assert_eq!(region.get_cloned(h), Some(v), "fresh insert resolves (I1)");
local.push((shard_id, v, h));
}
inserted.fetch_add(cap, Ordering::Relaxed);
per_worker_values.lock().unwrap().extend(local);
});
assert_eq!(inserted.load(Ordering::Relaxed), workers * cap);
assert_eq!(
region.len(),
workers * cap,
"len must equal total live inserts"
);
let all = per_worker_values.lock().unwrap();
for (shard_id, v, h) in all.iter() {
assert_eq!(h.shard(), *shard_id);
assert_eq!(
region.get_cloned(*h),
Some(*v),
"live value must resolve post-join"
);
}
}
#[test]
fn pinned_runner_then_cross_thread_remove() {
let region = Arc::new(ShardedRegion::<u64>::with_shards(4, 64));
let runner = match PinnedRunner::with_workers(®ion, 2) {
Some(r) => r,
None => {
eprintln!("skip: core_affinity::get_core_ids returned None on this host");
return;
}
};
let workers = runner.worker_count();
let handles = Arc::new(Mutex::new(Vec::<ShardedHandle<u64>>::new()));
runner.run_arc(®ion, |shard_id, region| {
let h = region.insert(u64::from(shard_id)).expect("capacity");
assert_eq!(h.shard(), shard_id);
handles.lock().unwrap().push(h);
});
assert_eq!(region.len(), workers);
let hs = handles.lock().unwrap().clone();
let mut removed = 0usize;
for h in &hs {
if region.remove(*h) {
removed += 1;
}
assert!(!region.remove(*h), "double remove must be a no-op false");
}
assert_eq!(
removed, workers,
"every worker's value was live and removable"
);
assert_eq!(region.len(), 0, "all removed → region empty");
}