Crate tenshift_core 

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

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);

Architecture

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.)

Re-exports

pub use pipeline::ErrorPolicy;

pub use pipeline::Pipeline;

pub use pipeline::PipelineConfig;

pub use pipeline::PipelineIterator;

pub use pipeline::PipelineStats;

pub use sources::DistributedSampler;

pub use sources::GlobSource;

Modules

error

Error types and the crate-wide error::Result alias. Error types for tenshift.

pipeline

Pipeline builder, runtime configuration, and iterator types. Pipeline - the composable data loading engine.

sample

Sample and tensor data structures exchanged between pipeline stages. Sample - the unit of data that flows through the pipeline.

source

Source traits for synchronous and asynchronous data origins. Source trait - the extension point for data origins.

sources

Built-in source implementations such as glob, memory, and JSONL readers. Built-in sources for common data origins.

transform

Stateless and stateful transform traits plus built-in transforms. Transform - composable operations on samples.