use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::sync::atomic::{AtomicU64, Ordering};
use arc_swap::ArcSwap;
use parking_lot::Mutex;
use crate::adaptive_ring::{AdaptiveError, AdaptiveRing, RingShape};
use crate::ordering::{default_stamp_kind, OrderingMode, StampKind};
use crate::shared_ring::RingError;
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum BackingTarget {
Anon,
File(PathBuf),
Shm(String),
}
#[derive(Debug, Clone, Default)]
pub struct RingConfig {
pub shape: Option<RingShape>,
pub capacity: Option<usize>,
pub locale: Option<BackingTarget>,
}
#[derive(Debug)]
pub enum CapacityMorphError {
InvalidCapacity,
CannotShrinkInFlight { in_flight: usize, new_capacity: usize },
Ring(RingError),
Adaptive(AdaptiveError),
}
impl From<RingError> for CapacityMorphError {
fn from(e: RingError) -> Self { Self::Ring(e) }
}
impl From<AdaptiveError> for CapacityMorphError {
fn from(e: AdaptiveError) -> Self { Self::Adaptive(e) }
}
impl std::fmt::Display for CapacityMorphError {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
Self::InvalidCapacity => write!(f, "capacity must be pow2 >= 2"),
Self::CannotShrinkInFlight { in_flight, new_capacity } => write!(
f,
"shrink rejected: {in_flight} in-flight items exceed new capacity {new_capacity}",
),
Self::Ring(e) => write!(f, "ring error during morph: {e:?}"),
Self::Adaptive(e) => write!(f, "adaptive ring error during morph: {e:?}"),
}
}
}
impl std::error::Error for CapacityMorphError {}
pub struct CapacityAdaptiveRing {
state: ArcSwap<RingState>,
pin_generation: AtomicU64,
capacity_atom: AtomicU64,
max_producers: usize,
max_consumers: usize,
backing_source: Mutex<BackingTarget>,
morph_seq: AtomicU64,
stamped: Option<StampKind>,
morph_lock: Mutex<()>,
warm: Mutex<Option<(usize, BackingTarget, Arc<AdaptiveRing>)>>,
warm_hits: AtomicU64,
stale_pops: AtomicU64,
}
struct RingState {
active: Arc<AdaptiveRing>,
stale: Vec<Arc<AdaptiveRing>>,
}
unsafe impl Send for CapacityAdaptiveRing {}
unsafe impl Sync for CapacityAdaptiveRing {}
impl CapacityAdaptiveRing {
pub fn create_anon(
max_producers: usize,
max_consumers: usize,
initial_capacity: usize,
) -> Result<Self, CapacityMorphError> {
if !initial_capacity.is_power_of_two() || initial_capacity < 2 {
return Err(CapacityMorphError::InvalidCapacity);
}
let ring = AdaptiveRing::create_anon(
max_producers,
max_consumers,
initial_capacity,
)?;
Ok(Self {
state: ArcSwap::from(Arc::new(RingState {
active: Arc::new(ring),
stale: Vec::new(),
})),
pin_generation: AtomicU64::new(0),
capacity_atom: AtomicU64::new(initial_capacity as u64),
max_producers,
max_consumers,
backing_source: Mutex::new(BackingTarget::Anon),
morph_seq: AtomicU64::new(0),
stamped: None,
morph_lock: Mutex::new(()),
warm: Mutex::new(None),
warm_hits: AtomicU64::new(0),
stale_pops: AtomicU64::new(0),
})
}
pub fn create_anon_stamped(
max_producers: usize,
max_consumers: usize,
initial_capacity: usize,
) -> Result<Self, CapacityMorphError> {
if !initial_capacity.is_power_of_two() || initial_capacity < 2 {
return Err(CapacityMorphError::InvalidCapacity);
}
let kind = default_stamp_kind();
let ring = AdaptiveRing::create_anon(
max_producers, max_consumers, initial_capacity,
)?
.with_ordering_stamps_kind(kind)
.map_err(CapacityMorphError::Ring)?;
Ok(Self {
state: ArcSwap::from(Arc::new(RingState {
active: Arc::new(ring),
stale: Vec::new(),
})),
pin_generation: AtomicU64::new(0),
capacity_atom: AtomicU64::new(initial_capacity as u64),
max_producers,
max_consumers,
backing_source: Mutex::new(BackingTarget::Anon),
morph_seq: AtomicU64::new(0),
stamped: Some(kind),
morph_lock: Mutex::new(()),
warm: Mutex::new(None),
warm_hits: AtomicU64::new(0),
stale_pops: AtomicU64::new(0),
})
}
pub fn create(
base_path: impl AsRef<Path>,
max_producers: usize,
max_consumers: usize,
initial_capacity: usize,
) -> Result<Self, CapacityMorphError> {
if !initial_capacity.is_power_of_two() || initial_capacity < 2 {
return Err(CapacityMorphError::InvalidCapacity);
}
let base = base_path.as_ref().to_path_buf();
let path = path_for_capacity(&base, initial_capacity);
let ring = AdaptiveRing::create(
&path,
max_producers,
max_consumers,
initial_capacity,
)?;
Ok(Self {
state: ArcSwap::from(Arc::new(RingState {
active: Arc::new(ring),
stale: Vec::new(),
})),
pin_generation: AtomicU64::new(0),
capacity_atom: AtomicU64::new(initial_capacity as u64),
max_producers,
max_consumers,
backing_source: Mutex::new(BackingTarget::File(base)),
morph_seq: AtomicU64::new(0),
stamped: None,
morph_lock: Mutex::new(()),
warm: Mutex::new(None),
warm_hits: AtomicU64::new(0),
stale_pops: AtomicU64::new(0),
})
}
pub fn create_stamped(
base_path: impl AsRef<Path>,
max_producers: usize,
max_consumers: usize,
initial_capacity: usize,
) -> Result<Self, CapacityMorphError> {
if !initial_capacity.is_power_of_two() || initial_capacity < 2 {
return Err(CapacityMorphError::InvalidCapacity);
}
let kind = default_stamp_kind();
let base = base_path.as_ref().to_path_buf();
let path = path_for_capacity(&base, initial_capacity);
let ring = AdaptiveRing::create(
&path, max_producers, max_consumers, initial_capacity,
)?
.with_ordering_stamps_kind(kind)
.map_err(CapacityMorphError::Ring)?;
Ok(Self {
state: ArcSwap::from(Arc::new(RingState {
active: Arc::new(ring),
stale: Vec::new(),
})),
pin_generation: AtomicU64::new(0),
capacity_atom: AtomicU64::new(initial_capacity as u64),
max_producers,
max_consumers,
backing_source: Mutex::new(BackingTarget::File(base)),
morph_seq: AtomicU64::new(0),
stamped: Some(kind),
morph_lock: Mutex::new(()),
warm: Mutex::new(None),
warm_hits: AtomicU64::new(0),
stale_pops: AtomicU64::new(0),
})
}
pub fn create_shmfs(
name_prefix: &str,
max_producers: usize,
max_consumers: usize,
initial_capacity: usize,
) -> Result<Self, CapacityMorphError> {
if !initial_capacity.is_power_of_two() || initial_capacity < 2 {
return Err(CapacityMorphError::InvalidCapacity);
}
let name = format!("{name_prefix}_cap_{initial_capacity}");
let ring = AdaptiveRing::create_shmfs(
&name,
max_producers,
max_consumers,
initial_capacity,
)?;
Ok(Self {
state: ArcSwap::from(Arc::new(RingState {
active: Arc::new(ring),
stale: Vec::new(),
})),
pin_generation: AtomicU64::new(0),
capacity_atom: AtomicU64::new(initial_capacity as u64),
max_producers,
max_consumers,
backing_source: Mutex::new(BackingTarget::Shm(name_prefix.to_owned())),
morph_seq: AtomicU64::new(0),
stamped: None,
morph_lock: Mutex::new(()),
warm: Mutex::new(None),
warm_hits: AtomicU64::new(0),
stale_pops: AtomicU64::new(0),
})
}
pub fn create_shmfs_stamped(
name_prefix: &str,
max_producers: usize,
max_consumers: usize,
initial_capacity: usize,
) -> Result<Self, CapacityMorphError> {
if !initial_capacity.is_power_of_two() || initial_capacity < 2 {
return Err(CapacityMorphError::InvalidCapacity);
}
let kind = default_stamp_kind();
let name = format!("{name_prefix}_cap_{initial_capacity}");
let ring = AdaptiveRing::create_shmfs(
&name, max_producers, max_consumers, initial_capacity,
)?
.with_ordering_stamps_kind(kind)
.map_err(CapacityMorphError::Ring)?;
Ok(Self {
state: ArcSwap::from(Arc::new(RingState {
active: Arc::new(ring),
stale: Vec::new(),
})),
pin_generation: AtomicU64::new(0),
capacity_atom: AtomicU64::new(initial_capacity as u64),
max_producers,
max_consumers,
backing_source: Mutex::new(BackingTarget::Shm(name_prefix.to_owned())),
morph_seq: AtomicU64::new(0),
stamped: Some(kind),
morph_lock: Mutex::new(()),
warm: Mutex::new(None),
warm_hits: AtomicU64::new(0),
stale_pops: AtomicU64::new(0),
})
}
pub fn current_capacity(&self) -> usize {
self.capacity_atom.load(Ordering::Acquire) as usize
}
pub fn pin_generation(&self) -> u64 {
self.pin_generation.load(Ordering::Acquire)
}
pub fn register_producer(&self) -> Result<usize, AdaptiveError> {
self.state.load().active.register_producer()
}
pub fn register_consumer(&self) -> Result<usize, AdaptiveError> {
self.state.load().active.register_consumer()
}
#[inline]
pub fn try_send(
&self,
producer_id: usize,
payload: &[u8],
) -> Result<(), RingError> {
self.state.load().active.try_send(producer_id, payload)
}
#[inline]
pub fn try_recv(
&self,
consumer_id: usize,
out: &mut [u8],
) -> Result<usize, RingError> {
let state = self.state.load();
for ring in &state.stale {
loop {
match ring.try_recv(consumer_id, out) {
Ok(n) => {
self.stale_pops.fetch_add(1, Ordering::Relaxed);
return Ok(n);
}
Err(_) => {
if ring.is_empty() {
break;
}
std::hint::spin_loop();
}
}
}
}
state.active.try_recv(consumer_id, out)
}
pub fn morph_capacity_to(
&self,
new_capacity: usize,
) -> Result<(), CapacityMorphError> {
self.morph_to_config(&RingConfig {
capacity: Some(new_capacity),
..RingConfig::default()
})
}
pub fn morph_to_config(
&self,
target: &RingConfig,
) -> Result<(), CapacityMorphError> {
let _morph_guard = self.morph_lock.lock();
let old_state = self.state.load_full();
let old = Arc::clone(&old_state.active);
let old_capacity = self.capacity_atom.load(Ordering::Acquire) as usize;
let old_shape = old.current_shape();
let old_locale = self.backing_source.lock().clone();
let new_capacity = target.capacity.unwrap_or(old_capacity);
if !new_capacity.is_power_of_two() || new_capacity < 2 {
return Err(CapacityMorphError::InvalidCapacity);
}
let new_shape = target.shape.unwrap_or(old_shape);
let new_locale =
target.locale.clone().unwrap_or_else(|| old_locale.clone());
if new_capacity == old_capacity
&& new_shape == old_shape
&& new_locale == old_locale
{
return Ok(());
}
if new_capacity == old_capacity && new_locale == old_locale {
return old.morph_to(new_shape).map_err(CapacityMorphError::Ring);
}
if new_locale != old_locale {
*self.backing_source.lock() = new_locale.clone();
}
let warm_hit = {
let mut warm = self.warm.lock();
warm.take_if(|(cap, loc, _)| {
*cap == new_capacity && *loc == new_locale
})
};
let new = match warm_hit {
Some((_, _, ring)) => {
self.warm_hits.fetch_add(1, Ordering::Relaxed);
ring
}
None => self.build_backing(new_capacity, &new_locale)?,
};
if self.stamped.is_some()
&& let (Some(new_region), Some(old_region)) =
(new.ordering_region(), old.ordering_region())
{
new_region.seed_from(old_region);
}
let n_producers = old.active_producers();
let n_consumers = old.active_consumers();
for _ in 0..n_producers {
new.register_producer()?;
}
for _ in 0..n_consumers {
new.register_consumer()?;
}
if new.current_shape() != new_shape {
new.morph_to(new_shape).map_err(CapacityMorphError::Ring)?;
}
self.pin_generation.fetch_add(1, Ordering::AcqRel);
let mut new_stale: Vec<Arc<AdaptiveRing>> =
old_state.stale.iter().filter(|r| !r.is_empty()).cloned().collect();
new_stale.push(old);
let new_state = RingState { active: new, stale: new_stale };
self.state.store(Arc::new(new_state));
self.capacity_atom
.store(new_capacity as u64, Ordering::Release);
Ok(())
}
fn build_backing(
&self,
capacity: usize,
locale: &BackingTarget,
) -> Result<Arc<AdaptiveRing>, CapacityMorphError> {
let seq = self.morph_seq.fetch_add(1, Ordering::AcqRel);
let mut ring = match locale {
BackingTarget::Anon => AdaptiveRing::create_anon(
self.max_producers,
self.max_consumers,
capacity,
)?,
BackingTarget::File(base) => AdaptiveRing::create(
path_for_capacity_seq(base, capacity, seq),
self.max_producers,
self.max_consumers,
capacity,
)?,
BackingTarget::Shm(prefix) => AdaptiveRing::create_shmfs(
&format!("{prefix}_cap_{capacity}_g{seq}"),
self.max_producers,
self.max_consumers,
capacity,
)?,
};
if let Some(kind) = self.stamped {
ring = ring
.with_ordering_stamps_kind(kind)
.map_err(CapacityMorphError::Ring)?;
}
Ok(Arc::new(ring))
}
pub fn prewarm(&self, capacity: usize) -> Result<(), CapacityMorphError> {
self.prewarm_config(&RingConfig {
capacity: Some(capacity),
..RingConfig::default()
})
}
pub fn prewarm_config(
&self,
target: &RingConfig,
) -> Result<(), CapacityMorphError> {
let capacity = target
.capacity
.unwrap_or_else(|| self.current_capacity());
if !capacity.is_power_of_two() || capacity < 2 {
return Err(CapacityMorphError::InvalidCapacity);
}
let locale = target
.locale
.clone()
.unwrap_or_else(|| self.backing_source.lock().clone());
if self
.warm
.lock()
.as_ref()
.is_some_and(|(c, l, _)| *c == capacity && *l == locale)
{
return Ok(());
}
let ring = self.build_backing(capacity, &locale)?;
*self.warm.lock() = Some((capacity, locale, ring));
Ok(())
}
pub fn warm_capacity(&self) -> Option<usize> {
self.warm.lock().as_ref().map(|(c, _, _)| *c)
}
pub fn warm_hits(&self) -> u64 {
self.warm_hits.load(Ordering::Relaxed)
}
pub fn stale_pops(&self) -> u64 {
self.stale_pops.load(Ordering::Relaxed)
}
pub fn clear_warm(&self) {
*self.warm.lock() = None;
}
pub fn pin_current_capacity(&self) -> PinnedCapacity<'_> {
let captured_gen = self.pin_generation.load(Ordering::Acquire);
let ring = Arc::clone(&self.state.load().active);
let capacity = self.capacity_atom.load(Ordering::Acquire) as usize;
PinnedCapacity {
parent: self,
pinned_generation: captured_gen,
ring,
capacity,
_not_sync: std::marker::PhantomData,
}
}
pub fn ring_handle(&self) -> Arc<AdaptiveRing> {
Arc::clone(&self.state.load().active)
}
pub fn is_stamped(&self) -> bool {
self.stamped.is_some()
}
pub fn ordering_mode(&self) -> Option<OrderingMode> {
self.state.load().active.ordering_mode()
}
pub fn set_ordering_mode(&self, mode: OrderingMode) -> Result<(), RingError> {
let state = self.state.load();
for ring in &state.stale {
ring.set_ordering_mode(mode)?;
}
state.active.set_ordering_mode(mode)
}
pub fn inversions(&self) -> u64 {
self.state.load().active.inversions()
}
}
pub struct PinnedCapacity<'a> {
parent: &'a CapacityAdaptiveRing,
pinned_generation: u64,
ring: Arc<AdaptiveRing>,
capacity: usize,
_not_sync: std::marker::PhantomData<std::cell::Cell<()>>,
}
impl<'a> PinnedCapacity<'a> {
pub fn is_still_valid(&self) -> bool {
self.parent.pin_generation.load(Ordering::Acquire) == self.pinned_generation
}
pub fn capacity(&self) -> usize { self.capacity }
pub fn generation(&self) -> u64 { self.pinned_generation }
pub fn ring(&self) -> &Arc<AdaptiveRing> { &self.ring }
}
fn path_for_capacity(base: &Path, capacity: usize) -> PathBuf {
let mut s = base.as_os_str().to_owned();
s.push(format!(".cap_{capacity}.bin"));
PathBuf::from(s)
}
fn path_for_capacity_seq(base: &Path, capacity: usize, seq: u64) -> PathBuf {
let mut s = base.as_os_str().to_owned();
s.push(format!(".cap_{capacity}_g{seq}.bin"));
PathBuf::from(s)
}
#[derive(Debug, Clone, Copy)]
pub struct CapacityPolicyObservation {
pub current_capacity: usize,
pub active_approx_len: usize,
pub total_slot_capacity: usize,
pub since_last_morph: std::time::Duration,
}
impl CapacityPolicyObservation {
pub fn fill_ratio(&self) -> f64 {
if self.total_slot_capacity == 0 {
return 0.0;
}
let ratio = self.active_approx_len as f64 / self.total_slot_capacity as f64;
if ratio > 1.0 { 1.0 } else { ratio }
}
}
pub trait CapacityPolicy: Send + Sync + 'static {
fn decide(&self, observation: &CapacityPolicyObservation) -> Option<usize>;
fn predict(&self, _observation: &CapacityPolicyObservation) -> Option<usize> {
None
}
}
pub struct DefaultCapacityPolicy {
pub grow_at: f64,
pub shrink_at: f64,
pub min_capacity: usize,
pub max_capacity: usize,
pub hysteresis: std::time::Duration,
}
impl Default for DefaultCapacityPolicy {
fn default() -> Self {
Self {
grow_at: 0.85,
shrink_at: 0.10,
min_capacity: 64,
max_capacity: 65536,
hysteresis: std::time::Duration::from_millis(100),
}
}
}
impl CapacityPolicy for DefaultCapacityPolicy {
fn decide(&self, obs: &CapacityPolicyObservation) -> Option<usize> {
if obs.since_last_morph < self.hysteresis {
return None;
}
let ratio = obs.fill_ratio();
if ratio >= self.grow_at && obs.current_capacity < self.max_capacity {
Some((obs.current_capacity * 2).min(self.max_capacity))
} else if ratio <= self.shrink_at && obs.current_capacity > self.min_capacity {
Some((obs.current_capacity / 2).max(self.min_capacity))
} else {
None
}
}
fn predict(&self, obs: &CapacityPolicyObservation) -> Option<usize> {
let ratio = obs.fill_ratio();
if ratio >= self.grow_at * 0.75 && obs.current_capacity < self.max_capacity {
Some((obs.current_capacity * 2).min(self.max_capacity))
} else if ratio <= self.shrink_at * 1.5 && obs.current_capacity > self.min_capacity {
Some((obs.current_capacity / 2).max(self.min_capacity))
} else {
None
}
}
}
pub struct CapacityAdaptiveRingSidecar {
handle: Option<std::thread::JoinHandle<()>>,
stop: Arc<std::sync::atomic::AtomicBool>,
morphs_triggered: Arc<AtomicU64>,
prewarms_issued: Arc<AtomicU64>,
}
impl CapacityAdaptiveRingSidecar {
pub fn spawn<P: CapacityPolicy>(
ring: Arc<CapacityAdaptiveRing>,
policy: P,
scan_interval: std::time::Duration,
) -> Self {
Self::spawn_gated(ring, policy, scan_interval, crate::policy_gate::GateConfig::default())
}
pub fn spawn_gated<P: CapacityPolicy>(
ring: Arc<CapacityAdaptiveRing>,
policy: P,
scan_interval: std::time::Duration,
gate_cfg: crate::policy_gate::GateConfig,
) -> Self {
let stop = Arc::new(std::sync::atomic::AtomicBool::new(false));
let morphs_triggered = Arc::new(AtomicU64::new(0));
let prewarms_issued = Arc::new(AtomicU64::new(0));
let stop_c = Arc::clone(&stop);
let morphs_c = Arc::clone(&morphs_triggered);
let prewarms_c = Arc::clone(&prewarms_issued);
let handle = std::thread::spawn(move || {
let mut last_morph = std::time::Instant::now();
let mut last_predicted: Option<usize> = None;
let mut gate = crate::policy_gate::ConfidenceGate::new(gate_cfg);
let mut last_peers = (0usize, 0usize);
let mut last_fill = 0.0f64;
let mut first_scan = true;
while !stop_c.load(Ordering::Acquire) {
let active = ring.ring_handle();
let obs = CapacityPolicyObservation {
current_capacity: ring.current_capacity(),
active_approx_len: active.approx_len(),
total_slot_capacity: active.total_slot_capacity(),
since_last_morph: last_morph.elapsed(),
};
let peers = (active.active_producers(), active.active_consumers());
drop(active);
let fill = obs.fill_ratio();
if !first_scan {
if peers != last_peers {
gate.shock();
}
if (fill - last_fill).abs() > 0.5 {
gate.shock();
}
}
last_peers = peers;
last_fill = fill;
first_scan = false;
if let Some(new_cap) = gate.observe(policy.decide(&obs))
&& ring.morph_capacity_to(new_cap).is_ok()
{
last_morph = std::time::Instant::now();
morphs_c.fetch_add(1, Ordering::Relaxed);
}
match policy.predict(&obs) {
Some(target) if target != ring.current_capacity() => {
if last_predicted == Some(target)
&& ring.warm_capacity() != Some(target)
&& ring.prewarm(target).is_ok()
{
prewarms_c.fetch_add(1, Ordering::Relaxed);
}
last_predicted = Some(target);
}
_ => last_predicted = None,
}
std::thread::sleep(scan_interval);
}
});
Self { handle: Some(handle), stop, morphs_triggered, prewarms_issued }
}
pub fn morphs_triggered(&self) -> u64 {
self.morphs_triggered.load(Ordering::Relaxed)
}
pub fn prewarms_issued(&self) -> u64 {
self.prewarms_issued.load(Ordering::Relaxed)
}
pub fn shutdown(mut self) {
self.stop.store(true, Ordering::Release);
if let Some(h) = self.handle.take() {
drop(h.join());
}
}
}
impl Drop for CapacityAdaptiveRingSidecar {
fn drop(&mut self) {
self.stop.store(true, Ordering::Release);
if let Some(h) = self.handle.take() {
drop(h.join());
}
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn create_anon_rejects_non_pow2() {
let r = CapacityAdaptiveRing::create_anon(1, 1, 100);
assert!(matches!(r, Err(CapacityMorphError::InvalidCapacity)));
}
#[test]
fn create_anon_rejects_capacity_below_two() {
let r = CapacityAdaptiveRing::create_anon(1, 1, 1);
assert!(matches!(r, Err(CapacityMorphError::InvalidCapacity)));
}
#[test]
fn anon_round_trip_after_create() {
let ring = CapacityAdaptiveRing::create_anon(1, 1, 64).unwrap();
ring.register_producer().unwrap();
ring.register_consumer().unwrap();
let payload = [0xAAu8; 56];
ring.try_send(0, &payload).unwrap();
let mut out = [0u8; 64];
let n = ring.try_recv(0, &mut out).unwrap();
assert!(n >= 56);
assert_eq!(&out[..56], &payload[..]);
}
#[test]
fn morph_grow_preserves_in_flight_items() {
let ring = CapacityAdaptiveRing::create_anon(1, 1, 64).unwrap();
ring.register_producer().unwrap();
ring.register_consumer().unwrap();
for i in 0..10u64 {
let mut payload = [0u8; 56];
payload[..8].copy_from_slice(&i.to_le_bytes());
ring.try_send(0, &payload).unwrap();
}
ring.morph_capacity_to(256).unwrap();
assert_eq!(ring.current_capacity(), 256);
assert_eq!(ring.pin_generation(), 1);
let mut got = Vec::new();
let mut out = [0u8; 64];
while ring.try_recv(0, &mut out).is_ok() {
let v = u64::from_le_bytes(out[..8].try_into().unwrap());
got.push(v);
}
got.sort();
assert_eq!(got, (0..10u64).collect::<Vec<_>>());
}
#[test]
fn morph_shrink_with_room_succeeds() {
let ring = CapacityAdaptiveRing::create_anon(1, 1, 256).unwrap();
ring.register_producer().unwrap();
ring.register_consumer().unwrap();
for i in 0..5u64 {
let mut payload = [0u8; 56];
payload[..8].copy_from_slice(&i.to_le_bytes());
ring.try_send(0, &payload).unwrap();
}
ring.morph_capacity_to(64).unwrap();
assert_eq!(ring.current_capacity(), 64);
let mut got = Vec::new();
let mut out = [0u8; 64];
while ring.try_recv(0, &mut out).is_ok() {
got.push(u64::from_le_bytes(out[..8].try_into().unwrap()));
}
got.sort();
assert_eq!(got, (0..5u64).collect::<Vec<_>>());
}
#[test]
fn morph_shrink_with_more_in_flight_than_new_capacity_succeeds() {
let ring = CapacityAdaptiveRing::create_anon(1, 1, 64).unwrap();
ring.register_producer().unwrap();
ring.register_consumer().unwrap();
for i in 0..40u64 {
let mut payload = [0u8; 56];
payload[..8].copy_from_slice(&i.to_le_bytes());
ring.try_send(0, &payload).unwrap();
}
ring.morph_capacity_to(16).expect("shrink succeeds via stale list");
assert_eq!(ring.current_capacity(), 16);
assert_eq!(ring.pin_generation(), 1);
let mut got = Vec::new();
let mut out = [0u8; 64];
while ring.try_recv(0, &mut out).is_ok() {
got.push(u64::from_le_bytes(out[..8].try_into().unwrap()));
}
assert_eq!(got, (0..40u64).collect::<Vec<_>>(),
"all 40 original items drained via stale list in send-order");
}
#[test]
fn morph_to_same_capacity_is_noop() {
let ring = CapacityAdaptiveRing::create_anon(1, 1, 64).unwrap();
let gen_before = ring.pin_generation();
ring.morph_capacity_to(64).unwrap();
assert_eq!(ring.pin_generation(), gen_before);
}
#[test]
fn morph_rejects_non_pow2_target() {
let ring = CapacityAdaptiveRing::create_anon(1, 1, 64).unwrap();
let r = ring.morph_capacity_to(100);
assert!(matches!(r, Err(CapacityMorphError::InvalidCapacity)));
}
#[test]
fn pin_invalidates_after_morph() {
let ring = CapacityAdaptiveRing::create_anon(1, 1, 64).unwrap();
ring.register_producer().unwrap();
ring.register_consumer().unwrap();
let pin = ring.pin_current_capacity();
assert!(pin.is_still_valid());
assert_eq!(pin.capacity(), 64);
ring.morph_capacity_to(128).unwrap();
assert!(!pin.is_still_valid());
let pin2 = ring.pin_current_capacity();
assert!(pin2.is_still_valid());
assert_eq!(pin2.capacity(), 128);
}
#[test]
fn multiple_grow_morphs_increment_generation_correctly() {
let ring = CapacityAdaptiveRing::create_anon(1, 1, 4).unwrap();
ring.register_producer().unwrap();
ring.register_consumer().unwrap();
assert_eq!(ring.pin_generation(), 0);
ring.morph_capacity_to(8).unwrap();
assert_eq!(ring.pin_generation(), 1);
ring.morph_capacity_to(16).unwrap();
assert_eq!(ring.pin_generation(), 2);
ring.morph_capacity_to(64).unwrap();
assert_eq!(ring.pin_generation(), 3);
assert_eq!(ring.current_capacity(), 64);
}
#[test]
fn stamped_capacity_morph_preserves_ordering_axis() {
let ring = CapacityAdaptiveRing::create_anon_stamped(2, 1, 64).unwrap();
assert!(ring.is_stamped());
ring.register_producer().unwrap();
ring.register_consumer().unwrap();
ring.set_ordering_mode(OrderingMode::MergeByStamp).unwrap();
for i in 0..6u64 {
let mut payload = [0u8; 48];
payload[..8].copy_from_slice(&i.to_le_bytes());
ring.try_send(0, &payload).unwrap();
}
ring.morph_capacity_to(256).unwrap();
assert_eq!(ring.ordering_mode(), Some(OrderingMode::MergeByStamp),
"the live mode flag must follow the capacity morph");
for i in 6..10u64 {
let mut payload = [0u8; 48];
payload[..8].copy_from_slice(&i.to_le_bytes());
ring.try_send(0, &payload).unwrap();
}
let mut out = [0u8; 64];
let mut got = Vec::new();
while ring.try_recv(0, &mut out).is_ok() {
got.push(u64::from_le_bytes(out[..8].try_into().unwrap()));
}
assert_eq!(got, (0..10u64).collect::<Vec<_>>(),
"ordering must hold across the capacity morph boundary");
assert_eq!(ring.inversions(), 0);
}
#[test]
fn cross_thread_concurrent_send_recv_through_morphs() {
use std::thread;
let ring = Arc::new(CapacityAdaptiveRing::create_anon(1, 1, 64).unwrap());
ring.register_producer().unwrap();
ring.register_consumer().unwrap();
let n = 5_000u64;
let r_prod = Arc::clone(&ring);
let prod = thread::spawn(move || {
for i in 0..n {
let mut payload = [0u8; 56];
payload[..8].copy_from_slice(&i.to_le_bytes());
while r_prod.try_send(0, &payload).is_err() {
std::hint::spin_loop();
}
}
});
let r_cons = Arc::clone(&ring);
let cons = thread::spawn(move || {
let mut got = Vec::with_capacity(n as usize);
let mut out = [0u8; 64];
while got.len() < n as usize {
if r_cons.try_recv(0, &mut out).is_ok() {
got.push(u64::from_le_bytes(out[..8].try_into().unwrap()));
}
}
got
});
let r_morph = Arc::clone(&ring);
let morph = thread::spawn(move || {
let targets = [128usize, 256, 128, 64];
for t in targets {
std::thread::sleep(std::time::Duration::from_micros(500));
r_morph.morph_capacity_to(t).expect("morph succeeds");
}
});
prod.join().unwrap();
morph.join().unwrap();
let mut got = cons.join().unwrap();
got.sort();
let expected: Vec<u64> = (0..n).collect();
assert_eq!(got, expected);
}
#[test]
fn prewarm_hit_consumes_cache_and_morph_works() {
let ring = CapacityAdaptiveRing::create_anon(1, 1, 64).unwrap();
ring.register_producer().unwrap();
ring.register_consumer().unwrap();
ring.prewarm(256).unwrap();
assert_eq!(ring.warm_capacity(), Some(256));
assert_eq!(ring.warm_hits(), 0);
for i in 0..10u64 {
let mut payload = [0u8; 56];
payload[..8].copy_from_slice(&i.to_le_bytes());
ring.try_send(0, &payload).unwrap();
}
ring.morph_capacity_to(256).unwrap();
assert_eq!(ring.warm_hits(), 1, "the morph must consume the prediction");
assert_eq!(ring.warm_capacity(), None, "the slot is one-shot");
assert_eq!(ring.current_capacity(), 256);
assert_eq!(ring.pin_generation(), 1);
ring.try_send(0, &[0xBBu8; 56]).unwrap();
let mut out = [0u8; 64];
let mut got = Vec::new();
while ring.try_recv(0, &mut out).is_ok() {
got.push(u64::from_le_bytes(out[..8].try_into().unwrap()));
}
assert_eq!(got.len(), 11);
assert_eq!(&got[..10], &(0..10u64).collect::<Vec<_>>()[..]);
}
#[test]
fn prewarm_mismatch_stays_cached_and_cold_path_runs() {
let ring = CapacityAdaptiveRing::create_anon(1, 1, 64).unwrap();
ring.register_producer().unwrap();
ring.register_consumer().unwrap();
ring.prewarm(512).unwrap();
ring.morph_capacity_to(256).unwrap();
assert_eq!(ring.warm_hits(), 0, "mismatched prediction must not be consumed");
assert_eq!(ring.warm_capacity(), Some(512), "mismatch stays cached");
assert_eq!(ring.current_capacity(), 256);
ring.morph_capacity_to(512).unwrap();
assert_eq!(ring.warm_hits(), 1, "the cached 512 serves the later morph");
assert_eq!(ring.warm_capacity(), None);
}
#[test]
fn prewarm_rejects_non_pow2() {
let ring = CapacityAdaptiveRing::create_anon(1, 1, 64).unwrap();
assert!(matches!(ring.prewarm(100), Err(CapacityMorphError::InvalidCapacity)));
assert!(matches!(ring.prewarm(1), Err(CapacityMorphError::InvalidCapacity)));
assert_eq!(ring.warm_capacity(), None);
}
#[test]
fn prewarm_same_capacity_is_idempotent_and_clear_drops() {
let ring = CapacityAdaptiveRing::create_anon(1, 1, 64).unwrap();
ring.prewarm(128).unwrap();
ring.prewarm(128).unwrap();
assert_eq!(ring.warm_capacity(), Some(128));
ring.prewarm(256).unwrap();
assert_eq!(ring.warm_capacity(), Some(256), "new prediction replaces the old");
ring.clear_warm();
assert_eq!(ring.warm_capacity(), None);
}
#[test]
fn warm_morph_preserves_stamps_and_shape() {
let ring = CapacityAdaptiveRing::create_anon_stamped(2, 1, 64).unwrap();
ring.register_producer().unwrap();
ring.register_consumer().unwrap();
ring.set_ordering_mode(OrderingMode::MergeByStamp).unwrap();
ring.ring_handle()
.morph_to(crate::adaptive_ring::RingShape::Mpsc)
.unwrap();
for i in 0..6u64 {
let mut payload = [0u8; 48];
payload[..8].copy_from_slice(&i.to_le_bytes());
ring.try_send(0, &payload).unwrap();
}
ring.prewarm(256).unwrap();
ring.morph_capacity_to(256).unwrap();
assert_eq!(ring.warm_hits(), 1);
assert_eq!(ring.ordering_mode(), Some(OrderingMode::MergeByStamp),
"live mode flag must follow a warm-hit morph");
assert_eq!(ring.ring_handle().current_shape(),
crate::adaptive_ring::RingShape::Mpsc,
"shape must be mirrored onto the warm backing");
for i in 6..10u64 {
let mut payload = [0u8; 48];
payload[..8].copy_from_slice(&i.to_le_bytes());
ring.try_send(0, &payload).unwrap();
}
let mut out = [0u8; 64];
let mut got = Vec::new();
while ring.try_recv(0, &mut out).is_ok() {
got.push(u64::from_le_bytes(out[..8].try_into().unwrap()));
}
assert_eq!(got, (0..10u64).collect::<Vec<_>>(),
"send order must hold across a warm-hit morph (seeded stamps)");
assert_eq!(ring.inversions(), 0);
}
#[test]
fn warm_morph_file_locale_round_trips() {
let dir = std::env::temp_dir().join(format!(
"subetha_warm_file_{}", std::process::id(),
));
std::fs::create_dir_all(&dir).unwrap();
let base = dir.join("warm_probe");
{
let ring = CapacityAdaptiveRing::create(&base, 1, 1, 64).unwrap();
ring.register_producer().unwrap();
ring.register_consumer().unwrap();
for i in 0..5u64 {
let mut payload = [0u8; 56];
payload[..8].copy_from_slice(&i.to_le_bytes());
ring.try_send(0, &payload).unwrap();
}
ring.prewarm(128).unwrap();
ring.morph_capacity_to(128).unwrap();
assert_eq!(ring.warm_hits(), 1);
ring.prewarm(64).unwrap();
ring.morph_capacity_to(64).unwrap();
assert_eq!(ring.warm_hits(), 2);
let mut out = [0u8; 64];
let mut got = Vec::new();
while ring.try_recv(0, &mut out).is_ok() {
got.push(u64::from_le_bytes(out[..8].try_into().unwrap()));
}
assert_eq!(got, (0..5u64).collect::<Vec<_>>());
}
drop(std::fs::remove_dir_all(&dir));
}
#[test]
fn default_policy_predict_bands() {
let policy = DefaultCapacityPolicy::default(); let obs = |len: usize, cap: usize| CapacityPolicyObservation {
current_capacity: cap,
active_approx_len: len,
total_slot_capacity: cap,
since_last_morph: std::time::Duration::ZERO,
};
assert_eq!(policy.predict(&obs(716, 1024)), Some(2048));
assert_eq!(policy.predict(&obs(512, 1024)), None);
assert_eq!(policy.predict(&obs(143, 1024)), Some(512));
assert_eq!(policy.predict(&obs(60000, 65536)), None);
assert_eq!(policy.predict(&obs(0, 64)), None);
let fresh = CapacityPolicyObservation {
since_last_morph: std::time::Duration::ZERO,
..obs(716, 1024)
};
assert_eq!(policy.decide(&fresh), None, "decide is hysteresis-gated");
assert_eq!(policy.predict(&fresh), Some(2048), "predict is not");
}
#[test]
fn policy_without_predict_override_never_prewarms() {
struct GrowOnly;
impl CapacityPolicy for GrowOnly {
fn decide(&self, obs: &CapacityPolicyObservation) -> Option<usize> {
(obs.fill_ratio() >= 0.85).then_some(obs.current_capacity * 2)
}
}
let ring = Arc::new(CapacityAdaptiveRing::create_anon(1, 1, 64).unwrap());
ring.register_producer().unwrap();
ring.register_consumer().unwrap();
let sidecar = CapacityAdaptiveRingSidecar::spawn(
Arc::clone(&ring), GrowOnly, std::time::Duration::from_millis(2),
);
for _ in 0..50 {
ring.try_send(0, &[0u8; 56]).ok();
}
std::thread::sleep(std::time::Duration::from_millis(50));
assert_eq!(sidecar.prewarms_issued(), 0,
"trait-default predict() must keep today's behavior");
sidecar.shutdown();
}
#[test]
fn sidecar_prewarms_on_sustained_trend_then_morph_hits_warm() {
struct Banded;
impl CapacityPolicy for Banded {
fn decide(&self, obs: &CapacityPolicyObservation) -> Option<usize> {
(obs.fill_ratio() >= 0.85).then_some(obs.current_capacity * 2)
}
fn predict(&self, obs: &CapacityPolicyObservation) -> Option<usize> {
(obs.fill_ratio() >= 0.60).then_some(obs.current_capacity * 2)
}
}
let ring = Arc::new(CapacityAdaptiveRing::create_anon(1, 1, 64).unwrap());
ring.register_producer().unwrap();
ring.register_consumer().unwrap();
let sidecar = CapacityAdaptiveRingSidecar::spawn(
Arc::clone(&ring), Banded, std::time::Duration::from_millis(2),
);
for _ in 0..45 {
ring.try_send(0, &[0u8; 56]).unwrap();
}
let deadline = std::time::Instant::now() + std::time::Duration::from_secs(2);
while ring.warm_capacity() != Some(128) {
assert!(std::time::Instant::now() < deadline,
"sidecar must prewarm 128 from the sustained trend");
std::thread::sleep(std::time::Duration::from_millis(2));
}
assert!(sidecar.prewarms_issued() >= 1);
for _ in 0..11 {
ring.try_send(0, &[0u8; 56]).unwrap();
}
let deadline = std::time::Instant::now() + std::time::Duration::from_secs(2);
while ring.current_capacity() != 128 {
assert!(std::time::Instant::now() < deadline,
"sidecar must morph to 128 once fill crosses decide");
std::thread::sleep(std::time::Duration::from_millis(2));
}
assert_eq!(ring.warm_hits(), 1,
"the sidecar-driven morph must consume the prewarmed backing");
sidecar.shutdown();
let mut out = [0u8; 64];
let mut n = 0;
while ring.try_recv(0, &mut out).is_ok() {
n += 1;
}
assert_eq!(n, 56);
}
#[test]
fn compound_capacity_plus_shape_is_one_generation() {
let ring = CapacityAdaptiveRing::create_anon(4, 1, 64).unwrap();
ring.register_producer().unwrap();
ring.register_consumer().unwrap();
for i in 0..10u64 {
let mut p = [0u8; 56];
p[..8].copy_from_slice(&i.to_le_bytes());
ring.try_send(0, &p).unwrap();
}
ring.morph_to_config(&RingConfig {
shape: Some(RingShape::Mpmc),
capacity: Some(512),
locale: None,
})
.unwrap();
assert_eq!(ring.pin_generation(), 1,
"two axes, ONE pin invalidation");
assert_eq!(ring.current_capacity(), 512);
assert_eq!(ring.ring_handle().current_shape(), RingShape::Mpmc);
for _ in 0..3 {
ring.register_producer().unwrap();
}
for pid in 1..4usize {
let mut p = [0u8; 56];
p[..8].copy_from_slice(&(100 + pid as u64).to_le_bytes());
ring.try_send(pid, &p).unwrap();
}
let mut out = [0u8; 64];
let mut got = Vec::new();
while ring.try_recv(0, &mut out).is_ok() {
got.push(u64::from_le_bytes(out[..8].try_into().unwrap()));
}
got.sort();
let mut expected: Vec<u64> = (0..10).collect();
expected.extend([101, 102, 103]);
assert_eq!(got, expected);
}
#[test]
fn shape_only_config_morphs_in_place_without_pin_bump() {
let ring = CapacityAdaptiveRing::create_anon(4, 1, 64).unwrap();
ring.register_producer().unwrap();
ring.register_consumer().unwrap();
ring.try_send(0, &[0x11u8; 56]).unwrap();
let active_before = ring.ring_handle();
ring.morph_to_config(&RingConfig {
shape: Some(RingShape::Mpsc),
..RingConfig::default()
})
.unwrap();
assert_eq!(ring.pin_generation(), 0,
"in-place shape morph must not invalidate the capacity pin");
assert!(Arc::ptr_eq(&active_before, &ring.ring_handle()),
"active backing must be the same instance");
assert_eq!(ring.ring_handle().current_shape(), RingShape::Mpsc);
let mut out = [0u8; 64];
assert!(ring.try_recv(0, &mut out).is_ok(),
"in-flight item survives the in-place shape morph");
}
#[test]
fn config_noop_when_every_axis_matches() {
let ring = CapacityAdaptiveRing::create_anon(1, 1, 64).unwrap();
ring.morph_to_config(&RingConfig::default()).unwrap();
ring.morph_to_config(&RingConfig {
shape: Some(RingShape::Spsc),
capacity: Some(64),
locale: Some(BackingTarget::Anon),
})
.unwrap();
assert_eq!(ring.pin_generation(), 0);
}
#[test]
fn compound_locale_change_drains_across_locales() {
let dir = std::env::temp_dir().join(format!(
"subetha_compound_locale_{}", std::process::id(),
));
std::fs::create_dir_all(&dir).unwrap();
let base = dir.join("compound");
{
let ring = CapacityAdaptiveRing::create_anon(1, 1, 64).unwrap();
ring.register_producer().unwrap();
ring.register_consumer().unwrap();
for i in 0..8u64 {
let mut p = [0u8; 56];
p[..8].copy_from_slice(&i.to_le_bytes());
ring.try_send(0, &p).unwrap();
}
ring.morph_to_config(&RingConfig {
shape: None,
capacity: Some(256),
locale: Some(BackingTarget::File(base.clone())),
})
.unwrap();
assert_eq!(ring.pin_generation(), 1);
assert_eq!(ring.current_capacity(), 256);
ring.try_send(0, &{
let mut p = [0u8; 56];
p[..8].copy_from_slice(&99u64.to_le_bytes());
p
}).unwrap();
let mut out = [0u8; 64];
let mut got = Vec::new();
while ring.try_recv(0, &mut out).is_ok() {
got.push(u64::from_le_bytes(out[..8].try_into().unwrap()));
}
let mut expected: Vec<u64> = (0..8).collect();
expected.push(99);
assert_eq!(got, expected);
ring.morph_capacity_to(512).unwrap();
let file_backings: Vec<_> = std::fs::read_dir(&dir).unwrap()
.filter_map(|e| e.ok())
.filter(|e| e.file_name().to_string_lossy().contains("cap_512"))
.collect();
assert!(!file_backings.is_empty(),
"post-retarget morphs must allocate file backings");
}
drop(std::fs::remove_dir_all(&dir));
}
#[test]
fn repatch_prewarm_config_full_target_hits() {
let ring = CapacityAdaptiveRing::create_anon(4, 1, 64).unwrap();
ring.register_producer().unwrap();
ring.register_consumer().unwrap();
let target = RingConfig {
shape: Some(RingShape::Mpmc),
capacity: Some(1024),
locale: None,
};
ring.prewarm_config(&target).unwrap();
assert_eq!(ring.warm_capacity(), Some(1024));
ring.morph_to_config(&target).unwrap();
assert_eq!(ring.warm_hits(), 1,
"repatch must consume the full-target prediction");
assert_eq!(ring.ring_handle().current_shape(), RingShape::Mpmc);
assert_eq!(ring.current_capacity(), 1024);
}
#[test]
fn warm_key_locale_mismatch_is_cold() {
let dir = std::env::temp_dir().join(format!(
"subetha_warm_locale_key_{}", std::process::id(),
));
std::fs::create_dir_all(&dir).unwrap();
{
let ring = CapacityAdaptiveRing::create_anon(1, 1, 64).unwrap();
ring.prewarm(256).unwrap();
ring.morph_to_config(&RingConfig {
shape: None,
capacity: Some(256),
locale: Some(BackingTarget::File(dir.join("keyed"))),
})
.unwrap();
assert_eq!(ring.warm_hits(), 0,
"an anon-built backing must never serve a file-locale morph");
assert_eq!(ring.warm_capacity(), Some(256),
"the mismatched prediction stays cached");
ring.clear_warm();
}
drop(std::fs::remove_dir_all(&dir));
}
#[test]
fn stale_pops_counts_transition_items() {
let ring = CapacityAdaptiveRing::create_anon(1, 1, 64).unwrap();
ring.register_producer().unwrap();
ring.register_consumer().unwrap();
for i in 0..7u64 {
let mut p = [0u8; 56];
p[..8].copy_from_slice(&i.to_le_bytes());
ring.try_send(0, &p).unwrap();
}
ring.morph_capacity_to(256).unwrap();
ring.try_send(0, &[0x22u8; 56]).unwrap();
let mut out = [0u8; 64];
while ring.try_recv(0, &mut out).is_ok() {}
assert_eq!(ring.stale_pops(), 7,
"exactly the pre-morph items traverse the stale walk");
}
}