rusty_engine/monitoring/

ring_buffer.rs

//! Lock-free ring buffer implementation for zero-latency metric recording
//!
//! This module provides a single-producer, multiple-consumer (SPMC) ring buffer
//! optimized for high-frequency metric recording with minimal overhead.
//!
//! # Architecture Overview
//!
//! The ring buffer uses atomic operations to avoid locks in the hot path, ensuring
//! minimal impact on trading system performance. It supports a single producer with
//! multiple concurrent consumers.
//!
//! **Key Design Features:**
//! - Zero-allocation in hot paths (all memory pre-allocated)
//! - Lock-free operations using compare-and-swap
//! - Cache-friendly data layout with alignment
//! - Overflow tracking for capacity monitoring
//!
//! # Capacity Requirements and Planning
//!
//! ## Power of 2 Requirement
//!
//! **MANDATORY**: Capacity must be a power of 2 (2, 4, 8, 16, 32, 64, 128, ...).
//! This enables:
//! - Bitwise AND operations (`& capacity_mask`) instead of expensive modulo
//! - ~10-20% performance improvement in hot paths
//! - Optimal cache line alignment for atomic operations
//!
//! ## Effective Capacity
//!
//! **Important**: Effective capacity is `capacity - 1` due to full/empty distinction.
//! The buffer reserves one slot to differentiate between full and empty states.
//!
//! ## Capacity Planning Formula
//!
//! ```text
//! Required_Capacity = Peak_Throughput_Per_Second × Max_Consumer_Delay_Seconds × Safety_Factor
//!
//! Where:
//! - Peak_Throughput_Per_Second: Maximum messages/second during bursts
//! - Max_Consumer_Delay_Seconds: Worst-case time between drain operations
//! - Safety_Factor: 2-5x buffer (3x recommended for production)
//! ```
//!
//! ## HFT Capacity Guidelines by Use Case
//!
//! | Use Case | Frequency | Recommended Capacity | Effective Items | Memory (8-byte) | Cache Level |
//! |----------|-----------|---------------------|-----------------|-----------------|-------------|
//! | Trade execution | Sub-μs | 64-128 | 63-127 | ~0.5-1KB | L1 |
//! | Order book updates | μs | 512-1024 | 511-1023 | ~4-8KB | L1/L2 |
//! | Risk calculations | ms | 2048-4096 | 2047-4095 | ~16-32KB | L2 |
//! | Strategy signals | s | 8192-16384 | 8191-16383 | ~64-128KB | L2/L3 |
//! | Historical logging | min+ | 32768+ | 32767+ | ~256KB+ | Memory |
//!
//! ## Performance Characteristics by Capacity
//!
//! - **64-256 items**: Ultra-low latency (~10-50ns), fits in L1 cache (32-64KB)
//! - **512-2048 items**: High performance (~50-100ns), excellent L2 utilization
//! - **4096-8192 items**: Balanced (~100-200ns), L3 cache friendly
//! - **16384+ items**: High capacity (~200ns+), may cause cache misses
//!
//! ## Memory Usage Calculation
//!
//! ```text
//! Total Memory = capacity × size_of::<T>() + overhead (~64 bytes)
//!
//! Examples:
//! - capacity=64, T=u64: ~576 bytes (fits in single cache line group)
//! - capacity=1024, T=u64: ~8KB (good L1 cache utilization)
//! - capacity=8192, T=u64: ~64KB (fits comfortably in L2 cache)
//! ```
//!
//! # Overflow Monitoring and Capacity Adjustment
//!
//! ## Overflow Implications
//!
//! Each overflow represents:
//! - **Lost metric data** that could impact trading decisions
//! - **Potential performance degradation** due to buffer pressure
//! - **Need for capacity adjustment** or consumer optimization
//!
//! ## Production Monitoring Guidelines
//!
//! - **Zero overflows**: Optimal capacity, consider reducing if over-provisioned
//! - **Rare overflows (< 0.1%)**: Acceptable for non-critical metrics
//! - **Regular overflows (> 1%)**: Increase capacity or optimize consumers
//! - **Frequent overflows (> 10%)**: Critical issue, system overload
//!
//! ## Capacity Adjustment Strategy
//!
//! Based on overflow rate per minute:
//! - **> 100 overflows/min**: Double capacity immediately
//! - **10-100 overflows/min**: Increase capacity by 50%
//! - **1-10 overflows/min**: Monitor trends, minor adjustment
//! - **0 overflows/min**: Current capacity sufficient
//!
//! # Usage Examples
//!
//! ## Basic Usage
//!
//! ```rust
//! use crate::monitoring::ring_buffer::SharedRingBuffer;
//!
//! // Create buffer for order book updates
//! let buffer = SharedRingBuffer::new(1024);
//!
//! // Producer: Push metrics
//! if !buffer.push(metric_data) {
//!     eprintln!("Buffer overflow! Consider increasing capacity");
//! }
//!
//! // Consumer: Drain in batches
//! let metrics = buffer.drain(64);
//! for metric in metrics {
//!     process_metric(metric);
//! }
//! ```
//!
//! ## Capacity Planning Examples
//!
//! ```rust
//! use crate::monitoring::ring_buffer::SharedRingBuffer;
//!
//! // Example 1: High-frequency order book (50K updates/sec, 1ms drain)
//! // Required: 50,000 × 0.001 = 50, Safety: 4x = 200, Next power of 2: 256
//! let order_book = SharedRingBuffer::new(256);
//!
//! // Example 2: Strategy signals (1K signals/sec, 10ms drain)
//! // Required: 1,000 × 0.01 = 10, Safety: 8x = 80, Next power of 2: 128
//! let strategy = SharedRingBuffer::new(128);
//!
//! // Example 3: Risk metrics (10K metrics/sec, 5ms drain)
//! // Required: 10,000 × 0.005 = 50, Safety: 5x = 250, Next power of 2: 512
//! let risk = SharedRingBuffer::new(512);
//! ```
//!
//! # Performance Optimization Guidelines
//!
//! ## Multi-Consumer Efficiency
//!
//! - Use `drain(batch_size)` instead of repeated `pop()` calls
//! - Recommended batch sizes: 16-64 items per drain operation
//! - Each consumer competes for items (no broadcasting)
//!
//! ## Overflow Handling Strategy
//!
//! When buffer overflows:
//! - Data is lost (ring buffer does not resize)
//! - Monitor `overflow_count()` for capacity planning
//! - Consider: larger capacity, faster consumers, or multiple buffers

