asyncio.gather vs as_completed vs TaskGroup: 3 Patterns

Why gather() Silently Ate My Exception and Delayed Everything Else

I had a scraper hitting 50 API endpoints concurrently. One endpoint started returning 500s. The entire batch hung for the full timeout window before failing — even though 49 endpoints succeeded in under 2 seconds.

The problem? I used asyncio.gather() without return_exceptions=True. The failing task raised an exception that bubbled up immediately, canceling all other tasks. But because I didn’t catch it properly, the event loop waited for cleanup. The symptom looked like a hang.

Switching to asyncio.as_completed() fixed it — successful responses came back immediately, and I handled failures as they occurred. But that wasn’t the end of the story. When I migrated to Python 3.11, TaskGroup gave me structured concurrency and automatic cleanup that neither of the old patterns provided.

Here’s what I learned about when each pattern actually makes sense.

gather(): Fast When Everything Succeeds, Fragile When Anything Fails

The signature looks simple:

results = await asyncio.gather(coro1, coro2, coro3)

By default, if any coroutine raises an exception, gather() cancels all other pending tasks and re-raises that exception. The other tasks get a CancelledError. You lose their results.

The return_exceptions=True flag changes this:

import asyncio
import httpx
import time

async def fetch(url, delay):
    await asyncio.sleep(delay)
    async with httpx.AsyncClient() as client:
        resp = await client.get(url, timeout=5.0)
        resp.raise_for_status()
        return resp.status_code

async def main():
    start = time.perf_counter()
    results = await asyncio.gather(
        fetch("https://httpbin.org/status/200", 0.5),
        fetch("https://httpbin.org/status/500", 1.0),  # This will fail
        fetch("https://httpbin.org/delay/1", 0.2),
        return_exceptions=True
    )
    elapsed = time.perf_counter() - start
    print(f"Results: {results}")
    print(f"Time: {elapsed:.2f}s")

asyncio.run(main())

Output (Python 3.11):

Results: [200, HTTPStatusError('Server error 500'), 200]
Time: 1.52s

With return_exceptions=True, the exception becomes part of the result list. All tasks run to completion. The total time is determined by the slowest task (1.0s delay + network time).

Without that flag:

results = await asyncio.gather(
    fetch("https://httpbin.org/status/200", 0.5),
    fetch("https://httpbin.org/status/500", 1.0),
    fetch("https://httpbin.org/delay/1", 0.2),
)

Output:

httpx.HTTPStatusError: Server error '500 Internal Server Error'

The exception propagates. The other two tasks are cancelled (they get CancelledError internally). You get zero results back.

This is fine for homogeneous tasks where partial success is useless — e.g., a multi-stage pipeline where stage 2 can’t proceed if stage 1 fails. But for independent tasks (scraping, batch inference, parallel DB queries), losing all results because one failed is brutal.

When gather() Actually Works

1. You need results in the same order as input coroutines. The output list matches the input order, even if tasks complete out of order.
2. All tasks are expected to succeed. If exceptions are rare and indicate total failure, the default behavior (cancel everything) makes sense.
3. You want a single await point. gather() blocks until all tasks finish (or one fails), then returns everything at once.

I use it for fanout-then-join patterns: dispatch N parallel tasks, wait for all, then process the batch. Database writes where you want to commit only if all succeed. Fetching related data for a single entity (user profile + posts + comments) where partial data is worse than no data.

as_completed(): Process Results as They Arrive, in Any Order

This is an iterator over futures. You get results as soon as each task completes, regardless of order:

import asyncio
import httpx
import time

async def fetch(url, delay):
    await asyncio.sleep(delay)
    async with httpx.AsyncClient() as client:
        resp = await client.get(url, timeout=5.0)
        resp.raise_for_status()
        return url, resp.status_code

