Как лучше работать с вычислительными задачами и потоками во Flask?

Question

[Everything_is_not_so_bad]

В некотором экшене во Flask отправляется JSON с порядка 25 элементов-строк для анализа тональности текста. Анализ представляет из себя вычислительную задачу. Поэтому, для ускорения возврата результата автор создает количество потоков, соответствующее количеству элементов, запускает их и ждет при помощи цикла while:

for item in items:
    threading.Thread(target=self.analyze, args=(item, language)).start()

while None in self.items:
    time.sleep(0.02)

Не будучи гуру, я посоветовал заменить конструкцию цикла на использование .join(), на что автор отвечает, что иначе во Flask будет блокироваться основной поток. Я не знаю что на это ответить, поскольку не знаю как под капотом это работает.

Хорошая ли практика? Что посоветуете?

Answer 1

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

Во-первых, while блокирует воркер так же, как и join. Во-вторых, Flask не предназначен для многопоточности и её в нём следует избегать. Лучше использовать асинхронность или очередь задач.

Comments

[Everything_is_not_so_bad]

Лучше использовать асинхронность или очередь задач.

Спасибо. Что посоветуете для версии Питона 3.10? Пару ссылок на доки/примеры, если можно.

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

Román Mirilaczvili, https://flask.palletsprojects.com/en/2.2.x/async-await/

[Everything_is_not_so_bad]

Сергей Горностаев,
Насчёт ссылки:

Async is beneficial when performing concurrent IO-bound tasks, but will probably not improve CPU-bound tasks.

То есть, нет смысла, я так понимаю. Так?

Ещё хотел прояснить вопрос насчёт блокирования: если нам важно дожидаться окончания вычисления, то как можно обходиться очередями? Например, кинуть все строки в очередь, у которой будут X обработчиков, затем колбэками обработчиков получать ответы, агрегировать их в какой-то структуре и ждать получения всех ответов в экшене? Не очень понимаю как всё это реализовывать.

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

Román Mirilaczvili,

То есть, нет смысла, я так понимаю. Так?

Верно. Впрочем, в Python и потоки не очень помогают с CPU-bound задачами.

как можно обходиться очередями?

С очередями лучше поменять интерфейс. Например отправлять задание в Celery и сразу возвращать клиенту идентификатор, по которому он может опрашивать бэкенд о готовности результата. Если это невозможно, то можно написать асинхронный обработчик, в котором в цикле проверять готовность результата, примерно как в коде вашего вопроса. Ещё можно написать собственную очередь задач:

1. Отправлять задания брокеру очередей.
   
2. Другим процессом Python вычитывать задания из очереди, выполнять их и отвечать в другую очередь.
   
3. В асинхронном обработчике запросов Flask с помощью асинхронного потребителя Pika ждать появления сообщений с ответами.
   

Жаль, что Celery не готова работать с asyncio. Ну и наконец вместо Flask можно взять какой-нибудь фреймворк лучше справляющийся с конкурентностью. Хотя с вычислительными задачами в любом питонячьем фреймворке просто не будет.

[Everything_is_not_so_bad]

В итоге переписано без Flask, при помощи асинхронной очереди.

[perepe1ka]

к чему тогда вопрос, если иначе можно.