Python notes

ch21.asynchronous_programming

Ch 21. Asynchronous Programming

Types of Coroutines in Python (from Python 3.5+)

  1. Native Coroutine:

    • Defined using async def.
    • Uses await to delegate to other native coroutines, similar to how classic coroutines use yield from.
    • Even without await in the body, async def always defines a native coroutine.
    • await can only be used inside native coroutines.

  2. Classic Coroutine:

    • A generator function that consumes data via my_coro.send(data) and reads the data using yield.
    • Can delegate to other classic coroutines using yield from.
    • Cannot be driven by await and is no longer supported by asyncio.

  3. Generator-based Coroutine:

    • A generator function decorated with @types.coroutine.
    • Introduced in Python 3.5 to make the generator compatible with the await keyword.

Asynchronous Generators

  • Defined using async def and yield in its body.
  • Returns an asynchronous generator object with the __anext__ coroutine method to retrieve the next item.

Example: Probing Domains with asyncio

Scenario: Checking if Python keyword-based domains (e.g., await.dev) are available.

  • Explanation:

    • The + indicates the domain was resolved via DNS.
    • No + means the domain did not resolve and may be available.

  • Key Details:

    • DNS probing is done using native coroutine objects.
    • The asynchronous operations reduce total execution time compared to sequential checks.
#!/usr/bin/env python3
import asyncio
import socket
from keyword import kwlist

MAX_KEYWORD_LEN = 4  # Limit the length of keywords to 4 characters

async def probe(domain: str) -> tuple[str, bool]:
    loop = asyncio.get_running_loop()  # Get the current event loop
    try:
        await loop.getaddrinfo(domain, None)  # Check domain's DNS info
    except socket.gaierror:
        return (domain, False)  # Domain not resolved
    return (domain, True)  # Domain resolved

async def main() -> None:
    names = (kw for kw in kwlist if len(kw) <= MAX_KEYWORD_LEN)  # Filter keywords
    domains = (f'{name}.dev'.lower() for name in names)  # Generate domain names
    coros = [probe(domain) for domain in domains]  # List of coroutine objects
    for coro in asyncio.as_completed(coros):  # Process as coroutines complete
        domain, found = await coro
        mark = '+' if found else ' '
        print(f'{mark} {domain}')

if __name__ == '__main__':
    asyncio.run(main())  # Start the event loop

Key Points

  • Keyword Length: Limit the domain names to Python keywords with a maximum of 4 characters.
  • probe function:
    • Probes a domain using asyncio to check if it resolves.
    • Returns a tuple containing the domain and a boolean indicating resolution status.
  • main coroutine:
    • Generates domain names based on Python keywords and probes each domain.
    • Uses asyncio.as_completed to process results as they complete (not in submission order).
    • Prints resolved domains with a + mark.
  • asyncio.get_running_loop (used in probe) was added in Python 3.7.
    • Raises RuntimeError if no loop is running.
    • Deprecated asyncio.get_event_loop in Python 3.10, which may create an event loop if necessary.

Guido’s Trick to Read Asynchronous Code

  • Tip: Ignore the async and await keywords when reading code to make it easier to follow. It will then resemble a regular sequential function.

  • Example: The coroutine below:

    async def probe(domain: str) -> tuple[str, bool]:
    loop = asyncio.get_running_loop()
    try:
        await loop.getaddrinfo(domain, None)
    except socket.gaierror:
        return (domain, False)
    return (domain, True)

    Can be understood as:

    def probe(domain: str) -> tuple[str, bool]:  # no async
    loop = asyncio.get_running_loop()
    try:
        loop.getaddrinfo(domain, None)  # no await
    except socket.gaierror:
        return (domain, False)
    return (domain, True)

The only difference: the asynchronous version magically avoids blocking.

How Coroutines Work

  • Suspension of Execution:

    • await suspends the coroutine’s execution, yielding control to the event loop.
    • While a coroutine (e.g., probe('if.dev')) is suspended, the event loop can drive other coroutines (e.g., probe('or.dev')).

  • Example Execution:

    • During await loop.getaddrinfo(domain, None) in probe('if.dev'), control is handed back to the event loop while waiting for the DNS resolution.
    • The event loop continues to work on other coroutines until it gets a response for getaddrinfo('if.dev', None).
    • Once the result is ready, the suspended probe('if.dev') coroutine resumes and continues execution from where it left off (handling exceptions or returning the result).

