tcontract_exec/exec.rs
1use std::num::NonZeroUsize;
2
3use crate::{BatchDecision, BatchPolicy, BatchQuery, ExecError, Pool, WorkspaceProvider};
4
5/// The resources one operation runs with: a placement token, optional reusable
6/// scratch, and the batch lane policy.
7///
8/// The [`threadteam::Exec`] token decides where the work runs and how many
9/// tasks may run at once. This type adds what is specific to contraction: the
10/// workspace the call may reuse and the [`BatchPolicy`]. It owns no placement
11/// code. To start parallel work take the token with [`Exec::token`] (a reborrow,
12/// so one `Exec` serves a sequence of calls) or [`Exec::into_token`], and call
13/// its `fork`, `for_ranges`, `broadcast` or `external`; build a child bundle
14/// for each forked job with [`Exec::from_token`].
15///
16/// Like the token it wraps, an `Exec` is neither `Clone` nor `Copy`, so two
17/// operations cannot both claim the full budget; operations take `&mut Exec`.
18///
19/// # Examples
20///
21/// ```
22/// use tcontract_exec::Exec;
23/// let mut exec = Exec::serial();
24/// assert_eq!(exec.budget(), 1);
25/// assert!(exec.token().is_serial());
26/// ```
27pub struct Exec<'a> {
28 token: threadteam::Exec<'a>,
29 workspace: Option<&'a dyn WorkspaceProvider>,
30 policy: BatchPolicy,
31}
32
33// A `&dyn WorkspaceProvider` is not `Debug`, so the derive cannot cover the
34// workspace. The provider is an anonymous implementation detail here, so only
35// its presence is shown.
36impl std::fmt::Debug for Exec<'_> {
37 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
38 f.debug_struct("Exec")
39 .field("token", &self.token)
40 .field("workspace", &self.workspace.is_some())
41 .field("policy", &self.policy)
42 .finish()
43 }
44}
45
46impl<'a> Exec<'a> {
47 /// The serial context: the calling thread only, call-local scratch.
48 pub fn serial() -> Exec<'static> {
49 Exec {
50 token: threadteam::Exec::serial(),
51 workspace: None,
52 policy: BatchPolicy::default(),
53 }
54 }
55
56 /// Serial execution with caller-owned reusable storage.
57 pub fn serial_with_workspace(workspace: &'a dyn WorkspaceProvider) -> Self {
58 Exec {
59 token: threadteam::Exec::serial(),
60 workspace: Some(workspace),
61 policy: BatchPolicy::default(),
62 }
63 }
64
65 /// `budget` slots of `pool`, with the pool's own workspace and the default
66 /// batch policy. This is the only place a budget is clamped: to the pool
67 /// width.
68 ///
69 /// # Errors
70 ///
71 /// [`ExecError::Placement`] when `budget == 0`.
72 ///
73 /// # Examples
74 ///
75 /// ```
76 /// use tcontract_exec::{Exec, Pool};
77 /// let sp = threadteam::Pool::builder().threads(2).build().unwrap();
78 /// let pool = Pool::borrow(&sp);
79 /// assert_eq!(Exec::new(&pool, 64).unwrap().budget(), pool.size());
80 /// assert!(Exec::new(&pool, 0).is_err());
81 /// ```
82 pub fn new(pool: &'a Pool<'_>, budget: usize) -> Result<Self, ExecError> {
83 Ok(Exec {
84 token: threadteam::Exec::new(pool.inner(), budget)?,
85 workspace: Some(pool.workspace()),
86 policy: BatchPolicy::default(),
87 })
88 }
89
90 /// Bundle an existing token with a workspace and a policy: the way to give a
91 /// forked job its own bundle, since `fork` hands out bare child tokens.
92 ///
93 /// # Examples
94 ///
95 /// ```
96 /// use tcontract_exec::{BatchPolicy, Exec};
97 /// let exec = Exec::from_token(threadteam::Exec::serial(), None, BatchPolicy::default());
98 /// assert_eq!(exec.budget(), 1);
99 /// ```
100 pub fn from_token(
101 token: threadteam::Exec<'a>,
102 workspace: Option<&'a dyn WorkspaceProvider>,
103 policy: BatchPolicy,
104 ) -> Self {
105 Exec {
106 token,
107 workspace,
108 policy,
109 }
110 }
111
112 /// The placement token, lent for one call. The bundle keeps its budget, so
113 /// several calls can follow one another.
114 pub fn token(&mut self) -> threadteam::Exec<'_> {
115 self.token.reborrow()
116 }
117
118 /// The placement token, consuming the bundle (for `fork`, `broadcast`, or
119 /// any entry that takes the token by value and outlives this borrow).
120 pub fn into_token(self) -> threadteam::Exec<'a> {
121 self.token
122 }
123
124 /// The same resources, lent for as long as `self` is exclusively borrowed.
125 pub fn reborrow(&mut self) -> Exec<'_> {
126 Exec {
127 token: self.token.reborrow(),
128 workspace: self.workspace,
129 policy: self.policy,
130 }
131 }
132
133 /// Narrow the budget to `min(budget, max)`; the dropped slots stay unused.
134 ///
135 /// # Examples
136 ///
137 /// ```
138 /// use tcontract_exec::Exec;
139 /// assert_eq!(Exec::serial().limit(4.try_into().unwrap()).budget(), 1);
140 /// ```
141 #[must_use]
142 pub fn limit(self, max: NonZeroUsize) -> Self {
143 Exec {
144 token: self.token.limit(max),
145 workspace: self.workspace,
146 policy: self.policy,
147 }
148 }
149
150 /// The batch lane policy this context applies.
151 ///
152 /// A serial context has one thread but still carries a policy, so a forced
153 /// strategy is resolved (and can be refused) at one thread as well.
154 ///
155 /// # Examples
156 ///
157 /// ```
158 /// use tcontract_exec::{BatchPolicy, BatchStrategy, Exec};
159 /// assert_eq!(Exec::serial().batch_policy().strategy, BatchStrategy::Auto);
160 /// let exec = Exec::serial().with_batch_policy(BatchPolicy::new(BatchStrategy::Outer));
161 /// assert_eq!(exec.batch_policy().strategy, BatchStrategy::Outer);
162 /// ```
163 pub fn batch_policy(&self) -> &BatchPolicy {
164 &self.policy
165 }
166
167 /// Return this context with a different batch lane policy.
168 ///
169 /// A serial context has one thread, so the policy cannot change its lane
170 /// count, but it is preserved: a forced strategy is still honoured (or
171 /// refused with a typed error) at one thread instead of silently running
172 /// `Auto`.
173 #[must_use]
174 pub fn with_batch_policy(mut self, policy: BatchPolicy) -> Self {
175 self.policy = policy;
176 self
177 }
178
179 /// Decide the batch route for `query` on this context's budget and policy.
180 ///
181 /// This is the one decision function every operation that owns batch lanes
182 /// calls; see [`crate::batch`]. Whether a *route* can honour the returned
183 /// decision is the route's own preflight (see
184 /// [`crate::batch::BatchDecision`]'s route contract); the decision itself
185 /// is arithmetic and never fails.
186 ///
187 /// # Examples
188 ///
189 /// ```
190 /// use tcontract_exec::{BatchQuery, Exec};
191 /// let d = Exec::serial().batch_decision(&BatchQuery::new(4, 1000.0));
192 /// assert_eq!((d.lanes, d.item_threads), (1, 1));
193 /// ```
194 pub fn batch_decision(&self, query: &BatchQuery) -> BatchDecision {
195 crate::batch::batch_decision(&self.policy, self, query)
196 }
197
198 /// Most threads an operation may occupy (1 for a serial context).
199 pub fn budget(&self) -> usize {
200 self.token.budget().get()
201 }
202
203 /// Storage the operations on this context may share, or none.
204 ///
205 /// A pool lends its own arena to every operation that runs on it; a serial
206 /// context owns nothing, so a caller that wants reuse keeps a provider of
207 /// its own and lends it through [`Exec::serial_with_workspace`].
208 pub fn workspace(&self) -> Option<&'a dyn WorkspaceProvider> {
209 self.workspace
210 }
211
212 /// The placement token, borrowed: for the rules that read its dispatch
213 /// costs without taking authority over it (`pays`, `width_for`, `dispatch_cost`).
214 pub fn token_ref(&self) -> &threadteam::Exec<'a> {
215 &self.token
216 }
217
218 /// Partition width for work estimated at `serial_ns` on one thread that the
219 /// region performs in `sections` dispatches: the width in `1..=budget` at which
220 /// it pays under the dispatch cost of this token (see
221 /// [`threadteam::Exec::width_for`]); `1` when none does.
222 ///
223 /// # Examples
224 ///
225 /// ```
226 /// use tcontract_exec::Exec;
227 /// assert_eq!(Exec::serial().width_for(1e9, 1), 1);
228 /// ```
229 pub fn width_for(&self, serial_ns: f64, sections: u32) -> usize {
230 self.token.width_for(ns_duration(serial_ns), sections).get()
231 }
232
233 /// [`Exec::width_for`] as if the budget were `budget` (clamped to the real one),
234 /// with the dispatch costs of this token: the rule for a stage that was
235 /// already narrowed to `budget` slots without narrowing the token.
236 ///
237 /// # Examples
238 ///
239 /// ```
240 /// use tcontract_exec::Exec;
241 /// assert_eq!(Exec::serial().width_for_within(4, 1e9, 1), 1);
242 /// ```
243 pub fn width_for_within(&self, budget: usize, serial_ns: f64, sections: u32) -> usize {
244 let budget = NonZeroUsize::new(budget.max(1))
245 .unwrap_or(NonZeroUsize::MIN)
246 .min(self.token.budget());
247 threadteam::width_for_by(ns_duration(serial_ns), sections, budget, |k| {
248 self.token.dispatch_cost(k)
249 })
250 .get()
251 }
252
253 /// Whether `serial_ns` of work in `sections` dispatches pays at `width`
254 /// participants (see [`threadteam::Exec::pays`]).
255 ///
256 /// # Examples
257 ///
258 /// ```
259 /// use tcontract_exec::Exec;
260 /// assert!(!Exec::serial().pays(1e9, 1, 2));
261 /// ```
262 pub fn pays(&self, serial_ns: f64, sections: u32, width: usize) -> bool {
263 let width = NonZeroUsize::new(width).unwrap_or(NonZeroUsize::MIN);
264 self.token.pays(ns_duration(serial_ns), sections, width)
265 }
266}
267
268/// A non-negative estimate in nanoseconds as a `Duration`: NaN and negative values
269/// are zero (they stay serial), and an overflowing value saturates.
270fn ns_duration(ns: f64) -> std::time::Duration {
271 if ns.is_nan() || ns <= 0.0 {
272 std::time::Duration::ZERO
273 } else if ns >= 1e18 {
274 std::time::Duration::from_secs(u64::MAX / 4)
275 } else {
276 std::time::Duration::from_secs_f64(ns * 1e-9)
277 }
278}