asyncio (RU)

18.5.4. Транспорты и протоколыAPI на основе колбэков​

Исходный код: Lib/asyncio/transports.py

Исходный код: Lib/asyncio/protocols.py

18.5.4.1. Транспорты​

Транспорты – это классы, предоставляемые asyncio для абстрагирования различных видов каналов связи. Обычно транспорт не создаётся самостоятельно; вместо этого вызывается метод AbstractEventLoop, который создаёт транспорт и пытается инициировать базовый канал связи, вызывая колбэк при успехе.

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

asyncio в настоящее время реализует транспорты для TCP, UDP, SSL и каналов подпроцессов. Доступные методы транспорта зависят от его типа.

Классы транспортов не являются потокобезопасными.

Изменено в версии 3.6: Параметр сокета TCP_NODELAY теперь установлен по умолчанию.

18.5.4.1.1. BaseTransport​

classasyncio.BaseTransport

Базовый класс для транспортов.

close()

Закрывает транспорт. Если транспорт имеет буфер для исходящих данных, буферизованные данные будут сброшены асинхронно. Новые данные приниматься не будут. После сброса всех буферизованных данных будет вызван метод connection_lost() протокола с аргументом None.

is_closing()

Возвращает True, если транспорт закрывается или закрыт.

Новое в версии 3.5.1.

get_extra_info(name, default=None)

Возвращает дополнительную информацию о транспорте. name – строка, представляющая запрашиваемую часть информации, специфичной для транспорта; default – значение, которое возвращается, если информация отсутствует.

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

  • сокет:

  • SSL-сокет:

    • 'compression': используемый алгоритм сжатия в виде строки, или None, если соединение не сжато; результат ssl.SSLSocket.compression()

    • 'cipher': кортеж из трёх значений, содержащий имя используемого шифра, версию протокола SSL, определяющую его использование, и количество используемых секретных бит; результат ssl.SSLSocket.cipher()

    • 'peercert': сертификат пира; результат ssl.SSLSocket.getpeercert()

    • 'sslcontext': экземпляр ssl.SSLContext

    • 'ssl_object': экземпляр ssl.SSLObject или ssl.SSLSocket

  • канал:

    • 'pipe': объект канала

  • подпроцесс:

set_protocol(протокол)

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

Новое в версии 3.5.3.

get_protocol()

Возвращает текущий протокол.

Новое в версии 3.5.3.

Изменено в версии 3.5.1: Информация 'ssl_object' была добавлена для SSL-сокетов.

18.5.4.1.2. ReadTransport​

classasyncio.ReadTransport

Интерфейс для транспортов только для чтения.

pause_reading()

Приостанавливает приёмную сторону транспорта. Данные не будут передаваться методу data_received() протокола до вызова resume_reading().

Изменено в версии 3.6.7: Метод является идемпотентным, т.е. его можно вызывать, когда транспорт уже приостановлен или закрыт.

resume_reading()

Возобновляет приёмную сторону. Метод data_received() протокола будет снова вызван, если для чтения доступны какие-либо данные.

Изменено в версии 3.6.7: Метод является идемпотентным, т.е. его можно вызывать, когда транспорт уже читает.

18.5.4.1.3. WriteTransport​

classasyncio.WriteTransport

Интерфейс для транспортов только для записи.

abort()

Немедленно закрывает транспорт, не дожидаясь завершения ожидающих операций. Буферизованные данные будут потеряны. Новые данные приниматься не будут. Метод connection_lost() протокола в конечном итоге будет вызван с None в качестве аргумента.

can_write_eof()

Возвращает True, если транспорт поддерживает write_eof(), и False в противном случае.

get_write_buffer_size()

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

get_write_buffer_limits()

Получает верхний (high) и нижний (low) пороги для управления потоком записи. Возвращает кортеж (low, high), где low и high – положительные числа байтов.

Для установки порогов используется set_write_buffer_limits().

Новое в версии 3.4.2.

set_write_buffer_limits(high=None, low=None)

