events.py
python
"""Цикл событий и политика цикла событий."""
# Содержит код из https://github.com/MagicStack/uvloop/tree/v0.16.0
# SPDX-License-Identifier: PSF-2.0 AND (MIT OR Apache-2.0)
# SPDX-FileCopyrightText: Copyright (c) 2015-2021 MagicStack Inc. http://magic.io
__all__ = (
"AbstractEventLoop",
"AbstractServer",
"Handle",
"TimerHandle",
"get_event_loop",
"set_event_loop",
"new_event_loop",
"_set_running_loop",
"get_running_loop",
"_get_running_loop",
)
import contextvars
import os
import signal
import socket
import subprocess
import sys
import threading
from . import format_helpers
class Handle:
"""Объект, возвращаемый методами регистрации колбэков."""
__slots__ = ('_callback', '_args', '_cancelled', '_loop',
'_source_traceback', '_repr', '__weakref__',
'_context')
def __init__(self, callback, args, loop, context=None):
if context is None:
context = contextvars.copy_context()
self._context = context
self._loop = loop
self._callback = callback
self._args = args
self._cancelled = False
self._repr = None
if self._loop.get_debug():
self._source_traceback = format_helpers.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_helpers._format_callback_source(
self._callback, self._args,
debug=self._loop.get_debug()))
if self._source_traceback:
frame = self._source_traceback[-1]
info.append(f'created at {frame[0]}:{frame[1]}')
return info
def __repr__(self):
if self._repr is not None:
return self._repr
info = self._repr_info()
return '<{}>'.format(' '.join(info))
def get_context(self):
return self._context
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 cancelled(self):
return self._cancelled
def _run(self):
try:
self._context.run(self._callback, *self._args)
except (SystemExit, KeyboardInterrupt):
raise
except BaseException as exc:
cb = format_helpers._format_callback_source(
self._callback, self._args,
debug=self._loop.get_debug())
msg = f'Exception in callback {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 # Необходимо для разрыва циклов при возникновении исключения.
# _ThreadSafeHandle используется для колбэков, запланированных с помощью call_soon_threadsafe
# и является потокобезопасным, в отличие от Handle, который не является потокобезопасным.
class _ThreadSafeHandle(Handle):
__slots__ = ('_lock',)
def __init__(self, callback, args, loop, context=None):
super().__init__(callback, args, loop, context)
self._lock = threading.RLock()
def cancel(self):
with self._lock:
return super().cancel()
def cancelled(self):
with self._lock:
return super().cancelled()
def _run(self):
# Цикл событий проверяет отмену без удержания блокировки
# Возможно, что дескриптор будет отменён после проверки
# но до вызова колбэка, поэтому проверьте его ещё раз после получения
# блокировки и вернитесь без вызова колбэка, если он отменён.
with self._lock:
if self._cancelled:
return
return super()._run()
class TimerHandle(Handle):
"""Объект, возвращаемый методами регистрации таймированных колбэков."""
__slots__ = ['_scheduled', '_when']
def __init__(self, when, callback, args, loop, context=None):
super().__init__(callback, args, loop, context)
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, f'when={self._when}')
return info
def __hash__(self):
return hash(self._when)
def __lt__(self, other):
if isinstance(other, TimerHandle):
return self._when < other._when
return NotImplemented
def __le__(self, other):
if isinstance(other, TimerHandle):
return self._when < other._when or self.__eq__(other)
return NotImplemented
def __gt__(self, other):
if isinstance(other, TimerHandle):
return self._when > other._when
return NotImplemented
def __ge__(self, other):
if isinstance(other, TimerHandle):
return self._when > other._when or self.__eq__(other)
return NotImplemented
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 cancel(self):
if not self._cancelled:
self._loop._timer_handle_cancelled(self)
super().cancel()
def when(self):
"""Возвращает запланированное время колбэка.
Время — это абсолютная временная метка, использующая ту же
привязку ко времени, что и loop.time().
"""
return self._when
class AbstractServer:
"""Абстрактный сервер, возвращаемый create_server()."""
def close(self):
"""Прекратить обслуживание. Существующие соединения остаются открытыми."""
raise NotImplementedError
def close_clients(self):
"""Закрыть все активные соединения."""
raise NotImplementedError
def abort_clients(self):
"""Закрыть все активные соединения немедленно."""
raise NotImplementedError
def get_loop(self):
"""Получить цикл событий, к которому прикреплён объект Server."""
raise NotImplementedError
def is_serving(self):
"""Возвращает True, если сервер принимает соединения."""
raise NotImplementedError
async def start_serving(self):
"""Начать приём соединений.
Этот метод является идемпотентным, поэтому его можно вызывать,
когда сервер уже обслуживает.
"""
raise NotImplementedError
async def serve_forever(self):
"""Начать приём соединений до отмены корутины.
Сервер закрывается при отмене корутины.
"""
raise NotImplementedError
async def wait_closed(self):
"""Корутина для ожидания закрытия сервиса."""
raise NotImplementedError
async def __aenter__(self):
return self
async def __aexit__(self, *exc):
self.close()
await self.wait_closed()
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
async def shutdown_asyncgens(self):
"""Завершает все активные асинхронные генераторы."""
raise NotImplementedError
async def shutdown_default_executor(self):
"""Планирует завершение исполнителя по умолчанию."""
raise NotImplementedError
# Методы, планирующие колбэки. Все они возвращают объекты Handle.
def _timer_handle_cancelled(self, handle):
"""Уведомление о том, что TimerHandle был отменён."""
raise NotImplementedError
def call_soon(self, callback, *args, context=None):
return self.call_later(0, callback, *args, context=context)
def call_later(self, delay, callback, *args, context=None):
raise NotImplementedError
def call_at(self, when, callback, *args, context=None):
raise NotImplementedError
def time(self):
raise NotImplementedError
def create_future(self):
raise NotImplementedError
# Метод, планирующий корутину: создать задачу.
def create_task(self, coro, **kwargs):
raise NotImplementedError
# Методы для взаимодействия с потоками.
def call_soon_threadsafe(self, callback, *args, context=None):
raise NotImplementedError
def run_in_executor(self, executor, func, *args):
raise NotImplementedError
def set_default_executor(self, executor):
raise NotImplementedError
# Методы сетевого ввода-вывода, возвращающие объекты Future.
async def getaddrinfo(self, host, port, *,
family=0, type=0, proto=0, flags=0):
raise NotImplementedError
async def getnameinfo(self, sockaddr, flags=0):
raise NotImplementedError
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):
raise NotImplementedError
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 и 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.
keep_alive, установленный в True, поддерживает соединения активными, включая периодическую передачу сообщений.
ssl_handshake_timeout — время в секундах, в течение которого SSL-сервер будет ждать завершения SSL-рукопожатия перед разрывом соединения. По умолчанию 60 с.
ssl_shutdown_timeout — время в секундах, в течение которого SSL-сервер будет ждать завершения процедуры завершения SSL перед разрывом соединения. По умолчанию 30 с.
start_serving, установленный в True (по умолчанию), заставляет созданный сервер немедленно начать приём подключений. Если установлен в False, пользователь должен await Server.start_serving() или Server.serve_forever(), чтобы сервер начал принимать подключения.
"""
raise NotImplementedError
async def sendfile(self, transport, file, offset=0, count=None,
*, fallback=True):
"""Отправить файл через транспорт.
Вернуть количество отправленных байт.
"""
raise NotImplementedError
async def start_tls(self, transport, protocol, sslcontext, *,
server_side=False,
server_hostname=None,
ssl_handshake_timeout=None,
ssl_shutdown_timeout=None):
"""Обновить транспорт до TLS.
Вернуть новый транспорт, который *protocol* должен начать использовать
немедленно.
"""
raise NotImplementedError
async def create_unix_connection(
self, protocol_factory, path=None, *,
ssl=None, sock=None,
server_hostname=None,
ssl_handshake_timeout=None,
ssl_shutdown_timeout=None):
raise NotImplementedError
async def create_unix_server(
self, protocol_factory, path=None, *,
sock=None, backlog=100, ssl=None,
ssl_handshake_timeout=None,
ssl_shutdown_timeout=None,
start_serving=True):
"""Корутина, которая создаёт сервер доменных сокетов UNIX.
Возвращаемое значение — объект Server, который можно использовать для остановки службы.
path — это строка, представляющая путь в файловой системе для привязки серверного сокета.
sock может быть указан для использования существующего объекта сокета.
backlog — максимальное количество ожидающих подключений, передаваемое в listen() (по умолчанию 100).
ssl может быть установлен в SSLContext для включения SSL поверх принятых соединений.
ssl_handshake_timeout — время в секундах, в течение которого SSL-сервер будет ждать завершения SSL-рукопожатия (по умолчанию 60 с).
ssl_shutdown_timeout — время в секундах, в течение которого SSL-сервер будет ждать завершения процедуры завершения SSL (по умолчанию 30 с).
start_serving, установленный в True (по умолчанию), заставляет созданный сервер немедленно начать приём подключений. Если установлен в False, пользователь должен await Server.start_serving() или Server.serve_forever(), чтобы сервер начал принимать подключения.
"""
raise NotImplementedError
async def connect_accepted_socket(
self, protocol_factory, sock,
*, ssl=None,
ssl_handshake_timeout=None,
ssl_shutdown_timeout=None):
"""Обработать принятое соединение.
Используется серверами, которые принимают соединения за пределами
asyncio, но используют asyncio для обработки соединений.
Этот метод является корутиной. После завершения корутина
возвращает пару (transport, protocol).
"""
raise NotImplementedError
async 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):
"""Корутина, которая создаёт конечную точку дейтаграмм.
Этот метод попытается установить конечную точку в фоновом режиме. При успехе корутина возвращает пару (транспорт, протокол).
protocol_factory должен быть вызываемым объектом, возвращающим экземпляр протокола.
Семейство сокетов AF_INET, socket.AF_INET6 или socket.AF_UNIX в зависимости от host (или family, если указан), тип сокета SOCK_DGRAM.
reuse_address сообщает ядру повторно использовать локальный сокет в состоянии TIME_WAIT, не дожидаясь истечения его тайм-аута. Если не указано, автоматически устанавливается в True на UNIX.
reuse_port сообщает ядру разрешить привязку этой конечной точки к тому же порту, что и другие существующие конечные точки, при условии, что все они устанавливают этот флаг при создании. Эта опция не поддерживается в Windows и некоторых UNIX. Если константа :py:data:`~socket.SO_REUSEPORT` не определена, то эта возможность не поддерживается.
allow_broadcast сообщает ядру разрешить этой конечной точке отправлять сообщения на широковещательный адрес.
sock может быть указан для использования существующего объекта сокета.
"""
raise NotImplementedError
# Каналы и подпроцессы.
async def connect_read_pipe(self, protocol_factory, pipe):
"""Зарегистрировать канал чтения в цикле событий. Установить канал в неблокирующий режим.
protocol_factory должен создавать объект с интерфейсом Protocol.
pipe — файлоподобный объект.
Вернуть пару (transport, protocol), где transport поддерживает
интерфейс ReadTransport."""
# Причина принимать объект, подобный файлу, а не просто файловый дескриптор
# это: нам нужно владеть каналом и закрыть его при завершении транспорта
# Могут возникнуть сложные ошибки, если передавать f.fileno(),
# закрыть fd в транспорте канала, а затем закрыть f и наоборот.
raise NotImplementedError
async def connect_write_pipe(self, protocol_factory, pipe):
"""Зарегистрировать канал записи в цикле событий.
protocol_factory должен создавать объект с интерфейсом BaseProtocol.
Канал — объект, подобный файлу, уже переведённый в неблокирующий режим.
Возвращает пару (транспорт, протокол), где транспорт поддерживает интерфейс WriteTransport."""
# Причина принимать объект, подобный файлу, а не просто файловый дескриптор
# это: нам нужно владеть каналом и закрыть его при завершении транспорта
# Могут возникнуть сложные ошибки, если передавать f.fileno(),
# закрыть fd в транспорте канала, а затем закрыть f и наоборот.
raise NotImplementedError
async def subprocess_shell(self, protocol_factory, cmd, *,
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
**kwargs):
raise NotImplementedError
async 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.
async def sock_recv(self, sock, nbytes):
raise NotImplementedError
async def sock_recv_into(self, sock, buf):
raise NotImplementedError
async def sock_recvfrom(self, sock, bufsize):
raise NotImplementedError
async def sock_recvfrom_into(self, sock, buf, nbytes=0):
raise NotImplementedError
async def sock_sendall(self, sock, data):
raise NotImplementedError
async def sock_sendto(self, sock, data, address):
raise NotImplementedError
async def sock_connect(self, sock, address):
raise NotImplementedError
async def sock_accept(self, sock):
raise NotImplementedError
async def sock_sendfile(self, sock, file, offset=0, count=None,
*, fallback=None):
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 _Local(threading.local):
_loop = None
_local = _Local()
# TLS для работающего цикла событий, используется _get_running_loop.
class _RunningLoop(threading.local):
loop_pid = (None, None)
_running_loop = _RunningLoop()
def get_running_loop():
"""Возвращает работающий цикл событий. Вызывает RuntimeError, если такового нет.
Эта функция специфична для потока.
"""
# ПРИМЕЧАНИЕ: эта функция реализована на C (см. _asynciomodule.c)
loop = _get_running_loop()
if loop is None:
raise RuntimeError('no running event loop')
return loop
def _get_running_loop():
"""Возвращает выполняющийся цикл событий или None.
Это низкоуровневая функция, предназначенная для использования циклами событий.
Эта функция привязана к потоку.
"""
# ПРИМЕЧАНИЕ: эта функция реализована на C (см. _asynciomodule.c)
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):
"""Устанавливает выполняющийся цикл событий.
Это низкоуровневая функция, предназначенная для использования циклами событий.
Эта функция привязана к потоку.
"""
# ПРИМЕЧАНИЕ: эта функция реализована на C (см. _asynciomodule.c)
_running_loop.loop_pid = (loop, os.getpid())
def _get_event_loop():
"""Возвращает цикл событий, установленный для текущего потока.
Вызывает RuntimeError, если для текущего потока не установлен цикл
событий. Это медленный путь get_event_loop(); работающий цикл
проверяется вызывающей стороной.
"""
loop = _local._loop
if loop is None:
raise RuntimeError('There is no current event loop in thread %r.'
% threading.current_thread().name)
return loop
def get_event_loop():
"""Возвращает цикл событий asyncio.
При вызове из корутины или колбэка (например, запланированного с помощью
call_soon или аналогичного API), эта функция всегда возвращает работающий
цикл событий.
Если работающий цикл событий не установлен, функция вернет цикл,
установленный через ``set_event_loop()``, или вызовет RuntimeError,
если цикл не был установлен.
"""
# ПРИМЕЧАНИЕ: эта функция реализована на C (см. _asynciomodule.c)
current_loop = _get_running_loop()
if current_loop is not None:
return current_loop
return _get_event_loop()
def set_event_loop(loop):
"""Устанавливает цикл событий для текущего потока в loop.
Если loop равен None, текущий цикл событий сбрасывается.
"""
if loop is not None and not isinstance(loop, AbstractEventLoop):
raise TypeError(f"loop must be an instance of AbstractEventLoop or None, not '{type(loop).__name__}'")
_local._loop = loop
def new_event_loop():
"""Создать и вернуть новый объект цикла событий."""
if sys.platform == 'win32':
from .windows_events import EventLoop
else:
from .unix_events import EventLoop
return EventLoop()
# Псевдоним для чисто Python-реализаций в целях тестирования.
_py__get_running_loop = _get_running_loop
_py__set_running_loop = _set_running_loop
_py_get_running_loop = get_running_loop
_py_get_event_loop = get_event_loop
try:
# get_event_loop() — одна из самых часто вызываемых
# функции в asyncio. Реализация на чистом Python
# примерно в 4 раза медленнее, чем ускоренная версия на C.
from _asyncio import (_get_running_loop, _set_running_loop,
get_running_loop, get_event_loop)
except ImportError:
pass
else:
# Псевдонимы для реализаций на C для тестирования.
_c__get_running_loop = _get_running_loop
_c__set_running_loop = _set_running_loop
_c_get_running_loop = get_running_loop
_c_get_event_loop = get_event_loop
if hasattr(os, 'fork'):
def on_fork():
# Сбросьте цикл событий и wakeupfd в дочернем процессе, созданном через fork.
global _local
_local = _Local()
_set_running_loop(None)
signal.set_wakeup_fd(-1)
os.register_at_fork(after_in_child=on_fork)