Примеры использования asyncio: HTTPServer?!

Не так давно зарелизилась новая версия Python 3.4 в changelog которой вошло много «вкусностей». Одна из таких — модуль asyncio, содержащий инфраструктуру пригодную для написания асинхронных сетевых приложений. Благодаря концепции сопрограмм (coroutines), код асинхронного приложения прост для понимания и поддержки.

В статье на примере простого TCP (Echo) сервера я постараюсь показать с чем едят asyncio, и рискну устранить «фатальный недостаток» этого модуля, а именно отсутствие реализации асинхронного HTTP сервера.

INTRO

Прямой конкурент и «брат» — это фреймворк tornado, который хорошо зарекомендовал себя и пользуется заслуженной популярностью. Однако на мой взгляд, asyncore выглядит проще, более логичен и продуман. Впрочем это не удивительно, ведь мы имеем дело со стандартной библиотекой языка.

Вы скажете, что на Python можно было и раньше писать асинхронные сервисы и будете правы. Но для этого требовались сторонние библиотеки и/или использования callback стиля программирования. Концепция coroutine доведенная в этой версии Python практически до совершенства позволяет писать линейный асинхронный код использую лишь возможности стандартных библиотек языка.

Сразу хочу оговориться, что все это я писал под Linux, однако все используемые компоненты кроссплатформенные и под Windows тоже должно заработать. Но версия Python 3.4 обязательна.

EchoServer

Пример Echo сервера есть в стандартной документации, но относится это к low-level API «Transports and protocols». Для «повседневного» использования рекомендуется high-level API «Streams». Пример кода TCP сервера в нем отсутствует, однако изучив пример из low-level API и посмотрев исходники того и другого модуля, написать простой TCP сервер не составляет труда.

import asyncio
import logging
import concurrent.futures

@asyncio.coroutine
def handle_connection(reader, writer):
    peername = writer.get_extra_info('peername')
    logging.info('Accepted connection from {}'.format(peername))
    while True:
        try:
            data = yield from asyncio.wait_for(reader.readline(), timeout=10.0)
            if data: 
                writer.write(data)
            else:
                logging.info('Connection from {} closed by peer'.format(peername))
                break
        except concurrent.futures.TimeoutError:
            logging.info('Connection from {} closed by timeout'.format(peername))
            break
    writer.close()

if __name__ == '__main__':
    loop = asyncio.get_event_loop()
    logging.basicConfig(level=logging.INFO)
    server_gen = asyncio.start_server(handle_connection, port=2007)
    server = loop.run_until_complete(server_gen)
    logging.info('Listening established on {0}'.format(server.sockets[0].getsockname()))
    try:
        loop.run_forever()
    except KeyboardInterrupt:
        pass # Press Ctrl+C to stop
    finally:
        server.close()
        loop.close()

Все достаточно очевидно, но есть пара нюансов на которые стоит обратить внимание.

    server_gen = asyncio.start_server(handle_connection, port=2007)
    server = loop.run_until_complete(server_gen)

В первой строке создается не сам сервер, а генератор, который при первом обращении к нему и недр asyncio создает и инициализирует по заданным параметрам TCP сервер. Вторая строка и есть пример такого обращения.

        try:
            data = yield from asyncio.wait_for(reader.readline(), timeout=10.0)
            if data: 
                writer.write(data)
            else:
                logging.info('Connection from {} closed by peer'.format(peername))
                break
        except concurrent.futures.TimeoutError:
            logging.info('Connection from {} closed by timeout'.format(peername))
            break

Функция-coroutine reader.readline() производит асинхронное чтение данных из входного потока. Но ожидание данных для чтения не ограниченно по времени, если нужно его прекратить по таймауту необходимо обернуть вызов функции-coroutine в asyncio.wait_for(). В этом случае по истечению заданного в секундах интервала времени будет поднято исключение concurrent.futures.TimeoutError, которое можно обработать необходимым образом.
Проверка что reader.readline() возвращает не пустое значение в данном примере обязательна. Иначе после разрыва соединения клиентом (connection reset by peer), попытки чтения и возврат пустого значения будут продолжаться до бесконечности.

А как же ООП?

С ООП тоже все хорошо. Достаточно обернуть методы использующие вызовы функций-coroutine в декоратор @asyncio.coroutine. Какие функции запускаются как coroutine в API явно указывается. Ниже пример реализующий класс EchoServer.

