Skip to content

Bridging Queues Between Threads and asyncio Tasks

Python ships two queues and neither spans the boundary between threads and the event loop. queue.Queue blocks the calling thread, which freezes the loop if a coroutine calls it. asyncio.Queue is not thread-safe, so a thread cannot use it at all. A service that mixes both worlds — a blocking SDK feeding async handlers, or async code dispatching work to a thread-based legacy component — needs a queue with two faces: blocking put/get for threads and awaitable put/get for coroutines, over one shared buffer with one shared bound. The janus library provides exactly this, and it is the right dependency for most teams. Understanding how it works is still worth an hour, because the same primitives appear whenever the two worlds meet, and because a 70-line version is sometimes preferable to a dependency. This guide builds one, tests it in both directions, and covers closing it without stranding either side.

Prerequisites

Which queue can each side use? A grid of 3 rows by 2 columns. Which queue can each side use? queue from a thread from a coroutine queue.Queue blocks the thread: fine blocks the loop: wrong asyncio.Queue not thread-safe awaits: fine mixed queue / janus blocking API awaitable API One buffer, one bound, two interfaces: that is the whole idea.

1. Understand why neither stdlib queue fits

The failure modes are worth seeing once: a blocking queue stalls the whole loop, and an asyncio queue used from a thread misses wake-ups.

import asyncio
import queue
import threading
import time


async def main() -> None:
    loop = asyncio.get_running_loop()
    blocking: queue.Queue[str] = queue.Queue()
    lag: list[float] = []

    async def heartbeat() -> None:
        while True:
            started = loop.time()
            await asyncio.sleep(0.01)
            lag.append(loop.time() - started - 0.01)

    beat = asyncio.create_task(heartbeat())
    threading.Timer(0.3, blocking.put, args=("late item",)).start()
    await asyncio.sleep(0.05)
    blocking.get()                                        # blocks the loop thread for 250 ms
    await asyncio.sleep(0.05)                             # let the delayed heartbeat record its lag
    beat.cancel()
    print(f"worst loop lag while waiting on queue.Queue.get(): {max(lag) * 1000:.0f} ms")


asyncio.run(main())

The heartbeat was late by roughly the wait — the loop was frozen, unable to run any other task. The mirror-image failure is using asyncio.Queue from a thread: put_nowait from another thread schedules the waiter's wake-up with a non-thread-safe call, so a consumer can wake very late or not at all, as measured in running multiple event loops in separate threads.

Verify: the reported lag is close to the time the loop spent inside blocking.get().

2. Build one buffer with two interfaces

The design: a deque protected by a threading.Lock, threading.Conditions for thread waiters, and asyncio.Events for loop waiters. Whichever side changes the buffer notifies both kinds of waiter, using call_soon_threadsafe when the change happened off the loop.

import asyncio
import collections
import threading


class Closed(RuntimeError):
    pass


class MixedQueue:
    """A bounded queue with a blocking side for threads and an async side for tasks."""

    def __init__(self, maxsize: int = 0) -> None:
        self._loop = asyncio.get_running_loop()           # constructed on the loop thread
        self._items: collections.deque = collections.deque()
        self._maxsize = maxsize
        self._mutex = threading.Lock()
        self._not_empty = threading.Condition(self._mutex)
        self._not_full = threading.Condition(self._mutex)
        self._async_not_empty = asyncio.Event()
        self._async_not_full = asyncio.Event()
        self._async_not_full.set()
        self._closed = False

    def _sync_flags(self) -> None:
        """Refresh the loop-side events. Must run on the loop thread, mutex held."""
        (self._async_not_empty.set if self._items else self._async_not_empty.clear)()
        full = self._maxsize and len(self._items) >= self._maxsize
        (self._async_not_full.clear if full else self._async_not_full.set)()

    def _wake_loop(self) -> None:
        """Schedule a flag refresh from a thread."""
        empty, full = not self._items, self._maxsize and len(self._items) >= self._maxsize

        def apply() -> None:
            (self._async_not_empty.clear if empty else self._async_not_empty.set)()
            (self._async_not_full.clear if full else self._async_not_full.set)()

        self._loop.call_soon_threadsafe(apply)

