1

Two coroutintes in code below, running in different threads, cannot communicate with each other by asyncio.Queue. After the producer inserts a new item in asyncio.Queue, the consumer cannot get this item from that asyncio.Queue, it gets blocked in method await self.n_queue.get().

I try to print the ids of asyncio.Queue in both consumer and producer, and I find that they are same.

import asyncio
import threading
import time


class Consumer:

    def __init__(self):
        self.n_queue = None
        self._event = None

    def run(self, loop):
        loop.run_until_complete(asyncio.run(self.main()))

    async def consume(self):
        while True:
            print("id of n_queue in consumer:", id(self.n_queue))
            data = await self.n_queue.get()
            print("get data ", data)
            self.n_queue.task_done()

    async def main(self):
        loop = asyncio.get_running_loop()
        self.n_queue = asyncio.Queue(loop=loop)
        task = asyncio.create_task(self.consume())
        await asyncio.gather(task)

    async def produce(self):
        print("id of queue in producer ", id(self.n_queue))
        await self.n_queue.put("This is a notification from server")


class Producer:
    def __init__(self, consumer, loop):
        self._consumer = consumer
        self._loop = loop

    def start(self):
        while True:
            time.sleep(2)
            self._loop.run_until_complete(self._consumer.produce())


if __name__ == '__main__':
    loop = asyncio.get_event_loop()
    print(id(loop))
    consumer = Consumer()
    threading.Thread(target=consumer.run, args=(loop,)).start()
    producer = Producer(consumer, loop)
    producer.start()


id of n_queue in consumer: 2255377743176
id of queue in producer  2255377743176
id of queue in producer  2255377743176
id of queue in producer  2255377743176

I try to debug step by step in asyncio.Queue, and I find after the method self._getters.append(getter) is invoked in asyncio.Queue, the item is inserted in queue self._getters. The following snippets are all from asyncio.Queue.


    async def get(self):
        """Remove and return an item from the queue.

        If queue is empty, wait until an item is available.
        """
        while self.empty():
            getter = self._loop.create_future()
            self._getters.append(getter)
            try:
                await getter
            except:
                # ...
                raise
        return self.get_nowait()

When a new item is inserted into asycio.Queue in producer, the methods below would be invoked. The variable self._getters has no items although it has same id in methods put() and set().


    def put_nowait(self, item):
        """Put an item into the queue without blocking.

        If no free slot is immediately available, raise QueueFull.
        """
        if self.full():
            raise QueueFull
        self._put(item)
        self._unfinished_tasks += 1
        self._finished.clear()
        self._wakeup_next(self._getters)

    def _wakeup_next(self, waiters):
        # Wake up the next waiter (if any) that isn't cancelled.
        while waiters:
            waiter = waiters.popleft()
            if not waiter.done():
                waiter.set_result(None)
                break

Does anyone know what's wrong with the demo code above? If the two coroutines are running in different threads, how could they communicate with each other by asyncio.Queue?

wu jushan
  • 71
  • 2
  • 7

1 Answers1

1

Short answer: no!

Because the asyncio.Queue needs to share the same event loop, but

An event loop runs in a thread (typically the main thread) and executes all callbacks and Tasks in its thread. While a Task is running in the event loop, no other Tasks can run in the same thread. When a Task executes an await expression, the running Task gets suspended, and the event loop executes the next Task.

see https://docs.python.org/3/library/asyncio-dev.html#asyncio-multithreading

Even though you can pass the event loop to threads, it might be dangerous to mix the different concurrency concepts. Still note, that passing the loop just means that you can add tasks to the loop from different threads, but they will still be executed in the main thread. However, adding tasks from threads can lead to race conditions in the event loop, because

Almost all asyncio objects are not thread safe, which is typically not a problem unless there is code that works with them from outside of a Task or a callback. If there’s a need for such code to call a low-level asyncio API, the loop.call_soon_threadsafe() method should be used

see https://docs.python.org/3/library/asyncio-dev.html#asyncio-multithreading

Typically, you should not need to run async functions in different threads, because they should be IO bound and therefore a single thread should be sufficient to handle the work load. If you still have some CPU bound tasks, you are able to dispatch them to different threads and make the result awaitable using asyncio.to_thread, see https://docs.python.org/3/library/asyncio-task.html#running-in-threads.

There are many questions already about this topic, see e.g. Send asyncio tasks to loop running in other thread or How to combine python asyncio with threads?

If you want to learn more about the concurrency concepts, I recommend to read https://medium.com/analytics-vidhya/asyncio-threading-and-multiprocessing-in-python-4f5ff6ca75e8

Carlos Horn
  • 1,115
  • 4
  • 17