# tenshift-core - Technical Spec
## Overview
`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.)
## Architecture
The crate is organized into the following public modules:
- `error`
- `pipeline`
- `sample`
- `source`
- `sources`
- `transform`
## Guarantees
- `#![forbid(unsafe_code)]` where applicable; see `src/lib.rs` for the exact lint preamble.
- All public types have doc comments.
- Error messages are actionable where applicable.
## Public API Summary
Key entry points are exported from `src/lib.rs` via `pub mod` and `pub use` re-exports.
Consult the module-level documentation in each source file for function signatures and usage examples.
## Error Handling
- `Error`