Two rules keep this correct. The mutex is held for a few statements only — never across an await or any I/O — so holding it briefly on the loop thread is acceptable. And loop-side flags are only ever set on the loop thread, either directly or through call_soon_threadsafe, because asyncio.Event is not thread-safe.

Verify: the class constructs on the loop thread, and _sync_flags is never called from a thread.

3. Implement both sides over the shared buffer

The thread side is a classic condition-variable queue. The async side waits on an event, then re-checks the state under the mutex — the same predicate loop, since an event is level-triggered and another waiter may have taken the item first.

import asyncio
import threading


class MixedQueue(MixedQueue):                              # continues the class from step 2
    # --- thread side -------------------------------------------------------
    def sync_put(self, item, timeout: float | None = None) -> None:
        with self._not_full:
            if self._maxsize:
                ready = self._not_full.wait_for(
                    lambda: self._closed or len(self._items) < self._maxsize, timeout)
                if not ready:
                    raise TimeoutError
            if self._closed:
                raise Closed
            self._items.append(item)
            self._not_empty.notify()
            self._wake_loop()

    def sync_get(self, timeout: float | None = None):
        with self._not_empty:
            if not self._not_empty.wait_for(lambda: self._items or self._closed, timeout):
                raise TimeoutError
            if not self._items:
                raise Closed
            item = self._items.popleft()
            self._not_full.notify()
            self._wake_loop()
            return item

    # --- loop side ---------------------------------------------------------
    async def async_put(self, item) -> None:
        while True:
            await self._async_not_full.wait()
            with self._mutex:
                if self._closed:
                    raise Closed
                if not self._maxsize or len(self._items) < self._maxsize:
                    self._items.append(item)
                    self._not_empty.notify()
                    self._sync_flags()
                    return

    async def async_get(self):
        while True:
            await self._async_not_empty.wait()
            with self._mutex:
                if self._items:
                    item = self._items.popleft()
                    self._not_full.notify()
                    self._sync_flags()
                    return item
                if self._closed:
                    raise Closed
                self._async_not_empty.clear()              # lost the race: wait again

Notifying the thread-side condition from the loop side is safe because the loop holds the same mutex while doing it. The reverse — the thread touching the asyncio.Events — goes through _wake_loop(). That asymmetry is the whole trick.

Verify: a thread blocked in sync_get wakes when a coroutine calls async_put, and a coroutine awaiting async_get wakes when a thread calls sync_put.

How a put on one side wakes the other 4 stacked layers from Shared buffer to Thread -> loop. How a put on one side wakes the other Shared buffer deque + maxsize threading.Lock Thread waiters Condition.wait_for notified under the mutex Loop waiters asyncio.Event set only on the loop thread Thread -> loop call_soon_threadsafe refresh the events Threads notify conditions directly; the loop's events are always touched by the loop.

4. Exercise both directions under load

A bridge that works in one direction often deadlocks in the other. Test both, with the bound small enough that each side actually blocks.

import asyncio
import threading
import time


async def main() -> None:
    bridge = MixedQueue(maxsize=16)

    def producer(n: int) -> None:
        for i in range(n):
            bridge.sync_put(("from-thread", i))

    thread = threading.Thread(target=producer, args=(500,), daemon=True)
    started = time.perf_counter()
    thread.start()
    received = [await bridge.async_get() for _ in range(500)]
    thread.join()
    print(f"thread -> loop: {len(received)} items, first {received[0]}, "
          f"last {received[-1]}, {time.perf_counter() - started:.2f}s")

    drained: list = []

    def consumer(n: int) -> None:
        for _ in range(n):
            drained.append(bridge.sync_get())

    thread = threading.Thread(target=consumer, args=(300,), daemon=True)
    thread.start()
    for i in range(300):
        await bridge.async_put(("from-loop", i))
    thread.join()
    print(f"loop -> thread: {len(drained)} items, last {drained[-1]}")


asyncio.run(main())

Five hundred items crossed from the thread to the loop and three hundred back, in order, with a bound of sixteen holding in both directions. Order is preserved because there is one buffer: unlike two separate queues, there is no window in which an item exists on neither side.

Verify: both directions transfer every item in order, and the run does not hang with a small maxsize.

5. Close without stranding either side

