base_events.py
python
"""Базовая реализация цикла событий.
Цикл событий может быть разделён на мультиплексор (часть, отвечающая за уведомление о событиях ввода-вывода) и собственно цикл событий, который оборачивает мультиплексор функциональностью для планирования колбэков, немедленно или в заданное время в будущем.
Всякий раз, когда публичный API принимает колбэк, последующие позиционные аргументы будут переданы колбэку при его вызове. Это позволяет избежать распространения тривиальных лямбда-функций, реализующих замыкания. Именованные аргументы для колбэка не поддерживаются; это осознанное проектное решение, оставляющее возможность для ключевых аргументов изменять смысл самого вызова API.
"""
import collections
import concurrent.futures
import heapq
import inspect
import itertools
import logging
import os
import socket
import subprocess
import threading
import time
import traceback
import sys
import warnings
import weakref
from . import compat
from . import coroutines
from . import events
from . import futures
from . import tasks
from .coroutines import coroutine
from .log import logger
__all__ = ['BaseEventLoop']
# Минимальное количество запланированных дескрипторов таймера перед очисткой
# отменённых дескрипторов.
_MIN_SCHEDULED_TIMER_HANDLES = 100
# Минимальная доля запланированных дескрипторов таймера, которые отменены
# перед выполнением очистки отменённых дескрипторов.
_MIN_CANCELLED_TIMER_HANDLES_FRACTION = 0.5
# Исключения, которые не должны вызывать обработчик исключений в методах фатальной ошибки
# (_fatal_error())
_FATAL_ERROR_IGNORE = (BrokenPipeError,
ConnectionResetError, ConnectionAbortedError)
_HAS_IPv6 = hasattr(socket, 'AF_INET6')
# Максимальное время ожидания, передаваемое в select, чтобы избежать ограничений ОС
MAXIMUM_SELECT_TIMEOUT = 24 * 3600
# Используется для объявления устаревшим и удаления параметра `loop.create_datagram_endpoint()`
# параметр *reuse_address*
_unset = object()
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 _is_stream_socket(sock_type):
if hasattr(socket, 'SOCK_NONBLOCK'):
# В Linux socket.type представляет собой битовую маску, которая может включать дополнительную информацию
# о сокете (например, бит SOCK_NONBLOCK), поэтому мы не можем сделать простое
# `sock_type == socket.SOCK_STREAM`, см.
# https://github.com/torvalds/linux/blob/v4.13/include/linux/net.h#L77
# для получения более подробной информации.
return (sock_type & 0xF) == socket.SOCK_STREAM
else:
return sock_type == socket.SOCK_STREAM
def _is_dgram_socket(sock_type):
if hasattr(socket, 'SOCK_NONBLOCK'):
# См. комментарий в `_is_stream_socket`.
return (sock_type & 0xF) == socket.SOCK_DGRAM
else:
return sock_type == socket.SOCK_DGRAM
def _ipaddr_info(host, port, family, type, proto):
# Попытаться пропустить 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 _is_stream_socket(type):
proto = socket.IPPROTO_TCP
elif _is_dgram_socket(type):
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, 0, 0)
else:
return af, type, proto, '', (host, port)
except OSError:
pass
# "host" не является IP-адресом.
return None
def _ensure_resolved(address, *, family=0, type=socket.SOCK_STREAM, proto=0,
flags=0, loop):
host, port = address[:2]
info = _ipaddr_info(host, port, family, type, proto)
if info is not None:
# "host" уже является разрешенным IP-адресом.
fut = loop.create_future()
fut.set_result([info])
return fut
else:
return loop.getaddrinfo(host, port, family=family, type=type,
proto=proto, flags=flags)
if hasattr(socket, 'TCP_NODELAY'):
def _set_nodelay(sock):
if (sock.family in {socket.AF_INET, socket.AF_INET6} and
_is_stream_socket(sock.type) and
sock.proto == socket.IPPROTO_TCP):
sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
else:
def _set_nodelay(sock):
pass
def _run_until_complete_cb(fut):
exc = fut._exception
if (isinstance(exc, BaseException)
and not isinstance(exc, Exception)):
# Issue #22429: run_forever() уже завершена, нет необходимости
# останавливать её.
return
fut._loop.stop()
class Server(events.AbstractServer):
def __init__(self, loop, sockets):
self._loop = loop
self.sockets = sockets
self._active_count = 0
self._waiters = []
def __repr__(self):
return '<%s sockets=%r>' % (self.__class__.__name__, self.sockets)
def _attach(self):
assert self.sockets is not None
self._active_count += 1
def _detach(self):
assert self._active_count > 0
self._active_count -= 1
if self._active_count == 0 and self.sockets is None:
self._wakeup()
def close(self):
sockets = self.sockets
if sockets is None:
return
self.sockets = None
for sock in sockets:
self._loop._stop_serving(sock)
if self._active_count == 0:
self._wakeup()
def _wakeup(self):
waiters = self._waiters
self._waiters = None
for waiter in waiters:
if not waiter.done():
waiter.set_result(waiter)
@coroutine
def wait_closed(self):
if self.sockets is None or self._waiters is None:
return
waiter = self._loop.create_future()
self._waiters.append(waiter)
yield from 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((not sys.flags.ignore_environment
and bool(os.environ.get('PYTHONASYNCIODEBUG'))))
# В режиме отладки, если выполнение колбэка или шага задачи
# если время выполнения в секундах превышает эту длительность, медленный колбэк/задача регистрируется.
self.slow_callback_duration = 0.1
self._current_handle = None
self._task_factory = None
self._coroutine_wrapper_set = False
if hasattr(sys, 'get_asyncgen_hooks'):
# Python >= 3.6
# Слабое множество всех асинхронных генераторов, которые
# итерируются циклом.
self._asyncgens = weakref.WeakSet()
else:
self._asyncgens = None
# Устанавливается в True при вызове `loop.shutdown_asyncgens`.
self._asyncgens_shutdown_called = False
def __repr__(self):
return ('<%s running=%s closed=%s debug=%s>'
% (self.__class__.__name__, self.is_running(),
self.is_closed(), self.get_debug()))
def create_future(self):
"""Создать объект Future, привязанный к циклу."""
return futures.Future(loop=self)
def create_task(self, coro):
"""Запланировать выполнение объекта корутины.
Возвращает объект задачи.
"""
self._check_closed()
if self._task_factory is None:
task = tasks.Task(coro, loop=self)
if task._source_traceback:
del task._source_traceback[-1]
else:
task = self._task_factory(self, coro)
return task
def set_task_factory(self, factory):
"""Установить фабрику задач, которая будет использоваться `loop.create_task()`.
Если factory равен None, будет установлена фабрика задач по умолчанию.
Если factory является вызываемым объектом, его сигнатура должна соответствовать
'(loop, coro)', где 'loop' будет ссылкой на активный
цикл событий, 'coro' — объектом корутины. Вызываемый объект
должен возвращать Future.
"""
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-транспорт."""
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
@coroutine
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 _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(
"asynchronous generator {!r} was scheduled after "
"loop.shutdown_asyncgens() call".format(agen),
ResourceWarning, source=self)
self._asyncgens.add(agen)
@coroutine
def shutdown_asyncgens(self):
"""Завершает все активные асинхронные генераторы."""
self._asyncgens_shutdown_called = True
if self._asyncgens is None or not len(self._asyncgens):
# Если версия Python <3.6 или у нас нет асинхронных
# генераторов в работе.
return
closing_agens = list(self._asyncgens)
self._asyncgens.clear()
shutdown_coro = tasks.gather(
*[ag.aclose() for ag in closing_agens],
return_exceptions=True,
loop=self)
results = yield from shutdown_coro
for result, agen in zip(results, closing_agens):
if isinstance(result, Exception):
self.call_exception_handler({
'message': 'an error occurred during closing of '
'asynchronous generator {!r}'.format(agen),
'exception': result,
'asyncgen': agen
})
def run_forever(self):
"""Выполняется до вызова stop()."""
self._check_closed()
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')
self._set_coroutine_wrapper(self._debug)
self._thread_id = threading.get_ident()
if self._asyncgens is not None:
old_agen_hooks = sys.get_asyncgen_hooks()
sys.set_asyncgen_hooks(firstiter=self._asyncgen_firstiter_hook,
finalizer=self._asyncgen_finalizer_hook)
try:
events._set_running_loop(self)
while True:
self._run_once()
if self._stopping:
break
finally:
self._stopping = False
self._thread_id = None
events._set_running_loop(None)
self._set_coroutine_wrapper(False)
if self._asyncgens is not None:
sys.set_asyncgen_hooks(*old_agen_hooks)
def run_until_complete(self, future):
"""Выполняется до завершения Future.
Если аргумент — корутина, она оборачивается в задачу.
ПРЕДУПРЕЖДЕНИЕ: вызывать run_until_complete() дважды с одной и той же корутиной
губительно — она будет обернута в две разные задачи, и это ни к чему хорошему не приведет.
Возвращает результат Future или возбуждает его исключение.
"""
self._check_closed()
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()
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
# В Python 3.3 и старше объекты с деструктором, являющиеся частью циклической ссылки,
# никогда не уничтожаются. В Python 3.4 это уже не так благодаря
# PEP 442.
if compat.PY34:
def __del__(self):
if not self.is_closed():
warnings.warn("unclosed event loop %r" % self, 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):
"""Планирует вызов колбэка в заданное время.
Возвращает Handle — непрозрачный объект с методом cancel(), который
можно использовать для отмены вызова.
Задержка может быть целым числом или числом с плавающей запятой, выраженным в секундах. Она
всегда указывается относительно текущего времени.
Каждый колбэк будет вызван ровно один раз. Если два колбэка
запланированы на одно и то же время, не определено, какой
будет вызван первым.
Любые позиционные аргументы после колбэка будут переданы
колбэку при его вызове.
"""
timer = self.call_at(self.time() + delay, callback, *args)
if timer._source_traceback:
del timer._source_traceback[-1]
return timer
def call_at(self, when, callback, *args):
"""Как call_later(), но использует абсолютное время.
Абсолютное время соответствует методу time() цикла событий.
"""
self._check_closed()
if self._debug:
self._check_thread()
self._check_callback(callback, 'call_at')
timer = events.TimerHandle(when, callback, args, self)
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):
"""Планирует вызов колбэка как можно скорее.
Работает как очередь FIFO: колбэки вызываются в
порядке их регистрации. Каждый колбэк будет вызван
ровно один раз.
Любые позиционные аргументы после колбэка будут переданы
колбэку при его вызове.
"""
self._check_closed()
if self._debug:
self._check_thread()
self._check_callback(callback, 'call_soon')
handle = self._call_soon(callback, args)
if handle._source_traceback:
del handle._source_traceback[-1]
return handle
def _check_callback(self, callback, method):
if (coroutines.iscoroutine(callback) or
coroutines.iscoroutinefunction(callback)):
raise TypeError(
"coroutines cannot be used with {}()".format(method))
if not callable(callback):
raise TypeError(
'a callable object was expected by {}(), got {!r}'.format(
method, callback))
def _call_soon(self, callback, args):
handle = events.Handle(callback, args, self)
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):
"""Как call_soon(), но потокобезопасен."""
self._check_closed()
if self._debug:
self._check_callback(callback, 'call_soon_threadsafe')
handle = self._call_soon(callback, args)
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
if executor is None:
executor = concurrent.futures.ThreadPoolExecutor()
self._default_executor = executor
return futures.wrap_future(executor.submit(func, *args), loop=self)
def set_default_executor(self, executor):
self._default_executor = executor
def _getaddrinfo_debug(self, host, port, family, type, proto, flags):
msg = ["%s:%r" % (host, port)]
if family:
msg.append('family=%r' % family)
if type:
msg.append('type=%r' % type)
if proto:
msg.append('proto=%r' % proto)
if flags:
msg.append('flags=%r' % flags)
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 = ('Getting address info %s took %.3f ms: %r'
% (msg, dt * 1e3, addrinfo))
if dt >= self.slow_callback_duration:
logger.info(msg)
else:
logger.debug(msg)
return addrinfo
def getaddrinfo(self, host, port, *,
family=0, type=0, proto=0, flags=0):
if self._debug:
return self.run_in_executor(None, self._getaddrinfo_debug,
host, port, family, type, proto, flags)
else:
return self.run_in_executor(None, socket.getaddrinfo,
host, port, family, type, proto, flags)
def getnameinfo(self, sockaddr, flags=0):
return self.run_in_executor(None, socket.getnameinfo, sockaddr, flags)
@coroutine
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):
"""Подключиться к 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 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')
f1 = _ensure_resolved((host, port), family=family,
type=socket.SOCK_STREAM, proto=proto,
flags=flags, loop=self)
fs = [f1]
if local_addr is not None:
f2 = _ensure_resolved(local_addr, family=family,
type=socket.SOCK_STREAM, proto=proto,
flags=flags, loop=self)
fs.append(f2)
else:
f2 = None
yield from tasks.wait(fs, loop=self)
infos = f1.result()
if not infos:
raise OSError('getaddrinfo() returned empty list')
if f2 is not None:
laddr_infos = f2.result()
if not laddr_infos:
raise OSError('getaddrinfo() returned empty list')
exceptions = []
for family, type, proto, cname, address in infos:
try:
sock = socket.socket(family=family, type=type, proto=proto)
sock.setblocking(False)
if f2 is not None:
for _, _, _, _, laddr in laddr_infos:
try:
sock.bind(laddr)
break
except OSError as exc:
exc = OSError(
exc.errno, 'error while '
'attempting to bind on address '
'{!r}: {}'.format(
laddr, exc.strerror.lower()))
exceptions.append(exc)
else:
sock.close()
sock = None
continue
if self._debug:
logger.debug("connect %r to %r", sock, address)
yield from self.sock_connect(sock, 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:
if len(exceptions) == 1:
raise exceptions[0]
else:
# Если у всех одинаковый 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:
if sock is None:
raise ValueError(
'host and port was not specified and no sock specified')
if not _is_stream_socket(sock.type):
# Мы допускаем AF_INET, AF_INET6, AF_UNIX, если они
# являются SOCK_STREAM.
# Мы поддерживаем передачу сокетов AF_UNIX, хотя у нас есть
# для этого специальный API: create_unix_connection.
# Запрет AF_UNIX в этом методе нарушает обратную
# совместимость.
raise ValueError(
'A Stream Socket was expected, got {!r}'.format(sock))
transport, protocol = yield from self._create_connection_transport(
sock, protocol_factory, ssl, server_hostname)
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
@coroutine
def _create_connection_transport(self, sock, protocol_factory, ssl,
server_hostname, server_side=False):
sock.setblocking(False)
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)
else:
transport = self._make_socket_transport(sock, protocol, waiter)
try:
yield from waiter
except:
transport.close()
raise
return transport, protocol
@coroutine
def create_datagram_endpoint(self, protocol_factory,
local_addr=None, remote_addr=None, *,
family=0, proto=0, flags=0,
reuse_address=_unset, reuse_port=None,
allow_broadcast=None, sock=None):
"""показать проблемные kwargs в сообщении исключения"""
if sock is not None:
if not _is_dgram_socket(sock.type):
raise ValueError(
'A UDP Socket was expected, got {!r}'.format(sock))
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_address=reuse_address, reuse_port=reuse_port,
allow_broadcast=allow_broadcast)
problems = ', '.join(
'{}={}'.format(k, v) for k, v in opts.items() if v)
raise ValueError(
'socket modifier keyword arguments can not be used '
'when sock is specified. ({})'.format(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)),)
else:
# Использование словаря с сохранением порядка
addr_infos = collections.OrderedDict()
for idx, addr in ((0, local_addr), (1, remote_addr)):
if addr is not None:
assert isinstance(addr, tuple) and len(addr) == 2, (
'2-tuple is expected')
infos = yield from _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 = []
# bpo-37228
if reuse_address is not _unset:
if reuse_address:
raise ValueError("Passing `reuse_address=True` is no "
"longer supported, as the usage of "
"SO_REUSEPORT in UDP poses a significant "
"security concern.")
else:
warnings.warn("The *reuse_address* parameter has been "
"deprecated as of 3.6.10 and is scheduled "
"for removal in 3.11.", DeprecationWarning,
stacklevel=2)
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:
yield from 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:
yield from waiter
except:
transport.close()
raise
return transport, protocol
@coroutine
def _create_server_getaddrinfo(self, host, port, family, flags):
infos = yield from _ensure_resolved((host, port), family=family,
type=socket.SOCK_STREAM,
flags=flags, loop=self)
if not infos:
raise OSError('getaddrinfo({!r}) returned empty list'.format(host))
return infos
@coroutine
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):
"""Создать TCP-сервер.
Параметр host может быть строкой, в этом случае TCP-сервер привязывается
к хосту и порту.
Параметр host также может быть последовательностью строк, и в этом случае
TCP-сервер привязывается ко всем хостам из последовательности. Если хост
встречается несколько раз (возможно косвенно, например, когда имена хостов
разрешаются в один и тот же IP-адрес), сервер привязывается только один раз к этому
хосту.
Возвращает объект Server, который можно использовать для остановки службы.
Этот метод является корутиной.
"""
if isinstance(ssl, bool):
raise TypeError('ssl argument must be an SSLContext or None')
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.Iterable)):
hosts = [host]
else:
hosts = host
fs = [self._create_server_getaddrinfo(host, port, family=family,
flags=flags)
for host in hosts]
infos = yield from tasks.gather(*fs, loop=self)
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)
if reuse_port:
_set_reuseport(sock)
# Отключает поддержку двойного стека 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:
raise OSError(err.errno, 'error while attempting '
'to bind on address %r: %s'
% (sa, err.strerror.lower()))
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 not _is_stream_socket(sock.type):
raise ValueError(
'A Stream Socket was expected, got {!r}'.format(sock))
sockets = [sock]
server = Server(self, sockets)
for sock in sockets:
sock.listen(backlog)
sock.setblocking(False)
self._start_serving(protocol_factory, sock, ssl, server, backlog)
if self._debug:
logger.info("%r is serving", server)
return server
@coroutine
def connect_accepted_socket(self, protocol_factory, sock, *, ssl=None):
"""Обработать принятое соединение.
Используется серверами, которые принимают соединения вне
asyncio, но используют asyncio для обработки соединений.
Этот метод является корутиной. После завершения корутина
возвращает пару (транспорт, протокол).
"""
if not _is_stream_socket(sock.type):
raise ValueError(
'A Stream Socket was expected, got {!r}'.format(sock))
transport, protocol = yield from self._create_connection_transport(
sock, protocol_factory, ssl, '', server_side=True)
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
@coroutine
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:
yield from waiter
except:
transport.close()
raise
if self._debug:
logger.debug('Read pipe %r connected: (%r, %r)',
pipe.fileno(), transport, protocol)
return transport, protocol
@coroutine
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:
yield from 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('stdin=%s' % _format_pipe(stdin))
if stdout is not None and stderr == subprocess.STDOUT:
info.append('stdout=stderr=%s' % _format_pipe(stdout))
else:
if stdout is not None:
info.append('stdout=%s' % _format_pipe(stdout))
if stderr is not None:
info.append('stderr=%s' % _format_pipe(stderr))
logger.debug(' '.join(info))
@coroutine
def subprocess_shell(self, protocol_factory, cmd, *, stdin=subprocess.PIPE,
stdout=subprocess.PIPE, stderr=subprocess.PIPE,
universal_newlines=False, shell=True, bufsize=0,
**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")
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 = yield from 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
@coroutine
def subprocess_exec(self, protocol_factory, program, *args,
stdin=subprocess.PIPE, stdout=subprocess.PIPE,
stderr=subprocess.PIPE, universal_newlines=False,
shell=False, bufsize=0, **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")
popen_args = (program,) + args
for arg in popen_args:
if not isinstance(arg, (str, bytes)):
raise TypeError("program arguments must be "
"a bytes or text string, not %s"
% type(arg).__name__)
protocol = protocol_factory()
debug_log = None
if self._debug:
# не логировать параметры: они могут содержать конфиденциальную информацию
# (пароль) и могут быть слишком длинными
debug_log = 'execute program %r' % program
self._log_subprocess(debug_log, stdin, stdout, stderr)
transport = yield from 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('A callable object or None is expected, '
'got {!r}'.format(handler))
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('{}: {}'.format(key, value))
logger.error('\n'.join(log_lines), exc_info=exc_info)
def call_exception_handler(self, context):
"""Вызвать обработчик исключений текущего цикла событий.
Аргумент context представляет собой словарь, содержащий следующие ключи:
- 'message': сообщение об ошибке;
- 'exception' (необязательно): объект исключения;
- 'future' (необязательно): экземпляр Future;
- 'handle' (необязательно): экземпляр Handle;
- 'protocol' (необязательно): экземпляр Protocol;
- 'transport' (необязательно): экземпляр Transport;
- 'socket' (необязательно): экземпляр Socket;
- 'asyncgen' (необязательно): асинхронный генератор, вызвавший
исключение.
В будущем могут быть добавлены новые ключи.
Примечание: не перегружайте этот метод в подклассе цикла событий.
Для настраиваемой обработки исключений используйте метод
`set_exception_handler()`.
"""
if self._exception_handler is None:
try:
self.default_exception_handler(context)
except Exception:
# Второй уровень защиты от непредвиденных ошибок
# в реализации по умолчанию, а также для подклассов
# циклов событий с переопределенным "default_exception_handler".
logger.error('Exception in default exception handler',
exc_info=True)
else:
try:
self._exception_handler(self, context)
except Exception as exc:
# Исключение в пользовательском обработчике исключений.
try:
# Попробуем обработчик по умолчанию.
self.default_exception_handler({
'message': 'Unhandled error in exception handler',
'exception': exc,
'context': context,
})
except Exception:
# Защита '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 в _scheduled (TimerHandle) или _ready."""
assert isinstance(handle, events.Handle), 'A Handle is required here'
if handle._cancelled:
return
assert not isinstance(handle, events.TimerHandle)
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:
# Вычислить желаемый таймаут.
when = self._scheduled[0]._when
timeout = min(max(0, when - self.time()), MAXIMUM_SELECT_TIMEOUT)
if self._debug and timeout != 0:
t0 = self.time()
event_list = self._selector.select(timeout)
dt = self.time() - t0
if dt >= 1.0:
level = logging.INFO
else:
level = logging.DEBUG
nevent = len(event_list)
if timeout is None:
logger.log(level, 'poll took %.3f ms: %s events',
dt * 1e3, nevent)
elif nevent:
logger.log(level,
'poll %.3f ms took %.3f ms: %s events',
timeout * 1e3, dt * 1e3, nevent)
elif dt >= 1.0:
logger.log(level,
'poll %.3f ms took %.3f ms: timeout',
timeout * 1e3, dt * 1e3)
else:
event_list = self._selector.select(timeout)
self._process_events(event_list)
# Обработать готовые колбэки 'later'.
end_time = self.time() + self._clock_resolution
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_wrapper(self, enabled):
try:
set_wrapper = sys.set_coroutine_wrapper
get_wrapper = sys.get_coroutine_wrapper
except AttributeError:
return
enabled = bool(enabled)
if self._coroutine_wrapper_set == enabled:
return
wrapper = coroutines.debug_wrapper
current_wrapper = get_wrapper()
if enabled:
if current_wrapper not in (None, wrapper):
warnings.warn(
"loop.set_debug(True): cannot set debug coroutine "
"wrapper; another wrapper is already set %r" %
current_wrapper, RuntimeWarning)
else:
set_wrapper(wrapper)
self._coroutine_wrapper_set = True
else:
if current_wrapper not in (None, wrapper):
warnings.warn(
"loop.set_debug(False): cannot unset debug coroutine "
"wrapper; another wrapper was set %r" %
current_wrapper, RuntimeWarning)
else:
set_wrapper(None)
self._coroutine_wrapper_set = False
def get_debug(self):
return self._debug
def set_debug(self, enabled):
self._debug = enabled
if self.is_running():
self._set_coroutine_wrapper(enabled)