Устанавливает верхний (high) и нижний (low) пороги для управления потоком записи.

Эти два значения (измеряемые в байтах) управляют тем, когда вызываются методы pause_writing() и resume_writing() протокола. Если заданы, нижний предел (low-water limit) должен быть меньше или равен верхнему пределу (high-water limit). Ни верхний, ни нижний предел не могут быть отрицательными.

pause_writing() вызывается, когда размер буфера становится больше или равен значению верхнего. Если запись была приостановлена, resume_writing() вызывается, когда размер буфера становится меньше или равен значению нижнего.

Значения по умолчанию зависят от реализации. Если задан только верхний предел, нижний предел по умолчанию принимает значение, зависящее от реализации, которое меньше или равно верхнему пределу. Установка верхнего в ноль также обнуляет нижний и приводит к вызову pause_writing() всякий раз, когда буфер становится непустым. Установка нижнего в ноль приводит к вызову resume_writing() только после опустошения буфера. Использование нуля для любого из пределов в целом неоптимально, так как уменьшает возможности для параллельного выполнения ввода-вывода и вычислений.

Используйте get_write_buffer_limits() для получения пределов.

write(data)

Записывает в транспорт некоторые данные (байты).

Этот метод не блокирует выполнение; он буферизует данные и обеспечивает их асинхронную отправку.

writelines(list_of_data)

Записывает список (или любой итерируемый объект) байтов данных в транспорт. Функционально это эквивалентно вызову write() для каждого элемента, полученного из итерируемого объекта, но может быть реализовано более эффективно.

write_eof()

Закрывает конец записи транспорта после сброса буферизованных данных. Данные всё ещё могут приниматься.

Этот метод может вызвать NotImplementedError, если транспорт (например, SSL) не поддерживает половинное закрытие.

18.5.4.1.4. DatagramTransport​

DatagramTransport.sendto(data, addr=None)

Отправляет байты данных удалённому узлу, заданному addr (адрес назначения, зависящий от транспорта). Если addr равно None, данные отправляются на адрес назначения, указанный при создании транспорта.

Этот метод не блокирует выполнение; он буферизует данные и обеспечивает их асинхронную отправку.

DatagramTransport.abort()

Немедленно закрывает транспорт, не дожидаясь завершения ожидающих операций. Буферизованные данные будут потеряны. Новые данные приниматься не будут. Метод connection_lost() протокола в конечном итоге будет вызван с None в качестве аргумента.

18.5.4.1.5. BaseSubprocessTransport​

classasyncio.BaseSubprocessTransport
get_pid()

Возвращает идентификатор процесса подпроцесса в виде целого числа.

get_pipe_transport(fd)

Возвращает транспорт для канала связи, соответствующего целочисленному файловому дескриптору fd:

  • 0: потоковый транспорт для чтения стандартного ввода (stdin) или None, если подпроцесс не был создан с stdin=PIPE

  • 1: потоковый транспорт для записи стандартного вывода (stdout) или None, если подпроцесс не был создан с stdout=PIPE

  • 2: потоковый транспорт для записи стандартного потока ошибок (stderr) или None, если подпроцесс не был создан с stderr=PIPE

  • другой fd: None

get_returncode()

Возвращает код возврата подпроцесса в виде целого числа или None, если он ещё не завершился, аналогично атрибуту subprocess.Popen.returncode.

kill()

Убивает подпроцесс, как в subprocess.Popen.kill().

В системах POSIX функция отправляет SIGKILL подпроцессу. В Windows этот метод является псевдонимом для terminate().

send_signal(signal)

Отправляет номер сигнала signal подпроцессу, как в subprocess.Popen.send_signal().

terminate()

Просит подпроцесс остановиться, как в subprocess.Popen.terminate(). Этот метод является псевдонимом для метода close().

На системах POSIX этот метод отправляет SIGTERM подпроцессу. На Windows для остановки подпроцесса вызывается функция TerminateProcess() из Windows API.

close()

