rusty_bin/monitor/

lockfree_stats.rs

//! Lock-free statistics collection for high-frequency monitoring
//!
//! This module provides lock-free atomic statistics collection optimized for
//! high-frequency trading environments where minimal latency is critical.

use parking_lot::RwLock;
use quanta::Instant as QuantaInstant;
use rusty_common::collections::FxHashMap;
use smartstring::alias::String as SmartString;
use std::sync::Arc;
use std::sync::atomic::{AtomicU64, Ordering};

/// Lock-free statistics collector using atomic operations
#[derive(Debug)]
pub struct LockFreeStatsCollector {
    // Core metrics using relaxed ordering for maximum performance
    events_processed: AtomicU64,
    trades_processed: AtomicU64,
    orderbooks_processed: AtomicU64,
    bytes_serialized: AtomicU64,
    simd_batches_processed: AtomicU64,
    buffer_reuse_count: AtomicU64,
    zero_copy_operations: AtomicU64,

    // Latency tracking with exponential moving average
    avg_latency_nanos: AtomicU64,

    // Error counters
    connection_errors: AtomicU64,
    parsing_errors: AtomicU64,
    serialization_errors: AtomicU64,

    // Per-exchange counters (using RwLock only for structure changes)
    exchange_stats: Arc<RwLock<FxHashMap<SmartString, ExchangeStatsAtomic>>>,

    // Start time for rate calculations
    start_time: QuantaInstant,
}

/// Atomic statistics for individual exchanges
#[derive(Debug)]
struct ExchangeStatsAtomic {
    messages_received: AtomicU64,
    bytes_received: AtomicU64,
    reconnections: AtomicU64,
    last_message_timestamp: AtomicU64,
    connection_duration_ms: AtomicU64,
}

impl ExchangeStatsAtomic {
    const fn new() -> Self {
        Self {
            messages_received: AtomicU64::new(0),
            bytes_received: AtomicU64::new(0),
            reconnections: AtomicU64::new(0),
            last_message_timestamp: AtomicU64::new(0),
            connection_duration_ms: AtomicU64::new(0),
        }
    }
}

/// Snapshot of current statistics (non-atomic for reading)
#[derive(Debug, Clone, Default)]
pub struct StatsSnapshot {
    /// Total number of events processed since startup
    pub events_processed: u64,
    /// Total number of trade messages processed
    pub trades_processed: u64,
    /// Total number of orderbook updates processed
    pub orderbooks_processed: u64,
    /// Total bytes serialized for outgoing messages
    pub bytes_serialized: u64,
    /// Number of SIMD batches processed for vectorized operations
    pub simd_batches_processed: u64,
    /// Number of buffer reuse operations to reduce memory allocations
    pub buffer_reuse_count: u64,
    /// Number of zero-copy operations performed for high-performance data handling
    pub zero_copy_operations: u64,
    /// Current average latency in nanoseconds using exponential moving average
    pub avg_latency_nanos: u64,
    /// Total number of connection errors encountered
    pub connection_errors: u64,
    /// Total number of parsing errors from malformed messages
    pub parsing_errors: u64,
    /// Total number of serialization errors during message encoding
    pub serialization_errors: u64,
    /// Current rate of events processed per second
    pub events_per_second: f64,
    /// Current rate of bytes processed per second
    pub bytes_per_second: f64,
    /// Compression ratio achieved for data transmission (0.0 if not calculated)
    pub compression_ratio: f64,
    /// Total uptime in seconds since statistics collection started
    pub uptime_seconds: f64,
    /// Per-exchange statistics breakdown
    pub exchange_stats: FxHashMap<SmartString, ExchangeStats>,
}

/// Non-atomic exchange statistics for snapshots
#[derive(Debug, Clone, Default)]
pub struct ExchangeStats {
    /// Total number of messages received from this exchange
    pub messages_received: u64,
    /// Total bytes received from this exchange
    pub bytes_received: u64,
    /// Number of reconnections to this exchange
    pub reconnections: u64,
    /// Timestamp of the last message received (nanoseconds since start)
    pub last_message_timestamp: u64,
    /// Total connection duration in milliseconds
    pub connection_duration_ms: u64,
    /// Current rate of messages received per second from this exchange
    pub messages_per_second: f64,
}

