wireshift-core 0.1.1

#![allow(missing_docs)]

use std::thread;
use std::time::Duration;

use wireshift::ops::Nop;
use wireshift::{BackendPreference, BufferPool, CompletionKind, Ring, RingConfig};

fn fallback_ring() -> Ring {
    Ring::new(RingConfig::default().with_backend(BackendPreference::Fallback)).expect("ring")
}

#[test]
fn core_batch_handles_empty_iterators() {
    let ring = fallback_ring();
    let results = ring.batch(std::iter::empty::<Nop>()).expect("batch");
    assert!(results.is_empty());
}

#[test]
fn core_batch_auto_drains_when_more_ops_than_capacity() {
    let ring = Ring::new(
        RingConfig::default()
            .with_backend(BackendPreference::Fallback)
            .with_queue_depth(2),
    )
    .expect("ring");

    let results = ring.batch((0..10).map(|_| Nop)).expect("batch");
    assert_eq!(results.len(), 10);
    assert_eq!(ring.in_flight().expect("in_flight"), 0);
}

#[test]
fn core_chain_context_can_run_inner_batch() {
    let ring = fallback_ring();
    let completed = ring
        .chain(|chain| Ok(chain.batch((0..6).map(|_| Nop))?.len()))
        .expect("chain");

    assert_eq!(completed, 6);
}

#[test]
fn core_complete_times_out_when_nothing_is_submitted() {
    let ring = fallback_ring();
    let error = ring
        .complete(Some(Duration::from_millis(20)))
        .expect_err("empty ring must time out");
    assert!(error.to_string().contains("timed out"));
}

#[test]
fn core_buffer_release_returns_capacity_to_pool() {
    let pool = BufferPool::new(8, 1).expect("pool");
    let mut owned = pool.acquire().expect("acquire");
    owned.as_mut_slice()[..4].copy_from_slice(b"test");
    let completed = owned
        .set_filled_len(4)
        .expect("filled")
        .into_submitted()
        .into_completed(4)
        .expect("completed");
    let released = completed.release();

    assert_eq!(released.capacity(), 8);
    assert_eq!(released.filled_len(), 0);
}

#[test]
fn core_parallel_buffer_pool_stress_keeps_all_threads_progressing() {
    let pool = BufferPool::new(128, 8).expect("pool");
    let mut workers = Vec::new();
    for _ in 0..8 {
        let pool = pool.clone();
        workers.push(thread::spawn(move || {
            let mut count = 0;
            for _ in 0..1_000 {
                let buffer = pool.acquire().expect("acquire");
                drop(buffer);
                count += 1;
            }
            count
        }));
    }

    let mut total_count = 0;
    for worker in workers {
        total_count += worker.join().expect("join");
    }
    assert_eq!(
        total_count, 8000,
        "all workers should complete 1000 iterations each"
    );
}

#[test]
fn core_completion_event_reports_completed_kind() {
    let ring = fallback_ring();
    ring.submit(Nop).expect("submit");
    let event = ring.complete(Some(Duration::from_secs(1))).expect("event");
    assert_eq!(event.kind(), CompletionKind::Completed);
    assert_eq!(event.op_name(), "nop");
}