Struct StreamProcessor 

pub struct StreamProcessor<T> { /* private fields */ }

Parallel streaming processor for applying operations to chunks

Implementations

impl<T> StreamProcessor<T>

where T: Send + Sync + 'static,

pub fn new(config: StreamConfig) -> Self

Create a new stream processor

Examples found in repository

examples/datasets_streaming_demo.rs (line 304)

fn demonstrate_parallel_processing() -> Result<(), Box<dyn std::error::Error>> {
    println!("⚡ PARALLEL STREAM PROCESSING");
    println!("{}", "-".repeat(40));

    let config = StreamConfig {
        chunk_size: 1500,
        buffer_size: 4,
        num_workers: 4,
        max_chunks: Some(8),
        ..Default::default()
    };

    println!("Parallel processing configuration:");
    println!("  Workers: {}", config.num_workers);
    println!("  Chunk size: {}", config.chunk_size);
    println!("  Buffer size: {}", config.buffer_size);

    // Create a simple processor that computes statistics
    let _processor: StreamProcessor<DataChunk> = StreamProcessor::new(config.clone());

    // Define a processing function
    let compute_stats = |chunk: DataChunk| -> Result<
        HashMap<String, f64>,
        Box<dyn std::error::Error + Send + Sync>,
    > {
        let mut stats = HashMap::new();

        // Compute basic statistics
        let mean = chunk.data.mean_axis(scirs2_core::ndarray::Axis(0)).unwrap();
        let std = chunk.data.std_axis(scirs2_core::ndarray::Axis(0), 0.0);

        stats.insert("mean_feature_0".to_string(), mean[0]);
        stats.insert("std_feature_0".to_string(), std[0]);
        stats.insert("n_samples".to_string(), chunk.n_samples() as f64);
        stats.insert("chunk_index".to_string(), chunk.chunk_index as f64);

        // Simulate some computation time
        std::thread::sleep(std::time::Duration::from_millis(100));

        Ok(stats)
    };

    println!("\nProcessing stream with parallel workers...");
    let start_time = Instant::now();

    let stream = stream_classification(12_000, 10, 3, config)?;

    // For demonstration, we'll process chunks sequentially with timing
    // In a real implementation, you'd use the processor.process_parallel method
    let mut stream_iter = stream;
    let mut chunk_results = Vec::new();

    while let Some(chunk) = stream_iter.next_chunk()? {
        let chunk_start = Instant::now();
        let chunk_id = chunk.chunk_index;
        let chunk_samples = chunk.n_samples();

        // Process chunk
        let stats = compute_stats(chunk)
            .map_err(|e| -> Box<dyn std::error::Error> { Box::new(std::io::Error::other(e)) })?;
        let chunk_time = chunk_start.elapsed();

        println!(
            "  Chunk {}: {} samples, {:.2}ms",
            chunk_id + 1,
            chunk_samples,
            chunk_time.as_millis()
        );

        chunk_results.push(stats);

        if chunk_results.len() >= 8 {
            break;
        }
    }

    let total_time = start_time.elapsed();

    println!("\nParallel Processing Results:");
    println!("  Total chunks: {}", chunk_results.len());
    println!("  Total time: {:.2}s", total_time.as_secs_f64());
    println!(
        "  Average time per chunk: {:.2}ms",
        total_time.as_millis() as f64 / chunk_results.len() as f64
    );

    // Aggregate statistics
    let total_samples: f64 = chunk_results
        .iter()
        .map(|stats| stats.get("n_samples").unwrap_or(&0.0))
        .sum();

    println!("  Total samples processed: {total_samples}");
    println!(
        "  Throughput: {:.1} samples/s",
        total_samples / total_time.as_secs_f64()
    );

    println!();
    Ok(())
}

pub fn process_parallel<F, R>( &self, iterator: StreamingIterator, processor: F, ) -> Result<Vec<R>>

where F: Fn(DataChunk) -> Result<R> + Send + Sync + Clone + 'static, R: Send + 'static,

Process chunks in parallel using a custom function