Теория: Задачи и структурная конкурентность

Цикл событий в asyncio

Цикл событий в asyncio, или event loop, — это центральный механизм, который управляет выполнением всех асинхронных задач в программе. Задача в asyncio — это корутина, которую цикл событий «запоминает» и может приостанавливать и возобновлять. Цикл событий отвечает за переключение между задачами, чтобы ни одна из них не блокировала работу остальных во время ожидания операций ввода-вывода или задержек.

В обычном синхронном коде выполнение останавливается на время долгой операции, например сетевого запроса или чтения файла. В асинхронной модели с использованием asyncio корутины не блокируют поток. Когда корутина встречает оператор await, она возвращает управление циклу событий. Затем цикл событий может выбрать для выполнения другую задачу, готовую к продолжению. Как только ожидаемая операция завершается, event loop возвращается к исходной корутине и продолжает её выполнение с того места, где она остановилась. Таким образом достигается конкурентное выполнение даже в одном потоке.

При запуске программы с помощью asyncio.run() создаётся и запускается новый event loop. Он берёт на себя планирование задач, наблюдает за их состоянием и переключает управление между ними. Цикл событий продолжает работу до тех пор, пока не будут завершены все задачи, а затем корректно закрывается. Разработчику обычно не нужно вручную управлять этим процессом, но важно понимать, что под капотом именно event loop обеспечивает асинхронное поведение.

Планирование задач (create_task)

Когда мы понимаем, что именно цикл событий отвечает за переключение между корутинами, возникает вопрос: как передать корутину на выполнение этому циклу. Обычный оператор await ожидает завершения корутины, блокируя выполнение текущей функции до получения результата. Чтобы запустить несколько задач одновременно и дать циклу событий самому управлять их переключением, используется функция asyncio.create_task().

При вызове asyncio.create_task() корутина превращается в объект задачи и регистрируется в цикле событий. Задача начинает выполняться, как только цикл передаст ей управление, а программа может продолжать создавать другие задачи или выполнять собственный код. Это позволяет не блокировать основной поток и добиваться конкурентного выполнения.

Рассмотрим пример. Допустим, есть корутина, которая имитирует сетевую задержку:

import asyncio


async def do_work(name: str, delay: int) -> None:
    print(f"{name} начата")
    await asyncio.sleep(delay)
    print(f"{name} завершена")


async def main() -> None:
    task1 = asyncio.create_task(do_work("Задача 1", 2))
    task2 = asyncio.create_task(do_work("Задача 2", 3))
    print("Обе задачи запущены")
    await task1
    await task2
    print("Все задачи завершены")


asyncio.run(main())

В этом примере обе задачи стартуют почти одновременно. Цикл событий сам решает, когда переключаться между ними. Пока одна задача «спит» внутри asyncio.sleep(), другая может выполняться. Благодаря этому общее время выполнения примерно равно времени самой долгой задачи — около 3 секунд. Если бы мы использовали простой await do_work(...) без create_task(), то сначала полностью выполнилась бы первая задача за 2 секунды, а затем вторая за 3 секунды. В таком случае программа завершилась бы примерно через 5 секунд, что наглядно показывает преимущество конкурентного запуска через create_task().

Важно понимать, что вызов create_task() не ждёт завершения задачи. Он только передаёт корутину в цикл событий и возвращает объект задачи. Если не сделать последующий await для этой задачи или не сохранить на неё ссылку, программа может завершиться раньше, чем задача выполнится. Поэтому почти всегда после создания задач их нужно явно ожидать или собирать результаты.

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

Именование задач для диагностики

При работе с несколькими задачами полезно давать им осмысленные имена для упрощения отладки и диагностики. Метод task.set_name() позволяет назначить задаче имя, которое будет отображаться в логах и трейсбеках:

import asyncio


async def do_work(delay: int) -> None:
    await asyncio.sleep(delay)
    print(f"Работа завершена за {delay} секунд")


async def main() -> None:
    task1 = asyncio.create_task(do_work(2))
    task1.set_name("fetch_user_data")

    task2 = asyncio.create_task(do_work(3))
    task2.set_name("fetch_user_posts")

    print(f"Запущена задача: {task1.get_name()}")
    print(f"Запущена задача: {task2.get_name()}")

    await task1
    await task2


asyncio.run(main())

Альтернативно, имя можно задать сразу при создании задачи:

import asyncio


async def do_work(delay: int) -> None:
    await asyncio.sleep(delay)


async def main() -> None:
    task1 = asyncio.create_task(do_work(2), name="fetch_user_data")
    task2 = asyncio.create_task(do_work(3), name="fetch_user_posts")

    await task1
    await task2


