Поймать прерывание клавиатуры, чтобы остановить многопроцессорный рабочий Python от работы в очереди

Question

Из нескольких сообщений, найденных на stackru, я создал этот код.

сценарий

Я хочу иметь multiprocessing.queue, который несколько слушателей "слушают"

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

проблема

Моя проблема с текущей версией, где я использую

signal.signal(signal.SIGINT, signal.SIG_IGN) 

Игнорировать Ctrl + C означает, что он также игнорируется основным процессом.

Есть идеи? Нужно ли использовать многопроцессорный рабочий пул? Некоторые примеры показывают, что мне, возможно, придется. Могу ли я тогда использовать очередь?

from multiprocessing import Pool, Process,Queue
import time
import signal
# http://docs.python.org/3.1/library/multiprocessing.html#multiprocessing.Queue
# http://docs.python.org/3.1/library/multiprocessing.html#multiprocessing.Process


class Worker(Process):
    def __init__(self, queue,ident):
        super(Worker, self).__init__()
        # Ignore Signals
        signal.signal(signal.SIGINT, signal.SIG_IGN)
        self.queue= queue
        self.idstr= str(ident)
        print "Ident" + self.idstr

    def run(self):
        print 'Worker started'
        # do some initialization here

        print 'Computing things!'
        for data in iter( self.queue.get, None ):
            print "#" + self.idstr + " : " + str(data)
            time.sleep(5)
            print "#" + self.idstr + "Queue Size: " + str(self.queue.qsize())

        print "Worker Done"

#### Main ####
request_queue = Queue(10)

for i in range(4):
    Worker( request_queue,i ).start()

try:
    for data in range(1000000):
        request_queue.put( data )
        #print "Queue Size: " + str(request_queue.qsize())
        # Sentinel objects to allow clean shutdown: 1 per worker.
    for i in range(4):
        request_queue.put( None ) 

except KeyboardInterrupt:
    print "Caught KeyboardInterrupt, terminating workers"
    while  request_queue.empty()==False:
         request_queue.get()
    request_queue.put( None )    

Answer 1

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

global STOP

import os, signal
def signal_handler(sig, frame):
    global STOP
    if STOP:
        signal.signal(signal.SIGINT, signal.SIG_IGN)
        os.kill(os.getpid(), signal.SIGTERM)
    STOP = True
signal.signal(signal.SIGINT, signal_handler)

Answer 2

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

1. Я указал обработчик сигнала для сигнала прерывания.
2. После получения первого сигнала Sig INT я игнорирую сигнал SIG Int.
3. Я переключаю флаг остановки на TRUE
4. Я вырываю цикл вставки очереди

5. Я вызываю функцию остановки, которая очищает очередь и вставляет стражу

   from multiprocessing import Pool, Process,Queue
   import time
   import signal
   # http://docs.python.org/3.1/library/multiprocessing.html#multiprocessing.Queue
   # http://docs.python.org/3.1/library/multiprocessing.html#multiprocessing.Process
   
   # Stop Flag for loop
   stop = False
   
   # Define SIGINT
   def signal_handler(sig, frame):
       print 'You pressed Ctrl+C!'
       global stop
       stop = True
       # Ignore more Ctrl+C
       signal.signal(signal.SIGINT, signal.SIG_IGN) 
   
   signal.signal(signal.SIGINT, signal_handler)
   
   def stopSentinel(request_queue):
       print "CTRL Stop Queue and insert None"
   
   # Empty Existing Queue
   while  request_queue.empty()==False:
        request_queue.get()
   
   # Put One None for each Worker
   for i in range(4):
       request_queue.put( None ) 
   
   
   class Worker(Process):
       def __init__(self, queue,ident):
           super(Worker, self).__init__()
   
           self.queue= queue
           self.idstr= str(ident)
           print "Ident" + self.idstr
   
       def run(self):
           print 'Worker started'
           # do some initialization here
   
           print 'Computing things!'
           for data in iter( self.queue.get, None ):
               print "#" + self.idstr + " : " + str(data)
               time.sleep(5)
               print "#" + self.idstr + "Queue Size: " + str(self.queue.qsize())
   
           print "Worker Done"
   
   
   
   #### Main #####
   request_queue = Queue(10)
   
   for i in range(4):
       Worker( request_queue,i ).start()
   
   #### Fill Queue with Data ####
   for data in range(1000000):
       request_queue.put( data )
       #print "Queue Size: " + str(request_queue.qsize())
       # Sentinel objects to allow clean shutdown: 1 per worker.
   
       # Check for Stop
       print "Check Breakout"
       if stop == True:
           print "Stop Break"
           break
   
   if stop == True:
       stopSentinel(request_queue)
   else:       
       print "Normal Stop" 
       for i in range(4):
           request_queue.put( None )