Шаг 215.
Python: сборник рецептов. Сети и веб-программирование. Разбираемся с вводом-выводом, управляемым событиями (event-driven I/O)

    На этом шаге мы рассмотрим один из способов решения этой задачи.

Задача

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

Решение

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

class EventHandler:
    def fileno(self):
        'Возвращает ассоциированный файловый дескриптор'
        raise NotImplemented('must implement')

    def wants_to_receive(self):
        'Возвращает True, если получение разрешено'
        return False

    def handle_receive(self):
        'Выполняет операцию получения'
        pass

    def wants_to_send(self):
        'Возвращает True, если отсылка запрошена' 
        return False

    def handle_send(self):
        'Отсылает исходящие данные'
        pass

    Экземпляры этого класса могут быть подключены к циклу, который выглядит так: 

import select

def event_loop(handlers):
    while True:
        wants_recv = [h for h in handlers if h.wants_to_receive()]
        wants_send = [h for h in handlers if h.wants_to_send()]
        can_recv, can_send, _ = select.select(wants_recv, wants_send, [])
        for h in can_recv:
            h.handle_receive()
        for h in can_send:
            h.handle_send()

    Вот и все! Секрет цикла событий - это вызов select(), который опрашивает файловые дескрипторы на предмет активности. Перед вызовом select() цикл событий просто опрашивает все обработчики, чтобы понять, какой из них хочет принимать или посылать. Далее он предоставляет получившиеся списки в select(). В результате select() возвращает список объектов, которые уже готовы принимать или посылать. Запускаются соответствующие методы handle_receive() или handle_send().

    Чтобы писать приложения, создаются специфические экземпляры класса EventHandler. Например, вот два простых обработчика, которые демонстрируют работу двух сетевых UDP-сервисов: 

import socket
import time

from eventhandler import EventHandler, event_loop

class UDPServer(EventHandler):
    def __init__(self, address):
        self.sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        self.sock.bind(address)

    def fileno(self):
        return self.sock.fileno()

    def wants_to_receive(self):
        return True

class UDPTimeServer(UDPServer):
    def handle_receive(self):
        msg, addr = self.sock.recvfrom(1)
        self.sock.sendto(time.ctime().encode('ascii'), addr)

class UDPEchoServer(UDPServer):
    def handle_receive(self):
        msg, addr = self.sock.recvfrom(8192)
        self.sock.sendto(msg, addr)

if __name__ == '__main__':
    handlers = [ UDPTimeServer(('',14000)), UDPEchoServer(('',15000))  ]
    event_loop(handlers)
Архив с этими файлами можно взять здесь.

    Чтобы протестировать этот код, вы можете попробовать соединиться с ним из другого интерпретатора Python: 

>>> from socket import *
>>> s = socket(AF_INET, SOCK_DGRAM)
>>> s.sendto(b'', ('localhost', 14000))
0
>>> s.recvfrom(128)
(b'Wed Oct 28 15:59:01 2026', ('127.0.0.1', 14000))
>>> s.sendto(b'Hello', ('localhost', 15000))
5
>>> s.recvfrom(128)
(b'Hello', ('127.0.0.1', 15000))
>>>

    Реализация TCP-сервера сложнее, поскольку каждый клиент вызывает создание нового объекта-обработчика. Вот пример TCP-эхо-клиента: 

import socket
from eventhandler import EventHandler, event_loop


class TCPServer(EventHandler):
    def __init__(self, address, client_handler, handler_list):
        self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        self.sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, True)
        self.sock.bind(address)
        self.sock.listen(1)
        self.client_handler = client_handler
        self.handler_list = handler_list

    def fileno(self):
        return self.sock.fileno()

    def wants_to_receive(self):
        return True

    def handle_receive(self):
        client, addr = self.sock.accept()
        # Добавляет клиента в список обработчиков цикла событий
        self.handler_list.append(self.client_handler(client, self.handler_list))


class TCPClient(EventHandler):
    def __init__(self, sock, handler_list):
        self.sock = sock
        self.handler_list = handler_list
        self.outgoing = bytearray()

    def fileno(self):
        return self.sock.fileno()

    def close(self):
        self.sock.close()
        # Удалиться из списка обработчиков цикла событий
        self.handler_list.remove(self)

    def wants_to_send(self):
        return True if self.outgoing else False

    def handle_send(self):
        nsent = self.sock.send(self.outgoing)
        self.outgoing = self.outgoing[nsent:]


class TCPEchoClient(TCPClient):
    def wants_to_receive(self):
        return True

    def handle_receive(self):
        data = self.sock.recv(8192)
        if not data:
            self.close()
        else:
            self.outgoing.extend(data)


