rusty_bin/monitor/

zerocopy_pipeline.rs

//! Zero-copy optimized data pipeline for high-frequency monitoring
//!
//! This module provides zerocopy optimizations for the monitoring system to eliminate
//! allocations in critical paths and achieve sub-microsecond latency.

use crate::monitor::collector::{MarketDataEvent, Result};
use crate::monitor::schema::{OrderBookRecord, OrderBookSerializer, TradeRecord, TradeSerializer};
use crate::monitor::storage::{CompressionMode, manager::StorageManager};
use flume::{Receiver, Sender};
use parking_lot::RwLock;
use quanta::Instant as QuantaInstant;
use rust_decimal::prelude::ToPrimitive;
use simd_aligned::VecSimd;
use std::sync::Arc;
use tokio::task::JoinHandle;
use wide::{CmpGt, f64x4};

/// Zero-copy optimized pipeline configuration
#[derive(Debug, Clone)]
pub struct ZerocopyPipelineConfig {
    /// Pre-allocated buffer pool size
    pub buffer_pool_size: usize,
    /// Buffer size for each worker
    pub worker_buffer_size: usize,
    /// SIMD-aligned batch size for processing
    pub simd_batch_size: usize,
    /// Number of worker tasks for processing events
    pub worker_count: usize,
    /// Compression mode for storage
    pub compression_mode: CompressionMode,
    /// Pre-allocated serialization buffer size
    pub serialization_buffer_size: usize,
}

impl Default for ZerocopyPipelineConfig {
    fn default() -> Self {
        Self {
            buffer_pool_size: 64,
            worker_buffer_size: 64 * 1024, // 64KB per worker
            simd_batch_size: 32,           // Process 32 events at once using SIMD
            worker_count: 4,
            compression_mode: CompressionMode::Realtime,
            serialization_buffer_size: 128 * 1024, // 128KB serialization buffer
        }
    }
}

/// Zero-copy optimized pipeline statistics
#[derive(Debug, Clone, Default)]
pub struct ZerocopyPipelineStats {
    /// Total number of market data events processed through the pipeline
    pub events_processed: u64,
    /// Total number of trade events processed
    pub trades_processed: u64,
    /// Total number of orderbook updates processed
    pub orderbooks_processed: u64,
    /// Total bytes serialized by the pipeline
    pub bytes_serialized: u64,
    /// Average compression ratio achieved (compressed_size / original_size)
    pub compression_ratio: f64,
    /// Average processing latency in nanoseconds
    pub avg_latency_nanos: u64,
    /// Number of SIMD-optimized batch operations performed
    pub simd_batches_processed: u64,
    /// Number of times buffers were reused from the pool
    pub buffer_reuse_count: u64,
    /// Total number of zero-copy operations performed
    pub zero_copy_operations: u64,
}

/// Pre-allocated buffer pool for zero-copy operations
#[derive(Debug)]
struct ZerocopyBufferPool {
    serialization_buffers: Vec<Vec<u8>>,
    compression_buffers: Vec<Vec<u8>>,
    simd_buffers: Vec<VecSimd<f64x4>>,
    available_serialization: Vec<usize>,
    available_compression: Vec<usize>,
    available_simd: Vec<usize>,
}

impl ZerocopyBufferPool {
    fn new(config: &ZerocopyPipelineConfig) -> Self {
        let pool_size = config.buffer_pool_size;

        // Pre-allocate serialization buffers
        let mut serialization_buffers = Vec::with_capacity(pool_size);
        let mut available_serialization = Vec::with_capacity(pool_size);
        for i in 0..pool_size {
            serialization_buffers.push(Vec::with_capacity(config.serialization_buffer_size));
            available_serialization.push(i);
        }

        // Pre-allocate compression buffers
        let mut compression_buffers = Vec::with_capacity(pool_size);
        let mut available_compression = Vec::with_capacity(pool_size);
        for i in 0..pool_size {
            compression_buffers.push(Vec::with_capacity(config.worker_buffer_size));
            available_compression.push(i);
        }

        // Pre-allocate SIMD-aligned buffers
        let mut simd_buffers = Vec::with_capacity(pool_size);
        let mut available_simd = Vec::with_capacity(pool_size);
        for i in 0..pool_size {
            simd_buffers.push(VecSimd::<f64x4>::with(0.0, config.simd_batch_size / 4));
            available_simd.push(i);
        }

        Self {
            serialization_buffers,
            compression_buffers,
            simd_buffers,
            available_serialization,
            available_compression,
            available_simd,
        }
    }

