use serde::{Deserialize, Serialize};
use thiserror::Error;
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct OperationReceipt<T> {
pub value: T,
pub work: WorkCounters,
}
#[derive(Debug, Error)]
#[error("{error}")]
pub struct OperationFailure<E> {
pub error: E,
pub work: Box<WorkCounters>,
}
impl<E> OperationFailure<E> {
#[must_use]
pub fn new(error: E, work: WorkCounters) -> Self {
Self {
error,
work: Box::new(work),
}
}
#[must_use]
pub fn before_work(error: E) -> Self {
Self::new(error, WorkCounters::default())
}
pub fn map_with_prior_work<T>(
self,
prior: WorkCounters,
map_error: impl FnOnce(E) -> T,
) -> OperationFailure<T>
where
T: From<WorkError>,
{
match prior.checked_add(*self.work) {
Ok(combined) => OperationFailure::new(map_error(self.error), combined),
Err(error) => OperationFailure::new(error.into(), prior),
}
}
}
pub type MeasuredResult<T, E> = Result<T, OperationFailure<E>>;
#[derive(Clone, Copy, Debug, Default, Deserialize, Eq, PartialEq, Serialize)]
#[serde(default, rename_all = "camelCase")]
pub struct WorkCounters {
pub authority_records_read: u64,
pub authority_records_appended: u64,
pub authority_bytes_read: u64,
pub authority_bytes_written: u64,
pub object_probes: u64,
pub backend_read_operations: u64,
pub backend_write_operations: u64,
pub durability_operations: u64,
pub page_reads: u64,
pub page_writes: u64,
pub object_bytes_read: u64,
pub object_bytes_written: u64,
pub bytes_hashed: u64,
pub bytes_copied: u64,
pub bytes_encoded: u64,
pub source_bytes_read: u64,
pub source_path_components: u64,
pub source_entries_visited: u64,
pub output_bytes: u64,
pub items_examined: u64,
pub items_returned: u64,
pub allocation_operations: u64,
pub peak_allocation_bytes: u64,
pub materializations: u64,
}
impl WorkCounters {
pub const UNBOUNDED: Self = Self {
authority_records_read: u64::MAX,
authority_records_appended: u64::MAX,
authority_bytes_read: u64::MAX,
authority_bytes_written: u64::MAX,
object_probes: u64::MAX,
backend_read_operations: u64::MAX,
backend_write_operations: u64::MAX,
durability_operations: u64::MAX,
page_reads: u64::MAX,
page_writes: u64::MAX,
object_bytes_read: u64::MAX,
object_bytes_written: u64::MAX,
bytes_hashed: u64::MAX,
bytes_copied: u64::MAX,
bytes_encoded: u64::MAX,
source_bytes_read: u64::MAX,
source_path_components: u64::MAX,
source_entries_visited: u64::MAX,
output_bytes: u64::MAX,
items_examined: u64::MAX,
items_returned: u64::MAX,
allocation_operations: u64::MAX,
peak_allocation_bytes: u64::MAX,
materializations: u64::MAX,
};
pub fn checked_add(self, other: Self) -> Result<Self, WorkError> {
Ok(Self {
authority_records_read: add(self.authority_records_read, other.authority_records_read)?,
authority_records_appended: add(
self.authority_records_appended,
other.authority_records_appended,
)?,
authority_bytes_read: add(self.authority_bytes_read, other.authority_bytes_read)?,
authority_bytes_written: add(
self.authority_bytes_written,
other.authority_bytes_written,
)?,
object_probes: add(self.object_probes, other.object_probes)?,
backend_read_operations: add(
self.backend_read_operations,
other.backend_read_operations,
)?,
backend_write_operations: add(
self.backend_write_operations,
other.backend_write_operations,
)?,
durability_operations: add(self.durability_operations, other.durability_operations)?,
page_reads: add(self.page_reads, other.page_reads)?,
page_writes: add(self.page_writes, other.page_writes)?,
object_bytes_read: add(self.object_bytes_read, other.object_bytes_read)?,
object_bytes_written: add(self.object_bytes_written, other.object_bytes_written)?,
bytes_hashed: add(self.bytes_hashed, other.bytes_hashed)?,
bytes_copied: add(self.bytes_copied, other.bytes_copied)?,
bytes_encoded: add(self.bytes_encoded, other.bytes_encoded)?,
source_bytes_read: add(self.source_bytes_read, other.source_bytes_read)?,
source_path_components: add(self.source_path_components, other.source_path_components)?,
source_entries_visited: add(self.source_entries_visited, other.source_entries_visited)?,
output_bytes: add(self.output_bytes, other.output_bytes)?,
items_examined: add(self.items_examined, other.items_examined)?,
items_returned: add(self.items_returned, other.items_returned)?,
allocation_operations: add(self.allocation_operations, other.allocation_operations)?,
peak_allocation_bytes: self.peak_allocation_bytes.max(other.peak_allocation_bytes),
materializations: add(self.materializations, other.materializations)?,
})
}
pub fn verify(self, budget: WorkBudget) -> Result<(), WorkError> {
let fields = [
(
"authority_records_read",
self.authority_records_read,
budget.authority_records_read,
),
(
"authority_records_appended",
self.authority_records_appended,
budget.authority_records_appended,
),
(
"authority_bytes_read",
self.authority_bytes_read,
budget.authority_bytes_read,
),
(
"authority_bytes_written",
self.authority_bytes_written,
budget.authority_bytes_written,
),
("object_probes", self.object_probes, budget.object_probes),
(
"backend_read_operations",
self.backend_read_operations,
budget.backend_read_operations,
),
(
"backend_write_operations",
self.backend_write_operations,
budget.backend_write_operations,
),
(
"durability_operations",
self.durability_operations,
budget.durability_operations,
),
("page_reads", self.page_reads, budget.page_reads),
("page_writes", self.page_writes, budget.page_writes),
(
"object_bytes_read",
self.object_bytes_read,
budget.object_bytes_read,
),
(
"object_bytes_written",
self.object_bytes_written,
budget.object_bytes_written,
),
("bytes_hashed", self.bytes_hashed, budget.bytes_hashed),
("bytes_copied", self.bytes_copied, budget.bytes_copied),
("bytes_encoded", self.bytes_encoded, budget.bytes_encoded),
(
"source_bytes_read",
self.source_bytes_read,
budget.source_bytes_read,
),
(
"source_path_components",
self.source_path_components,
budget.source_path_components,
),
(
"source_entries_visited",
self.source_entries_visited,
budget.source_entries_visited,
),
("output_bytes", self.output_bytes, budget.output_bytes),
("items_examined", self.items_examined, budget.items_examined),
("items_returned", self.items_returned, budget.items_returned),
(
"allocation_operations",
self.allocation_operations,
budget.allocation_operations,
),
(
"peak_allocation_bytes",
self.peak_allocation_bytes,
budget.peak_allocation_bytes,
),
(
"materializations",
self.materializations,
budget.materializations,
),
];
for (counter, observed, maximum) in fields {
if observed > maximum {
return Err(WorkError::BudgetExceeded {
counter,
observed,
maximum,
});
}
}
Ok(())
}
pub fn remaining(self, budget: WorkBudget) -> Result<WorkBudget, WorkError> {
self.verify(budget)?;
Ok(WorkBudget {
authority_records_read: budget.authority_records_read - self.authority_records_read,
authority_records_appended: budget.authority_records_appended
- self.authority_records_appended,
authority_bytes_read: budget.authority_bytes_read - self.authority_bytes_read,
authority_bytes_written: budget.authority_bytes_written - self.authority_bytes_written,
object_probes: budget.object_probes - self.object_probes,
backend_read_operations: budget.backend_read_operations - self.backend_read_operations,
backend_write_operations: budget.backend_write_operations
- self.backend_write_operations,
durability_operations: budget.durability_operations - self.durability_operations,
page_reads: budget.page_reads - self.page_reads,
page_writes: budget.page_writes - self.page_writes,
object_bytes_read: budget.object_bytes_read - self.object_bytes_read,
object_bytes_written: budget.object_bytes_written - self.object_bytes_written,
bytes_hashed: budget.bytes_hashed - self.bytes_hashed,
bytes_copied: budget.bytes_copied - self.bytes_copied,
bytes_encoded: budget.bytes_encoded - self.bytes_encoded,
source_bytes_read: budget.source_bytes_read - self.source_bytes_read,
source_path_components: budget.source_path_components - self.source_path_components,
source_entries_visited: budget.source_entries_visited - self.source_entries_visited,
output_bytes: budget.output_bytes - self.output_bytes,
items_examined: budget.items_examined - self.items_examined,
items_returned: budget.items_returned - self.items_returned,
allocation_operations: budget.allocation_operations - self.allocation_operations,
peak_allocation_bytes: budget.peak_allocation_bytes,
materializations: budget.materializations - self.materializations,
})
}
}
pub type WorkBudget = WorkCounters;
fn add(left: u64, right: u64) -> Result<u64, WorkError> {
left.checked_add(right).ok_or(WorkError::Overflow)
}
#[derive(Clone, Copy, Debug, Eq, Error, PartialEq)]
pub enum WorkError {
#[error("filesystem work accounting overflowed")]
Overflow,
#[error("work counter {counter} observed {observed}, exceeding {maximum}")]
BudgetExceeded {
counter: &'static str,
observed: u64,
maximum: u64,
},
}
#[cfg(test)]
#[path = "tests/performance.rs"]
mod tests;