Нужно спроектировать надёжный graceful shutdown для фонового воркера на asyncio/threading, у которого в момент получения сигнала остановки есть несколько активных задач в разных стадиях выполнения. Какие аспекты нужно учесть при проектировании такого механизма? Обсудите компромиссы, риски и критерии выбора — единственного правильного решения может не быть.

Короткий ответ

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

Как это работает подробнее

Обработчик сигнала (SIGTERM от оркестратора) должен перестать принимать новые задачи, дать текущим ограниченное время на завершение (через asyncio.wait_for или join(timeout=...)), а затем решить по оставшимся: принудительно отменить (cancel для корутин, флаг для потоков, так как потоки нельзя прервать насильно) или считать их «в процессе» и полагаться на внешний механизм переподхвата.

• Компромисс. Короткий таймаут graceful shutdown ускоряет деплои и освобождение ресурсов, но увеличивает риск принудительного прерывания задач, которые почти завершились; длинный таймаут снижает этот риск, но замедляет выкатки и может конфликтовать с таймаутами оркестратора — выбор конкретных значений зависит от идемпотентности задач, частоты деплоев и допустимой задержки при остановке.

Пример кода

example-284.pyКопировать

import asyncio, signal
async def worker_loop(stop_event):
    while not stop_event.is_set():
        await process_next_task()
async def main():
    stop_event = asyncio.Event()
    loop = asyncio.get_running_loop()
    for sig in (signal.SIGTERM, signal.SIGINT):
        loop.add_signal_handler(sig, stop_event.set)
    task = asyncio.create_task(worker_loop(stop_event))
    await stop_event.wait()  # сигнал получен, прекращаем принимать новые итерации
    try:
        await asyncio.wait_for(task, timeout=10)  # ждём graceful завершения текущей задачи
    except asyncio.TimeoutError:
        task.cancel()  # принудительная отмена, если не успели вовремя
asyncio.run(main())

Что использовать на практике

• Практика. Для задач, читающих сообщения из брокера, важно не подтверждать (ack) сообщение, пока обработка не завершена — тогда прерванная задача просто вернётся в очередь и будет переобработана, что требует идемпотентности обработки. Для потоков единственный надёжный способ — кооперативная проверка флага остановки внутри цикла работы на естественных границах между шагами.

Подводные камни

• Риски. Используют принудительное убийство процесса сразу без фазы graceful-периода, из-за чего задачи прерываются в произвольный момент без шанса на корректную очистку; не продумывают идемпотентность задач, из-за чего повторно обработанная задача производит побочный эффект (например, отправку письма) дважды.