staggered.py
python
"""Поддержка параллельного запуска корутин с разнесённым временем старта."""
__all__ = 'staggered_race',
import contextlib
from . import events
from . import exceptions as exceptions_mod
from . import locks
from . import tasks
from . import futures
async def staggered_race(coro_fns, delay, *, loop=None):
"""Запустить корутины с разнесённым временем старта и выбрать первую завершившуюся.
Этот метод принимает итерируемый объект корутинных функций. Первая запускается
немедленно. Далее, каждый раз, когда непосредственно предыдущая корутина завершается
ошибкой (возбуждает исключение) или когда проходит *delay* секунд, запускается
следующая корутина. Это продолжается до тех пор, пока одна из корутин не завершится
успешно (в этом случае все остальные отменяются) или пока все корутины не завершатся
ошибкой.
Предоставленные корутины должны вести себя следующим образом:
* Они должны только ``return`` в случае успешного завершения.
* Они всегда должны возбуждать исключение, если не завершились успешно.
В частности, если они обрабатывают отмену, им следует, вероятно,
перевозбуждать исключение, например так::
try:
# выполнение работы
except asyncio.CancelledError:
# отмена частично выполненной работы
raise
Args:
coro_fns: итерируемый объект корутинных функций, т.е. вызываемых объектов,
которые при вызове возвращают объект корутины. Используйте ``functools.partial`` или
лямбда-функции для передачи аргументов.
delay: время в секундах между запуском корутин. Если ``None``,
корутины будут выполняться последовательно.
loop: используемый цикл событий.
Returns:
кортеж *(winner_result, winner_index, exceptions)*, где
- *winner_result*: результат выигравшей корутины или ``None``,
если ни одна корутина не выиграла.
- *winner_index*: индекс выигравшей корутины в
``coro_fns`` или ``None``, если ни одна корутина не выиграла. Если
выигравшая корутина может вернуть None в случае успеха, *winner_index* можно использовать
для однозначного определения, выиграла ли какая-либо корутина.
- *exceptions*: список исключений, возвращённых корутинами.
``len(exceptions)`` равен количеству фактически запущенных корутин,
порядок совпадает с порядком в ``coro_fns``. Запись для выигравшей
корутины равна ``None``.
"""
loop = loop or events.get_running_loop()
parent_task = tasks.current_task(loop)
enum_coro_fns = enumerate(coro_fns)
winner_result = None
winner_index = None
unhandled_exceptions = []
exceptions = []
running_tasks = set()
on_completed_fut = None
def task_done(task):
running_tasks.discard(task)
futures.future_discard_from_awaited_by(task, parent_task)
if (
on_completed_fut is not None
and not on_completed_fut.done()
and not running_tasks
):
on_completed_fut.set_result(None)
if task.cancelled():
return
exc = task.exception()
if exc is None:
return
unhandled_exceptions.append(exc)
async def run_one_coro(ok_to_start, previous_failed) -> None:
# В нетерпеливых задачах это ожидание вызывает добавление этой задачи вызывающей задачей
# в running_tasks, в обычных задачах это ожидание является пустой операцией, которая
# не порождает future. См. gh-124309.
await ok_to_start.wait()
# Ждать завершения предыдущей задачи или истечения времени задержки
if previous_failed is not None:
with contextlib.suppress(exceptions_mod.TimeoutError):
# Используйте asyncio.wait_for() вместо asyncio.wait() здесь, чтобы
# если нас отменят в этой точке, Event.wait() также
# был отменён, иначе позже появится сообщение "Task destroyed but it is
# pending".
await tasks.wait_for(previous_failed.wait(), delay)
# Получить следующую корутину для запуска
try:
this_index, coro_fn = next(enum_coro_fns)
except StopIteration:
return
# Запустить задачу, которая выполнит следующую корутину
this_failed = locks.Event()
next_ok_to_start = locks.Event()
next_task = loop.create_task(run_one_coro(next_ok_to_start, this_failed))
futures.future_add_to_awaited_by(next_task, parent_task)
running_tasks.add(next_task)
next_task.add_done_callback(task_done)
# next_task была добавлена в running_tasks, поэтому next_task можно
# запускать.
next_ok_to_start.set()
# Подготовить место для размещения исключений этой корутины, если она не выиграет
exceptions.append(None)
assert len(exceptions) == this_index + 1
try:
result = await coro_fn()
except (SystemExit, KeyboardInterrupt):
raise
except BaseException as e:
exceptions[this_index] = e
this_failed.set() # Запустить следующую корутину
else:
# Сохранить результаты победителя
nonlocal winner_index, winner_result
assert winner_index is None
winner_index = this_index
winner_result = result
# Отменить все остальные задачи. Мы заботимся о том, чтобы не отменять текущую
# задачу. Если мы это сделаем, то, поскольку здесь нет `await`,
# а CancelledError обычно возбуждаются в точках ожидания, мы
# столкнёмся с любопытным пограничным случаем, когда текущая задача завершится
# как done() == True, cancelled() == False, exception() ==
# asyncio.CancelledError. Это поведение описано в
# https://bugs.python.org/issue30048
current_task = tasks.current_task(loop)
for t in running_tasks:
if t is not current_task:
t.cancel()
propagate_cancellation_error = None
try:
ok_to_start = locks.Event()
first_task = loop.create_task(run_one_coro(ok_to_start, None))
futures.future_add_to_awaited_by(first_task, parent_task)
running_tasks.add(first_task)
first_task.add_done_callback(task_done)
# first_task была добавлена в running_tasks, поэтому first_task можно запускать.
ok_to_start.set()
propagate_cancellation_error = None
# Убедиться, что не осталось запущенных задач при выходе из этой функции.
while running_tasks:
on_completed_fut = loop.create_future()
try:
await on_completed_fut
except exceptions_mod.CancelledError as ex:
propagate_cancellation_error = ex
for task in running_tasks:
task.cancel(*ex.args)
on_completed_fut = None
if __debug__ and unhandled_exceptions:
# Если run_one_coro вызывает необработанное исключение, это, вероятно,
# ошибка программирования, и я хочу её увидеть.
raise ExceptionGroup("staggered race failed", unhandled_exceptions)
if propagate_cancellation_error is not None:
raise propagate_cancellation_error
return winner_result, winner_index, exceptions
finally:
del exceptions, propagate_cancellation_error, unhandled_exceptions, parent_task