    fn get_serialization_buffer(&mut self) -> Option<usize> {
        self.available_serialization.pop()
    }

    fn return_serialization_buffer(&mut self, index: usize) {
        // Clear buffer but keep capacity
        self.serialization_buffers[index].clear();
        self.available_serialization.push(index);
    }

    fn get_compression_buffer(&mut self) -> Option<usize> {
        self.available_compression.pop()
    }

    fn return_compression_buffer(&mut self, index: usize) {
        // Clear buffer but keep capacity
        self.compression_buffers[index].clear();
        self.available_compression.push(index);
    }

    fn get_simd_buffer(&mut self) -> Option<usize> {
        self.available_simd.pop()
    }

    fn return_simd_buffer(&mut self, index: usize) {
        // Reset SIMD buffer
        for i in 0..self.simd_buffers[index].len() {
            self.simd_buffers[index][i] = f64x4::splat(0.0);
        }
        self.available_simd.push(index);
    }
}

/// Worker context with pre-allocated buffers
#[derive(Debug)]
struct ZerocopyWorkerContext {
    serialization_buffer: Vec<u8>,
    compression_buffer: Vec<u8>,
    simd_prices: VecSimd<f64x4>,
    simd_quantities: VecSimd<f64x4>,
    processing_timestamps: Vec<u64>,
}

impl ZerocopyWorkerContext {
    fn new(config: &ZerocopyPipelineConfig) -> Self {
        Self {
            serialization_buffer: Vec::with_capacity(config.serialization_buffer_size),
            compression_buffer: Vec::with_capacity(config.worker_buffer_size),
            simd_prices: VecSimd::<f64x4>::with(0.0, config.simd_batch_size / 4),
            simd_quantities: VecSimd::<f64x4>::with(0.0, config.simd_batch_size / 4),
            processing_timestamps: Vec::with_capacity(config.simd_batch_size),
        }
    }

    fn reset(&mut self) {
        self.serialization_buffer.clear();
        self.compression_buffer.clear();
        self.processing_timestamps.clear();

        // Reset SIMD buffers to zero
        for i in 0..self.simd_prices.len() {
            self.simd_prices[i] = f64x4::splat(0.0);
            self.simd_quantities[i] = f64x4::splat(0.0);
        }
    }
}

/// Zero-copy optimized data pipeline
#[derive(Debug)]
pub struct ZerocopyDataPipeline {
    config: ZerocopyPipelineConfig,
    storage_manager: Arc<StorageManager>,
    buffer_pool: Arc<RwLock<ZerocopyBufferPool>>,
    worker_handles: Vec<JoinHandle<()>>,
    stats: Arc<RwLock<ZerocopyPipelineStats>>,
    shutdown_sender: Option<Sender<()>>,
}

impl ZerocopyDataPipeline {
    /// Create a new zero-copy optimized data pipeline
    #[must_use]
    pub fn new(config: ZerocopyPipelineConfig, storage_manager: Arc<StorageManager>) -> Self {
        let buffer_pool = Arc::new(RwLock::new(ZerocopyBufferPool::new(&config)));

        Self {
            config,
            storage_manager,
            buffer_pool,
            worker_handles: Vec::new(),
            stats: Arc::new(RwLock::new(ZerocopyPipelineStats::default())),
            shutdown_sender: None,
        }
    }

