Logging from asyncio Tasks Without Blocking the Event Loop

Under asyncio, a slow log handler does not slow down logging — it stops the entire service. Every coroutine the loop was about to resume waits behind that write. This page shows how to detect the stall, how to move it off the loop with a queue, and how to keep request context attached to the records once a background thread owns the writing. It builds on context variables and thread safety, part of the Python logging fundamentals and structured data section.

Prerequisites

pip install "uvicorn[standard]>=0.30.0,<1.0.0" \
            "python-json-logger>=2.0.7,<4.0.0"

export PYTHONASYNCIODEBUG=1        # development only — reports slow callbacks
export LOG_QUEUE_MAXSIZE=10000

Implementation

Step 1 — Find the blocking handler before changing anything. asyncio's debug mode logs a warning whenever a callback occupies the loop for longer than loop.slow_callback_duration (0.1 s by default). Lower the threshold and the offending coroutine names itself.

import asyncio, logging

async def main() -> None:
    loop = asyncio.get_running_loop()
    loop.set_debug(True)
    loop.slow_callback_duration = 0.01          # 10 ms — tight enough to catch log writes
    await serve()

Expected Output:

WARNING asyncio Executing <Task finished coro=<handle_order()> ...> took 0.083 seconds

Eighty-three milliseconds inside handle_order with no await in the log path means the handler wrote synchronously. Confirm by pointing the same handler at /dev/null and re-running: if the warning disappears, the sink is the cause.

Step 2 — Put a QueueHandler in front of every sink. The application logger gets exactly one handler. Everything else — the file, the socket, the OTLP exporter — moves behind a QueueListener running on its own thread.

import logging
import logging.handlers
import queue

def install_async_safe_logging(*sinks: logging.Handler) -> logging.handlers.QueueListener:
    log_queue: queue.Queue = queue.Queue(maxsize=10_000)      # bounded, always
    listener = logging.handlers.QueueListener(
        log_queue, *sinks, respect_handler_level=True,
    )
    listener.start()

    root = logging.getLogger()
    for existing in list(root.handlers):
        root.removeHandler(existing)                          # the queue is the only handler now
    root.addHandler(logging.handlers.QueueHandler(log_queue))
    root.setLevel(logging.INFO)
    return listener

respect_handler_level=True lets the listener skip work per sink — an ERROR-only file handler will not format INFO records at all. Without it, every sink formats every record regardless of its own level.

Step 3 — Decide what a full queue does. The default is a silent discard through handleError. Make it explicit, shed the cheap records first, and count the loss so it shows up as a number rather than as a gap.

class SheddingQueueHandler(logging.handlers.QueueHandler):
    """Never blocks a coroutine; drops DEBUG and INFO first, and counts what it dropped."""

    dropped = 0

    def enqueue(self, record: logging.LogRecord) -> None:
        try:
            self.queue.put_nowait(record)
        except queue.Full:
            if record.levelno >= logging.WARNING:
                try:
                    self.queue.get_nowait()                   # evict one low-value record
                    self.queue.put_nowait(record)
                    return
                except queue.Empty:
                    pass
            type(self).dropped += 1                           # export this as a metric

Blocking is never the right answer here. A put() that waits turns a full queue into a stalled event loop, which is the exact failure the queue was added to prevent.

Step 4 — Keep the request context attached. asyncio copies the current context at task creation, so a value set before create_task is visible inside the task. Anything set afterwards is not, and anything set inside the task never escapes it.

import contextvars

request_id_var: contextvars.ContextVar[str] = contextvars.ContextVar("request_id")

async def handle(request) -> None:
    request_id_var.set(request.headers["x-request-id"])       # set BEFORE creating tasks
    task = asyncio.create_task(audit(request))                # inherits the context copy
    await respond(request)

async def audit(request) -> None:
    logging.getLogger("audit").info("recorded")               # carries the request id

Note that the record is enriched on the producing side, by a filter attached to the logger — not by the listener thread, which has no request context of its own. That placement rule is the same one that governs redaction, and it is covered in using contextvars for request tracing.

Step 5 — Drain on shutdown. The listener holds records that have not been written. Stop it from the application's shutdown hook, before the loop closes.

from contextlib import asynccontextmanager
from fastapi import FastAPI

@asynccontextmanager
async def lifespan(app: FastAPI):
    listener = install_async_safe_logging(logging.StreamHandler())
    yield
    listener.stop()                         # drains the queue, then joins the thread

app = FastAPI(lifespan=lifespan)

Configuration options

Verification

Prove the loop is no longer waiting, using a deliberately slow handler.

import asyncio, logging, time

class SlowHandler(logging.Handler):
    def emit(self, record):
        time.sleep(0.05)                    # a sink having a bad day

