Catch Keyboard Interrupt to stop Python multiprocessing worker from working on queue

Question

From several posts found on stackoverflow i created this code.

Scenario

I want to have a multiprocessing.queue on which several worker "listen"

In case of a keyboard interrupt, the main process should no longer put new items in the queue and with the help of the sentinel objects, the worker should gracefully be stopped.

Problem

My problem with the current version where i use

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

To ignore the Ctrl + C is that it is also ignored by the main process.

Any Ideas ? Do I need to use the multiprocessing worker pool ? Some examples indicate that i might have to. Can I then still use the queue ?

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 )

Have a look at this http://stackoverflow.com/questions/4205317/capture-keyboardinterrupt-in-python-without-try-except — Paco, Jun 26 '13 at 10:42
hey paco, yes i found it and i was thinking about how to integrate it. so That the queue would stopped be filling and the workers will be stopped. At a first thought i came to the conclusion that i still want to have the exception in the main thread — Dukeatcoding, Jun 26 '13 at 10:47

score 1 · Answer 1 · answered Dec 29 '13 at 17:21

Based on your solution (which is good), I added an additional layer of protection incase the main code is unresponsive and the user cancels twice:

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)

score 0 · Answer 2 · edited Dec 16 '21 at 06:49

I think I found a solution. Still I don't like that I get the SIGINT 1 time from main and 4 times from the Worker, but maybe I have to live with that.

I specified a signal handler for the Interrupt Signal.
After Receiving the first Sig INT I ignore more SIG Int signal
I switch the stop flag to TRUE
I break out the queue insert loop

I call stop function which clears the queue and inserts the stop sentinels

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 )

Catch Keyboard Interrupt to stop Python multiprocessing worker from working on queue

2 Answers2

Linked