impl LockFreeStatsCollector {
    /// Create a new lock-free statistics collector
    pub fn new() -> Self {
        Self {
            events_processed: AtomicU64::new(0),
            trades_processed: AtomicU64::new(0),
            orderbooks_processed: AtomicU64::new(0),
            bytes_serialized: AtomicU64::new(0),
            simd_batches_processed: AtomicU64::new(0),
            buffer_reuse_count: AtomicU64::new(0),
            zero_copy_operations: AtomicU64::new(0),
            avg_latency_nanos: AtomicU64::new(0),
            connection_errors: AtomicU64::new(0),
            parsing_errors: AtomicU64::new(0),
            serialization_errors: AtomicU64::new(0),
            exchange_stats: Arc::new(RwLock::new(FxHashMap::default())),
            start_time: QuantaInstant::now(),
        }
    }

    /// Record a processed event (lock-free, relaxed ordering for maximum performance)
    #[inline(always)]
    pub fn record_event(&self) {
        self.events_processed.fetch_add(1, Ordering::Relaxed);
    }

    /// Record processed trades in batch (lock-free)
    #[inline(always)]
    pub fn record_trades(&self, count: u64) {
        self.trades_processed.fetch_add(count, Ordering::Relaxed);
    }

    /// Record processed orderbooks in batch (lock-free)
    #[inline(always)]
    pub fn record_orderbooks(&self, count: u64) {
        self.orderbooks_processed
            .fetch_add(count, Ordering::Relaxed);
    }

    /// Record serialized bytes (lock-free)
    #[inline(always)]
    pub fn record_bytes_serialized(&self, bytes: u64) {
        self.bytes_serialized.fetch_add(bytes, Ordering::Relaxed);
    }

    /// Record SIMD batch processing (lock-free)
    #[inline(always)]
    pub fn record_simd_batch(&self) {
        self.simd_batches_processed.fetch_add(1, Ordering::Relaxed);
    }

    /// Record buffer reuse (lock-free)
    #[inline(always)]
    pub fn record_buffer_reuse(&self) {
        self.buffer_reuse_count.fetch_add(1, Ordering::Relaxed);
    }

    /// Record zero-copy operations (lock-free)
    #[inline(always)]
    pub fn record_zero_copy_ops(&self, count: u64) {
        self.zero_copy_operations
            .fetch_add(count, Ordering::Relaxed);
    }

    /// Update average latency using exponential moving average (lock-free)
    #[inline(always)]
    pub fn update_latency(&self, latency_nanos: u64) {
        // Use exponential moving average: new_avg = 0.9 * old_avg + 0.1 * new_value
        // Implemented with atomic operations to avoid locks
        let current_avg = self.avg_latency_nanos.load(Ordering::Relaxed);
        let new_avg = if current_avg == 0 {
            latency_nanos
        } else {
            (current_avg * 9 + latency_nanos) / 10
        };

        // Use compare_exchange_weak for better performance on contended systems
        let _ = self.avg_latency_nanos.compare_exchange_weak(
            current_avg,
            new_avg,
            Ordering::Relaxed,
            Ordering::Relaxed,
        );
    }

    /// Record connection error (lock-free)
    #[inline(always)]
    pub fn record_connection_error(&self) {
        self.connection_errors.fetch_add(1, Ordering::Relaxed);
    }

    /// Record parsing error (lock-free)
    #[inline(always)]
    pub fn record_parsing_error(&self) {
        self.parsing_errors.fetch_add(1, Ordering::Relaxed);
    }

    /// Record serialization error (lock-free)
    #[inline(always)]
    pub fn record_serialization_error(&self) {
        self.serialization_errors.fetch_add(1, Ordering::Relaxed);
    }

    /// Record exchange message (mostly lock-free, minimal lock contention)
    #[inline(always)]
    pub fn record_exchange_message(&self, exchange: &str, bytes: u64) {
        let exchange_key: SmartString = exchange.into();

        // Fast path: check if exchange stats already exist
        {
            let stats_read = self.exchange_stats.read();
            if let Some(exchange_stats) = stats_read.get(&exchange_key) {
                exchange_stats
                    .messages_received
                    .fetch_add(1, Ordering::Relaxed);
                exchange_stats
                    .bytes_received
                    .fetch_add(bytes, Ordering::Relaxed);
                exchange_stats.last_message_timestamp.store(
                    QuantaInstant::now().elapsed().as_nanos() as u64,
                    Ordering::Relaxed,
                );
                return;
            }
        }

        // Slow path: need to create new exchange stats (rare)
        let mut stats_write = self.exchange_stats.write();
        let exchange_stats = stats_write
            .entry(exchange_key)
            .or_insert_with(ExchangeStatsAtomic::new);

        exchange_stats
            .messages_received
            .fetch_add(1, Ordering::Relaxed);
        exchange_stats
            .bytes_received
            .fetch_add(bytes, Ordering::Relaxed);
        exchange_stats.last_message_timestamp.store(
            QuantaInstant::now().elapsed().as_nanos() as u64,
            Ordering::Relaxed,
        );
    }