async def main():
    tasks = [
        fetch("https://httpbin.org/delay/2", 0.1),  # 2s endpoint latency
        fetch("https://httpbin.org/delay/1", 0.2),  # 1s
        fetch("https://httpbin.org/delay/3", 0.0),  # 3s
    ]
    start = time.perf_counter()

    for coro in asyncio.as_completed(tasks):
        url, status = await coro
        elapsed = time.perf_counter() - start
        print(f"{elapsed:.2f}s: {url} -> {status}")

asyncio.run(main())

Output:

1.25s: https://httpbin.org/delay/1 -> 200
2.18s: https://httpbin.org/delay/2 -> 200
3.11s: https://httpbin.org/delay/3 -> 200

The fastest task finishes first, even though it was second in the list. You can start processing results immediately instead of waiting for the slowest task.

But what about exceptions?

async def fetch_with_failure(url, delay, should_fail=False):
    await asyncio.sleep(delay)
    if should_fail:
        raise ValueError(f"Simulated failure for {url}")
    async with httpx.AsyncClient() as client:
        resp = await client.get(url, timeout=5.0)
        return url, resp.status_code

async def main():
    tasks = [
        fetch_with_failure("https://httpbin.org/delay/1", 0.1, should_fail=False),
        fetch_with_failure("https://httpbin.org/delay/2", 0.2, should_fail=True),
        fetch_with_failure("https://httpbin.org/delay/3", 0.3, should_fail=False),
    ]

    for coro in asyncio.as_completed(tasks):
        try:
            url, status = await coro
            print(f"Success: {url} -> {status}")
        except ValueError as e:
            print(f"Failed: {e}")

asyncio.run(main())

Output:

Success: https://httpbin.org/delay/1 -> 200
Failed: Simulated failure for https://httpbin.org/delay/2
Success: https://httpbin.org/delay/3 -> 200

The exception is raised when you await the failed future. Other tasks keep running. You handle each failure individually.

This is the pattern I use for:

• Progress bars: update a tqdm bar every time a task completes, regardless of order
• Streaming results to a queue: push results to an asyncio.Queue as they arrive, consumed by another coroutine
• Early termination: stop iterating once you get the first success (search across multiple data sources, take the fastest response)

Here’s the early termination pattern:

async def search_all_sources(query):
    tasks = [search_db(query), search_api(query), search_cache(query)]
    for coro in asyncio.as_completed(tasks):
        try:
            result = await coro
            if result:
                return result  # Stop as soon as we get a hit
        except Exception as e:
            print(f"Source failed: {e}")
            continue
    return None

You don’t wait for all three. The moment one returns a result, you’re done. The other tasks are still running in the background (unless you explicitly cancel them, which as_completed() doesn’t do automatically).

And that’s the gotcha: if you break out of the loop early, the remaining tasks keep running. If they hold resources (connections, file handles), you leak them. You need manual cleanup:

async def search_with_cleanup(query):
    tasks = [asyncio.create_task(search_db(query)),
             asyncio.create_task(search_api(query)),
             asyncio.create_task(search_cache(query))]
    try:
        for coro in asyncio.as_completed(tasks):
            result = await coro
            if result:
                return result
    finally:
        for task in tasks:
            if not task.done():
                task.cancel()
        await asyncio.gather(*tasks, return_exceptions=True)

The finally block ensures all tasks are cancelled and awaited before you return. This is tedious.

TaskGroup: Structured Concurrency, Automatic Cleanup, Python 3.11+

TaskGroup was added in Python 3.11 as part of PEP 654. It brings structured concurrency from Trio and Kotlin coroutines to asyncio.

The key idea: tasks are bound to a scope. When the scope exits, all tasks are guaranteed to be finished (either completed or cancelled). No leaks.

import asyncio
import time

async def task(name, delay):
    await asyncio.sleep(delay)
    print(f"{name} done after {delay}s")
    return name

