Bounded Concurrency in Python asyncio: Why a Semaphore Is Not Enough

A semaphore can keep an API client at twenty concurrent requests while the process still holds a million pending tasks. The request limit is working. The admission policy is the problem.

This distinction matters whenever input arrives faster than downstream work completes: an import job, a document enrichment pipeline, a batch of embedding requests, or a service that reads records and calls another service for each one. Limiting active operations controls one resource. It does not, by itself, control the amount of waiting work.

We will build a small pipeline with a finite queue and a fixed worker population, then test the exact point where its producer stops advancing. The implementation and tests were run on CPython 3.13.4, using only the standard library. The example simulates I/O; it makes no claims about HTTP throughput or production latency.

The semaphore protects the call, after the task exists

Consider this sketch. call_remote is an application-provided async function, and items is a finite input collection:

limit = asyncio.Semaphore(20)

async def limited(item):
    async with limit:
        await call_remote(item)

await asyncio.gather(*(limited(item) for item in items))

The semaphore permits at most twenty entries into its protected section. That is the guarantee its counter provides. Python’s semaphore documentation describes the acquire and release behavior.

However, the starred argument expression consumes the iterable before the call to gather. The coroutines are then scheduled as tasks by gather. With a large collection, there can be a large population of tasks waiting to acquire those twenty permits. The gather API documents its coroutine scheduling behavior.

Each waiting operation brings bookkeeping and references to whatever its coroutine needs. Even a lazy input iterator loses its admission advantage once this expression expands it into arguments. The semaphore has no opportunity to stop that expansion.

For a modest, known batch, this can be a perfectly reasonable tradeoff. The problem starts when the input size is large, unbounded, or outside your control. At that point, the design needs a limit before it creates another independent unit of scheduled work.

Give active work and waiting work separate budgets

Use two numbers:

• W: the number of workers allowed to execute a handler.
• Q: the number of items allowed to wait in the queue.

For example, four workers and a queue capacity of eight describe a pipeline with four execution slots and eight waiting slots. A single producer feeds the queue. When those waiting slots fill, it must wait before admitting the next item.

source --> producer --> queue: up to Q items --> W workers --> awaited handler
               |
               +-- waits at put() when the queue is full

An asyncio.Queue with a positive maxsize makes put() wait when full. A zero limit means an unbounded queue, so the implementation below rejects zero. See the queue contract.

This gives the producer feedback from the consumers. A slower handler fills the queue; the full queue suspends the producer. There is no task-per-input list growing beside it.

A complete implementation

Save the following as bounded.py. Integer inputs keep the example focused. None is reserved as an end-of-stream marker; a generic implementation that accepts None as data needs a distinct sentinel.

"""Bounded admission for integer jobs; Python 3.13, standard library only."""

import asyncio
from collections.abc import AsyncIterable, Awaitable, Callable


async def run_bounded(
    source: AsyncIterable[int],
    handle: Callable[[int], Awaitable[None]],
    *,
    workers: int = 4,
    capacity: int = 16,
) -> None:
    if workers < 1 or capacity < 1:
        raise ValueError("workers and capacity must be positive")

    queue: asyncio.Queue[int | None] = asyncio.Queue(maxsize=capacity)

    async def produce() -> None:
        async for item in source:
            await queue.put(item)
        for _ in range(workers):
            await queue.put(None)

    async def work() -> None:
        while True:
            item = await queue.get()
            try:
                if item is None:
                    return
                await handle(item)
            finally:
                queue.task_done()

    async with asyncio.TaskGroup() as group:
        group.create_task(produce(), name="bounded-producer")
        for number in range(workers):
            group.create_task(work(), name=f"bounded-worker-{number}")


async def main() -> None:
    active = peak = completed = 0

    async def source():
        for number in range(100):
            yield number

    async def handle(item: int) -> None:
        nonlocal active, peak, completed
        active += 1
        peak = max(peak, active)
        try:
            await asyncio.sleep(0.001)  # Simulated I/O; not a throughput benchmark.
            completed += 1
        finally:
            active -= 1

    await run_bounded(source(), handle, workers=4, capacity=8)
    print(f"completed={completed}, peak_active={peak}")