use std::cell::UnsafeCell;
use std::mem::MaybeUninit;
use std::sync::Arc;
use std::sync::atomic::{AtomicUsize, Ordering};

/// A lock-free ring buffer for metric storage
///
/// This implementation uses atomic operations to avoid locks in the hot path,
/// ensuring minimal impact on trading system performance.
pub struct RingBuffer<T: Clone> {
    /// The buffer storage
    buffer: Vec<UnsafeCell<MaybeUninit<T>>>,
    /// Write position (only modified by single producer)
    write_pos: AtomicUsize,
    /// Read position (can be modified by multiple consumers)
    read_pos: AtomicUsize,
    /// Capacity mask (capacity - 1, for fast modulo)
    capacity_mask: usize,
    /// Total capacity
    capacity: usize,
    /// Overflow counter
    overflow_count: AtomicUsize,
}

unsafe impl<T: Clone + Send> Send for RingBuffer<T> {}
unsafe impl<T: Clone + Send> Sync for RingBuffer<T> {}

impl<T: Clone> RingBuffer<T> {
    /// Create a new ring buffer with the specified capacity
    ///
    /// # Arguments
    ///
    /// * `capacity` - Buffer capacity, **must be a power of 2** (2, 4, 8, 16, 32, 64, 128, ...)
    ///
    /// # Returns
    ///
    /// A new RingBuffer with effective capacity of `capacity - 1` items.
    ///
    /// # Panics
    ///
    /// Panics if capacity is not a power of 2 or is zero.
    ///
    /// # Example
    ///
    /// ```rust
    /// # use crate::monitoring::ring_buffer::RingBuffer;
    /// let buffer = RingBuffer::new(1024);  // Can hold 1023 items
    /// ```
    #[must_use]
    pub fn new(capacity: usize) -> Self {
        assert!(capacity.is_power_of_two(), "Capacity must be a power of 2");

        let mut buffer = Vec::with_capacity(capacity);
        for _ in 0..capacity {
            buffer.push(UnsafeCell::new(MaybeUninit::uninit()));
        }

        Self {
            buffer,
            write_pos: AtomicUsize::new(0),
            read_pos: AtomicUsize::new(0),
            capacity_mask: capacity - 1,
            capacity,
            overflow_count: AtomicUsize::new(0),
        }
    }