    /// Record exchange reconnection (mostly lock-free)
    #[inline(always)]
    pub fn record_exchange_reconnection(&self, exchange: &str) {
        let exchange_key: SmartString = exchange.into();

        // Fast path: check if exchange stats already exist
        {
            let stats_read = self.exchange_stats.read();
            if let Some(exchange_stats) = stats_read.get(&exchange_key) {
                exchange_stats.reconnections.fetch_add(1, Ordering::Relaxed);
                return;
            }
        }

        // Slow path: create new exchange stats
        let mut stats_write = self.exchange_stats.write();
        let exchange_stats = stats_write
            .entry(exchange_key)
            .or_insert_with(ExchangeStatsAtomic::new);
        exchange_stats.reconnections.fetch_add(1, Ordering::Relaxed);
    }

    /// Get a snapshot of current statistics (mostly lock-free reads)
    pub fn get_snapshot(&self) -> StatsSnapshot {
        let uptime_seconds = self.start_time.elapsed().as_secs_f64();

        // All atomic reads with relaxed ordering for maximum performance
        let events_processed = self.events_processed.load(Ordering::Relaxed);
        let bytes_serialized = self.bytes_serialized.load(Ordering::Relaxed);

        // Calculate rates
        let events_per_second = if uptime_seconds > 0.0 {
            events_processed as f64 / uptime_seconds
        } else {
            0.0
        };

        let bytes_per_second = if uptime_seconds > 0.0 {
            bytes_serialized as f64 / uptime_seconds
        } else {
            0.0
        };

        // Read exchange stats
        let mut exchange_stats = FxHashMap::default();
        {
            let stats_read = self.exchange_stats.read();
            for (exchange, atomic_stats) in stats_read.iter() {
                let messages_received = atomic_stats.messages_received.load(Ordering::Relaxed);
                let messages_per_second = if uptime_seconds > 0.0 {
                    messages_received as f64 / uptime_seconds
                } else {
                    0.0
                };

                exchange_stats.insert(
                    exchange.clone(),
                    ExchangeStats {
                        messages_received,
                        bytes_received: atomic_stats.bytes_received.load(Ordering::Relaxed),
                        reconnections: atomic_stats.reconnections.load(Ordering::Relaxed),
                        last_message_timestamp: atomic_stats
                            .last_message_timestamp
                            .load(Ordering::Relaxed),
                        connection_duration_ms: atomic_stats
                            .connection_duration_ms
                            .load(Ordering::Relaxed),
                        messages_per_second,
                    },
                );
            }
        }

        StatsSnapshot {
            events_processed,
            trades_processed: self.trades_processed.load(Ordering::Relaxed),
            orderbooks_processed: self.orderbooks_processed.load(Ordering::Relaxed),
            bytes_serialized,
            simd_batches_processed: self.simd_batches_processed.load(Ordering::Relaxed),
            buffer_reuse_count: self.buffer_reuse_count.load(Ordering::Relaxed),
            zero_copy_operations: self.zero_copy_operations.load(Ordering::Relaxed),
            avg_latency_nanos: self.avg_latency_nanos.load(Ordering::Relaxed),
            connection_errors: self.connection_errors.load(Ordering::Relaxed),
            parsing_errors: self.parsing_errors.load(Ordering::Relaxed),
            serialization_errors: self.serialization_errors.load(Ordering::Relaxed),
            events_per_second,
            bytes_per_second,
            compression_ratio: 0.0, // Calculated separately if needed
            uptime_seconds,
            exchange_stats,
        }
    }

    /// Reset all statistics (used for benchmarking)
    pub fn reset(&self) {
        self.events_processed.store(0, Ordering::Relaxed);
        self.trades_processed.store(0, Ordering::Relaxed);
        self.orderbooks_processed.store(0, Ordering::Relaxed);
        self.bytes_serialized.store(0, Ordering::Relaxed);
        self.simd_batches_processed.store(0, Ordering::Relaxed);
        self.buffer_reuse_count.store(0, Ordering::Relaxed);
        self.zero_copy_operations.store(0, Ordering::Relaxed);
        self.avg_latency_nanos.store(0, Ordering::Relaxed);
        self.connection_errors.store(0, Ordering::Relaxed);
        self.parsing_errors.store(0, Ordering::Relaxed);
        self.serialization_errors.store(0, Ordering::Relaxed);

        // Clear exchange stats
        self.exchange_stats.write().clear();
    }

    /// Get current events per second rate (lock-free)
    pub fn get_events_per_second(&self) -> f64 {
        let uptime = self.start_time.elapsed().as_secs_f64();
        if uptime > 0.0 {
            self.events_processed.load(Ordering::Relaxed) as f64 / uptime
        } else {
            0.0
        }
    }

