Python как убить потоки заблокированные на очереди сигналами?

Question

Я запускаю несколько потоков, работающих в очереди, и я хочу уничтожить их при отправке SIGINT (Ctrl + C). Каков наилучший способ справиться с этим?

targets = Queue.Queue()
threads_num = 10
threads = []

for i in threads_num:
    t = MyThread()
    t.setDaemon(True)
    threads.append(t)
    t.start()

targets.join()

Answer 1

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

Таким образом, вы можете использовать метод join потока - который поддерживает тайм-аут и не блокирует исключения - вместо того, чтобы ждать в методе join очереди.

Другими словами, сделайте что-то вроде этого:

term = Thread(target=someQueueVar.join)
term.daemon = True
term.start()
while (term.isAlive()):
    term.join(3600)

Теперь Ctrl + C завершит MainThread, после чего интерпретатор Python жестко убивает все потоки, помеченные как «демоны». Обратите внимание, что это означает, что вы должны установить «Thread.daemon» для всех других потоков или корректно завершить их, перехватив правильное исключение (KeyboardInterrupt или SystemExit) и выполнив все, что нужно для их выхода.

Также обратите внимание, что абсолютно необходимо , чтобы передать число в term.join(), так как в противном случае оно также будет игнорировать все исключения. Вы можете выбрать произвольно большое число.

Answer 2

Разве Ctrl + C SIGINT?

В любом случае вы можете установить обработчик для соответствующего сигнала, а в обработчике:

• установить глобальный флаг, который инструктирует рабочих выходить, и удостовериться, что они периодически проверяют его
• или поставить 10 жетонов отключения в очередь, и заставить рабочих выходить, когда они выбрасывают эту магиютокен
• или установите флаг, который указывает основному потоку выдавать эти токены, убедитесь, что основной поток проверяет этот флаг

и т. д.В основном это зависит от структуры приложения, которое вы прерываете.

Answer 3

Один из способов сделать это - установить обработчик сигнала для SIGTERM, который напрямую вызывает os._exit(signal.SIGTERM).Однако, если вы не укажете необязательный аргумент timeout для Queue.get, функция-обработчик сигнала не будет работать до тех пор, пока не вернется метод get.(Это совершенно недокументировано; я обнаружил это самостоятельно.) Таким образом, вы можете указать sys.maxint в качестве тайм-аута и поместить свой вызов Queue.get в цикл повторных попыток, чтобы чистота могла обойти это.

Answer 4

Мне удалось решить проблему, опустошив очередь на KeyboardInterrupt и позволив потокам изящно остановить себя.

Я не знаю, является ли это лучшим способом справиться с этим, но он прост и довольно чист.

targets = Queue.Queue()
threads_num = 10
threads = []

for i in threads_num:
    t = MyThread()
    t.setDaemon(True)
    threads.append(t)
    t.start()

while True:
    try:
        # If the queue is empty exit loop
        if self.targets.empty() is True:
            break

    # KeyboardInterrupt handler
    except KeyboardInterrupt:
        print "[X] Interrupt! Killing threads..."
        # Substitute the old queue with a new empty one and exit loop
        targets = Queue.Queue()
        break

# Join every thread on the queue normally
targets.join()

Answer 5

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