import asyncio
import logging
import concurrent.futures

class EchoServer(object):
    """Echo server class"""
    
    def __init__(self, host, port, loop=None):
        self._loop = loop or asyncio.get_event_loop() 
        self._server = asyncio.start_server(self.handle_connection, port=2007)
    
    def start(self, and_loop=True):
        self._server = self._loop.run_until_complete(self._server)
        logging.info('Listening established on {0}'.format(self._server.sockets[0].getsockname()))
        if and_loop:
            self._loop.run_forever()
    
    def stop(self, and_loop=True):
        self._server.close()
        if and_loop:
            self._loop.close()
    
    @asyncio.coroutine    
    def handle_connection(self, reader, writer):
        peername = writer.get_extra_info('peername')
        logging.info('Accepted connection from {}'.format(peername))
        while not reader.at_eof():
            try:
                data = yield from asyncio.wait_for(reader.readline(), timeout=10.0)
                writer.write(data)
            except concurrent.futures.TimeoutError:
                break
        writer.close()

if __name__ == '__main__':
    logging.basicConfig(level=logging.DEBUG)
    server = EchoServer('127.0.0.1', 2007)
    try:
        server.start()
    except KeyboardInterrupt:
        pass # Press Ctrl+C to stop
    finally:
        server.stop()

Как видно и в первом и во втором случае, код линейный и вполне читаемый. А во втором случае к тому же код оформлен в самодостаточный класс.

HTTP Server

Разобравшись со всем этим невольно возникает желание сделать что-то более существенное. Модуль asyncio предоставляет нам и такую возможность. В нем в отличии например от tornado не реализован HTTP сервер. Как говориться грех не попробовать исправить это упущение :)

Писать целиком с нуля HTTP сервер со всеми его классами типа HTTPRequest и т. п. — не спортивно, учитывая что есть масса готовых фреймворков работающих поверх протокола WSGI. Те кто в курсе справедливо заметят, что WSGI синхронный протокол. Это верно, но считать данные для environ и тело запроса можно асинхронно. Выдача результата в WSGI рекомендована в виде генератора, и это хорошо вписывается в концепцию coroutines используемую в asyncio.

Одним из фреймворков, который все делает правильно с отдачей контента является bottle. Так он например выдает содержимое файла не целиком, а порциями через генератор. Поэтому я выбрал его для тестирование разрабатываемого WSGI сервера и остался доволен результатом. Например демо приложение вполне оказалось способно отдавать большой файл на несколько клиентских подключений одновременно.

Полностью посмотреть что получилось можно у меня на github. Ни тестов, ни документации там пока нет, зато есть демо приложение использующее bottle фреймворк. Оно выдает список файлов в определенном каталоге и отдает выбранный в асинхронном режиме в независимости от размера. Так что если накидать в этот каталог фильмов, можно организовать небольшой видеохостинг :)

Хотелось бы сказать отдельное спасибо команде разработчиков CherryPy, в их код я часто поглядывал и кое-что взял целиком, что бы не придумывать «своих велосипедов».

import bottle
import os.path
from os import listdir
from bottle import route, template, static_file

root = os.path.abspath(os.path.dirname(__file__)) 


@route('/')
def index():
    tmpl = """<!DOCTYPE html>
<html>
<head><title>Bottle of Aqua</title></head>
</body>
<h3>List of files:</h3>
<ul>
  % for item in files:
    <li><a href="/files/{{item}}">{{item}}</a></li>
  % end
</ul>
</body>
</html>
"""
    files = [file_name for file_name in listdir(os.path.join(root, 'files'))
                        if os.path.isfile(os.path.join(root, 'files', file_name))]
    return template(tmpl, files=files)


@route('/files/<filename>')
def server_static(filename):
    return static_file(filename, root=os.path.join(root,'files'))


class AquaServer(bottle.ServerAdapter):
    """Bottle server adapter"""
    def run(self, handler):
        import asyncio
        import logging
        from aqua.wsgiserver import WSGIServer
        
        logging.basicConfig(level=logging.ERROR)
        loop = asyncio.get_event_loop()
        server = WSGIServer(handler, loop=loop)
        server.bind(self.host, self.port)

        try:
            loop.run_forever()
        except KeyboardInterrupt:
            pass # Press Ctrl+C to stop
        finally:
            server.unbindAll()
            loop.close()

if __name__ == '__main__':
    bottle.run(server=AquaServer, port=5000)

