diff --git a/scripts/bench-ab.sh b/scripts/bench-ab.sh index 36b8da6..3ed5767 100755 --- a/scripts/bench-ab.sh +++ b/scripts/bench-ab.sh @@ -10,6 +10,9 @@ # taskset -c 0-63 scripts/bench-ab.sh a3a9320 feat/wait-openmp 20 \ # --scenario acquire --widths 64 --calls 40 --impl threadteam # +# HEAD2= adds a third side, labelled `head2`, built and run in every round; the order of the +# sides rotates by round. +# # HEAD_VARIANTS adds head runs with extra arguments in every round, `;`-separated, each named by # its arguments, e.g. HEAD_VARIANTS='--spin-polls 300000;--own-hold-ms 100;--pressure-events 0'. # OMP=1 also runs the libgomp driver (benches/omp/wait_omp.c) in every round, with the same @@ -21,7 +24,7 @@ csv=${CSV:-docs/design/waiting-data/$host-ab.csv} log=${LOG:-docs/design/waiting-data/$host-ab.log} mkdir -p "$(dirname "$csv")" work=$(mktemp -d) -trap 'git worktree remove --force "$work/base" 2>/dev/null; git worktree remove --force "$work/head" 2>/dev/null; rm -rf "$work"' EXIT +trap 'for d in base head head2; do git worktree remove --force "$work/$d" 2>/dev/null; done; rm -rf "$work"' EXIT build() { # $1 = rev, $2 = dir name; prints the bench binary git worktree add -q --detach "$work/$2" "$1" (cd "$work/$2" && cargo build -j "${JOBS:-4}" --release --bench wait_scenarios >/dev/null 2>&1) @@ -29,6 +32,8 @@ build() { # $1 = rev, $2 = dir name; prints the bench binary } bin_base=$(build "$base" base) bin_head=$(build "$head" head) +bin_head2="" +[[ -n ${HEAD2:-} ]] && bin_head2=$(build "$HEAD2" head2) omp="" if [[ ${OMP:-0} == 1 ]]; then gcc -O2 -fopenmp "$work/head/benches/omp/wait_omp.c" -o "$work/wait_omp" @@ -42,12 +47,20 @@ arg() { # value of bench option $1 in "$@" after it, or $2 omp_scenario=$(arg --scenario all "$@"); omp_widths=$(arg --widths "$(nproc)" "$@") omp_calls=$(arg --calls 200 "$@"); omp_load=$(arg --load "" "$@") IFS=';' read -r -a variants <<<"${HEAD_VARIANTS:-}" -echo "base $(git rev-parse --short "$base") head $(git rev-parse --short "$head") rounds $rounds args $*" | tee -a "$log" +echo "base $(git rev-parse --short "$base") head $(git rev-parse --short "$head")${HEAD2:+ head2 $(git rev-parse --short "$HEAD2")} rounds $rounds args $*" | tee -a "$log" for r in $(seq 1 "$rounds"); do - order=(base head); (( r % 2 == 0 )) && order=(head base) + if [[ -n $bin_head2 ]]; then + case $(( r % 3 )) in + 1) order=(base head head2) ;; 2) order=(head head2 base) ;; *) order=(head2 base head) ;; + esac + else + order=(base head); (( r % 2 == 0 )) && order=(head base) + fi for side in "${order[@]}"; do if [[ $side == base ]]; then "$bin_base" --label "base#$r" --csv "$csv" "$@" >>"$log" + elif [[ $side == head2 ]]; then + "$bin_head2" --label "head2#$r" --csv "$csv" "$@" >>"$log" else "$bin_head" --label "head#$r" --csv "$csv" "$@" >>"$log" if [[ -n $omp ]]; then @@ -66,12 +79,18 @@ for r in $(seq 1 "$rounds"); do done echo "rows in $csv, waits in $log" python3 - "$csv" <<'PY' -import csv, collections, statistics, sys -rows = list(csv.DictReader(open(sys.argv[1]))) +import collections, statistics, sys +# `param` may hold commas (`work=5us,sections=200`), unquoted: the fields around it are fixed. cells = collections.defaultdict(list) -for r in rows: - name = r["label"].split("#")[0] - cells[(r["scenario"], r["width"], r["param"], r["metric"], name)].append(float(r["median_us"])) +for line in open(sys.argv[1]).read().splitlines()[1:]: + f = line.split(",") + host, label, cpus, load, scenario, impl, width = f[:7] + metric, median = f[-5], f[-4] + param = ",".join(f[7:-5]) + name = label.split("#")[0] + if impl not in ("threadteam", "libgomp"): + name = f"{name} ({impl})" + cells[(scenario, width, param, metric, name)].append(float(median)) print(f"{'scenario':<9} {'width':>5} {'param':<24} {'metric':<6} {'side':<40} {'n':>3} {'median of medians us':>21}") for k in sorted(cells): v = cells[k] diff --git a/src/awake.rs b/src/awake.rs index 0e6aa40..25334a4 100644 --- a/src/awake.rs +++ b/src/awake.rs @@ -354,7 +354,7 @@ impl Awake { let acq = Arc::new(Acq::new(missing)); for _ in 0..missing { let (aw, a) = (Arc::clone(self), Arc::clone(&acq)); - pool.inner.spawn(move || claim_job(&aw, &a)); + pool.inner().spawn(move || claim_job(&aw, &a)); } let deadline = Instant::now() + self.acquire_timeout; let mut spins = 0; diff --git a/src/lib.rs b/src/lib.rs index 326820b..01746d0 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -73,7 +73,7 @@ pub use park::{Barrier, Completion, Waiter}; pub use policy::{raise_current_thread_qos, Qos, WaitOverride, WaitPolicy}; #[doc(hidden)] pub use pool::WaitCounters; -pub use pool::{Pool, PoolBuilder}; +pub use pool::{spawned_workers, Pool, PoolBuilder}; pub use reserve::{IntoTeamError, Team}; pub use team::{Cancelled, Participant, TeamExec}; diff --git a/src/pool.rs b/src/pool.rs index 8bc2385..b8ebdd1 100644 --- a/src/pool.rs +++ b/src/pool.rs @@ -61,11 +61,25 @@ const SPIN_POLLS_DEFAULT: u32 = if cfg!(target_arch = "aarch64") { /// caller), where each call acquires its members with /// [`Exec::team`](crate::Exec::team) or uses plain tokens. /// -/// Dropping a `Pool` (or the `Team` that owns it) shuts its workers down and -/// joins them: when `drop` returns, every worker thread has exited, so the -/// thread-local destructors of the workers (including state set up by an -/// [`on_start`](PoolBuilder::on_start) hook) have already run. Without the -/// `parallel` feature there are no workers and nothing is joined. +/// Dropping a `Pool` (or the `Team` that owns it) ends its team members. What +/// happens to the worker threads depends on how the pool was built: +/// +/// * A pool with a [`thread_name`](PoolBuilder::thread_name), an +/// [`on_start`](PoolBuilder::on_start) hook, a +/// [`stack_size`](PoolBuilder::stack_size) or a [`qos`](PoolBuilder::qos) of +/// its own has **dedicated** workers: the drop shuts them down and joins +/// them, so when `drop` returns every worker thread has exited and its +/// thread-local destructors (including state set up by the hook) have run. +/// * A pool with the default thread settings has **shared** workers: the drop +/// waits until its members left, then keeps the idle workers in a +/// process-wide cache, keyed by width. The next `build` of a default pool of +/// the same width takes them instead of starting threads, and gives each +/// worker the calling thread's CPU set. A cached set of workers belongs to +/// one pool at a time. The cache holds at most as many threads as the +/// larger of the available parallelism and the widest pool built so far; +/// the oldest entries beyond that are joined. +/// +/// Without the `parallel` feature there are no workers and nothing is joined. /// /// A pool dropped from one of its own workers cannot wait for itself: that drop /// signals shutdown and returns without joining, so the workers exit shortly @@ -82,17 +96,26 @@ const SPIN_POLLS_DEFAULT: u32 = if cfg!(target_arch = "aarch64") { /// # Ok::<(), threadteam::Error>(()) /// ``` pub struct Pool { - // Field order matters: `inner` is dropped (shutdown signalled) before - // `joiner` waits for the threads. + /// `None` only inside `drop`, which moves the workers to the cache. #[cfg(feature = "parallel")] - pub(crate) inner: rayon::ThreadPool, + workers: Option, + /// Default thread settings: the workers go to the cache on drop. #[cfg(feature = "parallel")] - _joiner: Joiner, + shared: bool, pub(crate) width: NonZeroUsize, pub(crate) awake: Arc, } impl Pool { + /// The private Rayon pool. + #[cfg(feature = "parallel")] + pub(crate) fn inner(&self) -> &rayon::ThreadPool { + match &self.workers { + Some(w) => &w.inner, + None => unreachable!("the workers leave the pool only in drop"), + } + } + /// Starts configuring a pool. /// /// # Examples @@ -225,9 +248,101 @@ impl Drop for Pool { fn drop(&mut self) { // Idle members leave their loops so that the workers can be joined. self.awake.shutdown.store(true, Ordering::Release); + #[cfg(feature = "parallel")] + if let Some(workers) = self.workers.take() { + // A pool dropped on one of its own workers cannot wait for that + // worker's member loop: its workers are not reused. + if self.shared && workers.inner.current_thread_index().is_none() { + // No member loop of this pool may run on the threads when the + // next pool takes them. + let mut spins = 0u32; + while self.awake.resident() > 0 { + if spins < 64 { + spins += 1; + std::thread::yield_now(); + } else { + std::thread::sleep(Duration::from_micros(50)); + } + } + cache::put(workers); + } + } } } +/// A Rayon pool and its threads. Field order matters: `inner` is dropped +/// (shutdown signalled) before `_joiner` waits for the threads. +#[cfg(feature = "parallel")] +struct Workers { + inner: rayon::ThreadPool, + _joiner: Joiner, +} + +/// Idle shared workers, kept between pools of the same width. +#[cfg(feature = "parallel")] +mod cache { + use super::Workers; + use core::sync::atomic::{AtomicUsize, Ordering}; + use std::sync::{Mutex, PoisonError}; + + static IDLE: Mutex> = Mutex::new(Vec::new()); + /// The widest shared pool built so far. + static WIDEST: AtomicUsize = AtomicUsize::new(0); + + /// Takes the idle workers of a pool of `width` threads, if any are cached. + pub(super) fn take(width: usize) -> Option { + WIDEST.fetch_max(width, Ordering::Relaxed); + let mut idle = IDLE.lock().unwrap_or_else(PoisonError::into_inner); + let at = idle + .iter() + .position(|w| w.inner.current_num_threads() == width)?; + Some(idle.remove(at)) + } + + /// Keeps `workers` for a later pool; joins the oldest entries beyond the + /// cap (the larger of the available parallelism and the widest pool). + pub(super) fn put(workers: Workers) { + let cap = std::thread::available_parallelism() + .map_or(1, core::num::NonZeroUsize::get) + .max(WIDEST.load(Ordering::Relaxed)); + let evicted: Vec = { + let mut idle = IDLE.lock().unwrap_or_else(PoisonError::into_inner); + idle.push(workers); + let mut total: usize = idle.iter().map(|w| w.inner.current_num_threads()).sum(); + let mut n = 0; + while total > cap && n + 1 < idle.len() { + total -= idle[n].inner.current_num_threads(); + n += 1; + } + idle.drain(..n).collect() + }; + // Joined outside the lock. + drop(evicted); + } + + /// The widest shared pool built so far (tests). + #[cfg(test)] + pub(super) fn widest() -> usize { + WIDEST.load(Ordering::Relaxed) + } + + /// Threads currently cached (tests). + #[cfg(test)] + pub(super) fn cached_threads() -> usize { + let idle = IDLE.lock().unwrap_or_else(PoisonError::into_inner); + idle.iter().map(|w| w.inner.current_num_threads()).sum() + } +} + +/// Worker threads started by shared and dedicated pools since the process +/// started; a diagnostic for the reuse tests and benchmarks. +#[doc(hidden)] +pub fn spawned_workers() -> usize { + SPAWNED.load(Ordering::Relaxed) +} + +static SPAWNED: core::sync::atomic::AtomicUsize = core::sync::atomic::AtomicUsize::new(0); + /// The worker threads of a pool, joined when it is dropped. #[cfg(feature = "parallel")] struct Joiner { @@ -617,10 +732,47 @@ impl PoolBuilder { } #[cfg(feature = "parallel")] - fn build_checked(self) -> Result { + fn build_checked(mut self) -> Result { let n = self .threads .unwrap_or_else(|| std::thread::available_parallelism().map_or(1, NonZeroUsize::get)); + let shared = self.thread_name.is_none() + && self.on_start.is_none() + && self.stack_size == DEFAULT_STACK_BYTES + && self.qos.is_none(); + let workers = match shared.then(|| cache::take(n)).flatten() { + Some(workers) => { + // A reused worker takes the caller's CPU set, as a thread the + // caller spawned would. + if let Some(mask) = crate::pin::read_here() { + workers.inner.broadcast(|_| { + crate::pin::set_here(&mask); + }); + } + workers + } + None => self.spawn(n)?, + }; + let width = + NonZeroUsize::new(workers.inner.current_num_threads()).unwrap_or(NonZeroUsize::MIN); + Ok(Pool { + workers: Some(workers), + shared, + width, + awake: Arc::new(Awake::new( + width.get(), + self.wait, + self.acquire_timeout, + self.costs, + self.pin_participants, + self.spin_polls, + )), + }) + } + + /// Starts `n` new worker threads. + #[cfg(feature = "parallel")] + fn spawn(&mut self, n: usize) -> Result { let handles = std::sync::Arc::new(std::sync::Mutex::new(Vec::new())); let sink = std::sync::Arc::clone(&handles); let qos = self.qos.or(Qos::PLATFORM_DEFAULT); @@ -635,6 +787,7 @@ impl PoolBuilder { if let Some(size) = thread.stack_size() { b = b.stack_size(size); } + SPAWNED.fetch_add(1, Ordering::Relaxed); let handle = b.spawn(move || { if let Some(q) = qos { q.apply_to_current_thread(); @@ -647,10 +800,10 @@ impl PoolBuilder { } Ok(()) }); - if let Some(name) = self.thread_name { + if let Some(name) = self.thread_name.take() { builder = builder.thread_name(name); } - if let Some(hook) = self.on_start { + if let Some(hook) = self.on_start.take() { builder = builder.start_handler(hook); } // On failure `builder` drops its handler, but threads already started @@ -663,19 +816,9 @@ impl PoolBuilder { return Err(Error::PoolBuild(e.to_string())); } }; - let width = NonZeroUsize::new(inner.current_num_threads()).unwrap_or(NonZeroUsize::MIN); - Ok(Pool { + Ok(Workers { inner, _joiner: Joiner { handles }, - width, - awake: Arc::new(Awake::new( - width.get(), - self.wait, - self.acquire_timeout, - self.costs, - self.pin_participants, - self.spin_polls, - )), }) } @@ -721,7 +864,7 @@ mod tests { let last = Arc::clone(&pool); let (gate_tx, gate_rx) = mpsc::channel::<()>(); let (done_tx, done_rx) = mpsc::channel::<()>(); - pool.inner.spawn(move || { + pool.inner().spawn(move || { gate_rx.recv().unwrap(); drop(last); // the last reference: runs `Pool::drop` on this worker done_tx.send(()).unwrap(); @@ -732,6 +875,22 @@ mod tests { .recv_timeout(Duration::from_secs(10)) .expect("drop on a worker hung"); } + + /// The cache never holds more threads than its cap, whatever the number + /// of pools dropped. + #[test] + fn the_cache_keeps_at_most_its_cap_of_threads() { + let cap = std::thread::available_parallelism().map_or(1, NonZeroUsize::get); + let pools: Vec = (0..cap + 2) + .map(|_| Pool::builder().threads(1).build().unwrap()) + .collect(); + drop(pools); + // Other tests of this binary build wider pools concurrently, which + // can only raise the cap. + let cached = cache::cached_threads(); + let cap = cap.max(cache::widest()); + assert!(cached <= cap, "{cached} cached threads, cap {cap}"); + } } /// Counters of [`Pool::wait_counters`] (a diagnostic for the wait benchmarks). diff --git a/src/sched.rs b/src/sched.rs index b1f8a21..9861edd 100644 --- a/src/sched.rs +++ b/src/sched.rs @@ -10,7 +10,7 @@ use crate::track::Track; /// Whether the calling thread is a worker of `pool`. #[cfg(feature = "parallel")] pub(crate) fn in_pool(pool: &Pool) -> bool { - pool.inner.current_thread_index().is_some() + pool.inner().current_thread_index().is_some() } #[cfg(not(feature = "parallel"))] @@ -31,7 +31,7 @@ pub(crate) fn in_pool(_pool: &Pool) -> bool { /// re-pins at the next pin section. #[cfg(feature = "parallel")] pub(crate) fn unpin_worker(pool: &Pool) { - if let Some(worker) = pool.inner.current_thread_index() { + if let Some(worker) = pool.inner().current_thread_index() { pool.awake.unpin_member(worker); } } @@ -55,7 +55,7 @@ pub(crate) fn run_tasks(pool: &Pool, tasks: usize, track: & join_tree(0, tasks, &task); } else { track.note_install(); - pool.inner.install(|| join_tree(0, tasks, &task)); + pool.inner().install(|| join_tree(0, tasks, &task)); } } @@ -84,7 +84,7 @@ pub(crate) fn enter(pool: &Pool, track: &Track, f: impl FnOnce() -> R + f() } else { track.note_install(); - pool.inner.install(f) + pool.inner().install(f) } } @@ -98,7 +98,7 @@ pub(crate) fn broadcast(pool: &Pool, track: &Track, budget: usize, f: impl Fn(us // would deadlock a foreign-pool caller that steals a second broadcast onto // its own stack while waiting. track.note_install(); - pool.inner.broadcast(|ctx| { + pool.inner().broadcast(|ctx| { let i = ctx.index(); if i < budget { let _live = track.task(); diff --git a/tests/rayon_reuse.rs b/tests/rayon_reuse.rs new file mode 100644 index 0000000..e5686e4 --- /dev/null +++ b/tests/rayon_reuse.rs @@ -0,0 +1,186 @@ +//! Reuse of the private Rayon pool between pools of the same width (the +//! comparison experiment against own worker threads, tensor4all/faer#107). +#![cfg(feature = "parallel")] + +use std::collections::HashSet; +use std::sync::atomic::{AtomicUsize, Ordering}; +use std::sync::Mutex; +use std::thread::ThreadId; +use std::time::Duration; +use threadteam::{Exec, Pool}; + +/// A hang is a failure: abort the whole process if the test body stalls. +fn watchdog() -> impl Drop { + struct Stop(std::sync::mpsc::Sender<()>); + impl Drop for Stop { + fn drop(&mut self) { + let _ = self.0.send(()); + } + } + let (tx, rx) = std::sync::mpsc::channel::<()>(); + std::thread::spawn(move || { + if rx.recv_timeout(Duration::from_secs(60)).is_err() { + eprintln!("watchdog: test stalled"); + std::process::exit(101); + } + }); + Stop(tx) +} + +/// The cache and the spawn counter are process wide: the tests of this file +/// take turns. +static SERIAL: Mutex<()> = Mutex::new(()); + +fn worker_ids(pool: &Pool) -> HashSet { + let ids = Mutex::new(HashSet::new()); + Exec::new(pool, pool.width().get()) + .unwrap() + .broadcast(|_| { + ids.lock().unwrap().insert(std::thread::current().id()); + }) + .unwrap(); + ids.into_inner().unwrap() +} + +#[test] +fn a_pool_built_after_another_was_dropped_reuses_its_workers() { + let _w = watchdog(); + let _turn = SERIAL.lock().unwrap_or_else(|e| e.into_inner()); + let width = 4; + let build = || { + Pool::builder() + .threads(width) + .acquire_timeout(Duration::from_secs(5)) + .build() + .unwrap() + }; + let first = build(); + Exec::new(&first, width).unwrap().fork(width, |_, _| {}); + let ids = worker_ids(&first); + drop(first); + for round in 0..5 { + let before = threadteam::spawned_workers(); + let pool = build(); + let n = AtomicUsize::new(0); + Exec::new(&pool, width).unwrap().fork(width, |_, _| { + n.fetch_add(1, Ordering::SeqCst); + }); + assert_eq!(n.load(Ordering::SeqCst), width); + assert_eq!(worker_ids(&pool), ids, "round {round}: other threads"); + // A team on the reused workers, as faer's `with_team` takes one. + let team = pool.into_team(width).expect("idle reused workers"); + assert_eq!(team.size().get(), width); + drop(team); + assert_eq!( + threadteam::spawned_workers() - before, + 0, + "round {round}: a pool of the same width spawned threads" + ); + } +} + +#[test] +fn two_live_pools_of_one_width_never_share_workers() { + let _w = watchdog(); + let _turn = SERIAL.lock().unwrap_or_else(|e| e.into_inner()); + drop(Pool::builder().threads(2).build().unwrap()); + let a = Pool::builder().threads(2).build().unwrap(); + let b = Pool::builder().threads(2).build().unwrap(); + let (ia, ib) = (worker_ids(&a), worker_ids(&b)); + assert_eq!(ia.len(), 2); + assert_eq!(ib.len(), 2); + assert!(ia.is_disjoint(&ib), "{ia:?} {ib:?}"); +} + +#[test] +fn a_pool_with_hooks_gets_dedicated_workers_and_joins_them() { + let _w = watchdog(); + let _turn = SERIAL.lock().unwrap_or_else(|e| e.into_inner()); + let started = std::sync::Arc::new(AtomicUsize::new(0)); + for _ in 0..2 { + let s = std::sync::Arc::clone(&started); + let before = threadteam::spawned_workers(); + let pool = Pool::builder() + .threads(2) + .thread_name(|i| format!("dedicated-{i}")) + .on_start(move |_| { + s.fetch_add(1, Ordering::SeqCst); + }) + .build() + .unwrap(); + let names = Mutex::new(Vec::new()); + Exec::new(&pool, 2).unwrap().fork(2, |_, _| { + names + .lock() + .unwrap() + .push(std::thread::current().name().map(String::from)); + }); + drop(pool); + assert_eq!(threadteam::spawned_workers() - before, 2); + assert!(names + .into_inner() + .unwrap() + .iter() + .all(|n| n.as_deref().is_some_and(|n| n.starts_with("dedicated-")))); + } + // The drop joined the workers: their hooks ran once each, never again. + assert_eq!(started.load(Ordering::SeqCst), 4); +} + +/// A reused worker takes the CPU set of the caller that builds the pool, as a +/// thread that caller spawned would, not the set of whoever spawned it first. +#[cfg(target_os = "linux")] +#[test] +fn a_reused_worker_takes_the_building_callers_cpu_set() { + let _w = watchdog(); + let _turn = SERIAL.lock().unwrap_or_else(|e| e.into_inner()); + fn cpus() -> Vec { + unsafe { + let mut set: libc::cpu_set_t = std::mem::zeroed(); + libc::sched_getaffinity(0, std::mem::size_of::(), &mut set); + (0..libc::CPU_SETSIZE as usize) + .filter(|&c| libc::CPU_ISSET(c, &set)) + .collect() + } + } + fn set(cpus: &[usize]) -> bool { + unsafe { + let mut s: libc::cpu_set_t = std::mem::zeroed(); + libc::CPU_ZERO(&mut s); + for &c in cpus { + libc::CPU_SET(c, &mut s); + } + libc::sched_setaffinity(0, std::mem::size_of::(), &s) == 0 + } + } + let all = cpus(); + if all.len() < 4 { + eprintln!("skipped: needs 4 allowed CPUs"); + return; + } + // Run on a helper thread so the test thread's mask is never changed. + std::thread::spawn(move || { + let width = 3; + assert!(set(&all[..2])); + let first = Pool::builder().threads(width).build().unwrap(); + let ids = worker_ids(&first); + drop(first); + assert!(set(&all[2..4])); + let pool = Pool::builder().threads(width).build().unwrap(); + assert_eq!(worker_ids(&pool), ids, "the workers were not reused"); + let masks = Mutex::new(Vec::new()); + Exec::new(&pool, width) + .unwrap() + .broadcast(|_| masks.lock().unwrap().push(cpus())) + .unwrap(); + for m in masks.into_inner().unwrap() { + assert_eq!( + m, + all[2..4].to_vec(), + "a reused worker kept another caller's CPU set" + ); + } + }) + .join() + .unwrap(); +}