futures.py
python
"""Класс Future, аналогичный описанному в PEP 3148."""
__all__ = (
'Future', 'wrap_future', 'isfuture',
'future_add_to_awaited_by', 'future_discard_from_awaited_by',
)
import concurrent.futures
import contextvars
import logging
import sys
from types import GenericAlias
from . import base_futures
from . import events
from . import exceptions
from . import format_helpers
isfuture = base_futures.isfuture
_PENDING = base_futures._PENDING
_CANCELLED = base_futures._CANCELLED
_FINISHED = base_futures._FINISHED
STACK_DEBUG = logging.DEBUG - 1 # углублённая отладка
class Future:
"""Этот класс *почти* совместим с concurrent.futures.Future.
Отличия:
- Этот класс не является потокобезопасным.
- result() и exception() не принимают аргумент timeout и
возбуждают исключение, когда future ещё не завершён.
- Колбэки, зарегистрированные с помощью add_done_callback(), всегда вызываются
через call_soon() цикла событий.
- Этот класс не совместим с методами wait() и as_completed()
из пакета concurrent.futures.
"""
# Переменные класса, служащие значениями по умолчанию для переменных экземпляра.
_state = _PENDING
_result = None
_exception = None
_loop = None
_source_traceback = None
_cancel_message = None
# Сохранённый CancelledError для последующей привязки в качестве контекста исключения.
_cancelled_exc = None
# Это поле используется для двух целей:
# - Его наличие является маркером, объявляющим, что класс реализует
# протокол Future (т.е. предполагается совместимость по утиному типированию).
# Значение также должно быть не-None, чтобы позволить подклассу объявить
# что он несовместим, установив это значение в None.
# - Он устанавливается __iter__() ниже, чтобы Task.__step() мог определить
# разницу между
# `await Future()` или `yield from Future()` (правильно) vs.
# `yield Future()` (неправильно).
_asyncio_future_blocking = False
# Используется API capture_call_stack().
__asyncio_awaited_by = None
__log_traceback = False
def __init__(self, *, loop=None):
"""Инициализация future.
Необязательный аргумент event_loop позволяет явно задать объект
цикла событий, используемый future. Если он не предоставлен, future использует
цикл событий по умолчанию.
"""
if self._loop is not None:
raise RuntimeError(f"{self.__class__.__name__} object is already "
"initialized")
if loop is None:
self._loop = events.get_event_loop()
else:
self._loop = loop
self._callbacks = []
if self._loop.get_debug():
self._source_traceback = format_helpers.extract_stack(
sys._getframe(1))
def __repr__(self):
return base_futures._future_repr(self)
def __del__(self):
if not self.__log_traceback:
# set_exception() не был вызван, или result() или exception()
# уже потребили исключение
return
exc = self._exception
context = {
'message':
f'{self.__class__.__name__} exception was never retrieved',
'exception': exc,
'future': self,
}
if self._source_traceback:
context['source_traceback'] = self._source_traceback
self._loop.call_exception_handler(context)
__class_getitem__ = classmethod(GenericAlias)
@property
def _log_traceback(self):
return self.__log_traceback
@_log_traceback.setter
def _log_traceback(self, val):
if val:
raise ValueError('_log_traceback can only be set to False')
self.__log_traceback = False
@property
def _asyncio_awaited_by(self):
if self.__asyncio_awaited_by is None:
return None
return frozenset(self.__asyncio_awaited_by)
def get_loop(self):
"""Возвращает цикл событий, к которому привязан Future."""
loop = self._loop
if loop is None:
raise RuntimeError("Future object is not initialized.")
return loop
def _make_cancelled_error(self):
"""Создаёт CancelledError для возбуждения, если Future будет отменён.
Этот метод следует вызывать только один раз при обработке отмены,
поскольку он стирает сохранённое значение исключения контекста.
"""
if self._cancelled_exc is not None:
exc = self._cancelled_exc
self._cancelled_exc = None
return exc
if self._cancel_message is None:
exc = exceptions.CancelledError()
else:
exc = exceptions.CancelledError(self._cancel_message)
return exc
def cancel(self, msg=None):
"""Отменяет future и планирует колбэки.
Если future уже завершён или отменён, возвращает False. В противном
случае изменяет состояние future на «отменён», планирует колбэки и
возвращает True.
"""
self.__log_traceback = False
if self._state != _PENDING:
return False
self._state = _CANCELLED
self._cancel_message = msg
self.__schedule_callbacks()
return True
def __schedule_callbacks(self):
"""Внутренний метод: просит цикл событий вызвать все колбэки.
Колбэки планируются к вызову как можно скорее. Также очищает список
колбэков.
"""
callbacks = self._callbacks[:]
if not callbacks:
return
self._callbacks[:] = []
for callback, ctx in callbacks:
self._loop.call_soon(callback, self, context=ctx)
def cancelled(self):
"""Возвращает True, если future был отменён."""
return self._state == _CANCELLED
# Не реализуйте running(); см. http://bugs.python.org/issue18699
def done(self):
"""Возвращает True, если future завершён.
Завершён означает, что либо результат/исключение доступны, либо future
был отменён.
"""
return self._state != _PENDING
def result(self):
"""Возвращает результат, который представляет этот future.
Если future был отменён, возбуждается CancelledError. Если результат
future ещё не доступен, возбуждается InvalidStateError. Если future
завершён и имеет установленное исключение, возбуждается это исключение.
"""
if self._state == _CANCELLED:
raise self._make_cancelled_error()
if self._state != _FINISHED:
raise exceptions.InvalidStateError('Result is not ready.')
self.__log_traceback = False
if self._exception is not None:
raise self._exception.with_traceback(self._exception_tb)
return self._result
def exception(self):
"""Возвращает исключение, которое было установлено для этого future.
Исключение (или None, если исключение не установлено) возвращается
только если future завершён. Если future был отменён, возбуждается
CancelledError. Если future ещё не завершён, возбуждается
InvalidStateError.
"""
if self._state == _CANCELLED:
raise self._make_cancelled_error()
if self._state != _FINISHED:
raise exceptions.InvalidStateError('Exception is not set.')
self.__log_traceback = False
return self._exception
def add_done_callback(self, fn, *, context=None):
"""Добавляет колбэк, который будет вызван, когда future станет завершённым.
Колбэк вызывается с одним аргументом — объектом future. Если future
уже завершён на момент вызова, колбэк планируется с помощью call_soon.
"""
if self._state != _PENDING:
self._loop.call_soon(fn, self, context=context)
else:
if context is None:
context = contextvars.copy_context()
self._callbacks.append((fn, context))
# Новый метод, отсутствующий в PEP 3148.
def remove_done_callback(self, fn):
"""Удаляет все экземпляры колбэка из списка «вызвать при завершении».
Возвращает количество удалённых колбэков.
"""
filtered_callbacks = [(f, ctx)
for (f, ctx) in self._callbacks
if f != fn]
removed_count = len(self._callbacks) - len(filtered_callbacks)
if removed_count:
self._callbacks[:] = filtered_callbacks
return removed_count
# Так называемые внутренние методы (примечание: нет set_running_or_notify_cancel()).
def set_result(self, result):
"""Помечает future как завершённый и устанавливает его результат.
Если future уже завершён при вызове этого метода, возбуждается
InvalidStateError.
"""
if self._state != _PENDING:
raise exceptions.InvalidStateError(f'{self._state}: {self!r}')
self._result = result
self._state = _FINISHED
self.__schedule_callbacks()
def set_exception(self, exception):
"""Помечает future как завершённый и устанавливает исключение.
Если future уже завершён при вызове этого метода, возбуждается
InvalidStateError.
"""
if self._state != _PENDING:
raise exceptions.InvalidStateError(f'{self._state}: {self!r}')
if isinstance(exception, type):
exception = exception()
if isinstance(exception, StopIteration):
new_exc = RuntimeError("StopIteration interacts badly with "
"generators and cannot be raised into a "
"Future")
new_exc.__cause__ = exception
new_exc.__context__ = exception
exception = new_exc
self._exception = exception
self._exception_tb = exception.__traceback__
self._state = _FINISHED
self.__schedule_callbacks()
self.__log_traceback = True
def __await__(self):
if not self.done():
self._asyncio_future_blocking = True
yield self # Это указывает Task дождаться завершения.
if not self.done():
raise RuntimeError("await wasn't used with future")
return self.result() # Может также возбуждать исключение.
__iter__ = __await__ # сделать совместимым с 'yield from'.
# Необходим для целей тестирования.
_PyFuture = Future
def _get_loop(fut):
# Пытается вызвать Future.get_loop(), если он доступен.
# В противном случае использует старое свойство '_loop'.
try:
get_loop = fut.get_loop
except AttributeError:
pass
else:
return get_loop()
return fut._loop
def _set_result_unless_cancelled(fut, result):
"""Вспомогательный метод, устанавливающий результат только если future не была отменена."""
if fut.cancelled():
return
fut.set_result(result)
def _convert_future_exc(exc):
exc_class = type(exc)
if exc_class is concurrent.futures.CancelledError:
return exceptions.CancelledError(*exc.args).with_traceback(exc.__traceback__)
elif exc_class is concurrent.futures.InvalidStateError:
return exceptions.InvalidStateError(*exc.args).with_traceback(exc.__traceback__)
else:
return exc
def _set_concurrent_future_state(concurrent, source):
"""Копирует состояние из future в concurrent.futures.Future."""
assert source.done()
if source.cancelled():
concurrent.cancel()
if not concurrent.set_running_or_notify_cancel():
return
exception = source.exception()
if exception is not None:
concurrent.set_exception(_convert_future_exc(exception))
else:
result = source.result()
concurrent.set_result(result)
def _copy_future_state(source, dest):
"""Внутренний вспомогательный метод для копирования состояния из другого Future.
Другой Future должен быть concurrent.futures.Future.
"""
if dest.cancelled():
return
assert not dest.done()
done, cancelled, result, exception = source._get_snapshot()
assert done
if cancelled:
dest.cancel()
elif exception is not None:
dest.set_exception(_convert_future_exc(exception))
else:
dest.set_result(result)
def _chain_future(source, destination):
"""Соединяет два future так, что когда один завершается, то же происходит и с другим.
Результат (или исключение) источника будет скопирован в назначение.
Если назначение отменено, источник также отменяется.
Совместим с asyncio.Future и concurrent.futures.Future.
"""
if not isfuture(source) and not isinstance(source,
concurrent.futures.Future):
raise TypeError('A future is required for source argument')
if not isfuture(destination) and not isinstance(destination,
concurrent.futures.Future):
raise TypeError('A future is required for destination argument')
source_loop = _get_loop(source) if isfuture(source) else None
dest_loop = _get_loop(destination) if isfuture(destination) else None
def _set_state(future, other):
if isfuture(future):
_copy_future_state(other, future)
else:
_set_concurrent_future_state(future, other)
def _call_check_cancel(destination):
if destination.cancelled():
if source_loop is None or source_loop is events._get_running_loop():
source.cancel()
else:
source_loop.call_soon_threadsafe(source.cancel)
def _call_set_state(source):
if (destination.cancelled() and
dest_loop is not None and dest_loop.is_closed()):
return
if dest_loop is None or dest_loop is events._get_running_loop():
_set_state(destination, source)
else:
if dest_loop.is_closed():
return
dest_loop.call_soon_threadsafe(_set_state, destination, source)
destination.add_done_callback(_call_check_cancel)
source.add_done_callback(_call_set_state)
def wrap_future(future, *, loop=None):
"""Оборачивает объект concurrent.futures.Future."""
if isfuture(future):
return future
assert isinstance(future, concurrent.futures.Future), \
f'concurrent.futures.Future is expected, got {future!r}'
if loop is None:
loop = events.get_event_loop()
new_future = loop.create_future()
_chain_future(future, new_future)
return new_future
def future_add_to_awaited_by(fut, waiter, /):
"""Записывает, что `fut` ожидается объектом `waiter`."""
# Ради сохранения минимальности реализации и в предположении,
# что большинство пользователей asyncio используют встроенные объекты Future и Task
# (или их подклассы), мы поддерживаем только родные объекты Future
# и их подклассы.
#
# Подробнее: отслеживание требует хранения зависимости вызывающий-вызываемый
# где-то. Очевидный выбор — хранить эту
# информацию прямо в самом future в специальном атрибуте.
# Это означает, что нам пришлось бы требовать от всех совместимых по утиной типизации
# future реализации определённого атрибута, используемого asyncio для
# ведения учёта. Другое решение — хранить это в
# глобальном словаре. Недостаток в том, что это создаст
# сильные ссылки, и любой сценарий, где вызов "add" не
# сопровождается вызовом "discard", приведёт к утечке памяти.
# Использование WeakDict решило бы эту проблему, но усложнило бы
# код на C (_asynciomodule.c). Суть в том, что
# неясно, что вся эта работа оправдает усилия.
#
# Обратите внимание, что существует ускоренная версия этой функции
# которая переопределяет данную реализацию далее в этом файле.
if isinstance(fut, _PyFuture) and isinstance(waiter, _PyFuture):
if fut._Future__asyncio_awaited_by is None:
fut._Future__asyncio_awaited_by = set()
fut._Future__asyncio_awaited_by.add(waiter)
def future_discard_from_awaited_by(fut, waiter, /):
"""Записывает, что `fut` больше не ожидается объектом `waiter`."""
# См. комментарий в теле "future_add_to_awaited_by()" для
# подробностей реализации.
#
# Обратите внимание, что существует ускоренная версия этой функции
# которая переопределяет данную реализацию далее в этом файле.
if isinstance(fut, _PyFuture) and isinstance(waiter, _PyFuture):
if fut._Future__asyncio_awaited_by is not None:
fut._Future__asyncio_awaited_by.discard(waiter)
_py_future_add_to_awaited_by = future_add_to_awaited_by
_py_future_discard_from_awaited_by = future_discard_from_awaited_by
try:
import _asyncio
except ImportError:
pass
else:
# _CFuture необходим для тестов.
Future = _CFuture = _asyncio.Future
future_add_to_awaited_by = _asyncio.future_add_to_awaited_by
future_discard_from_awaited_by = _asyncio.future_discard_from_awaited_by
_c_future_add_to_awaited_by = future_add_to_awaited_by
_c_future_discard_from_awaited_by = future_discard_from_awaited_by