Crate amate 

Dynamic batching on worker threads, with no dependencies.

Named after Amaterasu, the sun goddess of Japanese myth, who gathers scattered light and sends it out at once, as amate gathers waiting requests into one batch.

Items queue to a few worker threads, each running one batch at a time. While some worker is idle, every item gets a batch of its own, started at once. Once all workers are busy, the queue fills, and the next worker to finish takes everything queued, up to Config::max_batch by weight, as one batch. If the queue runs dry before the batch is full, the worker waits for more, but never longer than the Batching::window after the first item was queued: a server waits only while an announced item is still on its way in (its body is being received or decoded, see Engine::arrival), a batch that sums over time waits out its window (Config::waits_unannounced). Items nobody wants any more (their client gave up) are skipped, and a batch that panics loses only its own items, where panics unwind: with panic = "abort" the process stops.

What a batch is made of, how heavy each item is, how long the window is and what to do with the batch are the Batching’s: the engine knows nothing about models or replies. It is the algorithm behind furnace’s dynamic batching.

use std::sync::mpsc;
use std::time::Duration;
use amate::{Batching, Config, Engine, Queued};

/// Sums the numbers of each batch and sends the totals on
struct Sum(mpsc::Sender<u64>);

impl Batching for Sum {
    type Item = u64;

    fn weight(&self, _: &u64) -> usize {
        1
    }

    fn window(&self) -> Duration {
        Duration::from_millis(2)
    }

    fn run(&self, batch: Vec<Queued<u64>>) {
        let _ = self.0.send(batch.iter().map(|queued| queued.item).sum());
    }
}

let (totals, received) = mpsc::channel();
let config = Config { max_batch: 8, workers: 2, capacity: Some(1024), waits_unannounced: false };
let engine = Engine::start(Sum(totals), config, |i| format!("sum-{i}"));
for n in 1..=10 {
    engine.submit(None, n).unwrap();
}
// The workers run what is queued, then stop, and `Sum` with them
drop(engine);
assert_eq!(received.iter().sum::<u64>(), 55);

Structs

Arrival

See Engine::arrival

Config

Engine

Handle to the workers; they stop when it is dropped

Queued

An item and when it was queued

Traits

Batching

What a batch is made of and what to do with one