Отмена asyncio.to_thread: как дождаться реального окончания работы

Корутина перестала ждать результат asyncio.to_thread, но файл продолжает обрабатываться. cancel() отменяет ожидание задачи; уже запущенная функция в потоке продолжает работать. Если считать её остановленной, следующий запуск может начать ту же работу одновременно.

Развести два срока жизни

У асинхронного ожидания и синхронной функции в потоке разные сроки жизни. Поток нельзя безопасно оборвать в произвольной инструкции средствами asyncio. Функция должна сама поддерживать остановку, например периодически проверять threading.Event. Если она вызывает чужую библиотеку, используйте её собственный таймаут или механизм отмены.

Пример не выполняет внешних действий. Поток ждёт сигнала release, затем фиксирует завершение. Асинхронный код отменяет Task после подтверждённого старта потока. Сразу после получения CancelledError функция ещё не закончилась; после release она завершается. События делают этот порядок наблюдаемым без угадывания миллисекунд.

Не удерживать цикл событий синхронным wait

Сам поток сообщает о старте через loop.call_soon_threadsafe(started.set). Это безопасно переносит уведомление в цикл событий. Ожидание завершения threading.Event вынесено в to_thread, поэтому основной цикл остаётся свободным. В finally обязательно разрешаем рабочему потоку выйти даже при сбое проверок.

В этом примере отмена означает «результат больше не нужен». Если смысл другой — «прекрати обработку», передавайте отдельный stop Event в рабочую функцию и проверяйте его между частями работы. Вызывать cancel повторно вместо такого протокола бесполезно.

import asyncio
import threading

async def main():
    loop = asyncio.get_running_loop()
    started = asyncio.Event()
    release = threading.Event()
    finished = threading.Event()
    def blocking_work():
        loop.call_soon_threadsafe(started.set)
        release.wait()
        finished.set()
        return 42
    task = asyncio.create_task(asyncio.to_thread(blocking_work))
    try:
        await started.wait()
        task.cancel()
        try:
            await task
        except asyncio.CancelledError:
            pass
        else:
            raise AssertionError("waiting task must be cancelled")
        assert not finished.is_set()
    finally:
        release.set()
    assert await asyncio.to_thread(finished.wait, 2)
    print("thread completed after cancelled wait")

asyncio.run(main())

Что сохранять вокруг побочного эффекта

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

Для CPU-нагрузки выбор to_thread также требует отдельного решения: обычный Python-код ограничен устройством интерпретатора и GIL в соответствующей сборке. Здесь цель механизма — вынести блокирующее ожидание. Пределы параллелизма и протокол завершения стоит задавать независимо.

Проверить порядок событий

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