Skip to content

Building an Async Worker Pool with TaskGroup

The classic asyncio worker pool — create N tasks that loop over a queue, await queue.join(), cancel the tasks — works in a demo and leaks in a service. A worker that raises disappears silently and the pool quietly runs at N−1. Callers that submitted a job have no handle to its result, so errors are logged somewhere far from the code that cares. When the service shuts down, cancelled workers leave half-processed jobs whose callers wait forever. Each problem has the same root: the workers' lifetime is not tied to anything. asyncio.TaskGroup provides that tie. A pool that owns its workers through a task group is an async context manager whose async with block bounds every worker's life, and whose exit waits for them, propagates fatal errors and cancels the rest. This guide builds that pool with a future per job, bounded submission, a clear split between errors that fail one job and errors that must stop the pool, and a guarantee that every submitted job ends with a result, an exception or a cancellation.

Prerequisites

A pool whose lifetime is a block 4 stacked layers from async with WorkerPool to Caller. A pool whose lifetime is a block async with WorkerPool scope of the pool exit waits for workers TaskGroup owns N workers fatal error cancels all Bounded queue job + future submit waits when full Caller awaits its future result, error or cancel Nothing the pool starts can outlive the block that created it.

1. Let a TaskGroup own the workers

The pool delegates its own __aenter__ and __aexit__ to a TaskGroup, starts the workers inside it, and uses a bounded queue for jobs. Each job carries a future that the worker resolves, so the submitter can await exactly its own result.

import asyncio
from collections.abc import Awaitable, Callable
from typing import Any


class PoolClosed(RuntimeError):
    pass


class FatalWorkerError(Exception):
    """Errors that mean the pool itself is broken (disk full, lost credentials)."""


class WorkerPool:
    def __init__(self, workers: int, queue_size: int = 100) -> None:
        self._workers = workers
        self._queue: asyncio.Queue = asyncio.Queue(maxsize=queue_size)
        self._group = asyncio.TaskGroup()
        self._closed = False

    async def __aenter__(self) -> "WorkerPool":
        await self._group.__aenter__()
        for i in range(self._workers):
            self._group.create_task(self._worker(), name=f"pool.worker:{i}")
        return self

    async def __aexit__(self, exc_type, exc, tb):
        if exc_type is None:
            await self.close()                              # normal exit: finish queued work
        self._closed = True
        try:
            return await self._group.__aexit__(exc_type, exc, tb)
        finally:
            self._cancel_queued()                           # nobody will run these any more

    async def submit(self, fn: Callable[..., Awaitable[Any]], *args: Any) -> asyncio.Future:
        if self._closed:
            raise PoolClosed("pool is closed")
        future = asyncio.get_running_loop().create_future()
        await self._queue.put((fn, args, future))           # waits when the queue is full
        return future

    async def close(self) -> None:
        if not self._closed:
            self._closed = True
            for _ in range(self._workers):
                await self._queue.put(None)                 # one stop signal per worker

    def _cancel_queued(self) -> None:
        while not self._queue.empty():
            job = self._queue.get_nowait()
            if job is not None and not job[2].done():
                job[2].cancel()

    async def _worker(self) -> None:
        while (job := await self._queue.get()) is not None:
            fn, args, future = job
            if future.cancelled():                           # the caller gave up while queued
                continue
            try:
                result = await fn(*args)
            except FatalWorkerError:
                future.cancel()
                raise                                        # stops the whole pool
            except Exception as exc:
                if not future.done():
                    future.set_exception(exc)                # isolated: only this job failed
            except BaseException:
                if not future.done():
                    future.cancel()                          # pool cancelled mid-job
                raise
            else:
                if not future.done():
                    future.set_result(result)

Three details carry the design. Exiting the block normally calls close(), which queues one stop signal per worker behind the jobs already queued, so submitted work finishes before the group exits. Any exception propagating out of a worker — deliberately only FatalWorkerError and cancellation — makes the task group cancel every other worker. And the finally in __aexit__ cancels futures still sitting in the queue, so no submitter waits on a job that will never run.

Verify: the class imports cleanly, and creating WorkerPool(workers=4) inside async with starts four tasks named pool.worker:0 to pool.worker:3.

2. Submit jobs and await per-job results

