A small worker pool, explained by execution

1. Fix the contract before writing threads

Our example processes document IDs. The processing function may return a value or raise an ordinary Exception subclass. We want several workers, but no more than a fixed number may run at once.

The single-coordinator restriction is deliberate. Supporting submit() racing with shutdown() requires an atomic closed-state boundary. Your closeable condition queue supplies that stronger primitive; the standard sentinel example below does not.

2. Begin with the sequential program

def process(document_id: str) -> str:
    return document_id.upper()


results = []
for document_id in ["a", "b", "c"]:
    results.append(process(document_id))

assert results == ["A", "B", "C"]

This gives a reference contract: three inputs produce three outputs. The order is deterministic because there is one execution path.

Adding workers changes two properties. Completion order can vary, and exceptions occur away from the coordinating thread. The concurrent version must make both behaviors explicit.

3. Add one queue and one worker

from queue import Queue
from threading import Thread


jobs = Queue(maxsize=2)
results = []


def worker():
    while True:
        item = jobs.get()
        result = process(item)
        results.append(result)
        jobs.task_done()


thread = Thread(target=worker)
thread.start()

for item in ["a", "b", "c"]:
    jobs.put(item)

jobs.join()

put() publishes work. get() claims work. task_done() records completion. jobs.join() waits until every published job has a matching completion.

This program processes all three jobs, but it never exits: after the third item, the worker loops back to get() and waits for a fourth item. The main thread has no shutdown message to send.

4. Trace the queue and unfinished count

Queue length and unfinished work measure different things. A worker removes an item before processing it, so an empty queue can coexist with active work.

5. Balance every get() with task_done()

This worker contains a liveness bug:

item = jobs.get()
result = process(item)  # may raise
results.append(result)
jobs.task_done()

If process() raises, execution skips task_done(). The unfinished count never reaches zero, so jobs.join() can wait forever.

The completion signal belongs in finally:

item = jobs.get()
try:
    result = process(item)
    results.append(result)
finally:
    jobs.task_done()

This restores the queue accounting, although the worker still dies from the uncaught exception. Section 7 will preserve both the accounting and the worker.

6. Stop workers with one sentinel each

For Python versions before Queue.shutdown(), a common worker-loop protocol uses a unique object as a control message:

STOP = object()


def worker():
    while True:
        item = jobs.get()
        try:
            if item is STOP:
                return
            results.append(process(item))
        finally:
            jobs.task_done()

The sentinel passes through the same queue as normal jobs. One worker retrieves it, balances it with task_done(), and returns from its target function.

With two workers, submit two sentinels:

jobs.join()  # normal jobs finished

for _ in workers:
    jobs.put(STOP)

jobs.join()  # sentinels consumed

for thread in workers:
    thread.join()

One sentinel cannot stop two workers. The first worker consumes it and exits; the second remains blocked in get().

The shutdown order also matters. Normal submissions finish before sentinels are inserted. Under this tutorial's contract, no producer can append a normal job behind the stop messages.

7. Record failures without losing the worker

Raw thread exceptions do not travel through Thread.join() to the coordinator. Catch ordinary job exceptions inside the loop and store the outcome under a lock:

try:
    result = self._process(item)
except Exception as exc:
    with self._outcomes_lock:
        self._errors.append((item, exc))
else:
    with self._outcomes_lock:
        self._results.append((item, result))
finally:
    self._jobs.task_done()

The inner try handles a job failure and allows the worker loop to continue. The outer finally guarantees completion accounting.

The result and error lists have multiple writers, so the same lock protects both. Their order reflects completion timing and should not be used as submission order.

8. Understand what queue capacity limits

Suppose there are two workers and the queue capacity is three. Once both workers have claimed a job, up to five jobs can be outstanding without a producer blocking:

2 jobs running in workers
3 jobs waiting in the queue
──────────────────────────
5 accepted, unfinished jobs

Worker count limits simultaneous processing. Queue capacity limits waiting work. put() blocks only when the queue itself is full; running jobs are no longer physically inside it.

This blocking behavior is backpressure: intake slows when downstream processing cannot keep pace. It prevents an unbounded queue from consuming memory while retaining every submitted item.

A two-worker trace

With two workers and queue capacity two, force both workers to pause while processing A and B:

Claiming C creates queue space even though Worker 1 immediately begins more work. Queue capacity controls waiting jobs; the unfinished count continues to include running jobs.

9. Complete implementation

from queue import Queue
from threading import Lock, Thread


class WorkerPool:
    def __init__(self, process, worker_count: int, capacity: int):
        if isinstance(worker_count, bool) or not isinstance(worker_count, int) or worker_count <= 0:
            raise ValueError("worker_count must be a positive integer")
        if isinstance(capacity, bool) or not isinstance(capacity, int) or capacity <= 0:
            raise ValueError("capacity must be a positive integer")

        self._process = process
        self._jobs = Queue(maxsize=capacity)
        self._stop = object()
        self._outcomes_lock = Lock()
        self._results = []
        self._errors = []
        self._closed = False

        self._workers = [
            Thread(target=self._worker, name=f"worker-{index}", daemon=False)
            for index in range(worker_count)
        ]
        for thread in self._workers:
            thread.start()

    def submit(self, item) -> None:
        if self._closed:
            raise RuntimeError("worker pool is closed")
        self._jobs.put(item)

    def shutdown(self) -> None:
        if self._closed:
            return

        self._closed = True
        self._jobs.join()

        for _ in self._workers:
            self._jobs.put(self._stop)

        self._jobs.join()

        for thread in self._workers:
            thread.join()

    def results(self):
        with self._outcomes_lock:
            return list(self._results)

    def errors(self):
        with self._outcomes_lock:
            return list(self._errors)

    def _worker(self) -> None:
        while True:
            item = self._jobs.get()
            try:
                if item is self._stop:
                    return

                try:
                    result = self._process(item)
                except Exception as exc:
                    with self._outcomes_lock:
                        self._errors.append((item, exc))
                else:
                    with self._outcomes_lock:
                        self._results.append((item, result))
            finally:
                self._jobs.task_done()

