Your cart is currently empty!
AsyncIO is deceptively simple—until it isn’t. You’re probably here because you hit one of these maddening errors:
RuntimeError: This event loop is already running
RuntimeError: Event loop is closed
Task was destroyed but it is pending!
This isn’t another surface-level tutorial. This is the in-depth guide on how to debug Python AsyncIO, specifically event loop-related issues. If you’re building FastAPI services, automation pipelines, or asynchronous scrapers, these insights will save you hours of debugging.
Table of Contents
- 🚀 TL;DR – Quick Fix Table
- The Core Problem: AsyncIO Event Loops Are Easy to Mismanage
- The Golden Rule: There Should Be One Loop to Rule Them All
- 🔥 The Most Common AsyncIO Event Loop Errors (And How to Debug Them)
- 💡 Debugging AsyncIO in Depth
- ✅ AsyncIO Best Practices (For Clean Event Loop Management)
- 🔗 External Resources
- TL;DR
🚀 TL;DR – Quick Fix Table
Error Message | Cause | Solution |
---|---|---|
RuntimeError: Event loop is closed | Loop closed but still accessed | Use asyncio.run() ; never reuse closed loops |
RuntimeError: This event loop is already running | Nested loops or conflicting runtime | Use nest_asyncio in Jupyter; avoid asyncio.run() inside FastAPI or ASGI apps |
Task was destroyed but it is pending! | Tasks left pending without proper shutdown | Always await tasks or explicitly cancel and gather them on shutdown |
Hanging coroutines / timeouts | Blocking sync calls or missing timeouts | Use asyncio.wait_for() ; never call sync code in async functions |
The Core Problem: AsyncIO Event Loops Are Easy to Mismanage
The event loop orchestrates coroutine execution. When you:
- Mismanage its lifecycle (starting/stopping)
- Mix blocking (sync) code with non-blocking (async) code
- Use conflicting frameworks (Jupyter, FastAPI, Celery)
You trigger event loop chaos. Fixing it requires understanding how loops work and how they fail.

The Golden Rule: There Should Be One Loop to Rule Them All
➡️ AsyncIO Basics Refresher
asyncio.run(coro())
: Starts the loop, runs the coroutine, cleans up.loop.run_until_complete(coro())
: Old-school; more control, but riskier.loop.create_task(coro())
: Schedules a coroutine inside a running loop.
AsyncIO expects:
- One loop per thread.
- One loop active at a time (per thread).
- Proper shutdown (cancel tasks, close loop).
🔥 The Most Common AsyncIO Event Loop Errors (And How to Debug Them)
1. RuntimeError: Event loop is closed
✅ What’s Happening
You’re trying to schedule or run a coroutine after the event loop has been explicitly or implicitly closed.
✅ Why It Happens
- You manually closed the loop with
loop.close()
. - An exception triggered shutdown prematurely.
- Cleanup ran before all tasks finished.
🛠️ Fix Strategy
Modern Python (3.7+): Use asyncio.run()
. It:
- Creates a loop.
- Runs your coroutine.
- Cancels remaining tasks.
- Closes the loop automatically.
pythonCopyEditasync def main():
await some_async_function()
asyncio.run(main()) # Handles the loop lifecycle safely
Legacy Python / Complex Apps:
- Manage lifecycle carefully.
- Cancel pending tasks before closing.
pythonCopyEditloop = asyncio.get_event_loop()
try:
loop.run_until_complete(main())
finally:
pending = asyncio.all_tasks(loop)
for task in pending:
task.cancel()
loop.run_until_complete(asyncio.gather(*pending, return_exceptions=True))
loop.close()
⚠️ Never reuse a closed loop. If you closed it, create a new loop (advanced use case, rarely necessary).
2. RuntimeError: This event loop is already running
✅ What’s Happening
You called asyncio.run()
(or loop.run_until_complete()
) on a loop that’s already running.
✅ Why It Happens
- Jupyter Notebook / IPython automatically runs an event loop.
- You nested
asyncio.run()
orloop.run_until_complete()
calls. - Third-party libraries (FastAPI, Starlette) already manage the event loop.
🛠️ Fix Strategy
Jupyter / IPython
Patch the loop with nest_asyncio
.
bashCopyEditpip install nest_asyncio
pythonCopyEditimport nest_asyncio
nest_asyncio.apply()
# Now you can run asyncio.run() without errors (not best practice, but workable)
Web Frameworks / Runtimes (FastAPI, Starlette, etc.)
- Never call
asyncio.run()
inside a handler. - Use the existing loop:
await your_coro()
directly.
Example (FastAPI Background Tasks)
pythonCopyEditfrom fastapi import FastAPI, BackgroundTasks
app = FastAPI()
async def expensive_async_job():
await asyncio.sleep(5)
@app.post("/run-task/")
async def run_task(background_tasks: BackgroundTasks):
background_tasks.add_task(expensive_async_job)
return {"message": "Task scheduled"}
Manual Control (Advanced)
pythonCopyEditloop = asyncio.get_event_loop()
# Check if it's running
if loop.is_running():
# Don't call run_until_complete here
task = loop.create_task(some_async_function())
else:
loop.run_until_complete(some_async_function())

