Asyncio gather: отмена оставшихся задач при ошибке

Безопасный опрос узлов

Модуль мониторинга опрашивает удалённые узлы сети, оборачивая корутины в задачи и запуская их параллельно через asyncio.gather(*tasks, return_exceptions=False). В штатной ситуации диспетчер получает упорядоченный список ответов, а признак сбоя остаётся пустым — None возвращается и в том случае, если исходный список задач изначально был пуст.

Однако опрос может сорваться. Причиной первого сбоя бывает как обычное исключение класса Exception, так и внутренняя самоотмена одной из дочерних задач через asyncio.CancelledError. При первом же таком исключении диспетчер перехватывает исходный объект сбоя и не оставляет брошенных фоновых процессов: он находит все ещё не завершившиеся задачи, вызывает для каждой cancel() и гарантированно дожидается их окончательной остановки через gather(*unfinished, return_exceptions=True).

В случае аварии частичные данные узлов отбрасываются, сменяясь значением None. Вместо них диспетчер сохраняет точный объект пойманного первого исключения и подсчитывает, сколько задач набора после полной остановки действительно перешло в состояние task.cancelled() is True. Реализуйте функцию безопасного опроса, возвращающую результаты штатного сбора либо исходную ошибку с точным числом подтверждённых отмен.

Что уже дано

В заготовке объявлена асинхронная функция safe_gather(factories). В окружении уже доступна синхронная вспомогательная функция coordinator_cancel_requested(), которая возвращает True, если у текущей задачи диспетчера есть активный внешний запрос на отмену (task.cancelling() > 0). Её не требуется реализовывать самому.

Что нужно сделать

Реализуйте асинхронную функцию safe_gather(factories).

Функция принимает список factories, содержащий от 0 до 50 вызываемых фабрик без аргументов. Каждый вызов фабрики возвращает новую завершаемую корутину, которая возвращает значение, выбрасывает Exception или внутри себя отменяется через asyncio.CancelledError.

Функция возвращает кортеж из трёх элементов: (results, error, cancelled_count).

Правила выполнения:

• Вызовите каждую фабрику из списка ровно один раз и оберните каждую полученную корутину в реальную задачу asyncio.Task.
• Запустите первичное ожидание задач через asyncio.gather(*tasks, return_exceptions=False).
• При полном успехе (включая пустой список factories) верните кортеж (ordered_values, None, 0), сохранив значения в порядке исходных фабрик.
• При первом обычном сбое (Exception) или внутренней самоотмене дочерней задачи (asyncio.CancelledError при условии, что coordinator_cancel_requested() возвращает False):
• сохраните точный объект первого пойманного исключения;
• найдите все задачи исходного набора, которые ещё не завершились (not task.done()), вызовите для них cancel() и дождитесь их завершения через await asyncio.gather(*unfinished, return_exceptions=True);
• верните кортеж (None, original_first_caught_error, confirmed_cancelled_count);
• частичные результаты успешных задач не возвращаются, а последующие ошибки очистки не заменяют первый пойманный объект сбоя;
• confirmed_cancelled_count — точное число задач исходного набора, у которых task.cancelled() is True (задача, перехватившая отмену и вернувшая значение или обычную ошибку, не считается отменённой);
• все созданные задачи должны полностью завершиться до выхода из функции.
• Если отмена пришла извне самому координатору (coordinator_cancel_requested() равен True), CancelledError распространяется дальше как отмена координатора, не превращаясь в отчёт о внутреннем дочернем сбое.

Ввод и вывод

Стандартные потоки ввода stdin и вывода stdout в решении не используются: считывать ввод и вызывать print() не нужно. После выполнения кода учащегося проверяющая система считывает из пространства имён модуля функцию верхнего уровня safe_gather и вызывает её с проверочными аргументами.

Пример использования

async def ok():
    await asyncio.sleep(0)
    return 0

async def empty_value():
    return None

async def example():
    values, error, cancelled = await safe_gather([ok, empty_value])
    print(values)
    print(error is None)
    print(cancelled)

asyncio.run(example())

Вывод:

[0, None]
True
0