Async & concurrency

Waiting on network/disk?
  many connections, async libs exist  -> asyncio
  sync library (requests, boto3, DB-API) -> thread pool
Burning CPU in Python code?
  -> ProcessPoolExecutor (safe default)
  -> InterpreterPoolExecutor / 3.14t threads (3.14+)
Both?  asyncio + loop.run_in_executor(process_pool, fn)

import asyncio
 
 
async def greet(name: str, delay: float) -> str:
    await asyncio.sleep(delay)  # yields to the loop
    return f"hi {name}"
 
 
async def main() -> None:
    msg = await greet("ada", 0.1)
    print(msg)
 
 
if __name__ == "__main__":
    asyncio.run(main())

import asyncio
 
 
async def fetch(n: int) -> int:
    await asyncio.sleep(0.1)
    return n * 2
 
 
async def main() -> None:
    # sequential: ~0.2 s, awaiting a coroutine runs it now
    a = await fetch(1)
    b = await fetch(2)
 
    # concurrent: ~0.1 s, tasks start on the next await
    t1 = asyncio.create_task(fetch(3), name="f3")
    t2 = asyncio.create_task(fetch(4))
    c, d = await t1, await t2
    print(a, b, c, d)
 
asyncio.run(main())

import asyncio
 
 
async def work(n: int) -> int:
    await asyncio.sleep(0.01 * n)
    if n == 3:
        raise ValueError(f"bad {n}")
    return n
 
 
async def main() -> None:
    async with asyncio.TaskGroup() as tg:
        tasks = [tg.create_task(work(n)) for n in (1, 2)]
    print([t.result() for t in tasks])  # [1, 2]
 
    try:
        async with asyncio.TaskGroup() as tg:
            for n in (1, 3, 5):
                tg.create_task(work(n))
    except* ValueError as eg:  # ExceptionGroup
        print("failed:", eg.exceptions)
 
    got = await asyncio.gather(
        work(1), work(3), return_exceptions=True
    )
    print(got)  # [1, ValueError('bad 3')]
 
asyncio.run(main())

import asyncio
 
 
async def slow() -> str:
    try:
        await asyncio.sleep(10)
        return "done"
    except asyncio.CancelledError:
        print("cleaning up")
        raise  # always re-raise
    finally:
        print("finally runs too")
 
 
async def main() -> None:
    try:
        async with asyncio.timeout(0.1):  # 3.11+
            await slow()
    except TimeoutError:  # builtin since 3.11
        print("timed out")
 
    t = asyncio.create_task(slow())
    await asyncio.sleep(0)
    t.cancel("shutting down")
    try:
        await t
    except asyncio.CancelledError:
        print("canceled:", t.cancelled())
 
asyncio.run(main())

import asyncio
 
lock = asyncio.Lock()
balance = 0
 
 
async def deposit(amount: int) -> None:
    global balance
    async with lock:  # read-modify-write across an await
        current = balance
        await asyncio.sleep(0)
        balance = current + amount
 
 
async def main() -> None:
    async with asyncio.TaskGroup() as tg:
        for _ in range(100):
            tg.create_task(deposit(1))
    print(balance)  # 100; without the lock: 1
 
asyncio.run(main())

import asyncio
from collections.abc import AsyncGenerator, AsyncIterator
from contextlib import aclosing, asynccontextmanager
 
 
async def ticks(n: int) -> AsyncGenerator[int]:
    for i in range(n):
        await asyncio.sleep(0.01)
        yield i
 
 
@asynccontextmanager
async def connection(url: str) -> AsyncIterator[str]:
    print("open", url)
    try:
        yield f"conn:{url}"
    finally:
        print("close", url)
 
 
async def main() -> None:
    async with connection("db://local") as conn:
        async with aclosing(ticks(5)) as it:
            evens = [i async for i in it if i % 2 == 0]
        print(conn, evens)  # conn:db://local [0, 2, 4]
 
asyncio.run(main())

import asyncio
import time
 
 
def blocking_io(path: str) -> int:
    time.sleep(0.2)  # stands in for a sync library call
    return len(path)
 
 
async def main() -> None:
    sizes = await asyncio.gather(
        asyncio.to_thread(blocking_io, "a.txt"),
        asyncio.to_thread(blocking_io, "bb.txt"),
    )
    print(sizes)  # [5, 6] after ~0.2 s, not 0.4 s
 
asyncio.run(main())

from concurrent.futures import (
    ThreadPoolExecutor,
    as_completed,
)
from urllib.request import urlopen
 
 
def status(url: str) -> tuple[str, int]:
    with urlopen(url, timeout=5) as res:
        return url, res.status
 
 
urls = ["https://example.com", "https://python.org"]
with ThreadPoolExecutor(max_workers=8) as pool:
    futs = [pool.submit(status, u) for u in urls]
    for fut in as_completed(futs):
        try:
            print(fut.result())
        except OSError as err:
            print("failed:", err)

import queue
import threading
 
jobs: queue.Queue[int] = queue.Queue()
results: list[int] = []
lock = threading.Lock()
 
 
def worker(stop: threading.Event) -> None:
    while not stop.is_set():
        try:
            n = jobs.get(timeout=0.1)
        except queue.Empty:
            continue
        with lock:
            results.append(n * n)
        jobs.task_done()
 
 
stop = threading.Event()
threads = [
    threading.Thread(target=worker, args=(stop,))
    for _ in range(4)
]
for t in threads:
    t.start()
for n in range(10):
    jobs.put(n)
jobs.join()  # every job done
stop.set()
for t in threads:
    t.join()
