Транспорты и протоколы

Иерархия транспортов

class asyncio.BaseTransport

Базовый класс для всех транспортов. Содержит методы, общие для всех транспортов asyncio.

class asyncio.WriteTransport(BaseTransport)

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

Экземпляры класса WriteTransport возвращаются методом цикла событий loop.connect_write_pipe(), а также используются методами, связанными с подпроцессами, например loop.subprocess_exec().

class asyncio.ReadTransport(BaseTransport)

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

Экземпляры класса ReadTransport возвращаются методом цикла событий loop.connect_read_pipe(), а также используются методами, связанными с подпроцессами, например loop.subprocess_exec().

class asyncio.Transport(WriteTransport, ReadTransport)

Интерфейс, представляющий двунаправленный транспорт, например TCP-соединение.

Пользователь не создаёт транспорт напрямую: он вызывает вспомогательную функцию, передавая ей фабрику протоколов и другие сведения, необходимые для создания транспорта и протокола.

Экземпляры класса Transport возвращаются методами цикла событий или используются ими, например loop.create_connection(), loop.create_unix_connection(), loop.create_server(), loop.sendfile() и т. д.

class asyncio.DatagramTransport(BaseTransport)

Транспорт для дейтаграммных соединений (UDP).

Экземпляры класса DatagramTransport возвращаются методом цикла событий loop.create_datagram_endpoint().

class asyncio.SubprocessTransport(BaseTransport)

Абстракция, представляющая соединение между родительским процессом и его дочерним процессом ОС.

Экземпляры класса SubprocessTransport возвращаются методами цикла событий loop.subprocess_shell() и loop.subprocess_exec().

Базовый транспорт

BaseTransport.close()

Закрыть транспорт.

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

BaseTransport.is_closing()

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

BaseTransport.get_extra_info(name, default=None)

Возвращает сведения о транспорте или используемых им базовых ресурсах.

name — строка, задающая тип специфичных для транспорта сведений, которые нужно получить.

default — значение, возвращаемое, если сведения недоступны либо транспорт не поддерживает их получение в текущей сторонней реализации цикла событий или на текущей платформе.

Например, следующий код пытается получить базовый объект сокета транспорта:

sock = transport.get_extra_info('socket')
if sock is not None:
    print(sock.getsockopt(...))

Категории сведений, которые можно запросить у некоторых транспортов:

• сокет:

  • 'peername': удалённый адрес, к которому подключён сокет; результат вызова socket.socket.getpeername() (None в случае ошибки)
  • 'socket': экземпляр socket.socket
  • 'sockname': собственный адрес сокета; результат вызова socket.socket.getsockname()

• сокет 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': объект канала

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

  • 'subprocess': экземпляр subprocess.Popen

BaseTransport.set_protocol(protocol)

Установить новый протокол.

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

BaseTransport.get_protocol()

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

Транспорты только для чтения

ReadTransport.is_reading()

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

Добавлено в версии 3.7.

ReadTransport.pause_reading()

Приостановить приём данных транспортом. Метод протокола protocol.data_received() не будет получать данные, пока не будет вызван resume_reading().

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

ReadTransport.resume_reading()

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

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

Транспорты только для записи

WriteTransport.abort()

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

WriteTransport.can_write_eof()

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

WriteTransport.get_write_buffer_size()

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

WriteTransport.get_write_buffer_limits()

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

Чтобы задать пороговые уровни, используйте set_write_buffer_limits().

Добавлено в версии 3.4.2.

WriteTransport.set_write_buffer_limits(high=None, low=None)

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

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

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

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

Чтобы получить пороговые уровни, используйте get_write_buffer_limits().

WriteTransport.write(data)

Записать в транспорт байты data.

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

WriteTransport.writelines(list_of_data)

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

WriteTransport.write_eof()

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

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

Дейтаграммные транспорты

DatagramTransport.sendto(data, addr=None)

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

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

Изменено в версии 3.13: Этот метод можно вызывать с пустым объектом bytes для отправки дейтаграммы нулевой длины. Также обновлён расчёт размера буфера, используемый для управления потоком, чтобы учитывать заголовок дейтаграммы.

DatagramTransport.abort()

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

Транспорты подпроцессов

SubprocessTransport.get_pid()

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

SubprocessTransport.get_pipe_transport(fd)

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

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

SubprocessTransport.get_returncode()

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

SubprocessTransport.kill()

Принудительно завершить подпроцесс.

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

См. также subprocess.Popen.kill().

SubprocessTransport.send_signal(signal)

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

SubprocessTransport.terminate()

Остановить подпроцесс.

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

См. также subprocess.Popen.terminate().

SubprocessTransport.close()

Завершить подпроцесс, вызвав метод kill().

Если подпроцесс ещё не завершился, закрыть транспорты каналов stdin, stdout и stderr.

