Struct StreamStats 

pub struct StreamStats {
    pub current_chunk: usize,
    pub total_chunks: Option<usize>,
    pub buffer_size: usize,
    pub buffer_capacity: usize,
    pub finished: bool,
}

Statistics about streaming operation

Fields

current_chunk: usize

Current chunk being processed

total_chunks: Option<usize>

Total number of chunks (if known)

buffer_size: usize

Number of chunks currently buffered

buffer_capacity: usize

Maximum buffer capacity

finished: bool

Whether streaming is finished

Implementations

impl StreamStats

pub fn progress_percent(&self) -> Option<f64>

Get progress as a percentage (if total is known)

Examples found in repository

examples/datasets_streaming_demo.rs (line 89)

fn demonstrate_basic_streaming() -> Result<(), Box<dyn std::error::Error>> {
    println!("📊 BASIC STREAMING OPERATIONS");
    println!("{}", "-".repeat(40));

    // Configure streaming
    let config = StreamConfig {
        chunk_size: 1000,           // 1K samples per chunk
        buffer_size: 3,             // Buffer 3 chunks
        num_workers: 4,             // Use 4 worker threads
        memory_limit_mb: Some(100), // Limit to 100MB
        enable_compression: false,
        enable_prefetch: true,
        max_chunks: Some(10), // Process only 10 chunks for demo
    };

    println!("Streaming Configuration:");
    println!("  Chunk size: {} samples", config.chunk_size);
    println!("  Buffer size: {} chunks", config.buffer_size);
    println!("  Workers: {}", config.num_workers);
    println!("  Memory limit: {:?} MB", config.memory_limit_mb);
    println!("  Max chunks: {:?}", config.max_chunks);

    // Create streaming classification dataset
    println!("\nStreaming synthetic classification data...");
    let mut stream = stream_classification(100_000, 20, 5, config.clone())?;

    let mut total_samples = 0;
    let mut chunk_count = 0;
    let mut class_distribution: HashMap<i32, usize> = HashMap::new();

    let start_time = Instant::now();

    while let Some(chunk) = stream.next_chunk()? {
        total_samples += chunk.n_samples();
        chunk_count += 1;

        // Analyze this chunk
        if let Some(target) = &chunk.target {
            for &class in target.iter() {
                *class_distribution.entry(class as i32).or_insert(0) += 1;
            }
        }

        // Print progress
        let stats = stream.stats();
        if let Some(progress) = stats.progress_percent() {
            println!(
                "  Chunk {}: {} samples (Progress: {:.1}%, Buffer: {:.1}%)",
                chunk.chunk_index + 1,
                chunk.n_samples(),
                progress,
                stats.buffer_utilization()
            );
        } else {
            println!(
                "  Chunk {}: {} samples (Buffer: {:.1}%)",
                chunk.chunk_index + 1,
                chunk.n_samples(),
                stats.buffer_utilization()
            );
        }

        // Simulate processing time
        std::thread::sleep(std::time::Duration::from_millis(50));

        if chunk.is_last {
            println!("  📋 Reached last chunk");
            break;
        }
    }

    let duration = start_time.elapsed();

    println!("\nStreaming Results:");
    println!("  Total chunks processed: {chunk_count}");
    println!("  Total samples: {total_samples}");
    println!("  Processing time: {:.2}s", duration.as_secs_f64());
    println!(
        "  Throughput: {:.1} samples/s",
        total_samples as f64 / duration.as_secs_f64()
    );
    println!("  Class distribution: {class_distribution:?}");

    println!();
    Ok(())
}

pub fn buffer_utilization(&self) -> f64

Get buffer utilization as a percentage

Examples found in repository

examples/datasets_streaming_demo.rs (line 95)

fn demonstrate_basic_streaming() -> Result<(), Box<dyn std::error::Error>> {
    println!("📊 BASIC STREAMING OPERATIONS");
    println!("{}", "-".repeat(40));

    // Configure streaming
    let config = StreamConfig {
        chunk_size: 1000,           // 1K samples per chunk
        buffer_size: 3,             // Buffer 3 chunks
        num_workers: 4,             // Use 4 worker threads
        memory_limit_mb: Some(100), // Limit to 100MB
        enable_compression: false,
        enable_prefetch: true,
        max_chunks: Some(10), // Process only 10 chunks for demo
    };

    println!("Streaming Configuration:");
    println!("  Chunk size: {} samples", config.chunk_size);
    println!("  Buffer size: {} chunks", config.buffer_size);
    println!("  Workers: {}", config.num_workers);
    println!("  Memory limit: {:?} MB", config.memory_limit_mb);
    println!("  Max chunks: {:?}", config.max_chunks);

    // Create streaming classification dataset
    println!("\nStreaming synthetic classification data...");
    let mut stream = stream_classification(100_000, 20, 5, config.clone())?;

    let mut total_samples = 0;
    let mut chunk_count = 0;
    let mut class_distribution: HashMap<i32, usize> = HashMap::new();

    let start_time = Instant::now();

    while let Some(chunk) = stream.next_chunk()? {
        total_samples += chunk.n_samples();
        chunk_count += 1;

        // Analyze this chunk
        if let Some(target) = &chunk.target {
            for &class in target.iter() {
                *class_distribution.entry(class as i32).or_insert(0) += 1;
            }
        }

        // Print progress
        let stats = stream.stats();
        if let Some(progress) = stats.progress_percent() {
            println!(
                "  Chunk {}: {} samples (Progress: {:.1}%, Buffer: {:.1}%)",
                chunk.chunk_index + 1,
                chunk.n_samples(),
                progress,
                stats.buffer_utilization()
            );
        } else {
            println!(
                "  Chunk {}: {} samples (Buffer: {:.1}%)",
                chunk.chunk_index + 1,
                chunk.n_samples(),
                stats.buffer_utilization()
            );
        }

        // Simulate processing time
        std::thread::sleep(std::time::Duration::from_millis(50));

        if chunk.is_last {
            println!("  📋 Reached last chunk");
            break;
        }
    }

    let duration = start_time.elapsed();

    println!("\nStreaming Results:");
    println!("  Total chunks processed: {chunk_count}");
    println!("  Total samples: {total_samples}");
    println!("  Processing time: {:.2}s", duration.as_secs_f64());
    println!(
        "  Throughput: {:.1} samples/s",
        total_samples as f64 / duration.as_secs_f64()
    );
    println!("  Class distribution: {class_distribution:?}");

    println!();
    Ok(())
}

Trait Implementations

impl Clone for StreamStats

fn clone(&self) -> StreamStats

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl Debug for StreamStats

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.