The Awaitable Concept

  • for Keyword: Works with iterables.

  • await Keyword: Works with awaitables.

  • As an end user of asyncio, these are the awaitables you will see on a daily basis:

    • A native coroutine object, which you get by calling a native coroutine function
    • An asyncio.Task, which you usually get by passing a coroutine object to asyncio.create_task()

Common Awaitables in asyncio 

import asyncio
import itertools

async def spin(msg: str) -> None:  # <1>
    for char in itertools.cycle(r'\|/-'):
        status = f'\r{char} {msg}'
        print(status, flush=True, end='')
        try:
            await asyncio.sleep(.1)  # <2>
        except asyncio.CancelledError:  # <3>
            break
    blanks = ' ' * len(status)
    print(f'\r{blanks}\r', end='')

async def slow() -> int:
    await asyncio.sleep(3)  # <4>
    return 42

def main() -> None:  # <1>
    result = asyncio.run(supervisor())  # <2>
    print(f'Answer: {result}')

async def supervisor() -> int:  # <3>
    spinner = asyncio.create_task(spin('thinking!'))  # <4>
    print(f'spinner object: {spinner}')  # <5>
    result = await slow()  # <6>
    spinner.cancel()  # <7>
    return result

if __name__ == '__main__':
    main()

  1. Native Coroutine Object:

    • Created by calling a native coroutine function (defined with async def).

  2. asyncio.Task:

    • Usually created by passing a coroutine object to asyncio.create_task().
    • Task Usage:
      • You don’t always need to await a Task.
      • Example: Use asyncio.create_task(one_coro()) to schedule a coroutine for concurrent execution without waiting for its result.
      • If you don’t expect to cancel the task or wait for it, there is no need to keep the Task object returned from create_task. Creating the task is enough to schedule the coroutine to run.

  3. await other_coro():

    • Runs other_coro immediately and waits for its result.
    • This is used when the result of the coroutine is required before continuing, e.g., res = await slow() in spinner_async.py.

Lower-level Awaitables

When implementing asynchronous libraries or contributing to asyncio itself, you may also deal with these lower-level awaitables:

  1. Objects with an __await__ method:

    • Return an iterator.
    • Example: An asyncio.Future instance (since asyncio.Task is a subclass of asyncio.Future).

  2. Objects from Other Languages:

    • These are written using the Python/C API and have a tp_as_async.am_await function, which returns an iterator (similar to the __await__ method).

  3. Generator-based Coroutine Objects:

    • These are still used in some codebases but are being deprecated.

Key Explanation from PEPs

  • PEP 492:

    • States that await uses the yield from mechanism but with an additional validation step.
    • Only accepts awaitable objects.

  • PEP 380:

    • Introduced the yield from construct, which is referred to in await implementation details.

Downloading with asyncio and HTTPX

  1. HTTPX:

    • An asynchronous HTTP client library (not part of the standard library).
    • Used to perform HTTP requests asynchronously with the AsyncClient.

  2. Asynchronous Design:

    • The script downloads 20 flags concurrently using native coroutines.
    • Designed to avoid blocking the event loop during network I/O.
import asyncio

from httpx import AsyncClient  # <1>

from flags import BASE_URL, save_flag, main  # <2>

def download_many(cc_list: list[str]) -> int:    # <1>
    return asyncio.run(supervisor(cc_list))      # <2>

async def supervisor(cc_list: list[str]) -> int:
    async with AsyncClient() as client:          # <3>
        to_do = [download_one(client, cc)
                 for cc in sorted(cc_list)]      # <4>
        res = await asyncio.gather(*to_do)       # <5>

    return len(res)                              # <6>

async def download_one(client: AsyncClient, cc: str):  # <3>
    image = await get_flag(client, cc)
    save_flag(image, f'{cc}.gif')
    print(cc, end=' ', flush=True)
    return cc

async def get_flag(client: AsyncClient, cc: str) -> bytes:  # <4>
    url = f'{BASE_URL}/{cc}/{cc}.gif'.lower()
    resp = await client.get(url, timeout=6.1,
                                  follow_redirects=True)  # <5>
    return resp.read()  # <6>

