Introduction

The RuntimeError: Event loop is closed error in Python asyncio occurs when code attempts to schedule a coroutine or task on an event loop that has already been shut down. This is especially common in long-running services, background workers, and when using libraries like aiohttp or asyncpg that maintain persistent connections. The error typically appears during application shutdown, test teardown, or when mixing synchronous and asynchronous code incorrectly.

Symptoms

You see errors like this in your logs:

bash
RuntimeError: Event loop is closed

Or during shutdown:

bash
Task was destroyed but it is pending!
task: <Task pending name='Task-2' coro=<Connection._listen() running at /usr/local/lib/python3.11/site-packages/asyncpg/pool.py:329>>
sys:1: RuntimeWarning: coroutine 'Connection._listen' was never awaited
RuntimeError: Event loop is closed

Or when using aiohttp:

bash
Unclosed client session
client_session: <aiohttp.client.ClientSession object at 0x7f9c2c1a3d90>
RuntimeError: Event loop is closed

The application may appear to work correctly during normal operation but fails during graceful shutdown, leaving resources leaked.

Common Causes

  • **Calling loop.close() while tasks are still running**: Manually closing the event loop before awaiting all pending tasks
  • **Using asyncio.get_event_loop() in Python 3.10+**: This deprecated function returns a closed loop after asyncio.run() completes
  • **Mixing loop.run_until_complete() with asyncio.run()**: Using both patterns in the same codebase causes loop lifecycle conflicts
  • Not awaiting cleanup in shutdown handlers: Async resources like database pools or HTTP clients not being closed before the loop shuts down
  • Background tasks still running during shutdown: Tasks spawned with asyncio.create_task() that outlive the main coroutine
  • Test framework teardown closing the loop prematurely: pytest-asyncio closing the loop while fixtures still have cleanup code

Step-by-Step Fix

Step 1: Use `asyncio.run()` as your single entry point

Replace manual event loop management with asyncio.run(), which handles the full lifecycle:

```python import asyncio

async def main(): async with aiohttp.ClientSession() as session: async with asyncpg.create_pool( host="localhost", database="myapp", user="postgres", password="secret", ) as pool: # Your application logic here await process_tasks(session, pool)

# Correct entry point - manages loop creation and cleanup if __name__ == "__main__": asyncio.run(main()) ```

Never use loop = asyncio.get_event_loop(); loop.run_until_complete(main()); loop.close() in Python 3.10+.

Step 2: Properly handle shutdown with signal handlers

For long-running services, use signal handlers to trigger graceful shutdown:

```python import asyncio import signal

async def main(): loop = asyncio.get_running_loop() shutdown_event = asyncio.Event()

def signal_handler(): print("Received shutdown signal, draining tasks...") shutdown_event.set()

for sig in (signal.SIGTERM, signal.SIGINT): loop.add_signal_handler(sig, signal_handler)

# Your main application loop while not shutdown_event.is_set(): try: await asyncio.wait_for( process_next_item(), timeout=1.0 ) except asyncio.TimeoutError: continue

# Graceful cleanup - close all async resources await cleanup_resources() print("Shutdown complete")

asyncio.run(main()) ```

Step 3: Cancel and await background tasks before shutdown

If you have background tasks, gather them and cancel before closing:

```python async def main(): background_tasks = set()

for i in range(5): task = asyncio.create_task(background_worker(i)) background_tasks.add(task) task.add_done_callback(background_tasks.discard)

try: await main_processing() finally: # Cancel all background tasks for task in background_tasks: task.cancel()

# Wait for them to finish their cancellation if background_tasks: await asyncio.gather(*background_tasks, return_exceptions=True)

asyncio.run(main()) ```

The finally block ensures cleanup happens even if the main coroutine raises an exception.

Prevention

  • Use asyncio.run() exclusively as the entry point, never manage loops manually
  • Always use async context managers (async with) for resources that need cleanup
  • Register all background tasks in a collection so they can be cancelled during shutdown
  • Use pytest-asyncio with @pytest.mark.asyncio for tests, not manual loop management
  • Add sys.excepthook logging to catch unhandled exceptions that silently close loops
  • Monitor for "Task was destroyed but it is pending" warnings in CI output