Build your own Future / CompletableFuture

01The design, in the six-step framework

1. Requirements and use cases

A producer thread computes a value and settles a future exactly once, with a value or an error. Consumers can block on get with a timeout, register a callback that runs once when the future settles (immediately if it already has), and build a derived future with then(fn) so results chain without blocking. A combinator turns a list of futures into one future of a list.

Out of scope: cancellation of the producer, an event loop or async/await integration, and a general thread pool (a two-line run_async stands in for one). Those are the follow-ups, not the core.

Constraints: many threads may call complete, get and on_done at the same time; a callback must never run twice or be lost, even if registered in the same instant the future settles; callbacks must run outside the future's lock so a callback that calls get on another future cannot deadlock; a callback that throws must not break the other callbacks or the settlement.

2. Core entities and relationships

• Future[T]: the one real class. A three-state machine (FutureState), a value or error slot, a list of pending callbacks and a condition variable.
• callback: any callable taking the settled future. then is built on top of it: the child future is settled by a callback registered on the parent.
• run_async: a producer helper that shows how a thread settles a future; a real executor would replace it.

A parent future aggregates its callbacks and only uses its child: once the child is settled the parent forgets it, so a long chain does not keep every intermediate value alive.

3. Interfaces and key methods

class Future(Generic[T]):
    def complete(self, value: T) -> bool:
        """Settle with a value; False if already settled."""
    def fail(self, error: BaseException) -> bool:
        """Settle with an error; False if already settled."""
    def get(self, timeout: float | None = None) -> T:
        """Block until settled; re-raise the error or TimeoutError."""
    def on_done(self, cb: Callable[["Future[T]"], None]) -> None:
        """Run cb exactly once, now if already settled, else at settlement."""
    def then(self, fn: Callable[[T], U]) -> "Future[U]":
        """A new future holding fn(value); errors propagate unchanged."""
    def done(self) -> bool: ...

Invariants: the state leaves PENDING at most once, and only inside _settle; once settled, _callbacks is empty forever and every later on_done runs its callback immediately; get never returns while PENDING.

4. Design patterns and the choices behind them

Settle-once as a compare-and-set

What varies is who settles first when two producers race, or when complete and fail both fire. _settle checks and changes the state under the lock, so the first caller wins and every later one gets False. That makes settlement idempotent, which is the whole reason the class is safe to hand to code you do not control.

Swap the callback list, then call outside the lock

The subtle requirement is the callback registered at the exact moment of settlement. on_done checks done() under the same lock that _settle uses to swap the list out; so a callback either lands in the list that _settle will drain, or it sees the future already settled and runs now. Running callbacks while holding the lock was rejected: a callback that blocks on get of another future whose settlement needs this lock would deadlock (see deadlock avoidance).

then as a child future plus an observer

then is the observer pattern reused: a forward callback maps the parent's outcome onto the child, turning an exception in fn into the child's failure. The alternative, a stored transformation applied inside get, was rejected because it would run fn on every reader and not at all if nobody reads.

Errors are values until get

A failed future stores the exception and re-raises it in get, in the consumer's thread, which is where a stack trace is useful. This is the boundary where an error turns back into an exception.

5. Core flow and code

1. f = run_async(load) creates a pending future and starts a thread.
2. The consumer calls g = f.then(parse): a child is created and forward is appended to f's callback list.
3. The consumer calls g.get(timeout=2) and waits on g's condition.
4. The loader thread calls f.complete(bytes): under f's lock the state becomes DONE, the callback list is swapped for an empty one, waiters are notified; outside the lock forward runs, calls parse and completes g.
5. g's _settle notifies its waiters; the consumer's wait_for returns and get returns the parsed value.
6. If parse raised, g would be FAILED and the consumer's get re-raises that exception.

import threading
from enum import Enum, auto
from typing import Callable, Generic, TypeVar

T = TypeVar("T"); U = TypeVar("U")

class FutureState(Enum):
    PENDING = auto(); DONE = auto(); FAILED = auto()