    /// Start the zero-copy pipeline processing
    pub async fn start(&mut self, event_receiver: Receiver<MarketDataEvent>) -> Result<()> {
        log::info!(
            "Starting zero-copy data pipeline with {} workers",
            self.config.worker_count
        );

        let (shutdown_sender, shutdown_receiver) = flume::unbounded();
        self.shutdown_sender = Some(shutdown_sender);

        // Start worker tasks with pre-allocated contexts
        for worker_id in 0..self.config.worker_count {
            let worker_handle = self
                .spawn_zerocopy_worker(worker_id, event_receiver.clone(), shutdown_receiver.clone())
                .await;
            self.worker_handles.push(worker_handle);
        }

        log::info!("Zero-copy data pipeline started successfully");
        Ok(())
    }

    /// Stop the zero-copy pipeline processing
    pub async fn stop(&mut self) -> Result<()> {
        log::info!("Stopping zero-copy data pipeline");

        // Send shutdown signal
        if let Some(shutdown_sender) = self.shutdown_sender.take() {
            let _ = shutdown_sender.send(());
        }

        // Wait for all workers to finish
        for handle in self.worker_handles.drain(..) {
            handle.abort();
            let _ = handle.await;
        }

        log::info!("Zero-copy data pipeline stopped");
        Ok(())
    }

    /// Get current zero-copy pipeline statistics
    pub fn get_stats(&self) -> ZerocopyPipelineStats {
        self.stats.read().clone()
    }

    /// Reset pipeline statistics
    pub fn reset_stats(&self) {
        *self.stats.write() = ZerocopyPipelineStats::default();
    }

    /// Spawn a zero-copy worker task
    async fn spawn_zerocopy_worker(
        &self,
        worker_id: usize,
        event_receiver: Receiver<MarketDataEvent>,
        shutdown_receiver: Receiver<()>,
    ) -> JoinHandle<()> {
        let storage_manager = self.storage_manager.clone();
        let buffer_pool = self.buffer_pool.clone();
        let stats = self.stats.clone();
        let config = self.config.clone();

        tokio::spawn(async move {
            log::info!("Starting zero-copy pipeline worker {worker_id}");

            let mut worker_context = ZerocopyWorkerContext::new(&config);
            let mut event_batch = Vec::with_capacity(config.simd_batch_size);

            loop {
                tokio::select! {
                    // Check for shutdown signal
                    _ = shutdown_receiver.recv_async() => {
                        log::info!("Zero-copy worker {worker_id} received shutdown signal");
                        break;
                    }

                    // Receive events and process in SIMD batches
                    event_result = event_receiver.recv_async() => {
                        match event_result {
                            Ok(event) => {
                                event_batch.push(event);

                                // Process batch when it reaches SIMD size
                                if event_batch.len() >= config.simd_batch_size {
                                    Self::process_zerocopy_batch(
                                        &event_batch,
                                        &mut worker_context,
                                        &storage_manager,
                                        &buffer_pool,
                                        &stats,
                                        &config,
                                    ).await;
                                    event_batch.clear();
                                }
                            }
                            Err(_) => {
                                log::info!("Zero-copy worker {worker_id} event channel closed");
                                break;
                            }
                        }
                    }
                }
            }

            // Process remaining events in batch
            if !event_batch.is_empty() {
                Self::process_zerocopy_batch(
                    &event_batch,
                    &mut worker_context,
                    &storage_manager,
                    &buffer_pool,
                    &stats,
                    &config,
                )
                .await;
            }

            log::info!("Zero-copy pipeline worker {worker_id} stopped");
        })
    }

