Reliability: Timeouts, Retries, Cancellation, Progress and Shutdown

import time
from concurrent.futures import ThreadPoolExecutor

finished = []


def slow_call() -> str:
    time.sleep(0.3)
    finished.append("slow")
    return "late answer"


with ThreadPoolExecutor(max_workers=1) as ex:
    fut = ex.submit(slow_call)
    try:
        fut.result(timeout=0.05)
    except TimeoutError:
        print("gave up waiting; cancel() ->", fut.cancel(), "running:", fut.running())
print("task still finished:", finished)
# gave up waiting; cancel() -> False running: True
# task still finished: ['slow']

import time
from concurrent.futures import ThreadPoolExecutor, wait


class Deadline:
    def __init__(self, seconds: float) -> None:
        self.at = time.monotonic() + seconds

    def remaining(self) -> float:
        return max(0.0, self.at - time.monotonic())

    def expired(self) -> bool:
        return self.remaining() == 0.0


def tool_call(name: str, deadline: Deadline) -> str:
    time.sleep(0.5 if name == "slow_search" else 0.02)  # real: client.call(..., timeout=deadline.remaining())
    return name


deadline = Deadline(0.2)
with ThreadPoolExecutor(max_workers=4) as ex:
    fs = {ex.submit(tool_call, n, deadline): n for n in ("calc", "sql", "slow_search")}
    done, not_done = wait(fs, timeout=deadline.remaining())
    print(sorted(fs[f] for f in done), [fs[f] for f in not_done])
# ['calc', 'sql'] ['slow_search']

import threading
import time
from concurrent.futures import ThreadPoolExecutor


class Stopped(Exception):
    pass


def paginate(source: str, stop: threading.Event) -> int:
    pages = 0
    while pages < 100:
        if stop.is_set():
            raise Stopped(f"{source}: stopped after {pages} pages")
        time.sleep(0.01)                                 # fetch one page (with a client timeout)
        pages += 1
        if stop.wait(0.0):                               # back-off sleeps: stop.wait(delay), never time.sleep
            raise Stopped(f"{source}: stopped after {pages} pages")
    return pages


stop = threading.Event()
with ThreadPoolExecutor(max_workers=2) as ex:
    futs = [ex.submit(paginate, s, stop) for s in ("crm", "wiki")]
    time.sleep(0.1)
    stop.set()
    outcomes = []
    for f in futs:
        try:
            outcomes.append(f.result())
        except Stopped as exc:
            outcomes.append(str(exc).split(":")[0])
print(outcomes)                                          # ['crm', 'wiki']

import random
import threading
import time
from concurrent.futures import ThreadPoolExecutor
from typing import Callable, Optional, Tuple, Type, TypeVar

R = TypeVar("R")


class TransientError(Exception):
    def __init__(self, msg: str, retry_after: Optional[float] = None) -> None:
        super().__init__(msg)
        self.retry_after = retry_after


def retry(fn: Callable[[], R], *, attempts: int = 5, base: float = 0.05, cap: float = 2.0,
          retry_on: Tuple[Type[BaseException], ...] = (TransientError, TimeoutError, ConnectionError),
          stop: Optional[threading.Event] = None) -> R:
    for attempt in range(attempts):
        try:
            return fn()
        except retry_on as exc:
            if attempt == attempts - 1:
                raise
            delay = random.uniform(0, min(cap, base * 2 ** attempt))
            delay = max(delay, getattr(exc, "retry_after", None) or 0.0)
            if stop is not None and stop.wait(delay):
                raise
            if stop is None:
                time.sleep(delay)
    raise AssertionError("unreachable")


calls = {"n": 0}
lock = threading.Lock()


def flaky() -> str:
    with lock:
        calls["n"] += 1
        n = calls["n"]
    if n < 3:
        raise TransientError("429", retry_after=0.01)
    return "ok"


with ThreadPoolExecutor(max_workers=1) as ex:
    print(ex.submit(retry, flaky).result(), calls["n"])  # ok 3

import threading
import time


class CircuitOpen(Exception):
    pass


class CircuitBreaker:
    def __init__(self, failures: int = 5, reset_after: float = 30.0) -> None:
        self.failures, self.reset_after = failures, reset_after
        self._count = 0
        self._opened_at: float = 0.0
        self._trial = False
        self._lock = threading.Lock()

    def call(self, fn, *args, **kwargs):
        with self._lock:
            if self._count >= self.failures:
                if time.monotonic() - self._opened_at < self.reset_after or self._trial:
                    raise CircuitOpen("dependency unavailable; failing fast")
                self._trial = True                       # half-open: exactly one trial call
        try:
            result = fn(*args, **kwargs)
        except Exception:
            with self._lock:
                self._count += 1
                self._trial = False
                if self._count >= self.failures:
                    self._opened_at = time.monotonic()
            raise
        with self._lock:
            self._count, self._trial = 0, False
        return result


breaker = CircuitBreaker(failures=3, reset_after=0.1)


def down() -> None:
    raise ConnectionError("vector db unreachable")