if __name__ == '__main__':
    main(download_many)

Main Functions Overview

  1. download_many Function:

    • Purpose: Acts as the entry point, running the supervisor coroutine.
    • Details:
      • Plain function to be passed to the main() function from flags.py.
      • Uses asyncio.run() to execute the event loop and drive the supervisor coroutine.

  2. supervisor Coroutine:

    • Purpose: Coordinates downloading of flags concurrently.
    • Details:
      • Uses AsyncClient (HTTPX) for asynchronous HTTP requests.
      • Builds a list of coroutines (to_do) by calling download_one for each flag.
      • Waits for all downloads to complete using asyncio.gather, which collects results in submission order.

  3. download_one Coroutine:

    • Purpose: Downloads a single flag image.
    • Details:
      • Calls get_flag to fetch the image data.
      • Saves the flag using save_flag() and displays the country code.
      • Returns the country code of the downloaded flag.

  4. get_flag Coroutine:

    • Purpose: Retrieves the flag image from the URL.
    • Details:
      • Constructs the flag image URL and makes an asynchronous GET request.
      • Returns the flag image as bytes.

Important Points

  • AsyncClient: Acts as an asynchronous context manager. Ensures proper setup and teardown of resources.
  • asyncio.gather: Waits for all coroutines (downloads) to finish and returns a list of results.
  • Concurrency: By using asyncio, multiple flag downloads occur simultaneously, greatly improving performance compared to sequential downloads.
  • Blocking I/O: The save_flag() function is a blocking I/O operation and should ideally be asynchronous. However, as Python does not have an async filesystem API, it is done synchronously here.
    • Thread Delegation: In more advanced usage (as discussed later in the book), blocking I/O like save_flag() can be delegated to a thread to prevent blocking the event loop.

The Secret of Native Coroutines: Humble Generators

  1. Difference from Classic Coroutines:

    • In native coroutines (like in flags_asyncio.py), there are no visible .send() calls or yield expressions, unlike classic coroutines.
    • Instead, native coroutines use await to suspend execution and wait for other coroutines or asynchronous tasks to complete.

  2. How Native Coroutines Work:

  • Event Loop: The asyncio event loop is responsible for making .send() calls that drive the coroutines.
  • Await Chains: Coroutines await other coroutines or awaitable objects, which eventually reach low-level awaitables like timers or network I/O operations.
    • These low-level awaitables are typically implemented deep in the libraries (such as HTTPX) or Python/C extensions.
  • Borrowed from yield from: await borrows most of its functionality from the yield from construct in classic coroutines, which also uses .send() to drive execution.
  1. Concurrency with asyncio:
    • Functions like asyncio.gather and asyncio.create_task allow multiple coroutines to run concurrently.
    • These functions enable concurrent execution of I/O operations using a single event loop and a single thread.

The All-or-Nothing Problem

  1. Asynchronous I/O Requirement:

    • To maximize performance with asyncio, every I/O function must be asynchronous. This ensures that control can be handed back to the event loop while waiting for I/O operations.
    • If a function blocks, it should either be rewritten as a coroutine or run in a separate thread or process (see "Delegating Tasks to Executors" for details).

  2. Rewriting Blocking Functions:

    • In flags_asyncio.py, the get_flag function had to be rewritten as a coroutine to utilize the asynchronous API of HTTPX.
    • Similarly, download_one from flags_threadpool.py (Example 20-3) could not be reused because it was designed for synchronous execution. In the asyncio version, it had to be rewritten as a coroutine to handle the await operations.

  3. Concurrency in flags_asyncio.py:

    • In supervisor, multiple download_one coroutine objects are created and driven concurrently by asyncio.gather. This allows for concurrent downloads without blocking the event loop.

Takeaways

  • Native coroutines are driven by the asyncio event loop, which makes the .send() calls internally, unlike classic coroutines where .send() is explicitly used.
  • To achieve peak performance with asyncio, all I/O-bound functions must be asynchronous or delegated to separate threads/processes if they cannot be made non-blocking.

