#![cfg(feature = "parallel")]
use std::sync::atomic::{AtomicBool, Ordering};
use strided_basic::{map_into, Exec, Pool, StridedArray};
const WORKER_PREFIX: &str = "strided-small-worker";
fn named_pool() -> Pool {
Pool::builder()
.threads(4)
.thread_name(|i| format!("{WORKER_PREFIX}-{i}"))
.build()
.unwrap()
}
fn on_pool_worker() -> bool {
std::thread::current()
.name()
.is_some_and(|name| name.starts_with(WORKER_PREFIX))
}
#[test]
fn small_operation_does_not_enter_the_target_pool() {
let pool = named_pool();
let source = StridedArray::<usize>::from_fn_col_major(&[8, 8], |index| index[0] + 8 * index[1]);
let mut destination = StridedArray::<usize>::col_major(&[8, 8]);
let caller = std::thread::current().id();
let left_caller = AtomicBool::new(false);
let saw_pool_worker = AtomicBool::new(false);
let exec = Exec::new(&pool, 4).unwrap();
let root = exec.ledger_root().unwrap();
map_into(
&mut destination.view_mut(),
&source.view(),
|value| {
if std::thread::current().id() != caller {
left_caller.store(true, Ordering::SeqCst);
}
if on_pool_worker() {
saw_pool_worker.store(true, Ordering::SeqCst);
}
value + 1
},
exec,
)
.unwrap();
assert!(!left_caller.load(Ordering::SeqCst));
assert!(!saw_pool_worker.load(Ordering::SeqCst));
assert_eq!(threadteam::ledger::installs(root), 0);
for index in 0..64 {
assert_eq!(destination.data()[index], index + 1);
}
}
#[test]
fn small_operation_inside_a_worker_of_the_target_pool_runs_in_place() {
let pool = named_pool();
let source = StridedArray::<usize>::from_fn_col_major(&[8, 8], |index| index[0] + 8 * index[1]);
let exec = Exec::new(&pool, 4).unwrap();
let root = exec.ledger_root().unwrap();
let left_worker = AtomicBool::new(false);
let saw_worker = AtomicBool::new(false);
exec.fork(2, |_job, child| {
assert!(
on_pool_worker(),
"a fork job of a pooled token runs on a worker"
);
saw_worker.store(true, Ordering::SeqCst);
let worker = std::thread::current().id();
let mut destination = StridedArray::<usize>::col_major(&[8, 8]);
map_into(
&mut destination.view_mut(),
&source.view(),
|value| {
if std::thread::current().id() != worker {
left_worker.store(true, Ordering::SeqCst);
}
value + 1
},
child,
)
.unwrap();
for index in 0..64 {
assert_eq!(destination.data()[index], index + 1);
}
});
assert!(saw_worker.load(Ordering::SeqCst));
assert!(!left_worker.load(Ordering::SeqCst));
assert_eq!(threadteam::ledger::installs(root), 1);
}