Batches

A batch endpoint gathers matching requests from every event loop into one queue and hands them to its handler in batches, which suits work that is cheaper per request in bulk, like GPU inference.

use std::time::Duration;
use ucall::*;

let mut server = Server::new(&[Interface::new("127.0.0.1:0").pipelined_requests(8)])?;
let batching = Batching { min_size: 16, max_size: 64, max_latency: Duration::from_millis(2) };
let infer = server.batch("/infer", Verb::Post, batching, |batch: Batch| {
    let inputs: Vec<&[u8]> = batch.iter().map(Request::body).collect();
    let scores: Vec<String> = inputs.iter().map(|input| input.len().to_string()).collect();
    for (request, score) in batch.into_iter().zip(scores) {
        let _ = request.reply(HTTPStatus::OK, &[], score.as_bytes());
    }
})?;

std::thread::scope(|scope| {
    scope.spawn(|| server.take_calls(0));
    scope.spawn(|| infer.take_batches());
    scope.spawn(|| {
        std::thread::sleep(Duration::from_millis(100));
        println!("{:?}", infer.statistics());
        server.stop();
    });
});

Batching dispatches as soon as min_size requests wait, never more than max_size at once, and a smaller batch once its oldest request waited max_latency; zero fields mean 1, 32 and 10 milliseconds. The handler runs on the thread polling the endpoint through take_batch or take_batches, never on an event loop, so a slow batch never stalls the network. A Batch lends its requests in arrival order: get and iter borrow them, take moves one out, and iterating it by value takes them all, forwards or reversed. Requests the handler neither replies to nor keeps answer 500 once it returns, and all of them do when it panics. Taken requests may be replied to later, from any thread, while their responses still leave each connection in request order.

The queue holds as many requests as the interfaces admit at once, connections times pipelined requests, and refuses more with 503. Server::stop releases the requests still queued without a reply, and a dropped BatchEndpoint leaves its queue unanswered until the server is freed. BatchEndpoint::statistics counts batches and requests, and whether each batch was dispatched on reaching max_size, on reaching min_size, or once max_latency passed, from any thread while the endpoint is polled.