Async Pitfalls
asyncio code fails in distinctive ways — a forgotten await, one blocking call, or an unhandled task exception can silently freeze or corrupt a program that looks correct.
Cooperative concurrency trades the safety of the GIL-serialized synchronous world for a new set of hazards: nothing preempts your code, so one blocking call stalls everything; forgetting await produces a coroutine object instead of a bug-visible crash; and cancellation and exceptions inside background tasks can vanish silently. This topic catalogs the pitfalls that trip up asyncio code that otherwise looks correct, and how to guard against each.
Blocking the Event Loop
asyncio runs one event loop per thread, and coroutines cooperate by voluntarily yielding control at await points. If a coroutine calls a synchronous, CPU-bound, or blocking-I/O function without awaiting anything, it holds the event loop hostage for that entire call — every other task, timer, and incoming I/O event on that loop simply waits, no matter how many async def functions are queued.
other_task's 'tick' prints don't interleave with bad_task as expected — time.sleep freezes the entire loop, so all ticks appear only after 'done'.
The fix: use the async equivalent (await asyncio.sleep(...) instead of time.sleep(...), an async HTTP client instead of a blocking one), or if you must call blocking code, run it off the loop with await asyncio.to_thread(fn, *args) (3.9+) or loop.run_in_executor.
asyncio.to_thread runs the blocking function in a worker thread and awaits its result, keeping the event loop free.
import asyncio
import time
async def bad_task():
print("start")
time.sleep(2) # BLOCKS the whole event loop for 2 seconds
print("done")
async def other_task():
for i in range(3):
print("tick", i)
await asyncio.sleep(0.5)
async def main():
await asyncio.gather(bad_task(), other_task())
asyncio.run(main())import asyncio
def blocking_computation(n: int) -> int:
total = 0
for i in range(n):
total += i * i
return total
async def main():
result = await asyncio.to_thread(blocking_computation, 10_000_000)
print(result)
asyncio.run(main())Forgetting await
Calling a coroutine function without await doesn't run its body — it just constructs a coroutine object, which sits there unused. This is one of asyncio's most common mistakes because Python doesn't reject it at call time; the bug surfaces later, or not at all, as missing side effects.
This prints 'done' but never 'saved 42' — the coroutine object from save_to_db(42) is created and discarded without ever running.
CPython issues a RuntimeWarning: coroutine 'save_to_db' was never awaited when such an unused coroutine object is garbage-collected — always investigate that warning immediately; it usually points at exactly this bug.
import asyncio
async def save_to_db(value: int) -> None:
await asyncio.sleep(0.1)
print(f"saved {value}")
async def main():
save_to_db(42) # BUG: missing await — nothing happens yet
print("done")
asyncio.run(main())Task Cancellation Surprises
Cancelling a task (task.cancel()) does not stop it instantly — it schedules a CancelledError to be raised at the task's next suspension point (its current or next await). A task that never awaits again (because it's blocking, or catches and swallows the exception) never actually stops. Catching CancelledError and not re-raising it is a common bug: it suppresses cancellation and leaves the task running.
The except block runs cleanup but must re-raise CancelledError — swallowing it would leave the task's cancellation unacknowledged.
Since Python 3.11, asyncio.CancelledError derives from BaseException, not Exception — so a bare except Exception: no longer accidentally swallows it (though a bare except: still would).
import asyncio
async def stubborn_task():
try:
while True:
await asyncio.sleep(0.1)
except asyncio.CancelledError:
print("cleaning up")
raise # re-raise so cancellation actually propagates
async def main():
task = asyncio.create_task(stubborn_task())
await asyncio.sleep(0.3)
task.cancel()
try:
await task
except asyncio.CancelledError:
print("task confirmed cancelled")
asyncio.run(main())Exceptions in Background Tasks Go Silent
A task created with asyncio.create_task() runs independently. If it raises and nothing ever awaits it (or checks its result), the exception is stored on the task object but never re-raised anywhere visible — the program can keep running as if nothing failed, and the exception only surfaces (as a logged warning) when the task object is garbage-collected.
The ValueError from flaky() is never raised into main's control flow — it only shows up as an unhandled-exception log message when garbage collection notices, easy to miss in real programs.
Fix: always eventually await tasks you create (directly, or via asyncio.gather/asyncio.TaskGroup), or attach a done-callback that checks task.exception(). Python 3.11's asyncio.TaskGroup is the modern preferred structure — it awaits all child tasks and re-raises their failures (wrapped in an ExceptionGroup if more than one fails) automatically when the async with block exits.
import asyncio
async def flaky():
await asyncio.sleep(0.1)
raise ValueError("something went wrong")
async def main():
task = asyncio.create_task(flaky())
await asyncio.sleep(0.5)
print("main finished, error above is easy to miss")
asyncio.run(main())import asyncio
async def flaky():
await asyncio.sleep(0.1)
raise ValueError("something went wrong")
async def main():
async with asyncio.TaskGroup() as tg:
tg.create_task(flaky())
print("unreachable if flaky fails")
try:
asyncio.run(main())
except* ValueError as eg:
print("caught:", eg.exceptions)Fire-and-Forget Tasks Getting Garbage Collected
asyncio only holds a *weak* reference to tasks it isn't otherwise tracking. If you call asyncio.create_task(coro()) without keeping a strong reference to the returned task object (assigning it to a variable, adding it to a set), the task can be garbage-collected mid-flight and silently stop running.
Keeping tasks in a set (and discarding them on completion) prevents premature garbage collection of fire-and-forget work.
# risky: no reference kept
asyncio.create_task(background_job()) # may be GC'd before it finishes
# safer: keep a strong reference
background_tasks: set[asyncio.Task] = set()
def spawn(coro):
task = asyncio.create_task(coro)
background_tasks.add(task)
task.add_done_callback(background_tasks.discard)
return taskShared Mutable State and Race Conditions
Because coroutines only switch at await points (never in the middle of a plain synchronous statement), asyncio code is less prone to fine-grained interleaving bugs than preemptive threads — but it is not immune. Two tasks that each do 'read state, await something, write state' can still race: the state can change during the await between the read and the write.
Both tasks read balance=100 before either awaits, then both write back based on that stale value — one withdrawal effectively gets lost.
Fix with an asyncio.Lock around the read-modify-write critical section so only one task executes it at a time between await points.
import asyncio
balance = 100
async def withdraw(amount: int) -> None:
global balance
current = balance
await asyncio.sleep(0.01) # simulated I/O — another task can run here
balance = current - amount # stale 'current' overwrites concurrent changes
async def main():
global balance
await asyncio.gather(withdraw(30), withdraw(50))
print("balance:", balance) # not necessarily 20 — a race
asyncio.run(main())lock = asyncio.Lock()
async def withdraw(amount: int) -> None:
global balance
async with lock:
current = balance
await asyncio.sleep(0.01)
balance = current - amountIncorrectly Mixing Sync and Async Code
Calling asyncio.run() from inside code that's already running an event loop raises RuntimeError: asyncio.run() cannot be called from a running event loop — a common mistake in notebooks, GUI callbacks, or nested async helpers. Conversely, calling a coroutine function and expecting it to just run synchronously (without await or an event loop) does nothing but build an unused coroutine object (see 'Forgetting await' above).
You also cannot directly await inside a plain synchronous function — await is only legal inside async def bodies. Bridging sync and async code requires an explicit boundary: asyncio.run() at the top level, asyncio.to_thread/run_in_executor to call blocking sync code from async code, or a library-provided sync wrapper.
# BUG: calling asyncio.run() from within an already-running loop
async def outer():
asyncio.run(inner()) # RuntimeError
# fix: just await it directly, no new loop needed
async def outer_fixed():
await inner()Practical Debugging Tips
Run with asyncio.run(main(), debug=True) (or set the PYTHONASYNCIODEBUG=1 environment variable) during development — debug mode logs slow callbacks that block the loop, warns about non-awaited coroutines more aggressively, and flags tasks destroyed while still pending.
Treat every RuntimeWarning: coroutine ... was never awaited as a real bug, not noise — it almost always means a missing await.
Prefer asyncio.TaskGroup (3.11+) or asyncio.gather(..., return_exceptions=False) over bare create_task calls you never join — structured concurrency makes exception propagation and cancellation automatic instead of something you must remember.
Best Practices Summary
Never call blocking I/O or CPU-heavy synchronous code directly inside a coroutine; offload it with asyncio.to_thread or a process pool.
Always keep a strong reference to tasks you create, and always eventually await or otherwise observe their result/exception.
Re-raise CancelledError after any cleanup inside an except block that catches it.
Guard genuinely shared mutable state with asyncio.Lock around any critical section that spans an await.
Prefer structured concurrency (TaskGroup, gather) over unmanaged fire-and-forget tasks.
Complete Example: Fixing a Broken Pipeline
Every coroutine call is awaited, blocking work is avoided, and TaskGroup ensures no task's failure is silently lost.
import asyncio
async def fetch(url: str) -> str:
await asyncio.sleep(0.1) # simulated network I/O, never blocking
return f"data from {url}"
async def process(url: str) -> None:
try:
result = await fetch(url) # awaited — no forgotten-await bug
print(result)
except Exception as exc:
print(f"failed to process {url}: {exc}")
async def main():
urls = ["a.example", "b.example", "c.example"]
async with asyncio.TaskGroup() as tg: # structured: exceptions propagate
for url in urls:
tg.create_task(process(url))
print("all fetches complete")
asyncio.run(main())Mental Model
Think of the event loop as a single chef who only ever works on one dish at a time and switches dishes only at points you explicitly mark with 'put this in the oven and check back' (await). If a task never says that — it just keeps chopping — the chef is stuck on it and every other dish burns. Every pitfall in this topic is really one root cause: something broke the promise of yielding control back promptly and honestly.
Quick Reference
Pitfall | Fix
--------------------------------|---------------------------------------------
blocking call in a coroutine | await asyncio.to_thread(fn, ...)
missing await | check RuntimeWarning: never awaited
task exception lost | await the task / use TaskGroup or gather
task cancelled but keeps running| re-raise CancelledError after cleanup
fire-and-forget task GC'd early | keep a strong reference (e.g. a set)
race on shared state | asyncio.Lock around the critical section
asyncio.run() inside a loop | just await the coroutine directlyKey Takeaways
asyncio concurrency is cooperative — one blocking call anywhere stalls the entire event loop, not just that task.
Calling a coroutine function without await silently does nothing but create an unused coroutine object; watch for the never-awaited warning.
Cancellation raises CancelledError at the next suspension point; catch it only to clean up, then re-raise it.
Unobserved task exceptions do not crash the program — they vanish unless the task is awaited, gathered, or grouped with TaskGroup.
Shared mutable state can still race across an await; protect multi-step critical sections with asyncio.Lock.
Bridge sync and async code deliberately at clear boundaries — never call asyncio.run() from inside a running loop.
async def handler(): time.sleep(1) # blocks the whole event loop async def handler(): await asyncio.sleep(1) # or asyncio.to_thread(...) for real blocking work time.sleep is a blocking call with no await point — it freezes every other task on the event loop, not just this coroutine.
async def main(): fetch_data() # coroutine object created, never run async def main(): await fetch_data()
Without await, the coroutine object is built and immediately discarded — its body never executes.
asyncio.create_task(background_job()) await asyncio.sleep(5) # task may vanish before completing (weak ref) task = asyncio.create_task(background_job()) tasks.add(task) task.add_done_callback(tasks.discard) await asyncio.sleep(5) asyncio only weakly references tasks it isn't tracking elsewhere; without a strong reference, the task can be garbage-collected mid-run.
try: await asyncio.sleep(10) except asyncio.CancelledError: pass # swallowed — cancellation is silently ignored try: await asyncio.sleep(10) except asyncio.CancelledError: cleanup() raise
Catching CancelledError without re-raising it defeats cancellation — the task's caller believes it stopped, but it may keep running.
Why does one blocking call inside a coroutine freeze unrelated tasks on the same event loop?
Because asyncio is cooperative single-threaded concurrency — a coroutine only yields control at await points; a blocking call has none, so it holds the loop's single thread until it returns.
What actually happens when you call a coroutine function without await?
It builds and returns a coroutine object without running any of its body; nothing happens until that object is awaited or otherwise driven, and Python may later warn it was never awaited.
Why is catching CancelledError without re-raising it a bug?
It suppresses the cancellation signal — the caller that called task.cancel() and awaited the task believes cancellation succeeded, but the task's loop can keep running.
Why can an exception raised inside a fire-and-forget task go completely unnoticed?
A task's exception is only re-raised when something awaits the task (or calls .result()/.exception()); an unobserved task just stores the exception and the program continues.
How can two coroutines race on a shared variable even though asyncio is single-threaded?
Because control can switch to another task at any await point; if a coroutine reads shared state, awaits, then writes it back, another task can change that state in between, causing a lost update.
What does asyncio.to_thread do and when should you use it?
It runs a blocking synchronous function in a worker thread and returns an awaitable for its result, letting you call blocking or CPU-bound code from a coroutine without freezing the event loop.
Why does calling asyncio.run() inside code already running an event loop raise an error?
asyncio.run() creates and manages its own fresh event loop and refuses to nest one inside an already-running loop; the fix is to just await the coroutine directly in the existing loop.
What does this print, and what happens to the RuntimeError?
It prints only 'main done'. The RuntimeError from worker() is never raised anywhere main can see — the task is never awaited — so it's silently lost (CPython logs an 'exception was never retrieved' message when the task is later garbage-collected, but the program doesn't crash).
Rewrite this buggy snippet, which is meant to fetch two URLs concurrently, so both coroutines actually run and any failure is raised to the caller: async def main():\n fetch('a')\n fetch('b') async def main(): async with asyncio.TaskGroup() as tg: tg.create_task(fetch("a")) tg.create_task(fetch("b"))
The original code never awaits fetch(), so neither coroutine runs. Using asyncio.TaskGroup (or asyncio.gather(fetch('a'), fetch('b'))) actually schedules and awaits both, and TaskGroup automatically propagates any exception raised by either task.
Write an async function safe_withdraw(amount) that decrements a module-level balance protected by an asyncio.Lock, simulating an await between the read and the write, so concurrent calls never lose an update.
import asyncio balance = 100 lock = asyncio.Lock() async def safe_withdraw(amount: int) -> None: global balance async with lock: current = balance await asyncio.sleep(0.01) balance = current - amount
The asyncio.Lock ensures only one call executes the read-await-write sequence at a time; a second concurrent call waits for the lock before reading balance, so it always sees the up-to-date value and no update is lost.
import asyncio
async def worker():
await asyncio.sleep(0.1)
raise RuntimeError("boom")
async def main():
asyncio.create_task(worker())
await asyncio.sleep(0.3)
print("main done")
asyncio.run(main())