Executors, futures and timers

01The intuition

The problem it solves

A price-comparison page asks five shops for a price, one after another. Each call takes a second, so the page takes five seconds. The obvious improvement starts five threads, but then the results have to be collected, and errors in a thread disappear.

import threading

results: dict[str, float] = {}

def fetch(shop: str) -> None:
    results[shop] = ask_shop(shop)          # an exception here is lost

def compare(shops: list[str]) -> dict[str, float]:
    threads = [threading.Thread(target=fetch, args=(s,)) for s in shops]
    for t in threads: t.start()
    for t in threads: t.join()
    return results

def ask_shop(shop: str) -> float:
    return 9.99

A thread per call, a global dictionary for answers, no timeouts, and a failed shop silently missing from the result. What the page wants is simpler to say than to build: run these five things in parallel, give me each answer or its error when I ask, and do not create a thread per request forever.

Build it one piece at a time

Step one: a future. It is a box for a result that does not exist yet. The worker fills it once; the caller asks for it and blocks until it is there. Build it from a condition variable, and make it hold either a value or an exception so errors travel with the result.

from typing import Any, Callable

class Future:
    def __init__(self) -> None:
        self._cond = threading.Condition()
        self._done = False
        self._value: Any = None
        self._error: BaseException | None = None

    def set_result(self, value: Any) -> None:
        with self._cond:
            if self._done:
                raise RuntimeError("already completed")
            self._done, self._value = True, value
            self._cond.notify_all()

    def set_exception(self, err: BaseException) -> None:
        with self._cond:
            self._done, self._error = True, err
            self._cond.notify_all()

    def result(self, timeout: float | None = None) -> Any:
        with self._cond:
            if not self._cond.wait_for(lambda: self._done, timeout):
                raise TimeoutError()
            if self._error:
                raise self._error
            return self._value

Step two: an executor. A fixed pool of worker threads takes jobs from a queue; submitting a job returns its future immediately. Threads are created once, not per request, and the pool size is a deliberate limit on parallelism.

from queue import Queue

class Executor:
    def __init__(self, workers: int) -> None:
        self._q: Queue = Queue()
        for _ in range(workers):
            threading.Thread(target=self._run, daemon=True).start()

    def submit(self, fn: Callable[..., Any], *args: Any) -> Future:
        fut = Future()
        self._q.put((fn, args, fut))
        return fut

    def _run(self) -> None:
        while True:
            fn, args, fut = self._q.get()
            try:
                fut.set_result(fn(*args))
            except BaseException as e:      # error travels to the caller
                fut.set_exception(e)

Step three: the page uses them. Submit all five, then collect, with a timeout per shop so one slow shop cannot hold the page, and with each failure reported per shop rather than swallowed. Python's concurrent.futures.ThreadPoolExecutor is this, with cancellation and callbacks added.

def compare(shops: list[str], pool: Executor) -> dict[str, float | str]:
    futures = {s: pool.submit(ask_shop, s) for s in shops}
    out: dict[str, float | str] = {}
    for shop, fut in futures.items():
        try:
            out[shop] = fut.result(timeout=2.0)
        except TimeoutError:
            out[shop] = "timed out"
        except Exception as e:
            out[shop] = f"error: {e}"
    return out

Five shops on a pool of two workers, each call one second:

Why it works

The future's invariant is that it completes at most once and that every waiter observes the same outcome; the condition variable turns "wait until done" into a blocking call without spinning. The executor's invariant is that every submitted job is run exactly once by some worker, which the queue guarantees. Together they separate three concerns that the first version tangled: starting work, limiting parallelism, and collecting results and errors.

Where it breaks: a job that submits another job to the same pool and waits for it can deadlock when all workers are waiting on each other; nested work needs a separate pool or an asynchronous design. A future nobody ever reads hides its exception, so log on set_exception. And daemon threads vanish at interpreter exit mid-job; a real pool has a shutdown that drains the queue.

wall time ≈ ceil(jobs / workers) × job time, so 5 jobs on 2 workers take 3 s; on 5 workers, 1 s

The trade-off you must say out loud

A pool caps parallelism, which protects the machine and the shops you call, but a cap that is too small turns parallel work back into a queue. Futures make errors explicit at the cost of a result call that can block or time out, so every caller must decide what to do on timeout. Say the pool size and the timeout as numbers. The follow-up that usually comes next is cancellation, which a future can only offer cooperatively (a flag the job checks), and composition, chaining a second step onto a future's completion.

Where it shows up in the 100 Google problems

• Build your own Future: the Future class above plus then() for chaining and exactly-once callbacks.
• Concurrent priority task scheduler: the executor with a priority queue and cancellation flags.
• Task scheduler with dependencies: a task's future completing is what makes its dependents ready.
• Timer with many concurrent timers: fired callbacks are submitted to a pool so a slow callback does not delay the next timer.

02Used in these problems

• Design a task scheduler with dependencies and retries
• Design a concurrent priority task scheduler
• Design a timer that supports many concurrent timers
• Build your own Future / CompletableFuture