Управление задачами и подпроцессами ThreadPoolExecutor сигналами UNIX

Практическое решение для Python: управление ThreadPoolExecutor и подпроцессами через SIGUSR1 и SIGTERM. Остановка планирования и корректное завершение задач.

Управление пулом долго работающих заданий-подпроцессов уже после запуска — распространенная задача: приостановить планирование, дождаться завершения текущей работы и затем возобновить выполнение или корректно выйти. Когда в Python для отправки вызовов подпроцессов используется ThreadPoolExecutor, первое желание — привязать горячую клавишу. Однако в терминалах WSL библиотеки-хуки вроде pynput или keyboard могут не получать нажатия как ожидается, а блокирующий ввод неприемлем. Подход ниже заменяет хоткеи сигналами UNIX — это надежный и удобный для автоматизации способ управления во время выполнения.

Минимальный пример, показывающий проблему управления

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

import concurrent.futures
import subprocess
import sys
import time


def do_work(arg):
    # Имитируем медленный вызов подпроцесса
    return subprocess.run([
        sys.executable,
        "-c",
        f"import time; time.sleep({arg})"
    ])


def main():
    durations = [5, 4, 6, 3, 5, 4]
    with concurrent.futures.ThreadPoolExecutor(max_workers=3) as pool:
        future_list = [pool.submit(do_work, d) for d in durations]
        for fut in concurrent.futures.as_completed(future_list):
            _ = fut.result()


if __name__ == "__main__":
    main()

После того как пул начинает отправлять задачи, в терминале WSL нельзя неблокирующим образом приостановить или остановить планирование следующей работы простым нажатием клавиши. Библиотеки, слушающие глобальные хоткеи, могут не видеть нажатия в этой среде, а использование input блокирует основной поток. Сигнализация через файлы здесь тоже нежелательна.

Что на самом деле происходит

Суть проблемы — не в самом executor, а в канале управления. Хуки горячих клавиш могут не доставлять события в процесс Python, запущенный в терминале WSL, и не остается надежного способа менять состояние без блокирующего чтения или опроса файлов. Как только задания отправлены, выполнение идет по принципу «запустил и забыл».

Решение: сигналы UNIX, имя процесса и явные обработчики

Практичный путь — управлять процессом Python снаружи с помощью сигналов UNIX. Сначала задайте процессу отличимое имя через setproctitle, чтобы можно было целиться по имени. Затем добавьте обработчики для SIGTERM и SIGUSR1. Сигналом SIGUSR1 можно приказать планировщику прекратить постановку новых задач, позволив текущим подпроцессам завершиться. Сигналом SIGTERM — выйти корректно. Поскольку у процесса есть узнаваемое имя, достаточно pkill -USR1 -f [exe_name], чтобы послать сигнал без ручного поиска PID.

Полная реализация

Код ниже задает имя процесса, реагирует на SIGUSR1 остановкой дальнейшего планирования и на SIGTERM — корректным завершением. Запущенные подпроцессы дожидаются окончания. Логика соответствует описанному подходу и делает управление явным и наблюдаемым.

import concurrent.futures
import signal
import subprocess
import sys
import threading
import time
from setproctitle import setproctitle


# Глобальные флаги управления, которые меняются обработчиками сигналов
halt_queue = threading.Event()
terminate_run = threading.Event()


def on_usr1(sig, frame):
    # Остановить постановку новых задач; текущие — дать завершиться
    halt_queue.set()


def on_term(sig, frame):
    # Запросить корректную остановку: прекратить добавление задач и перейти к завершению
    halt_queue.set()
    terminate_run.set()


def worker_job(delay_s):
    # Имитируем медленный вызов подпроцесса, не прерывая его при управляющих событиях
    return subprocess.run([
        sys.executable,
        "-c",
        f"import time; time.sleep({delay_s})"
    ])


def orchestrate():
    # Присваиваем процессу узнаваемое имя, чтобы нацеливаться через pkill -f
    setproctitle("batch-runner-example")

    # Подключаем обработчики сигналов
    signal.signal(signal.SIGUSR1, on_usr1)
    signal.signal(signal.SIGTERM, on_term)

    tasks = [5, 4, 6, 3, 5, 4, 7, 2]

    with concurrent.futures.ThreadPoolExecutor(max_workers=3) as exe:
        submitted = []

        for item in tasks:
            if halt_queue.is_set():
                break
            fut = exe.submit(worker_job, item)
            submitted.append(fut)

        for fut in concurrent.futures.as_completed(submitted):
            _ = fut.result()

    # Необязательная логика после завершения всех отправленных задач
    if terminate_run.is_set():
        # Корректное завершение уже выполнено
        pass


if __name__ == "__main__":
    orchestrate()

С этим подходом управление выполняется «снаружи внутрь» и легко автоматизируется. Один раз задайте имя процесса и отправляйте сигналы по имени — без блокировки stdin и без надежды на слушатели клавиш в терминале.

Как отправлять сигналы по имени

Сигналы можно отправлять, не разыскивая PID. Используйте имя процесса, заданное в скрипте. SIGUSR1 останавливает постановку новых задач, позволяя текущим подпроцессам завершиться. SIGTERM по умолчанию обеспечивает корректную остановку тем же механизмом.

# Приостановить постановку новых задач (текущие — дать завершиться)
pkill -USR1 -f batch-runner-example

# Запросить корректное завершение (сигнал по умолчанию — SIGTERM)
pkill -f batch-runner-example

При желании можно найти процесс через ps и отправить сигнал напрямую с помощью kill.

В Linux можно использовать ps, чтобы найти process_id, а затем kill process_id для отправки сигнала (по умолчанию SIGTERM). Приложение должно иметь обработчик сигналов, чтобы корректно отреагировать.

Как правило, сигналы, отправленные процессу Python, обрабатываются через обработчик сигналов (модуль стандартной библиотеки signal).

В качестве альтернативы самому параллельному исполнению некоторые практики рекомендуют GNU Parallel для запуска скриптов в несколько потоков. Подход выше делает акцент на управлении со стороны Python через сигналы.

Почему это важно

Сигналы Unix дают предсказуемый и малозатратный контроль там, где хуки клавиатуры ненадежны или нежелательны. Вы избегаете Ctrl+C, который немедленно убил бы подпроцессы, и обходите блокирующий ввод или опрос файлов. В итоге нагрузку можно приостанавливать извне и безопасно «сливать» без радикальных изменений модели исполнения.

Итоги

Если нужно приостанавливать планирование или корректно завершать работу в ThreadPoolExecutor, запускающем подпроцессы, отдавайте предпочтение сигналам, а не слушателям хоткеев в терминалах WSL/Linux. Задайте отдельное имя процесса с помощью setproctitle, добавьте обработчики SIGUSR1 и SIGTERM и контролируйте отправку задач через общий флаг. Управляйте выполнением командой pkill -USR1 -f [exe_name], чтобы прекратить постановку новых задач, и используйте стандартный SIGTERM для упорядоченного выхода. Так плоскость управления останется простой, сценаризуемой и устойчивой для длинных пакетов задач.