    /// Process a batch of events using zero-copy patterns
    async fn process_zerocopy_batch(
        batch: &[MarketDataEvent],
        worker_context: &mut ZerocopyWorkerContext,
        storage_manager: &StorageManager,
        buffer_pool: &Arc<RwLock<ZerocopyBufferPool>>,
        stats: &Arc<RwLock<ZerocopyPipelineStats>>,
        config: &ZerocopyPipelineConfig,
    ) {
        let start_time = QuantaInstant::now();
        worker_context.reset();

        // Separate trades and orderbooks for batch processing
        let mut trades = Vec::new();
        let mut orderbooks = Vec::new();

        for event in batch {
            match event {
                MarketDataEvent::Trade(trade) => trades.push(trade),
                MarketDataEvent::OrderBook(orderbook) => orderbooks.push(orderbook),
                _ => {} // Skip status/error events
            }
        }

        let mut total_bytes_serialized = 0u64;
        let mut zero_copy_operations = 0u64;

        // Process trades using SIMD batching
        if !trades.is_empty()
            && let Ok(bytes) = Self::process_trade_batch_simd(
                &trades,
                worker_context,
                storage_manager,
                buffer_pool,
            )
            .await
        {
            total_bytes_serialized += bytes;
            zero_copy_operations += trades.len() as u64;
        }

        // Process orderbooks using SIMD batching
        if !orderbooks.is_empty()
            && let Ok(bytes) = Self::process_orderbook_batch_simd(
                &orderbooks,
                worker_context,
                storage_manager,
                buffer_pool,
            )
            .await
        {
            total_bytes_serialized += bytes;
            zero_copy_operations += orderbooks.len() as u64;
        }

        // Update statistics
        let processing_time_nanos = start_time.elapsed().as_nanos() as u64;
        {
            let mut stats_guard = stats.write();
            stats_guard.events_processed += batch.len() as u64;
            stats_guard.trades_processed += trades.len() as u64;
            stats_guard.orderbooks_processed += orderbooks.len() as u64;
            stats_guard.bytes_serialized += total_bytes_serialized;
            stats_guard.avg_latency_nanos =
                (stats_guard.avg_latency_nanos * 9 + processing_time_nanos) / 10;
            stats_guard.simd_batches_processed += 1;
            stats_guard.zero_copy_operations += zero_copy_operations;
            stats_guard.buffer_reuse_count += 1;
        }

        log::debug!(
            "Processed zero-copy batch of {} events in {}ns",
            batch.len(),
            processing_time_nanos
        );
    }

    /// Process trade batch using SIMD operations
    async fn process_trade_batch_simd(
        trades: &[&TradeRecord],
        worker_context: &mut ZerocopyWorkerContext,
        storage_manager: &StorageManager,
        _buffer_pool: &Arc<RwLock<ZerocopyBufferPool>>,
    ) -> Result<u64> {
        if trades.is_empty() {
            return Ok(0);
        }

        let mut total_bytes = 0u64;

        // Fill SIMD buffers with trade data for vectorized processing
        let simd_chunks = trades.len() / 4;
        let prices_flat = worker_context.simd_prices.flat_mut();
        let quantities_flat = worker_context.simd_quantities.flat_mut();

        for (i, trade) in trades.iter().enumerate().take(simd_chunks * 4) {
            prices_flat[i] = trade.price.to_f64().unwrap_or_else(|| {
                #[cfg(debug_assertions)]
                eprintln!(
                    "Warning: Trade price conversion failed at index {}: {}",
                    i, trade.price
                );
                f64::NAN
            });
            quantities_flat[i] = trade.quantity.to_f64().unwrap_or_else(|| {
                #[cfg(debug_assertions)]
                eprintln!(
                    "Warning: Trade quantity conversion failed at index {}: {}",
                    i, trade.quantity
                );
                f64::NAN
            });
            worker_context
                .processing_timestamps
                .push(trade.timestamp_exchange);
        }

        // Use SIMD operations for validation and processing
        if simd_chunks > 0 {
            for chunk in 0..simd_chunks {
                let price_simd = worker_context.simd_prices[chunk];
                let quantity_simd = worker_context.simd_quantities[chunk];

                // SIMD validation: check for valid prices and quantities
                let zero_vec = f64x4::splat(0.0);
                let price_valid = price_simd.cmp_gt(zero_vec);
                let quantity_valid = quantity_simd.cmp_gt(zero_vec);
                let all_valid = price_valid & quantity_valid;

                // Process valid trades in the SIMD chunk
                if all_valid.all() {
                    // All trades in this chunk are valid, process them
                    for i in 0..4 {
                        let trade_index = chunk * 4 + i;
                        if trade_index < trades.len() {
                            // Serialize using zero-copy patterns
                            worker_context.serialization_buffer.clear();
                            if let Ok(serialized) =
                                TradeSerializer::serialize_trade(trades[trade_index])
                                && (storage_manager.write_trade(trades[trade_index]).await).is_ok()
                            {
                                total_bytes += serialized.len() as u64;
                            }
                        }
                    }
                }
            }
        }

        // Process remaining trades that don't fit into SIMD chunks
        for trade in trades.iter().skip(simd_chunks * 4) {
            worker_context.serialization_buffer.clear();
            if let Ok(serialized) = TradeSerializer::serialize_trade(trade)
                && (storage_manager.write_trade(trade).await).is_ok()
            {
                total_bytes += serialized.len() as u64;
            }
        }

        Ok(total_bytes)
    }