    /// Push an item into the ring buffer
    ///
    /// Returns `true` if the item was successfully pushed, `false` if the buffer was full.
    /// When full, the overflow counter is incremented and data is lost.
    ///
    /// # Arguments
    ///
    /// * `item` - The item to push into the buffer
    ///
    /// # Returns
    ///
    /// `true` if successful, `false` if buffer was full (overflow)
    #[inline(always)]
    pub fn push(&self, item: T) -> bool {
        let write_pos = self.write_pos.load(Ordering::Acquire);
        let read_pos = self.read_pos.load(Ordering::Acquire);

        // Check if buffer is full
        let next_write_pos = (write_pos + 1) & self.capacity_mask;
        let is_full = next_write_pos == read_pos;

        if is_full {
            // Buffer is full, increment overflow counter
            self.overflow_count.fetch_add(1, Ordering::Relaxed);
            // Optionally, we could force advance the read position here
            // to make room, but that would add complexity
            return false;
        }

        // Write the item
        unsafe {
            let slot_ptr = self.buffer[write_pos].get();
            (*slot_ptr).write(item);
        }

        // Update write position
        self.write_pos.store(next_write_pos, Ordering::Release);

        true
    }

    /// Pop an item from the ring buffer
    ///
    /// # Returns
    ///
    /// `Some(item)` if an item was available, `None` if the buffer was empty
    pub fn pop(&self) -> Option<T> {
        loop {
            let read_pos = self.read_pos.load(Ordering::Acquire);
            let write_pos = self.write_pos.load(Ordering::Acquire);

            // Check if buffer is empty
            if read_pos == write_pos {
                return None;
            }

            // Try to advance read position
            let next_read_pos = (read_pos + 1) & self.capacity_mask;

            // Use compare_exchange to ensure we're the only one reading this slot
            match self.read_pos.compare_exchange(
                read_pos,
                next_read_pos,
                Ordering::AcqRel,
                Ordering::Acquire,
            ) {
                Ok(_) => {
                    // Successfully claimed this slot
                    unsafe {
                        let slot = &*self.buffer[read_pos].get();
                        let item = slot.assume_init_read().clone();
                        return Some(item);
                    }
                }
                Err(_) => {
                    // Another thread beat us, retry
                    continue;
                }
            }
        }
    }

    /// Drain up to `max_items` from the buffer
    ///
    /// More efficient than repeatedly calling `pop()` for batch operations.
    ///
    /// # Arguments
    ///
    /// * `max_items` - Maximum number of items to drain
    ///
    /// # Returns
    ///
    /// Vector of drained items (may be fewer than `max_items`)
    pub fn drain(&self, max_items: usize) -> Vec<T> {
        let mut items = Vec::with_capacity(max_items.min(self.capacity));

        for _ in 0..max_items {
            match self.pop() {
                Some(item) => items.push(item),
                None => break,
            }
        }

        items
    }

    /// Returns the current number of items in the buffer
    pub fn len(&self) -> usize {
        let write_pos = self.write_pos.load(Ordering::Acquire);
        let read_pos = self.read_pos.load(Ordering::Acquire);

        if write_pos >= read_pos {
            write_pos - read_pos
        } else {
            self.capacity - read_pos + write_pos
        }
    }

    /// Returns `true` if the buffer contains no items
    pub fn is_empty(&self) -> bool {
        self.write_pos.load(Ordering::Acquire) == self.read_pos.load(Ordering::Acquire)
    }

    /// Returns the total number of overflow events
    pub fn overflow_count(&self) -> usize {
        self.overflow_count.load(Ordering::Relaxed)
    }

    /// Resets the overflow counter to zero
    pub fn reset_overflow_count(&self) {
        self.overflow_count.store(0, Ordering::Relaxed);
    }
}

