13.6 Coroutines and Event Loops in Python

NicePython

— 5 min read

In Python's asynchronous programming, coroutines and the event loop are core concepts that enable efficient multitasking within a single thread. Coroutines are special functions that can be paused and resumed at certain points, while the event loop manages the execution of these coroutines, scheduling them to run one at a time, allowing other tasks to run during I/O waits. Together, they enable non-blocking and concurrent execution of tasks.

In this section, we will explore how coroutines and event loops work, how they interact in the asyncio framework, and how to effectively use them to write efficient, asynchronous code.

13.6.1 Coroutines

A coroutine is a function that can be paused and resumed during its execution. Unlike regular functions that return a value and exit, coroutines use yield or await to pause execution, allowing other tasks to run. Coroutines are defined using the async def syntax and must be awaited to run.

Defining a Coroutine

To define a coroutine, use the async def syntax:

import asyncio

# Define an asynchronous coroutine
async def my_coroutine():
    print("Start")
    await asyncio.sleep(1)  # Simulate a non-blocking delay
    print("End")

Here:

  • async def defines the coroutine.
  • await pauses the coroutine's execution, allowing the event loop to run other tasks.

Running a Coroutine

A coroutine doesn't execute immediately like a regular function. You need to schedule it on the event loop using asyncio.run() or by wrapping it in a task.

# Run the coroutine using asyncio.run()
asyncio.run(my_coroutine())

Output:

Start
(1 second delay)
End

In this example:

  • asyncio.run() is used to execute the coroutine and run it until completion.

13.6.2 The Event Loop

The event loop is the central mechanism in asyncio responsible for managing and scheduling the execution of asynchronous tasks (coroutines, I/O operations, etc.). It runs in a loop, checking for tasks that are ready to execute, running them, and then suspending them when they hit an await point or need to wait for I/O operations.

How the Event Loop Works:

  1. Register: Coroutines or I/O tasks are registered with the event loop.
  2. Run: The event loop runs, executing the coroutines until they reach an await point.
  3. Pause/Resume: When a coroutine reaches an await point, the event loop pauses it and resumes other coroutines that are ready to continue.
  4. Complete: When all coroutines have completed, the event loop stops.

Example: Event Loop in Action

import asyncio

async def task1():
    print("Task 1 started")
    await asyncio.sleep(2)  # Simulate I/O
    print("Task 1 completed")

async def task2():
    print("Task 2 started")
    await asyncio.sleep(1)  # Simulate I/O
    print("Task 2 completed")

async def main():
    # Run two coroutines concurrently
    await asyncio.gather(task1(), task2())

# Run the main coroutine, which triggers the event loop
asyncio.run(main())

Output:

Task 1 started
Task 2 started
(1 second later)
Task 2 completed
(1 second later)
Task 1 completed

In this example:

  • The event loop runs task1 and task2 concurrently. Even though task1 takes 2 seconds, task2 completes after 1 second because the event loop efficiently switches between tasks.

13.6.3 await and Non-blocking I/O

The await keyword is used to pause a coroutine and wait for an awaitable (another coroutine, task, or I/O operation) to finish. While the coroutine is paused, the event loop can run other coroutines.

Example: Awaiting an Asynchronous Operation

import asyncio

async def fetch_data():
    print("Fetching data...")
    await asyncio.sleep(2)  # Simulate a network request or I/O
    print("Data fetched!")
    return "data"

async def process_data():
    data = await fetch_data()  # Wait for fetch_data to complete
    print(f"Processing {data}")

asyncio.run(process_data())

Output:

Fetching data...
(2 seconds later)
Data fetched!
Processing data

In this example:

  • The fetch_data() coroutine is awaited, pausing process_data() until the simulated I/O completes.

13.6.4 Creating and Managing Tasks

In asyncio, you can schedule coroutines to run concurrently by creating tasks. A task is a coroutine that has been scheduled on the event loop. Tasks run independently, and you can use them to perform multiple operations concurrently.

Creating Tasks with asyncio.create_task()

import asyncio

async def task1():
    print("Task 1 started")
    await asyncio.sleep(2)
    print("Task 1 completed")

async def task2():
    print("Task 2 started")
    await asyncio.sleep(1)
    print("Task 2 completed")

async def main():
    # Create two tasks that run concurrently
    task_1 = asyncio.create_task(task1())
    task_2 = asyncio.create_task(task2())

    # Wait for both tasks to complete
    await task_1
    await task_2