print(sorted(results))

from concurrent.futures import ProcessPoolExecutor
 
 
def cpu_heavy(n: int) -> int:
    return sum(i * i for i in range(n))
 
 
if __name__ == "__main__":  # required for spawn/forkserver
    with ProcessPoolExecutor() as pool:
        totals = list(pool.map(cpu_heavy, [10**6] * 8))
    print(totals[0])

import sys
import sysconfig
 
free_build = sysconfig.get_config_var("Py_GIL_DISABLED")
print("free-threaded build:", bool(free_build))
print("GIL on now:", sys._is_gil_enabled())

from concurrent import interpreters
from concurrent.futures import InterpreterPoolExecutor
 
 
def fib(n: int) -> int:
    return n if n < 2 else fib(n - 1) + fib(n - 2)
 
 
if __name__ == "__main__":
    interp = interpreters.create()
    print(interp.call(fib, 20))  # 6765, in isolation
    interp.close()
 
    with InterpreterPoolExecutor(max_workers=4) as pool:
        print(list(pool.map(fib, [24, 25, 26])))

import asyncio
from collections.abc import Coroutine
from typing import Any
 
background: set[asyncio.Task[Any]] = set()
 
 
def fire_and_forget(coro: Coroutine[Any, Any, Any]) -> None:
    task = asyncio.create_task(coro)
    background.add(task)  # strong reference
    task.add_done_callback(background.discard)

import asyncio
from collections.abc import Awaitable, Callable, Iterable
 
 
async def bounded[T, R](
    items: Iterable[T],
    fn: Callable[[T], Awaitable[R]],
    limit: int = 10,
) -> list[R]:
    sem = asyncio.Semaphore(limit)
 
    async def one(item: T) -> R:
        async with sem:
            return await fn(item)
 
    async with asyncio.TaskGroup() as tg:
        tasks = [tg.create_task(one(i)) for i in items]
    return [t.result() for t in tasks]

import asyncio
import random
from collections.abc import Awaitable, Callable
 
 
async def retry[T](
    fn: Callable[[], Awaitable[T]],
    *,
    attempts: int = 4,
    base: float = 0.2,
    cap: float = 5.0,
    retry_on: tuple[type[Exception], ...] = (OSError,),
) -> T:
    for attempt in range(attempts):
        try:
            return await fn()
        except retry_on:
            if attempt == attempts - 1:
                raise
            ceiling = min(cap, base * 2**attempt)
            await asyncio.sleep(random.uniform(0, ceiling))
    raise AssertionError("unreachable")

import asyncio
from collections.abc import Awaitable
 
 
async def with_timeout[T](
    aw: Awaitable[T], sec: float
) -> T | None:
    try:
        async with asyncio.timeout(sec):
            return await aw
    except TimeoutError:
        return None
 
 
async def main() -> None:
    delays = [0.05, 2.0, 0.1]
    results = await asyncio.gather(
        *(with_timeout(asyncio.sleep(d, d), 0.5)
          for d in delays)
    )
    print(results)  # [0.05, None, 0.1]
 
asyncio.run(main())

import asyncio
 
 
async def producer(q: asyncio.Queue[int]) -> None:
    for n in range(20):
        await q.put(n)  # waits while the queue is full
    q.shutdown()  # 3.13+: consumers stop once drained
 
 
async def consumer(name: str, q: asyncio.Queue[int]) -> None:
    while True:
        try:
            n = await q.get()
        except asyncio.QueueShutDown:
            return
        await asyncio.sleep(0.01)  # process n
        print(name, n)
 
 
async def main() -> None:
    q: asyncio.Queue[int] = asyncio.Queue(maxsize=5)
    async with asyncio.TaskGroup() as tg:
        tg.create_task(producer(q))
        for i in range(3):
            tg.create_task(consumer(f"w{i}", q))
 
asyncio.run(main())

import asyncio
import signal
 
 
async def serve(stop: asyncio.Event) -> None:
    while not stop.is_set():
        await asyncio.sleep(0.5)  # handle one unit of work
    print("draining...")
 
 
async def main() -> None:
    stop = asyncio.Event()
    loop = asyncio.get_running_loop()
    for sig in (signal.SIGINT, signal.SIGTERM):
        loop.add_signal_handler(sig, stop.set)
    try:
        await serve(stop)
    finally:
        print("closing connections")
 
if __name__ == "__main__":
    asyncio.run(main())

import asyncio
from concurrent.futures import ProcessPoolExecutor
 
 
def crunch(n: int) -> int:  # top level: must pickle
    return sum(i * i for i in range(n))
 
 
async def main() -> None:
    loop = asyncio.get_running_loop()
    with ProcessPoolExecutor() as pool:
        results = await asyncio.gather(*(
            loop.run_in_executor(pool, crunch, n)
            for n in (10**6, 2 * 10**6, 3 * 10**6)
        ))
    print(results)
 
if __name__ == "__main__":
    asyncio.run(main())

import asyncio
 
import httpx
 
 
async def fetch_all(urls: list[str]) -> dict[str, int]:
    limits = httpx.Limits(max_connections=10)
    timeout = httpx.Timeout(10.0)
    async with httpx.AsyncClient(
        limits=limits, timeout=timeout
    ) as client:
        async with asyncio.TaskGroup() as tg:
            tasks = {
                u: tg.create_task(client.get(u))
                for u in urls
            }
    return {u: t.result().status_code
            for u, t in tasks.items()}
 
 
urls = ["https://example.com", "https://www.python.org"]
print(asyncio.run(fetch_all(urls)))