asyncio.run(main())

Именование задач особенно полезно при диагностике утечек задач — ситуаций, когда задачи создаются, но никогда не завершаются и не отменяются. Для обнаружения таких проблем можно использовать функцию asyncio.all_tasks():

import asyncio


async def leaky_task() -> None:
    await asyncio.sleep(100)


async def main() -> None:
    # Создаём задачу, но не ждём её завершения
    asyncio.create_task(leaky_task(), name="potentially_leaked_task")

    # Проверяем активные задачи
    tasks = asyncio.all_tasks()
    for task in tasks:
        if not task.done():
            print(f"Незавершённая задача: {task.get_name()}")


asyncio.run(main())

Группы задач через TaskGroup

Требование версии: Функциональность TaskGroup доступна начиная с Python 3.11. Для более ранних версий используйте альтернативные подходы, описанные ниже.

В реальных программах часто возникает необходимость управлять не одной или двумя задачами, а целой их группой. Когда задачи логически связаны и должны выполняться вместе, удобнее управлять ими как единым блоком. Для этого в Python 3.11 появился инструмент asyncio.TaskGroup.

TaskGroup позволяет создавать несколько задач внутри контекстного менеджера. Все задачи, добавленные в группу, планируются на выполнение в цикле событий и начинают выполняться конкурентно в рамках одного прохода цикла. Это упрощает работу с кодом и делает его более надёжным, поскольку не нужно вручную собирать задачи и следить за тем, чтобы каждая из них была корректно завершена или обработана при ошибках.

Пример показывает, как можно запустить три задачи параллельно и дождаться их выполнения:

import asyncio


async def do_work(name: str, delay: int) -> None:
    print(f"{name} начата")
    await asyncio.sleep(delay)
    print(f"{name} завершена")


async def main() -> None:
    async with asyncio.TaskGroup() as tg:
        tg.create_task(do_work("Задача 1", 2))
        tg.create_task(do_work("Задача 2", 3))
        tg.create_task(do_work("Задача 3", 1))
    print("Все задачи из группы завершены")


asyncio.run(main())

При входе в блок async with создаётся группа задач, а при выходе из него программа ждёт, пока все задачи завершатся. Важная особенность: если одна из задач завершится с ошибкой, TaskGroup автоматически отменяет все остальные задачи в группе (так называемые «сиблинги»), отправляя им исключение CancelledError, которое будет обработано при следующем операторе await в каждой задаче. Все возникшие исключения затем оборачиваются в ExceptionGroup и пробрасываются наружу. Это особенно важно для поддержания целостности приложения и предотвращения «висячих» задач.

Работа с TaskGroup также удобна тем, что код остаётся компактным и читаемым. Не требуется вручную вызывать await для каждой задачи, как это было при использовании create_task(). Достаточно поместить все задачи в группу, и цикл событий выполнит их параллельно, контролируя их завершение.

Альтернатива для Python 3.10 и ниже

Если вы используете Python версии ниже 3.11, можно применить asyncio.gather() в сочетании с ручной обработкой отмены:

import asyncio


async def do_work(name: str, delay: int) -> None:
    print(f"{name} начата")
    await asyncio.sleep(delay)
    print(f"{name} завершена")


async def main() -> None:
    tasks = [
        asyncio.create_task(do_work("Задача 1", 2)),
        asyncio.create_task(do_work("Задача 2", 3)),
        asyncio.create_task(do_work("Задача 3", 1)),
    ]

    try:
        await asyncio.gather(*tasks)
    except Exception:
        # При ошибке отменяем все задачи
        for task in tasks:
            task.cancel()
        # Ждём завершения отмены
        await asyncio.gather(*tasks, return_exceptions=True)
        raise

    print("Все задачи завершены")


asyncio.run(main())

Однако этот подход требует больше кода и не предоставляет автоматическую обёртку исключений в ExceptionGroup.

Обработка ошибок с ExceptionGroup

Требование версии: Функциональность ExceptionGroup и синтаксис except* доступны начиная с Python 3.11.

Когда мы запускаем несколько задач одновременно с помощью TaskGroup, важно учитывать, что любая из них может завершиться с ошибкой. В асинхронном коде это особенно актуально, потому что сбой в одной задаче может повлиять на работу других. Если не обрабатывать ошибки правильно, можно оставить часть задач в подвешенном состоянии или потерять информацию о том, что пошло не так. Чтобы решать такие ситуации, в Python 3.11 появилась специальная структура — ExceptionGroup.

ExceptionGroup позволяет собрать несколько исключений, возникших в разных задачах, в один объект и передать его наружу. Это особенно полезно при работе с TaskGroup, когда несколько задач выполняются параллельно и более одной из них может упасть с ошибкой.

