lace_sans_io/runtime_support/
io_budget.rs

1//! Physical IO settings and the budget that bounds the IO buffers of
2//! connected operations and stores that stage received records.
3//!
4//! Each runtime owns one process-wide [`IoBudget`]. A connected operation
5//! reserves its codec's chunk buffers from it at setup and returns them once
6//! its physical IO settles; a Lace whose store copies file-range records
7//! reserves that copy buffer at adoption and returns it at close. Direct
8//! operations, in-memory record sinks and records are not counted.
9
10use std::sync::{Mutex, MutexGuard, PoisonError};
11
12use crate::core::ErrorCategory;
13use crate::iltp::{DEFAULT_CHUNK_BYTES, MIN_CHUNK_BYTES};
14use crate::runtime_values::Error;
15
16/// Runtime configuration of physical IO, not an operation option.
17#[derive(Debug, Clone, Copy, PartialEq, Eq)]
18pub struct IoSettings {
19    /// Bytes per transport read, output chunk, record read and store copy
20    /// piece; each buffer is allocated on first need and never longer than
21    /// its useful length. At least [`MIN_CHUNK_BYTES`].
22    pub chunk_bytes: usize,
23    /// Bytes that reserved IO buffers may hold at once; `usize::MAX` is
24    /// unbounded.
25    pub budget_bytes: usize,
26}
27
28impl IoSettings {
29    const DEFAULT: Self = Self {
30        chunk_bytes: DEFAULT_CHUNK_BYTES,
31        budget_bytes: usize::MAX,
32    };
33}
34
35impl Default for IoSettings {
36    fn default() -> Self {
37        Self::DEFAULT
38    }
39}
40
41struct IoState {
42    settings: IoSettings,
43    reserved: usize,
44}
45
46/// The current IO settings and the bytes reserved under them.
47pub struct IoBudget(Mutex<IoState>);
48
49impl Default for IoBudget {
50    fn default() -> Self {
51        Self::new()
52    }
53}
54
55impl IoBudget {
56    pub const fn new() -> Self {
57        Self(Mutex::new(IoState {
58            settings: IoSettings::DEFAULT,
59            reserved: 0,
60        }))
61    }
62
63    fn state(&self) -> MutexGuard<'_, IoState> {
64        self.0.lock().unwrap_or_else(PoisonError::into_inner)
65    }
66
67    /// Set the settings for work set up afterwards. Existing reservations
68    /// stay until they are returned; a lower budget only refuses new ones.
69    pub fn configure(&self, settings: IoSettings) -> Result<(), Error> {
70        if settings.chunk_bytes < MIN_CHUNK_BYTES {
71            return Err(Error::integration(format!(
72                "IO chunk size must be at least {MIN_CHUNK_BYTES} bytes"
73            )));
74        }
75        self.state().settings = settings;
76        Ok(())
77    }
78
79    /// The chunk size that work set up now uses.
80    pub fn chunk_bytes(&self) -> usize {
81        self.state().settings.chunk_bytes
82    }
83
84    /// Reserve `bytes`, or fail with `LimitExceeded` when the budget cannot
85    /// hold them.
86    pub fn reserve(&'static self, bytes: usize) -> Result<IoReservation, Error> {
87        let mut state = self.state();
88        match state.reserved.checked_add(bytes) {
89            Some(reserved) if reserved <= state.settings.budget_bytes => {
90                state.reserved = reserved;
91                Ok(IoReservation {
92                    budget: self,
93                    bytes,
94                })
95            }
96            _ => Err(Error::detailed(
97                ErrorCategory::LimitExceeded,
98                None,
99                None,
100                format!(
101                    "IO budget of {} bytes cannot reserve {bytes} more bytes ({} reserved)",
102                    state.settings.budget_bytes, state.reserved
103                ),
104            )),
105        }
106    }
107}
108
109/// Bytes reserved from an [`IoBudget`], returned on drop.
110#[derive(Debug)]
111pub struct IoReservation {
112    budget: &'static IoBudget,
113    bytes: usize,
114}
115
116impl std::fmt::Debug for IoBudget {
117    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
118        formatter.debug_struct("IoBudget").finish_non_exhaustive()
119    }
120}
121
122impl Drop for IoReservation {
123    fn drop(&mut self) {
124        self.budget.state().reserved -= self.bytes;
125    }
126}
lace_sans_io/runtime_support/io_budget.rs

lace_sans_io/runtime_support/
io_budget.rs
