Отмена задач: cancel и CancelledError

В прошлом уроке таймаут отменял корутину за кулисами. Теперь берём отмену в свои руки: пользователь закрыл вкладку — отменяем загрузку, пришёл новый апдейт — отменяем устаревший расчёт, кнопка «стоп» — останавливаем воркеры. У всего этого один механизм: метод task.cancel() у задачи и исключение CancelledError внутри корутины.

Главное, что стоит понять заранее: отмена в asyncio кооперативная. cancel() — это не выстрел, а вежливая просьба, доставляемая письмом с именем CancelledError. Задача сама решает, где остановится, — на ближайшем await — и сама прибирается. Отменить корутину силой нельзя; можно только попросить и дождаться.

Где вы встретитесь с отменой вживую? Почти в каждом асинхронном сервисе: пользователь нажал «стоп» — бот отменяет задачу рассылки; клиент закрыл вкладку — веб-сервер отменяет обработку запроса, чтобы не жечь ресурсы впустую; обход каталога нашёл нужное раньше конца — оставшиеся URL отменяются. И в каждом случае вопрос тот же: кто закроет файлы и снимет флаги, если работа остановилась на середине? На него и отвечает этот урок.

Первое знакомство: cancel и уборка в finally

Запустим «загрузку», которая собиралась работать 30 секунд, и отменим её через секунду. Уборку — закрытие файла, снятие флага — поместим в finally: этот блок выполняется и на успехе, и на падении, и на отмене:

import asyncio

async def downloader():
    try:
        print("Загрузка началась")
        await asyncio.sleep(30)               # очень долгая работа
        print("Загрузка завершилась сама")    # до этой строки не дойдёт
    finally:
        print("Уборка: закрыл файл и снял флаг")

async def main():
    task = asyncio.create_task(downloader())
    await asyncio.sleep(1)
    print("Передумали - отменяю")
    task.cancel()
    try:
        await task
    except asyncio.CancelledError:
        print("Задача подтвердила отмену")

asyncio.run(main())

Загрузка началась
Передумали - отменяю
Уборка: закрыл файл и снял флаг
Задача подтвердила отмену

Разбираем цепочку. task.cancel() доставляет CancelledError внутрь downloader — прямо на его await asyncio.sleep. Строка «Загрузка завершилась сама» не выполняется никогда: работа прервана. finally при этом честно отрабатывает — ресурс закрыт. Затем отменённая задача пробрасывает CancelledError дальше, и await task в main ловит его: так отменяющая сторона узнаёт, что задача действительно остановилась, а не успела завершиться.

Обрати внимание: await task нужен. cancel() только отправляет запрос — настоящий останов произойдёт, когда корутина дойдёт до своего await и получит исключение. Без await task вы не узнаете, когда уборка закончилась, а Python может ругаться на нечитанную отмену. Дождаться задачи после cancel() — та же гигиена, что закрыть файл после открытия.

Дожидание важно и для порядка запуска: следующий этап должен стартовать только после подтверждённой остановки предыдущего. Классика — перезапуск воркера с новыми настройками: отменили, дождались CancelledError (значит, соединения закрыты и флаги сняты), и только потом подняли новый. Если не дожидаться, два экземпляра воркера какое-то время живут параллельно и наступают друг другу на общий ресурс — баг, который воспроизводится «через раз» и отравляет жизнь именно этой непредсказуемостью.

Питфолл: проглоченная отмена

А что, если корутина поймает CancelledError и не будет его пробрасывать? Формально всё сработает: исключение перехвачено, программа не упала. По смыслу — катастрофа: отмена потеряна. Проверим, что увидит вызывающий код:

import asyncio

async def stubborn():
    try:
        await asyncio.sleep(30)
    except asyncio.CancelledError:
        print("Отмену поймал и проглотил")
        # нет raise - задача вернётся как успешная

async def main():
    task = asyncio.create_task(stubborn())
    await asyncio.sleep(1)
    task.cancel()
    try:
        await task
        print("await прошёл тихо: отмена потеряна")
    except asyncio.CancelledError:
        print("Отмена дошла до вызывающего")

asyncio.run(main())

Отмену поймал и проглотил
await прошёл тихо: отмена потеряна

Правильный паттерн: уборка и raise

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

import asyncio

async def worker():
    try:
        for i in range(1, 6):
            print("шаг", i)
            await asyncio.sleep(1)
    except asyncio.CancelledError:
        print("Останавливаюсь, сохраняю прогресс")
        raise                                 # отмена обязана продолжить путь

async def main():
    task = asyncio.create_task(worker())
    await asyncio.sleep(2)
    task.cancel()
    try:
        await task
    except asyncio.CancelledError:
        print("Задача остановлена корректно")

asyncio.run(main())

шаг 1
шаг 2
Останавливаюсь, сохраняю прогресс
Задача остановлена корректно

Два шага — и отмена: воркер успел напечатать шаг 1 и шаг 2, после чего вместо шага 3 получил CancelledError, оставил пометку о прогрессе и передал исключение дальше. Так строятся остановимые конвейеры: воркер ничего не знает о том, кто и зачем его отменяет, — он просто прибирается и подчиняется.

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