if __name__ == "__main__":
    asyncio.run(main())

Run it with:

python3 bounded.py

The verified output is:

completed=100, peak_active=4

There are five child tasks in this configuration: one producer and four workers. Each worker awaits one handler call before reading its next item. A handler that secretly creates background tasks would violate that design, so awaiting the real work is part of the handler contract.

On normal completion, the producer places one sentinel per worker after all input items. A worker returns when it receives its sentinel. Because each worker exits after consuming one, a fast worker cannot consume every sentinel and leave the others stranded.

The TaskGroup owns the producer and workers. It waits for them at scope exit; an ordinary child exception causes sibling cancellation, with failures surfaced as an exception group. Python’s TaskGroup documentation defines that behavior.

The success condition here is successful completion of the group. There is no separate queue.join() wait. The task_done() call balances queue accounting for each retrieved item, including sentinels. Calling it during error cleanup does not claim that the application operation succeeded.

The admission bound is W + Q + 1

The extra one is easy to miss. In the producer, the async iterator yields an item before the producer awaits queue.put(item). When the queue is full, that item already exists in the producer’s local state.

With all handlers blocked, the source can therefore advance through:

• W items held by workers;
• Q items waiting in the queue;
• one item held by the producer while it waits to enqueue it.

For two workers and four queue slots, the seventh item is the one that blocks admission. The eighth has not been requested from the source. This is a property of this implementation, with one producer and no prefetch inside the source.

That is an item-count bound, not a byte limit. Eight queue entries could contain eight small IDs or eight large documents. Nor does it account for a database driver that preloads a page, an HTTP client that buffers a response, a handler that accumulates output, or the original input collection already resident in memory.

A useful memory estimate for capacity planning is:

pipeline payload memory ≈ Q × queued-item size
                        + W × worker working set
                        + producer's pending item
                        + source/client/output buffers

Use measured upper bounds or conservative sizes when payloads vary. The formula is a budgeting aid, not a Python allocator model. Passing a small record ID through the queue and loading its payload inside a worker can make the waiting budget easier to predict.

Test where admission actually stops

A peak concurrency assertion alone would also pass for the semaphore example. To distinguish the designs, block every handler and count how many inputs the source is allowed to yield.

Save this focused test as test_admission.py next to bounded.py:

import asyncio
import unittest

from bounded import run_bounded


class AdmissionTest(unittest.IsolatedAsyncioTestCase):
    async def test_source_stops_at_workers_plus_capacity_plus_one(self):
            produced = 0
            full = asyncio.Event()
            release = asyncio.Event()

            async def source():
                nonlocal produced
                for item in range(100):
                    produced += 1
                    if produced == 7:  # Two active, four queued, one pending put.
                        full.set()
                    yield item

            async def handle(item):
                await release.wait()

            task = asyncio.create_task(run_bounded(source(), handle, workers=2, capacity=4))
            try:
                async with asyncio.timeout(2):
                    await full.wait()
                    for _ in range(10):
                        await asyncio.sleep(0)
                    self.assertEqual(produced, 7)
            finally:
                release.set()
                await task

PYTHONASYNCIODEBUG=1 python3 -W error -m unittest -v test_admission.py

The event marks the seventh input being produced. Yielding to the loop afterward gives an incorrectly unbounded producer opportunities to advance further. The assertion checks that it does not. Releasing the handlers in finally allows the pipeline to finish even if the assertion fails.

This test does not depend on a guessed network delay. The blocked handler and the source counter expose the admission boundary directly. The timeout is a test watchdog, not part of the pipeline’s operating policy.

The accompanying local validation also passed cases for all 100 inputs being handled once, the four-worker concurrency ceiling, an empty source, invalid limits, a producer exception, a worker exception, parent cancellation, and deliberately reversed completion order. These are controlled checks of the example’s behavior; they do not certify a particular database or HTTP integration.

Decide what failure means before adding retries

The example aborts the pipeline on an ordinary producer or handler exception. That policy is appropriate when the caller should see a failed run and decide how to recover. It is not an instruction to discard failures in a service that must retain every job.

