Python Asyncio: TaskGroup, Timeouts, and Bounded Concurrency Explained

Checked on October 3, 2026. The standalone examples were executed with Python 3.12.14. TaskGroup and asyncio.timeout require Python 3.11 or later. The examples simulate waiting locally; they make no network requests and do not measure real API performance.

An application needs several independent results. Running one request after another spends much of its time waiting, so a developer starts all requests together. The application becomes more responsive under light load, then encounters rate limits, incomplete results, and tasks that keep running after the user has left.

Concurrency changes the ownership of work as well as its timing. A useful implementation must explain who waits for tasks, what happens when one fails, how many may be active, and when the group should stop.

Short answer: use asyncio for workloads that can cooperatively wait, give related tasks a defined lifetime, bound active operations, and preserve cancellation. TaskGroup is useful when related work should fail together. Add an overall deadline and decide explicitly whether partial results are acceptable.

Async functions do not automatically run together

Calling an async def function produces a coroutine object. It does not by itself complete the work. An await or task scheduling step is required. The asyncio development guide describes warnings for coroutines that were created but never awaited.

Awaiting operation A and then operation B in a loop is sequential if B is not scheduled until A finishes. Creating tasks for both allows their waiting periods to overlap. That is valuable when the underlying operations support asynchronous waiting, such as an async HTTP client or database driver.

An await expression is also not a guarantee that another task will run immediately. An awaitable may already be ready. What matters is that execution can yield when it has to wait. A long calculation with no cooperative suspension can monopolize the event-loop thread despite living inside an async function.

The same applies to blocking library calls. Ordinary time.sleep, synchronous file operations, and a synchronous HTTP request do not become nonblocking because their caller uses async def. The development guide recommends moving blocking work out of the event-loop thread with an appropriate executor. Thread offloading can help blocking I/O; select a separate execution strategy for substantial CPU work based on the library and runtime.

Choose the failure contract first

Suppose a page needs both an account summary and its permission list. Displaying one without the other may be misleading, so the page treats the work as one unit. A different batch job may accept 99 successful item results and one recorded failure. Those are different contracts and should have different exception handling.

The coroutines and tasks documentation specifies the differing TaskGroup and gather behavior. Cancelling gather itself is a separate case from a child raising an error. Do not infer that every form of failure leaves the same tasks running.

TaskGroup waits for its children to finish as the context exits. That includes their cancellation cleanup. A successfully completed child is not undone when a sibling fails. A cancelled task may also have caused an external effect before cancellation arrived. A group controls task lifetime, not a database transaction.

A runnable example with bounded active work

This simulation produces doubled numbers while enforcing an active-operation limit. The waits merely give tasks opportunities to overlap. Result collection follows input order by retaining task handles, even though tasks may finish in another order.

import asyncio

async def run_batch(values, limit=2):
    if type(limit) is not int or limit < 1:
        raise ValueError("limit must be a positive integer")
    semaphore = asyncio.Semaphore(limit)
    active = 0
    peak = 0

    async def process(value):
        nonlocal active, peak
        async with semaphore:
            active += 1
            peak = max(peak, active)
            try:
                await asyncio.sleep(0.01)
                return value * 2
            finally:
                active -= 1

    async with asyncio.timeout(2.0):
        async with asyncio.TaskGroup() as group:
            tasks = [group.create_task(process(value)) for value in values]

    return [task.result() for task in tasks], peak, active

async def main():
    results, peak, active = await run_batch([3, 1, 4, 2])
    assert results == [6, 2, 8, 4]
    assert peak == 2 and active == 0
    empty, empty_peak, empty_active = await run_batch([])
    assert (empty, empty_peak, empty_active) == ([], 0, 0)
    print("Results:", results, "peak active:", peak)

asyncio.run(main())

The semaphore documentation describes an acquisition counter and recommends async with for releasing it reliably. Here the protected region includes the entire simulated operation. If the semaphore were released before awaiting the operation, it would not limit the operation's active duration.

The counter updates contain no intervening await and all participating coroutines run on the same event-loop thread. This exercise is not a recipe for shared counters across threads. Asyncio synchronization primitives themselves are not thread-safe.

A concurrency limit is not a rate limit

Two slots mean at most two tasks are inside the protected region. They do not mean two starts per second. When each task finishes quickly, many operations can begin within a second while only two are active at any instant.

An illustrative steady-state calculation is useful: two fully occupied slots with an average service time of 0.05 seconds could support about 40 completions per second under simplified assumptions. That is 2 / 0.05, not a benchmark or a prediction for a remote API. Queueing, variable latency, retries, and network overhead alter actual behavior.

A provider's request-per-period limit therefore needs time-based admission control in addition to a concurrency bound. If several application processes share an account limit, each process having its own semaphore does not enforce the shared account budget. Coordinate the policy at the scope where the limit applies.