Просит подпроцесс остановиться, вызывая метод terminate(), если подпроцесс ещё не завершился, и закрывает транспорты всех каналов (stdin, stdout и stderr).

18.5.4.2. Протоколы​

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

При наследовании класса протокола рекомендуется переопределять определённые методы. Эти методы являются колбэками: они вызываются транспортом при определённых событиях (например, при получении данных); не следует вызывать их самостоятельно, если только вы не реализуете транспорт.

Примечание

Все колбэки имеют пустые реализации по умолчанию. Поэтому вам нужно реализовать только те колбэки, которые соответствуют интересующим вас событиям.

18.5.4.2.1. Классы протоколов​

classasyncio.Protocol

Базовый класс для реализации потоковых протоколов (для использования, например, с транспортами TCP и SSL).

classasyncio.DatagramProtocol

Базовый класс для реализации дейтаграммных протоколов (для использования, например, с транспортами UDP).

classasyncio.SubprocessProtocol

Базовый класс для реализации протоколов, взаимодействующих с дочерними процессами (через набор однонаправленных каналов).

18.5.4.2.2. Колбэки соединения​

Эти колбэки могут вызываться на экземплярах Protocol, DatagramProtocol и SubprocessProtocol:

BaseProtocol.connection_made(транспорт)

Вызывается при установлении соединения.

Аргумент transport – это транспорт, представляющий соединение. Вы отвечаете за его сохранение (например, в виде атрибута), если это необходимо.

BaseProtocol.connection_lost(exc)

Вызывается при потере или закрытии соединения.

Аргумент является либо объектом исключения, либо None. Последнее означает, что получен обычный EOF, или соединение было прервано или закрыто этой стороной соединения.

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

Следующие колбэки могут вызываться только на экземплярах SubprocessProtocol:

SubprocessProtocol.pipe_data_received(fd, data)

Вызывается, когда дочерний процесс записывает данные в свой канал stdout или stderr. fd – целочисленный файловый дескриптор канала. data – непустой объект bytes, содержащий данные.

SubprocessProtocol.pipe_connection_lost(fd, exc)

Вызывается, когда один из каналов, связывающихся с дочерним процессом, закрывается. fd – целочисленный файловый дескриптор, который был закрыт.

SubprocessProtocol.process_exited()

Вызывается, когда дочерний процесс завершился.

18.5.4.2.3. Потоковые протоколы​

Следующие колбэки вызываются на экземплярах Protocol:

Protocol.data_received(data)

Вызывается, когда получены некоторые данные. data – непустой объект bytes, содержащий входящие данные.

Примечание

Буферизация, фрагментация или сборка данных зависит от транспорта. В общем случае не следует полагаться на конкретные особенности семантики; вместо этого стоит сделать парсинг достаточно общим и гибким. Однако данные всегда принимаются в правильном порядке.

Protocol.eof_received()

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

Этот метод может вернуть ложное значение (включая None), и в этом случае транспорт закроется сам. И наоборот, если этот метод возвращает истинное значение, закрытие транспорта остаётся на усмотрение протокола. Поскольку реализация по умолчанию возвращает None, она неявно закрывает соединение.

Примечание

Некоторые транспорты, такие как SSL, не поддерживают полузакрытые соединения; в таких случаях возврат true из этого метода не предотвратит закрытие соединения.

data_received() может вызываться произвольное количество раз в течение соединения. Однако eof_received() вызывается не более одного раза, и если он вызван, data_received() после него вызываться не будет.

Машина состояний:

18.5.4.2.4. Дейтаграммные протоколы​

Следующие колбэки вызываются на экземплярах DatagramProtocol.

DatagramProtocol.datagram_received(data, addr)

Вызывается при получении дейтаграммы. data – это объект bytes, содержащий входящие данные. addr – это адрес узла, отправляющего данные; точный формат зависит от транспорта.

DatagramProtocol.error_received(exc)

Вызывается, когда предыдущая операция отправки или получения вызывает OSError. exc – это экземпляр OSError.

