Build a Custom CompletableFuture: Async Primitive and Parallel Array Processing

Problem Statement

Modern asynchronous frameworks rely on future/promise abstractions to coordinate parallel tasks without blocking threads unnecessarily. In this problem, you will implement a thread-safe asynchronous primitive called SimpleFuture and use it to coordinate concurrent execution.

Implement the SimpleFuture class with the following methods:

1. complete(value: T) -> bool:
   • Sets the result of the future to value and transitions its state from pending to completed.
   • Invokes all registered callbacks with the resolved value.
   • If the future has already been completed, subsequent calls to complete must do nothing and return False. Returns True on the first successful completion.
2. get(timeout: float | None = None) -> T:
   • Blocks the calling thread until the future is completed and returns the value.
   • If timeout is provided (in seconds) and expires before completion, raises TimeoutError.
3. then(fn: Callable[[T], R]) -> SimpleFuture[R]:
   • Registers a transform function fn to run when this future completes, returning a new SimpleFuture holding the result of fn(value).
   • If this future is already completed when then is called, fn should be executed immediately (or scheduled).
4. all_of(futures: list[SimpleFuture[T]]) -> SimpleFuture[list[T]] (static method):
   • Returns a new future that completes only when all provided input futures complete.
   • Resolves with a list of results preserving the order of the input list.
   • If the input list is empty, returns an immediately completed future with an empty list.

All state transitions, callback registrations, and completions must be safe for concurrent access by multiple threads.

Interface

from typing import Generic, TypeVar, Callable, Optional, List
import threading

T = TypeVar('T')
R = TypeVar('R')

class SimpleFuture(Generic[T]):
    def __init__(self):
        pass

    def complete(self, value: T) -> bool:
        pass

    def get(self, timeout: Optional[float] = None) -> T:
        pass

    def then(self, fn: Callable[[T], R]) -> 'SimpleFuture[R]':
        pass

    @staticmethod
    def all_of(futures: List['SimpleFuture[T]']) -> 'SimpleFuture[List[T]]':
        pass

Examples

Example 1: Basic Completion and Chaining

f = SimpleFuture()
chained = f.then(lambda x: x * 2)

assert f.complete(10) is True
assert f.complete(20) is False  # Already completed
assert chained.get() == 20

Example 2: Fan-in with all_of

f1 = SimpleFuture()
f2 = SimpleFuture()
aggregate = SimpleFuture.all_of([f1, f2])

def worker(fut, val):
    fut.complete(val)

t1 = threading.Thread(target=worker, args=(f1, "A"))
t2 = threading.Thread(target=worker, args=(f2, "B"))
t1.start(); t2.start()
t1.join(); t2.join()

assert aggregate.get() == ["A", "B"]

Constraints

• Number of futures combined via all_of: 0 <= n <= 10^4.
• Any number of threads may concurrently call complete, get, or then on the same future instance.
• Callback functions fn passed to then are guaranteed not to raise exceptions.

Edge Cases

• then called after complete has already been called.
• all_of passed an empty list.
• Calling get with timeout=0.0 on an incomplete future must raise TimeoutError immediately.