Есть важная деталь: TaskGroup по умолчанию прерывает работу всех остальных задач, как только первая из них завершилась с исключением. Это значит, что если одна задача упала раньше остальных, последующие будут автоматически отменены и могут не успеть сгенерировать свою ошибку. ExceptionGroup может собрать несколько исключений, если соответствующие задачи успели сгенерировать ошибки до того, как TaskGroup обработал первую ошибку и инициировал отмену остальных задач. Таким образом, мы можем поймать несколько ошибок, если они произошли до того, как цикл событий успел отменить остальные задачи.

Рассмотрим пример:

import asyncio


async def task_ok() -> None:
    await asyncio.sleep(1)
    print("Задача без ошибок завершена")


async def task_fail1() -> None:
    await asyncio.sleep(0.5)
    raise ValueError("Ошибка в первой задаче")


async def task_fail2() -> None:
    await asyncio.sleep(0.51)
    raise RuntimeError("Ошибка во второй задаче")


async def main() -> None:
    try:
        async with asyncio.TaskGroup() as tg:
            tg.create_task(task_ok())
            tg.create_task(task_fail1())
            tg.create_task(task_fail2())
    except* Exception as eg:
        print("Поймано ExceptionGroup")
        for e in eg.exceptions:
            print(f"Тип: {type(e).__name__}, сообщение: {e}")


asyncio.run(main())

В этом примере две задачи завершаются с ошибками с разницей в 0.01 секунды. Первая ошибка (ValueError) приводит к тому, что TaskGroup начинает отменять остальные задачи. Но поскольку вторая задача уже почти завершалась, её ошибка тоже успеет попасть в ExceptionGroup. Мы можем видеть оба исключения при обработке блока except*.

Если бы вторая задача выполнялась значительно дольше, её отменили бы до того, как она смогла бы выбросить ошибку, и тогда в ExceptionGroup оказалась бы только первая ошибка. Это поведение важно учитывать при проектировании асинхронных программ.

Обработка исключений через ExceptionGroup делает код более надёжным и предсказуемым. Она позволяет корректно реагировать на несколько сбоев одновременно и при этом гарантировать, что ресурсы освобождаются, а программа остаётся в устойчивом состоянии.

Отмена подзадач и корректное завершение

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

В Python отмена задач реализована через исключение asyncio.CancelledError. Когда вызывается метод task.cancel(), в корутину, которую выполняет задача, отправляется сигнал отмены. При ближайшем выполнении оператора await внутри этой корутины будет поднято исключение CancelledError. Если его не перехватить, корутина завершится, что и требуется в большинстве случаев.

Пример демонстрирует базовую отмену задачи:

import asyncio


async def worker() -> None:
    try:
        print("Начало работы")
        await asyncio.sleep(5)
        print("Работа завершена")
    except asyncio.CancelledError:
        print("Задача была отменена, выполняем очистку ресурсов")
        raise  # Важно пробросить исключение дальше


async def main() -> None:
    task = asyncio.create_task(worker())
    await asyncio.sleep(1)
    task.cancel()
    try:
        await task
    except asyncio.CancelledError:
        print("Подтверждение отмены")


asyncio.run(main())

Здесь задача worker() запускается и начинает выполнять долгую операцию. Через секунду её отменяют методом cancel(). Когда корутина снова достигает точки ожидания (await asyncio.sleep(5)), она прерывается исключением CancelledError, а блок except внутри задачи позволяет выполнить финальные действия — например, закрыть файл или соединение. После этого исключение снова пробрасывается наружу, чтобы родительская задача знала об отмене.

Важно помнить, что отмена в asyncio является кооперативной. Если корутина выполняет длительный синхронный расчет без точек await, она может не реагировать на отмену до завершения расчета или явной проверки на прерывание. Поэтому корректное проектирование задач с периодическими точками ожидания — обязательное условие для быстрой реакции на отмену.

При использовании TaskGroup отмена происходит автоматически: если одна из задач завершилась с ошибкой, остальные задачи будут отменены. Каждая из них получит CancelledError, и у нас есть возможность обработать его внутри самой задачи, чтобы корректно завершить работу и освободить ресурсы.

Пример демонстрирует отмену в TaskGroup:

import asyncio


async def task1() -> None:
    try:
        await asyncio.sleep(2)
        print("Задача 1 завершена")
    except asyncio.CancelledError:
        print("Задача 1 отменена")
        raise  # Важно пробросить исключение дальше


async def task2() -> None:
    await asyncio.sleep(1)
    raise RuntimeError("Ошибка в задаче 2")


