Как работает Django channels?

Question

[wawa]

Разбираюсь с channels.
Читаю официальный тутор https://channels.readthedocs.io/en/latest/tutorial...
Как я понял один инстанс Consumer крутится в своем отдельном процессе/воркере и обслуживает одного клиента на протяжении всего времени соединения.
Т.е. 1 процесс == 1 Consumer() == 1 Connection == 1 клиент.
Если это так, то если я создам 10 воркеров (предполагается, что их много не должно быть, как я понял), то чатом пользоваться смогут только 10 человек одновременно.
Ну тогда грош цена всей этой технологии.
Очевидно в чем-то я ошибаюсь.
Что я упускаю?

В документации каналов опущен (либо размазан по всей доке) контекст работы всей этой системы. Я имею в виду контекст ОС. Можете объяснить как это работает оперируя такими сущностями как сокет, соединение, epool, процесс, тред, RAM? Буду очень благодарен.

Answer 1

[Сергей Горностаев]

Решает ли Django Channels проблему с блокирующими запросами?

Comments

[wawa]

Сергей Горностаев, конечно же я видел этот вопрос. Я раз 10 его перечитал, но там нет ответа на мой вопрос.

[Сергей Горностаев]

wawa, если у вас будет 10 воркеров и воркер выполняет свою работу за 50 миллисекунд, то подключены к чату могут быть одновременно сотни или даже тысячи пользователей, способных отправлять до 20 сообщений в секунду без возникновения backpressure.

[wawa]

Сергей Горностаев, спасибо за ответ. Я тоже так думал. Это по аналогии того как 10 воркеров uWSGI обрабатывают тысячи запросов, которые проксирует NGINX. Я это понимаю.
НО, как объясняет коллега выше, всё работает не так.
Честно говоря, я сейчас не могу понять почему это не работает так, как вы описали.

[Сергей Горностаев]

wawa, коллега выше объясняет то же самое. Есть асинхронный сервер Daphne, обрабатывающий соединения и общающийся с асинхронным кодом Channel по асинхронному протоколу ASGI. Синхронные обработчики существуют для простоты разработчика. Они как черепаха в салоне самолёта: сама черепаха летать не умеет и проползёт 5 метров за пять минут, если смотреть на неё находясь в том же салоне, но если смотреть с земли, то черепаха летит со скоростью 700 км/ч.

[wawa]

Сергей Горностаев, нет, коллега выше говорит об одной важной детали.
uWSGI воркер (обычный инстанс джанги в CPython процессе ОС) - stateless, и поэтому, отработав на один http запрос, будет готов обработать другой.
И поэтому неважно что воркеров меньше, чем клиентов, главное чтобы они успевали все запросы обработать. Т.е. здесь одно узкое место - производительность системы (бд например).
Консюмеры же (я про синхронные), работают каждый в своем треде как воркеры uWSGI работают каждый в своем процессе (- тут отличие незначительное). И они (консюмеры) !привязаны каждый к своему клиенту (т.е. stateful). Поэтому консюмер-тред в отстутствие данных от своего клиента !не может обрабатывать сообщения от других клиентов и просто !впустую жрёт ресурсы (накладные расходы на содержание треда). Почему так? Почему консюмеры не stateless?

[Сергей Горностаев]

wawa, с чего вы это взяли? Consumer'ы так же stateless. К тому же, реализованы они не поверх потоков. Так же один процесс, с одним главным потоком, в котором работает один синхронный Consumer. В случае асинхронных, в главном потоке воркера крутится цикл событий, на который навешано один или более Consumer'ов.

[wawa]

wawa, с чего вы это взяли? Consumer'ы так же stateless.

Сергей Горностаев, посмотрите на код консюмера из оф. тутора https://channels.readthedocs.io/en/latest/tutorial...
В методе connect создаются переменные room_group_name и room_name и они доступны в методе disconnect и не только. connect вызывается в самом начале, а дисконнект в самом конце. Т.е. инстанс консюмера должен жить очень долго (минутами, часами, ...)

from asgiref.sync import async_to_sync
from channels.generic.websocket import WebsocketConsumer
import json

