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

В статье на примере простого 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 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 хранилищам — это вполне может быть возможно.