Останавливаем пачку и проверяем честность

В реальных программах отменяют не по одной задаче, а пуулом: остановили бота — гасим всех воркеров. Каждому cancel(), затем общий await через gather с return_exceptions=True — флаг из урока 11 нужен, потому что отменённые задачи вернут CancelledError, и gather без флага упал бы на первой же:

import asyncio

async def worker(name):
    try:
        print(name, "работает")
        await asyncio.sleep(30)
    finally:
        print(name, "прибралась")

async def main():
    tasks = [asyncio.create_task(worker(n)) for n in ("Альфа", "Бета", "Гамма")]
    await asyncio.sleep(1)
    for t in tasks:
        t.cancel()
    await asyncio.gather(*tasks, return_exceptions=True)
    print("Все задачи остановлены")

asyncio.run(main())

Альфа работает
Бета работает
Гамма работает
Альфа прибралась
Бета прибралась
Гамма прибралась
Все задачи остановлены

Полезно знать и обратную ситуацию: отменять нечего. cancel() по уже завершившейся задаче безвреден — он возвращает False, и ничего не происходит. Это позволяет писать код остановки без проверок «а жива ли ещё задача»:

Заглянем под капот массовой остановки: CancelledError доставляются всем задачам почти одновременно, поэтому и уборки идут параллельно — каждая задача прибирается в своём темпе, и gather собирает их по мере готовности. Никакой согласованной очереди выключения нет: если одной задаче важно выключиться после другой (например, сначала закрыть приём заданий, потом воркеров), этот порядок выстраивают вручную — отдельными await, как в конвейере из урока 13, где продюсера дожидают до отмены воркеров. cancel() ничего не знает о ваших зависимостях — он просто раздаёт просьбы остановиться.

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

import asyncio

async def quick():
    await asyncio.sleep(1)
    return "готово"

async def main():
    task = asyncio.create_task(quick())
    await asyncio.sleep(2)            # задача уже завершилась сама
    print("cancel вернул:", task.cancel())
    print("Результат:", task.result())

asyncio.run(main())

cancel вернул: False
Результат: готово

Обратный полюс: отменить можно и задачу, которая ещё не сделала ни одного шага. cancel() до первого запуска сработает чисто: CancelledError будет доставлен корутине в её первом же шаге, и она не выполнит ни строчки:

import asyncio

async def job():
    print("эта строка не напечатается")
    await asyncio.sleep(1)

async def main():
    task = asyncio.create_task(job())
    task.cancel()                     # задача ещё не сделала ни шага
    try:
        await task
    except asyncio.CancelledError:
        print("Отменена до старта: ни шага не сделано")

asyncio.run(main())

Отменена до старта: ни шага не сделано

Уборка с флагами: состояние после отмены

Брифинг перед боем: отменённая задача оставляет после себя состояние — открытые файлы, поднятые флаги, арендованные подключения. finally — место, где состояние приводится в порядок. Проверим это явно: заведём объект с двумя флагами и убедимся, что после отмены оба сняты:

import asyncio

class Job:
    def __init__(self):
        self.file_open = True
        self.busy = True

async def worker(job):
    try:
        print("работаю, файл открыт:", job.file_open)
        await asyncio.sleep(30)
    finally:
        job.file_open = False
        job.busy = False
        print("уборка: файл закрыт, флаг снят")

async def main():
    job = Job()
    task = asyncio.create_task(worker(job))
    await asyncio.sleep(1)
    task.cancel()
    try:
        await task
    except asyncio.CancelledError:
        pass
    print("После отмены: файл открыт -", job.file_open, "| занят -", job.busy)

asyncio.run(main())

работаю, файл открыт: True
уборка: файл закрыт, флаг снят
После отмены: файл открыт - False | занят - False

Та же гигиена в боевом коде: сессии, файлы и соединения закрываются в finally — независимо от причины выхода. Фрагмент настоящего загрузчика (сети в песочнице нет — это скелет для локального запуска):

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

import asyncio
import aiohttp

async def download(url):
    session = aiohttp.ClientSession()
    try:
        async with session.get(url) as resp:
            return await resp.text()
    finally:
        await session.close()       # выполнится и при отмене

async def main():
    task = asyncio.create_task(download("https://example.com/"))
    await asyncio.sleep(2)
    task.cancel()

asyncio.run(main())

Что дальше

Отмена — это исключение, летящее по обычным правилам. Значит, в конкурентном коде летают и обычные ошибки: ValueError из одной корутины способен уронить весь gather. Следующий урок — исключения в конкурентном коде: как изолировать падения, собирать статистику и не дать одной сломанной задаче топить пачку. А применение всего этого вживую — в разделе про aiogram, где остановка polling отменяет воркеры-хендлеры точно так же.

Перед тем как двинуться дальше, короткая самопроверка: возьмите свою последнюю корутину с ресурсом — файлом, сессией, флагом — и мысленно отмените её на самом середине. Что останется после? Если уверенного ответа «чистое состояние» нет, вы знаете, какую строку кода открыть первой. Отмена — навык, а не теория.

Отмена в asyncio кооперативная: cancel() лишь просит остановиться, а убирается задача сама — в finally, который работает при любом исходе.