use serde::{Deserialize, Serialize};
use crate::error::{FootprintOverflow, MemoryError, MemoryResult};
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct MemoryDecision {
pub label: String,
pub fixed_bytes: u64,
pub bytes_per_event: u64,
pub chunk_events: usize,
pub estimated_peak_bytes: u64,
pub actual_high_water_bytes: Option<u64>,
pub strategy: String,
}
#[doc(hidden)]
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct MemoryFootprint {
pub fixed_bytes: u64,
pub bytes_per_event: u64,
}
impl MemoryFootprint {
pub const fn new(fixed_bytes: u64, bytes_per_event: u64) -> Self {
Self {
fixed_bytes,
bytes_per_event,
}
}
pub const fn fixed(fixed_bytes: u64) -> Self {
Self::new(fixed_bytes, 0)
}
pub const fn per_event(bytes_per_event: u64) -> Self {
Self::new(0, bytes_per_event)
}
pub fn from_usize(fixed_bytes: usize, bytes_per_event: usize) -> Self {
Self::from_usize_checked(fixed_bytes, bytes_per_event)
.unwrap_or(Self::new(u64::MAX, u64::MAX))
}
pub fn from_usize_checked(
fixed_bytes: usize,
bytes_per_event: usize,
) -> Result<Self, FootprintOverflow> {
Ok(Self::new(
checked_u64(fixed_bytes)?,
checked_u64(bytes_per_event)?,
))
}
pub const fn checked_add(self, other: Self) -> Result<Self, FootprintOverflow> {
let fixed_bytes = match self.fixed_bytes.checked_add(other.fixed_bytes) {
Some(value) => value,
None => return Err(FootprintOverflow::Addition),
};
let bytes_per_event = match self.bytes_per_event.checked_add(other.bytes_per_event) {
Some(value) => value,
None => return Err(FootprintOverflow::Addition),
};
Ok(Self::new(fixed_bytes, bytes_per_event))
}
pub const fn checked_scale(self, factor: u64) -> Result<Self, FootprintOverflow> {
let fixed_bytes = match self.fixed_bytes.checked_mul(factor) {
Some(value) => value,
None => return Err(FootprintOverflow::Multiplication),
};
let bytes_per_event = match self.bytes_per_event.checked_mul(factor) {
Some(value) => value,
None => return Err(FootprintOverflow::Multiplication),
};
Ok(Self::new(fixed_bytes, bytes_per_event))
}
pub fn checked_scale_usize(self, factor: usize) -> Result<Self, FootprintOverflow> {
self.checked_scale(checked_u64(factor)?)
}
pub fn checked_peak_bytes(self, events: usize) -> Result<u64, FootprintOverflow> {
let events = checked_u64(events)?;
let event_bytes = self
.bytes_per_event
.checked_mul(events)
.ok_or(FootprintOverflow::Multiplication)?;
self.fixed_bytes
.checked_add(event_bytes)
.ok_or(FootprintOverflow::Addition)
}
pub fn peak_bytes(self, events: usize) -> u64 {
self.checked_peak_bytes(events).unwrap_or(u64::MAX)
}
fn normalized(mut self) -> Self {
self.bytes_per_event = self.bytes_per_event.max(1);
self
}
}
#[doc(hidden)]
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct MemoryFitRequest {
pub label: String,
pub footprint: MemoryFootprint,
pub available_bytes: u64,
pub event_limit: usize,
pub strategy: String,
}
impl MemoryDecision {
pub fn fit(
label: impl Into<String>,
fixed_bytes: u64,
bytes_per_event: u64,
available_bytes: u64,
event_limit: usize,
strategy: impl Into<String>,
) -> MemoryResult<Self> {
MemoryFitRequest {
label: label.into(),
footprint: MemoryFootprint::new(fixed_bytes, bytes_per_event),
available_bytes,
event_limit,
strategy: strategy.into(),
}
.evaluate()
}
}
impl MemoryFitRequest {
pub fn evaluate(mut self) -> MemoryResult<MemoryDecision> {
self.footprint = self.footprint.normalized();
if self.event_limit == 0 {
return Ok(self.decision(0));
}
let event_capacity = self
.available_bytes
.saturating_sub(self.footprint.fixed_bytes);
let events =
saturating_usize(event_capacity / self.footprint.bytes_per_event).min(self.event_limit);
if events == 0 {
return Err(MemoryError::BudgetExceeded {
resource: self.label,
requested: self
.footprint
.fixed_bytes
.saturating_add(self.footprint.bytes_per_event),
remaining: self.available_bytes,
});
}
Ok(self.decision(events))
}
pub fn evaluate_resident(self, chunk_events: usize) -> MemoryResult<MemoryDecision> {
let peak = self.footprint.peak_bytes(self.event_limit);
if peak > self.available_bytes {
return Err(MemoryError::BudgetExceeded {
resource: self.label,
requested: peak,
remaining: self.available_bytes,
});
}
Ok(self.decision_with_peak(chunk_events, peak))
}
fn decision(self, events: usize) -> MemoryDecision {
let peak = self.footprint.peak_bytes(events);
self.decision_with_peak(events, peak)
}
fn decision_with_peak(self, chunk_events: usize, peak: u64) -> MemoryDecision {
MemoryDecision {
label: self.label,
fixed_bytes: self.footprint.fixed_bytes,
bytes_per_event: self.footprint.bytes_per_event,
chunk_events,
estimated_peak_bytes: peak,
actual_high_water_bytes: None,
strategy: self.strategy,
}
}
}
fn saturating_usize(value: u64) -> usize {
usize::try_from(value).unwrap_or(usize::MAX)
}
fn checked_u64(value: usize) -> Result<u64, FootprintOverflow> {
u64::try_from(value).map_err(|_| FootprintOverflow::Conversion)
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn fits_the_largest_safe_chunk() {
let decision = MemoryDecision::fit("test", 100, 8, 1_000, 1_000, "streaming").unwrap();
assert_eq!(decision.chunk_events, 112);
assert_eq!(decision.estimated_peak_bytes, 996);
}
#[test]
fn defines_zero_and_capacity_boundary_policy() {
let zero = MemoryDecision::fit("zero", 101, 0, 100, 0, "empty").unwrap();
assert_eq!(
(
zero.bytes_per_event,
zero.chunk_events,
zero.estimated_peak_bytes
),
(1, 0, 101)
);
let normalized = MemoryDecision::fit("normalized", 0, 0, 100, 200, "streaming").unwrap();
assert_eq!(
(
normalized.bytes_per_event,
normalized.chunk_events,
normalized.estimated_peak_bytes
),
(1, 100, 100)
);
assert!(MemoryDecision::fit("full", 100, 1, 100, 1, "streaming").is_err());
assert!(MemoryDecision::fit("full", 101, 1, 100, 1, "streaming").is_err());
}
#[test]
fn saturates_at_integer_boundaries() {
let decision =
MemoryDecision::fit("maximum", 0, 1, u64::MAX, usize::MAX, "resident").unwrap();
assert_eq!(
decision.chunk_events,
usize::try_from(u64::MAX).unwrap_or(usize::MAX)
);
assert_eq!(decision.estimated_peak_bytes, u64::MAX);
assert_eq!(
MemoryFootprint::new(u64::MAX, u64::MAX).peak_bytes(usize::MAX),
u64::MAX
);
assert_eq!(
MemoryDecision::fit("overflow", u64::MAX, 1, u64::MAX, 1, "streaming"),
Err(MemoryError::BudgetExceeded {
resource: "overflow".into(),
requested: u64::MAX,
remaining: u64::MAX,
})
);
}
#[test]
fn named_and_resident_requests_share_policy() {
let request = MemoryFitRequest {
label: "named".into(),
footprint: MemoryFootprint::from_usize(100, 8),
available_bytes: 1_000,
event_limit: 100,
strategy: "resident".into(),
};
assert_eq!(request.clone().evaluate().unwrap().chunk_events, 100);
let decision = request.evaluate_resident(25).unwrap();
assert_eq!(
(decision.chunk_events, decision.estimated_peak_bytes),
(25, 900)
);
}
#[test]
fn resident_requests_check_the_full_footprint() {
let request = MemoryFitRequest {
label: "resident".into(),
footprint: MemoryFootprint::new(100, 8),
available_bytes: 899,
event_limit: 100,
strategy: "resident".into(),
};
assert_eq!(
request.evaluate_resident(25),
Err(MemoryError::BudgetExceeded {
resource: "resident".into(),
requested: 900,
remaining: 899,
})
);
let zero_per_event = MemoryFitRequest {
label: "resident".into(),
footprint: MemoryFootprint::new(100, 0),
available_bytes: 100,
event_limit: 100,
strategy: "resident".into(),
}
.evaluate_resident(25)
.unwrap();
assert_eq!(
(
zero_per_event.bytes_per_event,
zero_per_event.estimated_peak_bytes
),
(0, 100)
);
}
}