submit() returns a future immediately after queuing — waiting only when the queue is full — so callers can submit many jobs and await them together, or await each one where it is needed.

import asyncio


async def resize(url: str, size: int) -> str:
    await asyncio.sleep(0.01)
    if url.endswith("bad"):
        raise ValueError(f"cannot decode {url}")
    return f"{url}@{size}"


async def main() -> None:
    async with WorkerPool(workers=4) as pool:
        futures = [await pool.submit(resize, f"img-{i}" + ("-bad" if i == 3 else ""), 128)
                   for i in range(10)]
        results = await asyncio.gather(*futures, return_exceptions=True)
    print([r if isinstance(r, str) else type(r).__name__ for r in results])


asyncio.run(main())

Nine images resize and the bad one reports ValueError in its own slot, while the pool keeps running. Because submit() awaits queue.put() on a bounded queue, a producer submitting faster than the pool can work is slowed at the source instead of filling memory — the bounded queue backpressure pattern applied at the pool's front door.

Verify: the printed list contains nine resized names and one ValueError at index 3.

3. Separate job failures from pool failures

Most exceptions belong to one job: bad input, a 404, a validation error. They go into that job's future and nowhere else. A few mean the pool cannot continue: the disk is full, credentials were revoked, the database is gone. Those should stop every worker and surface to the code that owns the pool.

import asyncio


async def main() -> None:
    async def disk_full() -> None:
        await asyncio.sleep(0)
        raise FatalWorkerError("disk full")

    try:
        async with WorkerPool(workers=2) as pool:
            in_flight = await pool.submit(asyncio.sleep, 1)
            await pool.submit(disk_full)
            queued = [await pool.submit(resize, f"late-{i}", 1) for i in range(3)]
            await asyncio.sleep(0.2)
    except* FatalWorkerError as group:
        print("pool stopped:", [str(e) for e in group.exceptions])
        print("in-flight job cancelled:", in_flight.cancelled(),
              "| queued jobs cancelled:", all(f.cancelled() for f in queued))


asyncio.run(main())

The fatal error propagates out of the async with as an ExceptionGroup, handled here with except* as described in handling ExceptionGroup from TaskGroup. The job still sleeping in the other worker was cancelled by the task group, and the three jobs that never started were cancelled by the pool's cleanup. Every future ended in a definite state.

Verify: the output reports disk full, and both the in-flight and queued futures report cancelled.

What should this exception do? A decision on What kind of failure is it with 3 outcomes. What should this exception do? What kind of failure is it? bad input, 404, validation fail this job only set_exception on future disk full, auth revoked stop the whole pool escape the worker pool is being cancelled cancel job future re-raise cancellation Keep the middle branch small; everything else is a job-level error.

4. Honour cancellation from submitters

A submitter that stops caring — its request timed out, its client disconnected — cancels its future. A cancelled future that is still queued should not consume a worker at all, and a caller-side timeout should not affect other jobs.

import asyncio


async def main() -> None:
    started: list[str] = []

    async def job(name: str) -> str:
        started.append(name)
        await asyncio.sleep(0.05)
        return name

    async with WorkerPool(workers=1) as pool:
        first = await pool.submit(job, "first")
        abandoned = await pool.submit(job, "abandoned")
        last = await pool.submit(job, "last")
        try:
            async with asyncio.timeout(0.01):
                await asyncio.shield(abandoned)          # wait briefly, then give up
        except TimeoutError:
            abandoned.cancel()                           # still queued behind 'first'
        print(await first, await last)
    print("jobs that actually ran:", started)            # ['first', 'last']


asyncio.run(main())

With one worker, abandoned was still queued when its caller gave up; the worker skipped it because its future was already cancelled, so the worker's time went to last. Shielding the await keeps the timeout from cancelling the future implicitly, which makes the cancellation an explicit decision in the caller.

Verify: the printed results are first last, and the list of jobs that ran omits abandoned.

5. Export what the pool is doing

A pool without metrics fails as a mystery. Queue depth shows whether the pool keeps up, busy workers show whether it is saturated, and job outcomes by kind show where errors come from. A thin subclass adds them without touching the core logic.

import asyncio
import collections
import time


