python-threadpool-executor-builder

Build thread pools for I/O-bound work that are correct before they are fast: pick threads only when they are the right tool, size the pool from latency and the provider's rate limit instead of guessing, bound what is queued, make every exception surface, stop cleanly on Ctrl-C, and keep shared clients and state thread-safe. scripts/executor_tool.py check finds the mechanical mistakes statically (19 rules); scripts/executor_tool.py scaffold prints complete, runnable recipes (bounded map, RPM limiter, retries, progress, per-thread sessions, graceful shutdown, context propagation, and an LLM eval harness that combines them).

#Trigger Conditions

Activate this skill when the user:

• Fans out blocking I/O from synchronous code: LLM or embedding API calls, HTTP scraping for a corpus, S3 uploads/downloads, vector-store upserts, database reads, tool calls in a sync agent, Streamlit/Gradio callbacks
• Writes or reviews ThreadPoolExecutor, executor.submit/map, as_completed, wait, Future, threading.Thread, threading.local, Lock, queue.Queue, asyncio.to_thread or run_in_executor
• Builds an eval harness or batch job over thousands of prompts that must respect RPM/TPM limits
• Sees exceptions disappear, memory grow while "submitting", 429 storms, a job that ignores Ctrl-C or never exits, RuntimeError: cannot schedule new futures after shutdown / after interpreter shutdown, BrokenThreadPool, Connection pool is full, discarding connection, SQLite objects created in a thread can only be used in that same thread, request ids missing from worker logs
• Asks how many threads to use, how to add timeouts/retries/rate limits, or whether a client is thread-safe

#Workflow

