python multiprocessing terminate

1You might run the child processes as daemons in the background.
2
3process.daemon = True
4Any errors and hangs (or an infinite loop) in a daemon process will not affect the main process, and it will only be terminated once the main process exits.
5
6This will work for simple problems until you run into a lot of child daemon processes which will keep reaping memories from the parent process without any explicit control.
7
8Best way is to set up a Queue to have all the child processes communicate to the parent process so that we can join them and clean up nicely. Here is some simple code that will check if a child processing is hanging (aka time.sleep(1000)), and send a message to the queue for the main process to take action on it:
9
10import multiprocessing as mp
11import time
12import queue
13
14running_flag = mp.Value("i", 1)
15
16def worker(running_flag, q):
17    count = 1
18    while True:
19        if running_flag.value:
20            print "working {0} ...".format(count)
21            count += 1
22            q.put(count)
23            time.sleep(1)
24            if count > 3:
25                # Simulate hanging with sleep
26                print "hanging..."
27                time.sleep(1000)
28
29def watchdog(q):
30    """
31    This check the queue for updates and send a signal to it
32    when the child process isn't sending anything for too long
33    """
34    while True:
35        try:
36            msg = q.get(timeout=10.0)
37        except queue.Empty as e:
38            print "[WATCHDOG]: Maybe WORKER is slacking"
39            q.put("KILL WORKER")
40
41def main():
42    """The main process"""
43    q = mp.Queue()
44
45    workr = mp.Process(target=worker, args=(running_flag, q))
46    wdog = mp.Process(target=watchdog, args=(q,))
47
48    # run the watchdog as daemon so it terminates with the main process
49    wdog.daemon = True
50
51    workr.start()
52    print "[MAIN]: starting process P1"
53    wdog.start()
54
55    # Poll the queue
56    while True:
57        msg = q.get()
58        if msg == "KILL WATCHDOG":
59            print "[MAIN]: Terminating slacking WORKER"
60            workr.terminate()
61            time.sleep(0.1)
62            if not workr.is_alive():
63                print "[MAIN]: WORKER is a goner"
64                workr.join(timeout=1.0)
65                print "[MAIN]: Joined WORKER successfully!"
66                q.close()
67                break # watchdog process daemon gets terminated
68
69if __name__ == '__main__':
70    main()
71Without terminating worker, attempt to join() it to the main process would have blocked forever since worker has never finished.