class ObservedPool(WorkerPool):
    def __init__(self, workers: int, queue_size: int = 100) -> None:
        super().__init__(workers, queue_size)
        self.busy = 0
        self.outcomes: collections.Counter[str] = collections.Counter()
        self.durations: list[float] = []

    async def submit(self, fn, *args):
        future = await super().submit(self._timed(fn), *args)
        return future

    def _timed(self, fn):
        async def run(*args):
            self.busy += 1
            started = time.perf_counter()
            try:
                result = await fn(*args)
                self.outcomes["ok"] += 1
                return result
            except Exception as exc:
                self.outcomes[type(exc).__name__] += 1
                raise
            finally:
                self.busy -= 1
                self.durations.append(time.perf_counter() - started)
        return run

    def snapshot(self) -> dict:
        return {"queue_depth": self._queue.qsize(), "busy_workers": self.busy,
                "outcomes": dict(self.outcomes)}


async def main() -> None:
    async with ObservedPool(workers=3) as pool:
        futures = [await pool.submit(resize, f"img-{i}" + ("-bad" if i % 4 == 0 else ""), 64)
                   for i in range(12)]
        await asyncio.sleep(0.005)
        print("during:", pool.snapshot())
        await asyncio.gather(*futures, return_exceptions=True)
    print("after: ", pool.snapshot())


asyncio.run(main())

Queue depth that trends upwards while busy workers equals the worker count is the signal to add workers or shed load; busy workers well below the count with a deep queue points at stalled workers instead. Sizing the count itself is covered in optimizing worker pool sizes for mixed I/O and CPU workloads.

Verify: the snapshot during processing shows queued jobs and three busy workers; afterwards the queue is empty, no workers are busy, and outcomes count 9 successes and 3 ValueErrors.

Outcomes of twelve jobs through an observed pool 2 bars comparing ok with the others. Outcomes of twelve jobs through an observed pool ok 9 jobs ValueError 3 jobs From the step 5 example: every fourth image is marked bad. Outcome counts by type point at where errors come from, per job kind.

Verification

The pool is structured correctly when:

  • Worker lifetime is scoped: workers exist only inside the pool's async with block, and exiting it waits for them.
  • Every job ends definitely: each submitted future resolves with a result, an exception or a cancellation — including when the pool stops early.
  • Job and pool failures are separated: ordinary exceptions stay in their job's future, while fatal ones stop the pool and propagate.
  • Submission is bounded: producers wait at submit() when the queue is full.
  • Abandoned jobs cost nothing: cancelled queued futures are skipped, and metrics show queue depth, busy workers and outcomes.

Pitfalls & edge cases

  • Letting ordinary exceptions escape the worker. Any exception that leaves a worker stops the whole task group. Keep the fatal set small and explicit.
  • Submitting after close. Jobs submitted after stop signals are queued would never run; the pool raises PoolClosed instead.
  • Awaiting futures inside the async with for jobs that submit more jobs. A job that submits to its own pool and waits can deadlock when all workers are busy doing the same. Use a separate pool or unbounded fan-out inside the job.
  • CPU-heavy jobs. Coroutine workers share one thread; CPU-bound jobs belong in an executor-backed pool, as in CPU-bound task offloading.
  • Restart expectations. A task-group pool stops on fatal errors by design; long-running services that should heal instead need a supervisor around it.

Frequently Asked Questions

How do I build a worker pool with asyncio.TaskGroup?

Wrap a TaskGroup in a class that acts as an async context manager: start N worker tasks in aenter, have them read jobs from a bounded asyncio.Queue, and in aexit enqueue one stop signal per worker before delegating to the TaskGroup's exit. The TaskGroup then waits for workers and propagates failures.

How does a caller get the result of a job submitted to an async worker pool?

Create an asyncio.Future for each submitted job, put it on the queue with the job, and return it to the caller. The worker sets the future's result or exception when the job finishes, so the caller can await that future or gather many of them.

Should one failing job stop the whole worker pool?

Usually not. Catch ordinary exceptions in the worker and store them in the job's future so only that caller sees the failure. Let a small, explicit set of fatal errors, such as a full disk or revoked credentials, escape the worker so the TaskGroup cancels the rest of the pool and reports the error.

What happens to queued jobs when an asyncio worker pool stops?

Nothing will run them, so their callers would wait forever unless the pool resolves them. On exit, drain the queue and cancel every remaining job's future, and cancel the futures of jobs that were in flight when their worker was cancelled.