Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
50 changes: 50 additions & 0 deletions src/production/perf_config.rs
Original file line number Diff line number Diff line change
Expand Up @@ -218,6 +218,12 @@ impl PerformanceConfig {
if self.buffers.max_size < self.buffers.read_size {
return Err("buffers.max_size must be >= read_size".to_string());
}
if self.connection_pool.max_connections == 0 {
return Err("connection_pool.max_connections must be > 0".to_string());
}
if self.connection_pool.buffer_pool_size == 0 {
return Err("connection_pool.buffer_pool_size must be > 0".to_string());
}
Ok(())
}
}
Expand All @@ -234,6 +240,8 @@ mod tests {
assert_eq!(config.response_pool.prewarm, 64);
assert_eq!(config.buffers.read_size, 8192);
assert_eq!(config.batching.min_pipeline_buffer, 60);
assert_eq!(config.connection_pool.max_connections, 10000);
assert_eq!(config.connection_pool.buffer_pool_size, 64);
}

#[test]
Expand Down Expand Up @@ -293,5 +301,47 @@ mod tests {
assert_eq!(config.num_shards, 8);
assert_eq!(config.response_pool.capacity, 256); // default
assert_eq!(config.buffers.read_size, 8192); // default
assert_eq!(config.connection_pool.max_connections, 10000); // default
assert_eq!(config.connection_pool.buffer_pool_size, 64); // default
}

#[test]
fn test_connection_pool_toml() {
let toml_str = r#"
num_shards = 16

[connection_pool]
max_connections = 5000
buffer_pool_size = 256
"#;

let config: PerformanceConfig = toml::from_str(toml_str).unwrap();
assert_eq!(config.connection_pool.max_connections, 5000);
assert_eq!(config.connection_pool.buffer_pool_size, 256);
}

#[test]
fn test_connection_pool_partial_toml() {
let toml_str = r#"
num_shards = 16

[connection_pool]
max_connections = 2000
"#;

let config: PerformanceConfig = toml::from_str(toml_str).unwrap();
assert_eq!(config.connection_pool.max_connections, 2000);
assert_eq!(config.connection_pool.buffer_pool_size, 64); // default
}

#[test]
fn test_validate_invalid_connection_pool() {
let mut config = PerformanceConfig::default();
config.connection_pool.max_connections = 0;
assert!(config.validate().is_err());

let mut config = PerformanceConfig::default();
config.connection_pool.buffer_pool_size = 0;
assert!(config.validate().is_err());
}
}
64 changes: 58 additions & 6 deletions src/redis/executor/mod.rs
Original file line number Diff line number Diff line change
Expand Up @@ -45,8 +45,6 @@ pub struct CommandExecutor {
pub(crate) data: AHashMap<String, Value>,
pub(crate) expirations: AHashMap<String, VirtualTime>,
pub(crate) current_time: VirtualTime,
#[allow(dead_code)]
pub(crate) key_count: usize,
pub(crate) commands_processed: usize,
pub(crate) simulation_start_epoch: i64,
/// Exact server start time in milliseconds (for precise PEXPIREAT/PXAT)
Expand All @@ -69,7 +67,6 @@ impl CommandExecutor {
data: AHashMap::new(),
expirations: AHashMap::new(),
current_time: VirtualTime::from_millis(0),
key_count: 0,
commands_processed: 0,
simulation_start_epoch: 0,
simulation_start_epoch_ms: 0,
Expand All @@ -88,7 +85,6 @@ impl CommandExecutor {
data: AHashMap::new(),
expirations: AHashMap::new(),
current_time: VirtualTime::from_millis(0),
key_count: 0,
commands_processed: 0,
simulation_start_epoch: 0,
simulation_start_epoch_ms: 0,
Expand Down Expand Up @@ -192,7 +188,7 @@ impl CommandExecutor {
self.current_time = current_time;

// Single-pass: retain unexpired keys, collect expired ones for data removal
let mut expired_keys = Vec::new();
let mut expired_keys = Vec::with_capacity(self.expirations.len() / 4);
self.expirations.retain(|k, &mut exp_time| {
if exp_time <= self.current_time {
expired_keys.push(k.clone());
Expand All @@ -219,13 +215,14 @@ impl CommandExecutor {
pre_exp_len.saturating_sub(count),
"Postcondition: expirations size must decrease by evicted count"
);
self.verify_invariants();
}

count
}

pub(crate) fn evict_expired_keys(&mut self) {
let mut expired_keys = Vec::new();
let mut expired_keys = Vec::with_capacity(self.expirations.len() / 4);
self.expirations.retain(|k, &mut exp_time| {
if exp_time <= self.current_time {
expired_keys.push(k.clone());
Expand All @@ -238,6 +235,61 @@ impl CommandExecutor {
for key in &expired_keys {
self.data.remove(key);
}

#[cfg(debug_assertions)]
self.verify_invariants();
}

/// Verify all CommandExecutor invariants hold.
/// Call in debug builds after mutations to catch consistency bugs early.
#[cfg(debug_assertions)]
pub(crate) fn verify_invariants(&self) {
// Invariant 1: Every key in expirations must exist in data
for key in self.expirations.keys() {
debug_assert!(
self.data.contains_key(key),
"Invariant violated: expiration key '{}' has no corresponding data entry",
key
);
}

// Invariant 2: No expired keys should remain in expirations after eviction
for (key, &exp_time) in &self.expirations {
debug_assert!(
exp_time > self.current_time,
"Invariant violated: expired key '{}' (exp={}, now={}) still in expirations",
key,
exp_time.as_millis(),
self.current_time.as_millis()
);
}

// Invariant 3: No empty collections in data
for (key, value) in &self.data {
match value {
Value::List(l) => debug_assert!(
!l.is_empty(),
"Invariant violated: empty list for key '{}'",
key
),
Value::Set(s) => debug_assert!(
!s.is_empty(),
"Invariant violated: empty set for key '{}'",
key
),
Value::Hash(h) => debug_assert!(
!h.is_empty(),
"Invariant violated: empty hash for key '{}'",
key
),
Value::SortedSet(z) => debug_assert!(
z.len() > 0,
"Invariant violated: empty sorted set for key '{}'",
key
),
_ => {}
}
}
}

pub(crate) fn get_value(&mut self, key: &str) -> Option<&Value> {
Expand Down
Loading