asyncio.gather vs create_task vs as_completed: How to run multiple coroutines concurrently in Python

Apr 20, 2026

I had three async functions that I wanted to run at the same time, and I kept getting confused about whether to use asyncio.gather(), asyncio.create_task(), or asyncio.as_completed(). I tried gather() first, but my code blocked until all tasks finished. Then I tried create_task(), but my tasks kept getting cancelled unexpectedly. What’s the difference, and when should I use each one?

The Problem: Which Tool for Concurrent Execution?

I was building an async data pipeline that needed to fetch data from multiple sources. My initial attempt looked like this:

import asyncio

async def fetch_data(source):
    # Simulate network delay
    await asyncio.sleep(1)
    return f"Data from {source}"

async def main():
    # I thought this would run them concurrently...
    result1 = await fetch_data("API 1")
    result2 = await fetch_data("API 2")
    result3 = await fetch_data("API 3")
    print([result1, result2, result3])

asyncio.run(main())

This ran sequentially and took 3 seconds total. That’s not what I wanted. I needed them to run in parallel. But then I discovered there are three different ways to do this in asyncio, each with different semantics.

Solution 1: asyncio.gather() for Batch Results

When I need all results at once and want them in a deterministic order, asyncio.gather() is the right choice.

import asyncio
import time

async def coro(numbers):
    await asyncio.sleep(min(numbers))
    return list(reversed(numbers))

async def main():
    task1 = asyncio.create_task(coro([10, 5, 2]))  # sleeps 2 seconds
    task2 = asyncio.create_task(coro([3, 2, 1]))   # sleeps 1 second
    print("Start:", time.strftime("%X"))
    result = await asyncio.gather(task1, task2)
    print("End:", time.strftime("%X"))
    print(f"result: {result}")
    # Output: [[2, 5, 10], [1, 2, 3]]  # Order matches input order

asyncio.run(main())

Start: 10:30:00
End: 10:30:02
result: [[2, 5, 10], [1, 2, 3]]

The key insight: even though task2 finished first (it only sleeps 1 second vs 2), the results are returned in the order I passed them to gather(). Total time was ~2 seconds because both ran concurrently.

When to use gather():

You need all results before proceeding
Order of results matters (matches input order)
Simplest API for batch operations

Gotcha: gather() blocks until all tasks complete. If one task takes 10 seconds and others take 1 second, you wait 10 seconds before getting any results.

Solution 2: asyncio.create_task() for Background Execution

When I want tasks running in the background while I continue other work, create_task() is the tool. But I learned a painful lesson about awaiting.

import asyncio

async def background_work(name):
    await asyncio.sleep(1)
    print(f"{name} completed!")
    return f"Result from {name}"

async def main():
    # WRONG: task created but never awaited!
    asyncio.create_task(background_work("Task 1"))
    asyncio.create_task(background_work("Task 2"))
    print("Main function done")

asyncio.run(main())

Main function done

The tasks never completed! When the event loop ends, any unawaited tasks get cancelled. This was a confusing bug until I understood what was happening.

The correct approach:

import asyncio

async def background_work(name):
    await asyncio.sleep(1)
    print(f"{name} completed!")
    return f"Result from {name}"

async def main():
    # RIGHT: schedule and store reference, then await
    task1 = asyncio.create_task(background_work("Task 1"))
    task2 = asyncio.create_task(background_work("Task 2"))

    # Can do other work here while tasks run
    print("Tasks started, doing other work...")
    await asyncio.sleep(0.5)
    print("Other work done, now waiting for tasks")

    # Must await to get results and prevent cancellation
    result1 = await task1
    result2 = await task2
    print(f"Results: {result1}, {result2}")

asyncio.run(main())

Tasks started, doing other work...
Other work done, now waiting for tasks
Task 1 completed!
Task 2 completed!
Results: Result from Task 1, Result from Task 2

When to use create_task():

Tasks should overlap with other work
Complex workflows where tasks run in background
You need fine-grained control over task lifecycle

Key difference from direct await:

import asyncio

async def work():
    await asyncio.sleep(1)
    return "done"

async def main():
    # Direct await: runs immediately, pauses caller until done
    result1 = await work()

    # create_task: schedules for background, caller continues
    task = asyncio.create_task(work())
    print("Task running in background...")
    result2 = await task  # Now wait for result

Solution 3: asyncio.as_completed() for Streaming Results

When I need to process results as they finish rather than waiting for all to complete, as_completed() is the answer.

import asyncio
import time

async def fetch_with_delay(name, delay):
    await asyncio.sleep(delay)
    return f"{name} (took {delay}s)"

async def main():
    tasks = [
        asyncio.create_task(fetch_with_delay("Fast", 1)),
        asyncio.create_task(fetch_with_delay("Medium", 2)),
        asyncio.create_task(fetch_with_delay("Slow", 3)),
    ]

    print("Start:", time.strftime("%X"))
    for task in asyncio.as_completed(tasks):
        result = await task
        print(f"Got result: {result} at {time.strftime('%X')}")
    print("End:", time.strftime("%X"))

asyncio.run(main())

Start: 10:30:00
Got result: Fast (took 1s) at 10:30:01
Got result: Medium (took 2s) at 10:30:02
Got result: Slow (took 3s) at 10:30:03
End: 10:30:03

