django.channels async consumer does not appear to execute asynchronously

Question

I have added django.channels to a django project in order to support long running processes that notify users of progress via websockets.

Everything appears to work fine except for the fact that the implementation of the long running process doesn't seem to respond asynchronously.

For testing I have created an AsyncConsumer that recognizes two types of messages 'run' and 'isBusy'.

The 'run' message handler sets a 'busy flag' sends back a 'process is running' message, waits asynchronously for 20 seconds resets the 'busy flag' and then sends back a 'process complete message'

The 'isBusy' message returns a message with the status of the busy flag.

My expectation is that if I send a run message I will receive immediately a 'process is running' message back and after 20 seconds I will receive a 'process complete' message. This works as expected.

I also expect that if I send a 'isBusy' message I will receive immediately a response with the state of the flag.

The observed behaviour is as follows:

a message 'run' is sent (from the client)
a message 'running please wait' is immediately received
a message 'isBusy' is sent (from the client)
the message reaches the web socket listener on the server side
nothing happens until the run handler finishes
a 'finished running' message is received on the client
followed immediately by a 'process isBusy:False' message

Here is the implementation of the Channel listener:

class BackgroundConsoleConsumer(AsyncConsumer):
    def __init__(self, scope):
        super().__init__(scope)
        self.busy = False

    async def run(self, message):
        print("run got message", message)
        self.busy = True
        await self.channel_layer.group_send('consoleChannel',{
                    "type":"consoleResponse",
                    "text":"running please wait"
                })
        await asyncio.sleep(20)
        self.busy = False
        await self.channel_layer.group_send('consoleChannel',{
                    "type":"consoleResponse",
                    "text": "finished running"
                })

    async def isBusy(self,message):
        print('isBusy got message', message)
        await self.channel_layer.group_send('consoleChannel',{
                    "type":"consoleResponse",
                    "text":  "process isBusy:{0}".format(self.busy)
                })

The channel is set up in the routing file as follows:

application = ProtocolTypeRouter({
    "websocket": AuthMiddlewareStack(
        URLRouter([
            url("^console/$", ConsoleConsumer),
        ])

    ),
    "channel": ChannelNameRouter({
        "background-console":BackgroundConsoleConsumer,
    }),
})

I run the channel with one worker (via ./manage.py runworker ).

The experiment was done with the django test server (via runserver).

Any ideas as to why the channel consumer does not appear to work asynchronously would be appreciated.

Can you show us the definition of run? The documentation doesn't seem to mention it, so I assume it is defined in your code and not inherited. — user4815162342
– user4815162342, Commented Feb 20, 2018 at 11:48
Hello, I am mentioning two functions run and isBusy, they are the functions shown in the code above BackgroundConsoleConsumer. I also mention running the channel with one worker... that is I start from a console a process for the channel with: ./manage.py runworker background-console where background-console is the name associated with the channel ( the second script in the description above) — raduw
– raduw, Commented Feb 20, 2018 at 11:51
The messages (run and isBusy) come to the BackgroundConsoleConsumer via a AsyncJsonWebsocketConsumer which listens for strings from connected clients and then sends messages to the background-console channel. So upon receiving a socket message I just do the following: await self.channel_layer.send('background-console', { 'type': 'run', 'data': { 'some-data': 1} }) — raduw
– raduw, Commented Feb 20, 2018 at 11:58
Sorry, I meant the call sites of run. The problem could be that run is being awaited instead of being started in the background, which causes isBusy to wait until it finishes. Maybe at some point you should be using loop.create_task instead of await. This is just a guess, since I'm not familiar with the channels architecture. — user4815162342
– user4815162342, Commented Feb 20, 2018 at 12:21
Nice work! Please write that up as an answer, it might be quite useful to others. Note that you might want to use loop.create_task (or the new asyncio.create_task) in preference to asyncio.ensure_future (as explained by Guido). — user4815162342
– user4815162342, Commented Feb 20, 2018 at 17:03

raduw · Accepted Answer · 2018-02-20 17:29:55Z

5

After a bit of digging around here is the problem and one solution to it.

A channel adds messages sent to it to a asyncio.Queue and processes them sequentially.

It is not enough to release the coroutine control (via a asyncio.sleep() or something similar), one must finish processing the message handler before a new message is received by the consumer.

Here is the fix to the previous example that behaves as expected (i.e. responds to the isBusy messages while processing the run long running task)

Thank you @user4815162342 for your suggestions.

class BackgroundConsoleConsumer(AsyncConsumer):
    def __init__(self, scope):
        super().__init__(scope)
        self.busy = False

    async def run(self, message):
        loop = asyncio.get_event_loop()
        loop.create_task(self.longRunning())

    async def longRunning(self):
        self.busy = True
        await self.channel_layer.group_send('consoleChannel',{
                    "type":"the.type",
                    "text": json.dumps({'message': "running please wait", 'author': 'background console process'})
                })
        print('before sleeping')
        await asyncio.sleep(20)
        print('after sleeping')
        self.busy = False
        await self.channel_layer.group_send('consoleChannel',{
                    "type":"the.type",
                    "text": json.dumps({'message': "finished running", 'author': 'background console process'})
                })

    async def isBusy(self,message):
        print('isBusy got message', message)
        await self.channel_layer.group_send('consoleChannel',{
                    "type":"the.type",
                    "text":  json.dumps({'message': "process isBusy:{0}".format(self.busy),
                                         'author': 'background console process'})
                })

answered Feb 20, 2018 at 17:29

raduw

1,1181 gold badge9 silver badges21 bronze badges

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

2 Comments

sebhaase Over a year ago

However ensure_future() - or starting with Python 3.7 asyncio.create_task []

sebhaase Over a year ago

However ensure_future() should be preferred over loop.create_task - or starting with Python 3.7 asyncio.create_task [docs.python.org/3/library/… - - ref. stackoverflow.com/questions/36342899/…

Collectives™ on Stack Overflow

django.channels async consumer does not appear to execute asynchronously

1 Answer 1

2 Comments

Your Answer

Linked

Hot Network Questions

Collectives™ on Stack Overflow

1 Answer 1

2 Comments

Your Answer

Sign up or log in

Post as a guest

Linked

Related