For HTTP work, inspect client connection-pool limits too. A semaphore of 100 with a connection pool of ten does not establish 100 useful simultaneous connections. Retries can further increase request count; see HTTP errors and safe retries for the operation-level policy.

Bound the amount of scheduled work as well

The example creates one task per input item. That is clear for four values. With millions of values, most tasks might sit waiting for semaphore slots while their task objects and arguments still occupy memory.

For a large stream, use a fixed number of workers consuming an asyncio.Queue with a positive maxsize. The queue documentation states that await put waits when the queue is full. This lets the producer slow down when consumers cannot keep up, instead of storing all future work as tasks.

Queue capacity is measured in items, not bytes. Ten enormous payloads can still be expensive. Keep compact identifiers in the queue where practical, and retrieve large content only when processing begins. Also define how the producer and workers terminate, how a failed item is recorded, and whether the remaining items should continue.

Calling task_done says the queue item is accounted for; it does not independently prove business success. A worker that records an accepted failure may mark an item done, while a system promising all items succeeded must separately inspect the failure record.

Test cancellation using a controlled failure

This second script coordinates tasks with an event instead of assuming a particular network or timing outcome. The failing task waits until its sibling starts, then raises a ValueError. The sibling remains waiting until cancellation.

import asyncio

async def main():
    started = asyncio.Event()
    never_released = asyncio.Event()
    state = {"cancelled": False, "cleaned": False, "failures": 0}

    async def waiting_task():
        started.set()
        try:
            await never_released.wait()
        except asyncio.CancelledError:
            state["cancelled"] = True
            raise
        finally:
            state["cleaned"] = True

    async def failing_task():
        await started.wait()
        raise ValueError("Synthetic validation failure")

    try:
        async with asyncio.TaskGroup() as group:
            group.create_task(waiting_task())
            group.create_task(failing_task())
    except* ValueError as errors:
        state["failures"] = len(errors.exceptions)

    assert state == {"cancelled": True, "cleaned": True, "failures": 1}
    print(state)

asyncio.run(main())

The except* syntax handles matching members of an exception group. PEP 654 explains why grouped failures need this mechanism and how unmatched members continue to propagate. Catch the exceptions you understand; a broad handler that converts every programming error into an empty result makes debugging much harder.

The cancelled child re-raises CancelledError after recording its state. Swallowing cancellation can interfere with task groups and timeouts. Prefer finally for releasing resources, and keep cleanup bounded. Cancellation does not forcibly interrupt arbitrary blocking code.

Put the timeout around the intended budget

An overall timeout outside the task group includes waiting for slots and active work in its deadline scope. A per-operation timeout placed inside the semaphore covers only the operation after admission. A timeout placed before acquisition can also cover slot waiting. Select the placement from the promise made to the caller.

import asyncio

async def main():
    cleaned = False
    try:
        async with asyncio.timeout(0.01):
            try:
                await asyncio.Event().wait()
            finally:
                cleaned = True
    except TimeoutError:
        assert cleaned
        print("Deadline expired; cleanup ran")
    else:
        raise AssertionError("The unreleased event should time out")

asyncio.run(main())

Catch TimeoutError outside the timeout context. The timeout works through cancellation, so its deadline is not a hard operating-system kill switch. Slow cleanup or blocked event-loop code can make the caller regain control later than the nominal deadline. Do not promise exact wall-clock termination from this pattern.

Record queue waiting, active duration, timeout counts, failure types, and peak concurrency separately. They reveal different bottlenecks. For this article, correctness is verified through results and state assertions; no speedup claim is inferred from millisecond sleeps.

Run the concurrency and cancellation checks

This download turns the guide's examples into one local exercise. It doubles four synthetic values, verifies a maximum of two active operations, and collects results in input order. It also checks an empty batch and rejects invalid concurrency limits. The waits happen on your computer; the script never contacts an API.

The more useful experiment is the controlled failure. One task waits until its sibling has started, then raises an error. Watch the sibling record cancellation, run its cleanup, and propagate cancellation correctly. A separate deadline check confirms that cleanup runs before the timeout is handled. Compare the printed object with the supplied expected result, then find which assertions would fail if cleanup were removed. These are correctness checks, not evidence of network speedup. A semaphore also does not enforce a requests-per-second limit or bound the number of scheduled tasks.

Download the asyncio reproduction pack (ZIP). Python 3.11 or later is required; the README records the version actually tested.

Sources and reproducibility

• Python 3.12 coroutines and tasks: task groups, gather, cancellation, and timeouts.
• Python asyncio development guide: blocking work, cooperative execution, and diagnostics.
• Python synchronization primitives: semaphore scope and release behavior.
• Python asyncio queues: bounded queues and completion accounting.
• PEP 654: exception groups and except*.

Each block runs as a separate Python script. In a notebook that already runs an event loop, invoke its main coroutine with the notebook's supported top-level await instead of starting another loop with asyncio.run.