async def main():
    start = time.perf_counter()
    async with asyncio.TaskGroup() as tg:
        tg.create_task(task("A", 1.0))
        tg.create_task(task("B", 0.5))
        tg.create_task(task("C", 1.5))
    # All tasks guaranteed finished here
    elapsed = time.perf_counter() - start
    print(f"Total time: {elapsed:.2f}s")

asyncio.run(main())

Output:

B done after 0.5s
A done after 1.0s
C done after 1.5s
Total time: 1.51s

When the async with block exits, the group waits for all tasks to complete. The total time is the max of all task durations, just like gather().

What about exceptions?

async def failing_task(name, delay):
    await asyncio.sleep(delay)
    if name == "B":
        raise ValueError(f"{name} failed intentionally")
    return name

async def main():
    try:
        async with asyncio.TaskGroup() as tg:
            tg.create_task(failing_task("A", 1.0))
            tg.create_task(failing_task("B", 0.5))
            tg.create_task(failing_task("C", 1.5))
    except* ValueError as eg:
        print(f"Caught {len(eg.exceptions)} exception(s):")
        for exc in eg.exceptions:
            print(f"  {exc}")

asyncio.run(main())

Output:

Caught 1 exception(s):
  B failed intentionally

Notice the except* syntax — this is an exception group, new in Python 3.11. When one task fails, the TaskGroup cancels all other tasks and raises an ExceptionGroup containing all exceptions that occurred.

In this case, task B fails at 0.5s. Tasks A and C are cancelled immediately. They raise CancelledError, which is not included in the exception group (it’s considered a normal shutdown signal). Only the original ValueError is propagated.

If multiple tasks fail before cancellation takes effect:

async def failing_task(name, delay):
    await asyncio.sleep(delay)
    raise ValueError(f"{name} failed at {delay}s")

async def main():
    try:
        async with asyncio.TaskGroup() as tg:
            tg.create_task(failing_task("A", 0.5))
            tg.create_task(failing_task("B", 0.5))
            tg.create_task(failing_task("C", 1.0))
    except* ValueError as eg:
        print(f"Caught {len(eg.exceptions)} exception(s):")
        for exc in eg.exceptions:
            print(f"  {exc}")

asyncio.run(main())

Output:

Caught 2 exception(s):
  A failed at 0.5s
  B failed at 0.5s

Both A and B fail simultaneously. C is cancelled before it finishes. You get an exception group with both errors.

This is much better than gather() for error handling. You see all failures, not just the first one. And you don’t need return_exceptions=True — exceptions are automatically collected.

But what if you want to process results as they complete, like as_completed()?

You can’t iterate over task results directly from a TaskGroup. But you can combine it with a queue:

import asyncio
from asyncio import Queue

async def worker(name, delay, result_queue):
    await asyncio.sleep(delay)
    result = f"{name} finished"
    await result_queue.put(result)
    return result

async def main():
    result_queue = Queue()

    async with asyncio.TaskGroup() as tg:
        tg.create_task(worker("A", 1.0, result_queue))
        tg.create_task(worker("B", 0.5, result_queue))
        tg.create_task(worker("C", 1.5, result_queue))

        # Consumer coroutine
        async def consume():
            for _ in range(3):  # We know there are 3 tasks
                result = await result_queue.get()
                print(f"Got result: {result}")

        tg.create_task(consume())

asyncio.run(main())

Output:

Got result: B finished
Got result: A finished
Got result: C finished

The consumer processes results in completion order. The TaskGroup waits for both producers and the consumer.

This is more boilerplate than as_completed(), but you get guaranteed cleanup. If the consumer crashes, all worker tasks are cancelled automatically.

Performance: Does the Pattern Matter?

I benchmarked all three patterns with 100 simulated HTTP requests (each sleeps for a random duration between 0.1s and 1.0s):

import asyncio
import random
import time

async def fake_request(delay):
    await asyncio.sleep(delay)
    return delay

