Thread Safety, Shared State, Context and asyncio Bridges

Which objects can be shared between pool threads, how to share the ones that cannot, the synchronization primitives and when each is right, how request ids and traces reach worker threads, and how threads and an event loop talk to each other.

#Contents

1. What the GIL does and does not protect
2. Client and library thread-safety
3. Four ways to give workers a client
4. Lock, RLock, Condition, Event, Semaphore, Barrier
5. queue.Queue
6. threading.local
7. contextvars
8. Logging from threads
9. Threads and asyncio
10. Free-threaded builds

#1. What the GIL does and does not protect

On the default CPython build one thread executes bytecode at a time, and many single operations on built-in types are atomic: lst.append(x), d[k] = v, d.get(k), q.put(x). Free-threaded builds (3.13t/3.14t) keep these single operations safe with per-object locks.

Compound operations are never atomic: counter += 1 (load, add, store), d[k] = d.get(k, 0) + 1, if k not in cache: cache[k] = compute(), iterating a dict while another thread inserts (RuntimeError: dictionary changed size during iteration). A thread switch between the read and the write loses an update. The window is small on the GIL build, which is why these bugs pass tests and appear in production; the demo below widens it deliberately:

import threading
import time
from concurrent.futures import ThreadPoolExecutor

stats = {"tokens": 0}
lock = threading.Lock()


def unsafe_add(n: int) -> None:
    current = stats["tokens"]          # read
    time.sleep(0.001)                  # any I/O or switch here...
    stats["tokens"] = current + n      # ...and this write overwrites someone else's


def safe_add(n: int) -> None:
    with lock:
        current = stats["tokens"]
        time.sleep(0.001)
        stats["tokens"] = current + n


for fn in (unsafe_add, safe_add):
    stats["tokens"] = 0
    with ThreadPoolExecutor(max_workers=8) as ex:
        list(ex.map(fn, [1] * 200))
    print(fn.__name__, stats["tokens"] == 200)
# unsafe_add False
# safe_add True

The best fix is usually structural: return values from tasks and aggregate in one thread. Locks are for state that genuinely must be shared (caches, rate limiters, connection pools). The linter's TPE006 flags read-modify-write on shared state in workers; TPE007 flags appends/item assignment on module-level containers.

#2. Client and library thread-safety

#3. Four ways to give workers a client

a) Per thread with threading.local + initializer (requests, sqlite, unknown SDKs):

import sqlite3
import threading
from concurrent.futures import ThreadPoolExecutor

_local = threading.local()
DB_PATH = ":memory:"                                    # a file path in real code


def init_worker() -> None:
    _local.conn = sqlite3.connect(DB_PATH)              # created in, and owned by, this worker thread


def lookup(doc_id: int) -> str:
    row = _local.conn.execute("SELECT ? || '-' || ?", (doc_id, threading.current_thread().name)).fetchone()
    return row[0]


with ThreadPoolExecutor(max_workers=3, thread_name_prefix="db", initializer=init_worker) as ex:
    rows = list(ex.map(lookup, range(6)))
print(len(rows), all("-db_" in r for r in rows))        # 6 True

Connections opened in the initializer are never closed explicitly (threads end at shutdown and the objects are collected). If you need deterministic closing, register each one in a list under a lock and close them after shutdown() (the thread-local-client recipe does this).

b) One shared thread-safe client (httpx, boto3 client, OpenAI/Anthropic SDK): create it before the pool, pass it in, size its connection pool to the worker count.

c) A pool of N clients in a queue.Queue when clients are expensive and not thread-safe:

import queue
from contextlib import contextmanager
from concurrent.futures import ThreadPoolExecutor


class ClientPool:
    def __init__(self, factory, size: int) -> None:
        self._q: queue.Queue = queue.Queue()
        for _ in range(size):
            self._q.put(factory())

    @contextmanager
    def client(self, timeout: float = 30.0):
        c = self._q.get(timeout=timeout)                # blocks when all are checked out
        try:
            yield c
        finally:
            self._q.put(c)                              # always returned, even on error


created = []
pool = ClientPool(lambda: created.append(1) or {"id": len(created)}, size=2)


def call(i: int) -> int:
    with pool.client() as c:
        return c["id"]


with ThreadPoolExecutor(max_workers=8) as ex:
    ids = set(ex.map(call, range(50)))
print(len(created), ids <= {1, 2})                      # 2 True

d) A single owner thread (sqlite writes, a non-thread-safe file writer): workers put records on a queue, one writer thread consumes them and is the only code touching the resource. See section 5.

