Skip to main content

threadteam/
exec.rs

1//! The splittable placement token.
2
3use crate::awake::Awake;
4#[cfg(feature = "parallel")]
5use crate::awake::LendGuard;
6use crate::cost::{pays_at, width_for_by, DispatchKind};
7use crate::error::{Error, Refused};
8use crate::policy::{WaitOverride, WaitPolicy};
9use crate::pool::Pool;
10use crate::reserve::LeaderGuard;
11use crate::sched;
12use crate::split;
13use crate::team::{split_raw, spread_raw, Plan, TeamExec};
14use crate::track::Track;
15use core::fmt;
16use core::num::NonZeroUsize;
17use core::ops::Range;
18use core::sync::atomic::{AtomicUsize, Ordering};
19use core::time::Duration;
20
21/// `budget` slots of a [`Pool`], or the calling thread alone.
22///
23/// An `Exec` is the only way kernels start parallel work. It is a *token*:
24///
25/// * It is neither `Clone` nor `Copy`, and every parallel entry consumes it, so
26///   a kernel cannot run two phases at full width at the same time. Use
27///   [`reborrow`](Exec::reborrow) to run several phases in sequence.
28/// * Only [`Exec::new`] mints a token, and it needs the `&Pool`; a token never
29///   hands its pool back out, so a kernel cannot mint a fresh full-budget token.
30/// * [`fork`](Exec::fork) divides the budget among its tasks (shares differ by
31///   at most one and sum to the budget), so the number of tasks running at once
32///   under one root token is at most its budget at every nesting depth.
33///
34/// A parent waiting for its children keeps no slot.
35///
36/// # Placement
37///
38/// Serial, budget one, or one task: the work runs inline on the calling thread
39/// and the scheduler is never called, so no global pool is started. The caller
40/// is a worker of the pool: the tasks run as a join tree on the caller, nothing is
41/// installed. The caller is a plain thread or a worker of a different pool: the
42/// pool is entered once and the caller waits; the tasks run on the target pool.
43/// No thread-local state is involved at any point.
44///
45/// # Examples
46///
47/// ```
48/// use std::sync::atomic::{AtomicUsize, Ordering};
49/// use threadteam::{Exec, Pool};
50///
51/// let pool = Pool::builder().threads(2).acquire_timeout(std::time::Duration::from_secs(10)).build()?;
52/// let exec = Exec::new(&pool, 2)?;
53/// let sum = AtomicUsize::new(0);
54/// exec.for_ranges(100, 1.try_into().unwrap(), |r| {
55///     sum.fetch_add(r.sum::<usize>(), Ordering::Relaxed);
56/// });
57/// assert_eq!(sum.into_inner(), 4950);
58/// # Ok::<(), threadteam::Error>(())
59/// ```
60///
61/// A token cannot be duplicated:
62///
63/// ```compile_fail
64/// fn needs_clone<T: Clone>() {}
65/// needs_clone::<threadteam::Exec<'static>>();
66/// ```
67///
68/// ```compile_fail
69/// fn needs_copy<T: Copy>() {}
70/// needs_copy::<threadteam::Exec<'static>>();
71/// ```
72///
73/// ```compile_fail
74/// let exec = threadteam::Exec::serial();
75/// exec.fork(1, |_, _| {});
76/// exec.fork(1, |_, _| {}); // `fork` consumed the token
77/// ```
78#[must_use = "an Exec is a budget of execution slots; pass it to a kernel"]
79pub struct Exec<'p> {
80    pool: Option<&'p Pool>,
81    budget: NonZeroUsize,
82    /// Produced by `fork` (directly or through `reborrow`).
83    child: bool,
84    track: Track,
85    /// Authority over reserved members, for tokens minted from a [`Team`](crate::Team).
86    team: Option<TeamAuth<'p>>,
87    /// Held by the root token of a `Team`: one leader at a time.
88    _leader: Option<LeaderGuard<'p>>,
89}
90
91/// The reserved members a token may dispatch to; `ids.len() + 1 == budget`.
92#[derive(Clone, Copy)]
93pub(crate) struct TeamAuth<'p> {
94    pub(crate) awake: &'p Awake,
95    pub(crate) ids: &'p [usize],
96    pub(crate) policy: WaitPolicy,
97    /// The root scope of the token is oversubscribed (see [`TeamExec`]).
98    /// Fixed when the root token is minted and carried into `reborrow`,
99    /// `limit`, `fork` children and every section.
100    pub(crate) over: bool,
101}
102
103impl Exec<'static> {
104    /// A token that runs everything inline on the calling thread.
105    ///
106    /// It never touches a scheduler, so using it cannot start any pool.
107    ///
108    /// # Examples
109    ///
110    /// ```
111    /// use threadteam::Exec;
112    ///
113    /// let exec = Exec::serial();
114    /// assert!(exec.is_serial());
115    /// let here = std::thread::current().id();
116    /// exec.fork(3, |_, _| assert_eq!(std::thread::current().id(), here));
117    /// ```
118    pub const fn serial() -> Exec<'static> {
119        Exec {
120            pool: None,
121            budget: NonZeroUsize::MIN,
122            child: false,
123            track: Track::NONE,
124            team: None,
125            _leader: None,
126        }
127    }
128}
129
130impl<'p> Exec<'p> {
131    /// A token of `budget` slots of `pool`, clamped to the pool's width.
132    ///
133    /// This is the only place a budget is clamped.
134    ///
135    /// # Errors
136    ///
137    /// [`Error::ZeroBudget`] when `budget` is zero.
138    ///
139    /// # Examples
140    ///
141    /// ```
142    /// use threadteam::{Error, Exec, Pool};
143    ///
144    /// let pool = Pool::builder().threads(2).acquire_timeout(std::time::Duration::from_secs(10)).build()?;
145    /// let exec = Exec::new(&pool, 64)?;
146    /// assert_eq!(exec.budget(), pool.width());
147    /// assert_eq!(Exec::new(&pool, 0).unwrap_err(), Error::ZeroBudget);
148    /// # Ok::<(), Error>(())
149    /// ```
150    pub fn new(pool: &'p Pool, budget: usize) -> Result<Exec<'p>, Error> {
151        let budget = NonZeroUsize::new(budget).ok_or(Error::ZeroBudget)?;
152        Ok(Exec {
153            pool: Some(pool),
154            budget: budget.min(pool.width()),
155            child: false,
156            track: Track::root(),
157            team: None,
158            _leader: None,
159        })
160    }
161
162    /// A root token of a [`Team`](crate::Team): `budget` is already clamped and
163    /// `ids.len() + 1 == budget`.
164    pub(crate) fn from_team(
165        pool: &'p Pool,
166        auth: TeamAuth<'p>,
167        guard: LeaderGuard<'p>,
168    ) -> Exec<'p> {
169        Exec {
170            pool: Some(pool),
171            budget: NonZeroUsize::new(auth.ids.len() + 1).unwrap_or(NonZeroUsize::MIN),
172            child: false,
173            track: Track::root(),
174            team: Some(auth),
175            _leader: Some(guard),
176        }
177    }
178
179    /// The number of slots this token stands for.
180    ///
181    /// # Examples
182    ///
183    /// ```
184    /// assert_eq!(threadteam::Exec::serial().budget().get(), 1);
185    /// ```
186    pub fn budget(&self) -> NonZeroUsize {
187        self.budget
188    }
189
190    /// The cost of one parallel section of `width` participants started now on
191    /// this token: the steady state (sections back to back, shares that carry
192    /// work). [`Duration::MAX`] when no section is possible: a serial token, or
193    /// `width < 2`. `width` is clamped to the budget.
194    ///
195    /// The kind of token is decided at call time:
196    ///
197    /// * the token has team authority (it comes from a [`Team`](crate::Team), or
198    ///   is a child of such a token): [`DispatchKind::Team`];
199    /// * a plain token whose caller is a worker of its pool:
200    ///   [`DispatchKind::InPool`];
201    /// * a plain token whose caller is outside the pool: [`DispatchKind::Outside`].
202    ///
203    /// The numbers come from the pool's [`DispatchCosts`](crate::DispatchCosts) (see
204    /// [`PoolBuilder::dispatch_costs`](crate::PoolBuilder::dispatch_costs)). The
205    /// call allocates nothing and reads no thread-local state of this crate.
206    /// The answer depends on the calling thread, so ask it on the thread that
207    /// will dispatch.
208    ///
209    /// # Examples
210    ///
211    /// ```
212    /// use std::num::NonZeroUsize;
213    /// use std::time::Duration;
214    /// use threadteam::Exec;
215    ///
216    /// let two = NonZeroUsize::new(2).unwrap();
217    /// assert_eq!(Exec::serial().dispatch_cost(two), Duration::MAX);
218    /// ```
219    pub fn dispatch_cost(&self, width: NonZeroUsize) -> Duration {
220        let width = width.min(self.budget);
221        if width.get() < 2 {
222            return Duration::MAX;
223        }
224        match (self.team, self.pool) {
225            (Some(t), _) => t.awake.costs.cost(DispatchKind::Team, width),
226            (None, Some(pool)) => {
227                let kind = if sched::in_pool(pool) {
228                    DispatchKind::InPool
229                } else {
230                    DispatchKind::Outside
231                };
232                pool.awake.costs.cost(kind, width)
233            }
234            (None, None) => Duration::MAX,
235        }
236    }
237
238    /// Whether a region of serial time `estimate` that opens `sections` parallel
239    /// sections of `width` participants pays on this token: `estimate >=
240    /// PAYS * sections * dispatch_cost(width)` (see [`pays_at`]). False for a
241    /// serial token.
242    ///
243    /// # Examples
244    ///
245    /// ```
246    /// use std::time::Duration;
247    /// use threadteam::Exec;
248    ///
249    /// let two = 2.try_into().unwrap();
250    /// assert!(!Exec::serial().pays(Duration::from_secs(1), 1, two));
251    /// ```
252    pub fn pays(&self, estimate: Duration, sections: u32, width: NonZeroUsize) -> bool {
253        pays_at(estimate, sections, self.dispatch_cost(width))
254    }
255
256    /// The width in `1..=budget` that minimises `sections * D(k) + estimate / k`
257    /// for this token, counting only widths at which the region
258    /// [`pays`](Self::pays); 1 when none does (see [`width_for_by`]).
259    ///
260    /// # Examples
261    ///
262    /// ```
263    /// use std::time::Duration;
264    /// use threadteam::Exec;
265    ///
266    /// assert_eq!(Exec::serial().width_for(Duration::from_secs(1), 1).get(), 1);
267    /// ```
268    pub fn width_for(&self, estimate: Duration, sections: u32) -> NonZeroUsize {
269        width_for_by(estimate, sections, self.budget, |k| self.dispatch_cost(k))
270    }
271
272    /// Whether no parallelism is possible: the budget is one slot.
273    ///
274    /// Kernels use this to take their serial path directly.
275    ///
276    /// # Examples
277    ///
278    /// ```
279    /// assert!(threadteam::Exec::serial().is_serial());
280    /// ```
281    pub fn is_serial(&self) -> bool {
282        self.budget.get() == 1
283    }
284
285    /// Whether this token carries team authority: it was minted by a
286    /// [`Team`](crate::Team) or is a child of such a token. Its sections are team
287    /// sections ([`DispatchKind::Team`]) and [`team`](Exec::team) returns the
288    /// carried (sub)team at once, without acquisition, so a consumer that wants a
289    /// [`TeamExec`] (faer's `Par::Team`) asks this first and keeps
290    /// [`external`](Exec::external) for plain tokens, whose `team` would reserve
291    /// members.
292    ///
293    /// # Examples
294    ///
295    /// ```
296    /// use threadteam::{Exec, Pool};
297    ///
298    /// assert!(!Exec::serial().carries_team());
299    /// let pool = Pool::builder().threads(2).acquire_timeout(std::time::Duration::from_secs(10)).build()?;
300    /// assert!(!Exec::new(&pool, 2)?.carries_team());
301    /// if let Ok(team) = pool.into_team(2) {
302    ///     assert!(team.exec(2).carries_team());
303    /// }
304    /// # Ok::<(), threadteam::Error>(())
305    /// ```
306    pub fn carries_team(&self) -> bool {
307        self.team.is_some()
308    }
309
310    /// Narrows the token to at most `max` slots.
311    ///
312    /// The result has budget `min(budget, max)` and the same pool; the dropped
313    /// slots are simply not used, so slots are conserved. Callers narrow before
314    /// [`broadcast`](Exec::broadcast) or [`external`](Exec::external) to match a team size chosen
315    /// by their own work model. Co-scheduling eligibility is unchanged by
316    /// `limit`: a root token stays eligible for `broadcast`, a child produced by
317    /// [`fork`](Exec::fork) stays ineligible.
318    ///
319    /// # Examples
320    ///
321    /// ```
322    /// use std::num::NonZeroUsize;
323    /// use threadteam::{Exec, Pool};
324    ///
325    /// let pool = Pool::builder().threads(4).acquire_timeout(std::time::Duration::from_secs(10)).build()?;
326    /// let exec = Exec::new(&pool, 4)?.limit(NonZeroUsize::new(2).unwrap());
327    /// assert!(exec.budget().get() <= 2);
328    /// # Ok::<(), threadteam::Error>(())
329    /// ```
330    pub fn limit(self, max: NonZeroUsize) -> Exec<'p> {
331        let budget = self.budget.min(max);
332        Exec {
333            budget,
334            team: self.team.map(|t| TeamAuth {
335                ids: &t.ids[..budget.get() - 1],
336                ..t
337            }),
338            ..self
339        }
340    }
341
342    /// A token with the same pool and budget, borrowed from `self`.
343    ///
344    /// Parallel entries consume the token, but a kernel often runs several
345    /// parallel phases in sequence. Each phase takes a reborrow. The parent is
346    /// exclusively borrowed and unusable while the child lives, so the slots are
347    /// held by one token at a time: this conserves slots exactly as handing the
348    /// token over would.
349    ///
350    /// # Examples
351    ///
352    /// ```
353    /// use threadteam::Exec;
354    ///
355    /// let mut exec = Exec::serial();
356    /// exec.reborrow().fork(2, |_, _| {});
357    /// exec.reborrow().fork(2, |_, _| {});
358    /// exec.fork(2, |_, _| {});
359    /// ```
360    pub fn reborrow(&mut self) -> Exec<'_> {
361        Exec {
362            pool: self.pool,
363            budget: self.budget,
364            child: self.child,
365            track: self.track.alias(),
366            team: self.team,
367            _leader: None,
368        }
369    }
370
371    /// Runs `jobs` jobs on `min(jobs, budget)` tasks; job `i` gets a child token.
372    ///
373    /// The budget is split into one share per task, shares differing by at most
374    /// one and summing to the budget. A job's child token carries its task's
375    /// share and may be passed to any kernel, which can `fork` again inside it.
376    /// When `jobs` exceeds the budget the tasks take job indices from a shared
377    /// counter, so the load balances; otherwise job `i` runs on task `i`.
378    ///
379    /// One task, a serial token, or `jobs <= 1` runs inline on the calling
380    /// thread without any scheduler call. Returns when every job has finished.
381    ///
382    /// # Examples
383    ///
384    /// ```
385    /// use std::sync::atomic::{AtomicUsize, Ordering};
386    /// use threadteam::{Exec, Pool};
387    ///
388    /// let pool = Pool::builder().threads(2).acquire_timeout(std::time::Duration::from_secs(10)).build()?;
389    /// let hits: Vec<AtomicUsize> = (0..5).map(|_| AtomicUsize::new(0)).collect();
390    /// Exec::new(&pool, 2)?.fork(5, |i, _child| {
391    ///     hits[i].fetch_add(1, Ordering::Relaxed);
392    /// });
393    /// assert!(hits.iter().all(|h| h.load(Ordering::Relaxed) == 1));
394    /// # Ok::<(), threadteam::Error>(())
395    /// ```
396    pub fn fork<F>(self, jobs: usize, f: F)
397    where
398        F: Fn(usize, Exec<'_>) + Sync,
399    {
400        if jobs == 0 {
401            return;
402        }
403        let budget = self.budget.get();
404        let tasks = jobs.min(budget);
405        let _released = self.track.release();
406        let next = AtomicUsize::new(0);
407        let this = &self;
408        let body = |k: usize, team_ids: Option<&[usize]>| {
409            let _live = this.track.task();
410            let share = split::share(budget, tasks, k);
411            let child = || Exec {
412                pool: this.pool,
413                // `share >= 1` because `tasks <= budget`.
414                budget: NonZeroUsize::new(share).unwrap_or(NonZeroUsize::MIN),
415                child: true,
416                track: this.track.child(),
417                team: this
418                    .team
419                    .zip(team_ids)
420                    .map(|(t, ids)| TeamAuth { ids, ..t }),
421                _leader: None,
422            };
423            if jobs <= tasks {
424                f(k, child());
425            } else {
426                loop {
427                    let job = next.fetch_add(1, Ordering::Relaxed);
428                    if job >= jobs {
429                        break;
430                    }
431                    f(job, child());
432                }
433            }
434        };
435        match (self.team, self.pool) {
436            (Some(t), _) if tasks > 1 => {
437                let plan = Plan::Even {
438                    size: budget,
439                    children: tasks,
440                };
441                split_raw(Some(t.awake), t.ids, &plan, t.over, |k, ids| {
442                    body(k, Some(ids));
443                });
444            }
445            (None, Some(pool)) if tasks > 1 => {
446                sched::run_tasks(pool, tasks, &self.track, |k| body(k, None));
447            }
448            _ => {
449                for k in 0..tasks {
450                    body(k, self.team.map(|t| t.ids));
451                }
452            }
453        }
454    }
455
456    /// Data-parallel loop over `0..len` in contiguous ranges.
457    ///
458    /// Uses at most `budget` tasks; every range is at least `grain` long except
459    /// possibly the last, and together the ranges cover `0..len` exactly once.
460    /// When one task suffices (`len <= grain`, a serial token) `f(0..len)` runs
461    /// inline on the calling thread without any scheduler call. `len == 0` does
462    /// nothing.
463    ///
464    /// # Examples
465    ///
466    /// ```
467    /// use std::sync::atomic::{AtomicUsize, Ordering};
468    /// use threadteam::{Exec, Pool};
469    ///
470    /// let pool = Pool::builder().threads(2).acquire_timeout(std::time::Duration::from_secs(10)).build()?;
471    /// let covered = AtomicUsize::new(0);
472    /// Exec::new(&pool, 2)?.for_ranges(10, 4.try_into().unwrap(), |r| {
473    ///     covered.fetch_add(r.len(), Ordering::Relaxed);
474    /// });
475    /// assert_eq!(covered.into_inner(), 10);
476    /// # Ok::<(), threadteam::Error>(())
477    /// ```
478    pub fn for_ranges<F>(self, len: usize, grain: NonZeroUsize, f: F)
479    where
480        F: Fn(Range<usize>) + Sync,
481    {
482        if len == 0 {
483            return;
484        }
485        let (chunk, tasks) = split::range_plan(len, grain.get(), self.budget.get());
486        let _released = self.track.release();
487        let track = &self.track;
488        let body = |k: usize| {
489            let _live = track.task();
490            f(split::chunk_range(len, chunk, k));
491        };
492        match (self.team, self.pool) {
493            (Some(t), _) if tasks > 1 => {
494                spread_raw(Some(t.awake), t.ids, tasks, t.over, body);
495            }
496            (None, Some(pool)) if tasks > 1 => sched::run_tasks(pool, tasks, track, body),
497            _ => {
498                for k in 0..tasks {
499                    body(k);
500                }
501            }
502        }
503    }
504
505    /// Co-scheduling for barrier-bearing kernels: runs `f(i)` for every
506    /// `i in 0..budget` with all participants running concurrently, or refuses.
507    ///
508    /// A serial or budget-one token runs `f(0)` inline and returns `Ok`.
509    /// Otherwise the pool is entered once and every participant runs on its own
510    /// worker, so `f` may block on a barrier shared by the `budget`
511    /// participants.
512    ///
513    /// # Errors
514    ///
515    /// Nothing runs and [`Refused`] is returned when co-scheduling cannot be
516    /// guaranteed:
517    ///
518    /// * [`Refused::ForkChild`]: the token was produced by [`fork`](Exec::fork)
519    ///   (or derived from such a token by `reborrow` or `limit`); its slots may
520    ///   share workers with siblings. Roots from [`Exec::new`], and their
521    ///   `reborrow`s and `limit`s, stay eligible.
522    ///
523    /// Callers must handle `Refused` by falling back to a barrier-free path.
524    /// * [`Refused::CallerInPool`]: the calling thread is a worker of the
525    ///   token's pool and would occupy a slot.
526    ///
527    /// # Guarantee and limits
528    ///
529    /// Concurrent broadcasts on one pool are queued on every worker in the same
530    /// order, so two of them cannot each hold half the workers. The participants
531    /// are the pool's workers `0..budget`; if unrelated tasks that never finish
532    /// occupy those workers, the broadcast waits for them. Keep barrier kernels on a pool whose workers are not held
533    /// by blocking work of another token.
534    ///
535    /// # Examples
536    ///
537    /// ```
538    /// use std::sync::Barrier;
539    /// use threadteam::{Exec, Pool};
540    ///
541    /// let pool = Pool::builder().threads(2).acquire_timeout(std::time::Duration::from_secs(10)).build()?;
542    /// let exec = Exec::new(&pool, 2)?;
543    /// let barrier = Barrier::new(exec.budget().get());
544    /// exec.broadcast(|_i| {
545    ///     barrier.wait(); // every participant is running, or this call refused
546    /// }).expect("called from a plain thread on a root token");
547    /// # Ok::<(), threadteam::Error>(())
548    /// ```
549    pub fn broadcast<F>(self, f: F) -> Result<(), Refused>
550    where
551        F: Fn(usize) + Sync,
552    {
553        if let Some(refused) = self.refusal() {
554            return Err(refused);
555        }
556        if let Some(t) = self.team {
557            // Reserved members are exclusively this token's: every participant
558            // runs at once, also for a child produced by `fork`.
559            let _released = self.track.release();
560            let track = &self.track;
561            spread_raw(Some(t.awake), t.ids, self.budget.get(), t.over, |i| {
562                let _live = track.task();
563                f(i);
564            });
565            return Ok(());
566        }
567        let pool = match self.pool {
568            Some(pool) if self.budget.get() > 1 => pool,
569            _ => {
570                let _live = self.track.task();
571                f(0);
572                return Ok(());
573            }
574        };
575        let _released = self.track.release();
576        #[cfg(feature = "parallel")]
577        sched::broadcast(pool, &self.track, self.budget.get(), f);
578        #[cfg(not(feature = "parallel"))]
579        {
580            let _ = pool;
581            f(0);
582        }
583        Ok(())
584    }
585
586    /// Whether [`broadcast`](Exec::broadcast) would run right now, without
587    /// consuming the token.
588    ///
589    /// True exactly when `broadcast` would not return [`Refused`]: the token is
590    /// not derived from [`fork`](Exec::fork) and the calling thread is not a
591    /// worker of its pool. A serial or budget-one token returns true, because
592    /// `broadcast` then runs `f(0)` inline. The answer depends on the calling
593    /// thread, so ask it on the thread that will broadcast.
594    ///
595    /// Use it before building an expensive setup that only a broadcast needs,
596    /// instead of paying for the setup and then being refused.
597    ///
598    /// # Examples
599    ///
600    /// ```
601    /// use threadteam::{Exec, Pool};
602    ///
603    /// let pool = Pool::builder().threads(2).acquire_timeout(std::time::Duration::from_secs(10)).build()?;
604    /// let exec = Exec::new(&pool, 2)?;
605    /// assert!(exec.can_broadcast());
606    /// exec.fork(1, |_, child| {
607    ///     let predicted = child.can_broadcast();
608    ///     assert_eq!(child.broadcast(|_| {}).is_ok(), predicted);
609    /// });
610    /// assert!(Exec::serial().can_broadcast());
611    /// # Ok::<(), threadteam::Error>(())
612    /// ```
613    pub fn can_broadcast(&self) -> bool {
614        self.refusal().is_none()
615    }
616
617    /// The single refusal rule shared by `broadcast` and `can_broadcast`.
618    fn refusal(&self) -> Option<Refused> {
619        if self.team.is_some() {
620            return None;
621        }
622        let pool = match self.pool {
623            Some(pool) if self.budget.get() > 1 => pool,
624            _ => return None,
625        };
626        if self.child {
627            return Some(Refused::ForkChild);
628        }
629        if sched::in_pool(pool) {
630            return Some(Refused::CallerInPool);
631        }
632        None
633    }
634
635    /// Hands the budget as a plain width to a library outside this stack.
636    ///
637    /// This is the only place a bare width exists. The token is consumed, the
638    /// pool is entered once (inline when the token is serial, has a budget of
639    /// one, or the caller already is a worker of the pool) and `f` receives the
640    /// budget, for example to build `faer::Par::rayon(width)` or to set a vendor
641    /// BLAS thread count. Whatever the library schedules with that width runs on
642    /// this pool, never on an ambient one. How the library divides the width is
643    /// its own responsibility. Without the `parallel` feature, `f` runs inline
644    /// and receives a width of one.
645    ///
646    /// # Placement
647    ///
648    /// The token's reserved members are **lent** for the call: each restores its
649    /// own affinity mask and serves the pool's jobs until the call returns. The
650    /// calling thread does the same when it is itself a pinned participant of
651    /// this pool (for example a member that leads a [`fork`](Exec::fork) child),
652    /// so the foreign work never keeps a team's single-CPU placement. Both
653    /// re-pin at the next pin section, that is at the next root team scope.
654    ///
655    /// # Examples
656    ///
657    /// ```
658    /// use threadteam::Exec;
659    ///
660    /// let width = Exec::serial().external(|width| width.get());
661    /// assert_eq!(width, 1);
662    /// ```
663    pub fn external<R: Send>(self, f: impl FnOnce(NonZeroUsize) -> R + Send) -> R {
664        #[cfg(feature = "parallel")]
665        if let Some(pool) = self.pool {
666            if self.budget.get() > 1 {
667                let width = self.budget;
668                let _released = self.track.release();
669                // A team token lends its reserved members to the pool for the
670                // call: the library's Rayon work runs on them as well.
671                let _lent = self.team.map(|t| LendGuard::new(t.awake, t.ids));
672                // The calling thread may itself be a pinned participant of this
673                // pool (the member that leads a `fork` child, or a token used
674                // from inside a share): it restores its own mask before it
675                // serves foreign work, as a lent member does.
676                sched::unpin_worker(pool);
677                return sched::enter(pool, &self.track, move || f(width));
678            }
679        }
680        f(NonZeroUsize::MIN)
681    }
682
683    /// Reserves `budget - 1` pool members for the scope of `f` and hands `f`
684    /// the exclusive [`TeamExec`] authority over them and the calling thread.
685    ///
686    /// The calling thread is participant 0 (the leader); the members are
687    /// pool workers that stay reserved, spinning and serving no Rayon job,
688    /// until `f` returns (or unwinds). A serial or budget-one token yields a
689    /// team of one without touching the pool.
690    ///
691    /// Acquisition never waits for a Rayon job that is already running. Idle
692    /// members of an earlier team (see [`WaitPolicy`]) are claimed with a
693    /// compare-and-swap; missing ones are requested from the pool, and the
694    /// call waits at most the pool's acquisition timeout
695    /// ([`PoolBuilder::acquire_timeout`](crate::PoolBuilder::acquire_timeout))
696    /// for idle workers to wake. If not all members are there by then, every
697    /// tentative claim is released and nothing runs. Before `f` is called,
698    /// every participant has acknowledged on its own thread.
699    ///
700    /// The token stays with the caller (this takes `&mut self`), so a refused
701    /// acquisition leaves it available for a barrier-free fallback such as
702    /// [`fork`](Exec::fork). It is conserved as usual: while `f` runs the token
703    /// is borrowed.
704    ///
705    /// This call opens a **root team scope**: it reads the calling thread's
706    /// affinity mask, which decides two things. With
707    /// [`PoolBuilder::pin_participants`](crate::PoolBuilder::pin_participants) on
708    /// it places every member on its own CPU of that mask for the scope (the
709    /// leader is never pinned).
710    /// Whether or not pinning is on, it records whether the scope is
711    /// *oversubscribed* (more participants than CPUs in the mask), which decides
712    /// how the members park and how a joining leader waits. See the design
713    /// record.
714    ///
715    /// # Errors
716    ///
717    /// [`Refused::Unavailable`] when the members could not be reserved.
718    ///
719    /// # Examples
720    ///
721    /// ```
722    /// use threadteam::{Exec, Pool, Refused};
723    ///
724    /// let pool = Pool::builder().threads(4).acquire_timeout(std::time::Duration::from_secs(10)).build()?;
725    /// let mut exec = Exec::new(&pool, 4)?;
726    /// match exec.team(|team| team.size().get()) {
727    ///     Ok(n) => assert!(n >= 1),
728    ///     Err(Refused::Unavailable) => exec.fork(4, |_, _| {}), // barrier-free fallback
729    ///     Err(_) => unreachable!(),
730    /// }
731    /// # Ok::<(), threadteam::Error>(())
732    /// ```
733    pub fn team<R>(
734        &mut self,
735        f: impl for<'team> FnOnce(TeamExec<'team>) -> R,
736    ) -> Result<R, Refused> {
737        self.team_with(WaitOverride::new(), f)
738    }
739
740    /// [`team`](Exec::team) with a scoped override of the waiting policy.
741    ///
742    /// The override applies to this operation only and is inherited by the
743    /// children of the team; the pool's defaults are not changed. It governs
744    /// the idle interval after this reservation ends and never weakens the
745    /// reservation itself.
746    ///
747    /// # Errors
748    ///
749    /// [`Refused::Unavailable`] when the members could not be reserved.
750    pub fn team_with<R>(
751        &mut self,
752        over: WaitOverride,
753        f: impl for<'team> FnOnce(TeamExec<'team>) -> R,
754    ) -> Result<R, Refused> {
755        if let Some(t) = self.team {
756            // Carried members: the (sub)team is available without acquisition.
757            let mut team = TeamExec {
758                awake: Some(t.awake),
759                ids: t.ids,
760                policy: t.policy.with(over),
761                track: self.track.alias(),
762                over: t.over,
763            };
764            // Only the root token minted by a `Team` opens a root scope; a
765            // child, a reborrow or a sub-team re-pins nothing of its own.
766            if self._leader.is_some() {
767                // Root-scope entry, on the executing thread: the root pressure
768                // and the members' spin budget are decided by *this* thread's CPU
769                // set. A token minted on another thread (or another mask) is
770                // paced by the thread that enters the scope, exactly as its pin
771                // section is placed by that thread. Descendants inherit it.
772                let oversubscribed = crate::pin::oversubscribed(team.size().get());
773                team.over = oversubscribed;
774                if let Some(awake) = team.awake {
775                    awake.set_spin_budget(team.ids, awake.spin_budget(oversubscribed));
776                }
777                if let Some(auth) = self.team.as_mut() {
778                    auth.over = oversubscribed;
779                }
780                if team.size().get() > 1 {
781                    pin_scope(&mut team);
782                }
783            }
784            return Ok(f(team));
785        }
786        let pool = match self.pool {
787            Some(pool) if self.budget.get() > 1 => pool,
788            _ => {
789                let policy = self.pool.map_or(WaitPolicy::DEFAULT, Pool::wait_policy);
790                return Ok(f(TeamExec {
791                    awake: None,
792                    ids: &[],
793                    policy: policy.with(over),
794                    track: self.track.alias(),
795                    over: false,
796                }));
797            }
798        };
799        let policy = pool.wait_policy().with(over);
800        // Root-scope entry: the CPU set of this calling thread decides both the
801        // placement of the participants and whether the scope is
802        // oversubscribed, whatever the pool's pinning setting.
803        let oversubscribed = crate::pin::oversubscribed(self.budget.get());
804        let reservation = pool.awake.acquire(
805            pool,
806            self.budget.get() - 1,
807            policy,
808            pool.awake.spin_budget(oversubscribed),
809        )?;
810        let mut team = TeamExec {
811            awake: Some(&pool.awake),
812            ids: &reservation.ids,
813            policy,
814            track: self.track.alias(),
815            over: oversubscribed,
816        };
817        // The pin section is also the acknowledgement; without pinning the
818        // participants acknowledge as before. The leader is never pinned, so
819        // there is nothing to restore when the scope ends or unwinds.
820        let pinned = team.size().get() > 1 && pin_scope(&mut team);
821        if !pinned {
822            team.acknowledge();
823        }
824        Ok(f(team))
825    }
826
827    /// Root id of this token in the [`ledger`](crate::ledger), if it has one.
828    ///
829    /// Tokens from [`Exec::new`] are roots; children and reborrows report the
830    /// root of their origin; [`Exec::serial`] has none.
831    ///
832    /// Available with the `ledger` cargo feature.
833    ///
834    /// # Examples
835    ///
836    /// ```
837    /// use threadteam::{Exec, Pool};
838    ///
839    /// let pool = Pool::builder().threads(1).acquire_timeout(std::time::Duration::from_secs(10)).build()?;
840    /// assert!(Exec::new(&pool, 1)?.ledger_root().is_some());
841    /// assert!(Exec::serial().ledger_root().is_none());
842    /// # Ok::<(), threadteam::Error>(())
843    /// ```
844    #[cfg(feature = "ledger")]
845    pub fn ledger_root(&self) -> Option<u64> {
846        self.track.root_id()
847    }
848}
849
850/// Runs the pin section of a root team scope: every member places itself on its
851/// own CPU of the calling thread's set. The leader is never pinned.
852///
853/// `false` when member pinning is off for this pool, when the platform has no
854/// CPU set (off Linux), or when the calling thread's mask could not be read:
855/// then nothing ran and the caller acknowledges as usual.
856fn pin_scope(team: &mut TeamExec<'_>) -> bool {
857    if !team.awake.is_some_and(|awake| awake.pin) {
858        return false;
859    }
860    // The leader reads its own mask, and never changes it: that set is where
861    // the members are placed.
862    let Some(set) = crate::pin::read_here() else {
863        return false;
864    };
865    team.pin_section(&set);
866    true
867}
868
869impl fmt::Debug for Exec<'_> {
870    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
871        f.debug_struct("Exec")
872            .field("budget", &self.budget)
873            .field("pooled", &self.pool.is_some())
874            .field("child", &self.child)
875            .field("team", &self.team.is_some())
876            .finish()
877    }
878}