Skip to content

Sending Results from Threads to an asyncio Queue

asyncio.to_thread() covers the common case: run a blocking call, get one result back. It does not cover the streaming case, which is just as common — a database driver that yields rows through a callback, a camera or serial SDK that pushes frames, a legacy library whose only API is for row in cursor: in a thread. Here the thread produces a series of results while an async consumer processes them, and the naive bridge — loop.call_soon_threadsafe(queue.put_nowait, item) for every item — works right up to the point where the producer is faster than the consumer. Then an unbounded queue grows until the process is killed. This guide builds the bridge properly: thread-safe delivery, real backpressure onto the producing thread, an explicit end-of-stream signal, exceptions that surface in the consumer, and a shutdown that stops the thread rather than leaking it.

Prerequisites

One item crossing from thread to loop 5 stages from SDK thread to release slot. One item crossing from thread to loop SDK thread acquire slot threadsafe hop schedule put loop thread queue.put_nowait consumer await get release slot producer may continue The slot is what makes the bound real for a thread that cannot await.

1. Deliver items with the one thread-safe call

The producing thread must never touch the asyncio.Queue directly: its put_nowait manipulates loop-owned state and its waiter wake-ups are not thread-safe. Schedule the put onto the loop instead.

import asyncio
import threading
import time


def sdk_stream(on_row, rows: int, stop: threading.Event) -> None:
    """A blocking library that pushes rows through a callback, in its own thread."""
    for i in range(rows):
        if stop.is_set():
            return
        on_row({"id": i})
    on_row(None)                                          # end of stream


class ThreadBridge:
    def __init__(self, maxsize: int = 100) -> None:
        self.loop = asyncio.get_running_loop()            # captured on the loop thread
        self.queue: asyncio.Queue = asyncio.Queue(maxsize=maxsize)

    def emit(self, item) -> None:                         # called from the SDK thread
        self.loop.call_soon_threadsafe(self.queue.put_nowait, item)

    async def __aiter__(self):
        while (item := await self.queue.get()) is not None:
            yield item


async def main() -> None:
    stop = threading.Event()
    bridge = ThreadBridge(maxsize=10**9)                  # effectively unbounded, for now
    thread = threading.Thread(target=sdk_stream, args=(bridge.emit, 2000, stop), daemon=True)
    thread.start()
    received, peak = 0, 0
    async for _row in bridge:
        received += 1
        peak = max(peak, bridge.queue.qsize())
        if received % 50 == 0:
            await asyncio.sleep(0.002)                    # a consumer that does real work
    thread.join()
    print(f"received {received} rows, peak queue depth {peak}")


asyncio.run(main())

It works — and the peak queue depth was about 1,700 of the 2,000 rows. The producing thread runs at full speed while the consumer awaits, so everything the thread produces piles up in memory. With a real stream of megabyte-sized frames, that is the whole heap.

Verify: the row count is correct and the peak queue depth is a large fraction of the total rows produced.

2. Push backpressure onto the producing thread

An asyncio.Queue's bound cannot help here: put_nowait from the loop either succeeds or raises, and the thread is what needs to wait. Give the bridge a threading.Semaphore with the same capacity: the thread acquires a slot before emitting, and the consumer releases one after taking an item.

import asyncio
import threading


class BoundedThreadBridge:
    def __init__(self, maxsize: int = 32) -> None:
        self.loop = asyncio.get_running_loop()
        self.queue: asyncio.Queue = asyncio.Queue(maxsize=maxsize)
        self._space = threading.Semaphore(maxsize)        # the thread's view of the bound

    def emit(self, item) -> None:                         # SDK thread
        self._space.acquire()                             # blocks the producer when full
        self.loop.call_soon_threadsafe(self.queue.put_nowait, item)

    async def __aiter__(self):
        while True:
            item = await self.queue.get()
            self._space.release()                         # one slot free: the thread may continue
            if item is None:
                return
            yield item


async def main() -> None:
    stop = threading.Event()
    bridge = BoundedThreadBridge(maxsize=32)
    thread = threading.Thread(target=sdk_stream, args=(bridge.emit, 2000, stop), daemon=True)
    thread.start()
    received, peak = 0, 0
    async for _row in bridge:
        received += 1
        peak = max(peak, bridge.queue.qsize())
        if received % 50 == 0:
            await asyncio.sleep(0.002)
    thread.join()
    print(f"received {received} rows, peak queue depth {peak}")