#4. Lock, RLock, Condition, Event, Semaphore, Barrier

Always use with lock: (released on exceptions). Acquire with a timeout in code that must not hang: if not lock.acquire(timeout=5): raise TimeoutError(...). When several locks are needed, take them in one global order everywhere. Deadlocks from lock order, non-reentrant re-acquisition and lost wake-ups are covered in threading-and-multiprocessing-deadlocks.md.

A Condition that lets the main thread wait until N results are ready (or a deadline passes):

import threading
import time
from concurrent.futures import ThreadPoolExecutor

cond = threading.Condition()
ready = []


def produce(i: int) -> None:
    time.sleep(0.01 * (i % 3))
    with cond:
        ready.append(i)
        cond.notify_all()


with ThreadPoolExecutor(max_workers=4) as ex:
    for i in range(10):
        ex.submit(produce, i).add_done_callback(lambda f: f.result())
    with cond:
        got_five = cond.wait_for(lambda: len(ready) >= 5, timeout=2)
    print(got_five)                                     # True

(The done-callback reads f.result() so a failure is at least logged by concurrent.futures; in real code keep the futures and check them.)

#5. queue.Queue

• Queue(maxsize=N): put blocks when full (backpressure), get blocks when empty; both take timeout=.
• task_done() + join(): join returns when every item put has been marked done. Forgetting task_done in an error path hangs join forever: put it in finally.
• Shut consumers down with a sentinel per consumer (or Queue.shutdown() on 3.13+, which makes get raise queue.ShutDown once the queue is empty).
• SimpleQueue: unbounded, no task_done, reentrant (safe from __del__/signal handlers).
• PriorityQueue: (priority, seq, item) tuples, with a sequence number so items never get compared.

Single-writer pattern: many workers produce, one thread owns the sqlite connection.

import queue
import sqlite3
import threading
from concurrent.futures import ThreadPoolExecutor

DONE = object()
rows: queue.Queue = queue.Queue(maxsize=256)
written = []


def writer() -> None:
    conn = sqlite3.connect(":memory:")                  # owned by this thread only
    conn.execute("CREATE TABLE chunks (id INTEGER, text TEXT)")
    while (item := rows.get()) is not DONE:
        conn.execute("INSERT INTO chunks VALUES (?, ?)", item)
    conn.commit()
    written.append(conn.execute("SELECT COUNT(*) FROM chunks").fetchone()[0])
    conn.close()


def chunk(i: int) -> None:
    rows.put((i, f"chunk {i}"))                         # blocks if the writer falls behind


w = threading.Thread(target=writer, name="sqlite-writer")
w.start()
try:
    with ThreadPoolExecutor(max_workers=8) as ex:
        list(ex.map(chunk, range(500)))
finally:
    rows.put(DONE)
    w.join()
print(written)                                          # [500]

#6. threading.local

threading.local() gives each thread its own attribute namespace. In a pool, a worker thread runs many tasks, so values persist across tasks on the same thread: perfect for clients, wrong for per-task state.

import threading
from concurrent.futures import ThreadPoolExecutor

_local = threading.local()


def handle(user: str) -> str:
    previous = getattr(_local, "user", None)            # leftover from an earlier task on this thread
    _local.user = user
    return f"{user} saw {previous}"


with ThreadPoolExecutor(max_workers=1) as ex:
    print(list(ex.map(handle, ["alice", "bob"])))       # ['alice saw None', 'bob saw alice']

Never store request/tenant/user data in threading.local inside pool tasks; pass it as an argument or use a ContextVar with copy_context().run (next section).

#contextvars

A ContextVar holds per-context values (request id, tenant, trace span, OpenTelemetry context). Pool tasks run in the worker thread's context, not the submitter's:

• The caller's values are not visible: .get() returns the default (or raises LookupError without one).
• A .set() inside a task persists into later tasks on the same worker thread.
• asyncio.to_thread copies the caller's context; executor.submit, executor.map, loop.run_in_executor and threading.Thread (on the default build) do not.

Fix: run each task in a copy of the submitter's context.

import contextvars
from concurrent.futures import ThreadPoolExecutor

tenant = contextvars.ContextVar("tenant", default="-")


def task(i: int) -> str:
    seen = tenant.get()
    tenant.set(f"leak-{i}")                              # leaks to later tasks unless run in a copy
    return seen


tenant.set("acme")
with ThreadPoolExecutor(max_workers=1) as ex:
    plain = [ex.submit(task, i).result() for i in range(2)]                              # noqa: TPE019
    copied = [ex.submit(contextvars.copy_context().run, task, i).result() for i in range(2)]
