Handling Ctrl+C with a Python Multiprocessing Pool • quietin

In theory, this applies to multiprocessing.dummy pools as well.

In Python 2.x, a function-based multiprocessing pool can enter kernel mode after join. Pressing Ctrl+C then cannot stop every process and exit. You have to press Ctrl+Z, find the remaining child processes, and kill them. First, here is code for which Ctrl+C does not work:

#!/usr/bin/env python
import multiprocessing
import os
import time

def do_work(x):
    print 'Work Started: %s' % os.getpid()
    time.sleep(10)
    return x * x

def main():
    pool = multiprocessing.Pool(4)
    try:
        result = pool.map_async(do_work, range(8))
        pool.close()
        pool.join()
        print result
    except KeyboardInterrupt:
        print 'parent received control-c'
        pool.terminate()
        pool.join()

if __name__ == "__main__":
    main()

After running this, pressing ^C kills none of the processes. Five processes remain: the parent plus four workers. Killing the parent makes all of them exit. Clearly, KeyboardInterrupt cannot be caught while using the pool this way. There are two solutions.

Option 1

The following is from pool.py in Python’s multiprocessing source. ApplyResult is the class a pool uses to hold the result of a function:

class ApplyResult(object):

    def __init__(self, cache, callback):
        self._cond = threading.Condition(threading.Lock())
        self._job = job_counter.next()
        self._cache = cache
        self._ready = False
        self._callback = callback
        cache[self._job] = self

This code also ignores ^C:

if __name__ == '__main__':
    import threading

    cond = threading.Condition(threading.Lock())
    cond.acquire()
    cond.wait()
    print "done"

Clearly, a threading.Condition(threading.Lock()) object cannot receive KeyboardInterrupt. A small change is enough: give cond.wait() a timeout. After map_async, pass that timeout through get by changing:

result = pool.map_async(do_work, range(4))

to:

result = pool.map_async(do_work, range(4)).get(1)

Then ^C can be received successfully. The value passed to get can be 1, 99999, or 0xffff.

Option 2

The other option is to write the process pool yourself with queues. Here is a piece of code to get a feel for it:

#!/usr/bin/env python
import multiprocessing, os, signal, time, Queue

def do_work():
    print 'Work Started: %d' % os.getpid()
    time.sleep(2)
    return 'Success'

def manual_function(job_queue, result_queue):
    signal.signal(signal.SIGINT, signal.SIG_IGN)
    while not job_queue.empty():
        try:
            job = job_queue.get(block=False)
            result_queue.put(do_work())
        except Queue.Empty:
            pass
        #except KeyboardInterrupt: pass

def main():
    job_queue = multiprocessing.Queue()
    result_queue = multiprocessing.Queue()

    for i in range(6):
        job_queue.put(None)

    workers = []
    for i in range(3):
        tmp = multiprocessing.Process(target=manual_function,
                                      args=(job_queue, result_queue))
        tmp.start()
        workers.append(tmp)

    try:
        for worker in workers:
            worker.join()
    except KeyboardInterrupt:
        print 'parent received ctrl-c'
        for worker in workers:
            worker.terminate()
            worker.join()

    while not result_queue.empty():
        print result_queue.get(block=False)

if __name__ == "__main__":
    main()

Option 3

Use a global eflag as a marker, bind SIG_INT to a handler that changes it, and make the worker function test eflag in its while condition. It can work, but honestly this approach is terrible. Threads are certainly viable here; processes would need a separately shared variable. I strongly do not recommend this approach.

A common wrong solution

This needs mentioning because I found people on SegmentFault being misled by it.

In theory, pass an initializer to the pool so child processes ignore SIGINT—that is, ^C—and then terminate the pool. The code is:

#!/usr/bin/env python
import multiprocessing
import os
import signal
import time

def init_worker():
    signal.signal(signal.SIGINT, signal.SIG_IGN)

def run_worker(x):
    print "child: %s" % os.getpid()
    time.sleep(20)
    return x * x

def main():
    pool = multiprocessing.Pool(4, init_worker)
    try:
        results = []
        print "Starting jobs"
        for x in range(8):
            results.append(pool.apply_async(run_worker, args=(x,)))

        time.sleep(5)
        pool.close()
        pool.join()
        print [x.get() for x in results]
    except KeyboardInterrupt:
        print "Caught KeyboardInterrupt, terminating workers"
        pool.terminate()
        pool.join()

if __name__ == "__main__":
    main()

But this only lets Ctrl+C interrupt while it is in time.sleep(5). Pressing ^C works during the first five seconds; once it reaches pool.join(), it stops working completely.

Recommendation

First, make sure multiprocessing is really necessary. For I/O-heavy programs, use multithreading or coroutines; for very CPU-heavy work, use multiprocessing. If you do need multiprocessing, concurrent.futures from Python 3 (also installable on Python 2.x) makes multithreaded and multiprocess code simpler. Its API is somewhat like Java’s concurrent framework.

I have verified that ProcessPoolExecutor does not have this Ctrl+C problem. If you need multiprocessing, I recommend using it.

References

Originally published on SegmentFault.