use std::collections::HashMap;
use std::fmt;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, Mutex, MutexGuard, PoisonError};
use crate::runtime::fs::FileIdentity;
#[derive(Debug)]
pub struct DiskQuota {
limit: u64,
used: AtomicU64,
unmeasured: bool,
open: Mutex<HashMap<FileIdentity, OpenFile>>,
}
#[derive(Debug, Default)]
struct OpenFile {
readers: u32,
writer: bool,
pending: u64,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Holder {
Reader,
Writer,
}
impl DiskQuota {
pub fn new(limit: u64, used: u64) -> Self {
Self {
limit,
used: AtomicU64::new(used),
unmeasured: false,
open: Mutex::default(),
}
}
pub fn unmeasured(limit: u64) -> Self {
Self {
unmeasured: true,
..Self::new(limit, 0)
}
}
pub fn limit(&self) -> u64 {
self.limit
}
pub fn used(&self) -> u64 {
self.used.load(Ordering::Acquire)
}
pub fn is_unmeasured(&self) -> bool {
self.unmeasured
}
pub fn reserve(&self, bytes: u64) -> Result<(), QuotaExceeded> {
if bytes == 0 {
return Ok(());
}
if self.unmeasured {
return Err(QuotaExceeded::Unmeasured { limit: self.limit });
}
let mut used = self.used.load(Ordering::Acquire);
loop {
let Some(next) = used.checked_add(bytes).filter(|next| *next <= self.limit) else {
return Err(QuotaExceeded::Over {
limit: self.limit,
used,
requested: bytes,
});
};
match self
.used
.compare_exchange_weak(used, next, Ordering::AcqRel, Ordering::Acquire)
{
Ok(_) => return Ok(()),
Err(actual) => used = actual,
}
}
}
pub fn record(&self, bytes: u64) {
let _ = self
.used
.fetch_update(Ordering::AcqRel, Ordering::Acquire, |used| {
Some(used.saturating_add(bytes))
});
}
pub fn release(&self, bytes: u64) {
let _ = self
.used
.fetch_update(Ordering::AcqRel, Ordering::Acquire, |used| {
Some(used.saturating_sub(bytes))
});
}
pub fn hold(self: &Arc<Self>, file: FileIdentity, holder: Holder) -> OpenFileGuard {
let mut open = self.open_files();
let entry = open.entry(file).or_default();
match holder {
Holder::Reader => entry.readers = entry.readers.saturating_add(1),
Holder::Writer => entry.writer = true,
}
OpenFileGuard {
quota: Arc::clone(self),
file,
holder,
}
}
pub fn is_held(&self, file: FileIdentity) -> bool {
self.open_files().contains_key(&file)
}
fn is_writers_own(&self, file: FileIdentity) -> bool {
self.open_files().get(&file).is_some_and(|held| held.writer)
}
pub fn release_file(&self, file: FileIdentity, bytes: u64) {
let mut open = self.open_files();
match open.get_mut(&file) {
Some(held) if held.writer => {}
Some(held) => held.pending = held.pending.saturating_add(bytes),
None => {
drop(open);
self.release(bytes);
}
}
}
fn release_writers_own_file(&self, file: FileIdentity, bytes: u64) {
let mut open = self.open_files();
match open.get_mut(&file) {
Some(held) if held.readers > 0 => held.pending = held.pending.saturating_add(bytes),
_ => {
drop(open);
self.release(bytes);
}
}
}
fn open_files(&self) -> MutexGuard<'_, HashMap<FileIdentity, OpenFile>> {
self.open.lock().unwrap_or_else(PoisonError::into_inner)
}
}
#[derive(Debug)]
pub struct OpenFileGuard {
quota: Arc<DiskQuota>,
file: FileIdentity,
holder: Holder,
}
impl Drop for OpenFileGuard {
fn drop(&mut self) {
let mut open = self.quota.open_files();
let Some(held) = open.get_mut(&self.file) else {
return;
};
match self.holder {
Holder::Reader => held.readers = held.readers.saturating_sub(1),
Holder::Writer => held.writer = false,
}
if held.readers == 0 && !held.writer {
let pending = held.pending;
open.remove(&self.file);
drop(open);
self.quota.release(pending);
}
}
}
#[derive(Debug)]
pub struct QuotaCharge {
quota: Option<Arc<DiskQuota>>,
allowance: u64,
drawn: u64,
reserved: u64,
replaces: Replaces,
}
#[derive(Debug, Clone, Copy)]
enum Replaces {
Nothing,
File(FileIdentity, u64),
InPlace,
}
impl Replaces {
fn file(replaced: Option<(FileIdentity, u64)>) -> Self {
replaced.map_or(Self::Nothing, |(file, bytes)| Self::File(file, bytes))
}
}
impl QuotaCharge {
pub fn new(quota: Option<Arc<DiskQuota>>, replaced: Option<(FileIdentity, u64)>) -> Self {
let allowance = match ("a, replaced) {
(Some(quota), Some((file, bytes))) if !quota.is_held(file) => bytes,
_ => 0,
};
Self {
quota,
allowance,
drawn: 0,
reserved: 0,
replaces: Replaces::file(replaced),
}
}
pub fn in_place(quota: Option<Arc<DiskQuota>>, rewritten: Option<(FileIdentity, u64)>) -> Self {
let (allowance, replaces) = match ("a, rewritten) {
(Some(quota), Some((file, bytes))) if !quota.is_writers_own(file) => {
(bytes, Replaces::InPlace)
}
_ => (0, Replaces::Nothing),
};
Self {
quota,
allowance,
drawn: 0,
reserved: 0,
replaces,
}
}
pub fn reserve(&mut self, bytes: u64) -> Result<(), QuotaExceeded> {
let from_allowance = bytes.min(self.allowance.saturating_sub(self.drawn));
let beyond = bytes.saturating_sub(from_allowance);
if let Some(quota) = &self.quota {
quota.reserve(beyond)?;
}
self.drawn = self.drawn.saturating_add(from_allowance);
self.reserved = self.reserved.saturating_add(beyond);
Ok(())
}
pub fn unreserve(&mut self, bytes: u64) {
let from_reserved = bytes.min(self.reserved);
if let Some(quota) = &self.quota {
quota.release(from_reserved);
}
self.reserved = self.reserved.saturating_sub(from_reserved);
self.drawn = self
.drawn
.saturating_sub(bytes.saturating_sub(from_reserved));
}
pub fn cover(&mut self, replaced: Option<(FileIdentity, u64)>) -> Result<(), QuotaExceeded> {
let replaced_bytes = replaced.map_or(0, |(_, bytes)| bytes);
let uncovered = self.drawn.saturating_sub(replaced_bytes);
if let Some(quota) = &self.quota {
quota.reserve(uncovered)?;
}
self.drawn = self.drawn.saturating_sub(uncovered);
self.reserved = self.reserved.saturating_add(uncovered);
self.replaces = Replaces::file(replaced);
Ok(())
}
pub fn commit(mut self) {
let Some(quota) = self.quota.take() else {
return;
};
match self.replaces {
Replaces::Nothing => {}
Replaces::File(file, bytes) => {
quota.release_file(file, bytes.saturating_sub(self.drawn));
}
Replaces::InPlace => quota.release(self.allowance.saturating_sub(self.drawn)),
}
}
pub fn settle_at(mut self, left: u64) {
let Some(quota) = self.quota.take() else {
return;
};
match self.replaces {
Replaces::InPlace => {
let counted = self.allowance.saturating_add(self.reserved);
if left < counted {
quota.release(counted - left);
} else {
quota.record(left - counted);
}
}
Replaces::Nothing | Replaces::File(..) => {
quota.release(self.reserved.saturating_sub(left));
}
}
}
pub fn keep(mut self) {
self.quota = None;
}
pub fn release_when_closed(mut self, written: FileIdentity) {
if let Some(quota) = self.quota.take() {
quota.release_writers_own_file(written, self.reserved);
}
}
}
impl Drop for QuotaCharge {
fn drop(&mut self) {
if let Some(quota) = self.quota.take() {
quota.release(self.reserved);
}
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum QuotaExceeded {
Over {
limit: u64,
used: u64,
requested: u64,
},
Unmeasured { limit: u64 },
}
impl fmt::Display for QuotaExceeded {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
Self::Over {
limit,
used,
requested,
} => write!(
f,
"the filesystem's size limit of {limit} bytes would be exceeded: {used} bytes are in use and this needs {requested} more"
),
Self::Unmeasured { limit } => write!(
f,
"the filesystem couldn't be measured against its size limit of {limit} bytes, so it is treated as full until a run can measure it"
),
}
}
}
impl std::error::Error for QuotaExceeded {}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn quota_reserves_within_the_limit_and_refuses_past_it() {
let quota = DiskQuota::new(100, 40);
quota.reserve(60).expect("fits exactly");
let refused = quota.reserve(1).expect_err("one byte over");
assert_eq!(
refused,
QuotaExceeded::Over {
limit: 100,
used: 100,
requested: 1,
}
);
quota.release(30);
assert_eq!(quota.used(), 70);
quota.release(1_000);
assert_eq!(quota.used(), 0, "release never underflows");
}
#[test]
fn a_dropped_charge_gives_back_what_it_reserved() {
let quota = Arc::new(DiskQuota::new(100, 0));
let mut charge = QuotaCharge::new(Some(Arc::clone("a)), None);
charge.reserve(70).expect("fits");
assert_eq!(quota.used(), 70);
drop(charge);
assert_eq!(quota.used(), 0);
}
#[test]
fn a_charge_covers_a_replaced_file_that_shrank() {
let quota = Arc::new(DiskQuota::new(100, 90));
let file = identity();
let mut charge = QuotaCharge::new(Some(Arc::clone("a)), Some((file, 90)));
charge.reserve(90).expect("drawn from the allowance");
assert_eq!(quota.used(), 90);
quota.release(90);
charge
.cover(Some((file, 0)))
.expect("fits once the file is empty");
charge.commit();
assert_eq!(quota.used(), 90);
let mut again = QuotaCharge::new(Some(Arc::clone("a)), Some((file, 90)));
again.reserve(90).expect("drawn from the allowance");
assert!(
again.cover(Some((file, 0))).is_err(),
"the file shrank and nothing freed its bytes"
);
}
#[test]
fn a_held_file_gives_no_allowance() {
let quota = Arc::new(DiskQuota::new(100, 60));
let file = identity();
let reader = quota.hold(file, Holder::Reader);
let mut charge = QuotaCharge::new(Some(Arc::clone("a)), Some((file, 60)));
assert!(
charge.reserve(60).is_err(),
"the replaced file stays on disk while the reader holds it"
);
charge.reserve(40).expect("the growth beyond it fits");
charge.commit();
assert_eq!(quota.used(), 100, "the old file is still counted");
drop(reader);
assert_eq!(quota.used(), 40, "and freed when the reader closes");
}
#[test]
fn a_file_rewritten_in_place_frees_its_old_bytes_while_held() {
let quota = Arc::new(DiskQuota::new(100, 60));
let file = identity();
let _reader = quota.hold(file, Holder::Reader);
let mut charge = QuotaCharge::in_place(Some(Arc::clone("a)), Some((file, 60)));
charge.reserve(30).expect("drawn from the old contents");
charge.commit();
assert_eq!(quota.used(), 30, "the old contents are gone, held or not");
}
#[test]
fn a_failed_rewrite_in_place_counts_what_it_left() {
let quota = Arc::new(DiskQuota::new(100, 40));
let mut charge = QuotaCharge::in_place(Some(Arc::clone("a)), Some((identity(), 40)));
charge.reserve(50).expect("room for the new contents");
assert_eq!(quota.used(), 50);
charge.settle_at(20);
assert_eq!(
quota.used(),
20,
"the old contents are gone, 20 new bytes landed"
);
let untouched = Arc::new(DiskQuota::new(100, 40));
let mut charge =
QuotaCharge::in_place(Some(Arc::clone(&untouched)), Some((identity(), 40)));
charge.reserve(10).expect("drawn from the old contents");
charge.settle_at(40);
assert_eq!(
untouched.used(),
40,
"a copy that failed before truncating left the file"
);
}
#[test]
fn a_writers_own_file_rewritten_in_place_is_left_for_the_writer() {
let quota = Arc::new(DiskQuota::new(100, 60));
let file = identity();
let _writer = quota.hold(file, Holder::Writer);
let mut charge = QuotaCharge::in_place(Some(Arc::clone("a)), Some((file, 60)));
charge.reserve(30).expect("room for the new contents");
charge.commit();
assert_eq!(
quota.used(),
90,
"the writer's charge still holds the old contents"
);
}
#[test]
fn a_removed_file_held_open_stays_counted_until_it_closes() {
let quota = Arc::new(DiskQuota::new(100, 60));
let file = identity();
let reader = quota.hold(file, Holder::Reader);
quota.release_file(file, 60);
assert_eq!(quota.used(), 60, "an open file's bytes are still on disk");
drop(reader);
assert_eq!(quota.used(), 0);
}
#[test]
fn a_writers_own_file_is_left_for_the_writer_to_settle() {
let quota = Arc::new(DiskQuota::new(100, 0));
let file = identity();
let writer = quota.hold(file, Holder::Writer);
quota.record(40);
quota.release_file(file, 40);
assert_eq!(quota.used(), 40);
drop(writer);
assert_eq!(
quota.used(),
40,
"releasing it is the writer's charge's job"
);
}
fn identity() -> FileIdentity {
let file = tempfile::tempfile().expect("tempfile");
let metadata = cap_std::fs::Metadata::from_file(&file).expect("metadata");
FileIdentity::of(&metadata).expect("identity")
}
}