Thread-Safe and Multiprocessing-Safe Logging in Python

Log lines emitted from a process pool arrive interleaved, truncated, or silently missing, because every worker holds an independent file descriptor onto the same destination and nothing coordinates their writes. This page is for backend engineers and SREs running multiprocessing.Pool, ProcessPoolExecutor, or forked gunicorn workers who need one clean, parseable log stream out of many processes. It is part of the context variables and thread safety reference within the Python Logging Fundamentals and Structured Data guide, and it picks up exactly where per-thread handler locking stops working: the process boundary.

Prerequisites

Everything on this page is standard library — logging.handlers.QueueHandler, logging.handlers.QueueListener, and multiprocessing — so there is nothing to install beyond a pinned interpreter. Pin the Python range anyway, because the multiprocessing default start method changed to spawn on macOS in 3.8 and on Linux in 3.14, and that difference decides whether a child inherits the parent's handlers.

# pyproject.toml — no third-party dependency is needed for this pattern
[project]
name = "worker-service"
requires-python = ">=3.11,<3.14"
dependencies = []          # QueueHandler/QueueListener ship with the stdlib

# These imports are all you need.
import logging
import logging.handlers
import multiprocessing

Two environment inputs are worth wiring up front: LOG_LEVEL (read once in the parent and passed to workers, so a child never re-reads configuration you have since changed) and an explicit start method, either PYTHONMULTIPROCESSING-free code that calls multiprocessing.set_start_method("spawn", force=True) in __main__, or a deliberate fork choice. Pinning the start method makes the handler-inheritance behaviour identical on every platform, which matters more here than raw process start latency. If you configure logging through a dictionary rather than code, the queue wiring drops straight into the handlers block described in configuring logging with dictConfig.

Why handler locks stop at the process boundary

Thread safety and process safety look identical from the outside — both show up as garbled output — but they are different problems with different fixes. Every logging.Handler creates a lock in its constructor and acquires it around emit, so two threads in one interpreter serialise cleanly on the same handler object. That guarantee is scoped to one address space. When a process forks, the child receives a copy of the handler object, a copy of the lock, and a duplicated file descriptor; when a process is spawned, it builds new objects from scratch. In either case two processes can be inside emit on the same file at the same instant, each convinced it holds the lock.

Whether that race actually corrupts a line depends on plumbing you do not control. A file opened in append mode writes atomically only up to the operating system's buffer limit, and a StreamHandler emits the formatted record and the terminator in a way that can flush as more than one syscall, so long JSON lines are the first to split. Rotating handlers are worse than plain ones: RotatingFileHandler and TimedRotatingFileHandler rename the active file during rollover, and every other process keeps writing into the renamed inode — records vanish into a file nobody tails. This is why the rotation guidance in best practices for log rotation in Python insists that exactly one process own the rollover.

The fix follows from the diagnosis. Do not try to make many writers safe — remove the writers. Workers only enqueue records; one process owns the real handlers and does all formatting and I/O. That is the same decoupling used for non-blocking logging with QueueHandler inside a single interpreter, extended across the process boundary.

Implementation

Step 1 — Create a shared queue in the parent. Use multiprocessing.Queue, which is picklable and proxied to children through inheritance at process creation. A plain queue.Queue lives in one process's heap and would not survive the trip.

import multiprocessing

# -1 means unbounded; the queue is created in the parent, before any worker exists.
log_queue: "multiprocessing.Queue" = multiprocessing.Queue(-1)

Step 2 — Configure each worker to log into the queue. Inside the worker, clear whatever handlers the root logger inherited and attach exactly one QueueHandler. Setting the level here matters: QueueHandler enqueues everything the logger admits, so filtering at the worker keeps debug traffic off the pipe entirely.

import logging
import logging.handlers