asyncio.run(main())

The same 2,000 rows arrived with a peak queue depth of 31 — the bound — because the producing thread blocked on acquire() whenever the consumer fell behind. Blocking a thread is exactly the right move here: that thread exists to run blocking code, and pausing it costs nothing but its stack.

Verify: the peak depth equals the configured bound rather than the number of rows, and no rows are lost.

Peak queue depth for 2,000 produced rows 2 bars comparing no backpressure with the others. Peak queue depth for 2,000 produced rows no backpressure ~1,700 rows buffered semaphore bound of 32 31 rows buffered Measured with the step 1 and 2 code: 2,000 rows, consumer pausing every 50 rows. Without a slot per item, the producer runs at full speed into memory.

3. Signal the end of the stream and propagate errors

A stream has three possible endings: it finished, it failed, or the consumer stopped caring. Encode the first two in the queue itself so the consumer sees them in order, instead of discovering them through a separate flag.

import asyncio
import threading
from dataclasses import dataclass


@dataclass
class StreamError:
    exception: BaseException


class ResultStream(BoundedThreadBridge):
    def fail(self, exc: BaseException) -> None:           # SDK thread
        self.emit(StreamError(exc))
        self.emit(None)

    async def __aiter__(self):
        while True:
            item = await self.queue.get()
            self._space.release()
            if item is None:
                return
            if isinstance(item, StreamError):
                raise item.exception                      # surfaces in the consumer's async for
            yield item


def flaky_sdk_stream(stream: ResultStream, rows: int, stop: threading.Event) -> None:
    try:
        for i in range(rows):
            if stop.is_set():
                return
            if i == 5:
                raise ConnectionError("cursor closed by the server")
            stream.emit({"id": i})
        stream.emit(None)
    except BaseException as exc:                          # the thread must not die silently
        stream.fail(exc)


async def main() -> None:
    stop = threading.Event()
    stream = ResultStream(maxsize=16)
    thread = threading.Thread(target=flaky_sdk_stream, args=(stream, 100, stop), daemon=True)
    thread.start()
    rows = []
    try:
        async for row in stream:
            rows.append(row)
    except ConnectionError as exc:
        print(f"received {len(rows)} rows, then: {exc}")
    thread.join()


asyncio.run(main())

The consumer received the first five rows and then the driver's ConnectionError, raised from its own async for with the original exception object. Wrapping the thread body in except BaseException matters: a thread that dies without signalling leaves the consumer awaiting a queue nothing will ever fill.

Verify: five rows arrive, the exception type and message are preserved, and the thread has exited.

4. Stop the producer when the consumer leaves

If the consumer breaks out of the loop — enough rows, a timeout, a cancelled request — the thread is still producing into a queue nobody drains, and it will block forever on the semaphore. Make the stream an async context manager that sets the stop flag and unblocks the thread.

import asyncio
import contextlib
import threading


class ManagedStream(ResultStream):
    def __init__(self, maxsize: int = 32) -> None:
        super().__init__(maxsize)
        self.stop = threading.Event()
        self.thread: threading.Thread | None = None

    def start(self, target, *args) -> None:
        self.thread = threading.Thread(target=target, args=(self, *args), daemon=True)
        self.thread.start()

    async def aclose(self, grace: float = 2.0) -> None:
        self.stop.set()                                   # ask the producer to finish
        self._space.release()                             # unblock it if it waits for space
        while not self.queue.empty():                     # drain so it can reach its exit
            self.queue.get_nowait()
            self._space.release()
        if self.thread is not None:
            await asyncio.to_thread(self.thread.join, grace)
            if self.thread.is_alive():
                raise TimeoutError("SDK thread did not stop")


def counting_stream(stream: ManagedStream, rows: int) -> None:
    produced = 0
    try:
        for i in range(rows):
            if stream.stop.is_set():
                break
            stream.emit({"id": i})
            produced += 1
        stream.emit(None)
    except BaseException as exc:
        stream.fail(exc)
    finally:
        stream.produced = produced


async def main() -> None:
    stream = ManagedStream(maxsize=8)
    stream.start(counting_stream, 100_000)
    taken = []
    try:
        async for row in stream:
            taken.append(row)
            if len(taken) == 20:
                break                                     # the consumer has what it needs
    finally:
        await stream.aclose()
    print(f"consumed {len(taken)} rows; producer stopped after {stream.produced} rows")


asyncio.run(main())

