asyncio.Queue Stuck With 1 Coroutine Adding to Queue, 1 Coroutine Getting from Queue

Question

In my simple asyncio code below, the app has one task self.add_item_loop_task continuously adding an integer to the asyncio.Queue named self.queue, while a second task self.get_item_loop_task continuously waits for something to be added to the queue and print it out.

However, this app only prints out 0 once when I run it, and gets stuck there. I believe the loop in self.get_item_loop_task is not proceeding. Why is this happening?

import asyncio

class App:
    def __init__(self):
        self.queue = asyncio.Queue()

    async def start(self):
        self.add_item_loop_task = asyncio.create_task(self.add_item_loop())
        self.get_item_loop_task = asyncio.create_task(self.get_item_loop())
        await asyncio.wait(
            [
                self.add_item_loop_task,
                self.get_item_loop_task,
            ]
        )

    async def stop(self):
        self.add_item_loop_task.cancel()
        self.get_item_loop_task.cancel()

    async def add_item_loop(self):
        i = 0
        while True:
            await self.queue.put(i)
            i += 1
            await asyncio.sleep(1)

    async def get_item_loop(self):
        while True:
            item = await self.queue.get()
            print(item)


app = App()
try:
    asyncio.run(app.start())
except KeyboardInterrupt:
    asyncio.run(app.stop())

Matt Fowler · Accepted Answer · 2021-06-13 13:07:52Z

8

This is caused by some dubious implementation details of asyncio. When you say self.queue = asyncio.Queue() this will actually create an event loop if one does not already exist. Meanwhile, when you call asyncio.run() it will always create a new event loop. This means if you create a queue before you call asyncio.run() you can get some strange behavior because there are two event loops, the one that your queue uses and the one that asyncio.run is using.

You can fix this issue by moving the creation of App into a coroutine function that you pass in to asyncio.run() like below. Doing this your application works as intended.

async def main():
    app = App()
    await app.start()

asyncio.run(main())

answered Jun 13, 2021 at 13:07

Matt Fowler

2,7932 gold badges18 silver badges20 bronze badges

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

Comments

Arjuna Deva · Accepted Answer · 2022-06-30 08:55:53Z

2

As @Matt Fowler explained the code is creating two simultaneous event loops. One created when initializing the class and the second one when running asyncio.run(main()).

You can simply move self.queue = asyncio.Queue() from the def __init__(self) method to the beginning async def start(self) method. The newly created asyncio.Queue() object will be available to both tasks.

The following code works:

import asyncio

class App:
# def __init__(self):
#     self.queue = asyncio.Queue()

async def start(self):
    # self.queue in the following line will be available to both tasks
    self.queue = asyncio.Queue()
    self.add_item_loop_task = asyncio.create_task(self.add_item_loop())
    self.get_item_loop_task = asyncio.create_task(self.get_item_loop())
    await asyncio.wait(
        [
            self.add_item_loop_task,
            self.get_item_loop_task,
        ]
    )

async def stop(self):
    self.add_item_loop_task.cancel()
    self.get_item_loop_task.cancel()

async def add_item_loop(self):
    i = 0
    while True:
        await self.queue.put(i)
        i += 1
        await asyncio.sleep(1)

async def get_item_loop(self):
    while True:
        item = await self.queue.get()
        print(item)

app = App()
try:
  asyncio.run(app.start())
except KeyboardInterrupt:
  asyncio.run(app.stop())

edited Jun 30, 2022 at 8:55

answered Jun 27, 2022 at 17:46

Arjuna Deva

2954 silver badges19 bronze badges

1 Comment

Nyxynyx Over a year ago

Sounds like a good alternative! Thanks!

Collectives™ on Stack Overflow

asyncio.Queue Stuck With 1 Coroutine Adding to Queue, 1 Coroutine Getting from Queue

2 Answers 2

Comments

1 Comment

Your Answer

Hot Network Questions

Collectives™ on Stack Overflow

2 Answers 2

Comments

1 Comment

Your Answer

Sign up or log in

Post as a guest

Related