if __name__ == '__main__':
    handlers = []
    handlers.append(TCPServer(('', 16000), TCPEchoClient, handlers))
    event_loop(handlers)
Архив с этим файлом можно взять здесь.

    Ключевой момент примера с TCP - это добавление и удаление клиентов из списка обработчика. На каждое соединение для клиента создается и добавляется в список новый обработчик. Когда соединение закрывается, каждый клиент должен позаботиться о том, чтобы удалить себя из списка.

    Если вы запустите эту программу и попробуете соединиться с ней с помощью Telnet или другого похожего инструмента, то увидите, как она эхом отправляет полученные данные обратно. Программа, по идее, должна легко работать с многочисленными клиентами.

Обсуждение

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

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

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

    Проблемы с блокированием или долгими вычислениями могут быть решены путем отсылки выполнения работы в отдельный поток или процесс. Однако координация потоков и процессов с циклом событий - это хитрая штука. Вот пример кода, который делает это с помощью модуля concurrent.futures: 

from concurrent.futures import ThreadPoolExecutor
import os


class ThreadPoolHandler(EventHandler):
    def __init__(self, nworkers):
        if os.name == 'posix':
            self.signal_done_sock, self.done_sock = socket.socketpair()
        else:
            server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            server.bind(('127.0.0.1', 0))
            server.listen(1)
            self.signal_done_sock = socket.socket(socket.AF_INET,
                     socket.SOCK_STREAM)
            self.signal_done_sock.connect(server.getsockname())

            self.done_sock, _ = server.accept()
            server.close()
            self.pending = []
            self.pool = ThreadPoolExecutor(nworkers)

    def fileno(self):
        return self.done_sock.fileno()

    # Функция обратного вызова, которая выполняется после завершения потока 
    def _complete(self, callback, r):
        self.pending.append((callback, r.result()))
        self.signal_done_sock.send(b'x')

    # Запускает функцию в пуле потоков
    def run(self, func, args=(), kwargs={},*,callback):
        r = self.pool.submit(func, *args, **kwargs)
        r.add_done_callback(lambda r: self._complete(callback, r))

    def wants_to_receive(self):
        return True

    # Запускает функции обратного вызова завершенной работы
    def handle_receive(self):
        # Вызывает все коллбэки в очереди
        for callback, result in self.pending:
            callback(result)
            self.done_sock.recv(1)
        self.pending = []

    В этом коде метод run() использован для отправки работы пулу вместе с функцией обратного вызова, которая должна быть вызвана по завершении. Затем работа отправляется в экземпляр ThreadPoolExecutor. Однако по-настоящему сложная проблема касается координации вычисленного результата и цикла событий. Чтобы это сделать, "под капотом" создается пара сокетов, которая используется как своего рода сигнальный механизм. Когда работа выполнена пулом потоков, выполняется метод _complete() в классе. Этот метод формирует очередь из ожидающего коллбэка и результата перед записью первого байта в один из этих сокетов. Метод fileno() запрограммирован возвращать другой сокет. Когда этот байт записан, он даст циклу событий сигнал о том, что что-то произошло. Когда запускается метод handle_receive(), выполняются все функции обратного вызова для ранее отправленной работы. Если честно, от этого может закружиться голова.

    Вот простой сервер, который демонстрирует использование пула потоков для выполнения длительных вычислений: 

# Очень плохая реализация поиска чисел Фибоначчи 
def fib(n):
    if n < 2:
        return 1
    else:
        return fib(n - 1) + fib(n - 2)

   
class UDPFibServer(UDPServer):
    def handle_receive(self):
        msg, addr = self.sock.recvfrom(128)
        n = int(msg)
        pool.run(fib, (n,), callback=lambda r: self.respond(r, addr))

    def respond(self, result, addr):
        self.sock.sendto(str(result).encode('ascii'), addr)

if __name__ == '__main__':
    pool = ThreadPoolHandler(16)
    handlers = [pool, UDPFibServer(('',16000))]
    event_loop(handlers)

    Чтобы опробовать сервер в работе, просто запустите его и поэкспериментируйте с другой программой на Python: 

from socket import *


sock = socket(AF_INET, SOCK_DGRAM)
for x in range(40):
    sock.sendto(str(x).encode('ascii'), ('localhost', 16000))
    resp = sock.recvfrom(8192)
    print(resp[0])

    У вас должно получиться запустить эту программу многократно, из разных окон, и все должно работать без остановки других программ, даже если все будет работать медленнее и медленнее с ростом значений чисел.

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

    В качестве альтернативы программированию на базе функций обратного вызова управляемый событиями код часто использует корутины.

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




Предыдущий шаг Содержание Следующий шаг