Notice the results come in fastest-first order, not input order. This is perfect for user-facing applications where showing partial results improves perceived performance.

When to use as_completed():

Process results incrementally as they arrive
User-facing apps needing responsiveness
When faster results should be handled first
Progress reporting or partial updates

Decision Matrix

After experimenting, I created this decision flowchart:

┌─────────────────────────────────────────────────────────────┐
│                    Need to run multiple                      │
│                      async functions?                        │
└────────────────────────┬────────────────────────────────────┘
                         │
                         ▼
         ┌───────────────────────────────┐
         │  Need results as they finish? │
         └───────────────┬───────────────┘
                  YES    │    NO
                         │
        ┌────────────────┴────────────────┐
        ▼                                  ▼
┌───────────────┐              ┌────────────────────┐
│ as_completed()│              │ Need all results   │
│               │              │ in input order?   │
└───────────────┘              └─────────┬──────────┘
                                YES       │     NO
                                           │
                          ┌────────────────┴────────────────┐
                          ▼                                 ▼
                 ┌───────────────┐              ┌────────────────┐
                 │   gather()    │              │  create_task() │
                 │               │              │  + manual await│
                 └───────────────┘              └────────────────┘

Common Mistakes I Made

Mistake 1: Forgetting to await create_task()

async def main():
    asyncio.create_task(some_coro())  # BUG: never awaited!
    # Task gets cancelled when main() exits

Fix: Always store the task and await it later, or use gather()/as_completed() to manage lifecycle.

Mistake 2: Using gather() for user-facing progress

I initially used gather() for a progress bar, but users saw nothing until all tasks completed. Switching to as_completed() let me update progress incrementally:

async def main():
    tasks = [asyncio.create_task(process_item(i)) for i in range(10)]

    completed = 0
    for task in asyncio.as_completed(tasks):
        await task
        completed += 1
        print(f"Progress: {completed}/10 items done")

Mistake 3: Mixing direct await with create_task inconsistently

async def main():
    # These are different!
    result1 = await coro()           # Runs immediately, blocks
    task = asyncio.create_task(coro())  # Schedules for later
    result2 = await task             # Waits for scheduled task

Both work, but they have different semantics. Direct await runs immediately and blocks. create_task() schedules for background execution.

Exception Handling Differences

Another important distinction: how each handles exceptions.

gather() with return_exceptions:

async def failing_task():
    raise ValueError("Oops!")

async def main():
    tasks = [asyncio.create_task(failing_task()),
             asyncio.create_task(asyncio.sleep(1, "success"))]

    # Default: exception propagates, other results lost
    # return_exceptions=True: exceptions returned as objects
    results = await asyncio.gather(*tasks, return_exceptions=True)
    for i, r in enumerate(results):
        if isinstance(r, Exception):
            print(f"Task {i} failed: {r}")
        else:
            print(f"Task {i} succeeded: {r}")

as_completed() propagates on await:

async def main():
    tasks = [asyncio.create_task(failing_task()),
             asyncio.create_task(asyncio.sleep(1, "success"))]

    for task in asyncio.as_completed(tasks):
        try:
            result = await task
            print(f"Success: {result}")
        except ValueError as e:
            print(f"Failed: {e}")

Performance Comparison

I ran benchmarks to compare the approaches:

Scenario: 10 tasks with delays 1-10 seconds

gather():        10s total (waits for slowest)
as_completed():  10s total (but results start at 1s)
create_task():   10s total (depends on how you await)

Key difference is NOT speed - it's WHEN you get results

The total execution time is similar. The difference is in when results become available and how you process them.

Beyond the Basics

Combining create_task with gather

I often use both together for complex workflows:

async def main():
    # Start background monitoring
    monitor = asyncio.create_task(monitor_system())

    # Run batch operations
    results = await asyncio.gather(
        fetch_users(),
        fetch_products(),
        fetch_orders()
    )

    # Do something with results while monitor continues
    process_results(results)

    # Clean up
    monitor.cancel()
    try:
        await monitor
    except asyncio.CancelledError:
        pass

Task Groups (Python 3.11+)

For Python 3.11+, TaskGroup provides structured concurrency with automatic cleanup:

async def main():
    async with asyncio.TaskGroup() as tg:
        task1 = tg.create_task(coro1())
        task2 = tg.create_task(coro2())
    # All tasks complete before exiting context
    # Exceptions are properly grouped and raised
    results = [task1.result(), task2.result()]

Summary

In this post, I explored the three main ways to run async tasks concurrently in Python. The key point is choosing based on your needs: asyncio.gather() for batch results in order, asyncio.create_task() for background execution with manual control, and asyncio.as_completed() for streaming results fastest-first.

Quick Reference:

Function	Returns	Order	Use Case
`gather()`	List of results	Input order	Batch operations
`create_task()`	Task object	N/A	Background work
`as_completed()`	Iterator	Completion order	Streaming results

Remember: always await tasks created with create_task(), or they’ll be cancelled when the event loop ends.

Final Words + More Resources

My intention with this article was to help others share my knowledge and experience. If you want to contact me, you can contact by email: Email me

Here are also the most important links from this article along with some further resources that will help you in this scope:

Oh, and if you found these resources useful, don’t forget to support me by starring the repo on GitHub!