3. Task was destroyed but it is pending!
✅ What’s Happening
A coroutine was still pending when the loop shut down or when it went out of scope without being awaited.
✅ Why It Happens
- You created tasks with
create_task()
but didn’tawait
orgather()
them. - Loop shutdown occurred before tasks completed.
- You didn’t cancel running tasks properly.
🛠️ Fix Strategy
- Always await tasks or use
asyncio.gather()
.
pythonCopyEditasync def main():
tasks = [asyncio.create_task(my_coro(i)) for i in range(5)]
await asyncio.gather(*tasks)
- On shutdown, cancel and clean up.
pythonCopyEdittasks = asyncio.all_tasks(loop)
for task in tasks:
task.cancel()
loop.run_until_complete(asyncio.gather(*tasks, return_exceptions=True))
- Return Exceptions for pending tasks:
pythonCopyEditresults = await asyncio.gather(*tasks, return_exceptions=True)
for result in results:
if isinstance(result, Exception):
handle_exception(result)
4. Hanging Coroutines / Timeouts
✅ What’s Happening
An await
is hanging because the underlying coroutine never finishes—common in bad I/O calls or when using sync libraries inside async functions.
✅ Why It Happens
- You accidentally used a blocking function (
requests.get()
instead ofaiohttp
). - Your coroutine has no timeout control.
- Poor error handling causes silent hangs.
🛠️ Fix Strategy
- Use Non-blocking Libraries
- Use
aiohttp
instead ofrequests
. - Use
aiomysql
orasyncpg
instead of sync DB drivers.
- Use
- Wrap coroutines with timeouts
pythonCopyEdittry:
await asyncio.wait_for(some_async_function(), timeout=5)
except asyncio.TimeoutError:
print("Task timed out.")
- Cancel stuck tasks
pythonCopyEdittask = asyncio.create_task(some_async_function())
try:
await asyncio.wait_for(task, timeout=5)
except asyncio.TimeoutError:
task.cancel()
await task
💡 Debugging AsyncIO in Depth
1. Debug Mode (Built-In)
Enabling debug mode gives more verbose logs on the event loop and coroutine execution.
pythonCopyEditasyncio.run(main(), debug=True)
Or:
pythonCopyEditloop = asyncio.get_event_loop()
loop.set_debug(True)
2. Use Logging (Not print)
pythonCopyEditimport logging
logging.basicConfig(level=logging.DEBUG)
logger = logging.getLogger(__name__)
logger.debug("Running async task")
3. Visualize Tasks
Use asyncio.all_tasks()
to inspect what’s pending:
pythonCopyEdittasks = asyncio.all_tasks(loop)
for t in tasks:
print(t)
✅ AsyncIO Best Practices (For Clean Event Loop Management)
- One Event Loop Per Thread. Always. Don’t share loops across threads unless you know what you’re doing.
- Use
asyncio.run()
in Scripts Not inside servers, not inside interactive sessions. - Cancel Outstanding Tasks Before Shutdown Prevent “destroyed but pending” warnings and resource leaks.
- Avoid
time.sleep()
Useawait asyncio.sleep()
inasync
functions. - Batch Async Tasks Don’t launch 10k tasks without
gather()
or a semaphore to limit concurrency.
❓ What does “RuntimeError: Event loop is closed” mean?
You’re trying to schedule or run tasks on a loop that’s already shut down. Use asyncio.run()
to manage loop lifecycle safely.
❓ How do I fix “This event loop is already running” in Jupyter?
Patch the loop with nest_asyncio
and avoid using asyncio.run()
; directly await
functions.
❓ Can AsyncIO run parallel tasks?
AsyncIO runs concurrent coroutines in one thread. For CPU-bound parallelism, use loop.run_in_executor()
or multiprocessing
.
pythonCopyEditawait loop.run_in_executor(None, blocking_function)
❓ How do I cancel tasks properly in AsyncIO?
Gather all tasks via asyncio.all_tasks()
, cancel them, and await their shutdown before closing the loop.
🔗 External Resources
TL;DR
- Use
asyncio.run()
for simple apps. - Never nest event loops.
- Cancel and clean up tasks on shutdown.
- Debug mode and logging are your friends.
- Understand your runtime’s event loop policy (Jupyter, FastAPI, etc.).
Staff picks
-
$21.00
Colorful Fox Tee
-
$21.00
Tech Cat Unisex Tee
-
$21.00
Bear Graphic Tee
-
$21.00
Sloth Tee
-
$21.00
Hedgehog Sunglasses Tee
-
$21.00
Squirrel Sunglasses Tee
-
Debugging Circular Imports in Python: Clean Project Layout
Circular imports are one of the most common, annoying, and avoidable problems…
-
Debugging Python AsyncIO Errors: Event Loop Problems Solved
AsyncIO is deceptively simple—until it isn’t. You’re probably here because you hit…
-
How to Fix Python Memory Leaks With tracemalloc
Struggling with a Python app that keeps eating up memory? Learn how…
-
Fixing “ModuleNotFoundError” in Python (Fast Debugging Guide)
Struggling with Python’s dreaded ModuleNotFoundError? This fast debugging guide covers exactly why…
-
How to Resolve ImportErrors and ModuleNotFoundErrors in Python Projects
Struggling with Python import errors? Learn how to fix ImportError and ModuleNotFoundError…
-
Stop Writing Python Like JavaScript – Common Mistakes and How to Fix Them
Python and JavaScript are not the same, and yet, I keep seeing…
Leave a Reply