The consumer took twenty rows out of a hundred thousand, and the producer stopped shortly after rather than filling a queue nobody reads. Releasing the semaphore and draining the queue in aclose() is what lets a thread blocked on acquire() notice the stop flag — a flag alone is not enough when the thread is waiting.

Verify: the producer's count is close to the number consumed plus the queue bound, and aclose() returns without a timeout.

5. Choose the right bridge for the shape of the work

Not everything needs a streaming bridge. Match the mechanism to the shape: one result, a stream of results, or many independent calls.

import asyncio
import concurrent.futures


def fetch_one(key: str) -> str:                           # a single blocking call
    return f"value:{key}"


async def main() -> None:
    # 1. one blocking call, one result
    print(await asyncio.to_thread(fetch_one, "a"))

    # 2. many independent blocking calls: a bounded pool, results as they arrive
    with concurrent.futures.ThreadPoolExecutor(max_workers=4) as pool:
        futures = [asyncio.wrap_future(pool.submit(fetch_one, k)) for k in "bcde"]
        for done in asyncio.as_completed(futures):
            print(await done)

    # 3. a stream of results from one long-lived thread: the bridge from this page
    stream = ManagedStream(maxsize=8)
    stream.start(counting_stream, 5)
    async for row in stream:
        print(row)
    await stream.aclose()


asyncio.run(main())

to_thread for one call, a pool with wrap_future for many, and the queue bridge only when a single thread produces a series of results. Reaching for the bridge when a pool would do adds a thread you must supervise; reaching for a pool when the SDK insists on one long-lived session does not work at all.

Verify: all three forms print their results, and only the third creates a thread your code has to stop.

Which thread bridge fits? A grid of 4 rows by 2 columns. Which thread bridge fits? shape of the work mechanism you must manage one blocking call asyncio.to_thread nothing many independent calls pool + wrap_future pool size stream from one session queue bridge thread, bound, shutdown callbacks with replies future per request correlation table Only the third and fourth rows need code you have to stop yourself.

Verification

The thread-to-asyncio bridge is correct when:

  • Only call_soon_threadsafe touches the loop from the producing thread.
  • Backpressure reaches the producer: peak queue depth equals the configured bound under a slow consumer, not the total item count.
  • Endings are explicit: completion and failure travel through the queue in order, and exceptions surface in the consumer.
  • Leaving stops the producer: breaking out of the consumer sets a stop flag, unblocks the thread and joins it within a grace period.
  • The simplest mechanism is used: to_thread or a pool where they suffice; the bridge only for genuine streams.

Pitfalls & edge cases

  • Calling queue.put_nowait from the thread. It mutates loop-owned state and its wake-ups are not thread-safe; the consumer can miss items or wake late.
  • A bound on the queue but not on the thread. QueueFull in a loop callback raises inside the loop, not in the producer; the semaphore is what actually slows the thread.
  • Deadlock on close. A thread blocked in acquire() never sees the stop flag; release the semaphore and drain the queue when closing.
  • Daemon threads at exit. A daemon thread is killed abruptly at interpreter shutdown, skipping its finally; close the stream explicitly during graceful shutdown.
  • Losing context. The producing thread does not share the request's context variables; capture request IDs when starting the thread, as in carrying contextvars across threads and executors.

Frequently Asked Questions

How do I put items on an asyncio.Queue from a thread?

Capture the loop on the loop thread and call loop.call_soon_threadsafe(queue.put_nowait, item) from the thread. Never call queue.put_nowait or queue.put directly from another thread: asyncio queues are not thread-safe, and their waiter wake-ups do not reach a loop that is asleep in its selector.

How do I apply backpressure to a producing thread?

Pair the asyncio queue with a threading.Semaphore initialised to the same capacity. The producing thread acquires a slot before emitting an item, which blocks it when the queue is full, and the async consumer releases a slot after taking each item. The queue's own bound cannot block a thread.

How should a producer thread signal that the stream is finished or failed?

Send sentinels through the same queue so ordering is preserved: a None, or a small marker object, for completion, and a marker carrying the exception for failure, which the consumer re-raises. Wrap the thread body in a try so a crashing producer always signals rather than leaving the consumer waiting forever.

What happens if the async consumer stops reading?

The producing thread fills the queue and then blocks on the semaphore indefinitely, so the thread leaks. Expose a close method that sets a stop flag, releases the semaphore and drains the queue so the thread can observe the flag, then joins the thread with a timeout.