1. CLASSIFY  -> Is the hot path waiting on I/O? (CPU-bound -> GIL skill / processes; async codebase -> asyncio)
2. MEASURE   -> p50/p95 latency per call, provider limits (RPM, TPM, concurrent), item count, payload size
3. SIZE      -> workers = target_rate x p95 latency (Little's law), capped by rate limit and connection pool
4. SHAPE     -> one executor per job/app (with-block or lifespan), bounded submission, per-thread clients
5. HARDEN    -> results AND errors collected, timeouts, retries for transient errors only, rate limiter,
               stop event + cancel_futures on Ctrl-C, contextvars copied, logs carry thread + request id
6. CHECK     -> python scripts/executor_tool.py check src/ --strict
7. VERIFY    -> deterministic tests (barriers, fake clocks), no leaked threads, measured throughput vs limit

#Step 1: Are Threads the Right Tool?

#Step 2: The Executor API in One Screen

concurrent.futures.TimeoutError is the builtin TimeoutError since 3.11. Full signatures, semantics and version tags: executor-api-reference.md.

#Step 3: Size the Pool from Latency and Limits

Little's law: in-flight requests = throughput x latency. To sustain 10 requests/s at a p95 of 2 s you need about 20 calls in flight, so ~20-25 workers. More threads than that do nothing but queue on the rate limiter.

def pool_size(rpm: float, p95_latency_s: float, connection_pool: int = 64, headroom: float = 1.2) -> int:
    """Workers needed to saturate a rate limit, capped by the connection pool."""
    needed = rpm / 60.0 * p95_latency_s * headroom
    return max(1, min(connection_pool, round(needed)))


print(pool_size(rpm=500, p95_latency_s=3.0))     # 30
print(pool_size(rpm=10_000, p95_latency_s=1.5))  # 64 (capped: raise the pool size or use asyncio)

The rate limiter, not the worker count, enforces the limit: always add a token bucket when a provider limits you.

#Step 4: Bound What You Submit

submit() never blocks and map() drains its input up front, so a loop over a 10-million-line JSONL file queues 10 million futures before the first finishes. Keep a sliding window of at most N futures:

import itertools
from concurrent.futures import FIRST_COMPLETED, ThreadPoolExecutor, wait


def call_llm(prompt: str) -> int:
    return len(prompt)                                   # stand-in for a blocking API call


def prompts():                                           # lazy source, e.g. a JSONL reader
    for i in range(10_000):
        yield f"prompt {i}"


total, max_in_flight = 0, 64
with ThreadPoolExecutor(max_workers=16, thread_name_prefix="llm") as ex:
    source = iter(prompts())
    pending = {ex.submit(call_llm, p) for p in itertools.islice(source, max_in_flight)}
    while pending:
        done, pending = wait(pending, return_when=FIRST_COMPLETED)
        for fut in done:
            total += fut.result()                        # surfaces task exceptions
        for p in itertools.islice(source, len(done)):    # refill one slot per finished task
            pending.add(ex.submit(call_llm, p))
print(total)                                             # 108890

On 3.14+ ex.map(fn, items, buffersize=64) gives the same bound for ordered results. A threading.BoundedSemaphore acquired before submit and released in a done-callback also works when you cannot restructure the loop. Patterns and trade-offs (ordered vs unordered, fail-fast vs collect, semaphore vs window vs queue.Queue(maxsize)): bounding-backpressure-and-rate-limits.md.

#Step 5: Never Lose an Exception

An exception raised in a task lives on its Future until someone calls result() or exception(). Every future must be read, and every error must be kept or re-raised:

from concurrent.futures import ThreadPoolExecutor, as_completed


def embed(doc: str) -> list:
    if not doc:
        raise ValueError("empty document")
    return [float(len(doc))]


docs = ["alpha", "", "gamma"]
vectors, errors = {}, {}
with ThreadPoolExecutor(max_workers=4, thread_name_prefix="embed") as ex:
    futures = {ex.submit(embed, d): i for i, d in enumerate(docs)}
    for fut in as_completed(futures):
        i = futures[fut]
        try:
            vectors[i] = fut.result()
        except Exception as exc:                         # keep it: count, log, report, maybe retry
            errors[i] = exc
print(sorted(vectors), {i: repr(e) for i, e in errors.items()})
# [0, 2] {1: "ValueError('empty document')"}

#Step 6: Timeouts, Retries, Rate Limits, Cancellation

• Threads cannot be killed. fut.result(timeout=5) stops waiting; the call keeps running and holds a worker. Put the real timeout in the I/O call (requests.get(url, timeout=(3.05, 30)), SDK timeout=), and use a threading.Event that long tasks check between steps.
• Retry only transient errors (429, 5xx, timeouts, connection resets) with exponential backoff and full jitter, honouring Retry-After; never retry a 400 or a context-length error. Sleep with stop.wait(delay), not time.sleep, so shutdown is immediate.
• Rate limit with one shared token bucket (RPM and, for LLMs, TPM): every attempt, including retries, takes a token. On a 429 with Retry-After, pause the whole bucket, not just one thread.
• Circuit breaker for a dependency that is down: after N consecutive failures fail fast for a cool-down.
• Ctrl-C: wait with a timeout in the main thread, then stop.set() and ex.shutdown(wait=True, cancel_futures=True); report partial results; exit 130.

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


def upload(part: int, stop: threading.Event) -> int:
    for _ in range(3):                                   # long task: check the stop flag between steps
        if stop.is_set():
            raise InterruptedError(f"part {part} stopped")
        time.sleep(0.01)
    return part


stop = threading.Event()
done_parts = []
with ThreadPoolExecutor(max_workers=4) as ex:
    pending = {ex.submit(upload, p, stop) for p in range(100)}
    try:
        while pending:
            done, pending = wait(pending, timeout=0.25, return_when=FIRST_COMPLETED)   # Ctrl-C lands here
            done_parts += [f.result() for f in done]
            if len(done_parts) >= 8:
                raise KeyboardInterrupt                  # simulate Ctrl-C for this example
    except KeyboardInterrupt:
        stop.set()
        ex.shutdown(wait=True, cancel_futures=True)
        print(f"stopped: {len(done_parts) >= 8}, cancelled: {sum(f.cancelled() for f in pending) > 50}")
# stopped: True, cancelled: True

Retry, token bucket, circuit breaker, deadlines, partial results and Ctrl-C in full: reliability-timeouts-retries-cancellation.md. Exception hierarchy design for retryable errors: python-exception-handling-designer.

#Step 7: Thread Safety of Clients and Shared State

import contextvars
import logging
import threading
from concurrent.futures import ThreadPoolExecutor

request_id = contextvars.ContextVar("request_id", default="-")
_local = threading.local()


def session() -> dict:
    """Per-thread client (a dict stands in for requests.Session())."""
    if not hasattr(_local, "s"):
        _local.s = {"owner": threading.current_thread().name}
    return _local.s


def fetch(url: str) -> str:
    return f"{request_id.get()} {url} via {session()['owner'].split('_')[0]}"


request_id.set("req-7")
with ThreadPoolExecutor(max_workers=2, thread_name_prefix="fetch", initializer=session) as ex:
    plain = ex.submit(fetch, "a").result()                        # noqa: TPE019 (the bug, for contrast)
    ctx = ex.submit(contextvars.copy_context().run, fetch, "b").result()
print(plain)   # - a via fetch
print(ctx)     # req-7 b via fetch

Locks, RLock, Condition, queues, contextvars and logging in depth: thread-safety-and-shared-state.md.

#Step 8: Mixing Threads and asyncio

asyncio.run() shuts the default executor down on exit (3.9+), waiting for running to_thread calls. The loop-thread deadlocks are covered in python-async-deadlock-debugger.

#Step 9: Free-Threaded Builds and InterpreterPoolExecutor

• On free-threaded CPython (3.13t experimental, 3.14t officially supported but still optional) threads run Python bytecode in parallel, so a thread pool can speed up CPU-bound work. Data races that the GIL used to hide (lost += updates, check-then-set) become frequent: fix TPE006 findings first. Diagnosis, scaling tests and extension compatibility: python-gil-bottleneck-analyzer.
• concurrent.futures.InterpreterPoolExecutor (3.14) runs each worker in its own sub-interpreter with its own GIL: CPU parallelism in one process, but functions and arguments must be shareable (pickled), and many C extensions do not support sub-interpreters yet. For I/O it offers nothing over threads.

#Symptom -> Cause -> Fix

Live debugging (thread dumps, py-spy, faulthandler, naming threads, metrics): debugging-thread-pools.md.

#AI-Engineering Patterns

Worked, runnable versions: ai-engineering-threadpools.md.

#Version Notes

#Output Format

When building or reviewing thread-pool code, answer with:

1. Verdict - threads vs asyncio vs processes for this workload, with the reason (I/O wait share, library).
2. Sizing - workers, max in-flight, rate limit and connection pool settings, with the Little's-law arithmetic.
3. Findings - numbered, with location, TPE rule id when the linter flagged it, the failure it causes (lost data, hang, memory blowup, 429s, race) and severity.
4. Code - complete, runnable before/after: executor lifetime, bounded submission, result/error collection, timeouts, retries, rate limiter, Ctrl-C handling, per-thread clients, context propagation.
5. Verification - the tests to add (deterministic, see testing-threaded-code.md), a thread-leak check, and the throughput to expect vs the provider limit.

#Scripts

# Lint (AST only; never imports or runs the analysed code)
python scripts/executor_tool.py check src/                            # files or directories
python scripts/executor_tool.py check worker.py --json                # machine-readable
cat job.py | python scripts/executor_tool.py check -                  # stdin
python scripts/executor_tool.py check src/ --strict                   # exit 1 on any finding (CI gate)
python scripts/executor_tool.py check src/ --select TPE001,TPE005 --ignore pool-size-smell
python scripts/executor_tool.py check src/ --min-severity medium
python scripts/executor_tool.py check --list-rules                    # why + fix snippet per rule (--json too)

Rules: TPE001 future-never-read, TPE002 executor-not-shutdown, TPE003 unbounded-submit, TPE004 map-over-generator, TPE005 map-results-unconsumed, TPE006 shared-read-modify-write, TPE007 shared-container-mutation (low), TPE008 shared-unsafe-client, TPE009 executor-self-wait, TPE010 pool-size-smell (low), TPE011 result-without-timeout (low), TPE012 shutdown-nowait-at-exit, TPE013 daemon-thread-side-effects, TPE014 polling-future-done, TPE015 swallowed-future-exception, TPE016 executor-per-call, TPE017 thread-not-joined, TPE018 cpu-bound-thread-pool (low), TPE019 contextvar-not-propagated. Silence a line with # noqa: TPE010. Exit codes: 0 clean (or not strict), 1 findings with --strict, 2 usage/read/syntax error (directory walks warn and skip unparsable files).

# Recipes: complete, typed, documented modules that run offline with stub work (verified with ast.parse)
python scripts/executor_tool.py scaffold --list
python scripts/executor_tool.py scaffold --recipe bounded-map > bounded_map.py && python bounded_map.py
python scripts/executor_tool.py scaffold --recipe llm-eval-harness --json      # {recipe, summary, code}

Recipes: bounded-map, rate-limited, retrying, progress, thread-local-client, graceful-shutdown (simulates Ctrl-C, exits 130), context-propagation, llm-eval-harness. Flags: --recipe NAME, --list, --json. Exit 2 for an unknown recipe or generated code that fails ast.parse.

#Additional Resources