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 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:

1. One loop per thread.
2. One loop active at a time (per thread).
3. 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() or loop.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’t await or gather() them.
• Loop shutdown occurred before tasks completed.
• You didn’t cancel running tasks properly.

🛠️ Fix Strategy

1. 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)

2. 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))

3. 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 of aiohttp).
• Your coroutine has no timeout control.
• Poor error handling causes silent hangs.

🛠️ Fix Strategy

1. Use Non-blocking Libraries
   • Use aiohttp instead of requests.
   • Use aiomysql or asyncpg instead of sync DB drivers.
2. Wrap coroutines with timeouts

pythonCopyEdittry:
    await asyncio.wait_for(some_async_function(), timeout=5)
except asyncio.TimeoutError:
    print("Task timed out.")

3. 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)

1. One Event Loop Per Thread. Always. Don’t share loops across threads unless you know what you’re doing.
2. Use asyncio.run() in Scripts Not inside servers, not inside interactive sessions.
3. Cancel Outstanding Tasks Before Shutdown Prevent “destroyed but pending” warnings and resource leaks.
4. Avoid time.sleep() Use await asyncio.sleep() in async functions.
5. 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

• Official Python AsyncIO Docs
• PEP 492 – Coroutines with async/await syntax
• FastAPI Async Guide

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.).