base_events.py
python
"""Базовая реализация цикла событий.
Цикл событий может быть разделён на мультиплексор (часть, отвечающая за уведомление о событиях ввода-вывода) и собственно цикл событий, который оборачивает мультиплексор функциональностью для планирования колбэков, немедленно или в заданное время в будущем.
Всякий раз, когда публичный API принимает колбэк, последующие позиционные аргументы будут переданы колбэку при его вызове. Это позволяет избежать распространения тривиальных лямбда-функций, реализующих замыкания. Именованные аргументы для колбэка не поддерживаются; это осознанное проектное решение, оставляющее возможность для ключевых аргументов изменять смысл самого вызова API.
"""
import collections
import contextvars
import collections.abc
import concurrent.futures
import errno
import heapq
import itertools
import math
import os
import socket
import stat
import subprocess
import sys
import threading
import time
import traceback
import warnings
import weakref
import inspect
try:
import ssl
except ImportError: # pragma: no cover
ssl = None
from . import constants
from . import coroutines
from . import events
from . import exceptions
from . import futures
from . import protocols
from . import sslproto
from . import staggered
from . import tasks
from . import timeouts
from . import transports
from . import trsock
from .log import logger
__all__ = 'BaseEventLoop','Server',
# Минимальное количество запланированных дескрипторов таймера перед очисткой
# отменённых дескрипторов.
_MIN_SCHEDULED_TIMER_HANDLES = 100
# Минимальная доля запланированных дескрипторов таймера, которые отменены
# перед выполнением очистки отменённых дескрипторов.
_MIN_CANCELLED_TIMER_HANDLES_FRACTION = 0.5
_HAS_IPv6 = hasattr(socket, 'AF_INET6')
# Максимальное время ожидания, передаваемое в select, чтобы избежать ограничений ОС
MAXIMUM_SELECT_TIMEOUT = 24 * 3600
def _format_handle(handle):
cb = handle._callback
if isinstance(getattr(cb, '__self__', None), tasks.Task):
# форматировать задачу
return repr(cb.__self__)
else:
return str(handle)
def _format_pipe(fd):
if fd == subprocess.PIPE:
return '<pipe>'
elif fd == subprocess.STDOUT:
return '<stdout>'
else:
return repr(fd)
def _set_reuseport(sock):
if not hasattr(socket, 'SO_REUSEPORT'):
raise ValueError('reuse_port not supported by socket module')
else:
try:
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEPORT, 1)
except OSError:
raise ValueError('reuse_port not supported by socket module, '
'SO_REUSEPORT defined but not implemented.')
def _ipaddr_info(host, port, family, type, proto, flowinfo=0, scopeid=0):
# Попытаться пропустить getaddrinfo, если "host" уже является IP-адресом. Пользователи могли
# обработать разрешение имён в своём коде и передать разрешённые IP-адреса.
if not hasattr(socket, 'inet_pton'):
return
if proto not in {0, socket.IPPROTO_TCP, socket.IPPROTO_UDP} or \
host is None:
return None
if type == socket.SOCK_STREAM:
proto = socket.IPPROTO_TCP
elif type == socket.SOCK_DGRAM:
proto = socket.IPPROTO_UDP
else:
return None
if port is None:
port = 0
elif isinstance(port, bytes) and port == b'':
port = 0
elif isinstance(port, str) and port == '':
port = 0
else:
# Если порт является именем службы, например "http", не пропускайте getaddrinfo.
try:
port = int(port)
except (TypeError, ValueError):
return None
if family == socket.AF_UNSPEC:
afs = [socket.AF_INET]
if _HAS_IPv6:
afs.append(socket.AF_INET6)
else:
afs = [family]
if isinstance(host, bytes):
host = host.decode('idna')
if '%' in host:
# inet_pton в Linux не принимает индекс зоны IPv6 после хоста,
# например, '::1%lo0'.
return None
for af in afs:
try:
socket.inet_pton(af, host)
# Хост уже разрешен.
if _HAS_IPv6 and af == socket.AF_INET6:
return af, type, proto, '', (host, port, flowinfo, scopeid)
else:
return af, type, proto, '', (host, port)
except OSError:
pass
# "host" не является IP-адресом.
return None
def _interleave_addrinfos(addrinfos, first_address_family_count=1):
"""Перемежать список кортежей addrinfo по семейству."""
# Сгруппировать адреса по семейству.
addrinfos_by_family = collections.OrderedDict()
for addr in addrinfos:
family = addr[0]
if family not in addrinfos_by_family:
addrinfos_by_family[family] = []
addrinfos_by_family[family].append(addr)
addrinfos_lists = list(addrinfos_by_family.values())
reordered = []
if first_address_family_count > 1:
reordered.extend(addrinfos_lists[0][:first_address_family_count - 1])
del addrinfos_lists[0][:first_address_family_count - 1]
reordered.extend(
a for a in itertools.chain.from_iterable(
itertools.zip_longest(*addrinfos_lists)
) if a is not None)
return reordered
def _run_until_complete_cb(fut):
if not fut.cancelled():
exc = fut.exception()
if isinstance(exc, (SystemExit, KeyboardInterrupt)):
# Issue #22429: run_forever() уже завершена, нет необходимости
# останавливать её.
return
futures._get_loop(fut).stop()
if hasattr(socket, 'TCP_NODELAY'):
def _set_nodelay(sock):
if (sock.family in {socket.AF_INET, socket.AF_INET6} and
sock.type == socket.SOCK_STREAM and
sock.proto == socket.IPPROTO_TCP):
sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
else:
def _set_nodelay(sock):
pass
def _check_ssl_socket(sock):
if ssl is not None and isinstance(sock, ssl.SSLSocket):
raise TypeError("Socket cannot be of type SSLSocket")
class _SendfileFallbackProtocol(protocols.Protocol):
def __init__(self, transp):
if not isinstance(transp, transports._FlowControlMixin):
raise TypeError("transport should be _FlowControlMixin instance")
self._transport = transp
self._proto = transp.get_protocol()
self._should_resume_reading = transp.is_reading()
self._should_resume_writing = transp._protocol_paused
transp.pause_reading()
transp.set_protocol(self)
if self._should_resume_writing:
self._write_ready_fut = self._transport._loop.create_future()
else:
self._write_ready_fut = None
async def drain(self):
if self._transport.is_closing():
raise ConnectionError("Connection closed by peer")
fut = self._write_ready_fut
if fut is None:
return
await fut
def connection_made(self, transport):
raise RuntimeError("Invalid state: "
"connection should have been established already.")
def connection_lost(self, exc):
if self._write_ready_fut is not None:
# Никогда не происходит, если пир отключается после отправки всего содержимого
# Таким образом, разрыв соединения всегда является исключением с точки зрения пользователя
if exc is None:
self._write_ready_fut.set_exception(
ConnectionError("Connection is closed by peer"))
else:
self._write_ready_fut.set_exception(exc)
self._proto.connection_lost(exc)
def pause_writing(self):
if self._write_ready_fut is not None:
return
self._write_ready_fut = self._transport._loop.create_future()
def resume_writing(self):
if self._write_ready_fut is None:
return
self._write_ready_fut.set_result(False)
self._write_ready_fut = None
def data_received(self, data):
raise RuntimeError("Invalid state: reading should be paused")
def eof_received(self):
raise RuntimeError("Invalid state: reading should be paused")
async def restore(self):
self._transport.set_protocol(self._proto)
if self._should_resume_reading:
self._transport.resume_reading()
if self._write_ready_fut is not None:
# Отменить future.
# В основном это не имеет эффекта, так как протокол переключается обратно,
# никакой код больше не должен ждать его.
self._write_ready_fut.cancel()
if self._should_resume_writing:
self._proto.resume_writing()
class Server(events.AbstractServer):
def __init__(self, loop, sockets, protocol_factory, ssl_context, backlog,
ssl_handshake_timeout, ssl_shutdown_timeout=None):
self._loop = loop
self._sockets = sockets
# Слабые ссылки, чтобы не нарушить способность Transport
# обнаруживать заброшенные транспорты
self._clients = weakref.WeakSet()
self._waiters = []
self._protocol_factory = protocol_factory
self._backlog = backlog
self._ssl_context = ssl_context
self._ssl_handshake_timeout = ssl_handshake_timeout
self._ssl_shutdown_timeout = ssl_shutdown_timeout
self._serving = False
self._serving_forever_fut = None
self._context = contextvars.copy_context()
def __repr__(self):
return f'<{self.__class__.__name__} sockets={self.sockets!r}>'
def _attach(self, transport):
assert self._sockets is not None
self._clients.add(transport)
def _detach(self, transport):
self._clients.discard(transport)
if len(self._clients) == 0 and self._sockets is None:
self._wakeup()
def _wakeup(self):
waiters = self._waiters
self._waiters = None
for waiter in waiters:
if not waiter.done():
waiter.set_result(None)
def _start_serving(self):
if self._serving:
return
self._serving = True
for sock in self._sockets:
sock.listen(self._backlog)
self._loop._start_serving(
self._protocol_factory, sock, self._ssl_context,
self, self._backlog, self._ssl_handshake_timeout,
self._ssl_shutdown_timeout, context=self._context)
def get_loop(self):
return self._loop
def is_serving(self):
return self._serving
@property
def sockets(self):
if self._sockets is None:
return ()
return tuple(trsock.TransportSocket(s) for s in self._sockets)
def close(self):
sockets = self._sockets
if sockets is None:
return
self._sockets = None
for sock in sockets:
self._loop._stop_serving(sock)
self._serving = False
if (self._serving_forever_fut is not None and
not self._serving_forever_fut.done()):
self._serving_forever_fut.cancel()
self._serving_forever_fut = None
if len(self._clients) == 0:
self._wakeup()
def close_clients(self):
for transport in self._clients.copy():
transport.close()
def abort_clients(self):
for transport in self._clients.copy():
transport.abort()
async def start_serving(self):
self._start_serving()
# Пропустить одну итерацию цикла, чтобы все 'loop.add_reader'
# прошли.
await tasks.sleep(0)
async def serve_forever(self):
if self._serving_forever_fut is not None:
raise RuntimeError(
f'server {self!r} is already being awaited on serve_forever()')
if self._sockets is None:
raise RuntimeError(f'server {self!r} is closed')
self._start_serving()
self._serving_forever_fut = self._loop.create_future()
try:
await self._serving_forever_fut
except exceptions.CancelledError:
try:
self.close()
self.close_clients()
await self.wait_closed()
finally:
raise
finally:
self._serving_forever_fut = None
async def wait_closed(self):
"""Ожидает, пока сервер будет закрыт и все соединения будут разорваны.
- Если сервер не закрыт, ждать.
- Если он закрыт, но есть активные соединения, ждать.
Любой, кто ожидает здесь, будет разблокирован, когда оба условия
(сервер закрыт и все соединения разорваны)
станут истинными, в любом порядке.
Историческая заметка: В версиях 3.11 и ранее это было сломано — возвращалось
немедленно, если сервер уже закрыт, даже если еще были активные соединения. Попытка исправления в 3.12.0
все еще была сломана — возвращалось немедленно, если сервер все еще
открыт и не было активных соединений. Надеемся, в 3.12.1
мы сделали это правильно.
"""
# Ожидающие разблокируются вызовом self._wakeup(), который вызывается
# из двух мест: self.close() и self._detach(), но только
# когда оба условия стали истинными. Чтобы сообщить, что это
# произошло, self._wakeup() устанавливает self._waiters в None.
if self._waiters is None:
return
waiter = self._loop.create_future()
self._waiters.append(waiter)
await waiter
class BaseEventLoop(events.AbstractEventLoop):
def __init__(self):
self._timer_cancelled_count = 0
self._closed = False
self._stopping = False
self._ready = collections.deque()
self._scheduled = []
self._default_executor = None
self._internal_fds = 0
# Идентификатор потока, выполняющего цикл событий, или None, если
# цикл событий не запущен
self._thread_id = None
self._clock_resolution = time.get_clock_info('monotonic').resolution
self._exception_handler = None
self.set_debug(coroutines._is_debug_mode())
# Сохранённое состояние хуков асинхронных генераторов.
self._old_agen_hooks = None
# В режиме отладки, если выполнение колбэка или шага задачи
# если время выполнения в секундах превышает эту длительность, медленный колбэк/задача регистрируется.
self.slow_callback_duration = 0.1
self._current_handle = None
self._task_factory = None
self._coroutine_origin_tracking_enabled = False
self._coroutine_origin_tracking_saved_depth = None
# Слабое множество всех асинхронных генераторов, которые
# итерируются циклом.
self._asyncgens = weakref.WeakSet()
# Устанавливается в True при вызове `loop.shutdown_asyncgens`.
self._asyncgens_shutdown_called = False
# Устанавливается в True при вызове `loop.shutdown_default_executor`.
self._executor_shutdown_called = False
def __repr__(self):
return (
f'<{self.__class__.__name__} running={self.is_running()} '
f'closed={self.is_closed()} debug={self.get_debug()}>'
)
def create_future(self):
"""Создать объект Future, привязанный к циклу."""
return futures.Future(loop=self)
def create_task(self, coro, **kwargs):
"""Запланировать или начать выполнение объекта корутины.
Вернуть объект задачи.
"""
self._check_closed()
if self._task_factory is not None:
return self._task_factory(self, coro, **kwargs)
task = tasks.Task(coro, loop=self, **kwargs)
if task._source_traceback:
del task._source_traceback[-1]
try:
return task
finally:
# gh-128552: предотвратить циклическую ссылку
# task.exception().__traceback__->BaseEventLoop.create_task->task
del task
def set_task_factory(self, factory):
"""Устанавливает фабрику задач, которая будет использоваться loop.create_task().
Если factory равен None, будет установлена фабрика задач по умолчанию.
Если factory является вызываемым объектом, его сигнатура должна соответствовать
'(loop, coro, **kwargs)', где 'loop' — ссылка на активный
цикл событий, 'coro' — объект корутины, а **kwargs
будут произвольными ключевыми аргументами, которые должны быть переданы
Task. Вызываемый объект должен возвращать Task.
"""
if factory is not None and not callable(factory):
raise TypeError('task factory must be a callable or None')
self._task_factory = factory
def get_task_factory(self):
"""Возвращает фабрику задач или None, если используется стандартная."""
return self._task_factory
def _make_socket_transport(self, sock, protocol, waiter=None, *,
extra=None, server=None):
"""Создать сокет-транспорт."""
raise NotImplementedError
def _make_ssl_transport(
self, rawsock, protocol, sslcontext, waiter=None,
*, server_side=False, server_hostname=None,
extra=None, server=None,
ssl_handshake_timeout=None,
ssl_shutdown_timeout=None,
call_connection_made=True,
context=None):
"""Создать SSL-транспорт."""
raise NotImplementedError
def _make_datagram_transport(self, sock, protocol,
address=None, waiter=None, extra=None):
"""Создать датаграммный транспорт."""
raise NotImplementedError
def _make_read_pipe_transport(self, pipe, protocol, waiter=None,
extra=None):
"""Создать транспорт для чтения из канала."""
raise NotImplementedError
def _make_write_pipe_transport(self, pipe, protocol, waiter=None,
extra=None):
"""Создать транспорт для записи в канал."""
raise NotImplementedError
async def _make_subprocess_transport(self, protocol, args, shell,
stdin, stdout, stderr, bufsize,
extra=None, **kwargs):
"""Создать транспорт подпроцесса."""
raise NotImplementedError
def _write_to_self(self):
"""Записать байт в self-pipe для пробуждения цикла событий.
Этот метод может быть вызван из другого потока.
Подкласс отвечает за реализацию self-pipe.
"""
raise NotImplementedError
def _process_events(self, event_list):
"""Обрабатывает события селектора."""
raise NotImplementedError
def _check_closed(self):
if self._closed:
raise RuntimeError('Event loop is closed')
def _check_default_executor(self):
if self._executor_shutdown_called:
raise RuntimeError('Executor shutdown has been called')
def _asyncgen_finalizer_hook(self, agen):
self._asyncgens.discard(agen)
if not self.is_closed():
self.call_soon_threadsafe(self.create_task, agen.aclose())
def _asyncgen_firstiter_hook(self, agen):
if self._asyncgens_shutdown_called:
warnings.warn(
f"asynchronous generator {agen!r} was scheduled after "
f"loop.shutdown_asyncgens() call",
ResourceWarning, source=self)
self._asyncgens.add(agen)
async def shutdown_asyncgens(self):
"""Завершает все активные асинхронные генераторы."""
self._asyncgens_shutdown_called = True
if not len(self._asyncgens):
# Если версия Python <3.6 или у нас нет асинхронных
# генераторов в работе.
return
closing_agens = list(self._asyncgens)
self._asyncgens.clear()
results = await tasks.gather(
*[ag.aclose() for ag in closing_agens],
return_exceptions=True)
for result, agen in zip(results, closing_agens):
if isinstance(result, Exception):
self.call_exception_handler({
'message': f'an error occurred during closing of '
f'asynchronous generator {agen!r}',
'exception': result,
'asyncgen': agen
})
async def shutdown_default_executor(self, timeout=None):
"""Планирует завершение работы исполнителя по умолчанию.
Параметр timeout задает время, которое будет предоставлено исполнителю
для завершения присоединения. Значение по умолчанию — None, что означает,
что исполнителю будет предоставлено неограниченное количество времени.
"""
self._executor_shutdown_called = True
if self._default_executor is None:
return
future = self.create_future()
thread = threading.Thread(target=self._do_shutdown, args=(future,))
thread.start()
try:
async with timeouts.timeout(timeout):
await future
except TimeoutError:
warnings.warn("The executor did not finishing joining "
f"its threads within {timeout} seconds.",
RuntimeWarning, stacklevel=2)
self._default_executor.shutdown(wait=False)
else:
thread.join()
def _do_shutdown(self, future):
try:
self._default_executor.shutdown(wait=True)
if not self.is_closed():
self.call_soon_threadsafe(futures._set_result_unless_cancelled,
future, None)
except Exception as ex:
if not self.is_closed() and not future.cancelled():
self.call_soon_threadsafe(future.set_exception, ex)
def _check_running(self):
if self.is_running():
raise RuntimeError('This event loop is already running')
if events._get_running_loop() is not None:
raise RuntimeError(
'Cannot run the event loop while another loop is running')
def _run_forever_setup(self):
"""Подготовить цикл выполнения к обработке событий.
Этот метод существует для того, чтобы подклассы пользовательского цикла событий (например, циклы событий,
интегрирующие цикл событий GUI с циклом событий Python) имели доступ ко всей
логике настройки цикла.
"""
self._check_closed()
self._check_running()
self._set_coroutine_origin_tracking(self._debug)
self._old_agen_hooks = sys.get_asyncgen_hooks()
self._thread_id = threading.get_ident()
sys.set_asyncgen_hooks(
firstiter=self._asyncgen_firstiter_hook,
finalizer=self._asyncgen_finalizer_hook
)
events._set_running_loop(self)
def _run_forever_cleanup(self):
"""Очистка после завершения цикла событий обработки событий.
Этот метод существует для того, чтобы подклассы пользовательского цикла событий (например, циклы событий,
интегрирующие цикл событий GUI с циклом событий Python) имели доступ ко всей
логике очистки цикла.
"""
self._stopping = False
self._thread_id = None
events._set_running_loop(None)
self._set_coroutine_origin_tracking(False)
# Восстановить все ранее существовавшие хуки асинхронных генераторов.
if self._old_agen_hooks is not None:
sys.set_asyncgen_hooks(*self._old_agen_hooks)
self._old_agen_hooks = None
def run_forever(self):
"""Выполняется до вызова stop()."""
self._run_forever_setup()
try:
while True:
self._run_once()
if self._stopping:
break
finally:
self._run_forever_cleanup()
def run_until_complete(self, future):
"""Выполняется до завершения Future.
Если аргумент — корутина, она оборачивается в задачу.
ПРЕДУПРЕЖДЕНИЕ: вызывать run_until_complete() дважды с одной и той же корутиной
губительно — она будет обернута в две разные задачи, и это ни к чему хорошему не приведет.
Возвращает результат Future или возбуждает его исключение.
"""
self._check_closed()
self._check_running()
new_task = not futures.isfuture(future)
future = tasks.ensure_future(future, loop=self)
if new_task:
# Исключение возбуждается, если future не завершился, поэтому
# нет необходимости логировать сообщение "destroy pending task"
future._log_destroy_pending = False
future.add_done_callback(_run_until_complete_cb)
try:
self.run_forever()
except:
if new_task and future.done() and not future.cancelled():
# Корутина возбудила BaseException. Поглощаем исключение
# чтобы не логировать предупреждение; у вызывающего кода нет доступа к
# локальной задаче.
future.exception()
raise
finally:
future.remove_done_callback(_run_until_complete_cb)
if not future.done():
raise RuntimeError('Event loop stopped before Future completed.')
return future.result()
def stop(self):
"""Остановить выполнение цикла событий.
Все уже запланированные колбэки всё равно будут выполнены. Это просто
сообщает run_forever, чтобы он прекратил циклический перебор после завершения итерации.
"""
self._stopping = True
def close(self):
"""Закрывает цикл событий.
Очищает очереди и завершает работу исполнителя,
но не ждет завершения исполнителя.
Цикл событий не должен быть запущен.
"""
if self.is_running():
raise RuntimeError("Cannot close a running event loop")
if self._closed:
return
if self._debug:
logger.debug("Close %r", self)
self._closed = True
self._ready.clear()
self._scheduled.clear()
self._executor_shutdown_called = True
executor = self._default_executor
if executor is not None:
self._default_executor = None
executor.shutdown(wait=False)
def is_closed(self):
"""Возвращает True, если цикл событий был закрыт."""
return self._closed
def __del__(self, _warn=warnings.warn):
if not self.is_closed():
_warn(f"unclosed event loop {self!r}", ResourceWarning, source=self)
if not self.is_running():
self.close()
def is_running(self):
"""Возвращает True, если цикл событий выполняется."""
return (self._thread_id is not None)
def time(self):
"""Возвращает время в соответствии с часами цикла событий.
Это число с плавающей запятой, выраженное в секундах от начала эпохи, но
эпоха, точность, правильность и дрейф не определены и могут
различаться для каждого цикла событий.
"""
return time.monotonic()
def call_later(self, delay, callback, *args, context=None):
"""Организовать вызов колбэка в заданное время.
Возвращает Handle — непрозрачный объект с методом cancel(), который
можно использовать для отмены вызова.
Задержка может быть int или float, выражена в секундах. Она
всегда задаётся относительно текущего момента времени.
Каждый колбэк будет вызван ровно один раз. Если два колбэка
запланированы на одно и то же время, порядок их вызова не определён.
Любые позиционные аргументы после колбэка будут переданы
колбэку при его вызове.
"""
if delay is None:
raise TypeError('delay must not be None')
timer = self.call_at(self.time() + delay, callback, *args,
context=context)
if timer._source_traceback:
del timer._source_traceback[-1]
return timer
def call_at(self, when, callback, *args, context=None):
"""Как call_later(), но использует абсолютное время.
Абсолютное время соответствует методу time() цикла событий.
"""
if when is None:
raise TypeError("when cannot be None")
self._check_closed()
if self._debug:
self._check_thread()
self._check_callback(callback, 'call_at')
timer = events.TimerHandle(when, callback, args, self, context)
if timer._source_traceback:
del timer._source_traceback[-1]
heapq.heappush(self._scheduled, timer)
timer._scheduled = True
return timer
def call_soon(self, callback, *args, context=None):
"""Планирует вызов колбэка как можно скорее.
Работает как очередь FIFO: колбэки вызываются в
порядке их регистрации. Каждый колбэк будет вызван
ровно один раз.
Любые позиционные аргументы после колбэка будут переданы
колбэку при его вызове.
"""
self._check_closed()
if self._debug:
self._check_thread()
self._check_callback(callback, 'call_soon')
handle = self._call_soon(callback, args, context)
if handle._source_traceback:
del handle._source_traceback[-1]
return handle
def _check_callback(self, callback, method):
if (coroutines.iscoroutine(callback) or
inspect.iscoroutinefunction(callback)):
raise TypeError(
f"coroutines cannot be used with {method}()")
if not callable(callback):
raise TypeError(
f'a callable object was expected by {method}(), '
f'got {callback!r}')
def _call_soon(self, callback, args, context):
handle = events.Handle(callback, args, self, context)
if handle._source_traceback:
del handle._source_traceback[-1]
self._ready.append(handle)
return handle
def _check_thread(self):
"""Проверяет, что текущий поток является потоком, выполняющим цикл событий.
Непотокобезопасные методы этого класса исходят из этого предположения и,
вероятно, будут работать некорректно при его нарушении.
Должен вызываться только когда (self._debug == True). Вызывающий код
отвечает за проверку этого условия по соображениям производительности.
"""
if self._thread_id is None:
return
thread_id = threading.get_ident()
if thread_id != self._thread_id:
raise RuntimeError(
"Non-thread-safe operation invoked on an event loop other "
"than the current one")
def call_soon_threadsafe(self, callback, *args, context=None):
"""Как call_soon(), но потокобезопасен."""
self._check_closed()
if self._debug:
self._check_callback(callback, 'call_soon_threadsafe')
handle = events._ThreadSafeHandle(callback, args, self, context)
self._ready.append(handle)
if handle._source_traceback:
del handle._source_traceback[-1]
if handle._source_traceback:
del handle._source_traceback[-1]
self._write_to_self()
return handle
def run_in_executor(self, executor, func, *args):
self._check_closed()
if self._debug:
self._check_callback(func, 'run_in_executor')
if executor is None:
executor = self._default_executor
# Проверять только при использовании исполнителя по умолчанию
self._check_default_executor()
if executor is None:
executor = concurrent.futures.ThreadPoolExecutor(
thread_name_prefix='asyncio'
)
self._default_executor = executor
return futures.wrap_future(
executor.submit(func, *args), loop=self)
def set_default_executor(self, executor):
if not isinstance(executor, concurrent.futures.ThreadPoolExecutor):
raise TypeError('executor must be ThreadPoolExecutor instance')
self._default_executor = executor
def _getaddrinfo_debug(self, host, port, family, type, proto, flags):
msg = [f"{host}:{port!r}"]
if family:
msg.append(f'family={family!r}')
if type:
msg.append(f'type={type!r}')
if proto:
msg.append(f'proto={proto!r}')
if flags:
msg.append(f'flags={flags!r}')
msg = ', '.join(msg)
logger.debug('Get address info %s', msg)
t0 = self.time()
addrinfo = socket.getaddrinfo(host, port, family, type, proto, flags)
dt = self.time() - t0
msg = f'Getting address info {msg} took {dt * 1e3:.3f}ms: {addrinfo!r}'
if dt >= self.slow_callback_duration:
logger.info(msg)
else:
logger.debug(msg)
return addrinfo
async def getaddrinfo(self, host, port, *,
family=0, type=0, proto=0, flags=0):
if self._debug:
getaddr_func = self._getaddrinfo_debug
else:
getaddr_func = socket.getaddrinfo
return await self.run_in_executor(
None, getaddr_func, host, port, family, type, proto, flags)
async def getnameinfo(self, sockaddr, flags=0):
return await self.run_in_executor(
None, socket.getnameinfo, sockaddr, flags)
async def sock_sendfile(self, sock, file, offset=0, count=None,
*, fallback=True):
if self._debug and sock.gettimeout() != 0:
raise ValueError("the socket must be non-blocking")
_check_ssl_socket(sock)
self._check_sendfile_params(sock, file, offset, count)
try:
return await self._sock_sendfile_native(sock, file,
offset, count)
except exceptions.SendfileNotAvailableError:
if not fallback:
raise
return await self._sock_sendfile_fallback(sock, file,
offset, count)
async def _sock_sendfile_native(self, sock, file, offset, count):
# Примечание: системный вызов sendfile не поддерживается для SSL-сокетов и
# файлов не-mmap, даже если sendfile поддерживается ОС
raise exceptions.SendfileNotAvailableError(
f"syscall sendfile is not available for socket {sock!r} "
f"and file {file!r} combination")
async def _sock_sendfile_fallback(self, sock, file, offset, count):
if hasattr(file, 'seek'):
file.seek(offset)
blocksize = (
min(count, constants.SENDFILE_FALLBACK_READBUFFER_SIZE)
if count else constants.SENDFILE_FALLBACK_READBUFFER_SIZE
)
buf = bytearray(blocksize)
total_sent = 0
try:
while True:
if count:
blocksize = min(count - total_sent, blocksize)
if blocksize <= 0:
break
view = memoryview(buf)[:blocksize]
read = await self.run_in_executor(None, file.readinto, view)
if not read:
break # EOF
await self.sock_sendall(sock, view[:read])
total_sent += read
return total_sent
finally:
if total_sent > 0 and hasattr(file, 'seek'):
file.seek(offset + total_sent)
def _check_sendfile_params(self, sock, file, offset, count):
if 'b' not in getattr(file, 'mode', 'b'):
raise ValueError("file should be opened in binary mode")
if not sock.type == socket.SOCK_STREAM:
raise ValueError("only SOCK_STREAM type sockets are supported")
if count is not None:
if not isinstance(count, int):
raise TypeError(
"count must be a positive integer (got {!r})".format(count))
if count <= 0:
raise ValueError(
"count must be a positive integer (got {!r})".format(count))
if not isinstance(offset, int):
raise TypeError(
"offset must be a non-negative integer (got {!r})".format(
offset))
if offset < 0:
raise ValueError(
"offset must be a non-negative integer (got {!r})".format(
offset))
async def _connect_sock(self, exceptions, addr_info, local_addr_infos=None):
"""Создать, привязать и подключить один сокет."""
my_exceptions = []
exceptions.append(my_exceptions)
family, type_, proto, _, address = addr_info
sock = None
try:
try:
sock = socket.socket(family=family, type=type_, proto=proto)
sock.setblocking(False)
if local_addr_infos is not None:
for lfamily, _, _, _, laddr in local_addr_infos:
# пропустить локальные адреса другого семейства
if lfamily != family:
continue
try:
sock.bind(laddr)
break
except OSError as exc:
msg = (
f'error while attempting to bind on '
f'address {laddr!r}: {str(exc).lower()}'
)
exc = OSError(exc.errno, msg)
my_exceptions.append(exc)
else: # все попытки привязки не удались
if my_exceptions:
raise my_exceptions.pop()
else:
raise OSError(f"no matching local address with {family=} found")
await self.sock_connect(sock, address)
return sock
except OSError as exc:
my_exceptions.append(exc)
raise
except:
if sock is not None:
try:
sock.close()
except OSError:
# Ошибка при закрытии недавно созданного сокета
# неважна, но может перезаписать более важную
# не-OSError ошибку. Поэтому игнорируйте её.
pass
raise
finally:
exceptions = my_exceptions = None
async def create_connection(
self, protocol_factory, host=None, port=None,
*, ssl=None, family=0,
proto=0, flags=0, sock=None,
local_addr=None, server_hostname=None,
ssl_handshake_timeout=None,
ssl_shutdown_timeout=None,
happy_eyeballs_delay=None, interleave=None,
all_errors=False):
"""Подключиться к TCP-серверу.
Создать потоковое транспортное соединение с заданным хостом и
портом: семейство сокетов AF_INET или socket.AF_INET6 в зависимости от хоста (или
family, если указано), тип сокета SOCK_STREAM. protocol_factory должен
быть вызываемым объектом, возвращающим экземпляр протокола.
Этот метод является корутиной, которая попытается установить
соединение в фоновом режиме. В случае успеха корутина
возвращает пару (transport, protocol).
"""
if server_hostname is not None and not ssl:
raise ValueError('server_hostname is only meaningful with ssl')
if server_hostname is None and ssl:
# Использовать host как значение по умолчанию для server_hostname. Это ошибка,
# если host пуст или не задан, например, когда
# был передан уже подключённый сокет или указан только порт
# указан. Чтобы избежать этой ошибки, можно передать
# server_hostname='' — это позволит обойти проверку имени хоста
# проверки. (Это также означает, что если хост является числовым
# IP/IPv6 адресом, мы попытаемся проверить именно этот
# адрес; это, скорее всего, не удастся, но возможно
# создать сертификат для конкретного IP-адреса, поэтому мы
# не судим об этом здесь.)
if not host:
raise ValueError('You must set server_hostname '
'when using ssl without a host')
server_hostname = host
if ssl_handshake_timeout is not None and not ssl:
raise ValueError(
'ssl_handshake_timeout is only meaningful with ssl')
if ssl_shutdown_timeout is not None and not ssl:
raise ValueError(
'ssl_shutdown_timeout is only meaningful with ssl')
if sock is not None:
_check_ssl_socket(sock)
if happy_eyeballs_delay is not None and interleave is None:
# Если используется Happy Eyeballs, по умолчанию чередуем адреса по семейству
interleave = 1
if host is not None or port is not None:
if sock is not None:
raise ValueError(
'host/port and sock can not be specified at the same time')
infos = await self._ensure_resolved(
(host, port), family=family,
type=socket.SOCK_STREAM, proto=proto, flags=flags, loop=self)
if not infos:
raise OSError('getaddrinfo() returned empty list')
if local_addr is not None:
laddr_infos = await self._ensure_resolved(
local_addr, family=family,
type=socket.SOCK_STREAM, proto=proto,
flags=flags, loop=self)
if not laddr_infos:
raise OSError('getaddrinfo() returned empty list')
else:
laddr_infos = None
if interleave:
infos = _interleave_addrinfos(infos, interleave)
exceptions = []
if happy_eyeballs_delay is None:
# не используем Happy Eyeballs
for addrinfo in infos:
try:
sock = await self._connect_sock(
exceptions, addrinfo, laddr_infos)
break
except OSError:
continue
else: # используем Happy Eyeballs
sock = (await staggered.staggered_race(
(
# нельзя использовать functools.partial, так как он хранит ссылку
# на исключения
lambda addrinfo=addrinfo: self._connect_sock(
exceptions, addrinfo, laddr_infos
)
for addrinfo in infos
),
happy_eyeballs_delay,
loop=self,
))[0] # нельзя использовать sock, _, _, так как он хранит ссылку на исключения
if sock is None:
exceptions = [exc for sub in exceptions for exc in sub]
try:
if all_errors:
raise ExceptionGroup("create_connection failed", exceptions)
if len(exceptions) == 1:
raise exceptions[0]
elif exceptions:
# Если у всех одинаковый str(), возбуждаем одно исключение.
model = str(exceptions[0])
if all(str(exc) == model for exc in exceptions):
raise exceptions[0]
# Возбуждаем комбинированное исключение, чтобы пользователь мог увидеть все
# различные сообщения об ошибках.
raise OSError('Multiple exceptions: {}'.format(
', '.join(str(exc) for exc in exceptions)))
else:
# Исключения не были собраны, возбудить ошибку тайм-аута
raise TimeoutError('create_connection failed')
finally:
exceptions = None
else:
if sock is None:
raise ValueError(
'host and port was not specified and no sock specified')
if sock.type != socket.SOCK_STREAM:
# Мы допускаем AF_INET, AF_INET6, AF_UNIX, если они
# являются SOCK_STREAM.
# Мы поддерживаем передачу сокетов AF_UNIX, хотя у нас есть
# для этого специальный API: create_unix_connection.
# Запрет AF_UNIX в этом методе нарушает обратную
# совместимость.
raise ValueError(
f'A Stream Socket was expected, got {sock!r}')
transport, protocol = await self._create_connection_transport(
sock, protocol_factory, ssl, server_hostname,
ssl_handshake_timeout=ssl_handshake_timeout,
ssl_shutdown_timeout=ssl_shutdown_timeout)
if self._debug:
# Получите сокет из транспорта, потому что SSL-транспорт закрывает
старый сокет и создает новый SSL-сокет
# Отправить файл через транспорт.
Возвращает общее количество отправленных байт.
Метод использует высокопроизводительный os.sendfile, если доступен.
file должен быть обычным файловым объектом, открытым в бинарном режиме.
offset указывает, с какого места начать чтение файла. Если указан,
count — это общее количество байт для передачи, в отличие от
отправки файла до достижения EOF. Позиция в файле обновляется при
возврате или в случае ошибки, и в этом случае file.tell()
можно использовать для определения количества отправленных
байт.
fallback, установленный в True, заставляет asyncio вручную читать и отправлять
файл, если платформа не поддерживает системный вызов sendfile
(например, Windows или SSL-сокет на Unix).
Вызывает SendfileNotAvailableError, если система не поддерживает
системный вызов sendfile, а fallback равен False.
sock = transport.get_extra_info('socket')
logger.debug("%r connected to %s:%r: (%r, %r)",
sock, host, port, transport, protocol)
return transport, protocol
async def _create_connection_transport(
self, sock, protocol_factory, ssl,
server_hostname, server_side=False,
ssl_handshake_timeout=None,
ssl_shutdown_timeout=None, context=None):
sock.setblocking(False)
context = context if context is not None else contextvars.copy_context()
protocol = protocol_factory()
waiter = self.create_future()
if ssl:
sslcontext = None if isinstance(ssl, bool) else ssl
transport = self._make_ssl_transport(
sock, protocol, sslcontext, waiter,
server_side=server_side, server_hostname=server_hostname,
ssl_handshake_timeout=ssl_handshake_timeout,
ssl_shutdown_timeout=ssl_shutdown_timeout,
context=context)
else:
transport = self._make_socket_transport(sock, protocol, waiter, context=context)
try:
await waiter
except:
transport.close()
raise
return transport, protocol
async def sendfile(self, transport, file, offset=0, count=None,
*, fallback=True):
"""Обновить транспорт до TLS.
Возвращает новый транспорт, который *protocol* должен начать использовать
немедленно.
"""
if transport.is_closing():
raise RuntimeError("Transport is closing")
mode = getattr(transport, '_sendfile_compatible',
constants._SendfileMode.UNSUPPORTED)
if mode is constants._SendfileMode.UNSUPPORTED:
raise RuntimeError(
f"sendfile is not supported for transport {transport!r}")
if mode is constants._SendfileMode.TRY_NATIVE:
try:
return await self._sendfile_native(transport, file,
offset, count)
except exceptions.SendfileNotAvailableError:
if not fallback:
raise
if not fallback:
raise exceptions.SendfileNotAvailableError(
f"fallback is disabled and native sendfile is not "
f"supported for transport {transport!r}")
return await self._sendfile_fallback(transport, file,
offset, count)
async def _sendfile_native(self, transp, file, offset, count):
raise exceptions.SendfileNotAvailableError(
"sendfile syscall is not supported")
async def _sendfile_fallback(self, transp, file, offset, count):
if hasattr(file, 'seek'):
file.seek(offset)
blocksize = (
min(count, constants.SENDFILE_FALLBACK_READBUFFER_SIZE)
if count else constants.SENDFILE_FALLBACK_READBUFFER_SIZE
)
buf = bytearray(blocksize)
total_sent = 0
proto = _SendfileFallbackProtocol(transp)
try:
while True:
if count:
blocksize = min(count - total_sent, blocksize)
if blocksize <= 0:
return total_sent
view = memoryview(buf)[:blocksize]
read = await self.run_in_executor(None, file.readinto, view)
if not read:
return total_sent # EOF
transp.write(view[:read])
await proto.drain()
total_sent += read
finally:
if total_sent > 0 and hasattr(file, 'seek'):
file.seek(offset + total_sent)
await proto.restore()
async def start_tls(self, transport, protocol, sslcontext, *,
server_side=False,
server_hostname=None,
ssl_handshake_timeout=None,
ssl_shutdown_timeout=None):
"""Приостановить раньше, чтобы "ssl_protocol.data_received()" не
"""
if ssl is None:
raise RuntimeError('Python ssl module is not available')
if not isinstance(sslcontext, ssl.SSLContext):
raise TypeError(
f'sslcontext is expected to be an instance of ssl.SSLContext, '
f'got {sslcontext!r}')
if not getattr(transport, '_start_tls_compatible', False):
raise TypeError(
f'transport {transport!r} is not supported by start_tls()')
waiter = self.create_future()
ssl_protocol = sslproto.SSLProtocol(
self, protocol, sslcontext, waiter,
server_side, server_hostname,
ssl_handshake_timeout=ssl_handshake_timeout,
ssl_shutdown_timeout=ssl_shutdown_timeout,
call_connection_made=False)
# успел вызваться до "ssl_protocol.connection_made()".
# Создать датаграммное соединение.
transport.pause_reading()
# gh-142352: переместить буферизованные данные StreamReader в SSLProtocol
if server_side:
from .streams import StreamReaderProtocol
if isinstance(protocol, StreamReaderProtocol):
stream_reader = getattr(protocol, '_stream_reader', None)
if stream_reader is not None:
buffer = stream_reader._buffer
if buffer:
ssl_protocol._incoming.write(buffer)
buffer.clear()
transport.set_protocol(ssl_protocol)
conmade_cb = self.call_soon(ssl_protocol.connection_made, transport)
resume_cb = self.call_soon(transport.resume_reading)
try:
await waiter
except BaseException:
transport.close()
conmade_cb.cancel()
resume_cb.cancel()
raise
return ssl_protocol._app_transport
async def create_datagram_endpoint(self, protocol_factory,
local_addr=None, remote_addr=None, *,
family=0, proto=0, flags=0,
reuse_port=None,
allow_broadcast=None, sock=None):
"""показать проблемные kwargs в сообщении исключения"""
if sock is not None:
if sock.type == socket.SOCK_STREAM:
raise ValueError(
f'A datagram socket was expected, got {sock!r}')
if (local_addr or remote_addr or
family or proto or flags or
reuse_port or allow_broadcast):
# У каталога могут быть права только на создание сокета.
opts = dict(local_addr=local_addr, remote_addr=remote_addr,
family=family, proto=proto, flags=flags,
reuse_port=reuse_port,
allow_broadcast=allow_broadcast)
problems = ', '.join(f'{k}={v}' for k, v in opts.items() if v)
raise ValueError(
f'socket modifier keyword arguments can not be used '
f'when sock is specified. ({problems})')
sock.setblocking(False)
r_addr = None
else:
if not (local_addr or remote_addr):
if family == 0:
raise ValueError('unexpected address family')
addr_pairs_info = (((family, proto), (None, None)),)
elif hasattr(socket, 'AF_UNIX') and family == socket.AF_UNIX:
for addr in (local_addr, remote_addr):
if addr is not None and not isinstance(addr, str):
raise TypeError('string is expected')
if local_addr and local_addr[0] not in (0, '\x00'):
try:
if stat.S_ISSOCK(os.stat(local_addr).st_mode):
os.remove(local_addr)
except FileNotFoundError:
pass
except OSError as err:
# объединить адрес по (family, protocol)
logger.error('Unable to check or remove stale UNIX '
'socket %r: %r',
local_addr, err)
addr_pairs_info = (((family, proto),
(local_addr, remote_addr)), )
else:
# Использование словаря с сохранением порядка
addr_infos = {} # каждый адрес должен иметь информацию для каждой пары (family, proto)
for idx, addr in ((0, local_addr), (1, remote_addr)):
if addr is not None:
if not (isinstance(addr, tuple) and len(addr) == 2):
raise TypeError('2-tuple is expected')
infos = await self._ensure_resolved(
addr, family=family, type=socket.SOCK_DGRAM,
proto=proto, flags=flags, loop=self)
if not infos:
raise OSError('getaddrinfo() returned empty list')
for fam, _, pro, _, address in infos:
key = (fam, pro)
if key not in addr_infos:
addr_infos[key] = [None, None]
addr_infos[key][idx] = address
# each addr has to have info for each (family, proto) pair
addr_pairs_info = [
(key, addr_pair) for key, addr_pair in addr_infos.items()
if not ((local_addr and addr_pair[0] is None) or
(remote_addr and addr_pair[1] is None))]
if not addr_pairs_info:
raise ValueError('can not get address information')
exceptions = []
for ((family, proto),
(local_address, remote_address)) in addr_pairs_info:
sock = None
r_addr = None
try:
sock = socket.socket(
family=family, type=socket.SOCK_DGRAM, proto=proto)
if reuse_port:
_set_reuseport(sock)
if allow_broadcast:
sock.setsockopt(
socket.SOL_SOCKET, socket.SO_BROADCAST, 1)
sock.setblocking(False)
if local_addr:
sock.bind(local_address)
if remote_addr:
if not allow_broadcast:
await self.sock_connect(sock, remote_address)
r_addr = remote_address
except OSError as exc:
if sock is not None:
sock.close()
exceptions.append(exc)
except:
if sock is not None:
sock.close()
raise
else:
break
else:
raise exceptions[0]
protocol = protocol_factory()
waiter = self.create_future()
transport = self._make_datagram_transport(
sock, protocol, r_addr, waiter)
if self._debug:
if local_addr:
logger.info("Datagram endpoint local_addr=%r remote_addr=%r "
"created: (%r, %r)",
local_addr, remote_addr, transport, protocol)
else:
logger.debug("Datagram endpoint remote_addr=%r created: "
"(%r, %r)",
remote_addr, transport, protocol)
try:
await waiter
except:
transport.close()
raise
return transport, protocol
async def _ensure_resolved(self, address, *,
family=0, type=socket.SOCK_STREAM,
proto=0, flags=0, loop):
host, port = address[:2]
info = _ipaddr_info(host, port, family, type, proto, *address[2:])
if info is not None:
# "host" уже является разрешенным IP-адресом.
return [info]
else:
return await loop.getaddrinfo(host, port, family=family, type=type,
proto=proto, flags=flags)
async def _create_server_getaddrinfo(self, host, port, family, flags):
infos = await self._ensure_resolved((host, port), family=family,
type=socket.SOCK_STREAM,
flags=flags, loop=self)
if not infos:
raise OSError(f'getaddrinfo({host!r}) returned empty list')
return infos
async def create_server(
self, protocol_factory, host=None, port=None,
*,
family=socket.AF_UNSPEC,
flags=socket.AI_PASSIVE,
sock=None,
backlog=100,
ssl=None,
reuse_address=None,
reuse_port=None,
keep_alive=None,
ssl_handshake_timeout=None,
ssl_shutdown_timeout=None,
start_serving=True):
"""Создать TCP-сервер.
Параметр host может быть строкой, в этом случае TCP-сервер привязывается
к хосту и порту.
Параметр host также может быть последовательностью строк, и в этом случае
TCP-сервер привязывается ко всем хостам последовательности. Если
хост встречается несколько раз (возможно косвенно, например, когда
имена хостов разрешаются в один и тот же IP-адрес), сервер привязывается
только один раз к этому хосту.
Возвращает объект Server, который можно использовать для остановки службы.
Этот метод является корутиной.
"""
if isinstance(ssl, bool):
raise TypeError('ssl argument must be an SSLContext or None')
if ssl_handshake_timeout is not None and ssl is None:
raise ValueError(
'ssl_handshake_timeout is only meaningful with ssl')
if ssl_shutdown_timeout is not None and ssl is None:
raise ValueError(
'ssl_shutdown_timeout is only meaningful with ssl')
if sock is not None:
_check_ssl_socket(sock)
if host is not None or port is not None:
if sock is not None:
raise ValueError(
'host/port and sock can not be specified at the same time')
if reuse_address is None:
reuse_address = os.name == "posix" and sys.platform != "cygwin"
sockets = []
if host == '':
hosts = [None]
elif (isinstance(host, str) or
not isinstance(host, collections.abc.Iterable)):
hosts = [host]
else:
hosts = host
fs = [self._create_server_getaddrinfo(host, port, family=family,
flags=flags)
for host in hosts]
infos = await tasks.gather(*fs)
infos = set(itertools.chain.from_iterable(infos))
completed = False
try:
for res in infos:
af, socktype, proto, canonname, sa = res
try:
sock = socket.socket(af, socktype, proto)
except socket.error:
# Предполагаем, что это неверная комбинация семейства/типа/протокола.
if self._debug:
logger.warning('create_server() failed to create '
'socket.socket(%r, %r, %r)',
af, socktype, proto, exc_info=True)
continue
sockets.append(sock)
if reuse_address:
sock.setsockopt(
socket.SOL_SOCKET, socket.SO_REUSEADDR, True)
# Начиная с Linux 6.12.9, SO_REUSEPORT не разрешен
# для других семейств адресов, отличных от AF_INET/AF_INET6.
if reuse_port and af in (socket.AF_INET, socket.AF_INET6):
_set_reuseport(sock)
if keep_alive:
sock.setsockopt(
socket.SOL_SOCKET, socket.SO_KEEPALIVE, True)
# Отключает поддержку двойного стека IPv4/IPv6 (включена по умолчанию
# на Linux), что позволяет одному сокету
# прослушивать оба семейства адресов.
if (_HAS_IPv6 and
af == socket.AF_INET6 and
hasattr(socket, 'IPPROTO_IPV6')):
sock.setsockopt(socket.IPPROTO_IPV6,
socket.IPV6_V6ONLY,
True)
try:
sock.bind(sa)
except OSError as err:
msg = ('error while attempting '
'to bind on address %r: %s'
% (sa, str(err).lower()))
if err.errno == errno.EADDRNOTAVAIL:
# Предполагаем, что семейство (family) не включено (bpo-30945).
sockets.pop()
sock.close()
if self._debug:
logger.warning(msg)
continue
raise OSError(err.errno, msg) from None
if not sockets:
raise OSError('could not bind on any address out of %r'
% ([info[4] for info in infos],))
completed = True
finally:
if not completed:
for sock in sockets:
sock.close()
else:
if sock is None:
raise ValueError('Neither host/port nor sock were specified')
if sock.type != socket.SOCK_STREAM:
raise ValueError(f'A Stream Socket was expected, got {sock!r}')
sockets = [sock]
for sock in sockets:
sock.setblocking(False)
server = Server(self, sockets, protocol_factory,
ssl, backlog, ssl_handshake_timeout,
ssl_shutdown_timeout)
if start_serving:
server._start_serving()
# Пропустить одну итерацию цикла, чтобы все 'loop.add_reader'
# прошли.
await tasks.sleep(0)
if self._debug:
logger.info("%r is serving", server)
return server
async def connect_accepted_socket(
self, protocol_factory, sock,
*, ssl=None,
ssl_handshake_timeout=None,
ssl_shutdown_timeout=None):
if sock.type != socket.SOCK_STREAM:
raise ValueError(f'A Stream Socket was expected, got {sock!r}')
if ssl_handshake_timeout is not None and not ssl:
raise ValueError(
'ssl_handshake_timeout is only meaningful with ssl')
if ssl_shutdown_timeout is not None and not ssl:
raise ValueError(
'ssl_shutdown_timeout is only meaningful with ssl')
_check_ssl_socket(sock)
transport, protocol = await self._create_connection_transport(
sock, protocol_factory, ssl, '', server_side=True,
ssl_handshake_timeout=ssl_handshake_timeout,
ssl_shutdown_timeout=ssl_shutdown_timeout)
if self._debug:
# Получите сокет из транспорта, потому что SSL-транспорт закрывает
старый сокет и создает новый SSL-сокет
# Отправить файл через транспорт.
Возвращает общее количество отправленных байт.
Метод использует высокопроизводительный os.sendfile, если доступен.
file должен быть обычным файловым объектом, открытым в бинарном режиме.
offset указывает, с какого места начать чтение файла. Если указан,
count — это общее количество байт для передачи, в отличие от
отправки файла до достижения EOF. Позиция в файле обновляется при
возврате или в случае ошибки, и в этом случае file.tell()
можно использовать для определения количества отправленных
байт.
fallback, установленный в True, заставляет asyncio вручную читать и отправлять
файл, если платформа не поддерживает системный вызов sendfile
(например, Windows или SSL-сокет на Unix).
Вызывает SendfileNotAvailableError, если система не поддерживает
системный вызов sendfile, а fallback равен False.
sock = transport.get_extra_info('socket')
logger.debug("%r handled: (%r, %r)", sock, transport, protocol)
return transport, protocol
async def connect_read_pipe(self, protocol_factory, pipe):
protocol = protocol_factory()
waiter = self.create_future()
transport = self._make_read_pipe_transport(pipe, protocol, waiter)
try:
await waiter
except:
transport.close()
raise
if self._debug:
logger.debug('Read pipe %r connected: (%r, %r)',
pipe.fileno(), transport, protocol)
return transport, protocol
async def connect_write_pipe(self, protocol_factory, pipe):
protocol = protocol_factory()
waiter = self.create_future()
transport = self._make_write_pipe_transport(pipe, protocol, waiter)
try:
await waiter
except:
transport.close()
raise
if self._debug:
logger.debug('Write pipe %r connected: (%r, %r)',
pipe.fileno(), transport, protocol)
return transport, protocol
def _log_subprocess(self, msg, stdin, stdout, stderr):
info = [msg]
if stdin is not None:
info.append(f'stdin={_format_pipe(stdin)}')
if stdout is not None and stderr == subprocess.STDOUT:
info.append(f'stdout=stderr={_format_pipe(stdout)}')
else:
if stdout is not None:
info.append(f'stdout={_format_pipe(stdout)}')
if stderr is not None:
info.append(f'stderr={_format_pipe(stderr)}')
logger.debug(' '.join(info))
async def subprocess_shell(self, protocol_factory, cmd, *,
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
universal_newlines=False,
shell=True, bufsize=0,
encoding=None, errors=None, text=None,
**kwargs):
if not isinstance(cmd, (bytes, str)):
raise ValueError("cmd must be a string")
if universal_newlines:
raise ValueError("universal_newlines must be False")
if not shell:
raise ValueError("shell must be True")
if bufsize != 0:
raise ValueError("bufsize must be 0")
if text:
raise ValueError("text must be False")
if encoding is not None:
raise ValueError("encoding must be None")
if errors is not None:
raise ValueError("errors must be None")
protocol = protocol_factory()
debug_log = None
if self._debug:
# не логировать параметры: они могут содержать конфиденциальную информацию
# (пароль) и могут быть слишком длинными
debug_log = 'run shell command %r' % cmd
self._log_subprocess(debug_log, stdin, stdout, stderr)
transport = await self._make_subprocess_transport(
protocol, cmd, True, stdin, stdout, stderr, bufsize, **kwargs)
if self._debug and debug_log is not None:
logger.info('%s: %r', debug_log, transport)
return transport, protocol
async def subprocess_exec(self, protocol_factory, program, *args,
stdin=subprocess.PIPE, stdout=subprocess.PIPE,
stderr=subprocess.PIPE, universal_newlines=False,
shell=False, bufsize=0,
encoding=None, errors=None, text=None,
**kwargs):
if universal_newlines:
raise ValueError("universal_newlines must be False")
if shell:
raise ValueError("shell must be False")
if bufsize != 0:
raise ValueError("bufsize must be 0")
if text:
raise ValueError("text must be False")
if encoding is not None:
raise ValueError("encoding must be None")
if errors is not None:
raise ValueError("errors must be None")
popen_args = (program,) + args
protocol = protocol_factory()
debug_log = None
if self._debug:
# не логировать параметры: они могут содержать конфиденциальную информацию
# (пароль) и могут быть слишком длинными
debug_log = f'execute program {program!r}'
self._log_subprocess(debug_log, stdin, stdout, stderr)
transport = await self._make_subprocess_transport(
protocol, popen_args, False, stdin, stdout, stderr,
bufsize, **kwargs)
if self._debug and debug_log is not None:
logger.info('%s: %r', debug_log, transport)
return transport, protocol
def get_exception_handler(self):
"""Возвращает обработчик исключений или None, если используется обработчик по умолчанию.
"""
return self._exception_handler
def set_exception_handler(self, handler):
"""Устанавливает handler в качестве нового обработчика исключений цикла событий.
Если handler равен None, будет установлен обработчик по умолчанию.
Если handler является вызываемым объектом, он должен иметь сигнатуру, соответствующую '(loop, context)', где 'loop' будет ссылкой на активный цикл событий, 'context' будет объектом dict (см. документацию `call_exception_handler()` для подробностей о context).
"""
if handler is not None and not callable(handler):
raise TypeError(f'A callable object or None is expected, '
f'got {handler!r}')
self._exception_handler = handler
def default_exception_handler(self, context):
"""Обработчик исключений по умолчанию.
Этот метод вызывается, когда происходит исключение и не установлен обработчик, и может быть вызван пользовательским обработчиком, который хочет отложить выполнение к поведению по умолчанию.
Этот обработчик по умолчанию логирует сообщение об ошибке и другую зависящую от контекста информацию. В режиме отладки также добавляется усеченная трассировка стека, показывающая, где был создан данный объект (например, handle, future или task), если таковой имеется.
Параметр context имеет то же значение, что и в `call_exception_handler()`.
"""
message = context.get('message')
if not message:
message = 'Unhandled exception in event loop'
exception = context.get('exception')
if exception is not None:
exc_info = (type(exception), exception, exception.__traceback__)
else:
exc_info = False
if ('source_traceback' not in context and
self._current_handle is not None and
self._current_handle._source_traceback):
context['handle_traceback'] = \
self._current_handle._source_traceback
log_lines = [message]
for key in sorted(context):
if key in {'message', 'exception'}:
continue
value = context[key]
if key == 'source_traceback':
tb = ''.join(traceback.format_list(value))
value = 'Object created at (most recent call last):\n'
value += tb.rstrip()
elif key == 'handle_traceback':
tb = ''.join(traceback.format_list(value))
value = 'Handle created at (most recent call last):\n'
value += tb.rstrip()
else:
value = repr(value)
log_lines.append(f'{key}: {value}')
logger.error('\n'.join(log_lines), exc_info=exc_info)
def call_exception_handler(self, context):
"""Вызвать обработчик исключений текущего цикла событий.
Аргумент context представляет собой словарь, содержащий следующие ключи:
- 'message': сообщение об ошибке;
- 'exception' (необязательно): объект исключения;
- 'future' (необязательно): экземпляр Future;
- 'task' (необязательно): экземпляр Task;
- 'handle' (необязательно): экземпляр Handle;
- 'protocol' (необязательно): экземпляр Protocol;
- 'transport' (необязательно): экземпляр Transport;
- 'socket' (необязательно): экземпляр Socket;
- 'source_traceback' (необязательно): traceback источника;
- 'handle_traceback' (необязательно): traceback дескриптора;
- 'asyncgen' (необязательно): асинхронный генератор, вызвавший исключение.
В будущем могут быть добавлены новые ключи.
Примечание: не перегружайте этот метод в подклассе цикла событий.
Для пользовательской обработки исключений используйте метод
`set_exception_handler()`.
"""
if self._exception_handler is None:
try:
self.default_exception_handler(context)
except (SystemExit, KeyboardInterrupt):
raise
except BaseException:
# Второй уровень защиты от непредвиденных ошибок
# в реализации по умолчанию, а также для подклассов
# циклов событий с переопределенным "default_exception_handler".
logger.error('Exception in default exception handler',
exc_info=True)
else:
try:
ctx = None
thing = context.get("task")
if thing is None:
# Хотя Future не имеют контекста,
# Task является подклассом Future,
# и иногда ключ 'future' содержит Task.
thing = context.get("future")
if thing is None:
# Хендлы также имеют контекст.
thing = context.get("handle")
if thing is not None and hasattr(thing, "get_context"):
ctx = thing.get_context()
if ctx is not None and hasattr(ctx, "run"):
ctx.run(self._exception_handler, self, context)
else:
self._exception_handler(self, context)
except (SystemExit, KeyboardInterrupt):
raise
except BaseException as exc:
# Исключение в пользовательском обработчике исключений.
try:
# Попробуем обработчик по умолчанию.
self.default_exception_handler({
'message': 'Unhandled error in exception handler',
'exception': exc,
'context': context,
})
except (SystemExit, KeyboardInterrupt):
raise
except BaseException:
# Защита 'default_exception_handler' на случай, если он
# перегружен.
logger.error('Exception in default exception handler '
'while handling an unexpected error '
'in custom exception handler',
exc_info=True)
def _add_callback(self, handle):
"""Добавить Handle в _ready."""
if not handle._cancelled:
self._ready.append(handle)
def _add_callback_signalsafe(self, handle):
"""Как _add_callback(), но вызывается из обработчика сигналов."""
self._add_callback(handle)
self._write_to_self()
def _timer_handle_cancelled(self, handle):
"""Уведомление о том, что TimerHandle был отменён."""
if handle._scheduled:
self._timer_cancelled_count += 1
def _run_once(self):
"""Выполнить одну полную итерацию цикла событий.
Это вызывает все готовые на данный момент колбэки, опрашивает I/O,
планирует полученные колбэки и, наконец, планирует
колбэки 'call_later'.
"""
sched_count = len(self._scheduled)
if (sched_count > _MIN_SCHEDULED_TIMER_HANDLES and
self._timer_cancelled_count / sched_count >
_MIN_CANCELLED_TIMER_HANDLES_FRACTION):
# Удалить отложенные вызовы, которые были отменены, если их количество
# слишком велико
new_scheduled = []
for handle in self._scheduled:
if handle._cancelled:
handle._scheduled = False
else:
new_scheduled.append(handle)
heapq.heapify(new_scheduled)
self._scheduled = new_scheduled
self._timer_cancelled_count = 0
else:
# Удалить отменённые отложенные вызовы из начала очереди.
while self._scheduled and self._scheduled[0]._cancelled:
self._timer_cancelled_count -= 1
handle = heapq.heappop(self._scheduled)
handle._scheduled = False
timeout = None
if self._ready or self._stopping:
timeout = 0
elif self._scheduled:
# Вычислить желаемый таймаут.
timeout = self._scheduled[0]._when - self.time()
if timeout > MAXIMUM_SELECT_TIMEOUT:
timeout = MAXIMUM_SELECT_TIMEOUT
elif timeout < 0:
timeout = 0
event_list = self._selector.select(timeout)
self._process_events(event_list)
# Необходимо для разрыва циклов при возникновении исключения.
event_list = None
# Обработать готовые колбэки 'later'.
now = self.time()
# Убедиться, что `end_time` строго возрастает
# когда разрешение часов слишком мало.
end_time = now + max(self._clock_resolution, math.ulp(now))
while self._scheduled:
handle = self._scheduled[0]
if handle._when >= end_time:
break
handle = heapq.heappop(self._scheduled)
handle._scheduled = False
self._ready.append(handle)
# Это единственное место, где колбэки действительно *вызываются*.
# Все остальные места просто добавляют их в ready.
# Примечание: Мы запускаем все запланированные на данный момент колбэки, но не те,
# колбэки, запланированные колбэками, запущенными в этот раз,
# они будут запущены в следующий раз (после очередного опроса ввода-вывода).
# Используйте идиому, потокобезопасную без использования блокировок.
ntodo = len(self._ready)
for i in range(ntodo):
handle = self._ready.popleft()
if handle._cancelled:
continue
if self._debug:
try:
self._current_handle = handle
t0 = self.time()
handle._run()
dt = self.time() - t0
if dt >= self.slow_callback_duration:
logger.warning('Executing %s took %.3f seconds',
_format_handle(handle), dt)
finally:
self._current_handle = None
else:
handle._run()
handle = None # Необходимо для разрыва циклов при возникновении исключения.
def _set_coroutine_origin_tracking(self, enabled):
if bool(enabled) == bool(self._coroutine_origin_tracking_enabled):
return
if enabled:
self._coroutine_origin_tracking_saved_depth = (
sys.get_coroutine_origin_tracking_depth())
sys.set_coroutine_origin_tracking_depth(
constants.DEBUG_STACK_DEPTH)
else:
sys.set_coroutine_origin_tracking_depth(
self._coroutine_origin_tracking_saved_depth)
self._coroutine_origin_tracking_enabled = enabled
def get_debug(self):
return self._debug
def set_debug(self, enabled):
self._debug = enabled
if self.is_running():
self.call_soon_threadsafe(self._set_coroutine_origin_tracking, enabled)