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}