How do ThreadPoolExecutor and ProcessPoolExecutor work?

# ThreadPoolExecutor: submit, map order, as_completed order, errors, shutdown.
# Not runnable in the browser: needs OS threads.
import os
import threading
import time
from concurrent.futures import ThreadPoolExecutor, as_completed, wait, FIRST_COMPLETED

def fetch(name, seconds):
    time.sleep(seconds)                 # a blocking I/O call releases the GIL
    return f"{name} ({threading.current_thread().name.split('_')[0]})"

jobs = [("slow", 0.3), ("fast", 0.1), ("medium", 0.2)]

with ThreadPoolExecutor(max_workers=3, thread_name_prefix="io") as pool:
    start = time.perf_counter()
    # map: results come back in INPUT order, whatever finishes first
    print("map:         ", list(pool.map(fetch, *zip(*jobs))))
    print("ran in parallel:", time.perf_counter() - start < 0.5)

    # as_completed: futures come back in COMPLETION order
    futures = {pool.submit(fetch, n, s): n for n, s in jobs}
    print("as_completed:", [futures[f] for f in as_completed(futures)])

    # wait: block until the first one finishes
    futs = [pool.submit(fetch, n, s) for n, s in jobs]
    done, not_done = wait(futs, return_when=FIRST_COMPLETED)
    print("first done:", [f.result() for f in done], "still running:", len(not_done))

# Exceptions are captured in the Future and re-raised by result()
def parse(text):
    return int(text)

with ThreadPoolExecutor(2) as pool:
    fut = pool.submit(parse, "x")
    print("exception stored:", type(fut.exception()).__name__)
    try:
        fut.result()
    except ValueError as e:
        print("result() raised:", e)

# The default pool size (3.13+: os.process_cpu_count() instead of os.cpu_count())
count = getattr(os, "process_cpu_count", os.cpu_count)() or 1
# (_max_workers is a private attribute, read here only to show the default)
print("default max_workers formula:", min(32, count + 4) == ThreadPoolExecutor()._max_workers)

# Leaving the with-block calls shutdown(wait=True): it waits for running work
pool = ThreadPoolExecutor(1)
f1 = pool.submit(time.sleep, 0.2)
f2 = pool.submit(time.sleep, 0.2)
pool.shutdown(wait=True, cancel_futures=False)
print("after shutdown(wait=True):", f1.done(), f2.done())

pool = ThreadPoolExecutor(1)
f1 = pool.submit(time.sleep, 0.2)
f2 = pool.submit(time.sleep, 0.2)        # still queued behind f1
time.sleep(0.05)                         # let the worker pick up f1
pool.shutdown(wait=True, cancel_futures=True)
print("cancel_futures=True:", f1.cancelled(), f2.cancelled())
try:
    pool.submit(print, "too late")
except RuntimeError as e:
    print("submit after shutdown:", e)

# ProcessPoolExecutor: real parallel CPU work, and the pickling rules.
# Not runnable in the browser: needs OS processes.
import math
import os
from concurrent.futures import ProcessPoolExecutor

def is_prime(n):
    if n < 2:
        return False
    for d in range(2, math.isqrt(n) + 1):
        if n % d == 0:
            return False
    return True

def count_primes(bounds):
    lo, hi = bounds
    return sum(1 for n in range(lo, hi) if is_prime(n))

def worker_pid(_):
    return os.getpid()

COUNTER = 0

def bump(_):
    global COUNTER
    COUNTER += 1          # changes the WORKER copy of the module
    return COUNTER

def fails(x):
    raise ValueError(f"bad input {x}")

if __name__ == "__main__":       # required: spawn and forkserver re-import this file
    chunks = [(i, i + 50_000) for i in range(0, 200_000, 50_000)]
    with ProcessPoolExecutor(max_workers=4) as pool:
        print("primes per chunk:", list(pool.map(count_primes, chunks)))
        print("total below 200000:", sum(pool.map(count_primes, chunks)))

        pids = set(pool.map(worker_pid, range(8)))
        print("ran outside the parent:", os.getpid() not in pids)

        # chunksize batches many small calls into one round trip per batch
        print("chunksize=100:", sum(pool.map(is_prime, range(1000), chunksize=100)))

        # an exception in a worker is pickled back and re-raised here
        try:
            pool.submit(fails, 7).result()
        except ValueError as e:
            print("worker error:", e)

        # arguments and functions travel by pickle: a lambda cannot
        try:
            pool.submit(lambda x: x * 2, 21).result()
        except Exception as e:
            print("lambda:", type(e).__name__)

        # globals are not shared: each worker process has its own copy
        list(pool.map(bump, range(8)))
        print("parent COUNTER after 8 bumps in workers:", COUNTER)

    # max_tasks_per_child (3.11+) replaces a worker after N tasks
    with ProcessPoolExecutor(max_workers=1, max_tasks_per_child=1) as pool:
        a = pool.submit(worker_pid, None).result()
        b = pool.submit(worker_pid, None).result()
        print("max_tasks_per_child=1 gave a fresh process:", a != b)