Worker Pool Implementations

A worker pool turns an unbounded stream of submitted work into a fixed, predictable amount of in-flight concurrency. In asyncio terms it is a set of long-lived worker coroutines, each looping on await queue.get(), draining a single shared asyncio.Queue until a sentinel or cancellation tears the pool down. The pool is the place where you put a hard ceiling on concurrency, attach backpressure, collect results, and decide what happens to a task that raises. This guide covers building that machinery for production: the N-worker-plus-queue baseline, bounded-queue backpressure, dynamic pool sizing, executor-backed pools for blocking and CPU work, and graceful drain — all with the diagnostics you need to operate them.

Architectural principles:

• Bounded concurrency is the whole point. The number of worker coroutines (and the executor's max_workers) is your concurrency ceiling. Without it, a burst of submissions becomes a burst of simultaneous connections, file descriptors, or processes, and the failure mode is exhaustion, not slowdown.
• The queue is the pressure valve. A bounded asyncio.Queue couples ingress rate to drain rate: when workers fall behind, await queue.put() blocks producers instead of letting memory grow. An unbounded queue silently converts backpressure into an OOM.
• Workers own their failures. A task raising inside a worker must not kill the worker loop or the pool. Decide per task: record the exception against a result handle, push to a dead-letter path, or retry — never let it propagate out of the worker's try.
• Shutdown is a protocol, not a kill. Stop accepting work, signal workers, let in-flight tasks finish (or hit a deadline), then join. Cancelling mid-flight is a fallback with a timeout, not the default.
• Pick the execution substrate by workload. Pure coroutine workers for async I/O; an executor-backed pool (run_in_executor) for blocking SDK calls or CPU-bound work that would otherwise stall the loop.

These five principles are not independent knobs — they interlock. A bounded queue is meaningless without a fixed worker ceiling to drain it, because an unbounded worker set turns every queued item into immediate concurrency. Per-task error isolation is what lets you keep workers long-lived, which is in turn what makes the drain protocol simple (you signal workers you already control rather than chasing ephemeral tasks). And the choice of substrate determines whether your "concurrency ceiling" is the worker count alone (coroutine pool) or the minimum of the worker count and an executor's max_workers. Hold all five in mind together; tuning one in isolation usually moves the bottleneck rather than removing it.

---

How a worker pool rides the event loop scheduler

A coroutine-based pool has no OS threads of its own. Each worker is an asyncio.Task registered with the loop; when a worker hits await queue.get() on an empty queue it suspends, and the loop is free to run other ready tasks. A put() that satisfies a waiting get() wakes exactly one waiter via the loop's internal future machinery — there is no busy-polling and no lock contention, because everything runs on the single loop thread cooperatively. This is why an asyncio pool can hold tens of thousands of "logical workers" cheaply: a suspended task is just a frame and a future, not a stack.

The catch is the flip side of cooperative scheduling: a worker that calls a blocking function never yields, so the loop stalls and every other worker stops too. That single fact dictates the structure of this whole topic — blocking and CPU-bound work must leave the loop thread via an executor, which is why an executor-backed pool is a first-class pattern below rather than an afterthought. For the broader map of how threads, processes, and tasks trade off against each other, start from the Concurrent Execution & Worker Patterns overview.

Worker pools build directly on two adjacent topics: the buffering and producer/consumer mechanics live in async queue management, and the rule for getting blocking or compute work off the loop is covered under CPU-bound task offloading. The concurrency ceiling itself is often expressed with an asyncio synchronization primitive such as a Semaphore when you want a limit without spawning a fixed worker set.

It also helps to be precise about what "worker" means here, because the word is overloaded. In a threaded pool a worker is an OS thread with its own stack, scheduled preemptively by the kernel; the GIL still serializes the bytecode they execute, so threads buy you overlap only when they spend time outside the interpreter (in I/O syscalls or C extensions that release the GIL). In a coroutine pool a worker is a Task — a suspended-resumable frame the loop multiplexes onto one thread cooperatively; there is no preemption, so a worker keeps the CPU until it awaits. That difference is why the threaded baseline tolerates an occasional blocking call (the kernel can still run other threads) while the coroutine baseline does not (a non-yielding worker freezes the loop). Choosing the substrate is therefore choosing a scheduling model, not just an API, and the patterns below are organized so the coroutine pool stays the default and blocking work is explicitly exported through an executor.

---

Pattern catalogue

N workers over a shared asyncio.Queue

The baseline. Spawn a fixed number of worker tasks that loop on queue.get(); producers put() work items. Concurrency is bounded by the worker count alone — no semaphore needed, because there are only N consumers. Use this when work items are roughly uniform and you want the simplest possible structure.

import asyncio

async def worker(name: str, queue: asyncio.Queue) -> None:
    while True:
        item = await queue.get()
        try:
            await asyncio.sleep(0.1)        # stand-in for real I/O
            print(f"{name} handled {item}")
        except Exception:
            # Never let a task error kill the worker loop.
            print(f"{name} failed on {item}")
        finally:
            queue.task_done()

async def main() -> None:
    queue: asyncio.Queue[int] = asyncio.Queue()
    workers = [asyncio.create_task(worker(f"w{i}", queue)) for i in range(4)]
    for item in range(20):
        await queue.put(item)
    await queue.join()                       # all items processed
    for w in workers:
        w.cancel()
    await asyncio.gather(*workers, return_exceptions=True)

When to use: a bounded batch of similar work items where you know the total upfront and just want to cap concurrency at N — image thumbnailing a directory, fanning out a fixed list of API calls. The queue.join()/cancel teardown is correct precisely because the producer is finite; once every item is task_done(), join() returns and you cancel the now-idle workers.

Trade-off: dead simple, but the worker count is fixed and the queue is unbounded — fine for a known finite batch, dangerous for an open-ended producer. There is also no result handle here: the workers print or side-effect rather than returning values to a caller. The moment you need backpressure, result collection, or an indefinite producer, move to the patterns below.

Bounded queue for backpressure

Give the queue a maxsize. Now await queue.put() blocks the producer once the buffer is full, propagating slowness upstream instead of accumulating memory. This is the single most important production change over the baseline: it makes the pool self-limiting under load. The blocking happens at the loop level — put() suspends the producer task on an internal future until a worker's task_done()-adjacent get() frees a slot — so there is no spin-waiting and the producer simply stops consuming its own input source until capacity returns.

When to use: any open-ended or high-volume producer — a socket reader, a Kafka consumer, a paginating API crawler — where ingress can outrun processing. The bounded queue converts "we received too much" into "we read our input slower", which is exactly the signal a well-behaved upstream (TCP flow control, a broker's consumer lag) is built to respond to.

import asyncio

async def producer(queue: asyncio.Queue) -> None:
    for item in range(1000):
        await queue.put(item)   # blocks here when workers fall behind
    for _ in range(4):
        await queue.put(None)   # one sentinel per worker

async def worker(queue: asyncio.Queue) -> None:
    while True:
        item = await queue.get()
        if item is None:
            queue.task_done()
            return
        try:
            await asyncio.sleep(0.05)
        finally:
            queue.task_done()

async def main() -> None:
    queue: asyncio.Queue[int | None] = asyncio.Queue(maxsize=50)
    async with asyncio.TaskGroup() as tg:
        for _ in range(4):
            tg.create_task(worker(queue))
        tg.create_task(producer(queue))

Trade-off: backpressure can deadlock if a producer is also a consumer of the same bounded queue, or if maxsize is smaller than a single batch a caller must enqueue atomically. Size maxsize to a few multiples of the worker count, not to "however much fits."

Dynamic pool sizing

When arrival rate swings widely, a fixed worker count is either idle or saturated. A dynamic pool adds workers when the queue backs up and retires idle ones when it drains — always with hysteresis (separate, gapped up/down thresholds) so the controller does not thrash on every sample. For coroutine workers the cost of a spawn is tiny (a create_task), so the main risk is not overhead but oscillation; the supervisor below scales up on sustained backlog (sampled on an interval) and lets workers self-retire after an idle timeout, which keeps the two decisions decoupled and naturally hysteretic.

import asyncio

class DynamicPool:
    def __init__(self, queue: asyncio.Queue, lo: int = 2, hi: int = 32):
        self.queue, self.lo, self.hi = queue, lo, hi
        self.workers: set[asyncio.Task] = set()

    def _spawn(self) -> None:
        t = asyncio.create_task(self._worker())
        self.workers.add(t)
        t.add_done_callback(self.workers.discard)

    async def _worker(self) -> None:
        while True:
            try:
                item = await asyncio.wait_for(self.queue.get(), timeout=2.0)
            except TimeoutError:
                if len(self.workers) > self.lo:
                    return            # retire an idle worker
                continue
            try:
                await asyncio.sleep(0.1)
            finally:
                self.queue.task_done()

    async def supervise(self) -> None:
        for _ in range(self.lo):
            self._spawn()
        while True:
            await asyncio.sleep(0.5)
            depth = self.queue.qsize()
            if depth > 2 * len(self.workers) and len(self.workers) < self.hi:
                self._spawn()          # scale up on sustained backlog

Trade-off: scaling has a cost — each spawn/retire touches the loop and, for executor-backed variants, the OS. Without hysteresis you get flapping; with too-slow hysteresis you under-react to bursts. Sizing the bounds correctly is its own topic, covered in optimizing worker pool sizes for mixed I/O and CPU workloads.

Executor-backed pool via run_in_executor

When tasks call blocking libraries (a sync DB driver, requests, heavy numpy), coroutine workers are useless — the call freezes the loop. Route those tasks through an executor so the blocking call runs on a thread (I/O) or process (CPU), while the coroutine worker awaits the wrapped future. Use asyncio.to_thread() for the common thread case, or a dedicated ProcessPoolExecutor for CPU work.

When to use: any pool whose unit of work touches a synchronous third-party library you cannot rewrite as async — the legacy database client, a vendor SDK that only ships a blocking interface, a C-extension transform. The coroutine worker becomes a thin shell: it owns the queue mechanics, backpressure, and error capture, while the actual work happens off the loop thread. This is also the bridge to dedicated CPU offloading, where the executor is a process pool rather than a thread pool.

import asyncio
from concurrent.futures import ProcessPoolExecutor

def cpu_heavy(n: int) -> int:
    return sum(i * i for i in range(n))      # runs in a separate process

async def worker(queue: asyncio.Queue, pool: ProcessPoolExecutor) -> None:
    loop = asyncio.get_running_loop()
    while True:
        n = await queue.get()
        if n is None:
            queue.task_done()
            return
        try:
            result = await loop.run_in_executor(pool, cpu_heavy, n)
            print(result)
        finally:
            queue.task_done()

async def main() -> None:
    queue: asyncio.Queue[int | None] = asyncio.Queue(maxsize=100)
    with ProcessPoolExecutor(max_workers=4) as pool:
        async with asyncio.TaskGroup() as tg:
            for _ in range(4):
                tg.create_task(worker(queue, pool))
            for n in (10**6, 2 * 10**6, 3 * 10**6):
                await queue.put(n)
            for _ in range(4):
                await queue.put(None)

Trade-off: the number of coroutine workers and the executor's max_workers are two separate ceilings — keep them aligned, or coroutine workers will pile up waiting on a too-small executor. Process pools also pay pickle/spawn overhead; see CPU-bound task offloading for when that cost is worth it.

Graceful drain and shutdown

Tearing a pool down correctly is a two-phase protocol: stop accepting new work, then let in-flight work finish under a deadline before cancelling. Sentinels (None per worker) are the cleanest signal for queue-based pools because they flow through the queue after all real work, guaranteeing each worker exits only once the backlog is empty.

import asyncio

async def drain(queue: asyncio.Queue, workers: list[asyncio.Task],
                deadline: float = 30.0) -> None:
    try:
        async with asyncio.timeout(deadline):
            await queue.join()                 # wait for in-flight to finish
    except TimeoutError:
        pass                                   # fall through to hard cancel
    for w in workers:
        w.cancel()
    await asyncio.gather(*workers, return_exceptions=True)

Trade-off: a deadline that is too short cancels healthy in-flight tasks; too long lets a single stuck task hold shutdown hostage. The cancel step relies on workers handling CancelledError cleanly — covered in cancellation patterns.

---

Resource boundaries

A worker pool exposes three independent limits, and getting any one wrong defeats the others:

• Worker count caps concurrent in-flight tasks. For coroutine pools this is the number of worker tasks; for an executor-backed pool you must also cap the executor (max_workers) — the smaller of the two wins.
• Queue maxsize caps buffered, not-yet-started work. Set it to roughly 2–4× the worker count: large enough to keep workers fed across jitter, small enough that backpressure engages before memory grows. An unbounded queue means there is effectively no backpressure at all.
• Per-task deadline caps how long any single item can run. Wrap the unit of work in asyncio.timeout() so one stuck task cannot occupy a worker forever and silently shrink your effective concurrency.

The relationship queue_depth ≈ arrival_rate × service_time − worker_count is Little's Law in disguise: if depth grows without bound at steady state, you are under-provisioned on workers, full stop — no queue size will save you. The queue only buffers transient mismatches between arrival and service rate; it cannot fix a sustained one. This is the most common sizing mistake in practice — operators bump maxsize to "stop the errors" when the queue fills, which only delays the OOM and hides the real signal that worker count is too low.

For coroutine pools there is a second, subtler ceiling that does not appear in any formula: the downstream dependency's own limit. A pool of 500 workers hammering a database with a 20-connection pool will see 480 workers blocked on connection acquisition at any instant, so your effective concurrency is 20 and the other 480 workers are pure overhead (and a latency cliff the moment the connection pool times out). Size the worker count to the narrowest resource in the chain — the connection pool, the rate limit, the remote concurrency cap — not to what your own process can physically run.

A Semaphore-bounded design (one limit, no fixed worker set) is the right alternative when work arrives as already-created coroutines rather than queue items; see synchronization primitives. The trade-off is that a semaphore gives you a ceiling but no buffer and no single place to attach result collection or drain logic — you get concurrency limiting for free but rebuild backpressure and lifecycle yourself.

---

Integrated production pool

The following pool combines the pieces: a bounded queue for backpressure, a fixed set of coroutine workers, per-task timeouts, result/error capture against a handle, sentinel-based graceful drain, and a TaskGroup for structured supervision.

import asyncio
import logging
import time
from dataclasses import dataclass, field
from typing import Any, Awaitable, Callable

log = logging.getLogger("pool")

@dataclass
class Job:
    fn: Callable[..., Awaitable[Any]]
    args: tuple = ()
    result: asyncio.Future = field(default_factory=asyncio.Future)

class WorkerPool:
    def __init__(self, workers: int = 8, maxsize: int = 32,
                 task_timeout: float = 10.0):
        self._n = workers
        self._timeout = task_timeout
        self._queue: asyncio.Queue[Job | None] = asyncio.Queue(maxsize=maxsize)
        self._tasks: list[asyncio.Task] = []
        self._submitted = self._completed = self._failed = 0
        self._closing = False

    async def submit(self, fn: Callable[..., Awaitable[Any]], *args) -> Any:
        if self._closing:
            raise RuntimeError("pool is draining; submission rejected")
        job = Job(fn, args)
        await self._queue.put(job)            # backpressure: blocks when full
        self._submitted += 1
        return await job.result               # caller awaits its own result

    async def _worker(self, wid: int) -> None:
        while True:
            job = await self._queue.get()
            try:
                if job is None:               # drain sentinel
                    return
                try:
                    async with asyncio.timeout(self._timeout):
                        value = await job.fn(*job.args)
                    job.result.set_result(value)
                    self._completed += 1
                except Exception as exc:      # per-task isolation
                    self._failed += 1
                    if not job.result.done():
                        job.result.set_exception(exc)
                    log.warning("worker %d task failed: %r", wid, exc)
            finally:
                self._queue.task_done()

    async def __aenter__(self) -> "WorkerPool":
        self._tasks = [asyncio.create_task(self._worker(i)) for i in range(self._n)]
        return self

    async def __aexit__(self, *exc) -> None:
        self._closing = True
        try:
            async with asyncio.timeout(30):
                await self._queue.join()      # let in-flight finish
        except TimeoutError:
            log.error("drain deadline exceeded; cancelling workers")
        for _ in self._tasks:
            await self._queue.put(None)       # wake idle workers to exit
        await asyncio.gather(*self._tasks, return_exceptions=True)
        log.info("pool closed: submitted=%d completed=%d failed=%d",
                 self._submitted, self._completed, self._failed)

    def stats(self) -> dict[str, int]:
        return {"queue_depth": self._queue.qsize(),
                "submitted": self._submitted,
                "completed": self._completed,
                "failed": self._failed,
                "in_flight": self._submitted - self._completed - self._failed}

async def demo() -> None:
    async def task(x: int) -> int:
        await asyncio.sleep(0.05)
        if x % 7 == 0:
            raise ValueError(f"bad input {x}")
        return x * x

    async with WorkerPool(workers=4, maxsize=16) as pool:
        results = await asyncio.gather(
            *(pool.submit(task, i) for i in range(20)),
            return_exceptions=True,
        )
    ok = [r for r in results if not isinstance(r, Exception)]
    print(f"ok={len(ok)} stats={pool.stats()}")

if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO)
    asyncio.run(demo())

Diagnostic Hook: Export stats() on a timer (every 5–10s) to your metrics backend and watch four series: queue_depth, in_flight, completed rate, and failed rate. Sustained queue_depth growth with in_flight == workers means you are worker-bound — add workers or shed load. in_flight < workers while queue_depth > 0 means workers are blocked on something off-loop (a sync call that should be in an executor). A climbing failed rate with flat completed is a poison-input or downstream-outage signal; route those to a dead-letter path rather than retrying in place.

---

Diagnostic Hook: operating the pool

• PYTHONASYNCIODEBUG=1 (or loop.set_debug(True)) logs any callback or task step that runs longer than loop.slow_callback_duration (default 0.1s) — the fastest way to catch a blocking call smuggled into a coroutine worker.
• Per-worker liveness: tag each worker task with task.set_name(f"pool-worker-{i}") and periodically log asyncio.all_tasks() names; a worker that vanishes from the set without the pool closing means its loop raised and exited — a bug you must never have.
• Backpressure visibility: alert when queue.qsize() stays above 0.8 * maxsize for more than a few sample windows — that is the pool telling you ingress is outrunning capacity before producers start stalling.
• Drain SLO: record time spent in __aexit__; if it regularly approaches the drain deadline, your per-task timeout is too generous or a downstream dependency is slow.

---

Failure modes

Failure mode	Root cause	Detection	Fix
Pool silently stalls, throughput → 0	A worker called a blocking function; the whole loop is frozen	PYTHONASYNCIODEBUG=1 slow-callback warnings; in_flight < workers with backlog	Move the blocking call to asyncio.to_thread() or an executor-backed worker
Unbounded memory growth	Unbounded queue absorbing a producer faster than workers drain	RSS climbs with queue.qsize(); no backpressure ever engages	Set asyncio.Queue(maxsize=...); let put() block producers
One bad task kills a worker	Exception escaped the worker's try/finally, task exits	Worker name disappears from asyncio.all_tasks(); effective concurrency drops	Wrap the unit of work; capture the error on the result handle, never re-raise into the loop
Shutdown hangs forever	queue.join() waits on a task stuck with no timeout	__aexit__ never returns; one task pinned in-flight	Wrap work in asyncio.timeout(); bound the drain with asyncio.timeout() then cancel
Coroutine workers pile up idle-waiting	Executor max_workers smaller than coroutine worker count	High in_flight, low CPU/throughput, executor queue grows	Align executor max_workers with the coroutine worker count
Scaling controller flaps	Up/down thresholds overlap (no hysteresis)	Worker count oscillates rapidly in metrics	Gap the thresholds and add a cooldown between resize actions

---

Frequently asked questions

How many worker coroutines should a pool have?

For pure async I/O the worker count is your concurrency ceiling against the downstream dependency, so size it to what that dependency tolerates (connection limits, rate limits) rather than to CPU cores. Apply Little's Law: at steady state queue depth grows unbounded if worker_count is below arrival_rate × service_time, so start there and tune against measured queue depth.

Why does my asyncio worker pool freeze completely under load?

Almost always a blocking call inside a coroutine worker. Because all workers share one loop thread, a single synchronous call (a sync DB driver, requests, heavy numpy) stalls every worker at once. Enable PYTHONASYNCIODEBUG=1 to get slow-callback warnings, then move the offending call to asyncio.to_thread() or an executor-backed worker.

Should I use a bounded asyncio.Queue or a Semaphore to limit concurrency?

Use a fixed set of workers draining a bounded queue when work arrives as plain items you enqueue and you want explicit backpressure on producers. Use a Semaphore when work already exists as coroutines and you just need to cap how many run at once without spawning a worker set. The queue gives you buffering and backpressure; the semaphore is lighter when you do not need either.

How do I shut a worker pool down without losing in-flight work?

Use a two-phase drain: stop accepting new submissions, then await queue.join() under an asyncio.timeout() deadline so in-flight tasks finish. After the deadline, push one sentinel per worker (or cancel) and gather the workers with return_exceptions=True. Never cancel mid-flight as the default path; cancellation is the fallback once the deadline is exceeded.

How do I run CPU-bound work in an async worker pool?

Coroutine workers cannot run CPU work without freezing the loop. Make the workers route the task through loop.run_in_executor() backed by a ProcessPoolExecutor so the computation runs in a separate process. Keep the coroutine worker count and the executor's max_workers aligned, since the smaller of the two is your real concurrency ceiling.

---

Related

• Concurrent Execution & Worker Patterns — up to the parent overview for how pools fit the full threads/processes/tasks trade-off.
• Optimizing worker pool sizes for mixed I/O and CPU workloads — the math and profiling for choosing worker and queue bounds.
• Async queue management — the producer/consumer and buffering mechanics underneath every pool.
• CPU-bound task offloading — when and how to move compute off the loop into a process pool.
• Cancellation patterns — making the drain-then-cancel shutdown path clean and leak-free.