Home / Questions / 476150
Accepted
Trauma
Asked: 2022-08-11 11:02:24 +0800 CST 2022-08-11 11:02:24 +0800 CST

Coroutines and asyncio.Queue

  • 772

I am experimenting with asyncio(in python3.8). First of all, a small test of how to use asyncio.Queue; simply put a data in the queue and get it again.

import asyncio

class QServer:
    def _getQueue( self, queueName ):
        queue = self._queues.get( queueName )

        if queue is None:
            queue = asyncio.Queue( )
            self._queues[queueName] = asyncio.Queue( ) 

        return queue

    def __init__( self ):
        self._queues = { }

    async def put( self, queueName, data ):
        queue = self._getQueue( queueName )
        await queue.put( data )

    async def get( self, queueName ):
        queue = self._getQueue( queueName )
        return await queue.get( )

async def main( ):
    QueueName = 'TestQueue'
    TestData = 'TestData'

    queue = QServer( )

    print( 'put' )
    await queue.put( QueueName, TestData )
    internal = queue._getQueue( QueueName )
    print( internal.qsize( ) )
    print( 'type' )
    assert( type( queue._queues[ QueueName ] ) == type( asyncio.Queue( ) ) )
    print( 'get' )
    assert( await queue.get( QueueName ) == TestData )
    print( 'End' )

asyncio.run( main( ) )

The output I expected is:

put
1
type
get
End

However, I am getting this other one:

put
0
type
get

And there the program stays on hold indefinitely, until I press Ctrl+Cand finish it.

To my uninitiated eyes, it looks like it put( ) 's not putting anything in the queue. And I have a suspicion that I am not correctly understanding how coroutines work .

  • What am I doing wrong ?
  • How do I solve it ?
python

1 Answers

  1. Best Answer

    You're going to pull your hair out when you read this, but... Your understanding of asyncio, coroutines and queues is correct, you just have a bug here, inside _getQueue():

            if queue is None:
            queue = asyncio.Queue( )
            self._queues[queueName] = asyncio.Queue( ) 
        return queue

    As you can see, you create a new queue, which you assign to the variable queue, and which is the one that you finally return (and on which the ) is going to be done put(), but nevertheless what you save in the dictionary self._queuesis not that queue that you have created, but a different new one, because you call again to asyncio.Queue(). Because of that, the queue you later do get()the , which is the one that is stored in that dictionary, is different from the one you used for the put().

    In short, that piece should be like this:

            if queue is None:
            queue = asyncio.Queue( )
            self._queues[queueName] = queue

        return queue

    With that change, now yes, the output is what you expect:

    put
1
type
get
End

    Bonuses

    If you want, and if you're using Python 3.8+, you can also use the assignment operator ( :=) which allows you to save a line, although I suspect at the cost of readability:

            if queue is None:
            self._queues[queueName] = (queue := asyncio.Queue())

        return queue

    Although certainly the most pythonic option would be to use defaultdict, which is a type of dictionary that automatically creates new elements (of the type you choose when creating the dictionary) when you ask for a key that doesn't exist.

    With this approach you save the ._getQueue()entire function and it would look like this:

    import asyncio
from collections import defaultdict

class QServer:
    def __init__( self ):
        # A defaultdict se le pasa qué constructor usar para crear elementos nuevos
        self._queues = defaultdict(asyncio.Queue)

    async def put( self, queueName, data ):
        queue = self._queues[queueName]
        await queue.put( data )

    async def get( self, queueName ):
        queue = self._queues[queueName]
        return await queue.get( )

async def main( ):
    QueueName = 'TestQueue'
    TestData = 'TestData'

    queue = QServer( )

    print( 'put' )
    await queue.put( QueueName, TestData )
    internal = queue._queues[ QueueName ]
    print( internal.qsize( ) )
    print( 'type' )
    assert( type( queue._queues[ QueueName ] ) == type( asyncio.Queue( ) ) )
    print( 'get' )
    assert( await queue.get( QueueName ) == TestData )
    print( 'End' )

asyncio.run( main( ) )
    • 10