A half-closed bridge is a hang: a thread blocked in sync_get or a coroutine awaiting async_get waits forever once the other side has gone. Closing must wake both kinds of waiter and make their state explicit.

import asyncio
import threading


class MixedQueue(MixedQueue):                              # continues the class
    def close(self) -> None:
        """Wake every waiter on both sides; further puts raise, drained gets raise."""
        with self._mutex:
            self._closed = True
            self._not_empty.notify_all()
            self._not_full.notify_all()

        def release_loop_waiters() -> None:
            self._async_not_empty.set()
            self._async_not_full.set()

        if threading.current_thread() is threading.main_thread() and self._loop.is_running():
            release_loop_waiters()
        else:
            self._loop.call_soon_threadsafe(release_loop_waiters)


async def main() -> None:
    bridge = MixedQueue(maxsize=4)
    bridge.close()
    try:
        bridge.sync_get(timeout=1)
    except Closed:
        print("sync_get on a closed, empty queue raises Closed")
    try:
        await bridge.async_get()
    except Closed:
        print("async_get on a closed, empty queue raises Closed")
    try:
        bridge.sync_put("x")
    except Closed:
        print("sync_put after close raises Closed")


asyncio.run(main())

Both sides fail fast rather than hanging, and items already in the buffer are still delivered before Closed is raised, so a consumer drains what is there and then learns the stream ended. Wire close() into the graceful shutdown sequence, and join the threads afterwards so their finally blocks run.

Verify: all three calls raise Closed rather than blocking, and a queue closed with items still buffered delivers them first.

Do you need a two-sided queue? A decision on How do the two worlds exchange data with 3 outcomes. Do you need a two-sided queue? How do the two worlds exchange data? thread produces, loop consumes one-way bridge queue + semaphore both directions mixed queue janus or this class call and result executor to_thread / wrap_future Most hybrid code needs the first or third row, not the second.

Verification

The bridge is sound when:

  • Neither side blocks the other's world: coroutines never call the blocking API, and threads never touch asyncio objects directly.
  • Both directions work under a small bound: items transfer in order, and each side blocks rather than buffering without limit.
  • Wake-ups are reliable: a waiter on either side is woken by a put or get on the other side, with no polling and no missed items.
  • Closing wakes everyone: buffered items are delivered, then both sides raise a clear error instead of hanging.
  • The mutex is never held across an await or any I/O call.

Pitfalls & edge cases

  • Constructing the queue off the loop thread. It captures the loop at construction; build it inside the running loop, or pass the loop explicitly.
  • Polling instead of waiting. A loop side implemented as while queue.empty(): await asyncio.sleep(0.01) adds latency and burns CPU; use events and condition variables.
  • Unbounded bridges. With maxsize=0 neither side ever waits, so a fast producer fills memory. Always set a bound in production.
  • Using it across event loops. The async side belongs to one loop; other loops need their own bridge, as in running multiple event loops in separate threads.
  • Reimplementing when a library will do. janus is maintained, tested and handles the edge cases; prefer it unless a dependency is genuinely unwanted.

Frequently Asked Questions

Can I use queue.Queue with asyncio?

Only from threads. Calling queue.Queue.get or put from a coroutine blocks the event loop thread until the operation completes, freezing every other task. Either call it through asyncio.to_thread, or use a queue that offers a blocking interface for threads and an awaitable interface for coroutines.

Is asyncio.Queue thread-safe?

No. Its methods must be called from the loop's thread. A thread may hand items to it only through loop.call_soon_threadsafe(queue.put_nowait, item), which also wakes the loop. Calling put_nowait directly from a thread can leave a waiting consumer unwoken until the loop happens to wake for another reason.

What is janus and when should I use it?

janus is a library providing a single queue with two interfaces: a synchronous one for threads and an asynchronous one for coroutines, sharing one buffer and one maximum size. Use it whenever both worlds must exchange items in a process; writing your own is reasonable only if you want to avoid the dependency.

How do I close a queue shared between threads and coroutines?

Set a closed flag under the shared mutex, notify every thread-side condition, and set the loop-side events through call_soon_threadsafe so awaiting coroutines wake. Deliver items already buffered, then raise a specific error from both sides so neither a thread nor a task waits forever.