Руководство по асинхронным приложениям

Асинхронные шаблоны проектирования плохо сочетаются со синхронной природой WSGI. Именно поэтому большинство асинхронных фреймворков (tornado, twisted, ...) реализуют специализированный API для экспонирования своих асинхронных возможностей. Bottle — это фреймворк WSGI и разделяет синхронную природу WSGI, но благодаря замечательному проекту gevent, всё ещё возможно создание асинхронных приложений с помощью bottle. Эта статья документирует использование Bottle с асинхронным WSGI.

Ограничения синхронного WSGI

Кратко, спецификация WSGI (pep 3333) определяет цикл запроса/ответа следующим образом: вызываемый объект приложения вызывается один раз для каждого запроса и должен вернуть итератор тела. Затем сервер итерирует по телу и записывает каждый фрагмент в сокет. Как только итератор тела исчерпан, соединение с клиентом закрывается.

Достаточно просто, но есть проблема: всё это происходит синхронно. Если вашему приложению нужно ждать данных (Ввод/вывод, сокеты, базы данных, ...), оно должно либо отдавать пустые строки (постоянное ожидание), либо блокировать текущую нить. Оба решения занимают обрабатывающую нить и не позволяют ей отвечать на новые запросы. Следовательно, в каждой нити может быть только один запрос.

Большинство серверов ограничивают количество нитей, чтобы избежать относительно высокой их стоимости. Распространены пулы из 20 или менее нитей. Как только все нити заняты, любое новое соединение приостанавливается. Сервер фактически недоступен для всех остальных. Если вы хотите реализовать чат, использующий длинные запросы AJAX для получения обновлений в реальном времени, вы столкнетесь с ограничением в 20 одновременных подключений. Это довольно маленький чат.

Зелёные потоки на помощь

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

Модуль gevent добавляет зелёные потоки в игру. Зелёные потоки ведут себя подобно традиционным нитям, но создаются очень дешево. Сервер на основе gevent может создавать тысячи зелёных потоков (по одному на каждое подключение) практически без накладных расходов. Блокирование отдельных зелёных потоков не влияет на способность сервера принимать новые запросы. Количество одновременных подключений практически не ограничено.

Это делает создание асинхронных приложений невероятно простым, потому что они выглядят и работают как синхронные приложения. Сервер на основе gevent на самом деле не асинхронный, а многопоточный. Вот пример:

from gevent import monkey; monkey.patch_all()

from time import sleep
from bottle import route, run

@route('/stream')
def stream():
    yield 'START'
    sleep(3)
    yield 'MIDDLE'
    sleep(5)
    yield 'END'

run(host='0.0.0.0', port=8080, server='gevent')

Первая строка важна. Она заставляет gevent подменить большинство блокирующих API Python, чтобы они не блокировали текущую нить, а вместо этого передавали управление следующему зелёному потоку. Она фактически заменяет потоки Python на псевдонитями на основе gevent. Вот почему вы всё ещё можете использовать time.sleep() , которые обычно блокируют всю нить. Если вы не уверены в подмене встроенных функций Python, вы можете использовать соответствующие функции gevent (gevent.sleep() в данном случае).

Если вы запустите этот скрипт и перенаправите свой браузер на http://localhost:8080/stream, вы должны увидеть START, MIDDLE, и END появляться по одному (вместо ожидания 8 секунд, чтобы увидеть их все сразу). Это работает точно так же, как и с обычными нитями, но теперь ваш сервер может обрабатывать тысячи одновременных запросов без проблем.

Обработчики событий

Очень распространённый шаблон проектирования в асинхронных фреймворках (включая tornado, twisted, node.js и другие) заключается в использовании неблокирующих API и привязке обработчиков к асинхронным событиям. Объект сокета остаётся открытым до явного закрытия, чтобы позволить обработчикам впоследствии записывать в сокет. Вот пример на основе библиотеки tornado:

class MainHandler(tornado.web.RequestHandler):
    @tornado.web.asynchronous
    def get(self):
        worker = SomeAsyncWorker()
        worker.on_data(lambda chunk: self.write(chunk))
        worker.on_finish(lambda: self.finish())

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

В Gevent+WSGI всё иначе: во-первых, раннее завершение не имеет преимущества, потому что у нас есть неограниченный пул (псевдо)нитей для приема новых подключений. Во-вторых, мы не можем прервать выполнение рано, потому что это закроет сокет (как требуется WSGI). В-третьих, мы должны вернуть итерируемый объект, чтобы соответствовать WSGI.

Для соответствия стандарту WSGI нам нужно только вернуть итерируемый объект тела, в который можно асинхронно записывать. С помощью gevent.queue мы можем симулировать отсоединённый сокет и переписать предыдущий пример следующим образом:

@route('/fetch')
def fetch():
    body = gevent.queue.Queue()
    worker = SomeAsyncWorker()
    worker.on_data(lambda chunk: body.put(chunk))
    worker.on_finish(lambda: body.put(StopIteration))
    return body

С точки зрения сервера, объект очереди является итерируемым. Он блокируется, если пуст, и останавливается, как только достигнет StopIteration. Это соответствует WSGI. Со стороны приложения объект очереди ведёт себя как неблокирующий сокет. В него можно записывать в любое время, передавать его и даже запускать новую (псевдо)нить, которая асинхронно записывает в него. Так чаще всего реализуется длинный опрос.

Наконец: WebSockets

Давайте забудем о низкоуровневых деталях на время и поговорим о WebSockets. Поскольку вы читаете эту статью, вы, вероятно, знаете, что такое WebSockets: двусторонний канал связи между браузером (клиент) и веб-приложением (сервер).

К счастью, пакет gevent-websocket выполняет всю сложную работу за нас. Вот простой конечный пункт WebSocket, который получает сообщения и просто отправляет их обратно клиенту:

from bottle import request, Bottle, abort
app = Bottle()

@app.route('/websocket')
def handle_websocket():
    wsock = request.environ.get('wsgi.websocket')
    if not wsock:
        abort(400, 'Expected WebSocket request.')

    while True:
        try:
            message = wsock.receive()
            wsock.send("Your message was: %r" % message)
        except WebSocketError:
            break

from gevent.pywsgi import WSGIServer
from geventwebsocket import WebSocketHandler, WebSocketError
server = WSGIServer(("0.0.0.0", 8080), app,
                    handler_class=WebSocketHandler)
server.serve_forever()

Цикл while выполняется до тех пор, пока клиент не закроет соединение. Вы поняли :)

API JavaScript для клиента также довольно прост:

<!DOCTYPE html>
<html>
<head>
  <script type="text/javascript">
    var ws = new WebSocket("ws://example.com:8080/websocket");
    ws.onopen = function() {
        ws.send("Hello, world");
    };
    ws.onmessage = function (evt) {
        alert(evt.data);
    };
  </script>
</head>
</html>