Your question is Thread-Safe Queue for Async Batching. Start with the requirements on the right.
Run and submit as often as you like. When you're ready, talk me through your approach or go straight to the code.
Amazon SageMaker inference services often improve throughput by processing requests in batches. Implement a thread-safe batching queue that accepts requests from producer threads and returns FIFO batches when either the batch reaches max_batch_size or the oldest queued request has waited at least max_wait_ms.
For deterministic evaluation, implement batch_requests, which simulates timestamped producer arrivals and returns the batches that would be emitted. Your implementation must use a thread-safe queue abstraction backed by threading.Condition or an equivalent synchronization mechanism.
Implement batch_requests(requests, max_batch_size, max_wait_ms). requests is a list of two-element lists [arrival_ms, request_id], sorted by nondecreasing arrival time. Return a list of batches, where each batch is a list of request IDs in FIFO order. Before adding a request arriving at time t, flush any ready batch at time t. After all requests arrive, flush remaining requests at the last arrival time plus max_wait_ms.
A batch may contain at most max_batch_size requests. Requests arriving at the exact expiration time are processed after the existing batch is flushed.
def batch_requests(requests, max_batch_size, max_wait_ms):