Исключения в конкурентном коде

В синхронном коде исключение предсказуемо: летит по стеку наверх, пока его не поймают. В конкурентном всё интереснее: одновременно работают десятки корутин, и любая из них может упасть. Куда полетит её ошибка? Кто её увидит? Что случится с соседями? В уроке про gather мы собрали пачку задач — сегодня пачка начнёт падать, и мы научимся это переживать.

Разница с синхронным миром не только в траектории ошибки, но и в привычках. В обычном коде try/except оборачивает один конкретный вызов, и читатель видит границы защиты. В пачке защита — свойство всей конструкции: один флаг в gather, и границы «кто за кого отвечает» меняются радикально. Поэтому соглашения из этого урока стоит проговаривать вслух на ревью: где мы ловим внутри корутины, где собираем с флагом, а где честно позволяем падению ронять всё — это три разных контракта одного и того же gather.

Одна ошибка роняет всю пачку

Поведение по умолчанию: gather пробрасывает первое исключение, возникшее среди задач. Запустим три корутины, где средняя падает с ValueError, и попробуем поймать это снаружи:

import asyncio

async def ok(name, delay):
    await asyncio.sleep(delay)
    return name + ": успех"

async def broken():
    await asyncio.sleep(1)
    raise ValueError("битые данные")

async def main():
    try:
        results = await asyncio.gather(
            ok("первая", 2),
            broken(),
            ok("третья", 3),
        )
        print("Результаты:", results)
    except ValueError as e:
        print("gather упал с ошибкой:", e)
        print("Результаты успешных задач потеряны")

asyncio.run(main())

gather упал с ошибкой: битые данные
Результаты успешных задач потеряны

Ошибка вылетела на первой секунде — и унесла с собой всё: результаты «первой» и «третьей» задач, которые готовы были через 2 и 3 секунды, до вызывающего кода не доехали. Для пачки независимых проверок это слишком радикально: почему проверка битого файла должна мешать отчёту по остальным?

Нюанс, который часто удивляет: упав из gather, исключение не останавливает остальных. Задачи «первая» и «третья» продолжали крутиться и честно отработали свои две-три секунды — просто их результаты уже некому читали. В долгих сценариях это незаметный расход ресурсов: пачка на десять минут падает на первой минуте, а остальные девять процессор молча обрабатывает то, что выброшено. Отсюда привычка: если gather без флага упал, а остальные задачи вам всё-таки нужны (или наоборот, нужно их прибить) — держите ссылки на задачи и управляйте ими явно: дожидайте через gather с флагом или отменяйте, как в уроке про отмену.

Классическое место, где это предупреждение заводится, — задачи, созданные и забытые: create_task без сохранения ссылки в переменную. Такой задачи даже cancel() не отправить — она живёт только в списке цикла, а её ошибка никому не видна. Правило из урока про create_task здесь работает вдвойне: ссылку на каждую задачу хранят (в списке, в словаре по id), а собирают пачкой через gather — тогда у каждой ошибки обязательно появится читатель.

return_exceptions=True: ошибки как значения

Флаг меняет контракт: gather больше ничего не пробрасывает. Каждое исключение кладётся в список результатов на место соответствующей задачи — как обычное значение. Проверяешь тип — и решаешь, что с ним делать:

import asyncio

async def ok(name, delay):
    await asyncio.sleep(delay)
    return name + ": успех"

async def broken():
    await asyncio.sleep(1)
    raise ValueError("битые данные")

async def main():
    results = await asyncio.gather(
        ok("первая", 2),
        broken(),
        ok("третья", 3),
        return_exceptions=True,
    )
    for r in results:
        if isinstance(r, Exception):
            print("Провал:", repr(r))
        else:
            print("Готово:", r)

asyncio.run(main())

Готово: первая: успех
Провал: ValueError('битые данные')
Готово: третья: успех

Список собран целиком: успехи на своих местах (порядок аргументов, как мы выяснили в уроке 8), а на месте упавшей — объект ValueError. Проверка isinstance — единственный различитель: строка это результат или ошибка. Теперь пачка независимых задач переживает любое количество падений, и отчёт всегда полный.

Обрати внимание: адресация по слотам работает и с ошибками. Зная, что results[1] — это про «вторую» задачу, можно собрать отчёт вида «источник такой-то: такая-то неудача» без всяких словарей с идентификаторами. С as_completed такой адресации нет — там метка должна лежать внутри результата; с gather с флагом позиция в списке — уже готовый ключ. Маленькая, но приятная экономия на целой прослойке сопоставления «кто есть кто».

Изоляция падений: отчёт по пачке

Типовая боевая задача: обойти пачку источников, собрать успешные и честно перечислить упавших. Фильтруем список результатов на два ведра и печатаем сводку:

import asyncio

async def check(url, broken):
    await asyncio.sleep(1)
    if broken:
        raise ConnectionError(url + " недоступен")
    return url + ": 200"

async def main():
    tasks = [
        check("site-1", False),
        check("site-2", True),
        check("site-3", False),
        check("site-4", True),
    ]
    results = await asyncio.gather(*tasks, return_exceptions=True)

    good = [r for r in results if not isinstance(r, Exception)]
    bad = [r for r in results if isinstance(r, Exception)]
    print("Успехов:", len(good), "| Провалов:", len(bad))
    for r in bad:
        print("Причина:", r)

asyncio.run(main())

Успехов: 2 | Провалов: 2
Причина: site-2 недоступен
Причина: site-4 недоступен

Два упавших мониторинга не отменили два удачных: отчёт показывает полную картину — и кого не достучались, и по какой причине. Именно так строят дашборды проверок и обходы каталогов: полный список с пометками, а не «упал на первом же».

