1use serde::{Deserialize, Serialize};
4use thiserror::Error;
5
6#[derive(Clone, Debug, Eq, PartialEq)]
8pub struct OperationReceipt<T> {
9 pub value: T,
11 pub work: WorkCounters,
13}
14
15#[derive(Debug, Error)]
17#[error("{error}")]
18pub struct OperationFailure<E> {
19 pub error: E,
21 pub work: Box<WorkCounters>,
23}
24
25impl<E> OperationFailure<E> {
26 #[must_use]
28 pub fn new(error: E, work: WorkCounters) -> Self {
29 Self {
30 error,
31 work: Box::new(work),
32 }
33 }
34
35 #[must_use]
37 pub fn before_work(error: E) -> Self {
38 Self::new(error, WorkCounters::default())
39 }
40
41 pub fn map_with_prior_work<T>(
47 self,
48 prior: WorkCounters,
49 map_error: impl FnOnce(E) -> T,
50 ) -> OperationFailure<T>
51 where
52 T: From<WorkError>,
53 {
54 match prior.checked_add(*self.work) {
55 Ok(combined) => OperationFailure::new(map_error(self.error), combined),
56 Err(error) => OperationFailure::new(error.into(), prior),
57 }
58 }
59}
60
61pub type MeasuredResult<T, E> = Result<T, OperationFailure<E>>;
63
64#[derive(Clone, Copy, Debug, Default, Deserialize, Eq, PartialEq, Serialize)]
66#[serde(default, rename_all = "camelCase")]
67pub struct WorkCounters {
68 pub authority_records_read: u64,
70 pub authority_records_appended: u64,
72 pub authority_bytes_read: u64,
74 pub authority_bytes_written: u64,
76 pub object_probes: u64,
78 pub backend_read_operations: u64,
80 pub backend_write_operations: u64,
82 pub durability_operations: u64,
84 pub page_reads: u64,
86 pub page_writes: u64,
88 pub object_bytes_read: u64,
90 pub object_bytes_written: u64,
92 pub bytes_hashed: u64,
94 pub bytes_copied: u64,
96 pub bytes_encoded: u64,
98 pub source_bytes_read: u64,
100 pub source_path_components: u64,
102 pub source_entries_visited: u64,
104 pub output_bytes: u64,
106 pub items_examined: u64,
108 pub items_returned: u64,
110 pub allocation_operations: u64,
112 pub peak_allocation_bytes: u64,
114 pub materializations: u64,
116}
117
118impl WorkCounters {
119 pub const UNBOUNDED: Self = Self {
121 authority_records_read: u64::MAX,
122 authority_records_appended: u64::MAX,
123 authority_bytes_read: u64::MAX,
124 authority_bytes_written: u64::MAX,
125 object_probes: u64::MAX,
126 backend_read_operations: u64::MAX,
127 backend_write_operations: u64::MAX,
128 durability_operations: u64::MAX,
129 page_reads: u64::MAX,
130 page_writes: u64::MAX,
131 object_bytes_read: u64::MAX,
132 object_bytes_written: u64::MAX,
133 bytes_hashed: u64::MAX,
134 bytes_copied: u64::MAX,
135 bytes_encoded: u64::MAX,
136 source_bytes_read: u64::MAX,
137 source_path_components: u64::MAX,
138 source_entries_visited: u64::MAX,
139 output_bytes: u64::MAX,
140 items_examined: u64::MAX,
141 items_returned: u64::MAX,
142 allocation_operations: u64::MAX,
143 peak_allocation_bytes: u64::MAX,
144 materializations: u64::MAX,
145 };
146
147 pub fn checked_add(self, other: Self) -> Result<Self, WorkError> {
153 Ok(Self {
154 authority_records_read: add(self.authority_records_read, other.authority_records_read)?,
155 authority_records_appended: add(
156 self.authority_records_appended,
157 other.authority_records_appended,
158 )?,
159 authority_bytes_read: add(self.authority_bytes_read, other.authority_bytes_read)?,
160 authority_bytes_written: add(
161 self.authority_bytes_written,
162 other.authority_bytes_written,
163 )?,
164 object_probes: add(self.object_probes, other.object_probes)?,
165 backend_read_operations: add(
166 self.backend_read_operations,
167 other.backend_read_operations,
168 )?,
169 backend_write_operations: add(
170 self.backend_write_operations,
171 other.backend_write_operations,
172 )?,
173 durability_operations: add(self.durability_operations, other.durability_operations)?,
174 page_reads: add(self.page_reads, other.page_reads)?,
175 page_writes: add(self.page_writes, other.page_writes)?,
176 object_bytes_read: add(self.object_bytes_read, other.object_bytes_read)?,
177 object_bytes_written: add(self.object_bytes_written, other.object_bytes_written)?,
178 bytes_hashed: add(self.bytes_hashed, other.bytes_hashed)?,
179 bytes_copied: add(self.bytes_copied, other.bytes_copied)?,
180 bytes_encoded: add(self.bytes_encoded, other.bytes_encoded)?,
181 source_bytes_read: add(self.source_bytes_read, other.source_bytes_read)?,
182 source_path_components: add(self.source_path_components, other.source_path_components)?,
183 source_entries_visited: add(self.source_entries_visited, other.source_entries_visited)?,
184 output_bytes: add(self.output_bytes, other.output_bytes)?,
185 items_examined: add(self.items_examined, other.items_examined)?,
186 items_returned: add(self.items_returned, other.items_returned)?,
187 allocation_operations: add(self.allocation_operations, other.allocation_operations)?,
188 peak_allocation_bytes: self.peak_allocation_bytes.max(other.peak_allocation_bytes),
189 materializations: add(self.materializations, other.materializations)?,
190 })
191 }
192
193 pub fn verify(self, budget: WorkBudget) -> Result<(), WorkError> {
199 let fields = [
200 (
201 "authority_records_read",
202 self.authority_records_read,
203 budget.authority_records_read,
204 ),
205 (
206 "authority_records_appended",
207 self.authority_records_appended,
208 budget.authority_records_appended,
209 ),
210 (
211 "authority_bytes_read",
212 self.authority_bytes_read,
213 budget.authority_bytes_read,
214 ),
215 (
216 "authority_bytes_written",
217 self.authority_bytes_written,
218 budget.authority_bytes_written,
219 ),
220 ("object_probes", self.object_probes, budget.object_probes),
221 (
222 "backend_read_operations",
223 self.backend_read_operations,
224 budget.backend_read_operations,
225 ),
226 (
227 "backend_write_operations",
228 self.backend_write_operations,
229 budget.backend_write_operations,
230 ),
231 (
232 "durability_operations",
233 self.durability_operations,
234 budget.durability_operations,
235 ),
236 ("page_reads", self.page_reads, budget.page_reads),
237 ("page_writes", self.page_writes, budget.page_writes),
238 (
239 "object_bytes_read",
240 self.object_bytes_read,
241 budget.object_bytes_read,
242 ),
243 (
244 "object_bytes_written",
245 self.object_bytes_written,
246 budget.object_bytes_written,
247 ),
248 ("bytes_hashed", self.bytes_hashed, budget.bytes_hashed),
249 ("bytes_copied", self.bytes_copied, budget.bytes_copied),
250 ("bytes_encoded", self.bytes_encoded, budget.bytes_encoded),
251 (
252 "source_bytes_read",
253 self.source_bytes_read,
254 budget.source_bytes_read,
255 ),
256 (
257 "source_path_components",
258 self.source_path_components,
259 budget.source_path_components,
260 ),
261 (
262 "source_entries_visited",
263 self.source_entries_visited,
264 budget.source_entries_visited,
265 ),
266 ("output_bytes", self.output_bytes, budget.output_bytes),
267 ("items_examined", self.items_examined, budget.items_examined),
268 ("items_returned", self.items_returned, budget.items_returned),
269 (
270 "allocation_operations",
271 self.allocation_operations,
272 budget.allocation_operations,
273 ),
274 (
275 "peak_allocation_bytes",
276 self.peak_allocation_bytes,
277 budget.peak_allocation_bytes,
278 ),
279 (
280 "materializations",
281 self.materializations,
282 budget.materializations,
283 ),
284 ];
285 for (counter, observed, maximum) in fields {
286 if observed > maximum {
287 return Err(WorkError::BudgetExceeded {
288 counter,
289 observed,
290 maximum,
291 });
292 }
293 }
294 Ok(())
295 }
296
297 pub fn remaining(self, budget: WorkBudget) -> Result<WorkBudget, WorkError> {
306 self.verify(budget)?;
307 Ok(WorkBudget {
308 authority_records_read: budget.authority_records_read - self.authority_records_read,
309 authority_records_appended: budget.authority_records_appended
310 - self.authority_records_appended,
311 authority_bytes_read: budget.authority_bytes_read - self.authority_bytes_read,
312 authority_bytes_written: budget.authority_bytes_written - self.authority_bytes_written,
313 object_probes: budget.object_probes - self.object_probes,
314 backend_read_operations: budget.backend_read_operations - self.backend_read_operations,
315 backend_write_operations: budget.backend_write_operations
316 - self.backend_write_operations,
317 durability_operations: budget.durability_operations - self.durability_operations,
318 page_reads: budget.page_reads - self.page_reads,
319 page_writes: budget.page_writes - self.page_writes,
320 object_bytes_read: budget.object_bytes_read - self.object_bytes_read,
321 object_bytes_written: budget.object_bytes_written - self.object_bytes_written,
322 bytes_hashed: budget.bytes_hashed - self.bytes_hashed,
323 bytes_copied: budget.bytes_copied - self.bytes_copied,
324 bytes_encoded: budget.bytes_encoded - self.bytes_encoded,
325 source_bytes_read: budget.source_bytes_read - self.source_bytes_read,
326 source_path_components: budget.source_path_components - self.source_path_components,
327 source_entries_visited: budget.source_entries_visited - self.source_entries_visited,
328 output_bytes: budget.output_bytes - self.output_bytes,
329 items_examined: budget.items_examined - self.items_examined,
330 items_returned: budget.items_returned - self.items_returned,
331 allocation_operations: budget.allocation_operations - self.allocation_operations,
332 peak_allocation_bytes: budget.peak_allocation_bytes,
333 materializations: budget.materializations - self.materializations,
334 })
335 }
336}
337
338pub type WorkBudget = WorkCounters;
340
341fn add(left: u64, right: u64) -> Result<u64, WorkError> {
342 left.checked_add(right).ok_or(WorkError::Overflow)
343}
344
345#[derive(Clone, Copy, Debug, Eq, Error, PartialEq)]
347pub enum WorkError {
348 #[error("filesystem work accounting overflowed")]
350 Overflow,
351 #[error("work counter {counter} observed {observed}, exceeding {maximum}")]
353 BudgetExceeded {
354 counter: &'static str,
356 observed: u64,
358 maximum: u64,
360 },
361}
362
363#[cfg(test)]
364#[path = "tests/performance.rs"]
365mod tests;