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.

import threading
import time

import ucall

server = ucall.Server(ucall.Interface("127.0.0.1:0", pipelined_requests=8))


@server.batch("/infer", min_size=16, max_size=64, max_latency_microseconds=2_000)
def infer(requests: list[ucall.Request]):
    scores = [str(len(request.body)).encode() for request in requests]
    for request, score in zip(requests, scores, strict=True):
        request.reply(score)


threads = [threading.Thread(target=server.take_calls), threading.Thread(target=infer.take_batches)]
for thread in threads:
    thread.start()
time.sleep(0.1)
print(infer.statistics())
server.stop()
for thread in threads:
    thread.join()

server.batch registers the function through add_batch_endpoint and replaces it with its BatchEndpoint, which server.batch_endpoints lists too. A batch is due as soon as min_size requests wait, holds at most max_size, and goes smaller once its oldest request waited max_latency_microseconds; zeros 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. It receives a list of Request objects in arrival order, and may reply in any order or keep requests to answer later from any thread. When the handler raises, every request it left unanswered answers 500.

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. endpoint.statistics() counts batches and requests, and how many batches were filled_batches on reaching max_size, ready_batches on reaching min_size, or expired_batches once max_latency_microseconds passed, from any thread while the endpoint is polled.