    /// Get current average latency in nanoseconds (lock-free)
    #[inline(always)]
    pub fn get_avg_latency_nanos(&self) -> u64 {
        self.avg_latency_nanos.load(Ordering::Relaxed)
    }
}

impl Default for LockFreeStatsCollector {
    fn default() -> Self {
        Self::new()
    }
}

// Thread-safe implementation
unsafe impl Send for LockFreeStatsCollector {}
unsafe impl Sync for LockFreeStatsCollector {}

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

    #[test]
    fn test_lockfree_stats_basic_operations() {
        let stats = LockFreeStatsCollector::new();

        // Test basic increments
        stats.record_event();
        stats.record_trades(5);
        stats.record_orderbooks(3);
        stats.record_bytes_serialized(1024);

        let snapshot = stats.get_snapshot();
        assert_eq!(snapshot.events_processed, 1);
        assert_eq!(snapshot.trades_processed, 5);
        assert_eq!(snapshot.orderbooks_processed, 3);
        assert_eq!(snapshot.bytes_serialized, 1024);
    }

    #[test]
    fn test_lockfree_stats_latency_tracking() {
        let stats = LockFreeStatsCollector::new();

        // Test latency updates
        stats.update_latency(1000);
        assert_eq!(stats.get_avg_latency_nanos(), 1000);

        stats.update_latency(2000);
        // Should be exponential moving average: (1000 * 9 + 2000) / 10 = 1100
        assert_eq!(stats.get_avg_latency_nanos(), 1100);
    }

    #[test]
    fn test_lockfree_stats_exchange_tracking() {
        let stats = LockFreeStatsCollector::new();

        // Test exchange message recording
        stats.record_exchange_message("binance", 512);
        stats.record_exchange_message("binance", 256);
        stats.record_exchange_message("coinbase", 1024);

        let snapshot = stats.get_snapshot();
        assert_eq!(snapshot.exchange_stats.len(), 2);

        let binance_stats = snapshot.exchange_stats.get("binance").unwrap();
        assert_eq!(binance_stats.messages_received, 2);
        assert_eq!(binance_stats.bytes_received, 768);

        let coinbase_stats = snapshot.exchange_stats.get("coinbase").unwrap();
        assert_eq!(coinbase_stats.messages_received, 1);
        assert_eq!(coinbase_stats.bytes_received, 1024);
    }

    #[test]
    fn test_lockfree_stats_concurrent_access() {
        let stats = Arc::new(LockFreeStatsCollector::new());
        let mut handles = Vec::new();

        // Spawn multiple threads to test concurrent access
        for i in 0..10 {
            let stats_clone = Arc::clone(&stats);
            let handle = thread::spawn(move || {
                for _ in 0..1000 {
                    stats_clone.record_event();
                    stats_clone.record_trades(1);
                    stats_clone.record_exchange_message("test", 100);
                    stats_clone.update_latency(i * 100);
                }
            });
            handles.push(handle);
        }

        // Wait for all threads to complete
        for handle in handles {
            handle.join().unwrap();
        }

        let snapshot = stats.get_snapshot();
        assert_eq!(snapshot.events_processed, 10000);
        assert_eq!(snapshot.trades_processed, 10000);

        let test_stats = snapshot.exchange_stats.get("test").unwrap();
        assert_eq!(test_stats.messages_received, 10000);
        assert_eq!(test_stats.bytes_received, 1000000);
    }

    #[test]
    fn test_lockfree_stats_reset() {
        let stats = LockFreeStatsCollector::new();

        // Add some data
        stats.record_event();
        stats.record_trades(5);
        stats.record_exchange_message("test", 1024);

        // Reset and verify
        stats.reset();
        let snapshot = stats.get_snapshot();
        assert_eq!(snapshot.events_processed, 0);
        assert_eq!(snapshot.trades_processed, 0);
        assert_eq!(snapshot.exchange_stats.len(), 0);
    }

    #[test]
    fn test_lockfree_stats_rates_calculation() {
        let stats = LockFreeStatsCollector::new();

        // Record some events
        for _ in 0..100 {
            stats.record_event();
        }
        for _ in 0..50 {
            stats.record_bytes_serialized(1024);
        }

        // Wait a bit to get meaningful rates
        thread::sleep(Duration::from_millis(100));

        let snapshot = stats.get_snapshot();
        assert!(snapshot.events_per_second > 0.0);
        assert!(snapshot.bytes_per_second > 0.0);
        assert!(snapshot.uptime_seconds > 0.0);
    }
}