Communication between async tasks and synchronous threads in python

Question

I am looking for the best solution for communication between async tasks and methods/functions that run in a thread pool executor from concurrent.futures. In previous synchronous projects, I would use the queue.Queue class. I assume that any method should be thread safe and therefore the asyncio.queue will not work.

I have seen people extend the queue.Queue class to do something like:

class async_queue(Queue):
  async def aput(self, item):
    self.put_nowait(item)

  async def aget(self):
    resp = await asyncio.get_event_loop().run_in_executor( None, self.get )
    return resp

Is there a better way?

This question is similar to: How to communicate between traditional thread and asyncio thread in Python?. If you believe it’s different, please edit the question, make it clear how it’s different and/or how the answers on that question are not helpful for your problem. — Ilya Egorov
– Ilya Egorov, Commented Nov 21, 2024 at 20:56

Alan W. Smith · Accepted Answer · 2023-07-31 18:54:14Z

13

I would recommend going the other way around: using the asyncio.Queue class to communicate between the two worlds. This has the advantage of not having to spend a slot in the thread pool on operations that take a long time to complete, such as a get().

Here is an example:

class Queue:
    def __init__(self):
        self._loop = asyncio.get_running_loop()
        self._queue = asyncio.Queue()

    def sync_put_nowait(self, item):
        self._loop.call_soon(self._queue.put_nowait, item)

    def sync_put(self, item):
        asyncio.run_coroutine_threadsafe(self._queue.put(item), self._loop).result()

    def sync_get(self):
        return asyncio.run_coroutine_threadsafe(self._queue.get(item), self._loop).result()

    def async_put_nowait(self, item):
        self._queue.put_nowait(item)

    async def async_put(self, item):
        await self._queue.put(item)

    async def async_get(self):
        return await self._queue.get()

The methods prefixed with sync_ are meant to be invoked by sync code (running outside the event loop thread). The ones prefixed with async_ are to be called by code running in the event loop thread, regardless of whether they are actually coroutines. (put_nowait, for example, is not a coroutine, but it still must be distinguished between a sync and an async version.)

edited Jul 31, 2023 at 18:54

Alan W. Smith

25.6k5 gold badges73 silver badges101 bronze badges

answered Jan 8, 2020 at 17:07

user4815162342

159k22 gold badges347 silver badges417 bronze badges

Sign up to request clarification or add additional context in comments.

2 Comments

bmitc Over a year ago

How does this work? The asyncio queues are not thread safe. They are meant to be used to communicate between coroutines. Since the coroutines all run in an event loop which runs on a single thread, this is why the lack of thread safety doesn't matter between coroutines. But it would matter with non-event loop code.

user4815162342 Over a year ago

@bmitc The "sync_" prefixed methods use run_coroutine_threadsafe to run the function inside the event loop. That way, when the self._queue.put (and get etc) starts actually executing, that will be in the event loop thread.

Ilya Egorov · Accepted Answer · 2025-01-23 08:19:13Z

3

The disadvantage of the solution via run_coroutine_threadsafe() given in the other answer is that a worker thread will actually depend on your event loop: the thread will wait for the event loop to process your put() and get(). If your event loop is doing a lot of work, this can lead to very long waits, up to seconds of real time.

I suggest using aiologic.Queue (I'm the creator of aiologic), which doesn't have this disadvantage: wherever you do put() or get(), the thread or task will only wait if it's really necessary (e.g. if you do get() when the queue is empty).

queue = aiologic.Queue(["first", "second"], maxsize=3)

Timer(1, queue.green_put, ["third"]).start()

print(await queue.async_get())  # "first"
print(await queue.async_get())  # "second"
print(await queue.async_get())  # after one second: "third"

And if you don't need to set the upperbound limit on the number of items that can be placed in the queue, try aiologic.SimpleQueue. With it you will get maximum performance when the queue is rarely empty and there are many consumers (and of course, nothing prevents you from using this queue in other cases).

edited Jan 23 at 8:19

answered Oct 19, 2024 at 21:45

Ilya Egorov

5211 gold badge3 silver badges12 bronze badges

1 Comment

user4815162342 Over a year ago

aoilogic looks really neat and well-thought-out - thanks for creating it!

Collectives™ on Stack Overflow

Communication between async tasks and synchronous threads in python

2 Answers 2

2 Comments

1 Comment

Your Answer

Linked

Hot Network Questions

Collectives™ on Stack Overflow

2 Answers 2

2 Comments

1 Comment

Your Answer

Sign up or log in

Post as a guest

Linked

Related