print(plain)    # ['-', 'leak-0']
print(copied)   # ['acme', 'acme']

• One copy_context() per task: sharing one copy between concurrently running tasks raises RuntimeError: cannot enter context: ... is already entered.
• A subclass that overrides submit covers map too (map calls submit): see scripts/executor_tool.py scaffold --recipe context-propagation.
• threading.Thread(target=contextvars.copy_context().run, args=(fn, *args)) for raw threads; 3.14 adds Thread(context=...), and free-threaded builds inherit a copy by default (-X thread_inherit_context).
• OpenTelemetry stores the active span in a ContextVar, so the same rule applies; its threading instrumentation package automates the copy if you prefer instrumentation over code.
• The linter flags workers that read a module-level ContextVar but were submitted without a copy (TPE019).

#8. Logging from threads

logging is thread-safe: each handler serializes emit with an RLock. What to add:

• %(threadName)s in the format, and thread_name_prefix on every executor, so lines read embed_3.
• Request ids via a logging.Filter that reads a ContextVar (works only if the context was propagated).
• For slow handlers (HTTP log shippers, cloud logging), a QueueHandler in front and a QueueListener that does the I/O in one thread, so worker threads never block on logging.

import logging
import logging.handlers
import queue
import sys
from concurrent.futures import ThreadPoolExecutor

log_q: queue.Queue = queue.Queue(-1)
stream = logging.StreamHandler(sys.stdout)
stream.setFormatter(logging.Formatter("%(threadName)s %(levelname)s %(message)s"))
listener = logging.handlers.QueueListener(log_q, stream)
log = logging.getLogger("ingest")
log.addHandler(logging.handlers.QueueHandler(log_q))
log.setLevel(logging.INFO)
log.propagate = False
listener.start()
try:
    with ThreadPoolExecutor(max_workers=1, thread_name_prefix="embed") as ex:
        ex.submit(log.info, "batch %d embedded", 7).result()
finally:
    listener.stop()                                     # flushes queued records
# embed_0 INFO batch 7 embedded

Unhandled exceptions in raw threads go to threading.excepthook (3.8+); install one that logs them: threading.excepthook = lambda a: log.error("thread %s died", a.thread.name, exc_info=(a.exc_type, a.exc_value, a.exc_traceback)). Pool tasks never reach it: their exceptions are stored on the Future.

#9. Threads and asyncio

import asyncio
import threading
from concurrent.futures import ThreadPoolExecutor


class LoopThread:
    """Run coroutines from sync/threaded code on one background event loop."""

    def __init__(self) -> None:
        self.loop = asyncio.new_event_loop()
        self.thread = threading.Thread(target=self.loop.run_forever, name="asyncio-loop")
        self.thread.start()

    def run(self, coro, timeout: float = 30.0):
        return asyncio.run_coroutine_threadsafe(coro, self.loop).result(timeout)

    def close(self) -> None:
        self.loop.call_soon_threadsafe(self.loop.stop)
        self.thread.join()
        self.loop.close()


async def async_embed(text: str) -> int:
    await asyncio.sleep(0.01)                           # an async-only SDK call
    return len(text)


portal = LoopThread()
try:
    with ThreadPoolExecutor(max_workers=4) as ex:      # sync workers calling async code
        print(list(ex.map(lambda t: portal.run(async_embed(t)), ["a", "bb", "ccc"])))   # [1, 2, 3]
finally:
    portal.close()

Never call portal.run (or any .result() on a run_coroutine_threadsafe future) from a coroutine running on that same loop: the loop waits for itself. asyncio.run() shuts the default executor down at exit (3.9+; waits up to 300 s since 3.12), so a stuck to_thread call delays exit. For thousands of concurrent calls use asyncio end to end: python-asyncio-pattern-advisor.

#10. Free-threaded builds

On python3.13t / python3.14t (sys._is_gil_enabled() returns False):

• Threads run Python code in parallel, so CPU-bound pools finally scale, and races that were rare become common. Every TPE006 finding is a real bug there.
• Built-in containers stay internally consistent (per-object locks), but compound operations still race.
• C extensions must declare support; importing one that does not re-enables the GIL (with a warning) unless PYTHON_GIL=0 / -X gil=0 forces it off.
• Threads inherit a copy of the creator's context by default (sys.flags.thread_inherit_context), but pool workers are created once and reused, so still copy the context per task.

Measuring whether a workload scales, and what to do when it does not: python-gil-bottleneck-analyzer.