Gracefully terminate multiprocessing based program

Question

I am working on a python service that spawns Process to handle the workload. Since I don't know at the start of the service how many workers I need, I chose to not use Pool. The following is a simplified version:

import multiprocessing as mp
import time
from datetime import datetime

def _print(s): # just my cheap logging utility
    print(f'{datetime.now()} - {s}')

def run_in_process(q, evt):
    _print(f'starting process job')
    while not evt.is_set(): # True
        try:
            x = q.get(timeout=2)
            _print(f'received {x}')
        except:
            _print(f'timed-out')


if __name__ == '__main__':

    with mp.Manager() as manager:
        q = manager.Queue()
        evt = manager.Event()
        p = mp.Process(target=run_in_process, args=(q, evt))
        p.start()
        time.sleep(2)

        data = 100
        while True:
            try:
                q.put(data)
                time.sleep(0.5)
                data += 1
                if data > 110:
                    break
            except KeyboardInterrupt:
                _print('finishing...')
                #p.terminate()
                break

        time.sleep(3)
        _print('setting event 0')
        evt.set()
        _print('joining process')
        p.join()

        _print('done')

The program works and exits gracefully, without any error messages. However, if I use Ctrl-C before I have all 10 events processed, I get the following error before it exits.

2022-04-01 12:41:06.866484 - received 101
2022-04-01 12:41:07.367628 - received 102
^C2022-04-01 12:41:07.507805 - timed-out
2022-04-01 12:41:07.507886 - finishing...
Process Process-2:
Traceback (most recent call last):
  File "//python3.7/multiprocessing/process.py", line 297, in _bootstrap
    self.run()
  File "//python3.7/multiprocessing/process.py", line 99, in run
    self._target(*self._args, **self._kwargs)
  File "mp.py", line 10, in run_in_process
    while not evt.is_set(): # True
  File "//python3.7/multiprocessing/managers.py", line 1088, in is_set
    return self._callmethod('is_set')
  File "//python3.7/multiprocessing/managers.py", line 819, in _callmethod
    kind, result = conn.recv()
  File "//python3.7/multiprocessing/connection.py", line 250, in recv
    buf = self._recv_bytes()
  File "//python3.7/multiprocessing/connection.py", line 407, in _recv_bytes
    buf = self._recv(4)
  File "//python3.7/multiprocessing/connection.py", line 379, in _recv
    chunk = read(handle, remaining)
ConnectionResetError: [Errno 104] Connection reset by peer
2022-04-01 12:41:10.511334 - setting event 0
Traceback (most recent call last):
  File "mp.py", line 42, in 
    evt.set()
  File "//python3.7/multiprocessing/managers.py", line 1090, in set
    return self._callmethod('set')
  File "//python3.7/multiprocessing/managers.py", line 818, in _callmethod
    conn.send((self._id, methodname, args, kwds))
  File "//python3.7/multiprocessing/connection.py", line 206, in send
    self._send_bytes(_ForkingPickler.dumps(obj))
  File "//python3.7/multiprocessing/connection.py", line 404, in _send_bytes
    self._send(header + buf)
  File "//python3.7/multiprocessing/connection.py", line 368, in _send
    n = write(self._handle, buf)
BrokenPipeError: [Errno 32] Broken pipe

A few observations:

The double error message looks exactly the same when I press Ctrl-C with my actual project. I think this is a good representation of my problem.
If I add p.terminate(), it doesn't change the behavior if the program is left to finish by itself. But if I press Ctrl-C halfway, I encounter the error message only once, I guess it's from the main thread/process.
If I change while not evt.is_set(): in run_in_process to an infinite loop: while Tre: and let the program finish its course I would continue to see periodic time-out prints which make sense. What I don't understand is that, if I press Ctrl-C, then the terminal will start spewing time-out without any time gap between them. What happened?

My ultimate question is: what is the correct way of construct this program so that when Ctrl-C is used (or a termination signal is generated to the program for that matter), the program stops gracefully?

user_4190567 · Answer

I found out a solution to this problem myself by using signal.

The idea is to set up a signal catcher to catch specific signals, such as signal.SIGINT, signal.SIGTERM.

import multiprocessing as mp
from threading import Event
import signal

if __name__ == '__main__':
    main_evt = Event()
    
    def stop_main_handler(signum, frame):
        if not main_evt.is_set():
            main_evt.set()

    signal.signal(signal.SIGINT, stop_main_handler)
    with mp.Manager() as manager:
        # creating mp queue, event and process
        q = manager.Queue()
        evt = manager.Event()
        p = mp.Process(target=..., args=(q, evt))
        p.start()
        while not main_evt.is_set():
            # processing data
        # cleanup
        evt.set()
        p.join()

Or you can wrap it in an object-oriented fashion:

class SignalCatcher(object):
    def __init__(self):
        self._main_evt = Event()
    
    def _stop_handler(self, signum, frame):
        if not self._main_evt.is_set():
            self._main_evt.set()
 
    def block_until_signaled(self):
        while not self._main_evt.is_set()
            time.sleep(2)

Then you can use it as follows:

if __name__ == '__main__':
    sc = SignalCatcher()
    # this has to be outside. It seems that there is another process 
    # created by multiprocessing library, if you put sc creation in
    # with-context, it would fail to signal each process. 
    with mp.Manager() as manager:
        # creating process and starting it
        # ...
        sc.block_until_signaled()
        
        # cleanup
        # ...

Gracefully terminate multiprocessing based program

1 Answers