Worker pool with futures

Q: Why threads, if Python has the GIL?

Infra tasks are mostly I/O: HTTP health checks, SSH commands, cloud API calls. A thread waiting on the network releases the GIL, so threads give real concurrency for this work. For CPU-bound work in Python you would use processes instead, and the pool design stays the same.

Q: How would this look in Go, and in production?

In Go the pool is a buffered chan func(), N goroutines ranging over it, and a sync.WaitGroup. close(ch) replaces the sentinels, because range exits once the channel is drained. A future is a result channel of size 1 per task, or you skip futures and use errgroup.Group with SetLimit(n). Production pools also bound the queue for backpressure, pass a context.Context for cancellation, and export queue depth and task latency as metrics.

Q: Why does Java's ExecutorService wrap errors in ExecutionException?

So the caller can tell an exception thrown by get() itself apart from one thrown by the task, and so the stack trace shows both threads. This version re-throws the original exception to keep the API close to the Python one. Either choice is fine in an interview if you say why.

hardScheduling and queues Must-do

Build a fixed-size worker pool: N threads that take tasks from a shared queue and run them, so a deploy tool can, for example, run health checks against 500 hosts with at most 20 in flight. Do not use concurrent.futures, ExecutorService or any other ready-made pool. Building one is the exercise.

API (Java: <T> TaskFuture<T> submit(Callable<T> fn), TaskFuture.get() / isDone(), shutdown(boolean wait))

◈ DIAGRAM

WorkerPool(num_workers: int)
submit(fn, *args, **kwargs) -> Future
shutdown(wait=True) -> None
 
Future.result(timeout=None) -> fn's return value   # blocks until finished
Future.done() -> bool

Rules

Exactly num_workers threads are started in the constructor. At most num_workers tasks run at the same time.
Tasks start in submission order (FIFO). They may finish in any order.
Future.result() blocks until the task finishes. It returns the value, or re-raises the exception the task raised. result(timeout) raises TimeoutError if the task is still running when the timeout expires.
A task that raises must not kill its worker thread. The pool keeps its full size.
shutdown() stops accepting work. Tasks already queued still run. With wait=True it returns only after every worker has exited. Calling it twice is harmless.
submit after shutdown raises RuntimeError (Java IllegalStateException).
num_workers < 1 raises ValueError.

The output lines are exact because each result() is read in submission order, whatever order the threads finish in.

Examples

Example 1

Input: pool = WorkerPool(2) f1 = pool.submit(square, 3) f2 = pool.submit(square, 4) f3 = pool.submit(check_host, "db-7") # raises ValueError f1.result() f2.result() f3.result() pool.shutdown() pool.submit(square, 5)

Output: 9 16 error: unreachable: db-7 rejected: pool is shut down

Explanation: The task's own exception comes out of result(). After shutdown, new work is refused.

Example 2

Input: pool = WorkerPool(3) futures = [pool.submit(slow_square, i) for i in range(9)] # each sleeps 50 ms pool.shutdown(wait=True) all(f.done() for f in futures) [f.result() for f in futures] peak_concurrency <= 3

Output: true [0, 1, 4, 9, 16, 25, 36, 49, 64] true

Explanation: shutdown(wait=True) lets all nine queued tasks finish before returning. slow_square records how many tasks were running at once, and it never exceeds the pool size.

Hints

Approach

Optimal

Three pieces:

Queue: an unbounded thread-safe FIFO (queue.Queue, LinkedBlockingQueue) holding (fn, args, future) items.
Workers: each thread loops on get(). It runs the task inside try/except and stores either the result or the exception in the future. Catching the exception is what keeps the worker alive after a bad task.
Future: result and error fields plus a threading.Event (Java: CountDownLatch(1)). The worker writes the fields and then sets the event. result() waits on the event and then reads the fields. The event's happens-before guarantee makes those writes visible to the waiting thread.

Shutdown: under a lock, set closed and push one _STOP sentinel per worker. Because the queue is FIFO, the sentinels come after every task already submitted, so queued work still runs. Each worker exits when it pulls a sentinel. submit checks closed under the same lock, so no task can be queued behind the sentinels, where it would never run.

ComplexityTime O(1) per submit; the tasks' own time on N threadsSpace O(queued tasks + N)

Python

import queue
import threading
 
 
class Future:
    def __init__(self):
        self._done = threading.Event()
        self._result = None
        self._error = None
 
    def _finish(self, result=None, error=None):
        self._result, self._error = result, error
        self._done.set()                  # publishes result/error to waiting threads
 
    def done(self):
        return self._done.is_set()
 
    def result(self, timeout=None):
        if not self._done.wait(timeout):
            raise TimeoutError("task not finished")
        if self._error is not None:
            raise self._error
        return self._result
 
 
class WorkerPool:
    _STOP = object()                      # sentinel: one per worker on shutdown
 
    def __init__(self, num_workers):
        if num_workers < 1:
            raise ValueError("num_workers must be at least 1")
        self._tasks = queue.Queue()
        self._lock = threading.Lock()
        self._closed = False
        self._workers = [threading.Thread(target=self._run, name=f"worker-{i}", daemon=True)
                         for i in range(num_workers)]
        for w in self._workers:
            w.start()
 
    def _run(self):
        while True:
            item = self._tasks.get()
            if item is self._STOP:
                return
            fn, args, kwargs, fut = item
            try:
                fut._finish(result=fn(*args, **kwargs))
            except Exception as exc:      # a failing task must not kill the worker
                fut._finish(error=exc)
 
    def submit(self, fn, *args, **kwargs):
        with self._lock:                  # no submit can slip in after the sentinels
            if self._closed:
                raise RuntimeError("pool is shut down")
            fut = Future()
            self._tasks.put((fn, args, kwargs, fut))
            return fut
 
    def shutdown(self, wait=True):
        with self._lock:
            if not self._closed:
                self._closed = True
                for _ in self._workers:   # FIFO: sentinels land after every queued task
                    self._tasks.put(self._STOP)
        if wait:
            for w in self._workers:
                w.join()

Follow-up questions

Bound the queue so submit blocks, or fails fast, when 1,000 tasks are already waiting. Which is better for a CLI and which for a server?
Add shutdown_now() that drops queued tasks and marks their futures cancelled.
Add a per-task timeout. Why can't you forcibly kill a running thread, and what do you do instead?

Frequently asked questions

Infra tasks are mostly I/O: HTTP health checks, SSH commands, cloud API calls. A thread waiting on the network releases the GIL, so threads give real concurrency for this work. For CPU-bound work in Python you would use processes instead, and the pool design stays the same.

In Go the pool is a buffered chan func(), N goroutines ranging over it, and a sync.WaitGroup. close(ch) replaces the sentinels, because range exits once the channel is drained. A future is a result channel of size 1 per task, or you skip futures and use errgroup.Group with SetLimit(n). Production pools also bound the queue for backpressure, pass a context.Context for cancellation, and export queue depth and task latency as metrics.

So the caller can tell an exception thrown by get() itself apart from one thrown by the task, and so the stack trace shows both threads. This version re-throws the original exception to keep the API close to the Python one. Either choice is fine in an interview if you say why.

Problem statement