Этот метод вызывается в редких случаях, когда транспорт (например, UDP) обнаруживает, что дейтаграмма не может быть доставлена получателю. Однако во многих случаях недоставленные дейтаграммы будут молча отброшены.

18.5.4.2.5. Колбэки управления потоком​

Эти колбэки могут вызываться на экземплярах Protocol, DatagramProtocol и SubprocessProtocol:

BaseProtocol.pause_writing()

Вызывается, когда буфер транспорта превышает верхнюю отметку.

BaseProtocol.resume_writing()

Вызывается, когда буфер транспорта опускается ниже нижней отметки.

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

Примечание

Если размер буфера равен верхней отметке, pause_writing() не вызывается – он должен быть строго больше. И наоборот, resume_writing() вызывается, когда размер буфера меньше или равен нижней отметке. Эти граничные условия важны для обеспечения ожидаемого поведения, когда любая из отметок равна нулю.

Примечание

В системах BSD (OS X, FreeBSD и т.д.) управление потоком не поддерживается для DatagramProtocol, потому что ошибки отправки, вызванные записью слишком большого количества пакетов, трудно обнаружить. Сокет всегда выглядит «готовым», и лишние пакеты отбрасываются; может быть возбуждено OSError с errno, установленным в errno.ENOBUFS, но это не гарантируется; если оно возбуждается, оно сообщается в DatagramProtocol.error_received(), но в остальном игнорируется.

18.5.4.2.6. Корутины и протоколы​

Корутины могут быть запланированы в методе протокола с помощью ensure_future(), но порядок выполнения не гарантируется. Протоколы не знают о корутинах, созданных в методах протокола, и поэтому не будут их ждать.

Для обеспечения надёжного порядка выполнения используйте потоковые объекты в корутине с yield from. Например, корутина StreamWriter.drain() может использоваться для ожидания, пока буфер записи не будет очищен.

18.5.4.3. Примеры протоколов​

18.5.4.3.1. Протокол TCP-эхо-клиента​

TCP-эхо-клиент, использующий метод AbstractEventLoop.create_connection(), отправляет данные и ждёт, пока соединение не будет закрыто.

python
import asyncio

class EchoClientProtocol(asyncio.Protocol):
    def __init__(self, message, loop):
        self.message = message
        self.loop = loop

    def connection_made(self, transport):
        transport.write(self.message.encode())
        print('Data sent: {!r}'.format(self.message))

    def data_received(self, data):
        print('Data received: {!r}'.format(data.decode()))

    def connection_lost(self, exc):
        print('The server closed the connection')
        print('Stop the event loop')
        self.loop.stop()

loop = asyncio.get_event_loop()
message = 'Hello World!'
coro = loop.create_connection(lambda: EchoClientProtocol(message, loop),
                              '127.0.0.1', 8888)
loop.run_until_complete(coro)
loop.run_forever()
loop.close()

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

Смотрите также

В примере TCP echo клиент, использующий потоки данных используется функция asyncio.open_connection().

18.5.4.3.2. Протокол TCP echo сервера​

TCP echo сервер, использующий метод AbstractEventLoop.create_server(), отправляет обратно полученные данные и закрывает соединение:

python
import asyncio

class EchoServerClientProtocol(asyncio.Protocol):
    def connection_made(self, transport):
        peername = transport.get_extra_info('peername')
        print('Connection from {}'.format(peername))
        self.transport = transport

    def data_received(self, data):
        message = data.decode()
        print('Data received: {!r}'.format(message))

        print('Send: {!r}'.format(message))
        self.transport.write(data)

        print('Close the client socket')
        self.transport.close()

loop = asyncio.get_event_loop()
# Каждое клиентское соединение создаст новый экземпляр протокола
coro = loop.create_server(EchoServerClientProtocol, '127.0.0.1', 8888)
server = loop.run_until_complete(coro)

# Обслуживать запросы до нажатия Ctrl+C
print('Serving on {}'.format(server.sockets[0].getsockname()))
try:
    loop.run_forever()
