Deep Dive into Python asyncio & Event Loop Execution

Python’s asyncio module is standard for writing high-concurrency network servers and asynchronous applications. However, many developers treat the event loop as a black box.

In this article, we decompose how coroutines, futures, and tasks interact with Python’s event loop.

Coroutines vs. Generators

Before async/await syntax was introduced in Python 3.5, coroutines were built on top of generators using yield from. Underneath the syntax, an async def function returns a coroutine object.

import asyncio
import time

async def fetch_data(source_id: int) -> dict:
    print(f"[{time.strftime('%X')}] Starting fetch for source {source_id}")
    await asyncio.sleep(1.5)  # Yields control back to the event loop
    print(f"[{time.strftime('%X')}] Completed fetch for source {source_id}")
    return {"source": source_id, "status": "success", "data": [10, 20, 30]}

async def main():
    start_time = time.perf_counter()
    # Scheduling concurrent execution
    results = await asyncio.gather(
        fetch_data(1),
        fetch_data(2),
        fetch_data(3)
    )
    elapsed = time.perf_counter() - start_time
    print(f"All fetches completed in {elapsed:.2f} seconds.")

if __name__ == "__main__":
    asyncio.run(main())

The Anatomy of the Event Loop

The event loop manages an execution queue. When await asyncio.sleep(1.5) is called:

1. The coroutine suspends itself and yields a Future object back to the loop.
2. The event loop registers a timer callback via OS-level primitives (epoll on Linux, kqueue on macOS).
3. The event loop proceeds to run other ready tasks in its ready queue.
4. When the timer expires, the loop puts the task back on the ready queue and invokes .send() to resume execution.

Custom Task Scheduling in Python

We can directly inspect task execution by managing the loop manually:

import asyncio
from typing import Coroutine, Any

def run_concurrent_tasks(coroutines: list[Coroutine[Any, Any, Any]]):
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    
    try:
        tasks = [loop.create_task(coro) for coro in coroutines]
        grouped_tasks = asyncio.gather(*tasks)
        return loop.run_until_complete(grouped_tasks)
    finally:
        loop.close()

# Code Session Example
if __name__ == "__main__":
    async def worker(name: str, delay: float):
        await asyncio.sleep(delay)
        return f"Worker {name} finished"

    res = run_concurrent_tasks([worker("A", 0.5), worker("B", 0.2)])
    print("Results:", res)

Key Takeaways

• Never call blocking sync I/O (e.g., requests.get) inside async coroutines. Use aiohttp or run blocking calls in asyncio.to_thread().
• Task creation is light: Tasks wrap coroutines and allow immediate scheduling on the loop.
• Single-threaded concurrency: AsyncIO provides concurrency, not parallel CPU execution. Use multiprocessing for CPU-bound tasks.