Asynchronous Context Managers

  1. Context Managers Recap:

    • In synchronous code, context managers use __enter__ and __exit__ methods to run setup and teardown code around a with block (e.g., file handling, database transactions).

  2. Limitation in Asynchronous Code:

    • In asynchronous programming, setup and teardown operations often involve I/O and should not block the event loop.
    • However, the classic with statement doesn’t support coroutines for __enter__ or __exit__.

  3. Solution: async with Statement:

    • PEP 492 introduced the async with statement, which works with asynchronous context managers—objects that implement __aenter__ and __aexit__ as coroutines.

Example: Asynchronous Database Transaction (asyncpg)

  • Original Synchronous-Style Code:

    tr = connection.transaction()
await tr.start()
try:
    await connection.execute("INSERT INTO mytable VALUES (1, 2, 3)")
except:
    await tr.rollback()
    raise
else:
    await tr.commit()

  • Rewritten with async with:

    async with connection.transaction():
    await connection.execute("INSERT INTO mytable VALUES (1, 2, 3)")

  • How It Works:

    • __aenter__: Starts the transaction with await self.start().
    • __aexit__: Rolls back or commits the transaction based on whether an exception occurred, using await for asynchronous I/O.
    • This allows asyncpg to handle multiple transactions concurrently, improving efficiency.

Asynchronous Context Manager in httpx

  • HTTPX's AsyncClient:

    • The AsyncClient class in httpx is also an asynchronous context manager.
    • It uses await in its __aenter__ and __aexit__ methods to manage setup (e.g., opening connections) and teardown (e.g., closing connections).

  • Usage in flags_asyncio.py:

    async with AsyncClient() as client:
    # Perform HTTP requests with the client

  • The async with block ensures that the AsyncClient is properly managed, opening and closing connections asynchronously without blocking the event loop.

  • By using async with, you avoid manually starting and stopping operations (like transactions) and simplify the flow of asynchronous code.

  • Side note: Connection Pooling in asyncpg: asyncpg also provides connection pooling, helping to manage PostgreSQL’s limited concurrency by maintaining a pool of internal connections without the need for external tools like pgbouncer.

Enhancing the asyncio Downloader

  1. Progress Display:

    • The script integrates the tqdm library to show a progress bar during downloads.
    • Instead of waiting for all downloads to finish (like with asyncio.gather), it processes each download as it completes using asyncio.as_completed, enabling real-time progress updates.

  2. Error Handling:

    • Handles various HTTP errors such as timeouts and HTTPStatusError (e.g., 404 Not Found).
    • Gracefully continues downloading the remaining flags even after encountering errors.

  3. Concurrency Control:

    • Uses a semaphore to limit the number of concurrent download requests.
    • The default is set to a low concurrency level (DEFAULT_CONCUR_REQ = 5) to avoid overwhelming servers and triggering HTTP errors (e.g., 503 Service Temporarily Unavailable).
    • Can handle up to 1000 concurrent requests (MAX_CONCUR_REQ = 1000).

Code Breakdown

get_flag Coroutine

  • Similar to the sequential version but adapted for asynchronous execution.

  • Makes an asynchronous HTTP request using httpx.AsyncClient.

  • Raises an error if the response status is not successful.

    async def get_flag(client: httpx.AsyncClient, base_url: str, cc: str) -> bytes:
    url = f'{base_url}/{cc}/{cc}.gif'.lower()
    resp = await client.get(url, timeout=3.1, follow_redirects=True)
    resp.raise_for_status()  # Raise HTTP error for unsuccessful status codes
    return resp.content

download_one Coroutine

  • Downloads a single flag with error handling and concurrency control using asyncio.Semaphore.

  • If an error occurs (e.g., 404 Not Found), it logs the error and continues.

  • Saves the flag image using asyncio.to_thread to avoid blocking the event loop with file I/O operations.

    async def download_one(client: httpx.AsyncClient, cc: str, base_url: str,
                       semaphore: asyncio.Semaphore, verbose: bool) -> DownloadStatus:
    try:
        # A semaphore controls the number of concurrent downloads.
        async with semaphore:  # Controls concurrent downloads
            image = await get_flag(client, base_url, cc)
    except httpx.HTTPStatusError as exc:
        res = exc.response
        if res.status_code == HTTPStatus.NOT_FOUND:
            status = DownloadStatus.NOT_FOUND
            msg = f'not found: {res.url}'
        else:
            raise  # Re-raise for other HTTP errors
    else:
        # Save flag in a separate thread
        await asyncio.to_thread(save_flag, image, f'{cc}.gif')
        status = DownloadStatus.OK
        msg = 'OK'
        if verbose and msg:
            print(cc, msg)
    return status