async def main() -> None:
    try:
        async with asyncio.TaskGroup() as tg:
            tg.create_task(task1())
            tg.create_task(task2())
    except* Exception as eg:
        print("Поймано ExceptionGroup:")
        for i, e in enumerate(eg.exceptions):
            print(f"  Исключение {i + 1}: {type(e).__name__}: {e}")


asyncio.run(main())

В этом примере task2() выбрасывает исключение через секунду. Цикл событий сообщает TaskGroup, что нужно отменить все остальные задачи. task1() получает CancelledError и завершает свою работу с сообщением о корректной отмене. При этом исключение из task2() пробрасывается наружу и может быть обработано через except*.

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

В приведённом примере очистка ресурсов выполняется в блоке except asyncio.CancelledError, что подходит для демонстрации обработки отмены. Однако в реальных приложениях стоит учитывать более широкий контекст управления ресурсами.

Блок except хорошо подходит, когда нужно выполнить действия именно при отмене — например, записать в лог информацию о прерывании операции или отправить уведомление о незавершённой задаче. Но если речь идёт об освобождении ресурсов (закрытие файлов, соединений с базой данных, освобождение памяти), то более надёжным решением будет использование блока finally или асинхронных контекстных менеджеров.

Блок finally гарантирует очистку ресурсов при любом исходе: успешном завершении, отмене или любом другом исключении:

import asyncio


async def worker_with_finally() -> None:
    try:
        print("Начало работы")
        await asyncio.sleep(5)
        print("Работа завершена")
    finally:
        print("Очистка ресурсов (выполняется всегда)")


async def main() -> None:
    task = asyncio.create_task(worker_with_finally())
    await asyncio.sleep(1)
    task.cancel()
    try:
        await task
    except asyncio.CancelledError:
        print("Задача отменена")


asyncio.run(main())

Ещё лучшим решением является использование асинхронных контекстных менеджеров, которые автоматически управляют жизненным циклом ресурсов:

import asyncio
from collections.abc import AsyncGenerator
from contextlib import asynccontextmanager


@asynccontextmanager
async def database_connection() -> AsyncGenerator[str, None]:
    print("Открытие соединения с базой данных")
    connection = "db_connection"
    try:
        yield connection
    finally:
        print("Закрытие соединения с базой данных")
        await asyncio.sleep(0.1)


async def worker_with_context() -> None:
    async with database_connection() as conn:
        print(f"Работа с {conn}")
        await asyncio.sleep(5)


async def main() -> None:
    task = asyncio.create_task(worker_with_context())
    await asyncio.sleep(1)
    task.cancel()
    try:
        await task
    except asyncio.CancelledError:
        print("Задача отменена, но ресурсы освобождены")


asyncio.run(main())

Контекстные менеджеры особенно удобны при работе с файлами, сетевыми соединениями и другими ресурсами, требующими явного освобождения.

Контрольный список при работе с задачами

Каждая созданная задача должна быть либо ожидаема, либо отменена — не оставляйте «потерянные» задачи, которые продолжают выполняться в фоне

Сохраняйте ссылки на все созданные задачи — если задача создана через create_task(), но на неё нет ссылки, она может завершиться с ошибкой незаметно

Используйте TaskGroup для логически связанных задач (Python 3.11+) — это гарантирует автоматическую отмену всех задач при ошибке в одной из них

Всегда пробрасывайте CancelledError после обработки — не подавляйте это исключение, если не понимаете последствий

Используйте finally или контекстные менеджеры для освобождения ресурсов — это надёжнее, чем обработка в except

Давайте задачам осмысленные имена — используйте task.set_name() или параметр name при создании задачи для упрощения отладки

Проверяйте незавершённые задачи в конце программы — используйте asyncio.all_tasks() для диагностики утечек задач

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

import asyncio


async def fetch_data(url: str) -> str:
    await asyncio.sleep(1)
    return f"Данные с {url}"


async def main() -> None:
    # Хорошо: создаём задачи и сохраняем ссылки
    task1 = asyncio.create_task(fetch_data("api.example.com/users"), name="fetch_users")
    task2 = asyncio.create_task(fetch_data("api.example.com/posts"), name="fetch_posts")

    # Хорошо: ожидаем завершения всех задач
    results = await asyncio.gather(task1, task2)
    print(results)

    # Проверяем, нет ли утечек задач
    tasks = [t for t in asyncio.all_tasks() if not t.done() and t != asyncio.current_task()]
    if tasks:
        print(f"Внимание: найдено {len(tasks)} незавершённых задач")
        for task in tasks:
            print(f"  - {task.get_name()}")


if __name__ == "__main__":
    asyncio.run(main())