Tornado. Асинхронное программирование

Для начинающих асинхронное программирование может показаться сложным, поэтому нужно освоить базовые понятия, чтобы избежать распространенных ошибок. В интернете есть много полезных ресурсов на которых описаны основные концепции асинхронного программирования. Но сейчас мы остановимся на асинхронном программировании с помощью Python фреймворка Tornado.

С домашней страницы Tornado:

Веб сервер проекта FriendFeed это относительно простой, не блокируемый веб сервер написанный на Python. Приложение FriendFeed написано на фреймворке подобно web.py или webapp от Google, но с дополнительными инструментами и оптимизацией для использования неблокируемого веб сервера. Tornado это общедоступная версия этого веб сервера с наиболее часто используемыми инструментами в FriendFeed. Фреймворк отличается от основных веб фреймворков(а так же от большинства Python фреймворков) тем что он не блокируемый и достаточно быстрый. Не блокируемый, потому что использует epoll или kqueue, и может обрабатывать тысячи одновременных постоянных соединений, что идеально для веб сервисов работающих в реальном времени. Мы спроектировали веб сервер специально для обработки открытых активных соединений в реальном времени. (Для более детальной информации о масштабировании серверов поддерживающих тысячи соединений, смотрите проблему C10K).

На первом шаге нужно уяснить на сколько необходим и вообще необходим ли асинхронный подход. Асинхронное программирование сложно воспринимается и более сложное, чем синхронное.

Вы должны использовать асинхронный подход, когда вашему приложению нужно следить за каким нибудь ресурсом и реагировать на изменение его состояния. Например, идеальным вариантом является сервер принимающий соединение через сокет. Или приложение, которое выполняет задачи периодически или отложенное выполнение. В качестве альтернативы можно использовать несколько потоков(или процессов) для управления несколькими задачами, но такая модель сложная.

Второй шаг это выяснить, можете ли вы пойти асинхронным путем. К сожалению, в Tornado, не все задачи могут быть выполнены асинхронно.

Торнадо однопоточен (но поддерживает несколько потоков при сложной конфигурации), поэтому любая "блокировка" будет блокировать весь сервер. Это означает, что блокирующая задача не позволит фреймворку выполнить следующую задачу, до того времени пока блокировка не будет снята. Выбор задач осуществляется в IOLoop, который, работает в единственном доступном потоке.

Для примера, это НЕВЕРНЫЙ подход использования IOLoop:

import time
from tornado.ioloop import IOLoop
from tornado import gen
 
def my_function(callback):
	print('do some work')
	# Эта строчка блокирует выполнение
	time.sleep(1)
	callback(123)
 
@gen.engine
def f():
	print('start')
	# Вызов my_function и возврат результата, как только вызовется "callback"
	# Результат - это аргумент, передаваемый в функцию обратного вызова(callback)
	result = yield gen.Task(my_function)
	print('result is', result)
	IOLoop.instance().stop()
 
if __name__ == "__main__":
	f()
	IOLoop.instance().start()

Обратите внимание, что my_function - вызывается корректно, но является блокируемой(строчка time.sleep), таким образом следующая операция не выполнится(вывод в консоль результата), до завершения текущей. Только тогда, когда будет получен результат выполнения, программа пойдет дальше.

Для сравнения, перепишем этот алгоритм, но будем использовать асинхронную версию time.sleep, то есть add_timeout:

import time
from tornado.ioloop import IOLoop
from tornado import gen

@gen.engine
def f():
	print('sleeping')
	yield gen.Task(IOLoop.instance().add_timeout, time.time() + 1)
	print('awake!')

if __name__ == "__main__":
	# Обратите внимание, что код выполняется "одновременно"
	IOLoop.instance().add_callback(f)
	IOLoop.instance().add_callback(f)
	IOLoop.instance().start()

В этом случае, первый вызов напечатает "sleeping", затем выполнение передастся в IOLoop и будет запланировано выполнение следующих задач, через 1 секунду. Когда контроль передастся в IOLoop, будет выполнен второй вызов, который напечатает "sleeping" снова, после этого управление передастся в IOLoop. После первой секунды ожидания IOLoop продолжает работать и напечатает "awake!", с того места, где был первый вызов функции. И в самом конце напечатается "awake!" второй раз. То есть последовательность вывода будет такая: "sleeping", "sleeping", "awake!", "awake!". Два вызова функций были одновременны(не параллельны, хотя!).

Что же, постает вопрос, "как можно создать асинхронно выполняющую функцию"?. В Tornado, каждая функция, которая имеет "callback" может использоваться с gen.engine.Task. Однако, будьте готовы к тому, что использование Task не делает вызов асинхронным! Здесь нет никакой магии, функция просто добавляется на выполнение, выполняется, и значение, что передается в callback, возвращается gen.Task. Смотрите ниже:

import time
from tornado.ioloop import IOLoop
from tornado import gen
 
def my_function(callback):
	print('do some work')
	# Эта строчка блокирует выполнение
	time.sleep(1)
	callback(123)
 
@gen.engine
def f():
	print('start')
	# Вызов my_function и возврат результата, как только вызовется "callback"
	# Результат - это аргумент, передаваемый в функцию обратного вызова(callback)
	result = yield gen.Task(my_function)
	print('result is', result)
	IOLoop.instance().stop()
 
if __name__ == "__main__":
	f()
	IOLoop.instance().start()

Большинство начинающих ожидают, что когда они напишут, например, такой код: Task(my_function), то my_func автоматически будет выполнятся асинхронно. Но это не так, Tornado так не работает. Так работает Go! Функции, которые должны быть асинхронными, должны использовать асинхронные библиотеки.

Имеется ввиду, что вызовы, подобные time.sleep или urllib2.urlopen или db.query должны быть заменены на эквивалентные асинхронные аналоги. Для примера, IOLoop.add_timeout вместо time.sleep, AsyncHTTPClient.fetch вместо urllib2.urlopen и т.д. Для запросов к базе данных ситуация более сложная и нужен специфический асинхронный драйвер. Например, Motor для MongoDB.

Источник: Asynchronous programming with Tornado