sklears_core/

streaming_lifetimes.rs

/// Explicit lifetime parameter examples for streaming operations
///
/// This module demonstrates advanced lifetime parameter usage in streaming
/// machine learning operations, providing clear lifetime relationships
/// and better compile-time safety.
use futures_core::{Future, Stream};
use std::pin::Pin;

// Type aliases for complex return types
type WindowedStream<'window, Output, E> = Pin<
    Box<
        dyn Stream<Item = std::result::Result<WindowedOutput<'window, Output>, E>> + Send + 'window,
    >,
>;

type ProcessingStream<'stream, Output, E> =
    Pin<Box<dyn Stream<Item = std::result::Result<Vec<Output>, E>> + Send + 'stream>>;

type ZeroCopyStream<'stream, Input, Output, E> = Pin<
    Box<
        dyn Stream<Item = std::result::Result<OutputView<'stream, Input, Output>, E>>
            + Send
            + 'stream,
    >,
>;

/// Streaming data processor with explicit lifetime relationships
///
/// This trait demonstrates advanced lifetime parameter patterns for streaming operations.
/// The lifetime parameters are explicitly documented to clarify their relationships:
///
/// - `'data`: Lifetime of the input data being processed
/// - `'config`: Lifetime of the configuration object
/// - `'processor`: Lifetime of the processor itself
pub trait StreamingProcessor<'data, 'config, 'processor, Input, Output>
where
    'data: 'processor,   // Data must live at least as long as the processor
    'config: 'processor, // Config must live at least as long as the processor
    Input: 'data,
{
    type Error: std::error::Error + Send + Sync + 'static;

    /// Process a single item with explicit lifetime management
    ///
    /// The lifetime parameters ensure that:
    /// - Input data lives for the duration of processing
    /// - Config is accessible throughout the operation
    /// - The processor maintains its state correctly
    fn process_item<'item>(
        &'processor self,
        item: &'item Input,
        config: &'config ProcessingConfig,
    ) -> Pin<Box<dyn Future<Output = std::result::Result<Output, Self::Error>> + Send + 'item>>
    where
        'processor: 'item, // Processor must outlive the item processing
        'config: 'item,    // Config must outlive the item processing
        'data: 'item,      // Data lifetime must cover item processing
        Input: 'item,      // Input must be valid for item processing
        Output: 'item; // Output is tied to item processing lifetime

    /// Process a stream of items with batching and explicit lifetime control
    ///
    /// This method shows how to handle complex lifetime relationships in streaming:
    /// - The input stream has its own lifetime ('stream)
    /// - The output stream is tied to multiple lifetimes
    /// - Memory management is explicit and safe
    fn process_stream<'stream>(
        &'processor self,
        input_stream: Pin<Box<dyn Stream<Item = Input> + Send + 'stream>>,
        config: &'config ProcessingConfig,
        batch_size: usize,
    ) -> ProcessingStream<'stream, Output, Self::Error>
    where
        'processor: 'stream, // Processor must outlive stream processing
        'config: 'stream,    // Config must outlive stream processing
        'data: 'stream,      // Data lifetime must cover stream processing
        Input: 'stream,      // Input items must be valid for stream duration
        Output: 'stream; // Output items are tied to stream lifetime

    /// Transform a stream with memory-efficient windowing
    ///
    /// Demonstrates lifetime management for windowed operations where:
    /// - Windows maintain references to multiple input items
    /// - Output lifetimes are carefully managed to prevent memory leaks
    /// - Window size affects memory lifetime requirements
    fn windowed_transform<'window>(
        &'processor self,
        input_stream: Pin<Box<dyn Stream<Item = &'window Input> + Send + 'window>>,
        window_size: usize,
        config: &'config ProcessingConfig,
    ) -> WindowedStream<'window, Output, Self::Error>
    where
        'processor: 'window, // Processor must outlive windowed operation
        'config: 'window,    // Config must outlive windowed operation
        'data: 'window,      // Data must outlive windowed operation
        Input: 'window,      // Input references must be valid for window duration
        Output: 'window; // Output is tied to window lifetime
}