except KeyboardInterrupt:
    pass

# Закрыть сервер
server.close()
loop.run_until_complete(server.wait_closed())
loop.close()

Transport.close() можно вызвать сразу после WriteTransport.write(), даже если данные ещё не отправлены на сокет: оба метода асинхронны. yield from не нужен, поскольку эти методы транспорта не являются корутинами.

Смотрите также

В примере TCP echo сервер, использующий потоки данных используется функция asyncio.start_server().

18.5.4.3.3. Протокол UDP echo клиента​

UDP echo клиент, использующий метод AbstractEventLoop.create_datagram_endpoint(), отправляет данные и закрывает транспорт после получения ответа:

python
import asyncio

class EchoClientProtocol:
    def __init__(self, message, loop):
        self.message = message
        self.loop = loop
        self.transport = None

    def connection_made(self, transport):
        self.transport = transport
        print('Send:', self.message)
        self.transport.sendto(self.message.encode())

    def datagram_received(self, data, addr):
        print("Received:", data.decode())

        print("Close the socket")
        self.transport.close()

    def error_received(self, exc):
        print('Error received:', exc)

    def connection_lost(self, exc):
        print("Socket closed, stop the event loop")
        loop = asyncio.get_event_loop()
        loop.stop()

loop = asyncio.get_event_loop()
message = "Hello World!"
connect = loop.create_datagram_endpoint(
    lambda: EchoClientProtocol(message, loop),
    remote_addr=('127.0.0.1', 9999))
transport, protocol = loop.run_until_complete(connect)
loop.run_forever()
transport.close()
loop.close()

18.5.4.3.4. Протокол UDP echo сервера​

UDP echo сервер, использующий метод AbstractEventLoop.create_datagram_endpoint(), отправляет обратно полученные данные:

python
import asyncio

class EchoServerProtocol:
    def connection_made(self, transport):
        self.transport = transport

    def datagram_received(self, data, addr):
        message = data.decode()
        print('Received %r from %s' % (message, addr))
        print('Send %r to %s' % (message, addr))
        self.transport.sendto(data, addr)

loop = asyncio.get_event_loop()
print("Starting UDP server")
# Будет создан один экземпляр протокола для обслуживания всех клиентских запросов
listen = loop.create_datagram_endpoint(
    EchoServerProtocol, local_addr=('127.0.0.1', 9999))
transport, protocol = loop.run_until_complete(listen)

try:
    loop.run_forever()
except KeyboardInterrupt:
    pass

transport.close()
loop.close()

18.5.4.3.5. Регистрация открытого сокета для ожидания данных с использованием протокола​

Ожидание получения данных сокетом с помощью метода AbstractEventLoop.create_connection() и протокола, а затем закрытие цикла событий

python
import asyncio
try:
    from socket import socketpair
except ImportError:
    from asyncio.windows_utils import socketpair

# Создать пару соединённых сокетов
rsock, wsock = socketpair()
loop = asyncio.get_event_loop()

class MyProtocol(asyncio.Protocol):
    transport = None

    def connection_made(self, transport):
        self.transport = transport

    def data_received(self, data):
        print("Received:", data.decode())

        # Мы закончили: закрываем транспорт (он вызовет connection_lost())
        self.transport.close()

    def connection_lost(self, exc):
        # Сокет закрыт, остановить цикл событий
        loop.stop()

# Зарегистрировать сокет для ожидания данных
connect_coro = loop.create_connection(MyProtocol, sock=rsock)
transport, protocol = loop.run_until_complete(connect_coro)

# Симулировать приём данных из сети
loop.call_soon(wsock.send, 'abc'.encode())

# Запустить цикл событий
loop.run_forever()

# Мы закончили, закрываем сокеты и цикл событий
rsock.close()
wsock.close()
loop.close()

Смотрите также

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

Пример регистрации открытого сокета для ожидания данных с использованием потоков использует высокоуровневые потоки, созданные функцией open_connection() в корутине.