    /// Process orderbook batch using SIMD operations
    async fn process_orderbook_batch_simd(
        orderbooks: &[&OrderBookRecord],
        worker_context: &mut ZerocopyWorkerContext,
        storage_manager: &StorageManager,
        _buffer_pool: &Arc<RwLock<ZerocopyBufferPool>>,
    ) -> Result<u64> {
        if orderbooks.is_empty() {
            return Ok(0);
        }

        let mut total_bytes = 0u64;

        // Process orderbooks with efficient serialization
        for orderbook in orderbooks {
            worker_context.serialization_buffer.clear();
            if let Ok(serialized) = OrderBookSerializer::serialize_orderbook(orderbook)
                && (storage_manager.write_orderbook(orderbook).await).is_ok()
            {
                total_bytes += serialized.len() as u64;
            }
        }

        Ok(total_bytes)
    }
}

impl Drop for ZerocopyDataPipeline {
    fn drop(&mut self) {
        // Send shutdown signal if still available
        if let Some(shutdown_sender) = self.shutdown_sender.take() {
            let _ = shutdown_sender.try_send(());
        }

        // Abort all worker handles
        for handle in &self.worker_handles {
            handle.abort();
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::monitor::config::storage::StorageConfig;

    use tempfile::TempDir;

    #[tokio::test]
    async fn test_zerocopy_pipeline_creation() {
        let temp_dir = TempDir::new().unwrap();
        let storage_config = StorageConfig {
            market_data_path: temp_dir.path().join("market_data"),
            ..Default::default()
        };
        let storage_manager = Arc::new(StorageManager::new(storage_config).await.unwrap());

        let config = ZerocopyPipelineConfig::default();
        let pipeline = ZerocopyDataPipeline::new(config, storage_manager);

        assert_eq!(pipeline.worker_handles.len(), 0);
        assert_eq!(pipeline.get_stats().events_processed, 0);
    }

    #[tokio::test]
    async fn test_zerocopy_buffer_pool() {
        let config = ZerocopyPipelineConfig::default();
        let mut pool = ZerocopyBufferPool::new(&config);

        // Test serialization buffer allocation and return
        let buffer_index = pool.get_serialization_buffer().unwrap();
        assert_eq!(
            pool.available_serialization.len(),
            config.buffer_pool_size - 1
        );

        pool.return_serialization_buffer(buffer_index);
        assert_eq!(pool.available_serialization.len(), config.buffer_pool_size);

        // Test SIMD buffer allocation
        let simd_index = pool.get_simd_buffer().unwrap();
        assert_eq!(pool.available_simd.len(), config.buffer_pool_size - 1);

        pool.return_simd_buffer(simd_index);
        assert_eq!(pool.available_simd.len(), config.buffer_pool_size);
    }

    #[test]
    fn test_worker_context_reset() {
        let config = ZerocopyPipelineConfig::default();
        let mut context = ZerocopyWorkerContext::new(&config);

        // Add some data
        context.serialization_buffer.extend_from_slice(b"test");
        context.processing_timestamps.push(123456789);

        // Reset should clear but maintain capacity
        let original_capacity = context.serialization_buffer.capacity();
        context.reset();

        assert_eq!(context.serialization_buffer.len(), 0);
        assert_eq!(context.serialization_buffer.capacity(), original_capacity);
        assert_eq!(context.processing_timestamps.len(), 0);
    }
}