Parallelize Blocking API Calls with a Thread Pool

Parallelize Blocking API Calls with a Thread Pool

In machine learning model serving and backend microservices, a common bottleneck is querying downstream services sequentially. When processing a batch of items (e.g., fetching embeddings, running inference, or querying external databases), blocking calls should run concurrently without overwhelming downstream services.

Implement a function execute_batch that processes a list of inputs using a bounded thread pool, preserving the original ordering of results while managing exceptions and timeouts.

Function Signature

from dataclasses import dataclass
from typing import Any, Callable, List, Optional, TypeVar

T = TypeVar("T")
R = TypeVar("R")

@dataclass
class TaskResult:
    index: int
    input: Any
    value: Optional[Any] = None
    error: Optional[Exception] = None
    timed_out: bool = False

def execute_batch(
    items: List[T],
    fn: Callable[[T], R],
    max_workers: int,
    timeout: Optional[float] = None,
) -> List[TaskResult]:
    pass

Requirements

1. Worker Bounds: At most max_workers threads may invoke fn concurrently.
2. Order Preservation: The returned list of TaskResult objects must correspond exactly to the indices and order of items (result[i].index == i).
3. Failure Isolation: An unhandled exception in fn(items[i]) should not abort execution for remaining items. Store the caught exception in TaskResult.error with value = None.
4. Overall Timeout: If timeout (in seconds) is provided, the total execution of execute_batch must not exceed timeout. Any task not completed within the deadline must have timed_out = True (or error set to a TimeoutError). Pending tasks waiting in the pool queue should be canceled immediately once the timeout expires.
5. Resource Cleanup: Thread pool resources must be cleaned up properly before returning.

Examples

Example 1: Latency Variance with Preserved Order

import time

def mock_call(x: int) -> int:
    # Later elements finish faster
    time.sleep((5 - x) * 0.05)
    return x * 10

results = execute_batch(items=[1, 2, 3, 4], fn=mock_call, max_workers=2)
assert [r.value for r in results] == [10, 20, 30, 40]
assert [r.index for r in results] == [0, 1, 2, 3]

Example 2: Errors and Timeouts

def faulty_call(x: int) -> int:
    if x == 2:
        raise ValueError("Invalid input")
    if x == 3:
        time.sleep(1.0)
    return x

results = execute_batch(
    items=[1, 2, 3],
    fn=faulty_call,
    max_workers=2,
    timeout=0.2
)

assert results[0].value == 1
assert isinstance(results[1].error, ValueError)
assert results[2].timed_out is True

Constraints

• 0 <= len(items) <= 1000
• 1 <= max_workers <= 64
• timeout is either None or a positive float.