class ChatConsumer(WebsocketConsumer):
    def connect(self):
        self.room_name = self.scope['url_route']['kwargs']['room_name']
        self.room_group_name = 'chat_%s' % self.room_name

        # Join room group
        async_to_sync(self.channel_layer.group_add)(
            self.room_group_name,
            self.channel_name
        )

        self.accept()

    def disconnect(self, close_code):
        # Leave room group
        async_to_sync(self.channel_layer.group_discard)(
            self.room_group_name,
            self.channel_name
        )

    # Receive message from WebSocket
    def receive(self, text_data):
        text_data_json = json.loads(text_data)
        message = text_data_json['message']

        # Send message to room group
        async_to_sync(self.channel_layer.group_send)(
            self.room_group_name,
            {
                'type': 'chat_message',
                'message': message
            }
        )

    # Receive message from room group
    def chat_message(self, event):
        message = event['message']

        # Send message to WebSocket
        self.send(text_data=json.dumps({
            'message': message
        }))

P.S: я работаю со 2ой версией channels

Answer 2

[Roman K]

Ну тогда грош цена всей этой технологии.

А, ну простите, закрываем проект.

Т.е. 1 процесс == 1 Consumer() == 1 Connection == 1 клиент.

С чего ты взял, что на каждый процесс приходится по одному consumer? На каждое соединение создаётся инстанс consumer'а, да. Да, он живёт всё время, пока поддерживается соединение. Но 1 воркер может поддерживать тысячи таких соединений. Как — читай про asyncio или исходники channels.

Comments

[wawa]

А, ну простите, закрываем проект.

...
Ну тогда грош цена всей этой технологии.
Очевидно в чем-то я ошибаюсь.
...

Roman Kitaev, прости, но ты имхо не дочитал мой вопрос.

WebsocketConsumer - синхронный. Да, я вижу, что есть и AsyncWebsocketConsumer, но тогда зачем синхронный нужен?
Он же бесполезен чуть менее, чем полностью. Тем не менее он есть.
И если его юзать, то верно ли "1 процесс == 1 Consumer() == 1 Connection == 1 клиент"?

[Roman K]

wawa, Нет, не верно.

прости, но ты имхо не дочитал мой вопрос.

прости, но ты имхо не дочитал документацию и исходники

WebsocketConsumer синхронный только для тебя. Под капотом есть функция sync_to_async, которая делает всю магию. А вот класс, который её имплементирует:

https://github.com/django/asgiref/blob/master/asgi...

Предвижу твой вопрос "а зачем тогда WebsocketConsumer, если есть AsyncWebsocketConsumer?", сразу отвечаю:
Посмотри, как реализован класс SyncToAsync и поймёшь, что AsyncWebsocketConsumer будет более производительным вариантом, т.к. SyncToAsync запускает run_in_executor

[wawa]

Чтобы разобраться в азах технологии - читать исходники? O_o
Насчет документации, да, ты прав - не дочитал.
Тем не менее, тредпул ограничен и самое главное ASGI_THREADS и кол-во воркеров фиксировано в работе аппа. Т.е. желая остаться в синхронном коде мне нужно запастисть тредами впрок? При каких числах это становится ужасной идеей? 1000, 2000? Что-то не очень хорошая схема выходит.
Что если использовать AsyncWebsocketConsumer и уже в нем использовать синхронные орм-запросы через тредпул? Ведь при увеличении коннектов к БД перфоманс растет только в самом начале, потом нет, и поэтому нет смысла в большом тредпуле. Так же эффективнее, верно?

[Roman K]

wawa, если ты хочешь реальной асинхронности — используй aiohttp и асинхронные драйвера для баз данных.

[wawa]

Roman Kitaev, Спасибо за ответы.
Можно еще один вопрос, для меня краеугольный в понимании каналов.
Почему [Daphne - консюмеры] не работают с вебсокетами также как [Nginx - uWSGI воркеры] работают с http?
Я имею в виду, что данные по вебсокету делятся на сообщения, которые делятся на фреймы. В итоге в моем наивном понимании:
1. Daphne, эффективно работая на epoll, собирает у себя в памяти(Redis?) фреймы
2. Получив последний фрейм, отправляет уже целое сообщение в воркер.
3. Воркер быстро обрабатывает сообщение и, если надо, отвечает Дафну. После чего он опять готов обрабатывать новые сообщения. Причем НЕВАЖНО от какого клиента.
И в итоге не нужно (в синхронном случае) для каждого клиента юзать отдельный тред на !протяжении! всего соединения. Важно лишь сколько сообщений приходит !одновременно!.
Почему это работает не так?