Important Notes

  1. Testing Concurrent Clients:
    • Be mindful when testing against external servers, as they might suspect a DoS attack due to the high rate of requests from asyncio clients.
    • Use local test servers to avoid causing issues with real servers
  2. Thread Delegation for I/O:
    • File I/O can block the event loop since it’s much slower than reading/writing from RAM.
    • asyncio.to_thread provides an easy way to offload such tasks to a thread pool, ensuring the event loop remains responsive.

Throttling Requests with asyncio.Semaphore

  1. Why Throttle Requests?

    • Network clients can overwhelm a server if too many requests are sent concurrently.
    • Throttling helps by limiting the number of concurrent requests, preventing issues like HTTP errors (e.g., 503 Service Temporarily Unavailable).

  2. How a Semaphore Helps:

    • A semaphore is a synchronization primitive that controls access to a resource by multiple coroutines.
    • It allows a set number of coroutines to proceed, while others wait until the semaphore's counter increases (i.e., other coroutines finish).
    • In this script, asyncio.Semaphore is used to limit how many coroutines (requests) can be active at a time.

supervisor Coroutine

  • Coordinates the download process, handling concurrency, progress, and error handling.

    async def supervisor(cc_list: list[str], base_url: str, verbose: bool, concur_req: int) -> Counter[DownloadStatus]:
    counter: Counter[DownloadStatus] = Counter()
    semaphore = asyncio.Semaphore(concur_req)  # Controls concurrent requests
    async with httpx.AsyncClient() as client:
        to_do = [download_one(client, cc, base_url, semaphore, verbose) for cc in sorted(cc_list)]  # Create coroutine list
        to_do_iter = asyncio.as_completed(to_do)  # Get coroutines as they complete

        if not verbose:
            to_do_iter = tqdm.tqdm(to_do_iter, total=len(cc_list))  # Add progress bar

        error: httpx.HTTPError | None = None
        for coro in to_do_iter:
            try:
                status = await coro  # Process completed coroutine
            except httpx.HTTPStatusError as exc:
                error_msg = f'HTTP error {exc.response.status_code} - {exc.response.reason_phrase}'
                error = exc
            except httpx.RequestError as exc:
                error_msg = f'{exc} {type(exc)}'.strip()
                error = exc
            except KeyboardInterrupt:
                break
            if error:
                status = DownloadStatus.ERROR
                if verbose:
                    url = str(error.request.url)
                    cc = Path(url).stem.upper()
                    print(f'{cc} error: {error_msg}')
            counter[status] += 1
        return counter

  • Semaphore Creation:

    • semaphore = asyncio.Semaphore(concur_req) limits the number of active download coroutines to concur_req (controlled by a command-line argument).

  • Handling Concurrent Downloads:

    • to_do creates a list of coroutines for each country code (flag).
    • asyncio.as_completed(to_do) ensures results are processed as they finish (not waiting for all to complete).
      • Instead of waiting for all coroutines to finish (as with asyncio.gather)
      • This allows real-time updates to the progress bar as each flag finishes downloading.

  • Progress Bar:

    • If verbosity is disabled, the tqdm progress bar wraps the iterator to_do_iter, tracking download progress.

  • Error Handling:

    • Captures and handles various HTTP errors (e.g., 404 Not Found or connection issues).
    • Errors are logged and counted, and processing continues with the remaining requests.
    • The syntax error: httpx.HTTPError | None = None is valid in Python 3.10 and later versions. It uses the new union type syntax introduced in PEP 593, which allows you to use the pipe symbol (|) to denote a union of types.

download_many Function

  • Invokes the supervisor coroutine using asyncio.run to execute the event loop.

  • Returns the final status count (number of successful/failed downloads).

    def download_many(cc_list: list[str], base_url: str, verbose: bool, concur_req: int) -> Counter[DownloadStatus]:
    coro = supervisor(cc_list, base_url, verbose, concur_req)
    counts = asyncio.run(coro)  # Run the coroutine
    return counts