def worker_init(queue: "multiprocessing.Queue", level: int = logging.INFO) -> None:
    root = logging.getLogger()
    root.handlers.clear()                                   # drop any inherited file handlers
    root.addHandler(logging.handlers.QueueHandler(queue))   # the worker's only handler
    root.setLevel(level)                                    # filter here, not on the far side


def do_work(item: str) -> None:
    # Formatting is deferred: the listener renders %s once, in one place.
    logging.getLogger("worker").info("processing %s", item)

Step 3 — Run the listener in the parent only. The QueueListener owns the real handlers and runs them on a background thread inside the parent, so there is exactly one writer. This is also where a JSON formatter belongs, following the field conventions in structured logging with the standard library.

import logging
import logging.handlers


def start_listener(queue: "multiprocessing.Queue") -> logging.handlers.QueueListener:
    handler = logging.StreamHandler()                       # swap for RotatingFileHandler in prod
    handler.setFormatter(
        logging.Formatter('{"level":"%(levelname)s","proc":"%(processName)s","msg":"%(message)s"}')
    )
    # respect_handler_level lets each real handler apply its own threshold.
    listener = logging.handlers.QueueListener(queue, handler, respect_handler_level=True)
    listener.start()                                        # spawns the draining thread
    return listener

Step 4 — Wire the pool to the queue and drain on shutdown. Pass the queue through the pool initializer so every worker installs its QueueHandler exactly once at startup, and stop the listener after the pool has joined so nothing is still in flight.

import multiprocessing

if __name__ == "__main__":                                  # required under the spawn start method
    multiprocessing.set_start_method("spawn", force=True)
    log_queue = multiprocessing.Queue(-1)
    listener = start_listener(log_queue)
    try:
        with multiprocessing.Pool(
            processes=3,
            initializer=worker_init,                        # module-level, therefore picklable
            initargs=(log_queue, logging.INFO),             # inherited at process creation
        ) as pool:
            pool.map(do_work, ["a", "b", "c"])              # the with-block joins the pool on exit
    finally:
        listener.stop()                                     # flush remaining records, join the thread

Expected Output:

{"level":"INFO","proc":"SpawnPoolWorker-1","msg":"processing a"}
{"level":"INFO","proc":"SpawnPoolWorker-2","msg":"processing b"}
{"level":"INFO","proc":"SpawnPoolWorker-3","msg":"processing c"}

Every line is whole because only the parent's listener touches the destination. Ordering across processes is not guaranteed — the queue interleaves producers — but no record is ever split, and processName tells you which worker emitted what. Note where the work now happens: rendering the format string and writing to disk runs once, in the listener thread, while workers do only the cheap part of building a LogRecord and putting it on the queue. Removing I/O contention from the workers is a throughput win on top of the correctness win.

One implementation detail of multiprocessing.Queue shapes shutdown: it is backed by an in-process buffer, a feeder thread, and an OS pipe. put returns as soon as the record is buffered, not when it reaches the parent, so a worker killed abruptly loses whatever its feeder thread had not yet flushed. Let workers exit normally and let the pool join before stopping the listener — the Pool context manager does the join for you when the with block exits without an exception. QueueHandler.prepare also reshapes each record before it travels: it formats the message, then clears args and exc_info so the record is picklable. That is why an unpicklable positional argument is usually harmless, while an unpicklable object attached through extra still raises at put time.

Carrying request context across the process boundary

Context variables do not cross processes. Under spawn the worker is a brand-new interpreter that inherits none of the parent's contextvars; under fork the child gets a snapshot at fork time, and every mutation afterwards is invisible in both directions. The intra-process model — set, token, reset — is covered in using contextvars for request tracing; across processes the only reliable transport is the argument list.

Pass correlation data explicitly and attach it with extra, so it lands on the record before the record enters the queue and is therefore already present when the listener formats it.

def do_work(item: str, request_id: str, trace_id: str) -> None:
    # extra keys become LogRecord attributes and must themselves be picklable.
    logging.getLogger("worker").info(
        "processing %s", item, extra={"request_id": request_id, "trace_id": trace_id}
    )

