futures.py
python
"""Класс Future, аналогичный описанному в PEP 3148."""
__all__ = ['CancelledError', 'TimeoutError', 'InvalidStateError',
'Future', 'wrap_future', 'isfuture']
import concurrent.futures
import logging
import sys
import traceback
from . import base_futures
from . import compat
from . import events
CancelledError = base_futures.CancelledError
InvalidStateError = base_futures.InvalidStateError
TimeoutError = base_futures.TimeoutError
isfuture = base_futures.isfuture
_PENDING = base_futures._PENDING
_CANCELLED = base_futures._CANCELLED
_FINISHED = base_futures._FINISHED
STACK_DEBUG = logging.DEBUG - 1 # углублённая отладка
class _TracebackLogger:
"""Вспомогательный объект для логирования трассировки при уничтожении, если он не был очищен.
Это решает неприятную проблему с Future и Task, у которых установлено исключение: если никто не запрашивает исключение, оно никогда не логируется. Это нарушает Дзен Python: «Ошибки никогда не должны проходить незамеченными. Если только они не были явно замалчены».
Однако мы не хотим логировать исключение сразу после вызова set_exception(): если вызывающий код написан правильно, он получит исключение и корректно обработает его. Но мы *действительно* хотим логировать его, если result() или exception() никогда не вызывались — иначе разработчики тратят много времени, гадая, почему их ошибочный код молча падает.
Ранняя попытка добавила метод __del__() в сам класс Future, но это дало обратный эффект, так как наличие __del__() мешает сборщику мусора разрывать циклические ссылки. Выход из этой безвыходной ситуации — не иметь метода __del__() в самом классе Future, а вместо этого иметь ссылку на вспомогательный объект с методом __del__(), который логирует трассировку, при этом мы гарантируем, что вспомогательный объект не участвует в циклах, и только Future имеет на него ссылку.
Вспомогательный объект добавляется при вызове set_exception(). Когда Future собирается сборщиком мусора, и вспомогательный объект присутствует, он также собирается, и его метод __del__() логирует трассировку. Когда вызывается метод result() или exception() Future (и вспомогательный объект присутствует), он удаляет вспомогательный объект, предварительно вызвав его метод clear(), чтобы предотвратить логирование.
Один из недостатков в том, что мы выполняем значительную работу по извлечению трассировки из исключения, даже если она никогда не логируется. Казалось бы, дешевле просто сохранить объект исключения, но он ссылается на трассировку, которая ссылается на стеки вызовов, которые могут ссылаться на Future, который ссылается на _TracebackLogger, и тогда _TracebackLogger окажется в цикле, чего мы и пытаемся избежать! В качестве оптимизации мы не производим немедленное форматирование исключения; мы делаем это только при вызове activate(), который откладывается до тех пор, пока не выполнятся все колбэки Future. Поскольку обычно у Future есть хотя бы один колбэк (обычно установленный через 'yield from'), и этот колбэк обычно извлекает результат, тем самым устраняя необходимость в форматировании исключения.
P.S. Я не претендую на авторство этого решения. Я впервые услышал о нём в обсуждении закрытия файлов при их сборке.
"""
__slots__ = ('loop', 'source_traceback', 'exc', 'tb')
def __init__(self, future, exc):
self.loop = future._loop
self.source_traceback = future._source_traceback
self.exc = exc
self.tb = None
def activate(self):
exc = self.exc
if exc is not None:
self.exc = None
self.tb = traceback.format_exception(exc.__class__, exc,
exc.__traceback__)
def clear(self):
self.exc = None
self.tb = None
def __del__(self):
if self.tb:
msg = 'Future/Task exception was never retrieved\n'
if self.source_traceback:
src = ''.join(traceback.format_list(self.source_traceback))
msg += 'Future/Task created at (most recent call last):\n'
msg += '%s\n' % src.rstrip()
msg += ''.join(self.tb).rstrip()
self.loop.call_exception_handler({'message': msg})
class Future:
"""Этот класс *почти* совместим с concurrent.futures.Future.
Отличия:
- Этот класс не является потокобезопасным.
- result() и exception() не принимают аргумент таймаута и
выбрасывают исключение, если future ещё не завершен.
- Колбэки, зарегистрированные через add_done_callback(), всегда вызываются
через call_soon() цикла событий.
- Этот класс не совместим с методами wait() и as_completed()
из пакета concurrent.futures.
(В Python 3.4 или новее мы сможем унифицировать реализации.)
"""
# Переменные класса, служащие значениями по умолчанию для переменных экземпляра.
_state = _PENDING
_result = None
_exception = None
_loop = None
_source_traceback = None
# Это поле используется для двух целей:
# - Его наличие является маркером, объявляющим, что класс реализует
# протокол Future (т.е. предполагается совместимость по утиному типированию).
# Значение также должно быть не-None, чтобы позволить подклассу объявить
# что он несовместим, установив это значение в None.
# - Устанавливается методом __iter__() ниже, чтобы Task._step() мог различить
# разница между `yield from Future()` (правильно) и
# `yield Future()` (неправильно).
_asyncio_future_blocking = False
_log_traceback = False
def __init__(self, *, loop=None):
"""Инициализация future.
Необязательный аргумент event_loop позволяет явно задать объект
цикла событий, используемый future. Если он не предоставлен, future использует
цикл событий по умолчанию.
"""
if loop is None:
self._loop = events.get_event_loop()
else:
self._loop = loop
self._callbacks = []
if self._loop.get_debug():
self._source_traceback = events.extract_stack(sys._getframe(1))
_repr_info = base_futures._future_repr_info
def __repr__(self):
return '<%s %s>' % (self.__class__.__name__, ' '.join(self._repr_info()))
# В Python 3.3 и старше объекты с деструктором, являющиеся частью циклической ссылки,
# никогда не уничтожаются. В Python 3.4 это уже не так благодаря
# PEP 442.
if compat.PY34:
def __del__(self):
if not self._log_traceback:
# set_exception() не был вызван, или result() или exception()
# уже потребили исключение
return
exc = self._exception
context = {
'message': ('%s exception was never retrieved'
% self.__class__.__name__),
'exception': exc,
'future': self,
}
if self._source_traceback:
context['source_traceback'] = self._source_traceback
self._loop.call_exception_handler(context)
def cancel(self):
"""Отменяет future и планирует колбэки.
Если future уже завершён или отменён, возвращает False. В противном
случае изменяет состояние future на «отменён», планирует колбэки и
возвращает True.
"""
self._log_traceback = False
if self._state != _PENDING:
return False
self._state = _CANCELLED
self._schedule_callbacks()
return True
def _schedule_callbacks(self):
"""Внутренний метод: просит цикл событий вызвать все колбэки.
Колбэки планируются к вызову как можно скорее. Также очищает список
колбэков.
"""
callbacks = self._callbacks[:]
if not callbacks:
return
self._callbacks[:] = []
for callback in callbacks:
self._loop.call_soon(callback, self)
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 CancelledError
if self._state != _FINISHED:
raise InvalidStateError('Result is not ready.')
self._log_traceback = False
if self._exception is not None:
raise self._exception
return self._result
def exception(self):
"""Возвращает исключение, которое было установлено для этого future.
Исключение (или None, если исключение не установлено) возвращается
только если future завершён. Если future был отменён, возбуждается
CancelledError. Если future ещё не завершён, возбуждается
InvalidStateError.
"""
if self._state == _CANCELLED:
raise CancelledError
if self._state != _FINISHED:
raise InvalidStateError('Exception is not set.')
self._log_traceback = False
return self._exception
def add_done_callback(self, fn):
"""Добавляет колбэк, который будет вызван, когда future станет завершённым.
Колбэк вызывается с одним аргументом — объектом future. Если future
уже завершён на момент вызова, колбэк планируется с помощью call_soon.
"""
if self._state != _PENDING:
self._loop.call_soon(fn, self)
else:
self._callbacks.append(fn)
# Новый метод, отсутствующий в PEP 3148.
def remove_done_callback(self, fn):
"""Удаляет все экземпляры колбэка из списка «вызвать при завершении».
Возвращает количество удалённых колбэков.
"""
filtered_callbacks = [f for f 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 InvalidStateError('{}: {!r}'.format(self._state, self))
self._result = result
self._state = _FINISHED
self._schedule_callbacks()
def set_exception(self, exception):
"""Помечает future как завершённый и устанавливает исключение.
Если future уже завершён при вызове этого метода, возбуждается
InvalidStateError.
"""
if self._state != _PENDING:
raise InvalidStateError('{}: {!r}'.format(self._state, self))
if isinstance(exception, type):
exception = exception()
if type(exception) is StopIteration:
raise TypeError("StopIteration interacts badly with generators "
"and cannot be raised into a Future")
self._exception = exception
self._state = _FINISHED
self._schedule_callbacks()
if compat.PY34:
self._log_traceback = True
else:
self._tb_logger = _TracebackLogger(self, exception)
# Организовать активацию регистратора после того, как все колбэки
# смогут вызвать result() или exception().
self._loop.call_soon(self._tb_logger.activate)
def __iter__(self):
if not self.done():
self._asyncio_future_blocking = True
yield self # Это указывает Task дождаться завершения.
assert self.done(), "yield from wasn't used with future"
return self.result() # Может также возбуждать исключение.
if compat.PY35:
__await__ = __iter__ # сделать совместимым с выражением 'await'
# Необходим для целей тестирования.
_PyFuture = Future
def _set_result_unless_cancelled(fut, result):
"""Вспомогательный метод, устанавливающий результат только если future не была отменена."""
if fut.cancelled():
return
fut.set_result(result)
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(exception)
else:
result = source.result()
concurrent.set_result(result)
def _copy_future_state(source, dest):
"""Внутренний вспомогательный метод для копирования состояния из другого Future.
Другой Future может быть concurrent.futures.Future.
"""
assert source.done()
if dest.cancelled():
return
assert not dest.done()
if source.cancelled():
dest.cancel()
else:
exception = source.exception()
if exception is not None:
dest.set_exception(exception)
else:
result = source.result()
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 = source._loop if isfuture(source) else None
dest_loop = destination._loop 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 dest_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 source_loop:
_set_state(destination, source)
else:
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), \
'concurrent.futures.Future is expected, got {!r}'.format(future)
if loop is None:
loop = events.get_event_loop()
new_future = loop.create_future()
_chain_future(future, new_future)
return new_future
try:
import _asyncio
except ImportError:
pass
else:
# _CFuture необходим для тестов.
Future = _CFuture = _asyncio.Future