asyncio.run(main())

Output:

Task 1 started
Task 2 started
(1 second later)
Task 2 completed
(1 second later)
Task 1 completed

In this example:

  • asyncio.create_task() schedules the coroutines to run concurrently, and the event loop handles their execution.
  • Both tasks are created, and await task_1 and await task_2 ensure that the main coroutine waits for them to finish.

13.6.5 Managing the Event Loop

In addition to using asyncio.run(), you can manage the event loop directly to control when tasks start, stop, or are canceled. This is useful in more complex applications where you need to manage coroutines manually.

Example: Accessing the Event Loop

import asyncio

async def my_coroutine():
    print("Running a coroutine")
    await asyncio.sleep(1)
    print("Coroutine finished")

# Get the current event loop
loop = asyncio.get_event_loop()

# Schedule the coroutine to run
loop.run_until_complete(my_coroutine())

# Close the loop after running the coroutine
loop.close()

In this example:

  • asyncio.get_event_loop() returns the current event loop, and run_until_complete() runs the specified coroutine.
  • This method is used in more advanced applications where you need finer control over the event loop, though asyncio.run() is generally preferred for most use cases.

13.6.6 Running Tasks Concurrently with asyncio.gather()

asyncio.gather() is used to run multiple coroutines concurrently and wait for all of them to complete. It allows you to efficiently manage tasks without having to manually create tasks and manage them individually.

Example: Using asyncio.gather()

import asyncio

async def task1():
    print("Task 1 started")
    await asyncio.sleep(2)
    print("Task 1 completed")

async def task2():
    print("Task 2 started")
    await asyncio.sleep(1)
    print("Task 2 completed")

async def main():
    # Run two tasks concurrently using asyncio.gather
    await asyncio.gather(task1(), task2())

# Run the main coroutine
asyncio.run(main())

In this example:

  • asyncio.gather() runs multiple coroutines concurrently and waits for both tasks to finish.

13.6.7 Handling Timeouts with asyncio.wait_for()

asyncio.wait_for() sets a maximum time limit for a coroutine to complete. If the coroutine takes longer than the specified time, a TimeoutError is raised.

Example: Using asyncio.wait_for()

import asyncio

async def slow_task():
    print("Starting slow task...")
    await asyncio.sleep(5)  # Simulate a long-running task
    print("Slow task finished")

async def main():
    try:
        # Set a timeout of 3 seconds
        await asyncio.wait_for(slow_task(), timeout=3)
    except asyncio.TimeoutError:
        print("Task timed out!")

# Run the main coroutine
asyncio.run(main())

Output:

Starting slow task...
Task timed out!

In this example:

  • asyncio.wait_for() ensures that if the slow_task() coroutine takes more than 3 seconds, it will raise a TimeoutError.

13.6.8 Working with Asynchronous Generators

An asynchronous generator allows you to yield values asynchronously. Asynchronous generators are defined using async def and

the yield keyword, and they are iterated using async for.

Example: Asynchronous Generator

import asyncio

# Define an asynchronous generator
async def async_generator():
    for i in range(5):
        await asyncio.sleep(1)  # Simulate async work
        yield i

# Main coroutine
async def main():
    # Use async for to iterate over the async generator
    async for value in async_generator():
        print(value)

# Run the main coroutine
asyncio.run(main())

Output:

0
1
2
3
4

In this example:

  • async_generator() is an asynchronous generator that yields values every second.
  • async for is used to iterate over the values asynchronously.

13.6.9 Summary

  • Coroutines: Special functions defined with async def that can be paused and resumed using await.
  • Event Loop: The mechanism that schedules and manages the execution of coroutines, ensuring that tasks are run concurrently without blocking each other.
  • await: Used to pause a coroutine and allow other tasks to run while waiting for an I/O operation or another coroutine to complete.
  • Tasks: Units of work that are scheduled on the event loop, allowing multiple coroutines to run concurrently.
  • asyncio.gather(): A convenient way to run multiple coroutines concurrently.
  • asyncio.create_task(): Used to schedule a coroutine to run as a background task.
  • Timeouts: asyncio.wait_for() can be used to impose time limits on the execution of coroutines.
  • Asynchronous Generators: Enable the yielding of values asynchronously, with async for used for iteration.

By mastering coroutines and the event loop, you can write highly efficient Python programs that handle many tasks concurrently, without the need for multiple threads or processes, making them ideal for I/O-bound and event-driven applications.

Read more