При написании кода WSGI сервера, я не заметил каких-то нюансов, которые можно бы было отнести на счет модуля asyncio. Единственный момент, это особенность браузеров (например хрома), сбрасывать запрос если он видит что начинает получать большой файл. Очевидно это сделано с целью переключения на более оптимизированный способ загрузки больших файлов, ибо следом запрос повторяется и файл начинает приниматься штатно. Но первый сброшенный запрос вызывает исключение ConnectionResetError, если отдача файла по нему уже началась с помощь вызова функции StreamWriter.write(). Этот случай надо обрабатывать и закрывать соединение с помощью StreamWriter.close().

Производительность

Для сравнительного теста я выбрал утилиту siege. В качестве подопытных выступили «наш пациент» (он же aqua :) в связке с bottle, достаточно популярный Waitress WSGI сервер тоже в связке с bottle и конечно же Tornado. В качестве приложения был минимально возможный helloword. Тесты проводил со следующими параметрами: 100 и 1000 одновременных подключений; длительность теста 10 секунд; три варианта размера отдаваемых данных соответственно 13 байт, 13 килобайт и 13 мегабайт. Ниже результат:

Ну что сказать? Tornado конечно рулит, но «наш пациент» кажется вырывается вперед на больших файлах и улучшил относительные показатели на большем числе соединений. К тому же он уверено обошел waitress (с его четырьмя дочерними процессами по числу ядер), который не на плохом счету среди разработчиков. Не скажу что моё тестирование адекватно на 100%, но как оценочное наверно сгодиться.

Ниже лог еще четырех более жестких тестов. 100 и 1000 подключений, 13Mb тело, 60 секунд продолжительность. Который так же показывает, что с большими файлами сервер под asyncio работает увереннее.

$ siege -c 100 -b -t 60S http://127.0.0.1:5000/
** SIEGE 2.70
** Preparing 100 concurrent users for battle.
The server is now under siege...
Lifting the server siege...      done.
Transactions:               1556 hits
Availability:              99.87 %
Elapsed time:              59.21 secs
Data transferred:       20228.00 MB
Response time:              3.69 secs
Transaction rate:          26.28 trans/sec
Throughput:           341.63 MB/sec
Concurrency:               96.87
Successful transactions:        1556
Failed transactions:               2
Longest transaction:            4.53
Shortest transaction:           1.15

$ siege -c 1000 -b -t 60S http://127.0.0.1:5000/
** SIEGE 2.70
** Preparing 1000 concurrent users for battle.
Transactions:               1570 hits
Availability:              60.18 %
Elapsed time:              59.84 secs
Data transferred:       20410.00 MB
Response time:              5.56 secs
Transaction rate:          26.24 trans/sec
Throughput:           341.08 MB/sec
Concurrency:              145.80
Successful transactions:        1570
Failed transactions:            1039
Longest transaction:           20.44
Shortest transaction:           0.00

$ siege -c 100 -b -t 60S http://127.0.0.1:5002/
** SIEGE 2.70
** Preparing 100 concurrent users for battle.
The server is now under siege...
Lifting the server siege...      done.
Transactions:                942 hits
Availability:             100.00 %
Elapsed time:              59.69 secs
Data transferred:       12246.00 MB
Response time:              5.97 secs
Transaction rate:          15.78 trans/sec
Throughput:           205.16 MB/sec
Concurrency:               94.25
Successful transactions:         942
Failed transactions:               0
Longest transaction:           14.82
Shortest transaction:           0.57

$ siege -c 1000 -b -t 60S http://127.0.0.1:5002/
** SIEGE 2.70
** Preparing 1000 concurrent users for battle.
The server is now under siege...
Lifting the server siege...      done.
Transactions:                857 hits
Availability:              55.65 %
Elapsed time:              59.07 secs
Data transferred:       11141.00 MB
Response time:             20.14 secs
Transaction rate:          14.51 trans/sec
Throughput:           188.61 MB/sec
Concurrency:              292.16
Successful transactions:         857
Failed transactions:             683
Longest transaction:           51.19
Shortest transaction:           3.26

OUTRO

Асинхронный вебсервер с использованием asyncio имеет право на жизнь. Возможно говорить об использовании таких серверов в серьезных проектах пока рано, но после тестирования, обкатки и с появлением асинхронных драйверов asyncio к базам данных и key-value хранилищам — это вполне может быть возможно.

Автор: Alesh

Источник