tenshift_core/

lib.rs

#![warn(missing_docs)]
#![warn(clippy::pedantic)]
#![allow(
    clippy::module_name_repetitions,
    clippy::must_use_candidate,
    clippy::missing_errors_doc,
)]
#![forbid(unsafe_code)]
//! `tenshift-core` builds thread-safe, backpressure-aware data loading pipelines
//! for iterative processing workloads such as training loops, evaluation passes,
//! and large offline transforms.
//!
//! A pipeline connects a [`source::Source`] to parallel worker stages and a
//! collector stage with bounded channels between them. Those bounded channels
//! are the backpressure mechanism: when the consumer slows down, upstream
//! stages stop overproducing instead of growing memory without bound.
//!
//! # Quick Start
//!
//! ```rust
//! use tenshift_core::sample::{Sample, Tensor};
//! use tenshift_core::sources::MemorySource;
//! use tenshift_core::Pipeline;
//!
//! let samples: Vec<Sample> = (0..100)
//!     .map(|i| {
//!         Sample::new()
//!             .with("x", Tensor::f32(&vec![i as f32; 10], vec![10]))
//!             .with("y", Tensor::i64(&[(i % 10) as i64], vec![1]))
//!     })
//!     .collect();
//!
//! let mut pipeline = Pipeline::from_source(MemorySource::new("train", samples))
//!     .workers(4)
//!     .prefetch(8)
//!     .batch(32)
//!     .start()?;
//!
//! let mut batches = 0;
//! for batch in &mut pipeline {
//!     assert!(!batch.is_empty());
//!     batches += 1;
//! }
//!
//! assert_eq!(batches, 4);
//! # Ok::<(), tenshift_core::error::Error>(())
//! ```
//!
//! # Architecture
//!
//! ```text
//! Source thread ──▸ N worker threads ──▸ Collector thread ──▸ Consumer
//!   (I/O)           (parallel map/       (serial shuffle/     (caller)
//!                    filter)              batch, flush)
//! ```
//!
//! # Auto-Scaling Defaults
//!
//! The pipeline automatically configures itself based on available resources:
//!
//! ## Prefetch Buffer Size
//! **Default:** `num_cpus() * 2`, minimum 8
//!
//! This heuristic balances two competing concerns:
//! - **Worker starvation prevention:** If the prefetch buffer is too small, workers
//!   finish their assigned chunks and stall waiting for the collector to consume
//!   results. A larger buffer keeps all workers fed even when the consumer is slow.
//! - **Memory pressure:** Each slot in the prefetch buffer holds a full chunk of
//!   samples. On high-core-count machines (64+), a fixed small buffer would cause
//!   workers to sit idle.
//!
//! The `* 2` multiplier provides headroom for bursty throughput, while the `min 8`
//! floor ensures reasonable behavior on single-core or container-constrained
//! environments.
//!
//! ## Channel Chunk Size
//! **Default:** 64 samples per chunk
//!
//! Samples flow through channels in chunks rather than individually:
//! - **Overhead amortization:** Sending 64 samples across a channel costs nearly
//!   the same as sending 1, dramatically reducing synchronization overhead.
//! - **Cache locality:** Workers process chunks sequentially, improving cache hit
//!   rates for transform chains.
//! - **Backpressure granularity:** 64 provides a sweet spot - large enough to batch
//!   productively, small enough that bounded channels still provide meaningful
//!   backpressure to prevent OOM.
//!
//! # Shutdown Semantics
//!
//! Tenshift guarantees clean shutdown in all error scenarios:
//!
//! ## Producer Panic (Source Thread)
//! If the source thread panics (e.g., filesystem failure), the panic is caught
//! and converted to an error message. The `shutdown` atomic flag is set, signaling
//! all workers and the collector to drain and exit. No zombie threads remain.
//!
//! ## Consumer Panic (Iterator Drop)
//! If the consumer thread panics or drops the [`PipelineIterator`] without
//! exhausting it, the iterator's `Drop` implementation sets `shutdown` and clears
//! the receiver channel. This unblocks any producer threads blocked on full
//! channels, allowing graceful thread termination.
//!
//! ## Channel Disconnect
//! All channels are bounded. If any stage disconnects (e.g., worker thread death),
//! subsequent sends fail immediately. Workers detect send failures and exit their
//! loops. The collector detects receive failures and flushes remaining data before
//! terminating.
//!
//! ## Explicit Stop
//! Call [`PipelineIterator::stop()`] to signal shutdown without dropping the
//! iterator. This allows checking final [`PipelineStats`] after cancellation.
//!
//! # Extension Points
//!
//! The community extends tenshift through two traits:
//!
//! - **[`source::Source`]**  -  Add a new data origin (database, S3, video, etc.)
//! - **[`transform::Transform`]**  -  Add a new operation (augmentation, normalization, etc.)

#![cfg_attr(
    not(test),
    deny(
        clippy::unwrap_used,
        clippy::expect_used,
        clippy::todo,
        clippy::unimplemented,
        clippy::panic
    )
)]
// Builder methods all return Self  -  requiring #[must_use] on 40+ methods is noise.
// Trait name() methods returning &str are idiomatic even when the impl returns a literal.
// u64-to-usize casts: this crate targets 64-bit ML workloads, not 32-bit embedded.
#![allow(
    clippy::return_self_not_must_use,
    clippy::needless_pass_by_value,
    clippy::unnecessary_literal_bound
)]

/// Error types and the crate-wide [`error::Result`] alias.
pub mod error;
/// Pipeline builder, runtime configuration, and iterator types.
pub mod pipeline;
/// Sample and tensor data structures exchanged between pipeline stages.
pub mod sample;
/// Source traits for synchronous and asynchronous data origins.
pub mod source;
/// Built-in source implementations such as glob, memory, and JSONL readers.
pub mod sources;
/// Stateless and stateful transform traits plus built-in transforms.
pub mod transform;

/// Error handling behavior when a source item or transform fails.
pub use pipeline::ErrorPolicy;
/// Builder for a thread-safe, backpressure-aware data loading pipeline.
pub use pipeline::Pipeline;
/// Runtime configuration for pipeline execution.
pub use pipeline::PipelineConfig;
/// Running pipeline iterator that yields output batches.
pub use pipeline::PipelineIterator;
/// Observability snapshot for a running or completed pipeline.
pub use pipeline::PipelineStats;
#[cfg(feature = "csv")]
/// CSV-backed [`source::Source`] implementation.
pub use sources::CsvSource;
/// Source wrapper that shards items across distributed ranks.
pub use sources::DistributedSampler;
/// Glob-backed file source that emits one sample per matched file.
pub use sources::GlobSource;
#[cfg(feature = "uring")]
/// `io_uring`-accelerated file source.
pub use sources::RingSource;

/// Compile-checks the README quick-start example as a doctest.
#[cfg(doctest)]
#[doc = include_str!("../README.md")]
struct ReadmeExamples;