/// A thread-safe wrapper around RingBuffer for shared access
pub struct SharedRingBuffer<T: Clone> {
    inner: Arc<RingBuffer<T>>,
}

impl<T: Clone> SharedRingBuffer<T> {
    /// Creates a new shared ring buffer with the specified capacity
    ///
    /// This is the primary interface for creating ring buffers in the HFT system.
    /// The buffer is wrapped in an `Arc` for efficient sharing between threads.
    ///
    /// # Arguments
    ///
    /// * `capacity` - Buffer capacity, **must be a power of 2** (2, 4, 8, 16, 32, 64, 128, ...)
    ///
    /// # Returns
    ///
    /// A SharedRingBuffer with effective capacity of `capacity - 1` items.
    ///
    /// # Panics
    ///
    /// Panics if capacity is not a power of 2 or is zero.
    ///
    /// # Example
    ///
    /// ```rust
    /// # use crate::monitoring::ring_buffer::SharedRingBuffer;
    /// let buffer = SharedRingBuffer::new(1024);  // For high-frequency data
    /// ```
    #[must_use]
    pub fn new(capacity: usize) -> Self {
        Self {
            inner: Arc::new(RingBuffer::new(capacity)),
        }
    }

    /// Push an item into the ring buffer
    ///
    /// # Arguments
    ///
    /// * `item` - The item to push
    ///
    /// # Returns
    ///
    /// `true` if successful, `false` if buffer was full
    pub fn push(&self, item: T) -> bool {
        self.inner.push(item)
    }

    /// Pop an item from the ring buffer
    ///
    /// # Returns
    ///
    /// `Some(item)` if available, `None` if buffer was empty
    pub fn pop(&self) -> Option<T> {
        self.inner.pop()
    }

    /// Drain up to `max_items` from the ring buffer
    ///
    /// More efficient than repeatedly calling `pop()` for batch operations.
    ///
    /// # Arguments
    ///
    /// * `max_items` - Maximum number of items to drain
    ///
    /// # Returns
    ///
    /// Vector of drained items (may be fewer than requested)
    pub fn drain(&self, max_items: usize) -> Vec<T> {
        self.inner.drain(max_items)
    }

    /// Returns the current number of items in the buffer
    pub fn len(&self) -> usize {
        self.inner.len()
    }

    /// Returns `true` if the buffer contains no items
    pub fn is_empty(&self) -> bool {
        self.inner.is_empty()
    }

    /// Returns the number of times the ring buffer has overflowed
    ///
    /// An overflow occurs when `push()` is called on a full buffer, resulting in data loss.
    /// This counter is critical for monitoring buffer saturation and capacity planning.
    ///
    /// # Returns
    ///
    /// The total number of overflow events since buffer creation or last reset.
    ///
    /// # Example
    ///
    /// ```rust
    /// # use crate::monitoring::ring_buffer::SharedRingBuffer;
    /// let buffer = SharedRingBuffer::new(64);
    ///
    /// // Monitor overflow rate
    /// let initial = buffer.overflow_count();
    /// // ... time passes ...
    /// let overflows = buffer.overflow_count() - initial;
    /// if overflows > 0 {
    ///     eprintln!("Warning: {} overflows detected", overflows);
    /// }
    /// ```
    pub fn overflow_count(&self) -> usize {
        self.inner.overflow_count()
    }
}

