events.py
python
"""Цикл событий и политика цикла событий."""
__all__ = ['AbstractEventLoopPolicy',
'AbstractEventLoop', 'AbstractServer',
'Handle', 'TimerHandle',
'get_event_loop_policy', 'set_event_loop_policy',
'get_event_loop', 'set_event_loop', 'new_event_loop',
'get_child_watcher', 'set_child_watcher',
'_set_running_loop', '_get_running_loop',
]
import functools
import inspect
import os
import reprlib
import socket
import subprocess
import sys
import threading
import traceback
from . import compat
from . import constants
def _get_function_source(func):
if compat.PY34:
func = inspect.unwrap(func)
elif hasattr(func, '__wrapped__'):
func = func.__wrapped__
if inspect.isfunction(func):
code = func.__code__
return (code.co_filename, code.co_firstlineno)
if isinstance(func, functools.partial):
return _get_function_source(func.func)
if compat.PY34 and isinstance(func, functools.partialmethod):
return _get_function_source(func.func)
return None
def _format_args_and_kwargs(args, kwargs):
"""Форматирование аргументов функции и именованных аргументов.
Особый случай для одного параметра: ('hello',) форматируется как ('hello').
"""
# использовать reprlib для ограничения длины вывода
items = []
if args:
items.extend(reprlib.repr(arg) for arg in args)
if kwargs:
items.extend('{}={}'.format(k, reprlib.repr(v))
for k, v in kwargs.items())
return '(' + ', '.join(items) + ')'
def _format_callback(func, args, kwargs, suffix=''):
if isinstance(func, functools.partial):
suffix = _format_args_and_kwargs(args, kwargs) + suffix
return _format_callback(func.func, func.args, func.keywords, suffix)
if hasattr(func, '__qualname__') and func.__qualname__:
func_repr = func.__qualname__
elif hasattr(func, '__name__') and func.__name__:
func_repr = func.__name__
else:
func_repr = repr(func)
func_repr += _format_args_and_kwargs(args, kwargs)
if suffix:
func_repr += suffix
return func_repr
def _format_callback_source(func, args):
func_repr = _format_callback(func, args, None)
source = _get_function_source(func)
if source:
func_repr += ' at %s:%s' % source
return func_repr
def extract_stack(f=None, limit=None):
"""Замена traceback.extract_stack(), которая выполняет только
необходимую работу для режима отладки asyncio.
"""
if f is None:
f = sys._getframe().f_back
if limit is None:
# Ограничьте объём работы разумным значением, так как extract_stack()
# может вызываться для каждой корутины и future в режиме отладки.
limit = constants.DEBUG_STACK_DEPTH
stack = traceback.StackSummary.extract(traceback.walk_stack(f),
limit=limit,
lookup_lines=False)
stack.reverse()
return stack
class Handle:
"""Объект, возвращаемый методами регистрации колбэков."""
__slots__ = ('_callback', '_args', '_cancelled', '_loop',
'_source_traceback', '_repr', '__weakref__')
def __init__(self, callback, args, loop):
self._loop = loop
self._callback = callback
self._args = args
self._cancelled = False
self._repr = None
if self._loop.get_debug():
self._source_traceback = extract_stack(sys._getframe(1))
else:
self._source_traceback = None
def _repr_info(self):
info = [self.__class__.__name__]
if self._cancelled:
info.append('cancelled')
if self._callback is not None:
info.append(_format_callback_source(self._callback, self._args))
if self._source_traceback:
frame = self._source_traceback[-1]
info.append('created at %s:%s' % (frame[0], frame[1]))
return info
def __repr__(self):
if self._repr is not None:
return self._repr
info = self._repr_info()
return '<%s>' % ' '.join(info)
def cancel(self):
if not self._cancelled:
self._cancelled = True
if self._loop.get_debug():
# Сохраняйте представление в режиме отладки, чтобы сохранить колбэк и
# параметры. Например, для записи предупреждения
# "Executing <Handle...> took 2.5 second"
self._repr = repr(self)
self._callback = None
self._args = None
def _run(self):
try:
self._callback(*self._args)
except Exception as exc:
cb = _format_callback_source(self._callback, self._args)
msg = 'Exception in callback {}'.format(cb)
context = {
'message': msg,
'exception': exc,
'handle': self,
}
if self._source_traceback:
context['source_traceback'] = self._source_traceback
self._loop.call_exception_handler(context)
self = None # Необходимо для разрыва циклов при возникновении исключения.
class TimerHandle(Handle):
"""Объект, возвращаемый методами регистрации таймированных колбэков."""
__slots__ = ['_scheduled', '_when']
def __init__(self, when, callback, args, loop):
assert when is not None
super().__init__(callback, args, loop)
if self._source_traceback:
del self._source_traceback[-1]
self._when = when
self._scheduled = False
def _repr_info(self):
info = super()._repr_info()
pos = 2 if self._cancelled else 1
info.insert(pos, 'when=%s' % self._when)
return info
def __hash__(self):
return hash(self._when)
def __lt__(self, other):
return self._when < other._when
def __le__(self, other):
if self._when < other._when:
return True
return self.__eq__(other)
def __gt__(self, other):
return self._when > other._when
def __ge__(self, other):
if self._when > other._when:
return True
return self.__eq__(other)
def __eq__(self, other):
if isinstance(other, TimerHandle):
return (self._when == other._when and
self._callback == other._callback and
self._args == other._args and
self._cancelled == other._cancelled)
return NotImplemented
def __ne__(self, other):
equal = self.__eq__(other)
return NotImplemented if equal is NotImplemented else not equal
def cancel(self):
if not self._cancelled:
self._loop._timer_handle_cancelled(self)
super().cancel()
class AbstractServer:
"""Абстрактный сервер, возвращаемый create_server()."""
def close(self):
"""Прекратить обслуживание. Существующие соединения остаются открытыми."""
return NotImplemented
def wait_closed(self):
"""Корутина для ожидания закрытия сервиса."""
return NotImplemented
class AbstractEventLoop:
"""Абстрактный цикл событий."""
# Запуск и остановка цикла событий.
def run_forever(self):
"""Запускать цикл событий, пока не вызвана stop()."""
raise NotImplementedError
def run_until_complete(self, future):
"""Запустить цикл событий до завершения Future.
Вернуть результат Future или возбудить его исключение.
"""
raise NotImplementedError
def stop(self):
"""Остановить цикл событий как можно скорее.
Насколько именно скоро — зависит от реализации, но
не должно быть запланировано новых колбэков ввода-вывода.
"""
raise NotImplementedError
def is_running(self):
"""Вернуть, запущен ли в данный момент цикл событий."""
raise NotImplementedError
def is_closed(self):
"""Возвращает True, если цикл событий был закрыт."""
raise NotImplementedError
def close(self):
"""Закрыть цикл.
Цикл не должен быть запущен.
Эта операция идемпотентна и необратима.
После этого вызова не следует вызывать другие методы.
"""
raise NotImplementedError
def shutdown_asyncgens(self):
"""Завершает все активные асинхронные генераторы."""
raise NotImplementedError
# Методы, планирующие колбэки. Все они возвращают объекты Handle.
def _timer_handle_cancelled(self, handle):
"""Уведомление о том, что TimerHandle был отменён."""
raise NotImplementedError
def call_soon(self, callback, *args):
return self.call_later(0, callback, *args)
def call_later(self, delay, callback, *args):
raise NotImplementedError
def call_at(self, when, callback, *args):
raise NotImplementedError
def time(self):
raise NotImplementedError
def create_future(self):
raise NotImplementedError
# Метод, планирующий корутину: создать задачу.
def create_task(self, coro):
raise NotImplementedError
# Методы для взаимодействия с потоками.
def call_soon_threadsafe(self, callback, *args):
raise NotImplementedError
def run_in_executor(self, executor, func, *args):
raise NotImplementedError
def set_default_executor(self, executor):
raise NotImplementedError
# Методы сетевого ввода-вывода, возвращающие объекты Future.
def getaddrinfo(self, host, port, *, family=0, type=0, proto=0, flags=0):
raise NotImplementedError
def getnameinfo(self, sockaddr, flags=0):
raise NotImplementedError
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):
raise NotImplementedError
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 и port.
Возвращаемое значение — объект Server, который можно использовать для остановки
службы.
Если host является пустой строкой или None, подразумеваются все интерфейсы,
и будет возвращён список нескольких сокетов (скорее всего,
один для IPv4 и другой для IPv6). Параметр host также может быть
последовательностью (например, списком) хостов для привязки.
family может быть установлен в AF_INET или AF_INET6, чтобы заставить
сокет использовать IPv4 или IPv6. Если не задан, будет определён
из host (по умолчанию AF_UNSPEC).
flags — битовая маска для getaddrinfo().
sock может быть опционально указан для использования уже существующего
объекта сокета.
backlog — максимальное количество ожидающих соединений, передаваемое в
listen() (по умолчанию 100).
ssl может быть установлен в SSLContext для включения SSL на
принимаемых соединениях.
reuse_address указывает ядру повторно использовать локальный сокет в
состоянии TIME_WAIT, не дожидаясь истечения его естественного тайм-аута.
Если не указано, автоматически устанавливается в True на
UNIX.
reuse_port указывает ядру разрешить привязку этой конечной точки к
тому же порту, к которому привязаны другие существующие конечные точки, при условии,
что все они устанавливают этот флаг при создании. Эта опция не
поддерживается на Windows.
"""
raise NotImplementedError
def create_unix_connection(self, protocol_factory, path, *,
ssl=None, sock=None,
server_hostname=None):
raise NotImplementedError
def create_unix_server(self, protocol_factory, path, *,
sock=None, backlog=100, ssl=None):
"""Корутина, создающая сервер сокетов домена UNIX.
Возвращаемое значение — объект Server, который можно использовать для остановки
службы.
path — это строка, представляющая путь в файловой системе для привязки
серверного сокета.
sock может быть опционально указан для использования уже существующего
объекта сокета.
backlog — максимальное количество ожидающих соединений, передаваемое в
listen() (по умолчанию 100).
ssl может быть установлен в SSLContext для включения SSL на
принимаемых соединениях.
"""
raise NotImplementedError
def create_datagram_endpoint(self, protocol_factory,
local_addr=None, remote_addr=None, *,
family=0, proto=0, flags=0,
reuse_address=None, reuse_port=None,
allow_broadcast=None, sock=None):
"""Корутина, создающая датаграммную конечную точку.
Этот метод попытается установить конечную точку в фоновом режиме.
При успешном выполнении корутина возвращает пару (transport, protocol).
protocol_factory должен быть вызываемым объектом, возвращающим экземпляр протокола.
Семейство сокетов AF_INET или socket.AF_INET6 в зависимости от host (или
family, если указан), тип сокета SOCK_DGRAM.
reuse_address указывает ядру повторно использовать локальный сокет в
состоянии TIME_WAIT, не дожидаясь истечения его естественного тайм-аута.
Если не указано, автоматически устанавливается в True на
UNIX.
reuse_port указывает ядру разрешить привязку этой конечной точки к
тому же порту, к которому привязаны другие существующие конечные точки, при условии,
что все они устанавливают этот флаг при создании. Эта опция не
поддерживается на Windows и некоторых UNIX. Если константа
:py:data:`~socket.SO_REUSEPORT` не определена, то эта
возможность не поддерживается.
allow_broadcast указывает ядру разрешить этой конечной точке отправлять
сообщения на широковещательный адрес.
sock может быть опционально указан для использования уже существующего
объекта сокета.
"""
raise NotImplementedError
# Каналы и подпроцессы.
def connect_read_pipe(self, protocol_factory, pipe):
"""Зарегистрировать канал чтения в цикле событий. Установить канал в неблокирующий режим.
protocol_factory должен создавать объект с интерфейсом Protocol.
pipe — файлоподобный объект.
Вернуть пару (transport, protocol), где transport поддерживает
интерфейс ReadTransport."""
# Причина принимать объект, подобный файлу, а не просто файловый дескриптор
# это: нам нужно владеть каналом и закрыть его при завершении транспорта
# Могут возникнуть сложные ошибки, если передавать f.fileno(),
# закрыть fd в транспорте канала, затем закрыть f и наоборот.
raise NotImplementedError
def connect_write_pipe(self, protocol_factory, pipe):
"""Зарегистрировать канал записи в цикле событий.
protocol_factory должен создавать объект с интерфейсом BaseProtocol.
Канал является объектом, подобным файлу, уже переведенным в неблокирующий режим.
Возвращает пару (transport, protocol), где transport поддерживает интерфейс WriteTransport."""
# Причина принимать объект, подобный файлу, а не просто файловый дескриптор
# это: нам нужно владеть каналом и закрыть его при завершении транспорта
# Могут возникнуть сложные ошибки, если передавать f.fileno(),
# закрыть fd в транспорте канала, затем закрыть f и наоборот.
raise NotImplementedError
def subprocess_shell(self, protocol_factory, cmd, *, stdin=subprocess.PIPE,
stdout=subprocess.PIPE, stderr=subprocess.PIPE,
**kwargs):
raise NotImplementedError
def subprocess_exec(self, protocol_factory, *args, stdin=subprocess.PIPE,
stdout=subprocess.PIPE, stderr=subprocess.PIPE,
**kwargs):
raise NotImplementedError
# Методы регистрации колбэков на основе готовности.
# Методы add_*() возвращают None.
# Методы remove_*() возвращают True, если что-то было удалено,
# False, если нечего было удалять.
def add_reader(self, fd, callback, *args):
raise NotImplementedError
def remove_reader(self, fd):
raise NotImplementedError
def add_writer(self, fd, callback, *args):
raise NotImplementedError
def remove_writer(self, fd):
raise NotImplementedError
# Асинхронные методы ввода-вывода, возвращающие Future.
def sock_recv(self, sock, nbytes):
raise NotImplementedError
def sock_sendall(self, sock, data):
raise NotImplementedError
def sock_connect(self, sock, address):
raise NotImplementedError
def sock_accept(self, sock):
raise NotImplementedError
# Обработка сигналов.
def add_signal_handler(self, sig, callback, *args):
raise NotImplementedError
def remove_signal_handler(self, sig):
raise NotImplementedError
# Фабрика задач.
def set_task_factory(self, factory):
raise NotImplementedError
def get_task_factory(self):
raise NotImplementedError
# Обработчики ошибок.
def get_exception_handler(self):
raise NotImplementedError
def set_exception_handler(self, handler):
raise NotImplementedError
def default_exception_handler(self, context):
raise NotImplementedError
def call_exception_handler(self, context):
raise NotImplementedError
# Управление флагом отладки.
def get_debug(self):
raise NotImplementedError
def set_debug(self, enabled):
raise NotImplementedError
class AbstractEventLoopPolicy:
"""Абстрактная политика доступа к циклу событий."""
def get_event_loop(self):
"""Получить цикл событий для текущего контекста.
Возвращает объект цикла событий, реализующий интерфейс BaseEventLoop,
или вызывает исключение, если цикл событий не был установлен для
текущего контекста, и текущая политика не предусматривает его создание.
Никогда не должен возвращать None."""
raise NotImplementedError
def set_event_loop(self, loop):
"""Устанавливает цикл событий для текущего контекста равным loop."""
raise NotImplementedError
def new_event_loop(self):
"""Создаёт и возвращает новый объект цикла событий в соответствии с
правилами данной политики. Если необходимо сделать этот цикл событий
циклом для текущего контекста, необходимо явно вызвать set_event_loop."""
raise NotImplementedError
# Обработка дочерних процессов (только Unix).
def get_child_watcher(self):
"Возвращает наблюдатель за дочерними процессами."
raise NotImplementedError
def set_child_watcher(self, watcher):
"""Устанавливает наблюдатель за дочерними процессами."""
raise NotImplementedError
class BaseDefaultEventLoopPolicy(AbstractEventLoopPolicy):
"""Реализация политики по умолчанию для доступа к циклу событий.
В этой политике каждый поток имеет собственный цикл событий. Однако
мы автоматически создаём цикл событий по умолчанию только для
главного потока; остальные потоки по умолчанию не имеют цикла событий.
Другие политики могут иметь иные правила (например, один глобальный
цикл событий, или автоматическое создание цикла событий на поток, или
использование некоторого другого понятия контекста, с которым
ассоциируется цикл событий).
"""
_loop_factory = None
class _Local(threading.local):
_loop = None
_set_called = False
def __init__(self):
self._local = self._Local()
def get_event_loop(self):
"""Получить цикл событий.
Может быть None или экземпляром EventLoop.
"""
if (self._local._loop is None and
not self._local._set_called and
isinstance(threading.current_thread(), threading._MainThread)):
self.set_event_loop(self.new_event_loop())
if self._local._loop is None:
raise RuntimeError('There is no current event loop in thread %r.'
% threading.current_thread().name)
return self._local._loop
def set_event_loop(self, loop):
"""Устанавливает цикл событий."""
self._local._set_called = True
assert loop is None or isinstance(loop, AbstractEventLoop)
self._local._loop = loop
def new_event_loop(self):
"""Создаёт новый цикл событий.
Для того чтобы сделать его текущим, необходимо вызвать
set_event_loop().
"""
return self._loop_factory()
# Политика цикла событий. Сама политика всегда глобальна, даже если
# правила политики гласят, что цикл событий существует на поток (или другой
# понятие контекста). Политика по умолчанию устанавливается при первом
# вызове get_event_loop_policy().
_event_loop_policy = None
# Блокировка для защиты создания политики цикла событий на лету.
_lock = threading.Lock()
# TLS для работающего цикла событий, используется _get_running_loop.
class _RunningLoop(threading.local):
loop_pid = (None, None)
_running_loop = _RunningLoop()
def _get_running_loop():
"""Возвращает выполняющийся цикл событий или None.
Это низкоуровневая функция, предназначенная для использования циклами событий.
Эта функция привязана к потоку.
"""
running_loop, pid = _running_loop.loop_pid
if running_loop is not None and pid == os.getpid():
return running_loop
def _set_running_loop(loop):
"""Устанавливает выполняющийся цикл событий.
Это низкоуровневая функция, предназначенная для использования циклами событий.
Эта функция привязана к потоку.
"""
_running_loop.loop_pid = (loop, os.getpid())
def _init_event_loop_policy():
global _event_loop_policy
with _lock:
if _event_loop_policy is None: # pragma: no branch
from . import DefaultEventLoopPolicy
_event_loop_policy = DefaultEventLoopPolicy()
def get_event_loop_policy():
"""Возвращает текущую политику цикла событий."""
if _event_loop_policy is None:
_init_event_loop_policy()
return _event_loop_policy
def set_event_loop_policy(policy):
"""Устанавливает текущую политику цикла событий.
Если policy равен None, восстанавливается политика по умолчанию."""
global _event_loop_policy
assert policy is None or isinstance(policy, AbstractEventLoopPolicy)
_event_loop_policy = policy
def get_event_loop():
"""Возвращает цикл событий asyncio.
При вызове из корутины или колбэка (например, запланированного с помощью call_soon
или аналогичного API) эта функция всегда возвращает выполняющийся цикл событий.
Если выполняющийся цикл событий не установлен, функция вернёт
результат вызова `get_event_loop_policy().get_event_loop()`.
"""
current_loop = _get_running_loop()
if current_loop is not None:
return current_loop
return get_event_loop_policy().get_event_loop()
def set_event_loop(loop):
"""Эквивалентно вызову get_event_loop_policy().set_event_loop(loop)."""
get_event_loop_policy().set_event_loop(loop)
def new_event_loop():
"""Эквивалентно вызову get_event_loop_policy().new_event_loop()."""
return get_event_loop_policy().new_event_loop()
def get_child_watcher():
"""Эквивалентно вызову get_event_loop_policy().get_child_watcher()."""
return get_event_loop_policy().get_child_watcher()
def set_child_watcher(watcher):
"""Эквивалентно вызову
get_event_loop_policy().set_child_watcher(watcher)."""
return get_event_loop_policy().set_child_watcher(watcher)