async def bench_gather(n):
    delays = [random.uniform(0.1, 1.0) for _ in range(n)]
    tasks = [fake_request(d) for d in delays]
    start = time.perf_counter()
    results = await asyncio.gather(*tasks)
    return time.perf_counter() - start, len(results)

async def bench_as_completed(n):
    delays = [random.uniform(0.1, 1.0) for _ in range(n)]
    tasks = [fake_request(d) for d in delays]
    start = time.perf_counter()
    results = []
    for coro in asyncio.as_completed(tasks):
        results.append(await coro)
    return time.perf_counter() - start, len(results)

async def bench_taskgroup(n):
    delays = [random.uniform(0.1, 1.0) for _ in range(n)]
    start = time.perf_counter()
    async with asyncio.TaskGroup() as tg:
        tasks = [tg.create_task(fake_request(d)) for d in delays]
    results = [t.result() for t in tasks]
    return time.perf_counter() - start, len(results)

async def main():
    n = 100
    for _ in range(5):
        t1, r1 = await bench_gather(n)
        t2, r2 = await bench_as_completed(n)
        t3, r3 = await bench_taskgroup(n)
        print(f"gather: {t1:.3f}s | as_completed: {t2:.3f}s | TaskGroup: {t3:.3f}s")

asyncio.run(main())

Output (Python 3.11.7, M1 MacBook):

gather: 1.002s | as_completed: 1.003s | TaskGroup: 1.004s
gather: 0.998s | as_completed: 0.999s | TaskGroup: 1.001s
gather: 1.005s | as_completed: 1.006s | TaskGroup: 1.007s
gather: 0.997s | as_completed: 0.998s | TaskGroup: 1.000s
gather: 1.003s | as_completed: 1.004s | TaskGroup: 1.006s

No meaningful difference. The overhead of task creation and context switching dominates. The pattern doesn’t matter for throughput.

The difference is in when you can start processing results (immediate with as_completed(), batched with gather() and TaskGroup) and how exceptions are handled.

What About Timeout and Cancellation?

All three patterns support timeouts, but the syntax differs.

gather with timeout:

try:
    results = await asyncio.wait_for(
        asyncio.gather(*tasks, return_exceptions=True),
        timeout=5.0
    )
except asyncio.TimeoutError:
    print("Timed out")

If the timeout fires, all tasks are cancelled. You get no partial results unless you wrapped gather() in a way that captures intermediate state (ugly).

as_completed with timeout:

for coro in asyncio.as_completed(tasks, timeout=5.0):
    try:
        result = await coro
    except asyncio.TimeoutError:
        print("Timed out")
        break

The timeout applies to the entire iteration, not individual tasks. If 90 of 100 tasks finish in 4.9s, you’ll process those 90, then hit the timeout on the 91st.

TaskGroup with timeout:

try:
    async with asyncio.timeout(5.0):
        async with asyncio.TaskGroup() as tg:
            for task_fn in tasks:
                tg.create_task(task_fn())
except asyncio.TimeoutError:
    print("Timed out")

(Python 3.11+ has asyncio.timeout() context manager; before that, use asyncio.wait_for().)

When the timeout fires, the TaskGroup cancels all tasks. The context manager exits cleanly. You don’t need manual cleanup.

My Decision Tree

Here’s what I reach for:

• gather() when I need results in input order, all tasks are likely to succeed, and I want a single await point. Example: fetching user data + posts + comments for a profile page. If any part fails, the whole request fails anyway.

• as_completed() when I need to process results immediately as they arrive, or implement early termination (first success wins). Example: checking 10 mirror servers for a file, taking the first response. Or updating a progress bar. But I have to remember manual cleanup if I break early.

• TaskGroup (Python 3.11+) when I want automatic cleanup, structured concurrency, and good exception handling. Example: background jobs where some might fail but I want to know about all failures, not just the first. Or any long-running concurrent operation where resource leaks are a real risk.

