Django Channels unable to receive events during asyncio.sleep?

user_6259805 0
Viewed 588

I am using AsyncWebSocketConsumer and I want to put consumer to sleep for 10 seconds while handling one specific event. So, on receiving that event I am calling await asyncio.sleep(10). But problem with this is that it is blocking events to this consumer instance for next 10 seconds.

My question is that if I am using AsyncConsumer, shouldn't it continue sending future events even though I am on async sleep?

I have been going through the code of daphne server and found out following:

    def onMessage(self, payload, isBinary):
        # If we're muted, do nothing.
        if self.muted:
            logger.debug("Muting incoming frame on %s", self.client_addr)
            return
        logger.debug("WebSocket incoming frame on %s", self.client_addr)
        self.last_ping = time.time()
        if isBinary:
            self.application_queue.put_nowait(
                {"type": "websocket.receive", "bytes": payload}
            )
        else:
            self.application_queue.put_nowait(
                {"type": "websocket.receive", "text": payload.decode("utf8")}
            )

which means on receiving new event from frontend, it creates a new task and add it to async queue for consumer to consume. So, why sleep on one async task blocking other task? Am I missing something here?

daphne django-channels python-asyncio django
1 Answers

By the producer-consumer pattern, each consumer consumes the queue serially: get from queue > process > get from queue...
At the process phase for a given item (or "event") you are putting the consumer to sleep, so the process is not finished until the sleep is done. You can't consume the next item, that is why your consumer gets blocked.
You can make a new and dedicated consumer for that specific "event", that way the sleep will not block the processing of other events. See asyncio.Queue, have your AsyncWebSocketConsumer own an async queue and fill that with .put_nowait() and have a daemon coroutine (a new consumer) consume that with .get(), that new consumer can await asyncio.sleep(10) before getting the next item.
Note that queue should have a carefully defined maxsize, since it's items are popped every 10 seconds letting it grow without limits may consume much memory. When que queue is full, put_nowait() raises asyncio.QueueFull, catch it in AsyncWebSocketConsumer and just throw away the new item or find a way to tell the producer to slow down.

ps: I don't know how django channels works, I have never used it. (#twisted tag is not appropriate for this question).

user_121734510
Related