selector_events.py
python
"""Цикл событий, использующий селектор и связанные классы.
Селектор — это мультиплексор типа "уведомить, когда готов". О подклассе, который
также включает поддержку обработки сигналов, см. подмодуль unix_events.
"""
__all__ = 'BaseSelectorEventLoop',
import collections
import errno
import functools
import itertools
import os
import selectors
import socket
import warnings
import weakref
try:
import ssl
except ImportError: # pragma: no cover
ssl = None
from . import base_events
from . import constants
from . import events
from . import futures
from . import protocols
from . import sslproto
from . import transports
from . import trsock
from .log import logger
_HAS_SENDMSG = hasattr(socket.socket, 'sendmsg')
if _HAS_SENDMSG:
try:
SC_IOV_MAX = os.sysconf('SC_IOV_MAX')
except OSError:
# Откат к send
_HAS_SENDMSG = False
def _test_selector_event(selector, fd, event):
# Проверить, отслеживает ли селектор события типа 'event'
# для файлового дескриптора 'fd'.
try:
key = selector.get_key(fd)
except KeyError:
return False
else:
return bool(key.events & event)
class BaseSelectorEventLoop(base_events.BaseEventLoop):
"""Цикл событий селектора.
Смотрите events.AbstractEventLoop для спецификации API.
"""
def __init__(self, selector=None):
super().__init__()
if selector is None:
selector = selectors.DefaultSelector()
logger.debug('Using selector: %s', selector.__class__.__name__)
self._selector = selector
self._make_self_pipe()
self._transports = weakref.WeakValueDictionary()
def _make_socket_transport(self, sock, protocol, waiter=None, *,
extra=None, server=None, context=None):
self._ensure_fd_no_transport(sock)
return _SelectorSocketTransport(self, sock, protocol, waiter,
extra, server, context=context)
def _make_ssl_transport(
self, rawsock, protocol, sslcontext, waiter=None,
*, server_side=False, server_hostname=None,
extra=None, server=None,
ssl_handshake_timeout=constants.SSL_HANDSHAKE_TIMEOUT,
ssl_shutdown_timeout=constants.SSL_SHUTDOWN_TIMEOUT,
context=None,
):
self._ensure_fd_no_transport(rawsock)
ssl_protocol = sslproto.SSLProtocol(
self, protocol, sslcontext, waiter,
server_side, server_hostname,
ssl_handshake_timeout=ssl_handshake_timeout,
ssl_shutdown_timeout=ssl_shutdown_timeout,
)
_SelectorSocketTransport(self, rawsock, ssl_protocol,
extra=extra, server=server, context=context)
return ssl_protocol._app_transport
def _make_datagram_transport(self, sock, protocol,
address=None, waiter=None, extra=None):
self._ensure_fd_no_transport(sock)
return _SelectorDatagramTransport(self, sock, protocol,
address, waiter, extra)
def close(self):
if self.is_running():
raise RuntimeError("Cannot close a running event loop")
if self.is_closed():
return
self._close_self_pipe()
super().close()
if self._selector is not None:
self._selector.close()
self._selector = None
def _close_self_pipe(self):
self._remove_reader(self._ssock.fileno())
self._ssock.close()
self._ssock = None
self._csock.close()
self._csock = None
self._internal_fds -= 1
def _make_self_pipe(self):
# Само-сокет, на самом деле. :-)
self._ssock, self._csock = socket.socketpair()
self._ssock.setblocking(False)
self._csock.setblocking(False)
self._internal_fds += 1
self._add_reader(self._ssock.fileno(), self._read_from_self)
def _process_self_data(self, data):
pass
def _read_from_self(self):
while True:
try:
data = self._ssock.recv(4096)
if not data:
break
self._process_self_data(data)
except InterruptedError:
continue
except BlockingIOError:
break
def _write_to_self(self):
# Это может быть вызвано из другого потока, возможно после
# _close_self_pipe() была вызвана или даже пока она
# выполняется. Защита от того, что self._csock равен None или закрыт. Когда
# сокет закрыт, send() вызывает OSError (с errno, установленным в
# EBADF, но не будем полагаться на точный код ошибки).
csock = self._csock
if csock is None:
return
try:
csock.send(b'\0')
except OSError:
if self._debug:
logger.debug("Fail to write a null byte into the "
"self-pipe socket",
exc_info=True)
def _start_serving(self, protocol_factory, sock,
sslcontext=None, server=None, backlog=100,
ssl_handshake_timeout=constants.SSL_HANDSHAKE_TIMEOUT,
ssl_shutdown_timeout=constants.SSL_SHUTDOWN_TIMEOUT, context=None):
self._add_reader(sock.fileno(), self._accept_connection,
protocol_factory, sock, sslcontext, server, backlog,
ssl_handshake_timeout, ssl_shutdown_timeout, context)
def _accept_connection(
self, protocol_factory, sock,
sslcontext=None, server=None, backlog=100,
ssl_handshake_timeout=constants.SSL_HANDSHAKE_TIMEOUT,
ssl_shutdown_timeout=constants.SSL_SHUTDOWN_TIMEOUT, context=None):
# Этот метод вызывается только один раз за каждый такт цикла событий, когда
# прослушивающий сокет вызвал событие EVENT_READ. Может быть несколько
# подключений, ожидающих .accept(), поэтому он вызывается в цикле.
# См. https://bugs.python.org/issue27906 для получения дополнительных сведений.
for _ in range(backlog + 1):
try:
conn, addr = sock.accept()
if self._debug:
logger.debug("%r got a new connection from %r: %r",
server, addr, conn)
conn.setblocking(False)
except ConnectionAbortedError:
# Отбрасывать соединения, которые были прерваны до accept().
continue
except (BlockingIOError, InterruptedError):
# Досрочный выход из-за сигнала или
# буфер accept сокета пуст.
return
except OSError as exc:
# Некуда отправить ошибку, поэтому просто логируем её.
if exc.errno in (errno.EMFILE, errno.ENFILE,
errno.ENOBUFS, errno.ENOMEM):
# Некоторые платформы (например, Linux) продолжают сообщать файловый дескриптор как
# готов, поэтому мы временно удаляем обработчик чтения.
# Повторим попытку через некоторое время.
self.call_exception_handler({
'message': 'socket.accept() out of system resource',
'exception': exc,
'socket': trsock.TransportSocket(sock),
})
self._remove_reader(sock.fileno())
self.call_later(constants.ACCEPT_RETRY_DELAY,
self._start_serving,
protocol_factory, sock, sslcontext, server,
backlog, ssl_handshake_timeout,
ssl_shutdown_timeout, context)
else:
raise # Цикл событий перехватит, залогирует и проигнорирует его.
else:
extra = {'peername': addr}
conn_context = context.copy() if context is not None else None
accept = self._accept_connection2(
protocol_factory, conn, extra, sslcontext, server,
ssl_handshake_timeout, ssl_shutdown_timeout, context=conn_context)
self.create_task(accept, context=conn_context)
async def _accept_connection2(
self, protocol_factory, conn, extra,
sslcontext=None, server=None,
ssl_handshake_timeout=constants.SSL_HANDSHAKE_TIMEOUT,
ssl_shutdown_timeout=constants.SSL_SHUTDOWN_TIMEOUT, context=None):
protocol = None
transport = None
try:
protocol = protocol_factory()
waiter = self.create_future()
if sslcontext:
transport = self._make_ssl_transport(
conn, protocol, sslcontext, waiter=waiter,
server_side=True, extra=extra, server=server,
ssl_handshake_timeout=ssl_handshake_timeout,
ssl_shutdown_timeout=ssl_shutdown_timeout,
context=context)
else:
transport = self._make_socket_transport(
conn, protocol, waiter=waiter, extra=extra,
server=server, context=context)
try:
await waiter
except BaseException:
transport.close()
# gh-109534: Когда исключение возникает в объекте SSLProtocol,
# исключение, установленное в этом future, может удерживать объект протокола в памяти
# и вызывать циклическую ссылку.
waiter = None
raise
# Теперь обработка соединения возлагается на протокол.
except (SystemExit, KeyboardInterrupt):
raise
except BaseException as exc:
if self._debug:
context = {
'message':
'Error on transport creation for incoming connection',
'exception': exc,
}
if protocol is not None:
context['protocol'] = protocol
if transport is not None:
context['transport'] = transport
self.call_exception_handler(context)
def _ensure_fd_no_transport(self, fd):
fileno = fd
if not isinstance(fileno, int):
try:
fileno = int(fileno.fileno())
except (AttributeError, TypeError, ValueError):
# Этот код соответствует функции selectors._fileobj_to_fd.
raise ValueError(f"Invalid file object: {fd!r}") from None
transport = self._transports.get(fileno)
if transport and not transport.is_closing():
raise RuntimeError(
f'File descriptor {fd!r} is used by transport '
f'{transport!r}')
def _add_reader(self, fd, callback, *args, context=None):
self._check_closed()
handle = events.Handle(callback, args, self, context=context)
key = self._selector.get_map().get(fd)
if key is None:
self._selector.register(fd, selectors.EVENT_READ,
(handle, None))
else:
mask, (reader, writer) = key.events, key.data
self._selector.modify(fd, mask | selectors.EVENT_READ,
(handle, writer))
if reader is not None:
reader.cancel()
return handle
def _remove_reader(self, fd):
if self.is_closed():
return False
key = self._selector.get_map().get(fd)
if key is None:
return False
mask, (reader, writer) = key.events, key.data
mask &= ~selectors.EVENT_READ
if not mask:
self._selector.unregister(fd)
else:
self._selector.modify(fd, mask, (None, writer))
if reader is not None:
reader.cancel()
return True
else:
return False
def _add_writer(self, fd, callback, *args, context=None):
self._check_closed()
handle = events.Handle(callback, args, self, context=context)
key = self._selector.get_map().get(fd)
if key is None:
self._selector.register(fd, selectors.EVENT_WRITE,
(None, handle))
else:
mask, (reader, writer) = key.events, key.data
self._selector.modify(fd, mask | selectors.EVENT_WRITE,
(reader, handle))
if writer is not None:
writer.cancel()
return handle
def _remove_writer(self, fd):
"""Удалить колбэк записи."""
if self.is_closed():
return False
key = self._selector.get_map().get(fd)
if key is None:
return False
mask, (reader, writer) = key.events, key.data
# Удалить и запись, и соединитель.
mask &= ~selectors.EVENT_WRITE
if not mask:
self._selector.unregister(fd)
else:
self._selector.modify(fd, mask, (reader, None))
if writer is not None:
writer.cancel()
return True
else:
return False
def add_reader(self, fd, callback, *args):
"""Добавить колбэк чтения."""
self._ensure_fd_no_transport(fd)
self._add_reader(fd, callback, *args)
def remove_reader(self, fd):
"""Удалить колбэк чтения."""
self._ensure_fd_no_transport(fd)
return self._remove_reader(fd)
def add_writer(self, fd, callback, *args):
"""Добавить колбэк записи.."""
self._ensure_fd_no_transport(fd)
self._add_writer(fd, callback, *args)
def remove_writer(self, fd):
"""Удалить колбэк записи."""
self._ensure_fd_no_transport(fd)
return self._remove_writer(fd)
async def sock_recv(self, sock, n):
"""Получить данные из сокета.
Возвращаемое значение — объект bytes, представляющий полученные данные.
Максимальное количество данных, получаемых за один раз, задаётся
nbytes.
"""
base_events._check_ssl_socket(sock)
if self._debug and sock.gettimeout() != 0:
raise ValueError("the socket must be non-blocking")
try:
return sock.recv(n)
except (BlockingIOError, InterruptedError):
pass
fut = self.create_future()
fd = sock.fileno()
self._ensure_fd_no_transport(fd)
handle = self._add_reader(fd, self._sock_recv, fut, sock, n)
fut.add_done_callback(
functools.partial(self._sock_read_done, fd, handle=handle))
return await fut
def _sock_read_done(self, fd, fut, handle=None):
if handle is None or not handle.cancelled():
self.remove_reader(fd)
def _sock_recv(self, fut, sock, n):
# _sock_recv() может добавить себя как колбэк ввода-вывода, если операция не может
# быть выполнена немедленно. Не используйте её напрямую, вызывайте sock_recv().
if fut.done():
return
try:
data = sock.recv(n)
except (BlockingIOError, InterruptedError):
return # повторите попытку позже
except (SystemExit, KeyboardInterrupt):
raise
except BaseException as exc:
fut.set_exception(exc)
else:
fut.set_result(data)
async def sock_recv_into(self, sock, buf):
"""Получить данные из сокета.
Полученные данные записываются в *buf* (буфер для записи).
Возвращаемое значение — количество записанных байт.
"""
base_events._check_ssl_socket(sock)
if self._debug and sock.gettimeout() != 0:
raise ValueError("the socket must be non-blocking")
try:
return sock.recv_into(buf)
except (BlockingIOError, InterruptedError):
pass
fut = self.create_future()
fd = sock.fileno()
self._ensure_fd_no_transport(fd)
handle = self._add_reader(fd, self._sock_recv_into, fut, sock, buf)
fut.add_done_callback(
functools.partial(self._sock_read_done, fd, handle=handle))
return await fut
def _sock_recv_into(self, fut, sock, buf):
# _sock_recv_into() может добавить себя в качестве I/O колбэка, если операция
# не может быть выполнена немедленно. Не используйте её напрямую; вызывайте
# sock_recv_into().
if fut.done():
return
try:
nbytes = sock.recv_into(buf)
except (BlockingIOError, InterruptedError):
return # повторите попытку позже
except (SystemExit, KeyboardInterrupt):
raise
except BaseException as exc:
fut.set_exception(exc)
else:
fut.set_result(nbytes)
async def sock_recvfrom(self, sock, bufsize):
"""Получить дейтаграмму из сокета дейтаграмм.
Возвращаемое значение — кортеж (байты, адрес), представляющий
полученную дейтаграмму и адрес, с которого она пришла.
Максимальный объём данных, получаемых за один раз, задаётся nbytes.
"""
base_events._check_ssl_socket(sock)
if self._debug and sock.gettimeout() != 0:
raise ValueError("the socket must be non-blocking")
try:
return sock.recvfrom(bufsize)
except (BlockingIOError, InterruptedError):
pass
fut = self.create_future()
fd = sock.fileno()
self._ensure_fd_no_transport(fd)
handle = self._add_reader(fd, self._sock_recvfrom, fut, sock, bufsize)
fut.add_done_callback(
functools.partial(self._sock_read_done, fd, handle=handle))
return await fut
def _sock_recvfrom(self, fut, sock, bufsize):
# _sock_recvfrom() может добавлять себя в качестве колбэка ввода-вывода, если операция
# не может быть выполнена немедленно. Не используйте её напрямую; вызывайте
# sock_recvfrom().
if fut.done():
return
try:
result = sock.recvfrom(bufsize)
except (BlockingIOError, InterruptedError):
return # повторите попытку позже
except (SystemExit, KeyboardInterrupt):
raise
except BaseException as exc:
fut.set_exception(exc)
else:
fut.set_result(result)
async def sock_recvfrom_into(self, sock, buf, nbytes=0):
"""Получить данные из сокета.
Полученные данные записываются в *buf* (буфер для записи).
Возвращаемое значение — кортеж (количество записанных байт, адрес).
"""
base_events._check_ssl_socket(sock)
if self._debug and sock.gettimeout() != 0:
raise ValueError("the socket must be non-blocking")
if not nbytes:
nbytes = len(buf)
try:
return sock.recvfrom_into(buf, nbytes)
except (BlockingIOError, InterruptedError):
pass
fut = self.create_future()
fd = sock.fileno()
self._ensure_fd_no_transport(fd)
handle = self._add_reader(fd, self._sock_recvfrom_into, fut, sock, buf,
nbytes)
fut.add_done_callback(
functools.partial(self._sock_read_done, fd, handle=handle))
return await fut
def _sock_recvfrom_into(self, fut, sock, buf, bufsize):
# _sock_recv_into() может добавить себя в качестве I/O колбэка, если операция
# не может быть выполнена немедленно. Не используйте её напрямую; вызывайте
# sock_recv_into().
if fut.done():
return
try:
result = sock.recvfrom_into(buf, bufsize)
except (BlockingIOError, InterruptedError):
return # повторите попытку позже
except (SystemExit, KeyboardInterrupt):
raise
except BaseException as exc:
fut.set_exception(exc)
else:
fut.set_result(result)
async def sock_sendall(self, sock, data):
"""Отправить данные в сокет.
Сокет должен быть подключен к удаленному сокету. Этот метод
продолжает отправлять данные из data до тех пор, пока все данные не будут
отправлены или не произойдет ошибка. В случае успеха возвращается None. При ошибке
вызывается исключение, и невозможно определить, сколько данных,
если таковые были, успешно обработано принимающей стороной
соединения.
"""
base_events._check_ssl_socket(sock)
if self._debug and sock.gettimeout() != 0:
raise ValueError("the socket must be non-blocking")
try:
n = sock.send(data)
except (BlockingIOError, InterruptedError):
n = 0
if n == len(data):
# все данные отправлены
return
fut = self.create_future()
fd = sock.fileno()
self._ensure_fd_no_transport(fd)
# использовать трюк со списком в замыкании для хранения изменяемого состояния
handle = self._add_writer(fd, self._sock_sendall, fut, sock,
memoryview(data), [n])
fut.add_done_callback(
functools.partial(self._sock_write_done, fd, handle=handle))
return await fut
def _sock_sendall(self, fut, sock, view, pos):
if fut.done():
# Отмена Future может быть запланирована на предыдущей итерации цикла
return
start = pos[0]
try:
n = sock.send(view[start:])
except (BlockingIOError, InterruptedError):
return
except (SystemExit, KeyboardInterrupt):
raise
except BaseException as exc:
fut.set_exception(exc)
return
start += n
if start == len(view):
fut.set_result(None)
else:
pos[0] = start
async def sock_sendto(self, sock, data, address):
"""Отправить данные в сокет.
Сокет должен быть подключен к удаленному сокету. Этот метод
продолжает отправлять данные из data до тех пор, пока все данные не будут
отправлены или не произойдет ошибка. В случае успеха возвращается None. При ошибке
вызывается исключение, и невозможно определить, сколько данных,
если таковые были, успешно обработано принимающей стороной
соединения.
"""
base_events._check_ssl_socket(sock)
if self._debug and sock.gettimeout() != 0:
raise ValueError("the socket must be non-blocking")
try:
return sock.sendto(data, address)
except (BlockingIOError, InterruptedError):
pass
fut = self.create_future()
fd = sock.fileno()
self._ensure_fd_no_transport(fd)
# использовать трюк со списком в замыкании для хранения изменяемого состояния
handle = self._add_writer(fd, self._sock_sendto, fut, sock, data,
address)
fut.add_done_callback(
functools.partial(self._sock_write_done, fd, handle=handle))
return await fut
def _sock_sendto(self, fut, sock, data, address):
if fut.done():
# Отмена Future может быть запланирована на предыдущей итерации цикла
return
try:
n = sock.sendto(data, 0, address)
except (BlockingIOError, InterruptedError):
return
except (SystemExit, KeyboardInterrupt):
raise
except BaseException as exc:
fut.set_exception(exc)
else:
fut.set_result(n)
async def sock_connect(self, sock, address):
"""Подключиться к удаленному сокету по адресу.
Этот метод является корутиной.
"""
base_events._check_ssl_socket(sock)
if self._debug and sock.gettimeout() != 0:
raise ValueError("the socket must be non-blocking")
if sock.family == socket.AF_INET or (
base_events._HAS_IPv6 and sock.family == socket.AF_INET6):
resolved = await self._ensure_resolved(
address, family=sock.family, type=sock.type, proto=sock.proto,
loop=self,
)
_, _, _, _, address = resolved[0]
fut = self.create_future()
self._sock_connect(fut, sock, address)
try:
return await fut
finally:
# Необходимо для разрыва циклов при возникновении исключения.
fut = None
def _sock_connect(self, fut, sock, address):
fd = sock.fileno()
try:
sock.connect(address)
except (BlockingIOError, InterruptedError):
# Ошибка #23618: Когда C-функция connect() завершается ошибкой с EINTR,
# соединение выполняется в фоне. Мы должны ждать, пока сокет
# не станет доступным для записи, чтобы получить уведомление, когда соединение установится или
# завершится ошибкой.
self._ensure_fd_no_transport(fd)
handle = self._add_writer(
fd, self._sock_connect_cb, fut, sock, address)
fut.add_done_callback(
functools.partial(self._sock_write_done, fd, handle=handle))
except (SystemExit, KeyboardInterrupt):
raise
except BaseException as exc:
fut.set_exception(exc)
else:
fut.set_result(None)
finally:
fut = None
def _sock_write_done(self, fd, fut, handle=None):
if handle is None or not handle.cancelled():
self.remove_writer(fd)
def _sock_connect_cb(self, fut, sock, address):
if fut.done():
return
try:
err = sock.getsockopt(socket.SOL_SOCKET, socket.SO_ERROR)
if err != 0:
# Перейти к любому блоку except ниже.
raise OSError(err, f'Connect call failed {address}')
except (BlockingIOError, InterruptedError):
# сокет все еще зарегистрирован, колбэк будет повторен позже
pass
except (SystemExit, KeyboardInterrupt):
raise
except BaseException as exc:
fut.set_exception(exc)
else:
fut.set_result(None)
finally:
fut = None
async def sock_accept(self, sock):
"""Принять соединение.
Сокет должен быть привязан к адресу и ожидать
соединения. Возвращаемое значение - пара (conn, address), где
conn - новый объект сокета, пригодный для отправки и получения данных по
соединению, а address - адрес, привязанный к сокету на
другом конце соединения.
"""
base_events._check_ssl_socket(sock)
if self._debug and sock.gettimeout() != 0:
raise ValueError("the socket must be non-blocking")
fut = self.create_future()
self._sock_accept(fut, sock)
return await fut
def _sock_accept(self, fut, sock):
fd = sock.fileno()
try:
conn, address = sock.accept()
conn.setblocking(False)
except (BlockingIOError, InterruptedError):
self._ensure_fd_no_transport(fd)
handle = self._add_reader(fd, self._sock_accept, fut, sock)
fut.add_done_callback(
functools.partial(self._sock_read_done, fd, handle=handle))
except (SystemExit, KeyboardInterrupt):
raise
except BaseException as exc:
fut.set_exception(exc)
else:
fut.set_result((conn, address))
async def _sendfile_native(self, transp, file, offset, count):
del self._transports[transp._sock_fd]
resume_reading = transp.is_reading()
transp.pause_reading()
await transp._make_empty_waiter()
try:
return await self.sock_sendfile(transp._sock, file, offset, count,
fallback=False)
finally:
transp._reset_empty_waiter()
if resume_reading:
transp.resume_reading()
self._transports[transp._sock_fd] = transp
def _process_events(self, event_list):
for key, mask in event_list:
fileobj, (reader, writer) = key.fileobj, key.data
if mask & selectors.EVENT_READ and reader is not None:
if reader._cancelled:
self._remove_reader(fileobj)
else:
self._add_callback(reader)
if mask & selectors.EVENT_WRITE and writer is not None:
if writer._cancelled:
self._remove_writer(fileobj)
else:
self._add_callback(writer)
def _stop_serving(self, sock):
self._remove_reader(sock.fileno())
sock.close()
class _SelectorTransport(transports._FlowControlMixin,
transports.Transport):
max_size = 256 * 1024 # Размер буфера, передаваемый в recv().
# Атрибут, используемый в деструкторе: должен быть установлен, даже если конструктор
# не вызывается (см. _SelectorSslTransport, который может начать с выброса
# исключения)
_sock = None
def __init__(self, loop, sock, protocol, extra=None, server=None, context=None):
super().__init__(extra, loop)
self._extra['socket'] = trsock.TransportSocket(sock)
try:
self._extra['sockname'] = sock.getsockname()
except OSError:
self._extra['sockname'] = None
if 'peername' not in self._extra:
try:
self._extra['peername'] = sock.getpeername()
except socket.error:
self._extra['peername'] = None
self._sock = sock
self._sock_fd = sock.fileno()
self._context = context
self._protocol_connected = False
self.set_protocol(protocol)
self._server = server
self._buffer = collections.deque()
self._buffer_size = 0
self._conn_lost = 0 # Устанавливается, когда запланирован вызов connection_lost.
self._closing = False # Устанавливается при вызове close().
self._paused = False # Устанавливается при вызове pause_reading()
if self._server is not None:
self._server._attach(self)
loop._transports[self._sock_fd] = self
def __repr__(self):
info = [self.__class__.__name__]
if self._sock is None:
info.append('closed')
elif self._closing:
info.append('closing')
info.append(f'fd={self._sock_fd}')
# проверка, был ли транспорт закрыт
if self._loop is not None and not self._loop.is_closed():
polling = _test_selector_event(self._loop._selector,
self._sock_fd, selectors.EVENT_READ)
if polling:
info.append('read=polling')
else:
info.append('read=idle')
polling = _test_selector_event(self._loop._selector,
self._sock_fd,
selectors.EVENT_WRITE)
if polling:
state = 'polling'
else:
state = 'idle'
bufsize = self.get_write_buffer_size()
info.append(f'write=<{state}, bufsize={bufsize}>')
return '<{}>'.format(' '.join(info))
def abort(self):
self._force_close(None)
def set_protocol(self, protocol):
self._protocol = protocol
self._protocol_connected = True
def get_protocol(self):
return self._protocol
def is_closing(self):
return self._closing
def is_reading(self):
return not self.is_closing() and not self._paused
def pause_reading(self):
if not self.is_reading():
return
self._paused = True
self._loop._remove_reader(self._sock_fd)
if self._loop.get_debug():
logger.debug("%r pauses reading", self)
def resume_reading(self):
if self._closing or not self._paused:
return
self._paused = False
self._add_reader(self._sock_fd, self._read_ready)
if self._loop.get_debug():
logger.debug("%r resumes reading", self)
def close(self):
if self._closing:
return
self._closing = True
self._loop._remove_reader(self._sock_fd)
if not self._buffer:
self._conn_lost += 1
self._loop._remove_writer(self._sock_fd)
self._call_soon(self._call_connection_lost, None)
def __del__(self, _warn=warnings.warn):
if self._sock is not None:
_warn(f"unclosed transport {self!r}", ResourceWarning, source=self)
self._sock.close()
if self._server is not None:
self._server._detach(self)
def _fatal_error(self, exc, message='Fatal error on transport'):
# Должен вызываться только из обработчика исключений.
if isinstance(exc, OSError):
if self._loop.get_debug():
logger.debug("%r: %s", self, message, exc_info=True)
else:
self._loop.call_exception_handler({
'message': message,
'exception': exc,
'transport': self,
'protocol': self._protocol,
})
self._force_close(exc)
def _force_close(self, exc):
if self._conn_lost:
return
if self._buffer:
self._buffer.clear()
self._buffer_size = 0
self._loop._remove_writer(self._sock_fd)
if not self._closing:
self._closing = True
self._loop._remove_reader(self._sock_fd)
self._conn_lost += 1
self._call_soon(self._call_connection_lost, exc)
def _call_connection_lost(self, exc):
try:
if self._protocol_connected:
self._protocol.connection_lost(exc)
finally:
self._sock.close()
self._sock = None
self._protocol = None
self._loop = None
server = self._server
if server is not None:
server._detach(self)
self._server = None
def get_write_buffer_size(self):
return self._buffer_size
def _add_reader(self, fd, callback, *args):
if not self.is_reading():
return
self._loop._add_reader(fd, callback, *args, context=self._context)
def _add_writer(self, fd, callback, *args):
self._loop._add_writer(fd, callback, *args, context=self._context)
def _call_soon(self, callback, *args):
self._loop.call_soon(callback, *args, context=self._context)
class _SelectorSocketTransport(_SelectorTransport):
_start_tls_compatible = True
_sendfile_compatible = constants._SendfileMode.TRY_NATIVE
def __init__(self, loop, sock, protocol, waiter=None,
extra=None, server=None, context=None):
self._read_ready_cb = None
super().__init__(loop, sock, protocol, extra, server, context)
self._eof = False
self._empty_waiter = None
if _HAS_SENDMSG:
self._write_ready = self._write_sendmsg
else:
self._write_ready = self._write_send
# Отключить алгоритм Нагла — маленькие записи будут
# отправляться без ожидания TCP-подтверждения. Это обычно
# снижает задержку (в некоторых случаях значительно.)
base_events._set_nodelay(self._sock)
self._call_soon(self._protocol.connection_made, self)
# начинать чтение только после вызова connection_made()
self._call_soon(self._add_reader, self._sock_fd, self._read_ready)
if waiter is not None:
# пробуждать ожидающего только после вызова connection_made()
self._call_soon(futures._set_result_unless_cancelled, waiter, None)
def set_protocol(self, protocol):
if isinstance(protocol, protocols.BufferedProtocol):
self._read_ready_cb = self._read_ready__get_buffer
else:
self._read_ready_cb = self._read_ready__data_received
super().set_protocol(protocol)
def _read_ready(self):
self._read_ready_cb()
def _read_ready__get_buffer(self):
if self._conn_lost:
return
try:
buf = self._protocol.get_buffer(-1)
if not len(buf):
raise RuntimeError('get_buffer() returned an empty buffer')
except (SystemExit, KeyboardInterrupt):
raise
except BaseException as exc:
self._fatal_error(
exc, 'Fatal error: protocol.get_buffer() call failed.')
return
try:
nbytes = self._sock.recv_into(buf)
except (BlockingIOError, InterruptedError):
return
except (SystemExit, KeyboardInterrupt):
raise
except BaseException as exc:
self._fatal_error(exc, 'Fatal read error on socket transport')
return
if not nbytes:
self._read_ready__on_eof()
return
try:
self._protocol.buffer_updated(nbytes)
except (SystemExit, KeyboardInterrupt):
raise
except BaseException as exc:
self._fatal_error(
exc, 'Fatal error: protocol.buffer_updated() call failed.')
def _read_ready__data_received(self):
if self._conn_lost:
return
try:
data = self._sock.recv(self.max_size)
except (BlockingIOError, InterruptedError):
return
except (SystemExit, KeyboardInterrupt):
raise
except BaseException as exc:
self._fatal_error(exc, 'Fatal read error on socket transport')
return
if not data:
self._read_ready__on_eof()
return
try:
self._protocol.data_received(data)
except (SystemExit, KeyboardInterrupt):
raise
except BaseException as exc:
self._fatal_error(
exc, 'Fatal error: protocol.data_received() call failed.')
def _read_ready__on_eof(self):
if self._loop.get_debug():
logger.debug("%r received EOF", self)
try:
keep_open = self._protocol.eof_received()
except (SystemExit, KeyboardInterrupt):
raise
except BaseException as exc:
self._fatal_error(
exc, 'Fatal error: protocol.eof_received() call failed.')
return
if keep_open:
# Мы держим соединение открытым, чтобы
# протокол мог ещё записать, но мы всё ещё не можем
# принимать больше, поэтому удаляем колбэк чтения.
self._loop._remove_reader(self._sock_fd)
else:
self.close()
def write(self, data):
if not isinstance(data, (bytes, bytearray, memoryview)):
raise TypeError(f'data argument must be a bytes, bytearray, or memoryview '
f'object, not {type(data).__name__!r}')
if self._eof:
raise RuntimeError('Cannot call write() after write_eof()')
if self._empty_waiter is not None:
raise RuntimeError('unable to write; sendfile is in progress')
if not data:
return
if self._conn_lost:
if self._conn_lost >= constants.LOG_THRESHOLD_FOR_CONNLOST_WRITES:
logger.warning('socket.send() raised exception.')
self._conn_lost += 1
return
if not self._buffer:
# Оптимизация: попробовать отправить сейчас.
try:
n = self._sock.send(data)
except (BlockingIOError, InterruptedError):
pass
except (SystemExit, KeyboardInterrupt):
raise
except BaseException as exc:
self._fatal_error(exc, 'Fatal write error on socket transport')
return
else:
data = memoryview(data)[n:]
if not data:
return
# Не все было записано; зарегистрировать обработчик записи.
self._add_writer(self._sock_fd, self._write_ready)
# Добавить это в буфер.
self._buffer.append(data)
self._buffer_size += len(data)
self._maybe_pause_protocol()
def _get_sendmsg_buffer(self):
return itertools.islice(self._buffer, SC_IOV_MAX)
def _write_sendmsg(self):
assert self._buffer, 'Data should not be empty'
if self._conn_lost:
return
try:
nbytes = self._sock.sendmsg(self._get_sendmsg_buffer())
self._adjust_leftover_buffer(nbytes)
except (BlockingIOError, InterruptedError):
pass
except (SystemExit, KeyboardInterrupt):
raise
except BaseException as exc:
self._loop._remove_writer(self._sock_fd)
self._buffer.clear()
self._buffer_size = 0
self._fatal_error(exc, 'Fatal write error on socket transport')
if self._empty_waiter is not None:
self._empty_waiter.set_exception(exc)
else:
self._maybe_resume_protocol() # Можно добавить в буфер.
if not self._buffer:
self._loop._remove_writer(self._sock_fd)
if self._empty_waiter is not None:
self._empty_waiter.set_result(None)
if self._closing:
self._call_connection_lost(None)
elif self._eof:
self._sock.shutdown(socket.SHUT_WR)
def _adjust_leftover_buffer(self, nbytes: int) -> None:
self._buffer_size -= nbytes
buffer = self._buffer
while nbytes:
b = buffer.popleft()
b_len = len(b)
if b_len <= nbytes:
nbytes -= b_len
else:
buffer.appendleft(b[nbytes:])
break
def _write_send(self):
assert self._buffer, 'Data should not be empty'
if self._conn_lost:
return
try:
buffer = self._buffer.popleft()
n = self._sock.send(buffer)
if n != len(buffer):
# Не все данные были записаны
self._buffer.appendleft(buffer[n:])
self._buffer_size -= n
except (BlockingIOError, InterruptedError):
self._buffer.appendleft(buffer)
return
except (SystemExit, KeyboardInterrupt):
raise
except BaseException as exc:
self._loop._remove_writer(self._sock_fd)
self._buffer.clear()
self._buffer_size = 0
self._fatal_error(exc, 'Fatal write error on socket transport')
if self._empty_waiter is not None:
self._empty_waiter.set_exception(exc)
else:
self._maybe_resume_protocol() # Можно добавить в буфер.
if not self._buffer:
self._loop._remove_writer(self._sock_fd)
if self._empty_waiter is not None:
self._empty_waiter.set_result(None)
if self._closing:
self._call_connection_lost(None)
elif self._eof:
self._sock.shutdown(socket.SHUT_WR)
def write_eof(self):
if self._closing or self._eof:
return
self._eof = True
if not self._buffer:
self._sock.shutdown(socket.SHUT_WR)
def writelines(self, list_of_data):
if self._eof:
raise RuntimeError('Cannot call writelines() after write_eof()')
if self._empty_waiter is not None:
raise RuntimeError('unable to writelines; sendfile is in progress')
if not list_of_data:
return
if self._conn_lost:
if self._conn_lost >= constants.LOG_THRESHOLD_FOR_CONNLOST_WRITES:
logger.warning('socket.send() raised exception.')
self._conn_lost += 1
return
for data in list_of_data:
self._buffer.append(memoryview(data))
self._buffer_size += len(data)
self._write_ready()
# Если весь буфер не может быть записан, зарегистрируйте обработчик записи
if self._buffer:
self._add_writer(self._sock_fd, self._write_ready)
self._maybe_pause_protocol()
def can_write_eof(self):
return True
def _call_connection_lost(self, exc):
try:
super()._call_connection_lost(exc)
finally:
self._write_ready = None
if self._empty_waiter is not None:
self._empty_waiter.set_exception(
ConnectionError("Connection is closed by peer"))
def _make_empty_waiter(self):
if self._empty_waiter is not None:
raise RuntimeError("Empty waiter is already set")
self._empty_waiter = self._loop.create_future()
if not self._buffer:
self._empty_waiter.set_result(None)
return self._empty_waiter
def _reset_empty_waiter(self):
self._empty_waiter = None
def close(self):
self._read_ready_cb = None
super().close()
class _SelectorDatagramTransport(_SelectorTransport, transports.DatagramTransport):
_header_size = 8
def __init__(self, loop, sock, protocol, address=None,
waiter=None, extra=None):
super().__init__(loop, sock, protocol, extra)
self._address = address
self._buffer_size = 0
self._call_soon(self._protocol.connection_made, self)
# начинать чтение только после вызова connection_made()
self._call_soon(self._add_reader, self._sock_fd, self._read_ready)
if waiter is not None:
# пробуждать ожидающего только после вызова connection_made()
self._call_soon(futures._set_result_unless_cancelled, waiter, None)
def get_write_buffer_size(self):
return self._buffer_size
def _read_ready(self):
if self._conn_lost:
return
try:
data, addr = self._sock.recvfrom(self.max_size)
except (BlockingIOError, InterruptedError):
pass
except OSError as exc:
self._protocol.error_received(exc)
except (SystemExit, KeyboardInterrupt):
raise
except BaseException as exc:
self._fatal_error(exc, 'Fatal read error on datagram transport')
else:
self._protocol.datagram_received(data, addr)
def sendto(self, data, addr=None):
if not isinstance(data, (bytes, bytearray, memoryview)):
raise TypeError(f'data argument must be a bytes-like object, '
f'not {type(data).__name__!r}')
if self._address:
if addr not in (None, self._address):
raise ValueError(
f'Invalid address: must be None or {self._address}')
addr = self._address
if self._conn_lost and self._address:
if self._conn_lost >= constants.LOG_THRESHOLD_FOR_CONNLOST_WRITES:
logger.warning('socket.send() raised exception.')
self._conn_lost += 1
return
if not self._buffer:
# Сначала попытаться отправить немедленно.
try:
if self._extra['peername']:
self._sock.send(data)
else:
self._sock.sendto(data, addr)
return
except (BlockingIOError, InterruptedError):
self._add_writer(self._sock_fd, self._sendto_ready)
except OSError as exc:
self._protocol.error_received(exc)
return
except (SystemExit, KeyboardInterrupt):
raise
except BaseException as exc:
self._fatal_error(
exc, 'Fatal write error on datagram transport')
return
# Убедитесь, что буферизуемые данные неизменяемы.
self._buffer.append((bytes(data), addr))
self._buffer_size += len(data) + self._header_size
self._maybe_pause_protocol()
def _sendto_ready(self):
while self._buffer:
data, addr = self._buffer.popleft()
self._buffer_size -= len(data) + self._header_size
try:
if self._extra['peername']:
self._sock.send(data)
else:
self._sock.sendto(data, addr)
except (BlockingIOError, InterruptedError):
self._buffer.appendleft((data, addr)) # Повторить попытку позже.
self._buffer_size += len(data) + self._header_size
break
except OSError as exc:
self._protocol.error_received(exc)
return
except (SystemExit, KeyboardInterrupt):
raise
except BaseException as exc:
self._fatal_error(
exc, 'Fatal write error on datagram transport')
return
self._maybe_resume_protocol() # Можно добавить в буфер.
if not self._buffer:
self._loop._remove_writer(self._sock_fd)
if self._closing:
self._call_connection_lost(None)