Using Semaphores for Concurrency Control

  • asyncio.Semaphore Usage:

    • The semaphore limits the number of concurrent coroutines by ensuring that only concur_req downloads are active at once.
    async def download_one(..., semaphore: asyncio.Semaphore, ...):
    async with semaphore:  # Acquire the semaphore
        # Perform download operation

  • Why This Is Important:

    • It prevents overloading the server with too many requests at once.
    • By managing concurrency, the program remains efficient without triggering server-side errors like rate-limiting or overloading.

Error Handling and Exception Scope

  • Extracting Errors:

    • If an error occurs, the error variable captures the exception, and the relevant message is logged after the try/except block.

    • Python's try/except block doesn't create a new scope, but exceptions captured in the except block (exc) are scoped to that block. This is why the error is saved in the error variable.

Making Multiple Requests for Each Flag Download

The new goal is to download:

  1. The flag image.
  2. The country metadata (JSON file) to get the country name and include it in the file name.

In a threaded environment, this would be done by making two blocking requests. However, in an asynchronous script, await allows you to handle both requests sequentially while keeping local variables in scope, avoiding the complexity of callbacks.

get_country Coroutine

  • This coroutine retrieves the metadata.json file, which contains the country name.

    async def get_country(client: httpx.AsyncClient, base_url: str, cc: str) -> str:
    url = f'{base_url}/{cc}/metadata.json'.lower()
    resp = await client.get(url, timeout=3.1, follow_redirects=True)
    resp.raise_for_status()
    metadata = resp.json()
    return metadata['country']

  • Fetches the JSON file at the URL: <base_url>/<cc>/metadata.json.

  • Uses await client.get(...) to asynchronously retrieve the file.

  • Returns the country name from the metadata dictionary, extracted from the JSON response.

Modified download_one Coroutine (Example 21-9)

  • The download_one coroutine now handles two sequential requests:

    1. Fetches the flag image.
    2. Fetches the country metadata to get the country name.
    async def download_one(client: httpx.AsyncClient, cc: str, base_url: str,
                       semaphore: asyncio.Semaphore, verbose: bool) -> DownloadStatus:
    try:
        async with semaphore:
            image = await get_flag(client, base_url, cc)
        async with semaphore:
            country = await get_country(client, base_url, cc)
    except httpx.HTTPStatusError as exc:
        res = exc.response
        if res.status_code == HTTPStatus.NOT_FOUND:
            status = DownloadStatus.NOT_FOUND
            msg = f'not found: {res.url}'
        else:
            raise
    else:
        filename = country.replace(' ', '_')  # Replace spaces with underscores for filenames
        await asyncio.to_thread(save_flag, image, f'{filename}.gif')
        status = DownloadStatus.OK
        msg = 'OK'
        if verbose and msg:
            print(cc, msg)
    return status

  • Two Asynchronous Requests:

    • First, it fetches the flag image using await get_flag(client, base_url, cc).
    • Then, it fetches the country metadata using await get_country(client, base_url, cc).

  • Semaphore for Throttling:

    • Both get_flag and get_country are wrapped in async with semaphore to ensure concurrency control. This ensures no more than the allowed number of requests happen concurrently.
    • Semaphore blocks are separate for each request to keep the time the semaphore is held to a minimum.

  • Creating the Filename:

    • The country name is used to create the file name, replacing spaces with underscores for better readability.
    • The flag image is then saved using await asyncio.to_thread(save_flag, image, f'{filename}.gif').

  • Error Handling:

    • If either the flag or metadata request fails with an HTTP error, it handles the error gracefully and logs it.

Why Not Use asyncio.gather?

  • You could use asyncio.gather to make both requests (for the flag and metadata) in parallel, but in this case, it’s unnecessary:
    • Reason: If fetching the flag fails, there’s no point in fetching the metadata. It makes more sense to fetch them sequentially in this case.

Summary of Improvements

  1. Sequential Asynchronous Requests:

    • download_one now makes two HTTP requests, one for the flag and one for the metadata, using await for both.

  2. Concurrency Control with Semaphores:

    • The semaphore is used to throttle both the flag and metadata requests, limiting how many are active concurrently.

  3. Simplified Code without Callbacks:

    • The use of await makes the code easier to follow compared to a callback-based approach, where the flow would have been split across multiple functions.

  4. Efficient File Naming:

    • The country name is used to generate a more descriptive filename for the saved flag image.