class Future(Generic[T]):
    def __init__(self) -> None:
        self._cv = threading.Condition()
        self._state = FutureState.PENDING
        self._value: T | None = None
        self._error: BaseException | None = None
        self._callbacks: list[Callable[["Future[T]"], None]] = []

    # ---- settlement: exactly once -------------------------------------
    def _settle(self, value: T | None, error: BaseException | None) -> bool:
        with self._cv:
            if self._state is not FutureState.PENDING:
                return False
            self._value, self._error = value, error
            self._state = FutureState.FAILED if error else FutureState.DONE
            callbacks, self._callbacks = self._callbacks, []
            self._cv.notify_all()
        for cb in callbacks:                 # outside the lock
            self._safe_call(cb)
        return True

    def complete(self, value: T) -> bool:
        return self._settle(value, None)

    def fail(self, error: BaseException) -> bool:
        return self._settle(None, error)

    # ---- observation ---------------------------------------------------
    def done(self) -> bool:
        return self._state is not FutureState.PENDING

    def get(self, timeout: float | None = None) -> T:
        with self._cv:
            if not self._cv.wait_for(self.done, timeout):
                raise TimeoutError("future not settled")
            if self._error is not None:
                raise self._error
            return self._value  # type: ignore[return-value]

    def on_done(self, cb: Callable[["Future[T]"], None]) -> None:
        with self._cv:
            if not self.done():
                self._callbacks.append(cb)
                return
        self._safe_call(cb)                  # already settled: run now, still outside the lock

    def _safe_call(self, cb: Callable[["Future[T]"], None]) -> None:
        try:
            cb(self)
        except Exception:
            pass                             # a listener's bug never breaks settlement

    # ---- composition ---------------------------------------------------
    def then(self, fn: Callable[[T], U]) -> "Future[U]":
        child: Future[U] = Future()

        def forward(parent: "Future[T]") -> None:
            if parent._error is not None:
                child.fail(parent._error)
                return
            try:
                child.complete(fn(parent._value))   # type: ignore[arg-type]
            except Exception as e:                   # noqa: BLE001
                child.fail(e)

        self.on_done(forward)
        return child

    @staticmethod
    def all_of(futures: list["Future[T]"]) -> "Future[list[T]]":
        result: Future[list[T]] = Future()
        lock = threading.Lock()
        remaining = [len(futures)]
        if not futures:
            result.complete([]); return result

        def one_done(f: "Future[T]") -> None:
            if f._error is not None:
                result.fail(f._error); return
            with lock:
                remaining[0] -= 1
                last = remaining[0] == 0
            if last:
                result.complete([x._value for x in futures])  # type: ignore[misc]

        for f in futures:
            f.on_done(one_done)
        return result


def run_async(fn: Callable[[], T]) -> Future[T]:
    """Tiny executor: runs fn on a thread and settles the future."""
    fut: Future[T] = Future()
    def body() -> None:
        try:
            fut.complete(fn())
        except Exception as e:  # noqa: BLE001
            fut.fail(e)
    threading.Thread(target=body, daemon=True).start()
    return fut

6. Extensions, concurrency and what interviewers probe

• Cancellation: add a CANCELLED state reachable only from PENDING via the same compare-and-set; the producer polls done() and stops, since a Python thread cannot be killed.
• Running callbacks on a pool: today they run on the settling thread. Accepting an executor in on_done moves them off the producer without changing the exactly-once guarantee.
• Where the lock matters: state change plus list swap in _settle, and the check-then-append in on_done. Both must use the same lock or a callback can be lost in the gap.
• all_of: the first failure fails the combined future immediately; the remaining counter is protected by its own small lock because callbacks arrive from several threads.
• Test first: register a callback before and after settlement and assert each runs once; two threads racing complete(1) and complete(2), assert exactly one True; a callback that raises followed by one that records, assert the second still runs.
• Common mistake: calling callbacks inside the lock, or checking done() without the lock in on_done, which loses a callback registered in the same microsecond as settlement.

02Where this question was reported

• Land the Lab · 'Build your own CompletableFuture'
• aceoffer.app · 'A custom future with multithreading'