tasks.py
python
"""Поддержка задач, корутин и планировщика."""
__all__ = ['Task',
'FIRST_COMPLETED', 'FIRST_EXCEPTION', 'ALL_COMPLETED',
'wait', 'wait_for', 'as_completed', 'sleep', 'async',
'gather', 'shield', 'ensure_future', 'run_coroutine_threadsafe',
]
import concurrent.futures
import functools
import inspect
import warnings
import weakref
from . import base_tasks
from . import compat
from . import coroutines
from . import events
from . import futures
from .coroutines import coroutine
class Task(futures.Future):
"""Корутина, обёрнутая в Future."""
# Важный инвариант, поддерживаемый, пока задача не выполнена:
#
# - Либо _fut_waiter равен None, и запланирован вызов _step();
# - или _fut_waiter является некоторым Future, и _step() *не* запланирован.
#
# Единственный переход от последнего к первому происходит через
# _wakeup(). Когда _fut_waiter не None, один из его колбэков
# должен быть _wakeup().
# Слабое множество, содержащее все живые задачи.
_all_tasks = weakref.WeakSet()
# Словарь, содержащий задачи, которые в данный момент активны в
# во всех работающих циклах событий. {EventLoop: Task}
_current_tasks = {}
# Если False, не записывать сообщение, если задача уничтожена, тогда как её
# статус всё ещё ожидает (pending)
_log_destroy_pending = True
@classmethod
def current_task(cls, loop=None):
"""Возвращает текущую выполняющуюся задачу в цикле событий или None.
По умолчанию возвращается текущая задача для текущего цикла событий.
None возвращается, когда вызов происходит не в контексте задачи.
"""
if loop is None:
loop = events.get_event_loop()
return cls._current_tasks.get(loop)
@classmethod
def all_tasks(cls, loop=None):
"""Возвращает множество всех задач для цикла событий.
По умолчанию возвращаются все задачи для текущего цикла событий.
"""
if loop is None:
loop = events.get_event_loop()
return {t for t in cls._all_tasks if t._loop is loop}
def __init__(self, coro, *, loop=None):
assert coroutines.iscoroutine(coro), repr(coro)
super().__init__(loop=loop)
if self._source_traceback:
del self._source_traceback[-1]
self._coro = coro
self._fut_waiter = None
self._must_cancel = False
self._loop.call_soon(self._step)
self.__class__._all_tasks.add(self)
# На Python 3.3 и старше объекты с деструктором, являющиеся частью
# цикла ссылок, никогда не уничтожаются. Это уже не так на
# Python 3.4 благодаря PEP 442.
if compat.PY34:
def __del__(self):
if self._state == futures._PENDING and self._log_destroy_pending:
context = {
'task': self,
'message': 'Task was destroyed but it is pending!',
}
if self._source_traceback:
context['source_traceback'] = self._source_traceback
self._loop.call_exception_handler(context)
futures.Future.__del__(self)
def _repr_info(self):
return base_tasks._task_repr_info(self)
def get_stack(self, *, limit=None):
"""Возвращает список стековых фреймов для корутины этой задачи.
Если корутина не завершена, возвращается стек, в котором она приостановлена.
Если корутина успешно завершилась или была отменена, возвращается пустой список.
Если корутина была завершена исключением, возвращается список фреймов трассировки.
Фреймы всегда упорядочены от старых к новым.
Необязательный аргумент limit задаёт максимальное количество фреймов для
возврата; по умолчанию возвращаются все доступные фреймы. Его значение
отличается в зависимости от того, возвращается стек или трассировка:
возвращаются самые новые фреймы стека, но самые старые фреймы трассировки.
(Это соответствует поведению модуля traceback.)
По причинам, не зависящим от нас, для приостановленной корутины возвращается
только один стековый фрейм.
"""
return base_tasks._task_get_stack(self, limit)
def print_stack(self, *, limit=None, file=None):
"""Печатает стек или трассировку для корутины этой задачи.
Вывод аналогичен выводу модуля traceback для фреймов, полученных с помощью get_stack().
Аргумент limit передаётся в get_stack(). Аргумент file — это поток ввода/вывода,
в который записывается вывод; по умолчанию вывод записывается в sys.stderr.
"""
return base_tasks._task_print_stack(self, limit, file)
def cancel(self):
"""Запрашивает отмену этой задачи.
Это организует выброс CancelledError в обёрнутую корутину на следующем цикле событий.
Затем корутина получает возможность очиститься или даже отклонить запрос
с помощью try/except/finally.
В отличие от Future.cancel, это не гарантирует, что задача будет отменена:
исключение может быть перехвачено и обработано, что задержит отмену задачи или
предотвратит её полностью. Задача также может вернуть значение или
вызвать другое исключение.
Сразу после вызова этого метода Task.cancelled() будет возвращать False
(если задача ещё не была отменена). Задача будет помечена как отменённая,
когда обёрнутая корутина завершится с исключением CancelledError
(даже если cancel() не вызывался).
"""
self._log_traceback = False
if self.done():
return False
if self._fut_waiter is not None:
if self._fut_waiter.cancel():
# Открепить self._fut_waiter; это может быть задача, которая
# перехватывает и игнорирует отмену, поэтому нам, возможно, придётся
# отменить её снова позже.
return True
# Должно быть так, что self._step уже запланирован.
self._must_cancel = True
return True
def _step(self, exc=None):
assert not self.done(), \
'_step(): already done: {!r}, {!r}'.format(self, exc)
if self._must_cancel:
if not isinstance(exc, futures.CancelledError):
exc = futures.CancelledError()
self._must_cancel = False
coro = self._coro
self._fut_waiter = None
self.__class__._current_tasks[self._loop] = self
# Вызвать либо coro.throw(exc), либо coro.send(None).
try:
if exc is None:
# Мы используем метод `send` напрямую, потому что корутины
# не имеют методов `__iter__` и `__next__`.
result = coro.send(None)
else:
result = coro.throw(exc)
except StopIteration as exc:
if self._must_cancel:
# Задача отменяется прямо перед остановкой корутины.
self._must_cancel = False
self.set_exception(futures.CancelledError())
else:
self.set_result(exc.value)
except futures.CancelledError:
super().cancel() # То есть Future.cancel(self).
except Exception as exc:
self.set_exception(exc)
except BaseException as exc:
self.set_exception(exc)
raise
else:
blocking = getattr(result, '_asyncio_future_blocking', None)
if blocking is not None:
# Future, переданная через yield, должна быть получена из Future.__iter__().
if result._loop is not self._loop:
self._loop.call_soon(
self._step,
RuntimeError(
'Task {!r} got Future {!r} attached to a '
'different loop'.format(self, result)))
elif blocking:
if result is self:
self._loop.call_soon(
self._step,
RuntimeError(
'Task cannot await on itself: {!r}'.format(
self)))
else:
result._asyncio_future_blocking = False
result.add_done_callback(self._wakeup)
self._fut_waiter = result
if self._must_cancel:
if self._fut_waiter.cancel():
self._must_cancel = False
else:
self._loop.call_soon(
self._step,
RuntimeError(
'yield was used instead of yield from '
'in task {!r} with {!r}'.format(self, result)))
elif result is None:
# Голый yield передаёт управление на одну итерацию цикла событий.
self._loop.call_soon(self._step)
elif inspect.isgenerator(result):
# Передавать через yield генератор — просто ошибка.
self._loop.call_soon(
self._step,
RuntimeError(
'yield was used instead of yield from for '
'generator in task {!r} with {!r}'.format(
self, result)))
else:
# Передавать через yield что-то ещё — ошибка.
self._loop.call_soon(
self._step,
RuntimeError(
'Task got bad yield: {!r}'.format(result)))
finally:
self.__class__._current_tasks.pop(self._loop)
self = None # Необходимо для разрыва циклов при возникновении исключения.
def _wakeup(self, future):
try:
future.result()
except Exception as exc:
# Это также может быть отменой.
self._step(exc)
else:
# Не передавайте значение `future.result()` явно,
# так как `Future.__iter__` и `Future.__await__` не нуждаются в нём.
# Если мы вызовем `_step(value, None)` вместо `_step()`,
# цикл выполнения Python будет использовать вызов метода `.send(value)`,
# вместо `__next__()`, что медленнее для future-объектов,
# которые возвращают итераторы, не являющиеся генераторами, из своего `__iter__`.
self._step()
self = None # Необходимо для разрыва циклов при возникновении исключения.
_PyTask = Task
try:
import _asyncio
except ImportError:
pass
else:
# _CTask необходим для тестов.
Task = _CTask = _asyncio.Task
# wait() и as_completed() подобны тем, что в PEP 3148.
FIRST_COMPLETED = concurrent.futures.FIRST_COMPLETED
FIRST_EXCEPTION = concurrent.futures.FIRST_EXCEPTION
ALL_COMPLETED = concurrent.futures.ALL_COMPLETED
@coroutine
def wait(fs, *, loop=None, timeout=None, return_when=ALL_COMPLETED):
"""Ожидает завершения Futures и корутин, заданных через fs.
Последовательность futures не должна быть пустой.
Корутины будут обёрнуты в задачи.
Возвращает два множества Future: (done, pending).
Использование:
done, pending = yield from asyncio.wait(fs)
Примечание: это не вызывает TimeoutError! Futures, которые не завершились
по истечении тайм-аута, возвращаются во втором множестве.
"""
if futures.isfuture(fs) or coroutines.iscoroutine(fs):
raise TypeError("expect a list of futures, not %s" % type(fs).__name__)
if not fs:
raise ValueError('Set of coroutines/Futures is empty.')
if return_when not in (FIRST_COMPLETED, FIRST_EXCEPTION, ALL_COMPLETED):
raise ValueError('Invalid return_when value: {}'.format(return_when))
if loop is None:
loop = events.get_event_loop()
fs = {ensure_future(f, loop=loop) for f in set(fs)}
return (yield from _wait(fs, timeout, return_when, loop))
def _release_waiter(waiter, *args):
if not waiter.done():
waiter.set_result(None)
@coroutine
def wait_for(fut, timeout, *, loop=None):
"""Ожидать завершения единственного Future или корутины с тайм-аутом.
Корутина будет обернута в задачу.
Возвращает результат Future или корутины. При возникновении тайм-аута
отменяет задачу и вызывает TimeoutError. Чтобы избежать отмены задачи,
оберните её в shield().
Если ожидание отменено, задача также отменяется.
Эта функция является корутиной.
"""
if loop is None:
loop = events.get_event_loop()
if timeout is None:
return (yield from fut)
waiter = loop.create_future()
timeout_handle = loop.call_later(timeout, _release_waiter, waiter)
cb = functools.partial(_release_waiter, waiter)
fut = ensure_future(fut, loop=loop)
fut.add_done_callback(cb)
try:
# ждать, пока future не завершится или не истечёт тайм-аут
try:
yield from waiter
except futures.CancelledError:
fut.remove_done_callback(cb)
fut.cancel()
raise
if fut.done():
return fut.result()
else:
fut.remove_done_callback(cb)
fut.cancel()
raise futures.TimeoutError()
finally:
timeout_handle.cancel()
@coroutine
def _wait(fs, timeout, return_when, loop):
"""Внутренний вспомогательный метод для wait() и wait_for().
Аргумент fs должен быть коллекцией Futures.
"""
assert fs, 'Set of Futures is empty.'
waiter = loop.create_future()
timeout_handle = None
if timeout is not None:
timeout_handle = loop.call_later(timeout, _release_waiter, waiter)
counter = len(fs)
def _on_completion(f):
nonlocal counter
counter -= 1
if (counter <= 0 or
return_when == FIRST_COMPLETED or
return_when == FIRST_EXCEPTION and (not f.cancelled() and
f.exception() is not None)):
if timeout_handle is not None:
timeout_handle.cancel()
if not waiter.done():
waiter.set_result(None)
for f in fs:
f.add_done_callback(_on_completion)
try:
yield from waiter
finally:
if timeout_handle is not None:
timeout_handle.cancel()
done, pending = set(), set()
for f in fs:
f.remove_done_callback(_on_completion)
if f.done():
done.add(f)
else:
pending.add(f)
return done, pending
# Это *не* @coroutine! Это просто итератор (возвращающий Future через yield).
def as_completed(fs, *, loop=None, timeout=None):
"""Возвращает итератор, значениями которого являются корутины.
При ожидании полученных корутин вы получите результаты (или
исключения!) исходных Futures (или корутин) в том порядке,
в котором и как только они завершатся.
Это отличается от PEP 3148; правильный способ использования:
for f in as_completed(fs):
result = yield from f # 'yield from' может возбудить исключение.
# Используйте результат.
Если указан тайм-аут, 'yield from' возбудит
TimeoutError, если тайм-аут наступит до завершения всех Futures.
Примечание: futures 'f' не обязательно являются элементами fs.
"""
if futures.isfuture(fs) or coroutines.iscoroutine(fs):
raise TypeError("expect a list of futures, not %s" % type(fs).__name__)
loop = loop if loop is not None else events.get_event_loop()
todo = {ensure_future(f, loop=loop) for f in set(fs)}
from .queues import Queue # Импорт здесь, чтобы избежать циклического импорта.
done = Queue(loop=loop)
timeout_handle = None
def _on_timeout():
for f in todo:
f.remove_done_callback(_on_completion)
done.put_nowait(None) # Поставить фиктивное значение в очередь для _wait_for_one().
todo.clear() # Нельзя делать todo.remove(f) в цикле.
def _on_completion(f):
if not todo:
return # _on_timeout() был здесь первым.
todo.remove(f)
done.put_nowait(f)
if not todo and timeout_handle is not None:
timeout_handle.cancel()
@coroutine
def _wait_for_one():
f = yield from done.get()
if f is None:
# Фиктивное значение от _on_timeout().
raise futures.TimeoutError
return f.result() # Может возбудить f.exception().
for f in todo:
f.add_done_callback(_on_completion)
if todo and timeout is not None:
timeout_handle = loop.call_later(timeout, _on_timeout)
for _ in range(len(todo)):
yield _wait_for_one()
@coroutine
def sleep(delay, result=None, *, loop=None):
"""Корутина, которая завершается через заданное время (в секундах)."""
if delay == 0:
yield
return result
if loop is None:
loop = events.get_event_loop()
future = loop.create_future()
h = future._loop.call_later(delay,
futures._set_result_unless_cancelled,
future, result)
try:
return (yield from future)
finally:
h.cancel()
def async_(coro_or_future, *, loop=None):
"""Оборачивает корутину в future.
Если аргумент является Future, он возвращается напрямую.
Эта функция устарела в версии 3.5. Используйте asyncio.ensure_future() вместо неё.
"""
warnings.warn("asyncio.async() function is deprecated, use ensure_future()",
DeprecationWarning,
stacklevel=2)
return ensure_future(coro_or_future, loop=loop)
# Подавить DeprecationWarning:
globals()['async'] = async_
async_.__name__ = 'async'
del async_
def ensure_future(coro_or_future, *, loop=None):
"""Оборачивает корутину или ожидаемый объект в future.
Если аргумент — Future, он возвращается напрямую.
"""
if futures.isfuture(coro_or_future):
if loop is not None and loop is not coro_or_future._loop:
raise ValueError('loop argument must agree with Future')
return coro_or_future
elif coroutines.iscoroutine(coro_or_future):
if loop is None:
loop = events.get_event_loop()
task = loop.create_task(coro_or_future)
if task._source_traceback:
del task._source_traceback[-1]
return task
elif compat.PY35 and inspect.isawaitable(coro_or_future):
return ensure_future(_wrap_awaitable(coro_or_future), loop=loop)
else:
raise TypeError('An asyncio.Future, a coroutine or an awaitable is '
'required')
@coroutine
def _wrap_awaitable(awaitable):
"""Вспомогательный метод для asyncio.ensure_future().
Оборачивает ожидаемый объект (объект с __await__) в корутину,
которая позже будет обёрнута в Task методом ensure_future().
"""
return (yield from awaitable.__await__())
class _GatheringFuture(futures.Future):
"""Вспомогательный метод для gather().
Переопределяет cancel() для отмены всех дочерних задач и ведёт себя
больше как Task.cancel(), который не помечает себя как отменённый
немедленно.
"""
def __init__(self, children, *, loop=None):
super().__init__(loop=loop)
self._children = children
self._cancel_requested = False
def cancel(self):
if self.done():
return False
ret = False
for child in self._children:
if child.cancel():
ret = True
if ret:
# Если какие-либо дочерние задачи были фактически отменены, мы должны
# распространить запрос на отмену независимо от
# аргумента *return_exceptions*. См. issue 32684.
self._cancel_requested = True
return ret
def gather(*coros_or_futures, loop=None, return_exceptions=False):
"""Возвращает future, агрегирующий результаты переданных корутин
или futures.
Корутины будут обёрнуты в future и запланированы в цикле
событий. Они не обязательно будут запланированы в том же порядке,
в котором были переданы.
Все futures должны использовать один и тот же цикл событий. Если все задачи
успешно выполнены, результатом возвращённого future будет список
результатов (в порядке исходной последовательности, не обязательно
в порядке поступления результатов). Если *return_exceptions* равно True,
исключения в задачах обрабатываются так же, как успешные
результаты, и включаются в список результатов; в противном случае первое
возбуждённое исключение будет немедленно распространено на возвращённый
future.
Отмена: если внешний Future отменён, все дочерние (которые
ещё не завершены) также отменяются. Если какой-либо дочерний
отменён, это рассматривается так, как если бы он возбудил CancelledError --
внешний Future *не* отменяется в этом случае. (Это сделано для
предотвращения отмены одного дочернего элемента, которая может привести к отмене
других дочерних элементов.)
"""
if not coros_or_futures:
if loop is None:
loop = events.get_event_loop()
outer = loop.create_future()
outer.set_result([])
return outer
arg_to_fut = {}
for arg in set(coros_or_futures):
if not futures.isfuture(arg):
fut = ensure_future(arg, loop=loop)
if loop is None:
loop = fut._loop
# Вызывающий не может управлять этим future, предупреждение "destroy pending task"
# не должно выдаваться.
fut._log_destroy_pending = False
else:
fut = arg
if loop is None:
loop = fut._loop
elif fut._loop is not loop:
raise ValueError("futures are tied to different event loops")
arg_to_fut[arg] = fut
children = [arg_to_fut[arg] for arg in coros_or_futures]
nchildren = len(children)
outer = _GatheringFuture(children, loop=loop)
nfinished = 0
results = [None] * nchildren
def _done_callback(i, fut):
nonlocal nfinished
if outer.done():
if not fut.cancelled():
# Пометить исключение как полученное.
fut.exception()
return
if fut.cancelled():
res = futures.CancelledError()
if not return_exceptions:
outer.set_exception(res)
return
elif fut._exception is not None:
res = fut.exception() # Пометить исключение как полученное.
if not return_exceptions:
outer.set_exception(res)
return
else:
res = fut._result
results[i] = res
nfinished += 1
if nfinished == nchildren:
if outer._cancel_requested:
outer.set_exception(futures.CancelledError())
else:
outer.set_result(results)
for i, fut in enumerate(children):
fut.add_done_callback(functools.partial(_done_callback, i))
return outer
def shield(arg, *, loop=None):
"""Ожидает future, защищая его от отмены.
Выражение
res = yield from shield(something())
полностью эквивалентно выражению
res = yield from something()
*за исключением* того, что если корутина, содержащая его, отменяется, то
задача, выполняемая в something(), не отменяется. С точки зрения
something() отмена не произошла. Но её вызывающий код всё ещё
отменён, поэтому выражение yield from всё равно возбуждает
CancelledError. Примечание: Если something() отменяется другим способом,
это всё равно отменит shield().
Если вы хотите полностью игнорировать отмену (не рекомендуется),
вы можете объединить shield() с блоком try/except следующим образом:
try:
res = yield from shield(something())
except CancelledError:
res = None
"""
inner = ensure_future(arg, loop=loop)
if inner.done():
# Сокращение.
return inner
loop = inner._loop
outer = loop.create_future()
def _done_callback(inner):
if outer.cancelled():
if not inner.cancelled():
# Пометить результат inner как полученный.
inner.exception()
return
if inner.cancelled():
outer.cancel()
else:
exc = inner.exception()
if exc is not None:
outer.set_exception(exc)
else:
outer.set_result(inner.result())
inner.add_done_callback(_done_callback)
return outer
def run_coroutine_threadsafe(coro, loop):
"""Отправить объект корутины в заданный цикл событий.
Возвращает concurrent.futures.Future для доступа к результату.
"""
if not coroutines.iscoroutine(coro):
raise TypeError('A coroutine object is required')
future = concurrent.futures.Future()
def callback():
try:
futures._chain_future(ensure_future(coro, loop=loop), future)
except Exception as exc:
if future.set_running_or_notify_cancel():
future.set_exception(exc)
raise
loop.call_soon_threadsafe(callback)
return future