async def probe() -> float:
    ticks = 0
    async def heartbeat():
        nonlocal ticks
        while True:
            await asyncio.sleep(0.001)
            ticks += 1
    task = asyncio.create_task(heartbeat())
    start = time.perf_counter()
    for _ in range(20):
        logging.getLogger("probe").info("record")
    await asyncio.sleep(0)
    task.cancel()
    return time.perf_counter() - start

Expected Output:

direct SlowHandler : 1.003 s for 20 records — the loop was blocked the whole time
via QueueHandler   : 0.002 s for 20 records — the writes are still happening, elsewhere

Add a test that asserts the enqueue path stays fast even when the sink is unavailable, so a future refactor that removes the queue fails loudly rather than during an incident.

Common mistakes

The loop stalls only under load

Error signature: P99 latency spikes with no corresponding CPU increase, and asyncio debug mode reports slow callbacks in unrelated coroutines. Root cause: a synchronous handler on the loop thread; the stall is invisible at low concurrency because nothing was waiting behind it. Remediation: move every sink behind a QueueListener, as in non-blocking logging with QueueHandler.

Background task records have no request ID

Error signature: records from create_task work carry every field except the request context. Root cause: the contextvar was set after the task was created, so the task's context copy predates the value. Remediation: set the context first, then create tasks. Where a task must outlive the request, pass the ID explicitly as an argument.

Records disappear under burst load

Error signature: gaps in a sequence that should be contiguous, with no error anywhere. Root cause: a full queue discarding silently through handleError. Remediation: override enqueue, shed by level, and export the drop counter as a metric — a loss you cannot count is a loss you will not notice.

Finding the blocking call

The queue removes the symptom; the blocking call is still there and worth finding, because a log sink is rarely the only synchronous thing in an async service. Three techniques, in increasing order of effort.

asyncio debug mode with a lowered threshold. The cheapest and usually sufficient. It names the coroutine whose callback exceeded the threshold, which in most cases is enough to identify the line.

loop.set_debug(True)
loop.slow_callback_duration = 0.02        # 20 ms — staging, under load

A stack dump triggered by lag. When debug mode names something generic — a framework's own task wrapper, run_in_executor — the next step is to capture what the loop thread was actually doing. A watchdog thread that samples the loop thread's stack when lag exceeds a threshold gives that directly.

import faulthandler
import threading
import time

def watchdog(loop, threshold: float = 0.5, interval: float = 0.1) -> None:
    """Dump every thread's stack when the loop falls behind."""
    while True:
        due = loop.time() + interval
        time.sleep(interval)
        if loop.time() - due > threshold:
            faulthandler.dump_traceback()      # writes to stderr, from another thread
threading.Thread(target=watchdog, args=(loop,), daemon=True).start()

Note the shape: the watchdog runs on a separate thread, because a coroutine cannot observe a loop that is not running it. That is also why it can dump the loop thread's stack while the loop is still stuck in the offending call, which is exactly the moment the stack is informative.

A sampling profiler in production. Tools that sample a running process without instrumenting it will show where wall-clock time is going on the loop thread, and a synchronous call inside a coroutine stands out immediately as a frame with no await above it. This is the heaviest option and the one that finds the cases the other two miss.

The usual suspects

In a Python service, the same handful of calls account for most loop stalls, and knowing the list shortens the search considerably.

A synchronous database driver used because the async one was not available for a particular feature. A file read that looks harmless — a template, a certificate, a configuration reload — performed on the request path. A requests call inside a coroutine, usually in a helper written before the service went async. JSON serialisation of a large response, which is CPU-bound and never yields. An import performed lazily inside a function, which pays disk I/O and module execution on whichever request arrives first after a deploy. And garbage collection, which is not a call at all but blocks the thread that triggered it and is measurable through the GC pause metrics.

The remedy is the same in every case except the last: move the work to a thread with asyncio.to_thread or a dedicated executor, and remember that the context does not follow it unless you copy it explicitly.

import asyncio
import contextvars

async def read_template(path: str) -> str:
    ctx = contextvars.copy_context()                       # keep the log context
    return await asyncio.to_thread(ctx.run, _read_sync, path)

That pattern converts a stall that delays every concurrent request into work that occupies one thread and delays nobody — which is the same trade the log queue makes, applied to the rest of the service.

Related

• Context variables and thread safety in Python logging — the parent guide: contextvars, threads, and record enrichment.
• Using contextvars for request tracing — setting and resetting the values this page relies on.
• Non-blocking logging with QueueHandler — the queue wiring in full, including shutdown draining.
• Async logging with Loguru's enqueue — the same problem solved by a library flag.
• Async tracing patterns in Python — the same context-copy semantics, applied to spans.

Frequently Asked Questions