Базовые протоколы

class asyncio.BaseProtocol

Базовый протокол с методами, общими для всех протоколов.

class asyncio.Protocol(BaseProtocol)

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

class asyncio.BufferedProtocol(BaseProtocol)

Базовый класс для реализации потоковых протоколов с ручным управлением буфером приёма.

class asyncio.DatagramProtocol(BaseProtocol)

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

class asyncio.SubprocessProtocol(BaseProtocol)

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

Базовый протокол

Все протоколы asyncio могут реализовывать обратные вызовы базового протокола.

Обратные вызовы соединения

Обратные вызовы соединения вызываются для всех протоколов ровно один раз при каждом успешном соединении. Все остальные обратные вызовы протокола могут вызываться только между этими двумя методами.

BaseProtocol.connection_made(transport)

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

Аргумент transport — это транспорт, представляющий соединение. Протокол отвечает за сохранение ссылки на свой транспорт.

BaseProtocol.connection_lost(exc)

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

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

Обратные вызовы управления потоком

Транспорты могут вызывать обратные вызовы управления потоком, чтобы приостановить или возобновить запись, выполняемую протоколом.

Подробнее см. документацию метода set_write_buffer_limits().

BaseProtocol.pause_writing()

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

BaseProtocol.resume_writing()

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

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

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

Потоковые протоколы

Методы цикла событий, такие как loop.create_server(), loop.create_unix_server(), loop.create_connection(), loop.create_unix_connection(), loop.connect_accepted_socket(), loop.connect_read_pipe() и loop.connect_write_pipe(), принимают фабрики, возвращающие потоковые протоколы.

Protocol.data_received(data)

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

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

Этот метод может вызываться произвольное количество раз, пока соединение открыто.

Однако protocol.eof_received() вызывается не более одного раза. После вызова eof_received() метод data_received() больше не вызывается.

Protocol.eof_received()

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

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

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

Конечный автомат:

start -> connection_made
    [-> data_received]*
    [-> eof_received]?
-> connection_lost -> end

Буферизованные потоковые протоколы

Буферизованные протоколы можно использовать с любым методом цикла событий, поддерживающим потоковые протоколы.

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

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

BufferedProtocol.get_buffer(sizehint)

Вызывается для выделения нового буфера приёма.

sizehint — рекомендуемый минимальный размер возвращаемого буфера. Допускается возвращать буферы меньшего или большего размера, чем предлагает значение sizehint. Если оно равно -1, размер буфера может быть произвольным. Возвращать буфер нулевого размера нельзя.

Метод get_buffer() должен возвращать объект, реализующий буферный протокол.

BufferedProtocol.buffer_updated(nbytes)

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

nbytes — общее количество байтов, записанных в буфер.

BufferedProtocol.eof_received()

См. документацию метода protocol.eof_received().

get_buffer() может вызываться произвольное количество раз в течение соединения. Однако protocol.eof_received() вызывается не более одного раза; если он был вызван, get_buffer() и buffer_updated() больше не вызываются.

Конечный автомат:

start -> connection_made
    [-> get_buffer
        [-> buffer_updated]?
    ]*
    [-> eof_received]?
-> connection_lost -> end

Протоколы дейтаграмм

Экземпляры протоколов дейтаграмм следует создавать с помощью фабрик протоколов, передаваемых методу loop.create_datagram_endpoint().

DatagramProtocol.datagram_received(data, addr)

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

DatagramProtocol.error_received(exc)

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

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

Протоколы подпроцессов

Экземпляры протоколов подпроцессов следует создавать с помощью фабрик протоколов, передаваемых методам loop.subprocess_exec() и loop.subprocess_shell().

SubprocessProtocol.pipe_data_received(fd, data)

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

fd — целочисленный файловый дескриптор канала.

data — непустой объект bytes, содержащий полученные данные.

SubprocessProtocol.pipe_connection_lost(fd, exc)

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

fd — целочисленный файловый дескриптор закрытого канала.

SubprocessProtocol.process_exited()

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

Этот метод может быть вызван до методов pipe_data_received() и pipe_connection_lost().

Эхо-сервер TCP

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

import asyncio


class EchoServerProtocol(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()


async def main():
    # Get a reference to the event loop as we plan to use
    # low-level APIs.
    loop = asyncio.get_running_loop()

    server = await loop.create_server(
        EchoServerProtocol,
        '127.0.0.1', 8888)

    async with server:
        await server.serve_forever()


asyncio.run(main())

Эхо-клиент TCP

Эхо-клиент TCP с использованием метода loop.create_connection() отправляет данные и ожидает закрытия соединения:

import asyncio


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

    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')
        self.on_con_lost.set_result(True)


async def main():
    # Get a reference to the event loop as we plan to use
    # low-level APIs.
    loop = asyncio.get_running_loop()

    on_con_lost = loop.create_future()
    message = 'Hello World!'

    transport, protocol = await loop.create_connection(
        lambda: EchoClientProtocol(message, on_con_lost),
        '127.0.0.1', 8888)

    # Wait until the protocol signals that the connection
    # is lost and close the transport.
    try:
        await on_con_lost
    finally:
        transport.close()


asyncio.run(main())

Эхо-сервер UDP

Эхо-сервер UDP с использованием метода loop.create_datagram_endpoint() отправляет полученные данные обратно:

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)