/// Configuration for streaming processing operations
#[derive(Debug, Clone)]
pub struct ProcessingConfig {
    /// Buffer size for streaming operations
    pub buffer_size: usize,
    /// Maximum memory usage per batch (bytes)
    pub max_memory_per_batch: usize,
    /// Enable memory-efficient processing
    pub memory_efficient: bool,
    /// Timeout for individual operations
    pub operation_timeout: std::time::Duration,
}

/// Windowed output that maintains lifetime relationships
///
/// This struct demonstrates how to maintain safe lifetime relationships
/// in windowed streaming operations where the output references
/// input data across multiple time steps.
#[derive(Debug)]
pub struct WindowedOutput<'window, T> {
    /// The computed output value
    pub value: T,
    /// Window metadata with lifetime tied to input
    pub window_info: WindowInfo<'window>,
    /// Processing timestamp
    pub timestamp: std::time::Instant,
}

/// Window information with explicit lifetime management
#[derive(Debug)]
pub struct WindowInfo<'window> {
    /// Window size used for computation
    pub size: usize,
    /// References to the first and last items in the window
    /// This demonstrates safe lifetime management for references
    pub window_bounds: Option<WindowBounds<'window>>,
    /// Window statistics
    pub stats: WindowStats,
}

/// Window bounds with lifetime-managed references
#[derive(Debug)]
pub struct WindowBounds<'window> {
    /// Reference to the first item in the window
    pub first: Option<&'window dyn std::fmt::Debug>,
    /// Reference to the last item in the window
    pub last: Option<&'window dyn std::fmt::Debug>,
}

/// Statistics computed over the window
#[derive(Debug, Clone)]
pub struct WindowStats {
    /// Number of items processed
    pub count: usize,
    /// Processing duration
    pub duration: std::time::Duration,
    /// Memory usage during processing
    pub memory_used: usize,
}

/// Advanced streaming ML pipeline with explicit lifetime management
///
/// This trait demonstrates complex lifetime relationships in ML pipelines
/// where multiple processing stages have different lifetime requirements.
pub trait StreamingMLPipeline<'data, 'model, Input, Output>
where
    'data: 'model, // Data must outlive the model
    Input: 'data,
{
    type Model: 'model;
    type Error: std::error::Error + Send + Sync + 'static;
    type IntermediateOutput: 'model;

    /// Multi-stage processing with explicit lifetime control
    ///
    /// This method shows how to chain multiple processing stages
    /// while maintaining safe lifetime relationships throughout.
    fn process_pipeline<'pipeline>(
        &'model self,
        input_stream: Pin<Box<dyn Stream<Item = Input> + Send + 'pipeline>>,
        stages: &'pipeline [PipelineStage<'model, Self::Model>],
    ) -> Pin<Box<dyn Stream<Item = std::result::Result<Output, Self::Error>> + Send + 'pipeline>>
    where
        'model: 'pipeline, // Model must outlive pipeline execution
        'data: 'pipeline,  // Data must outlive pipeline execution
        Input: 'pipeline,  // Input must be valid for pipeline duration
        Output: 'pipeline, // Output is tied to pipeline lifetime
        Self::IntermediateOutput: 'pipeline;

    /// Parallel processing with lifetime-aware work distribution
    ///
    /// Demonstrates how to safely distribute work across multiple threads
    /// while maintaining lifetime guarantees for shared data.
    fn process_parallel<'parallel>(
        &'model self,
        input_stream: Pin<Box<dyn Stream<Item = Input> + Send + 'parallel>>,
        worker_count: usize,
        config: &'parallel ProcessingConfig,
    ) -> Pin<Box<dyn Stream<Item = std::result::Result<Output, Self::Error>> + Send + 'parallel>>
    where
        'model: 'parallel,        // Model must outlive parallel processing
        'data: 'parallel,         // Data must outlive parallel processing
        Input: 'parallel + Clone, // Input must be cloneable for distribution
        Output: 'parallel,        // Output is tied to parallel processing lifetime
        Self::Model: Sync,        // Model must be thread-safe for parallel access
        Self: Sync; // Pipeline must be thread-safe

    /// Adaptive processing with dynamic lifetime management
    ///
    /// Shows how to handle dynamic lifetime requirements where
    /// processing parameters may change based on input characteristics.
    fn process_adaptive<'adaptive>(
        &'model self,
        input_stream: Pin<Box<dyn Stream<Item = Input> + Send + 'adaptive>>,
        adaptation_fn: &'adaptive dyn Fn(&Input) -> AdaptationParams,
    ) -> Pin<Box<dyn Stream<Item = std::result::Result<Output, Self::Error>> + Send + 'adaptive>>
    where
        'model: 'adaptive, // Model must outlive adaptive processing
        'data: 'adaptive,  // Data must outlive adaptive processing
        Input: 'adaptive,  // Input must be valid for adaptive processing
        Output: 'adaptive; // Output is tied to adaptive processing lifetime
}

