tenshift-core

Thread-safe, backpressure-aware data loading pipeline for iterative processing. The core engine of the tenshift ML data loading ecosystem.

Installation

cargo add tenshift-core

Quick Example

use tenshift_core::{Pipeline, sample::{Sample, Tensor}, sources::MemorySource};

let samples: Vec<Sample> = (0..10_000)
    .map(|i| Sample::new()
        .with("x", Tensor::f32(&vec![i as f32; 784], vec![1, 28, 28]))
        .with("y", Tensor::i64(&[i as i64 % 10], vec![1])))
    .collect();

let mut iter = Pipeline::from_source(MemorySource::new("mnist", samples))
    .workers(4)
    .shuffle(1000)
    .batch(32)
    .prefetch(4)
    .start()?;

for batch in &mut iter {
    // Process batch of 32 samples
}
# Ok::<(), tenshift_core::error::Error>(())

Architecture Overview

Source thread ──▶ N worker threads ──▶ Collector thread ──▶ Consumer
   (I/O)          (parallel map)       (shuffle/batch)      (User)

Bounded channels provide backpressure
RAII cleanup on drop - no zombie processes
Zero-copy Arc tensor passing between threads
Optional io_uring acceleration via wireshift feature

Extension Guide

Add a Data Source

use tenshift_core::{source::{Source, SourceIterator}, error::Result, sample::Sample};

struct MySource;

impl Source for MySource {
    fn open(&self) -> Result<Box<dyn SourceIterator>> { unimplemented!("return your iterator here") }
    fn name(&self) -> &str { "my-source" }
}

Add a Transform

use tenshift_core::{transform::{Transform, TransformResult}, sample::Sample};

struct Normalize;

impl Transform for Normalize {
    fn apply(&self, sample: Sample) -> TransformResult {
        TransformResult::Sample(sample)
    }
    fn name(&self) -> &str { "normalize" }
}

Community contributions welcome for Parquet, HuggingFace Hub, S3/GCS sources.

tenshift-core 0.1.3