impl<T: Clone> Clone for SharedRingBuffer<T> {
    fn clone(&self) -> Self {
        Self {
            inner: Arc::clone(&self.inner),
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::sync::Arc;
    use std::thread;

    #[test]
    fn test_ring_buffer_basic() {
        let rb = RingBuffer::new(4);

        // Test empty buffer
        assert!(rb.is_empty());
        assert_eq!(rb.len(), 0);
        assert!(rb.pop().is_none());

        // Test push and pop
        assert!(rb.push(1));
        assert!(rb.push(2));
        assert_eq!(rb.len(), 2);

        assert_eq!(rb.pop(), Some(1));
        assert_eq!(rb.pop(), Some(2));
        assert!(rb.pop().is_none());
    }

    #[test]
    fn test_ring_buffer_wrap_around() {
        let rb = RingBuffer::new(4);

        // CRITICAL TEST: Verify overflow counter starts at 0
        assert_eq!(rb.overflow_count(), 0, "Overflow counter should start at 0");

        // Fill buffer (capacity 4 can hold 3 items due to full/empty distinction)
        assert!(rb.push(1));
        assert!(rb.push(2));
        assert!(rb.push(3));

        // CRITICAL: No overflow should occur when buffer is exactly full but not overflowing
        assert_eq!(
            rb.overflow_count(),
            0,
            "Overflow counter should remain 0 when buffer is full but not overflowing"
        );

        // Pop some items to make space
        assert_eq!(rb.pop(), Some(1));
        assert_eq!(rb.pop(), Some(2));

        // CRITICAL: Verify overflow counter remains 0 after popping
        assert_eq!(
            rb.overflow_count(),
            0,
            "Overflow counter should remain 0 after popping items"
        );

        // Push more items (wrap around) - we now have 1 item (3) so we can push 2 more
        // This tests the wrap-around behavior where write_pos wraps back to 0
        assert!(rb.push(4));
        assert_eq!(
            rb.overflow_count(),
            0,
            "Overflow counter should remain 0 during wrap-around (push 4)"
        );

        assert!(rb.push(5));
        assert_eq!(
            rb.overflow_count(),
            0,
            "Overflow counter should remain 0 during wrap-around (push 5)"
        );

        // Pop one item to make space for one more
        assert_eq!(rb.pop(), Some(3));
        assert_eq!(
            rb.overflow_count(),
            0,
            "Overflow counter should remain 0 after popping during wrap-around"
        );

        // Now we can push one more item (another wrap-around operation)
        assert!(rb.push(6));
        assert_eq!(
            rb.overflow_count(),
            0,
            "Overflow counter should remain 0 during second wrap-around (push 6)"
        );

        // Verify remaining order and final overflow counter state
        assert_eq!(rb.pop(), Some(4));
        assert_eq!(rb.pop(), Some(5));
        assert_eq!(rb.pop(), Some(6));
        assert!(rb.pop().is_none());

        // FINAL CRITICAL VERIFICATION: Overflow counter should still be 0
        // This confirms that wrap-around operations are NOT overflow conditions
        assert_eq!(
            rb.overflow_count(),
            0,
            "Overflow counter should remain 0 throughout normal wrap-around operations"
        );
    }

    #[test]
    fn test_ring_buffer_overflow() {
        let rb = RingBuffer::new(4);

        // Fill buffer completely (capacity - 1 items)
        assert!(rb.push(1));
        assert!(rb.push(2));
        assert!(rb.push(3));

        // This should fail (buffer full)
        assert!(!rb.push(4));
        assert_eq!(rb.overflow_count(), 1);
    }

    #[test]
    fn test_ring_buffer_concurrent() {
        let rb = Arc::new(RingBuffer::new(1024));
        let rb_clone = Arc::clone(&rb);

        // Producer thread
        let producer = thread::spawn(move || {
            for i in 0..1000 {
                while !rb_clone.push(i) {
                    // Buffer full, yield
                    thread::yield_now();
                }
            }
        });

        // Consumer thread
        let consumer = thread::spawn(move || {
            let mut items = Vec::new();
            while items.len() < 1000 {
                if let Some(item) = rb.pop() {
                    items.push(item);
                } else {
                    thread::yield_now();
                }
            }
            items
        });

        producer.join().unwrap();
        let items = consumer.join().unwrap();

        // Verify we got all items in order
        assert_eq!(items.len(), 1000);
        for (i, &item) in items.iter().enumerate() {
            assert_eq!(item, i);
        }
    }

    #[test]
    fn test_shared_ring_buffer() {
        let srb = SharedRingBuffer::new(8);
        let srb_clone = srb.clone();

        assert!(srb.push(42));
        assert_eq!(srb_clone.pop(), Some(42));
    }

    #[test]
    #[should_panic(expected = "Capacity must be a power of 2")]
    fn test_non_power_of_two_capacity() {
        let _rb = RingBuffer::<i32>::new(10);
    }
}