/// Pipeline stage with model lifetime management
#[derive(Debug)]
pub struct PipelineStage<'model, Model> {
    /// Name of the processing stage
    pub name: &'static str,
    /// Model used in this stage (with lifetime bound to pipeline)
    pub model: &'model Model,
    /// Stage-specific configuration
    pub config: StageConfig,
}

/// Configuration for individual pipeline stages
#[derive(Debug, Clone)]
pub struct StageConfig {
    /// Enable this stage
    pub enabled: bool,
    /// Stage-specific parameters
    pub parameters: std::collections::HashMap<String, f64>,
    /// Memory limit for this stage
    pub memory_limit: Option<usize>,
}

/// Parameters for adaptive processing
#[derive(Debug, Clone)]
pub struct AdaptationParams {
    /// Batch size adaptation
    pub batch_size: usize,
    /// Memory allocation hint
    pub memory_hint: usize,
    /// Processing priority
    pub priority: ProcessingPriority,
}

/// Processing priority levels
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
pub enum ProcessingPriority {
    Low,
    Normal,
    High,
    Critical,
}

/// Zero-copy streaming operations with lifetime guarantees
///
/// This trait demonstrates zero-copy streaming patterns with explicit
/// lifetime management to ensure memory safety without performance overhead.
pub trait ZeroCopyStreaming<'data, Input, Output>
where
    Input: 'data,
    Output: 'data,
{
    type Error: std::error::Error + Send + Sync + 'static;

    /// Process data without copying, maintaining lifetime relationships
    ///
    /// This method shows how to process data in-place while ensuring
    /// that all lifetime relationships are maintained correctly.
    fn process_zero_copy<'processing>(
        &self,
        data: &'processing mut [Input],
        output_buffer: &'processing mut [Output],
    ) -> Pin<Box<dyn Future<Output = std::result::Result<usize, Self::Error>> + Send + 'processing>>
    where
        'data: 'processing, // Data lifetime must cover processing
        Input: 'processing, // Input must be valid for processing duration
        Output: 'processing; // Output buffer must be valid for processing

    /// Stream processing with zero-copy views
    ///
    /// Demonstrates streaming with zero-copy views where the output
    /// contains references to the input data rather than copies.
    fn stream_zero_copy_views<'stream>(
        &self,
        input_stream: Pin<Box<dyn Stream<Item = &'stream [Input]> + Send + 'stream>>,
    ) -> ZeroCopyStream<'stream, Input, Output, Self::Error>
    where
        'data: 'stream, // Data must outlive streaming
        Input: 'stream, // Input slices must be valid for stream duration
        Output: 'stream; // Output views are tied to stream lifetime
}