# The parent reads context once, then fans it out with the work items.
work = [("a", request_id, trace_id), ("b", request_id, trace_id)]
pool.starmap(do_work, work)

If the parent re-establishes the identifiers inside each worker instead — for example by calling contextvar.set(request_id) at the top of do_work — a filter on the worker's logger can inject them automatically, which is the pattern used for adding trace IDs to log records. Either way the identifier must be carried over the boundary; nothing inherits it.

Configuration options

Verification

The single-writer invariant is cheap to assert. Run this inside a worker (or from the initializer) to prove that no inherited handler survived — the failure mode responsible for most "it works locally, corrupts in production" reports.

def assert_worker_logging() -> None:
    handlers = logging.getLogger().handlers
    assert len(handlers) == 1, handlers                              # exactly one handler
    assert isinstance(handlers[0], logging.handlers.QueueHandler)    # and it must not do I/O

Expected Output: the function returns silently. An AssertionError listing a FileHandler or StreamHandler means that worker still writes directly to a destination and will race the listener.

For an end-to-end check, stress the pipeline and verify the output rather than eyeballing it. Have each worker emit a few thousand records whose message carries its own process identity and a monotonic counter, then parse the result: every line must be valid JSON, and every (worker, counter) pair must appear exactly once.

import json


def verify_output(path: str, workers: int, per_worker: int) -> None:
    seen: set[tuple[str, int]] = set()
    with open(path, encoding="utf-8") as fh:
        for line in fh:                                  # a split line fails json.loads here
            record = json.loads(line)
            seen.add((record["proc"], record["seq"]))
    assert len(seen) == workers * per_worker, len(seen)  # nothing lost, nothing duplicated

Expected Output: no exception, and wc -l on the file equals workers * per_worker. A json.JSONDecodeError proves a second writer is interleaving bytes; a short count without a decode error usually means the listener was stopped before the queue drained, or a worker was killed with records still in its feeder thread.

Common mistakes

• Error signature: RuntimeError: Queue objects should only be shared between processes through inheritance, raised when submitting a task. Root cause: a multiprocessing.Queue was passed as an argument to pool.map or apply_async instead of at process creation. Remediation: pass it through initializer/initargs, or switch to a multiprocessing.Manager().Queue() proxy if it genuinely must travel as a task argument.

• Error signature: a JSON log line ends mid-token, or two records share one line, only under load. Root cause: a worker inherited or re-added a real handler — commonly because logging was configured at import time and fork copied it — so two file descriptors write the same file. Remediation: call root.handlers.clear() in the worker initializer, attach only a QueueHandler, and assert the invariant with the check above.

• Error signature: the last second of logs is missing after every run, with no error reported. Root cause: the process exited while records were still in the queue or the listener's buffer. Remediation: join the pool (let the with block exit) and then call listener.stop() from a finally block, so the sentinel is enqueued and the drain completes before the interpreter shuts down.

• Error signature: PicklingError or TypeError: cannot pickle ... raised from the logger.info call itself. Root cause: an unpicklable object was attached via extra; unlike positional args, extra attributes survive QueueHandler.prepare and must cross the pipe intact. Remediation: bind identifiers and plain data only, converting rich objects to strings at the call site — the same discipline that keeps a production configuration portable, as covered in how to configure Python logging for production.

Related

• Context variables and thread safety — the parent reference on request-scoped state across coroutines, threads, and processes.
• Using contextvars for request tracing — the intra-process model whose guarantees end at the boundary described here.
• Non-blocking logging with QueueHandler — the same queue/listener split applied inside a single interpreter.
• Best practices for log rotation in Python — why rollover must be owned by exactly one process, and what to use when it cannot be.
• Async and non-blocking logging with Loguru enqueue — the Loguru equivalent of the single-writer pattern, with the queue managed for you.

Frequently Asked Questions