use super::{ResidencyPermit, ResidencyReason};
use crate::AsyncObjectStore;
use crate::cancellation::CancellationToken;
use crate::foundation::{GenerationId, OperationId, VolumeId};
use crate::performance::{WorkBudget, WorkCounters, WorkError};
use crate::storage::{
ObjectFailure, ObjectId, ObjectReadRequest, ObjectReadRetention, ObjectReceipt, ObjectResult,
ObjectStoreError,
};
use std::collections::{BTreeMap, VecDeque};
use std::future::Future;
use thiserror::Error;
const BASIS_POINTS: u128 = 10_000;
#[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
pub struct StorageLocationId([u8; 16]);
impl StorageLocationId {
#[must_use]
pub const fn from_bytes(bytes: [u8; 16]) -> Self {
Self(bytes)
}
#[must_use]
pub const fn into_bytes(self) -> [u8; 16] {
self.0
}
}
#[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
pub enum StorageTier {
ProcessMemory,
NodeLocal,
SharedCache,
DurableOrigin,
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub struct ObjectResidency {
pub object_id: ObjectId,
pub location_id: StorageLocationId,
pub tier: StorageTier,
pub source_priority: u16,
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub struct PromotionDestination {
pub location_id: StorageLocationId,
pub tier: StorageTier,
pub writable: bool,
pub maximum_object_bytes: u64,
pub priority: u16,
pub cost_units_per_byte: u64,
}
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct PromotionCandidate {
pub operation_id: OperationId,
pub volume_id: VolumeId,
pub generation_id: GenerationId,
pub request: ObjectReadRequest,
pub accepted_tiers: Vec<StorageTier>,
pub reason: ResidencyReason,
}
impl PromotionCandidate {
#[must_use]
pub fn from_residency(permit: ResidencyPermit, accepted_tiers: Vec<StorageTier>) -> Self {
let candidate = permit.candidate();
Self {
operation_id: candidate.operation_id,
volume_id: candidate.volume_id,
generation_id: candidate.generation_id,
request: candidate.request,
accepted_tiers,
reason: candidate.reason,
}
}
}
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct PromotionPlan {
pub candidate: PromotionCandidate,
pub source: ObjectResidency,
pub destination: PromotionDestination,
pub estimated_cost_units: u64,
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub struct PromotionSpeculatorOptions {
pub maximum_active_operations: u32,
pub maximum_active_bytes: u64,
pub maximum_active_cost_units: u64,
pub maximum_residency_facts: u32,
pub maximum_destinations: u32,
pub maximum_accepted_tiers: u32,
pub outcome_window: u32,
pub minimum_usefulness_samples: u32,
pub minimum_usefulness_basis_points: u16,
}
impl Default for PromotionSpeculatorOptions {
fn default() -> Self {
Self {
maximum_active_operations: 4,
maximum_active_bytes: 16 * 1024 * 1024,
maximum_active_cost_units: u64::MAX,
maximum_residency_facts: 64,
maximum_destinations: 32,
maximum_accepted_tiers: 4,
outcome_window: 32,
minimum_usefulness_samples: 4,
minimum_usefulness_basis_points: 5_000,
}
}
}
impl PromotionSpeculatorOptions {
fn validate(self) -> Result<Self, PromotionSpeculatorError> {
if self.maximum_active_operations == 0
|| self.maximum_active_bytes == 0
|| self.maximum_active_cost_units == 0
|| self.maximum_residency_facts == 0
|| self.maximum_destinations == 0
|| self.maximum_accepted_tiers == 0
|| self.outcome_window == 0
|| self.minimum_usefulness_samples == 0
{
return Err(PromotionSpeculatorError::ZeroBound);
}
if self.minimum_usefulness_samples > self.outcome_window {
return Err(PromotionSpeculatorError::UsefulnessWindowTooSmall);
}
if u128::from(self.minimum_usefulness_basis_points) > BASIS_POINTS {
return Err(PromotionSpeculatorError::InvalidBasisPoints);
}
Ok(self)
}
}
#[derive(Clone, Copy, Debug, Error, Eq, PartialEq)]
pub enum PromotionSpeculatorError {
#[error("promotion speculation bounds must be positive")]
ZeroBound,
#[error("promotion usefulness window is smaller than its minimum sample count")]
UsefulnessWindowTooSmall,
#[error("promotion usefulness basis points exceed 10,000")]
InvalidBasisPoints,
#[error("promotion speculation accounting overflowed")]
Overflow,
#[error("promotion speculation plan is not active")]
UnknownPlan,
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum PromotionRejection {
WrongVolume,
StaleGeneration,
InvalidRequest,
InputCapacity,
MissingSource,
DuplicateObject,
DuplicateOperation,
ActiveCapacity,
NoDestination,
LowUsefulness,
}
#[derive(Clone, Debug, Eq, PartialEq)]
pub enum PromotionAdmission {
Satisfied(ObjectResidency),
Planned(PromotionPlan),
Rejected(PromotionRejection),
}
#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
pub struct PromotionMetrics {
pub candidates: u64,
pub satisfied: u64,
pub planned: u64,
pub active: u64,
pub active_bytes: u64,
pub active_cost_units: u64,
pub useful: u64,
pub wasted: u64,
pub rejected: u64,
}
#[derive(Clone)]
pub struct PromotionSpeculator {
options: PromotionSpeculatorOptions,
volume_id: VolumeId,
generation_id: GenerationId,
active: BTreeMap<OperationId, PromotionPlan>,
outcomes: VecDeque<bool>,
metrics: PromotionMetrics,
}
impl PromotionSpeculator {
pub fn new(
options: PromotionSpeculatorOptions,
volume_id: VolumeId,
generation_id: GenerationId,
) -> Result<Self, PromotionSpeculatorError> {
Ok(Self {
options: options.validate()?,
volume_id,
generation_id,
active: BTreeMap::new(),
outcomes: VecDeque::new(),
metrics: PromotionMetrics::default(),
})
}
pub fn plan(
&mut self,
candidate: PromotionCandidate,
residency: &[ObjectResidency],
destinations: &[PromotionDestination],
) -> Result<PromotionAdmission, PromotionSpeculatorError> {
let mut next = self.clone();
let admission = next.plan_in_place(candidate, residency, destinations)?;
*self = next;
Ok(admission)
}
fn plan_in_place(
&mut self,
candidate: PromotionCandidate,
residency: &[ObjectResidency],
destinations: &[PromotionDestination],
) -> Result<PromotionAdmission, PromotionSpeculatorError> {
self.metrics.candidates = checked_increment(self.metrics.candidates)?;
if let Some(rejection) = self.basic_rejection(&candidate, residency, destinations) {
return self.reject(rejection);
}
if let Some(current) = best_satisfied(&candidate, residency) {
self.metrics.satisfied = checked_increment(self.metrics.satisfied)?;
return Ok(PromotionAdmission::Satisfied(current));
}
let Some(source) = best_source(candidate.request.object_id, residency) else {
return self.reject(PromotionRejection::MissingSource);
};
let Some(destination) = best_destination(&candidate, source.location_id, destinations)
else {
return self.reject(PromotionRejection::NoDestination);
};
if !self.useful_enough() {
return self.reject(PromotionRejection::LowUsefulness);
}
let estimated_cost_units = candidate
.request
.maximum_bytes
.checked_mul(destination.cost_units_per_byte)
.ok_or(PromotionSpeculatorError::Overflow)?;
if self.active.len() >= self.options.maximum_active_operations as usize
|| self
.metrics
.active_bytes
.checked_add(candidate.request.maximum_bytes)
.is_none_or(|bytes| bytes > self.options.maximum_active_bytes)
|| self
.metrics
.active_cost_units
.checked_add(estimated_cost_units)
.is_none_or(|cost| cost > self.options.maximum_active_cost_units)
{
return self.reject(PromotionRejection::ActiveCapacity);
}
let plan = PromotionPlan {
candidate,
source,
destination,
estimated_cost_units,
};
self.metrics.planned = checked_increment(self.metrics.planned)?;
self.metrics.active = checked_increment(self.metrics.active)?;
self.metrics.active_bytes = self
.metrics
.active_bytes
.checked_add(plan.candidate.request.maximum_bytes)
.ok_or(PromotionSpeculatorError::Overflow)?;
self.metrics.active_cost_units = self
.metrics
.active_cost_units
.checked_add(plan.estimated_cost_units)
.ok_or(PromotionSpeculatorError::Overflow)?;
self.active
.insert(plan.candidate.operation_id, plan.clone());
Ok(PromotionAdmission::Planned(plan))
}
pub fn finish(
&mut self,
operation_id: OperationId,
useful: bool,
) -> Result<(), PromotionSpeculatorError> {
let mut next = self.clone();
next.finish_in_place(operation_id, useful)?;
*self = next;
Ok(())
}
fn finish_in_place(
&mut self,
operation_id: OperationId,
useful: bool,
) -> Result<(), PromotionSpeculatorError> {
let plan = self
.active
.remove(&operation_id)
.ok_or(PromotionSpeculatorError::UnknownPlan)?;
self.metrics.active = self
.metrics
.active
.checked_sub(1)
.ok_or(PromotionSpeculatorError::Overflow)?;
self.metrics.active_bytes = self
.metrics
.active_bytes
.checked_sub(plan.candidate.request.maximum_bytes)
.ok_or(PromotionSpeculatorError::Overflow)?;
self.metrics.active_cost_units = self
.metrics
.active_cost_units
.checked_sub(plan.estimated_cost_units)
.ok_or(PromotionSpeculatorError::Overflow)?;
if useful {
self.metrics.useful = checked_increment(self.metrics.useful)?;
} else {
self.metrics.wasted = checked_increment(self.metrics.wasted)?;
}
self.outcomes.push_back(useful);
while self.outcomes.len() > self.options.outcome_window as usize {
self.outcomes.pop_front();
}
Ok(())
}
pub fn preempt_for_foreground(&mut self) -> Result<Vec<OperationId>, PromotionSpeculatorError> {
let mut next = self.clone();
let operations = next.preempt_in_place()?;
*self = next;
Ok(operations)
}
fn preempt_in_place(&mut self) -> Result<Vec<OperationId>, PromotionSpeculatorError> {
let operations: Vec<_> = self.active.keys().copied().collect();
for operation_id in &operations {
self.finish_in_place(*operation_id, false)?;
}
Ok(operations)
}
pub fn replace_generation(
&mut self,
generation_id: GenerationId,
) -> Result<Vec<OperationId>, PromotionSpeculatorError> {
let mut next = self.clone();
let preempted = next.preempt_in_place()?;
next.generation_id = generation_id;
*self = next;
Ok(preempted)
}
#[must_use]
pub const fn metrics(&self) -> PromotionMetrics {
self.metrics
}
fn basic_rejection(
&self,
candidate: &PromotionCandidate,
residency: &[ObjectResidency],
destinations: &[PromotionDestination],
) -> Option<PromotionRejection> {
if candidate.volume_id != self.volume_id {
return Some(PromotionRejection::WrongVolume);
}
if candidate.generation_id != self.generation_id {
return Some(PromotionRejection::StaleGeneration);
}
if candidate.request.maximum_bytes == 0
|| candidate.accepted_tiers.is_empty()
|| candidate.accepted_tiers.len() > self.options.maximum_accepted_tiers as usize
|| has_duplicate_tiers(&candidate.accepted_tiers)
|| residency
.iter()
.any(|fact| fact.object_id != candidate.request.object_id)
|| has_duplicate_residency_locations(residency)
|| has_duplicate_destination_locations(destinations)
{
return Some(PromotionRejection::InvalidRequest);
}
if residency.len() > self.options.maximum_residency_facts as usize
|| destinations.len() > self.options.maximum_destinations as usize
{
return Some(PromotionRejection::InputCapacity);
}
if self
.active
.values()
.any(|plan| plan.candidate.request.object_id == candidate.request.object_id)
{
return Some(PromotionRejection::DuplicateObject);
}
if self.active.contains_key(&candidate.operation_id) {
return Some(PromotionRejection::DuplicateOperation);
}
None
}
fn useful_enough(&self) -> bool {
if self.outcomes.len() < self.options.minimum_usefulness_samples as usize {
return true;
}
let useful = self.outcomes.iter().filter(|useful| **useful).count() as u128;
useful * BASIS_POINTS
>= self.outcomes.len() as u128
* u128::from(self.options.minimum_usefulness_basis_points)
}
fn reject(
&mut self,
rejection: PromotionRejection,
) -> Result<PromotionAdmission, PromotionSpeculatorError> {
self.metrics.rejected = checked_increment(self.metrics.rejected)?;
Ok(PromotionAdmission::Rejected(rejection))
}
}
pub trait PromotionExecutor {
fn promote(
&self,
plan: &PromotionPlan,
budget: WorkBudget,
cancellation: &CancellationToken,
) -> impl Future<Output = ObjectResult<()>>;
}
pub struct StorePromotionExecutor<S, D> {
source_location_id: StorageLocationId,
destination_location_id: StorageLocationId,
source: S,
destination: D,
}
impl<S, D> StorePromotionExecutor<S, D> {
pub fn new(
source_location_id: StorageLocationId,
source: S,
destination_location_id: StorageLocationId,
destination: D,
) -> Result<Self, StorePromotionExecutorError> {
if source_location_id == destination_location_id {
return Err(StorePromotionExecutorError::SameLocation);
}
Ok(Self {
source_location_id,
destination_location_id,
source,
destination,
})
}
#[must_use]
pub const fn source(&self) -> &S {
&self.source
}
#[must_use]
pub const fn destination(&self) -> &D {
&self.destination
}
}
#[derive(Clone, Copy, Debug, Error, Eq, PartialEq)]
pub enum StorePromotionExecutorError {
#[error("promotion source and destination locations must differ")]
SameLocation,
}
impl<S: AsyncObjectStore, D: AsyncObjectStore> PromotionExecutor for StorePromotionExecutor<S, D> {
async fn promote(
&self,
plan: &PromotionPlan,
budget: WorkBudget,
cancellation: &CancellationToken,
) -> ObjectResult<()> {
validate_executor_plan(plan, self.source_location_id, self.destination_location_id)?;
cancellation
.check()
.map_err(|_| ObjectFailure::before_work(ObjectStoreError::Cancelled))?;
let request = plan.candidate.request;
let read = self
.source
.read(
request.object_id,
request.maximum_bytes,
budget,
cancellation,
)
.await?;
let live_bytes = match read.value.retention {
ObjectReadRetention::Shared => 0,
ObjectReadRetention::Owned { logical_bytes } => logical_bytes,
};
let mut remaining = read
.work
.remaining(budget)
.map_err(|error| ObjectFailure::new(error.into(), read.work))?;
remaining.peak_allocation_bytes = remaining
.peak_allocation_bytes
.checked_sub(live_bytes)
.ok_or_else(|| {
ObjectFailure::new(
ObjectStoreError::Work(WorkError::BudgetExceeded {
counter: "peak_allocation_bytes",
observed: live_bytes,
maximum: budget.peak_allocation_bytes,
}),
read.work,
)
})?;
let prior = read.work;
let put = self
.destination
.put(request.object_id, read.value.bytes, remaining, cancellation)
.await
.map_err(|failure| merge_promotion_failure(prior, live_bytes, failure))?;
let work = merge_promotion_work(prior, live_bytes, put.work)
.map_err(|error| ObjectFailure::new(error.into(), prior))?;
work.verify(budget)
.map_err(|error| ObjectFailure::new(error.into(), work))?;
Ok(ObjectReceipt { value: (), work })
}
}
pub async fn execute_promotion<E: PromotionExecutor>(
speculator: &PromotionSpeculator,
executor: &E,
plan: &PromotionPlan,
budget: WorkBudget,
cancellation: &CancellationToken,
) -> ObjectResult<()> {
if speculator.active.get(&plan.candidate.operation_id) != Some(plan) {
return Err(crate::performance::OperationFailure::before_work(
ObjectStoreError::Rejected(
"promotion speculation plan is stale or inactive".to_owned(),
),
));
}
executor.promote(plan, budget, cancellation).await
}
fn validate_executor_plan(
plan: &PromotionPlan,
source_location_id: StorageLocationId,
destination_location_id: StorageLocationId,
) -> ObjectResult<()> {
if plan.source.location_id != source_location_id
|| plan.destination.location_id != destination_location_id
|| plan.source.object_id != plan.candidate.request.object_id
|| source_location_id == destination_location_id
{
return Err(ObjectFailure::before_work(ObjectStoreError::Rejected(
"promotion plan does not match configured stores".to_owned(),
)));
}
Ok(ObjectReceipt {
value: (),
work: WorkCounters::default(),
})
}
fn merge_promotion_work(
prior: WorkCounters,
live_bytes: u64,
nested: WorkCounters,
) -> Result<WorkCounters, WorkError> {
let simultaneous_peak = live_bytes
.checked_add(nested.peak_allocation_bytes)
.ok_or(WorkError::Overflow)?;
let mut combined = prior.checked_add(nested)?;
combined.peak_allocation_bytes = combined.peak_allocation_bytes.max(simultaneous_peak);
Ok(combined)
}
fn merge_promotion_failure(
prior: WorkCounters,
live_bytes: u64,
failure: ObjectFailure,
) -> ObjectFailure {
match merge_promotion_work(prior, live_bytes, *failure.work) {
Ok(combined) => ObjectFailure::new(failure.error, combined),
Err(error) => ObjectFailure::new(error.into(), prior),
}
}
fn best_satisfied(
candidate: &PromotionCandidate,
residency: &[ObjectResidency],
) -> Option<ObjectResidency> {
candidate.accepted_tiers.iter().find_map(|tier| {
residency
.iter()
.filter(|fact| fact.object_id == candidate.request.object_id && fact.tier == *tier)
.min_by_key(|fact| (fact.source_priority, fact.location_id))
.copied()
})
}
fn best_source(object_id: ObjectId, residency: &[ObjectResidency]) -> Option<ObjectResidency> {
residency
.iter()
.filter(|fact| fact.object_id == object_id)
.min_by_key(|fact| (fact.source_priority, fact.location_id))
.copied()
}
fn best_destination(
candidate: &PromotionCandidate,
source_location: StorageLocationId,
destinations: &[PromotionDestination],
) -> Option<PromotionDestination> {
candidate.accepted_tiers.iter().find_map(|tier| {
destinations
.iter()
.filter(|destination| {
destination.tier == *tier
&& destination.location_id != source_location
&& destination.writable
&& destination.maximum_object_bytes >= candidate.request.maximum_bytes
})
.min_by_key(|destination| (destination.priority, destination.location_id))
.copied()
})
}
fn has_duplicate_tiers(tiers: &[StorageTier]) -> bool {
tiers
.iter()
.enumerate()
.any(|(index, tier)| tiers.get(..index).unwrap_or(&[]).contains(tier))
}
fn has_duplicate_residency_locations(residency: &[ObjectResidency]) -> bool {
residency.iter().enumerate().any(|(index, fact)| {
residency
.get(..index)
.unwrap_or(&[])
.iter()
.any(|prior| prior.location_id == fact.location_id)
})
}
fn has_duplicate_destination_locations(destinations: &[PromotionDestination]) -> bool {
destinations.iter().enumerate().any(|(index, destination)| {
destinations
.get(..index)
.unwrap_or(&[])
.iter()
.any(|prior| prior.location_id == destination.location_id)
})
}
fn checked_increment(value: u64) -> Result<u64, PromotionSpeculatorError> {
value
.checked_add(1)
.ok_or(PromotionSpeculatorError::Overflow)
}
#[cfg(test)]
#[path = "tests/promotion.rs"]
mod tests;