Delegate Tasks to Executors

  • Node.js Advantage: In Node.js, all I/O (network and file) is asynchronous, but Python lacks asynchronous APIs for file I/O. This means file operations in Python can block the event loop, degrading performance.
  • Solution in Python: Use asyncio with executors (e.g., thread pools) to delegate blocking tasks like file I/O to separate threads or processes, ensuring that the main event loop remains non-blocking.

Example: Saving Flags Asynchronously

Original Code (Python 3.9+)

  • In the download_one coroutine (Example 21-6), this line saves the downloaded image to disk without blocking the event loop:

    await asyncio.to_thread(save_flag, image, f'{cc}.gif')

  • asyncio.to_thread:

    • Introduced in Python 3.9.
    • Simplifies offloading a blocking task (like file I/O) to a thread without manually managing the executor.
    • Allows passing keyword arguments directly, making it more flexible than the older run_in_executor.

Supporting Python 3.7 and 3.8: Using run_in_executor

For compatibility with Python 3.7 and 3.8, you can replace asyncio.to_thread with run_in_executor. Example 21-10 shows how to do this:

loop = asyncio.get_running_loop()
loop.run_in_executor(None, save_flag, image, f'{cc}.gif')

Explanation

  1. Getting the Event Loop:

    • loop = asyncio.get_running_loop() retrieves the current event loop.

  2. Running in an Executor:

    • loop.run_in_executor(None, save_flag, image, f'{cc}.gif') offloads the save_flag function to a thread pool (or a process pool if specified).
    • The first argument (None) specifies the executor to use. Passing None selects the default ThreadPoolExecutor.

  3. Argument Handling:

    • run_in_executor allows passing positional arguments directly to the function (save_flag in this case).
    • If you need to pass keyword arguments, you must use functools.partial.

Under the Hood: run_in_executor in asyncio

  • run_in_executor is used by asyncio internally for blocking calls, like getaddrinfo (for DNS lookups). This function is blocking because it depends on network latency and DNS resolution time.

  • Thread Pool Usage:

    • Thread pools are often used behind asynchronous APIs, as they provide a pragmatic way to manage blocking I/O while maintaining a non-blocking interface via coroutines.
    • Example: The Motor MongoDB driver wraps a threaded core with an async/await interface to manage database I/O efficiently.

When to Use ProcessPoolExecutor

  1. For CPU-Intensive Tasks:

    • If a task is CPU-bound (e.g., heavy computation), using a ThreadPoolExecutor won’t help due to Python’s Global Interpreter Lock (GIL), which limits CPU-bound threads.
    • Instead, you can use a ProcessPoolExecutor, which runs tasks in separate processes, bypassing the GIL.

  2. How to Use:

    • In Supervisor: Start the ProcessPoolExecutor in a central coroutine (like supervisor) and pass it to tasks that need to use it.
    • This reduces the overhead of starting new processes for each task and allows better resource management.

Caution: run_in_executor Cancellation Issues

Caleb’s Warning

  • Cancellation Issues:

    • Tasks running in an executor (like a thread pool) don’t have a true cancellation mechanism. This can lead to situations where:
      • Long-running tasks (in a thread) might block the program from shutting down cleanly.
      • The asyncio.run function waits for the executor to finish, and if a task doesn’t stop, it could block the program indefinitely.

  • Implication:

    • Even though a coroutine might appear cancellable, tasks offloaded to a thread are not truly cancellable, and you must ensure they finish on their own or implement a manual cancellation mechanism.

    Caleb humorously suggests renaming run_in_executor to run_in_executor_uncancellable to reflect this behavior.

Summary

  • Delegating Blocking Tasks: For tasks like file I/O, you can delegate work to a thread pool using asyncio.to_thread (Python 3.9+) or run_in_executor (Python 3.7/3.8).

  • Thread vs Process Pools:

    • Use ThreadPoolExecutor for I/O-bound tasks (e.g., file I/O).
    • Use ProcessPoolExecutor for CPU-bound tasks (e.g., heavy computation) to avoid the GIL.

  • Cancellation Caveat: Be cautious with run_in_executor because tasks running in threads are not truly cancellable, which can cause shutdown issues in some cases.

Previous ch20.concurrent_executors Next pydonts