/// Zero-copy output view with lifetime management
///
/// This struct demonstrates how to create zero-copy views of processed data
/// while maintaining safe lifetime relationships between input and output.
#[derive(Debug)]
pub struct OutputView<'view, Input, Output> {
    /// Processed output data
    pub output: Output,
    /// Optional reference to original input (zero-copy)
    pub input_ref: Option<&'view Input>,
    /// Processing metadata
    pub metadata: ViewMetadata,
}

/// Metadata for zero-copy views
#[derive(Debug, Clone)]
pub struct ViewMetadata {
    /// Processing time
    pub processing_time: std::time::Duration,
    /// Memory overhead (should be minimal for zero-copy)
    pub memory_overhead: usize,
    /// Whether the operation was truly zero-copy
    pub zero_copy: bool,
}

impl Default for ProcessingConfig {
    fn default() -> Self {
        Self {
            buffer_size: 8192,
            max_memory_per_batch: 1024 * 1024, // 1MB
            memory_efficient: true,
            operation_timeout: std::time::Duration::from_secs(30),
        }
    }
}

impl Default for StageConfig {
    fn default() -> Self {
        Self {
            enabled: true,
            parameters: std::collections::HashMap::new(),
            memory_limit: None,
        }
    }
}

impl Default for AdaptationParams {
    fn default() -> Self {
        Self {
            batch_size: 1000,
            memory_hint: 1024 * 1024, // 1MB
            priority: ProcessingPriority::Normal,
        }
    }
}

/// Lifetime documentation utilities
///
/// This module provides documentation helpers for understanding
/// lifetime relationships in complex streaming operations.
pub mod lifetime_docs {
    /// Documents the lifetime relationships in a streaming operation
    ///
    /// # Lifetime Parameters
    /// - `'data`: The lifetime of the source data being processed
    /// - `'processor`: The lifetime of the processing pipeline
    /// - `'config`: The lifetime of configuration objects
    /// - `'output`: The lifetime of the output data
    ///
    /// # Lifetime Relationships
    /// - `'data: 'processor` - Source data must outlive the processor
    /// - `'config: 'processor` - Configuration must outlive the processor
    /// - `'processor: 'output` - Processor must outlive output generation
    ///
    /// # Memory Safety Guarantees
    /// - No dangling references in streaming operations
    /// - Proper cleanup of resources when lifetimes end
    /// - Zero-copy operations maintain reference validity
    /// - Batched processing respects memory constraints
    pub fn document_streaming_lifetimes() {
        // This function serves as documentation for lifetime patterns
        // used throughout the streaming operations in this module.
    }

    /// Documents common lifetime anti-patterns to avoid
    ///
    /// # Anti-patterns
    /// - Returning references with insufficient lifetime bounds
    /// - Storing references without proper lifetime constraints
    /// - Using 'static where shorter lifetimes would suffice
    /// - Mixing owned and borrowed data without clear lifetime bounds
    pub fn document_lifetime_antipatterns() {
        // This function documents what NOT to do with lifetimes
        // in streaming machine learning operations.
    }
}

#[allow(non_snake_case)]
#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn test_processing_config_default() {
        let config = ProcessingConfig::default();
        assert_eq!(config.buffer_size, 8192);
        assert_eq!(config.max_memory_per_batch, 1024 * 1024);
        assert!(config.memory_efficient);
    }

    #[test]
    fn test_stage_config_default() {
        let config = StageConfig::default();
        assert!(config.enabled);
        assert!(config.parameters.is_empty());
        assert!(config.memory_limit.is_none());
    }

    #[test]
    fn test_adaptation_params_default() {
        let params = AdaptationParams::default();
        assert_eq!(params.batch_size, 1000);
        assert_eq!(params.memory_hint, 1024 * 1024);
        assert_eq!(params.priority, ProcessingPriority::Normal);
    }

    #[test]
    fn test_processing_priority_ordering() {
        assert!(ProcessingPriority::Critical > ProcessingPriority::High);
        assert!(ProcessingPriority::High > ProcessingPriority::Normal);
        assert!(ProcessingPriority::Normal > ProcessingPriority::Low);
    }
}