outcomes = []
for _ in range(5):
    try:
        breaker.call(down)
    except (ConnectionError, CircuitOpen) as exc:
        outcomes.append(type(exc).__name__)
time.sleep(0.12)
try:
    breaker.call(lambda: "recovered")
    outcomes.append("closed")
except CircuitOpen:
    outcomes.append("still open")
print(outcomes)
# ['ConnectionError', 'ConnectionError', 'ConnectionError', 'CircuitOpen', 'CircuitOpen', 'closed']

import time
from concurrent.futures import ThreadPoolExecutor, as_completed

try:
    from tqdm import tqdm
except ImportError:
    tqdm = None


def grade(i: int) -> bool:
    time.sleep(0.002)
    return i % 9 != 0


items = list(range(200))
ok = failed = 0
with ThreadPoolExecutor(max_workers=8) as ex:
    futures = [ex.submit(grade, i) for i in items]
    bar = tqdm(total=len(futures), unit="item") if tqdm else None
    for n, fut in enumerate(as_completed(futures), start=1):
        if fut.result():
            ok += 1
        else:
            failed += 1
        if bar:
            bar.update(1)
            bar.set_postfix(ok=ok, failed=failed)
        elif n % 100 == 0:
            print(f"{n}/{len(futures)} ok={ok} failed={failed}", flush=True)
    if bar:
        bar.close()
# 100/200 ok=... failed=...   (counts depend on completion order)
# 200/200 ok=177 failed=23

import sys
from concurrent.futures import ThreadPoolExecutor, as_completed
from dataclasses import dataclass
from typing import Optional


@dataclass
class Outcome:
    key: str
    value: Optional[int] = None
    error: Optional[BaseException] = None


def parse(doc: str) -> int:
    if doc.endswith(".bad"):
        raise ValueError(f"cannot parse {doc}")
    return len(doc)


docs = ["a.pdf", "b.bad", "c.pdf", "d.bad"]
outcomes = []
with ThreadPoolExecutor(max_workers=4) as ex:
    futures = {ex.submit(parse, d): d for d in docs}
    for fut in as_completed(futures):
        try:
            outcomes.append(Outcome(futures[fut], value=fut.result()))
        except Exception as exc:
            outcomes.append(Outcome(futures[fut], error=exc))

errors = [o.error for o in outcomes if o.error]
print(sum(o.error is None for o in outcomes), "ok;", len(errors), "failed")   # 2 ok; 2 failed
if errors and sys.version_info >= (3, 11):
    try:
        raise ExceptionGroup("2 documents failed", errors)    # one raise carries every traceback
    except ExceptionGroup as group:                           # callers can also use except* (3.11)
        print(len(group.exceptions), "errors grouped")        # 2 errors grouped

import threading
import time
from concurrent.futures import FIRST_COMPLETED, ThreadPoolExecutor, wait


def call(i: int) -> int:
    time.sleep(0.002)
    if i >= 20:
        raise PermissionError("401 invalid API key")        # systemic failure from item 20 on
    return i


stop = threading.Event()
ok, failed, max_fail_ratio, min_samples = 0, 0, 0.2, 20
with ThreadPoolExecutor(max_workers=4) as ex:
    pending = {ex.submit(call, i) for i in range(1000)}
    while pending and not stop.is_set():
        done, pending = wait(pending, return_when=FIRST_COMPLETED)
        for f in done:
            if f.exception() is None:
                ok += 1
            else:
                failed += 1
        total = ok + failed
        if total >= min_samples and failed / total > max_fail_ratio:
            stop.set()
            ex.shutdown(wait=True, cancel_futures=True)
print(f"aborted={stop.is_set()} ok={ok} failed<100={failed < 100}")
# aborted=True ok=20 failed<100=True

import signal
import threading
import time
from concurrent.futures import FIRST_COMPLETED, ThreadPoolExecutor, wait


class Shutdown(Exception):
    pass


def on_sigterm(signum, frame):                            # runs in the main thread
    raise Shutdown(f"signal {signum}")


signal.signal(signal.SIGTERM, on_sigterm)
stop = threading.Event()


def work(i: int, stop: threading.Event) -> int:
    for _ in range(4):
        if stop.is_set():
            return -1
        time.sleep(0.01)
    return i


completed = []
try:
    with ThreadPoolExecutor(max_workers=4) as ex:
        pending = {ex.submit(work, i, stop) for i in range(200)}
        threading.Timer(0.2, signal.raise_signal, args=(signal.SIGTERM,)).start()   # demo: SIGTERM later
        try:
            while pending:
                done, pending = wait(pending, timeout=0.25, return_when=FIRST_COMPLETED)
                completed += [f.result() for f in done]
        except (KeyboardInterrupt, Shutdown):
            stop.set()
            ex.shutdown(wait=True, cancel_futures=True)
            raise
except (KeyboardInterrupt, Shutdown) as exc:
    print(f"stopped by {type(exc).__name__}: {len(completed) < 200} completed early, exit code 130")
# stopped by Shutdown: True completed early, exit code 130