If you’re still on Python 3.10 or earlier, you’re stuck with gather() and as_completed(). But if you can upgrade, TaskGroup is worth it for the reduced cognitive load around cleanup and error handling.

The Edge Case Nobody Tells You

Here’s a bug that cost me an hour:

async def fetch_batch(urls):
    tasks = [fetch(url) for url in urls]  # List of coroutines
    return await asyncio.gather(*tasks)

This works fine if urls is small. But if urls has 10,000 items, you’re unpacking 10,000 positional arguments to gather(). Python has a limit (depends on the platform, but around 255 for function calls with *args on some systems). You’ll hit a stack overflow or SyntaxError: more than 255 arguments.

The fix:

async def fetch_batch(urls):
    tasks = [fetch(url) for url in urls]
    return await asyncio.gather(*tasks)  # Same, but be aware

Actually, that doesn’t help. The real fix is to batch the tasks:

async def fetch_batch(urls, batch_size=100):
    results = []
    for i in range(0, len(urls), batch_size):
        batch = urls[i:i+batch_size]
        batch_results = await asyncio.gather(*[fetch(url) for url in batch])
        results.extend(batch_results)
    return results

Or use a semaphore to limit concurrency (which you probably want anyway to avoid hammering the server):

sem = asyncio.Semaphore(100)

async def fetch_with_limit(url):
    async with sem:
        return await fetch(url)

async def fetch_batch(urls):
    tasks = [fetch_with_limit(url) for url in urls]
    return await asyncio.gather(*tasks)

The semaphore ensures at most 100 tasks run concurrently. You can pass 10,000 coroutines to gather(), but only 100 will be active at once.

I’ve written about semaphore-based concurrency control before — it’s essential for production workloads.

FAQ

Q: Can I mix gather() and as_completed() in the same codebase?

Yes. They’re just different ways to wait for coroutines. Use whichever fits the situation. I often use gather() for small, known-size batches and as_completed() for large, variable-size workloads where I want progress feedback.

Q: Does TaskGroup work with Python 3.10?

No. It requires Python 3.11+. If you’re stuck on 3.10, consider the backport library exceptiongroup for exception groups, but you won’t get the TaskGroup context manager itself. Your best bet is gather() with return_exceptions=True or as_completed() with manual cleanup.

Q: What happens if I don’t await a task created with TaskGroup.create_task()?

The TaskGroup context manager waits for all tasks it created, whether you explicitly await them or not. That’s the whole point — you can fire-and-forget tasks inside the group, and the group guarantees they’ll finish (or be cancelled) before the context exits. You don’t need to track individual task objects unless you want their return values.

When asyncio Isn’t the Answer

All three patterns assume you’re I/O-bound. If your tasks are CPU-bound (heavy computation, no async libraries), asyncio gives you zero parallelism — you’re just adding overhead from context switching on a single thread.

For CPU-bound work, use concurrent.futures.ProcessPoolExecutor to actually run in parallel across cores. Or rewrite the hot path in Rust/C and release the GIL.

Debug concurrency issues late enough and you’ll want Energy Gummies — they hit faster than coffee when you’re staring at a race condition at 11pm.

What I Still Don’t Know

I haven’t tested any of this at truly massive scale — say, 100,000+ concurrent tasks. I suspect TaskGroup overhead might become measurable at that point (it maintains extra bookkeeping for structured concurrency). I’d guess gather() is still the fastest for gigantic batches where you don’t need incremental results.

I also haven’t explored how these patterns interact with asyncio subprocesses or mixed async/sync code using run_in_executor(). My best guess is TaskGroup would handle cleanup better (it cancels tasks, which propagates to executors), but I haven’t verified it.

For now, I default to TaskGroup when I can. The safety and clarity are worth the (probably negligible) overhead. If you’re still on 3.10, reach for as_completed() when you need streaming results and gather() when you don’t. And always set a timeout — unbounded concurrency is a recipe for resource exhaustion.