Why the closed flag is sufficient here

The contract gives lifecycle ownership to one coordinator: it finishes calling submit() before calling shutdown(). Therefore, the check in submit() cannot race with shutdown.

If arbitrary producer threads may submit concurrently with shutdown, this implementation is incomplete. A producer could pass the closed check, pause, and enqueue after stop messages. Use a queue whose close operation is synchronized with put(), such as your condition-based closeable queue or Python 3.13+ Queue.shutdown().

10. Tests that check the contract

def process(value):
    if value == 3:
        raise ValueError("bad input")
    return value * 10


pool = WorkerPool(process, worker_count=2, capacity=2)

for value in range(6):
    pool.submit(value)

pool.shutdown()

assert sorted(pool.results()) == [
    (0, 0),
    (1, 10),
    (2, 20),
    (4, 40),
    (5, 50),
]

errors = pool.errors()
assert len(errors) == 1
assert errors[0][0] == 3
assert isinstance(errors[0][1], ValueError)

assert all(not thread.is_alive() for thread in pool._workers)

try:
    pool.submit(99)
    assert False, "submit should fail after shutdown"
except RuntimeError:
    pass

The test sorts results because completion order is unspecified. It checks that a failed job is visible, that other jobs complete, and that every worker terminates.

This happy-path test calls shutdown() directly. If you deliberately remove task_done(), run the whole test file as a child process with a timeout. A helper thread is insufficient: after the assertion fails, that non-daemon thread and the workers can still keep the Python process alive.

# run_pool_check.py
import subprocess
import sys


try:
    subprocess.run(
        [sys.executable, "pool_check.py"],
        check=True,
        timeout=2,
    )
except subprocess.TimeoutExpired:
    print("pool_check.py did not terminate: inspect task_done() and shutdown")
    raise

Force a producer to experience backpressure

Events let the test hold the worker in a known state. With one worker processing A and a capacity-one queue holding B, submitting C cannot finish until the worker is released:

from queue import Full
from threading import Event, Thread


processing_started = Event()
release_worker = Event()
queue_was_full = Event()
submit_finished = Event()


def gated_process(value):
    processing_started.set()
    release_worker.wait()
    return value


pool = WorkerPool(gated_process, worker_count=1, capacity=1)
pool.submit("A")
assert processing_started.wait(timeout=1)  # worker holds A

pool.submit("B")                           # B fills the queue
assert pool._jobs.full()                   # safe: worker is gated


def submit_c():
    try:
        pool._jobs.put_nowait("C")
        raise AssertionError("expected the capacity-one queue to be full")
    except Full:
        queue_was_full.set()

    pool.submit("C")                       # waits for queue space
    submit_finished.set()


submitter = Thread(target=submit_c)
submitter.start()

try:
    assert queue_was_full.wait(timeout=1)
    assert submitter.is_alive()
    assert not submit_finished.is_set()
finally:
    release_worker.set()                   # A finishes; worker claims B
    submitter.join(timeout=1)

assert not submitter.is_alive()
assert submit_finished.is_set()

pool.shutdown()                            # B and C finish; worker stops
assert sorted(pool.results()) == [
    ("A", "A"),
    ("B", "B"),
    ("C", "C"),
]

The non-blocking probe proves that the capacity-one queue was full. The worker cannot remove B while it is held by the event, so the following public submit("C") must wait. The finally block releases the worker even if an assertion fails.

11. The same workload with ThreadPoolExecutor

from concurrent.futures import ThreadPoolExecutor, as_completed


with ThreadPoolExecutor(max_workers=2) as executor:
    future_to_value = {
        executor.submit(process, value): value
        for value in range(6)
    }

    results = []
    errors = []

    for future in as_completed(future_to_value):
        value = future_to_value[future]
        try:
            results.append((value, future.result()))
        except Exception as exc:
            errors.append((value, exc))

A future carries one job's value or exception back to the coordinator. The context manager stops accepting submissions and waits for submitted work when leaving the block.

The executor removes most worker-loop and shutdown plumbing. Repeated calls to its general submit() interface do not expose a public bounded-queue capacity. Python 3.14 added buffersize to Executor.map(), which pauses input iteration when that buffer is full; that option applies to mapped input, not individual submit() calls. The custom pool remains useful when bounded submission is part of the API contract.

12. Your exercise

• Can you explain why queue length can be zero while unfinished work is positive?
• Can you explain why every get(), including a sentinel, needs task_done()?
• Can you explain why one sentinel is needed per worker?
• Can you state the lifecycle assumption that makes the closed flag safe?
• Can you identify what ThreadPoolExecutor gives you and what backpressure control it does not?

Official references

• Synchronized queue documentation
• Threading primitives documentation
• Executors and futures documentation