use std::{
env,
path::PathBuf,
sync::{
atomic::{AtomicUsize, Ordering},
Arc,
},
};
pub const DEFAULT_MEMORY_BUDGET: usize = 1024 * 1024 * 1024;
pub const DEFAULT_SPILL_THRESHOLD: f64 = 0.8;
pub const DEFAULT_TARGET_PARTITION_BYTES: usize = 64 * 1024 * 1024;
pub const MIN_OPERATOR_MEMORY: usize = 4 * 1024 * 1024;
#[derive(Debug, Clone)]
pub struct MemoryConfig {
pub budget_bytes: usize,
pub temp_directory: PathBuf,
pub spill_threshold: f64,
pub target_partition_bytes: usize,
}
impl Default for MemoryConfig {
fn default() -> Self {
let budget_bytes = env::var("VIBESQL_MEMORY_LIMIT")
.ok()
.and_then(|s| parse_memory_size(&s))
.unwrap_or(DEFAULT_MEMORY_BUDGET);
let temp_directory = env::var("VIBESQL_TEMP_DIR")
.ok()
.map(PathBuf::from)
.unwrap_or_else(|| env::temp_dir().join("vibesql"));
let spill_threshold = env::var("VIBESQL_SPILL_THRESHOLD")
.ok()
.and_then(|s| s.parse::<f64>().ok())
.filter(|&t| (0.0..=1.0).contains(&t))
.unwrap_or(DEFAULT_SPILL_THRESHOLD);
let target_partition_bytes = env::var("VIBESQL_PARTITION_SIZE")
.ok()
.and_then(|s| parse_memory_size(&s))
.unwrap_or(DEFAULT_TARGET_PARTITION_BYTES);
Self { budget_bytes, temp_directory, spill_threshold, target_partition_bytes }
}
}
impl MemoryConfig {
pub fn with_budget(budget_bytes: usize) -> Self {
Self { budget_bytes, ..Default::default() }
}
pub fn with_temp_dir(mut self, path: PathBuf) -> Self {
self.temp_directory = path;
self
}
pub fn with_spill_threshold(mut self, threshold: f64) -> Self {
self.spill_threshold = threshold.clamp(0.0, 1.0);
self
}
}
#[derive(Debug, Clone)]
pub struct MemoryStats {
pub budget_bytes: usize,
pub reserved_bytes: usize,
pub peak_bytes: usize,
pub bytes_spilled: usize,
pub spill_count: usize,
pub active_reservations: usize,
pub spill_threshold: f64,
}
impl MemoryStats {
pub fn utilization(&self) -> f64 {
if self.budget_bytes == 0 {
0.0
} else {
self.reserved_bytes as f64 / self.budget_bytes as f64
}
}
pub fn is_under_pressure(&self) -> bool {
self.utilization() >= self.spill_threshold
}
pub fn available_bytes(&self) -> usize {
self.budget_bytes.saturating_sub(self.reserved_bytes)
}
}
impl std::fmt::Display for MemoryStats {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
write!(
f,
"Memory: {}/{} ({:.1}%), peak: {}, spilled: {} ({} ops)",
format_bytes(self.reserved_bytes),
format_bytes(self.budget_bytes),
self.utilization() * 100.0,
format_bytes(self.peak_bytes),
format_bytes(self.bytes_spilled),
self.spill_count,
)
}
}
fn format_bytes(bytes: usize) -> String {
if bytes >= 1024 * 1024 * 1024 {
format!("{:.2}GB", bytes as f64 / (1024.0 * 1024.0 * 1024.0))
} else if bytes >= 1024 * 1024 {
format!("{:.2}MB", bytes as f64 / (1024.0 * 1024.0))
} else if bytes >= 1024 {
format!("{:.2}KB", bytes as f64 / 1024.0)
} else {
format!("{}B", bytes)
}
}
fn parse_memory_size(s: &str) -> Option<usize> {
let s = s.trim().to_uppercase();
if let Ok(n) = s.parse::<usize>() {
return Some(n);
}
let (num_str, multiplier) = if let Some(num) = s.strip_suffix("GB") {
(num, 1024 * 1024 * 1024)
} else if let Some(num) = s.strip_suffix("G") {
(num, 1024 * 1024 * 1024)
} else if let Some(num) = s.strip_suffix("MB") {
(num, 1024 * 1024)
} else if let Some(num) = s.strip_suffix("M") {
(num, 1024 * 1024)
} else if let Some(num) = s.strip_suffix("KB") {
(num, 1024)
} else if let Some(num) = s.strip_suffix("K") {
(num, 1024)
} else {
return None;
};
num_str.trim().parse::<usize>().ok().map(|n| n * multiplier)
}
pub struct MemoryController {
config: MemoryConfig,
reserved: AtomicUsize,
active_reservations: AtomicUsize,
bytes_spilled: AtomicUsize,
spill_count: AtomicUsize,
peak_memory: AtomicUsize,
}
impl MemoryController {
pub fn new(config: MemoryConfig) -> Self {
Self {
config,
reserved: AtomicUsize::new(0),
active_reservations: AtomicUsize::new(0),
bytes_spilled: AtomicUsize::new(0),
spill_count: AtomicUsize::new(0),
peak_memory: AtomicUsize::new(0),
}
}
pub fn with_defaults() -> Self {
Self::new(MemoryConfig::default())
}
pub fn with_budget(budget_bytes: usize) -> Self {
Self::new(MemoryConfig::with_budget(budget_bytes))
}
pub fn budget(&self) -> usize {
self.config.budget_bytes
}
pub fn reserved(&self) -> usize {
self.reserved.load(Ordering::Relaxed)
}
pub fn available(&self) -> usize {
let budget = self.config.budget_bytes;
let reserved = self.reserved.load(Ordering::Relaxed);
budget.saturating_sub(reserved)
}
pub fn spill_threshold_bytes(&self) -> usize {
(self.config.budget_bytes as f64 * self.config.spill_threshold) as usize
}
pub fn should_spill(&self) -> bool {
self.reserved() >= self.spill_threshold_bytes()
}
pub fn temp_directory(&self) -> &PathBuf {
&self.config.temp_directory
}
pub fn target_partition_bytes(&self) -> usize {
self.config.target_partition_bytes
}
pub fn create_reservation(self: &Arc<Self>) -> MemoryReservation {
self.active_reservations.fetch_add(1, Ordering::Relaxed);
MemoryReservation { controller: Arc::clone(self), reserved: 0 }
}
pub fn record_spill(&self, bytes: usize) {
self.bytes_spilled.fetch_add(bytes, Ordering::Relaxed);
self.spill_count.fetch_add(1, Ordering::Relaxed);
}
pub fn bytes_spilled(&self) -> usize {
self.bytes_spilled.load(Ordering::Relaxed)
}
pub fn spill_count(&self) -> usize {
self.spill_count.load(Ordering::Relaxed)
}
pub fn peak_memory(&self) -> usize {
self.peak_memory.load(Ordering::Relaxed)
}
pub fn active_reservations(&self) -> usize {
self.active_reservations.load(Ordering::Relaxed)
}
pub fn stats(&self) -> MemoryStats {
MemoryStats {
budget_bytes: self.config.budget_bytes,
reserved_bytes: self.reserved.load(Ordering::Relaxed),
peak_bytes: self.peak_memory.load(Ordering::Relaxed),
bytes_spilled: self.bytes_spilled.load(Ordering::Relaxed),
spill_count: self.spill_count.load(Ordering::Relaxed),
active_reservations: self.active_reservations.load(Ordering::Relaxed),
spill_threshold: self.config.spill_threshold,
}
}
fn try_reserve(&self, bytes: usize) -> bool {
let budget = self.config.budget_bytes;
loop {
let current = self.reserved.load(Ordering::Relaxed);
let new_reserved = current.saturating_add(bytes);
if new_reserved > budget {
return false;
}
match self.reserved.compare_exchange_weak(
current,
new_reserved,
Ordering::SeqCst,
Ordering::Relaxed,
) {
Ok(_) => {
self.update_peak_memory(new_reserved);
return true;
}
Err(_) => continue, }
}
}
fn update_peak_memory(&self, current: usize) {
let mut peak = self.peak_memory.load(Ordering::Relaxed);
while current > peak {
match self.peak_memory.compare_exchange_weak(
peak,
current,
Ordering::Relaxed,
Ordering::Relaxed,
) {
Ok(_) => break,
Err(actual) => peak = actual,
}
}
}
fn release(&self, bytes: usize) {
self.reserved.fetch_sub(bytes, Ordering::Relaxed);
}
fn release_reservation(&self) {
self.active_reservations.fetch_sub(1, Ordering::Relaxed);
}
}
pub struct MemoryReservation {
controller: Arc<MemoryController>,
reserved: usize,
}
impl MemoryReservation {
pub fn reserved(&self) -> usize {
self.reserved
}
pub fn try_grow(&mut self, additional: usize) -> bool {
if self.controller.try_reserve(additional) {
self.reserved = self.reserved.saturating_add(additional);
true
} else {
false
}
}
pub fn shrink(&mut self, bytes: usize) {
let to_release = bytes.min(self.reserved);
self.controller.release(to_release);
self.reserved = self.reserved.saturating_sub(to_release);
}
pub fn release_all(&mut self) {
if self.reserved > 0 {
self.controller.release(self.reserved);
self.reserved = 0;
}
}
pub fn should_spill(&self) -> bool {
self.controller.should_spill()
}
pub fn would_exceed_budget(&self, additional: usize) -> bool {
let current_total = self.controller.reserved();
current_total.saturating_add(additional) > self.controller.budget()
}
pub fn budget(&self) -> usize {
self.controller.budget()
}
pub fn temp_directory(&self) -> &PathBuf {
self.controller.temp_directory()
}
pub fn target_partition_bytes(&self) -> usize {
self.controller.target_partition_bytes()
}
pub fn record_spill(&self, bytes: usize) {
self.controller.record_spill(bytes);
}
}
impl Drop for MemoryReservation {
fn drop(&mut self) {
if self.reserved > 0 {
self.controller.release(self.reserved);
}
self.controller.release_reservation();
}
}
unsafe impl Send for MemoryController {}
unsafe impl Sync for MemoryController {}
#[cfg(test)]
mod tests {
use std::sync::Arc;
use super::*;
#[test]
fn test_memory_size_parsing() {
assert_eq!(parse_memory_size("1024"), Some(1024));
assert_eq!(parse_memory_size("1KB"), Some(1024));
assert_eq!(parse_memory_size("1K"), Some(1024));
assert_eq!(parse_memory_size("1MB"), Some(1024 * 1024));
assert_eq!(parse_memory_size("1M"), Some(1024 * 1024));
assert_eq!(parse_memory_size("1GB"), Some(1024 * 1024 * 1024));
assert_eq!(parse_memory_size("1G"), Some(1024 * 1024 * 1024));
assert_eq!(parse_memory_size("4GB"), Some(4 * 1024 * 1024 * 1024));
assert_eq!(parse_memory_size("512mb"), Some(512 * 1024 * 1024));
assert_eq!(parse_memory_size(" 100 "), Some(100));
assert_eq!(parse_memory_size("invalid"), None);
}
#[test]
fn test_controller_creation() {
let controller = Arc::new(MemoryController::with_budget(1024 * 1024));
assert_eq!(controller.budget(), 1024 * 1024);
assert_eq!(controller.reserved(), 0);
assert_eq!(controller.available(), 1024 * 1024);
}
#[test]
fn test_reservation_try_grow() {
let controller = Arc::new(MemoryController::with_budget(1024));
let mut reservation = controller.create_reservation();
assert!(reservation.try_grow(512));
assert_eq!(reservation.reserved(), 512);
assert_eq!(controller.reserved(), 512);
assert!(reservation.try_grow(256));
assert_eq!(reservation.reserved(), 768);
assert_eq!(controller.reserved(), 768);
assert!(!reservation.try_grow(512));
assert_eq!(reservation.reserved(), 768);
assert_eq!(controller.reserved(), 768);
assert!(reservation.try_grow(256));
assert_eq!(reservation.reserved(), 1024);
}
#[test]
fn test_reservation_shrink() {
let controller = Arc::new(MemoryController::with_budget(1024));
let mut reservation = controller.create_reservation();
reservation.try_grow(1024);
assert_eq!(reservation.reserved(), 1024);
reservation.shrink(512);
assert_eq!(reservation.reserved(), 512);
assert_eq!(controller.reserved(), 512);
reservation.shrink(1024);
assert_eq!(reservation.reserved(), 0);
assert_eq!(controller.reserved(), 0);
}
#[test]
fn test_reservation_drop_releases_memory() {
let controller = Arc::new(MemoryController::with_budget(1024));
{
let mut reservation = controller.create_reservation();
reservation.try_grow(512);
assert_eq!(controller.reserved(), 512);
}
assert_eq!(controller.reserved(), 0);
}
#[test]
fn test_multiple_reservations() {
let controller = Arc::new(MemoryController::with_budget(1024));
let mut res1 = controller.create_reservation();
let mut res2 = controller.create_reservation();
assert!(res1.try_grow(300));
assert!(res2.try_grow(300));
assert_eq!(controller.reserved(), 600);
assert!(res1.try_grow(200));
assert!(res2.try_grow(200));
assert_eq!(controller.reserved(), 1000);
assert!(!res1.try_grow(100));
assert!(!res2.try_grow(100));
drop(res1);
assert_eq!(controller.reserved(), 500);
assert!(res2.try_grow(400));
assert_eq!(controller.reserved(), 900);
}
#[test]
fn test_should_spill() {
let config = MemoryConfig {
budget_bytes: 1000,
spill_threshold: 0.8, temp_directory: std::env::temp_dir(),
target_partition_bytes: DEFAULT_TARGET_PARTITION_BYTES,
};
let controller = Arc::new(MemoryController::new(config));
let mut reservation = controller.create_reservation();
reservation.try_grow(700);
assert!(!reservation.should_spill());
reservation.try_grow(100);
assert!(reservation.should_spill());
}
#[test]
fn test_spill_tracking() {
let controller = Arc::new(MemoryController::with_budget(1024));
assert_eq!(controller.bytes_spilled(), 0);
controller.record_spill(100);
assert_eq!(controller.bytes_spilled(), 100);
controller.record_spill(200);
assert_eq!(controller.bytes_spilled(), 300);
}
#[test]
fn test_active_reservation_count() {
let controller = Arc::new(MemoryController::with_budget(1024));
assert_eq!(controller.active_reservations(), 0);
let _res1 = controller.create_reservation();
assert_eq!(controller.active_reservations(), 1);
let _res2 = controller.create_reservation();
assert_eq!(controller.active_reservations(), 2);
drop(_res1);
assert_eq!(controller.active_reservations(), 1);
drop(_res2);
assert_eq!(controller.active_reservations(), 0);
}
#[test]
fn test_concurrent_reservations() {
use std::thread;
let controller = Arc::new(MemoryController::with_budget(10_000));
let mut handles = vec![];
for _ in 0..10 {
let controller = Arc::clone(&controller);
handles.push(thread::spawn(move || {
let mut reservation = controller.create_reservation();
reservation.try_grow(500);
thread::sleep(std::time::Duration::from_millis(10));
reservation.reserved()
}));
}
let reserved: usize = handles.into_iter().map(|h| h.join().unwrap()).sum();
assert_eq!(reserved, 5000);
assert_eq!(controller.reserved(), 0);
}
#[test]
fn test_memory_stats() {
let config = MemoryConfig {
budget_bytes: 1000,
spill_threshold: 0.8,
temp_directory: std::env::temp_dir(),
target_partition_bytes: DEFAULT_TARGET_PARTITION_BYTES,
};
let controller = Arc::new(MemoryController::new(config));
let stats = controller.stats();
assert_eq!(stats.budget_bytes, 1000);
assert_eq!(stats.reserved_bytes, 0);
assert_eq!(stats.peak_bytes, 0);
assert_eq!(stats.bytes_spilled, 0);
assert_eq!(stats.spill_count, 0);
assert_eq!(stats.active_reservations, 0);
assert_eq!(stats.spill_threshold, 0.8);
let mut res = controller.create_reservation();
res.try_grow(500);
let stats = controller.stats();
assert_eq!(stats.reserved_bytes, 500);
assert_eq!(stats.peak_bytes, 500);
assert_eq!(stats.active_reservations, 1);
assert!((stats.utilization() - 0.5).abs() < 0.001);
assert_eq!(stats.available_bytes(), 500);
assert!(!stats.is_under_pressure());
res.try_grow(400);
let stats = controller.stats();
assert!(stats.is_under_pressure()); }
#[test]
fn test_peak_memory_tracking() {
let controller = Arc::new(MemoryController::with_budget(1000));
{
let mut res = controller.create_reservation();
res.try_grow(800);
assert_eq!(controller.peak_memory(), 800);
}
assert_eq!(controller.reserved(), 0);
assert_eq!(controller.peak_memory(), 800);
{
let mut res = controller.create_reservation();
res.try_grow(900);
assert_eq!(controller.peak_memory(), 900);
}
{
let mut res = controller.create_reservation();
res.try_grow(100);
assert_eq!(controller.peak_memory(), 900);
}
}
#[test]
fn test_memory_stats_display() {
let stats = MemoryStats {
budget_bytes: 1024 * 1024 * 1024,
reserved_bytes: 512 * 1024 * 1024,
peak_bytes: 950 * 1024 * 1024,
bytes_spilled: 2 * 1024 * 1024 * 1024,
spill_count: 3,
active_reservations: 2,
spill_threshold: 0.8,
};
let display = format!("{}", stats);
assert!(display.contains("512.00MB"));
assert!(display.contains("1.00GB"));
assert!(display.contains("50.0%"));
assert!(display.contains("950.00MB"));
assert!(display.contains("2.00GB"));
assert!(display.contains("3 ops"));
}
#[test]
fn test_spill_count_tracking() {
let controller = Arc::new(MemoryController::with_budget(1024));
assert_eq!(controller.spill_count(), 0);
assert_eq!(controller.bytes_spilled(), 0);
controller.record_spill(100);
assert_eq!(controller.spill_count(), 1);
assert_eq!(controller.bytes_spilled(), 100);
controller.record_spill(200);
assert_eq!(controller.spill_count(), 2);
assert_eq!(controller.bytes_spilled(), 300);
controller.record_spill(50);
assert_eq!(controller.spill_count(), 3);
assert_eq!(controller.bytes_spilled(), 350);
}
#[test]
fn test_format_bytes_helper() {
let make_stats = |bytes| MemoryStats {
budget_bytes: bytes,
reserved_bytes: bytes,
peak_bytes: bytes,
bytes_spilled: 0,
spill_count: 0,
active_reservations: 0,
spill_threshold: 0.8,
};
let s = format!("{}", make_stats(500));
assert!(s.contains("500B"));
let s = format!("{}", make_stats(2048));
assert!(s.contains("2.00KB"));
let s = format!("{}", make_stats(5 * 1024 * 1024));
assert!(s.contains("5.00MB"));
let s = format!("{}", make_stats(3 * 1024 * 1024 * 1024));
assert!(s.contains("3.00GB"));
}
}