"""Что именно делает `gather` со вторым исключением.

ЗАЧЕМ ЭТОТ ФАЙЛ. Урок говорил: «gather второе исключение теряет насовсем».
Аудит линии возразил: документация обещает другое — при
`return_exceptions=False` вызывающий немедленно получает ПЕРВОЕ исключение, а
остальные awaitables не отменяются и продолжают работу. То есть «теряется» оно
не в gather, а в вызывающем коде, который никуда больше не смотрит. Разница не
словесная: если задачи сохранены, их ошибки наблюдаемы — и это ровно та
поправка, ради которой пишется скрипт.

ЧТО ЗДЕСЬ ПРОВЕРЯЕТСЯ ПРОГОНОМ, А НЕ ПЕРЕСКАЗОМ:

  1. отменяются ли соседи после первого исключения;
  2. доходят ли они до конца и виден ли их результат;
  3. наблюдаемо ли ВТОРОЕ исключение, если задачи сохранены;
  4. что печатает Python, если их не сохранить и не посмотреть;
  5. чем от этого отличается TaskGroup.

ПОЧЕМУ ВЕЗДЕ ЖУРНАЛ СОБЫТИЙ, А НЕ ЗАМЕР ВРЕМЕНИ. Вопрос урока — семантика, а
не стоимость: важно, ЧТО произошло и в каком порядке, а не за сколько
наносекунд. Поэтому каждый корутин пишет строку в общий список, и печатается
он целиком.

ЗАПУСК: python3.13 bench/async-await/gather_semantics.py
Вывод по версиям: runs/gather_semantics-3.11.txt и соседние.
"""

import asyncio
import gc
import sys

LOG: list[str] = []


def show(title: str) -> None:
    print()
    print(title)
    print("-" * len(title))


async def fails_fast(name: str) -> str:
    LOG.append(f"{name}: начал")
    await asyncio.sleep(0.01)
    LOG.append(f"{name}: возбуждает ошибку")
    raise ValueError(f"ошибка из {name}")


async def fails_late(name: str) -> str:
    LOG.append(f"{name}: начал")
    await asyncio.sleep(0.05)
    LOG.append(f"{name}: возбуждает ошибку")
    raise RuntimeError(f"ошибка из {name}")


async def finishes_late(name: str) -> str:
    LOG.append(f"{name}: начал")
    await asyncio.sleep(0.08)
    LOG.append(f"{name}: дошёл до конца")
    return f"{name} готов"


# ---------------------------------------------------------------- 1
async def block1() -> None:
    """Первое исключение прилетает сразу; что происходит с соседями."""
    LOG.clear()
    slow = asyncio.create_task(finishes_late("медленный"))
    late = asyncio.create_task(fails_late("падающий позже"))
    fast = asyncio.create_task(fails_fast("падающий первым"))

    try:
        await asyncio.gather(fast, late, slow)
    except BaseException as exc:
        LOG.append(f"вызывающий получил: {type(exc).__name__}: {exc}")

    LOG.append(f"сразу после gather: медленный done={slow.done()}, поздний done={late.done()}")

    # Ждём столько, чтобы соседи успели завершиться, и смотрим на них.
    await asyncio.sleep(0.15)
    LOG.append(f"медленный: cancelled={slow.cancelled()}, результат={slow.result()!r}")
    LOG.append(f"поздний: cancelled={late.cancelled()}, исключение={late.exception()!r}")

    for line in LOG:
        print(" ", line)


# ---------------------------------------------------------------- 2
async def block2() -> None:
    """Те же задачи, но ссылок на них никто не держит и никто не смотрит.

    «Ошибка потерялась» — утверждение проверяемое, и проверять его нужно не
    глазами. Незабранное исключение задачи цикл событий отдаёт своему
    обработчику ошибок (loop.call_exception_handler, «Task exception was never
    retrieved»). Поэтому обработчик здесь подменяется и считает вызовы: ноль
    означает, что об ошибке не сообщил никто и нигде, а не что её «не было
    видно на экране».
    """
    LOG.clear()
    reported: list[str] = []
    loop = asyncio.get_running_loop()
    previous = loop.get_exception_handler()
    loop.set_exception_handler(lambda _loop, ctx: reported.append(str(ctx.get("message"))))

    try:
        try:
            await asyncio.gather(
                fails_fast("падающий первым"),
                fails_late("падающий позже"),
            )
        except BaseException as exc:
            LOG.append(f"вызывающий получил: {type(exc).__name__}: {exc}")

        # Ждём и заставляем сборщик мусора добраться до брошенных задач:
        # именно при уничтожении задачи цикл сообщает о незабранной ошибке.
        await asyncio.sleep(0.15)
        gc.collect()
        await asyncio.sleep(0)
    finally:
        loop.set_exception_handler(previous)

    LOG.append(f"сообщений от цикла событий о незабранных ошибках: {len(reported)}")
    for message in reported:
        LOG.append(f"  сообщение: {message}")
    for line in LOG:
        print(" ", line)


# ---------------------------------------------------------------- 3
async def block3() -> None:
    """return_exceptions=True: исключения возвращаются как значения."""
    LOG.clear()
    got = await asyncio.gather(
        fails_fast("падающий первым"),
        fails_late("падающий позже"),
        finishes_late("медленный"),
        return_exceptions=True,
    )
    for item in got:
        LOG.append(f"вернулось: {item!r}")
    for line in LOG:
        print(" ", line)


# ---------------------------------------------------------------- 4
async def block4() -> None:
    """TaskGroup: другая семантика — отменяет соседей и собирает группу.

    Ссылки на задачи сохраняются намеренно: «отменяет соседей» — утверждение о
    состоянии соседей, и проверяется оно у них самих, а не по тексту ошибки.
    """
    LOG.clear()
    slow = late = None
    try:
        async with asyncio.TaskGroup() as tg:
            tg.create_task(fails_fast("падающий первым"))
            late = tg.create_task(fails_late("падающий позже"))
            slow = tg.create_task(finishes_late("медленный"))
    except BaseException as exc:
        LOG.append(f"вызывающий получил: {type(exc).__name__}: {exc}")
        subs = getattr(exc, "exceptions", ())
        LOG.append(f"в группе исключений: {[type(s).__name__ for s in subs]}")

    await asyncio.sleep(0.15)
    LOG.append(f"медленный: cancelled={slow.cancelled()}")
    LOG.append(f"поздний: cancelled={late.cancelled()}")
    for line in LOG:
        print(" ", line)


async def main() -> None:
    show("1. gather(return_exceptions=False): что стало с соседями")
    await block1()

    show("2. те же задачи без ссылок: вторая ошибка не наблюдается")
    await block2()

    show("3. gather(return_exceptions=True): исключение как значение")
    await block3()

    show("4. TaskGroup: отмена соседей и ExceptionGroup")
    await block4()


if __name__ == "__main__":
    print(f"Python {sys.version.split()[0]} ({sys.implementation.name})")
    asyncio.run(main())