Checkpoint: Four Sources at Once

Time to put the day together. Four servers hold logs. One is quick, one is a little slower, one refuses the connection, and one simply never answers. The analyzer needs whatever it can get, in about the time of a single fetch, and a clear line saying what happened to each server.

Day 6 checkpoint

async def fetch_safely(server, timeout):
    """One server's Result. Expected network failures become data; bugs escape."""
    try:
        async with asyncio.timeout(timeout):
            lines = await fetch_log(server)
    except TimeoutError:
        return Result(server, "timeout", error=f"no answer after {timeout}s")
    except ConnectionError as error:
        return Result(server, "failed", error=str(error))
    return Result(server, "ok", tuple(lines))


async def collect(servers, timeout):
    """A Result for every server, fetched all at once, in the order given.

    Expected failures are Results. A bug escapes, and cancels the other fetches.
    """
    async with asyncio.TaskGroup() as group:
        tasks = [group.create_task(fetch_safely(server, timeout)) for server in servers]
    return [task.result() for task in tasks]


def error_counts(results):
    """{server: number of ERROR lines} for every server that answered."""
    return {r.server: sum(1 for line in r.lines if line.split()[2] == "ERROR")
            for r in results if r.status == "ok"}


def report(results):
    """One line of text per server, as the tests show."""
    return [f"{r.server:<10}{r.status:<10}{f'{len(r.lines)} lines' if r.status == 'ok' else r.error}"
            for r in results]

The Day 6 reference solution

Today was standalone, so this is the complete collector rather than the analyzer. main.py fetches all four servers, prints the report, then runs the tests.

Working on your own machine? Download the Day 6 files and carry on there. You need Python 3.12 or later, and nothing else.

Tomorrow the analyzer returns, to be finished, logged and given a command line. Fetching logs from servers would plug in as one more source of lines, since parse_lines accepts any iterable of text.

Going further: threads, processes and real HTTP

asyncio overlaps waiting, on one thread. Python has two other tools for doing several things at once, and each suits a different kind of work.

Threads, for blocking calls. Many libraries block: requests.get, time.sleep, most database drivers. Run each call on its own thread and the waits overlap, without rewriting the library. asyncio.to_thread does this from async code. concurrent.futures.ThreadPoolExecutor does it from ordinary code:

import asyncio
import time
from concurrent.futures import ThreadPoolExecutor

SERVERS = ["auth", "payments", "search"]


def blocking_fetch(server):
    time.sleep(0.3)                     # a blocking call, like requests.get(...)
    return f"{server} log"


async def with_asyncio():
    return await asyncio.gather(*(asyncio.to_thread(blocking_fetch, s) for s in SERVERS))


def with_threads():
    with ThreadPoolExecutor() as pool:
        return list(pool.map(blocking_fetch, SERVERS))


for name, run in [("to_thread", lambda: asyncio.run(with_asyncio())), ("thread pool", with_threads)]:
    start = time.perf_counter()
    print(f"{name:12} {run()} in {time.perf_counter() - start:.1f}s")

Both take 0.3 seconds for three 0.3-second waits.

Processes, for heavy calculation. In the standard build of CPython, only one thread runs Python code at a time. That rule is the GIL, the global interpreter lock. So threads overlap waits, but not calculation. Separate processes each have their own interpreter, and really do run in parallel on several CPU cores:

import time
from concurrent.futures import ProcessPoolExecutor


def crunch(n):
    return sum(i * i for i in range(n))


if __name__ == "__main__":                  # required: each process imports this file
    jobs = [20_000_000] * 4
    start = time.perf_counter()
    [crunch(n) for n in jobs]
    print(f"one process:   {time.perf_counter() - start:.1f}s")
    start = time.perf_counter()
    with ProcessPoolExecutor() as pool:
        list(pool.map(crunch, jobs))
    print(f"four processes: {time.perf_counter() - start:.1f}s")

On a four-core laptop, the second number is roughly a third of the first. Processes cost more to start, and everything sent to them has to be copied across, so they pay off only for real work.

There’s also a free-threaded build of CPython without the GIL, where threads can calculate in parallel too. It arrived as an experiment in 3.13 and is officially supported from 3.14, as a separate, optional build. It isn’t the default yet, and some libraries don’t support it yet, so check before relying on it.

Real HTTP. The standard library’s urllib blocks, so from async code, run it in a thread:

import asyncio
import urllib.request

URLS = ["https://www.python.org", "https://docs.python.org/3/", "https://pypi.org"]


def get_status(url):
    with urllib.request.urlopen(url, timeout=5) as response:
        return response.status


async def main():
    results = await asyncio.gather(*(asyncio.to_thread(get_status, url) for url in URLS),
                                   return_exceptions=True)
    for url, result in zip(URLS, results):
        print(url, result)


asyncio.run(main())

For real projects, the third-party httpx library has a native async client (httpx.AsyncClient), which needs no threads at all. On macOS with Python from python.org, a CERTIFICATE_VERIFY_FAILED error means the certificates haven’t been installed yet. Run Install Certificates.command from the Python folder in Applications.

What you can now do

• Fetch many sources concurrently with a time limit, and report on every one of them, whatever happened.
• Keep the async part of a program small, with plain functions around it.
• Choose between asyncio, threads and processes by the kind of work.