Tasks, gather and TaskGroup
30 min · Intermediate
Need help? Ask a tutor on WhatsApp (opens in a new tab)One awaited call at a time is no faster than normal code. The gain comes when many waits overlap, and tasks are how you start them together.
create_task starts work now
Task is a coroutine that the event loop has scheduled to run. It starts running at the next await, and you await the task later to get its result.
Awaiting a coroutine runs it to the end before the next line. asyncio.create_task instead schedules it and returns at once, so several can wait together.
import asyncio
async def fetch(name): await asyncio.sleep(0.1) return name.upper()
async def main(): t1 = asyncio.create_task(fetch("a")) t2 = asyncio.create_task(fetch("b")) print(await t1, await t2)
asyncio.run(main())Await each task later to get its result.
Example 1 · Start two tasks, await both
import asyncio
async def lookup(code): await asyncio.sleep(0.05) return {"IN": "India", "US": "United States"}[code]
async def main(): first = asyncio.create_task(lookup("IN")) second = asyncio.create_task(lookup("US")) print(await first) print(await second)
asyncio.run(main())India United States
What this code does
- Both lookups start waiting as soon as the first
awaitruns. - We then collect them in a fixed order, so the output is fixed.
Keep a reference to every task you create. A task with no reference can be removed from memory before it finishes.
gather: run many, keep the order
gather runs several awaitables at the same time and returns their results in a list, in the order you passed them. It finishes when all of them have finished.
await asyncio.gather(a(), b(), c()) runs them all together and returns a list of results in the order you gave them, not the order they finished.
import asyncio
async def job(n): await asyncio.sleep(0.1) return n * n
async def main(): results = await asyncio.gather(*(job(n) for n in range(4))) print(results)
asyncio.run(main())The picture compares it with awaiting in a row.
Example 2 · gather keeps order even when jobs finish out of order
import asyncioimport time
async def job(n, delay): await asyncio.sleep(delay) return n
async def main(): start = time.perf_counter() results = await asyncio.gather(job(1, 0.3), job(2, 0.2), job(3, 0.1)) print(results) print("under 0.5 s:", time.perf_counter() - start < 0.5)
asyncio.run(main())[1, 2, 3] under 0.5 s: True
What this code does
- Job 3 finishes first, but
gatherlists results in the order the jobs were given. - The total is about 0.3 s, the longest wait. In a row it would be 0.6 s.
- We print only whether it was fast, because exact times vary.
* unpacks the generator into separate arguments. Three waits of one second take about one second with gather, not three.
When one task fails
return_exceptions is a gather option. When True, an error inside one task is returned as a value in the result list, so the other results are not lost.
By default, the first error in a gather is raised to the caller, and the other tasks keep running unseen.
async def main(): results = await asyncio.gather( job(1), bad_job(), job(3), return_exceptions=True, ) return results # [1, ValueError(...), 9]With return_exceptions=True, each error becomes an item in the result list.
Example 3 · Keep the good results with return_exceptions
import asyncio
async def parse(text): await asyncio.sleep(0.01) return int(text)
async def main(): results = await asyncio.gather(parse("4"), parse("x"), parse("6"), return_exceptions=True) print([type(r).__name__ if isinstance(r, Exception) else r for r in results])
asyncio.run(main())[4, 'ValueError', 6]
What this code does
- The bad text raises
ValueError, which gather stores in the list. - The list keeps the good results and the position of the failure.
Then check each item with isinstance(item, Exception). This keeps the good results when only some calls fail.
Limit how many run at once
Semaphore is a counter that limits how many tasks may be inside a block at once. A task waits for a free permit before entering and gives it back on leaving.
Starting 10,000 tasks at once can flood a server. A semaphore is a counter: a task must take a permit before it runs and returns it after.
sem = asyncio.Semaphore(3)
async def limited(n): async with sem: await asyncio.sleep(0.1) return nasync with sem: takes and returns the permit.
Example 4 · A semaphore caps the running tasks
import asyncio
running = 0most = 0
async def worker(sem): global running, most async with sem: running += 1 most = max(most, running) await asyncio.sleep(0.05) running -= 1
async def main(): sem = asyncio.Semaphore(2) await asyncio.gather(*(worker(sem) for _ in range(6))) print("most at once:", most)
asyncio.run(main())most at once: 2
What this code does
- Six workers start, but only two permits exist.
runningnever goes above 2, so the print is always 2.- No lock is needed for the counters: a switch can only happen at
await.
At most 3 limited calls are inside the block at any time. Create the semaphore inside the running loop, for example in main.
TaskGroup: tasks that live and fail together
TaskGroup is a scope that owns a set of tasks. When the async with block ends it has waited for every task, and one failure cancels the rest.
asyncio.TaskGroup (Python 3.11) starts tasks inside an async with block and waits for all of them at the end.
If one task fails, the others are cancelled and the errors are raised together.
Example 5 · A TaskGroup collects results
import asyncio
async def square(n): await asyncio.sleep(0.01) return n * n
async def main(): async with asyncio.TaskGroup() as tg: tasks = [tg.create_task(square(n)) for n in (1, 2, 3)] print([t.result() for t in tasks])
asyncio.run(main())[1, 4, 9]
What this code does
- The block ends only when all three tasks have finished.
- Each task's
result()is then ready to read, in the order we created them.
Prefer a TaskGroup for new code. It cannot forget a task, and it does not leave unseen errors behind the way a plain gather can.
Next step
Check your understanding
10 quick questions