Suppose item 12 fails while items 13 and 14 have already completed. The caller receives a failure, but the completed side effects remain completed. Restarting the source from item zero can repeat them. A fixed worker pool controls resource use; it does not create a transaction around remote operations.

For durable processing, decide where progress is recorded and how a repeated item is recognized. If the source is a broker, queue admission is not successful processing: an item sitting in this in-memory queue can disappear with the process. Broker acknowledgement and recovery require their own design. The earlier article on idempotent .NET RabbitMQ consumers examines that separate boundary.

Some applications should continue after selected failures. In that case, catch the expected application error inside the handler, durably record a failed outcome, and return according to an explicit policy. Avoid a blanket exception handler that silently turns data loss into a successful run.

When several workers fail, the error report may contain multiple exceptions. An except* handler can select matching members without discarding unrelated failures; PEP 654 explains the exception-group model. Flattening everything into the first error can hide a second failure that matters for recovery.

Cancellation needs a resource owner

Keep cancellation cooperative. A handler should release resources in finally and normally propagate CancelledError. It should not independently cancel its worker as a substitute for reporting an application failure. Cancellation has special treatment in task groups. Python’s cancellation guidance describes the cleanup contract.

The pipeline accepts an async iterable, but it does not own arbitrary resources behind that iterable. If the caller owns an async generator with cleanup code, it can wrap the call in contextlib.aclosing. For a database cursor or client session, use that object’s supported async context manager. The aclosing documentation covers deterministic async-generator cleanup after early exit.

That ownership detail matters on an exception: the producer might be suspended while putting an already-yielded item into a full queue. The generator is then paused at its yield, so cancelling the producer is not itself an explicit close of the generator.

The cleanup test in the local suite wraps the source with aclosing, forces a worker failure, and checks both source cleanup and sibling-handler cleanup. Checking only that an exception was raised would miss leaked resource ownership.

A full queue makes time budgets visible

Queueing time is part of the time a caller waits. A two-second timeout started inside the handler excludes time already spent waiting for a worker.

If an item has an end-to-end deadline, carry a monotonic deadline with it and check the remaining budget when a worker picks it up. Reject expired work before starting an expensive operation. Decide separately whether an expired item should abort this whole pipeline or become a recorded per-item failure.

A whole-pipeline timeout may also be useful, but it does not undo completed writes or requests. The application still needs to reconcile uncertain outcomes. Python’s timeout contexts provide cancellation mechanics; the meaning of an expired operation belongs to the application.

There is another practical limit: a blocking handler prevents the event loop from servicing other tasks. The queue cannot correct blocking I/O or a long CPU loop on that thread. Python’s asyncio development guide discusses moving blocking work off the event-loop thread. Evaluate that boundary separately from queue sizing.

Choose capacity for a workload you can describe

The defaults in the example are demonstration values. They are not a recommendation for an API, database, or model endpoint.

Start W from the downstream resource budget. If the client has fewer usable connections than workers, some workers will simply wait inside the client. Increasing W again may add memory and scheduling overhead without adding useful throughput.

Set Q to absorb an acceptable burst, then observe how long items wait. A permanently full queue says the producer is repeatedly reaching the admission limit. Making it larger gives the backlog more room; it does not make the handler finish faster.

For a concrete planning example, assume measured sustained processing capacity is 200 items per second and a burst delivers 1,000 additional items while ordinary arrivals pause. The extra work needs about five seconds of service even before accounting for variability. If those items expire after one second, a queue large enough to hold the burst merely stores work that will become stale. Those numbers are hypothetical, chosen to make the capacity tradeoff explicit.

Track active handlers, queue depth, queue waiting time, handler duration, failure counts, expired items, and memory together. A task count that stays flat is useful evidence, but it is not enough if each task starts retaining larger objects.

Finally, this pipeline preserves the queue’s admission order, not completion order. A slow item can finish after later items. Strict output order needs another policy, such as bounded batches or admission tied to a reorder window. An unbounded dictionary of completed results would reintroduce the memory problem at the output end.

The design question to carry into a review is precise: when all workers are busy and the queue is full, where does the next item wait? In this implementation, it waits in the single producer, before another worker task exists. That is the boundary the admission test protects.