Обрабатывай внутри корутины, когда можешь

Флаг return_exceptions — инструмент отчётности, а не замена обработки. Если ошибка предсказуема и есть разумный запасной план, ловите её прямо в корутине: тогда она вообще не доходит до gather, и список остаётся однородным:

import asyncio

async def safe_job(name):
    try:
        await asyncio.sleep(1)
        if name == "Бета":
            raise ValueError("брак")
        return name + ": ок"
    except ValueError as e:
        return name + ": обработали сами (" + str(e) + ")"

async def main():
    results = await asyncio.gather(safe_job("Альфа"), safe_job("Бета"))
    for r in results:
        print(r)

asyncio.run(main())

Альфа: ок
Бета: обработали сами (брак)

Правило выбора простое: знаешь, как жить с этой ошибкой, — обрабатывай в корутине; ошибка должна быть видна наверху — пропускай и собирай через return_exceptions=True. Пойманный баг — не подарок, а маскировка: он тихо поплывёт по данным и найдётся по испорченному отчёту через неделю. Пусть баги падают громко: пачка с флагом всё равно соберёт их в список, а трейсбек в логе покажет настоящее место. А когда задача создаётся через create_task без gather, исключение живёт внутри объекта задачи до её await:

import asyncio

async def boom():
    raise RuntimeError("упала в тени")

async def main():
    task = asyncio.create_task(boom())
    await asyncio.sleep(0)        # задача успевает упасть
    print("main жив, исключение ещё не поднимали")
    try:
        await task                # здесь оно вылетает
    except RuntimeError as e:
        print("Поймали при await:", e)

asyncio.run(main())

main жив, исключение ещё не поднимали
Поймали при await: упала в тени

Задача упала мгновенно, но main об этом не узнал, пока не сделал await task. Это и есть контракт: Task хранит исключение и отдаёт его первому, кто дождётся. Никто не дождётся — получите предупреждение из питфолла выше.

У объекта задачи есть и «безопасный» способ спросить про ошибку — метод task.exception(): он возвращает объект исключения, если задача упала, None, если всё хорошо, и не бросает ничего. Удобно для проверок в духе «а всё ли в порядке с фоновыми задачами» без try/except вокруг каждого await. Помните только: получив исключение из exception(), вы всё ещё отвечаете за его судьбу — хотя бы залогировать, иначе снова скатитесь в «never retrieved».

В пачке могут летать исключения разных типов — фильтр isinstance различает и их. Для полноты картины соберём смешанную пачку и посмотрим на каждый слот:

import asyncio

async def job(name, exc):
    await asyncio.sleep(1)
    if exc is not None:
        raise exc
    return name + ": ок"

async def main():
    results = await asyncio.gather(
        job("Альфа", None),
        job("Бета", ValueError("плохой формат")),
        job("Гамма", KeyError("нет ключа")),
        return_exceptions=True,
    )
    for r in results:
        print(type(r).__name__ + ":", r)

asyncio.run(main())

str: Альфа: ок
ValueError: плохой формат
KeyError: 'нет ключа'

Когда ошибок в пачке много и они разнотипные, удобный приём — Counter по именам классов: сводка «сколько чего упало» для дашборда собирается в две строки:

import asyncio
from collections import Counter

async def job(exc):
    await asyncio.sleep(1)
    if exc is not None:
        raise exc
    return "ок"

async def main():
    results = await asyncio.gather(
        job(ValueError("а")),
        job(KeyError("б")),
        job(None),
        job(ValueError("в")),
        return_exceptions=True,
    )
    fails = Counter(type(r).__name__ for r in results if isinstance(r, Exception))
    ok = sum(1 for r in results if not isinstance(r, Exception))
    print("Успехов:", ok)
    for name, count in fails.items():
        print(name + ":", count)

asyncio.run(main())

Успехов: 1
ValueError: 2
KeyError: 1

Такая сводка — готовая метрика: число провалов по типам пишется в мониторинг, и алерт «ValueError стало втрое больше» срабатывает раньше, чем пользователи заметят, что данные перестали приезжать. Заметьте, как мало кода: Counter по именам классов плюс одна сумма — а на выходе картина здоровья пачки. В больших проектах к этому добавляют долю провалов от общего числа, но принцип тот же: собрал список с флагом — преврати его в числа, а не только в логи.

Сводная шпаргалка по трём стратегиям работы с падениями в пачке:

import asyncio
import aiohttp

async def fetch(session, url):
    try:
        async with session.get(url) as resp:
            return url, "ok"
    except aiohttp.ClientError as e:
        return url, "network: " + str(e)

async def main():
    urls = ["https://example.com/", "https://broken.example/"]
    async with aiohttp.ClientSession() as session:
        results = await asyncio.gather(*(fetch(session, u) for u in urls))
    for url, status in results:
        print(url, "->", status)

asyncio.run(main())

Что дальше

Пачка научилась переживать падения: ошибки собираются как значения, отчёт всегда полный. Дальше — о ложной конкурентности: что произойдёт, если внутри async-кода вызвать обычный time.sleep. Спойлер из первого урока: весь цикл замрёт, и никакие задачи друг друга не спасут. Разбор с таймстампами — в уроке 12.

Домашняя проверка на минуту: найдите в своём коде все gather и для каждого ответьте, какой из трёх контрактов там работает — падение наверх, флаг с отчётом или полная изоляция внутри корутин. Если ответить не получается без чтения кода, контракт стоит закрепить комментарием в одну строку.

Исключение в конкурентном коде — это значение, которое надо доставить: Task хранит его до первого await, gather раскладывает по слотам.