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

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

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

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

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

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

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

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

Модуль 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(body.put)
    worker.on_finish(lambda: body.put(StopIteration))
    worker.start()
    return body

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

Наконец: WebSocket

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

К счастью, пакет 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>