Python 中 Ctrl+C 不能終止 Multiprocessing Pool 的解決方案

lmxdawn 發(fā)布于2019-07-24 18:18 / 3485人閱讀

摘要：解決方法有兩種。代碼然而這段代碼只有在運行在處的時候才能用中斷，即前你按有效，一旦后則完全無效建議先確認是否真的需要用到多進程，如果是多的程序建議用多線程或協(xié)程，計算特別多則用多進程。

本文理論上對multiprocessing.dummy的Pool同樣有效。

python2.x中multiprocessing提供的基于函數(shù)進程池，join后陷入內(nèi)核態(tài)，按下ctrl+c不能停止所有的進程并退出。即必須ctrl+z后找到殘留的子進程，把它們干掉。先看一段ctrl+c無效的代碼：

#!/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()

這段代碼運行后，按^c一個進程也殺不掉，最后會殘留包括主進程在內(nèi)共5個進程(1+4)，kill掉主進程能讓其全部退出。很明顯，使用進程池時KeyboardInterrupt不能被進程捕捉。解決方法有兩種。

方案一

下面這段是python源碼里multiprocessing下的pool.py中的一段，ApplyResult就是Pool用來保存函數(shù)運行結(jié)果的類

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

而下面這段代碼也是^c無效的代碼

if __name__ == "__main__":
    import threading

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

很明顯，threading.Condition(threading.Lock())對象無法接收KeyboardInterrupt，但稍微修改一下，給cond.wait()一個timeout參數(shù)即可，這個timeout可以在map_async后用get傳遞，把

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

改為

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

就能成功接收^c了，get里面填1填99999還是0xffff都行

方案二

另一種方法當(dāng)然就是自己寫進程池了，需要使用隊列，貼一段代碼感受下

#!/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()

方案三

使用一個全局變量eflag作標(biāo)識，讓SIG_INT信號綁定一個處理函數(shù)，在其中對eflag的值更改，線程的函數(shù)中以eflag的值判定作為while的條件，把語句寫在循環(huán)里，老實說這個方案雖然可以用，但是簡直太差勁。線程肯定是可行的，進程應(yīng)該還需要多帶帶共享變量，~~非常不推薦的方式~~

常見的錯誤方案

這個必須要提一下，我發(fā)現(xiàn)segmentfault上都有人被誤導(dǎo)了

理論上，在Pool初始化時傳遞一個initializer函數(shù)，讓子進程忽略SIGINT信號，也就是^c，然后Pool進行terminate處理。代碼

#!/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()

然而這段代碼只有在運行在time.sleep(5)處的時候才能用ctrl+c中斷，即前5s你按^c有效，一旦pool.join()后則完全無效！

建議

先確認是否真的需要用到多進程，如果是IO多的程序建議用多線程或協(xié)程，計算特別多則用多進程。如果非要用多進程，可以利用Python3的concurrent.futures包(python2.x也能裝)，編寫更加簡單易用的多線程/多進程代碼，其使用和Java的concurrent框架有些相似.
經(jīng)過親自驗證，ProcessPoolExecutor是沒有^c的問題的，要用多進程建議使用它

參考

http://bryceboe.com/2010/08/26/python-multiprocessing-and-keyboardinterrupt/#georges

http://stackoverflow.com/questions/1408356/keyboard-interrupts-with-pythons-multiprocessing-pool#comment12678760_6191991