Struct BatchProcessor 

pub struct BatchProcessor<T: Send + 'static> { /* private fields */ }

Batch processor for inference requests

Implementations

impl<T: Send + 'static> BatchProcessor<T>

pub fn new(config: BatchConfig) -> Self

Create a new batch processor

pub async fn submit(&self, payload: T) -> Result<BatchResult<T>, BatchError>

Submit a request and wait for the result

pub async fn run<F, Fut>(&self, executor: F)

where F: FnMut(Vec<T>) -> Fut, Fut: Future<Output = Vec<T>>,

Run the dynamic-batching serving loop: wait for a batch, collect it, run executor on the payloads, and deliver each request’s response.

This is the piece that makes BatchProcessor::submit resolve — without a running run (or manual collect_batch/complete_batch), submitted futures never complete. It is the serving primitive an inference/expert worker spawns as a task; it loops until the task is cancelled.

executor receives the batched payloads in arrival order and must return exactly one response per input, in the same order. (If it returns fewer, the surplus requests are cancelled rather than answered.)

let p = processor.clone();
tokio::spawn(async move {
    p.run(|batch: Vec<Req>| async move { model.forward(batch).await }).await;
});
let result = processor.submit(req).await?; // resolved by the loop

pub fn queue_len(&self) -> usize

Get current queue length

pub fn stats(&self) -> BatchStats

Get batch statistics

pub fn collect_batch(&self) -> Vec<(T, Sender<BatchResult<T>>, Instant)>

Collect a batch of requests (up to max_batch_size)

pub fn complete_batch(&self, results: Vec<(Sender<BatchResult<T>>, Instant, T)>)

Complete a batch with results.

Each entry is (response_sender, arrival_instant, response_payload).

pub async fn wait_for_batch(&self) -> bool

Wait for a batch to be ready

Trait Implementations

impl<T: Send + 'static> Clone for BatchProcessor<T>

fn clone(&self) -> Self

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.