async def main():
    print("Starting UDP server")

    # Get a reference to the event loop as we plan to use
    # low-level APIs.
    loop = asyncio.get_running_loop()

    # One protocol instance will be created to serve all
    # client requests.
    transport, protocol = await loop.create_datagram_endpoint(
        EchoServerProtocol,
        local_addr=('127.0.0.1', 9999))

    try:
        await asyncio.sleep(3600)  # Serve for 1 hour.
    finally:
        transport.close()


asyncio.run(main())

Эхо-клиент UDP

Эхо-клиент UDP с использованием метода loop.create_datagram_endpoint() отправляет данные и закрывает транспорт при получении ответа:

import asyncio


class EchoClientProtocol:
    def __init__(self, message, on_con_lost):
        self.message = message
        self.on_con_lost = on_con_lost
        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("Connection closed")
        self.on_con_lost.set_result(True)


async def main():
    # Get a reference to the event loop as we plan to use
    # low-level APIs.
    loop = asyncio.get_running_loop()

    on_con_lost = loop.create_future()
    message = "Hello World!"

    transport, protocol = await loop.create_datagram_endpoint(
        lambda: EchoClientProtocol(message, on_con_lost),
        remote_addr=('127.0.0.1', 9999))

    try:
        await on_con_lost
    finally:
        transport.close()


asyncio.run(main())

Подключение существующих сокетов

Дождитесь получения данных сокетом с помощью метода loop.create_connection() и протокола:

import asyncio
import socket


class MyProtocol(asyncio.Protocol):

    def __init__(self, on_con_lost):
        self.transport = None
        self.on_con_lost = on_con_lost

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

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

        # We are done: close the transport;
        # connection_lost() will be called automatically.
        self.transport.close()

    def connection_lost(self, exc):
        # The socket has been closed
        self.on_con_lost.set_result(True)


async def main():
    # Get a reference to the event loop as we plan to use
    # low-level APIs.
    loop = asyncio.get_running_loop()
    on_con_lost = loop.create_future()

    # Create a pair of connected sockets
    rsock, wsock = socket.socketpair()

    # Register the socket to wait for data.
    transport, protocol = await loop.create_connection(
        lambda: MyProtocol(on_con_lost), sock=rsock)

    # Simulate the reception of data from the network.
    loop.call_soon(wsock.send, 'abc'.encode())

    try:
        await protocol.on_con_lost
    finally:
        transport.close()
        wsock.close()

asyncio.run(main())

loop.subprocess_exec() и SubprocessProtocol

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

Подпроцесс создаётся методом loop.subprocess_exec():

import asyncio
import sys

class DateProtocol(asyncio.SubprocessProtocol):
    def __init__(self, exit_future):
        self.exit_future = exit_future
        self.output = bytearray()
        self.pipe_closed = False
        self.exited = False

    def pipe_connection_lost(self, fd, exc):
        self.pipe_closed = True
        self.check_for_exit()

    def pipe_data_received(self, fd, data):
        self.output.extend(data)

    def process_exited(self):
        self.exited = True
        # process_exited() method can be called before
        # pipe_connection_lost() method: wait until both methods are
        # called.
        self.check_for_exit()

    def check_for_exit(self):
        if self.pipe_closed and self.exited:
            self.exit_future.set_result(True)

async def get_date():
    # Get a reference to the event loop as we plan to use
    # low-level APIs.
    loop = asyncio.get_running_loop()

    code = 'import datetime; print(datetime.datetime.now())'
    exit_future = asyncio.Future(loop=loop)

    # Create the subprocess controlled by DateProtocol;
    # redirect the standard output into a pipe.
    transport, protocol = await loop.subprocess_exec(
        lambda: DateProtocol(exit_future),
        sys.executable, '-c', code,
        stdin=None, stderr=None)

    # Wait for the subprocess exit using the process_exited()
    # method of the protocol.
    await exit_future

    # Close the stdout pipe.
    transport.close()

    # Read the output which was collected by the
    # pipe_data_received() method of the protocol.
    data = bytes(protocol.output)
    return data.decode('ascii').rstrip()

date = asyncio.run(get_date())
print(f"Current date: {date}")

См. также тот же пример, написанный с использованием высокоуровневых API.