python的多處理池中的鍵盤中斷

Question

如何處理python的多處理池中的KeyboardInterrupt事件？下面是一個簡單的例子：

from multiprocessing import Pool 
from time import sleep 
from sys import exit 

def slowly_square(i): 
    sleep(1) 
    return i*i 

def go(): 
    pool = Pool(8) 
    try: 
     results = pool.map(slowly_square, range(40)) 
    except KeyboardInterrupt: 
     # **** THIS PART NEVER EXECUTES. **** 
     pool.terminate() 
     print "You cancelled the program!" 
     sys.exit(1) 
    print "\nFinally, here are the results: ", results 

if __name__ == "__main__": 
    go() 

當運行上面的代碼，當我按下^C的KeyboardInterrupt被提出，但這一過程只是掛起，在這一點上，我不得不從外部殺死它。

我希望能夠隨時按^C，並使所有進程正常退出。

Answer 1

這是一個Python錯誤。在等待threading.Condition.wait（）中的條件時，KeyboardInterrupt永遠不會被髮送。攝製：

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

的KeyboardInterrupt異常之前，不會等待交付（）返回時，它永遠不會返回，因此中斷不會發生。 KeyboardInterrupt幾乎肯定會中斷一個等待狀態。

請注意，如果指定超時，則不會發生這種情況; cond.wait（1）會立即收到中斷。所以，一個解決方法是指定一個超時。要做到這一點，與

results = pool.map_async(slowly_square, range(40)).get(9999999) 

或類似的替代

results = pool.map(slowly_square, range(40)) 

。

Answer 2

奇怪的是，它看起來像你還必須在兒童中處理KeyboardInterrupt。我本來期望這個工作，因爲寫的......請嘗試更改slowly_square到：

def slowly_square(i): 
    try: 
     sleep(1) 
     return i * i 
    except KeyboardInterrupt: 
     print 'You EVIL bastard!' 
     return 0 

如你預期這應該工作。

Answer 3

由於某些原因，只有繼承自基類Exception類的例外才能正常處理。作爲一種變通方法，你可能會重新提高你KeyboardInterrupt爲Exception實例：

from multiprocessing import Pool 
import time 

class KeyboardInterruptError(Exception): pass 

def f(x): 
    try: 
     time.sleep(x) 
     return x 
    except KeyboardInterrupt: 
     raise KeyboardInterruptError() 

def main(): 
    p = Pool(processes=4) 
    try: 
     print 'starting the pool map' 
     print p.map(f, range(10)) 
     p.close() 
     print 'pool map complete' 
    except KeyboardInterrupt: 
     print 'got ^C while pool mapping, terminating the pool' 
     p.terminate() 
     print 'pool is terminated' 
    except Exception, e: 
     print 'got exception: %r, terminating the pool' % (e,) 
     p.terminate() 
     print 'pool is terminated' 
    finally: 
     print 'joining pool processes' 
     p.join() 
     print 'join complete' 
    print 'the end' 

if __name__ == '__main__': 
    main() 

通常你會得到以下輸出：

staring the pool map 
[0, 1, 2, 3, 4, 5, 6, 7, 8, 9] 
pool map complete 
joining pool processes 
join complete 
the end 

所以，如果你打^C，您將獲得：

staring the pool map 
got ^C while pool mapping, terminating the pool 
pool is terminated 
joining pool processes 
join complete 
the end 

Answer 4

我發現，目前最好的解決方案是不使用multiprocessing.pool功能，而是使用自己的池功能。我提供了一個演示apply_async錯誤的示例以及一個演示如何完全避免使用池功能的示例。

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

Answer 5

從我最近發現，最好的解決辦法是成立工作進程完全忽略SIGINT，並限制所有的清理代碼父進程。這解決了空閒和繁忙工作進程的問題，並且不需要在子進程中使用錯誤處理代碼。

import signal 

... 

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

... 

def main() 
    pool = multiprocessing.Pool(size, init_worker) 

    ... 

    except KeyboardInterrupt: 
     pool.terminate() 
     pool.join() 

說明和完整的示例代碼可在分別http://noswap.com/blog/python-multiprocessing-keyboardinterrupt/和http://github.com/jreese/multiprocessing-keyboardinterrupt找到。

Answer 6

通常這種簡單的結構，適用於按Ctrl - ç在游泳池：

def signal_handle(_signal, frame): 
    print "Stopping the Jobs." 

signal.signal(signal.SIGINT, signal_handle) 

正如幾個類似的帖子指出：

Capture keyboardinterrupt in Python without try-except

Answer 7

似乎有兩個問題在多處理令人討厭時會造成異常。第一個（由Glenn注意到）是你需要使用map_async而不是map以獲得即時響應（即，不完成整個列表的處理）。第二個（由Andrey指出）是多處理不會捕獲不會從Exception（例如，SystemExit）繼承的異常。所以這裏是我的解決方案，處理這兩個：

import sys 
import functools 
import traceback 
import multiprocessing 

def _poolFunctionWrapper(function, arg): 
    """Run function under the pool 

    Wrapper around function to catch exceptions that don't inherit from 
    Exception (which aren't caught by multiprocessing, so that you end 
    up hitting the timeout). 
    """ 
    try: 
     return function(arg) 
    except: 
     cls, exc, tb = sys.exc_info() 
     if issubclass(cls, Exception): 
      raise # No worries 
     # Need to wrap the exception with something multiprocessing will recognise 
     import traceback 
     print "Unhandled exception %s (%s):\n%s" % (cls.__name__, exc, traceback.format_exc()) 
     raise Exception("Unhandled exception: %s (%s)" % (cls.__name__, exc)) 

def _runPool(pool, timeout, function, iterable): 
    """Run the pool 

    Wrapper around pool.map_async, to handle timeout. This is required so as to 
    trigger an immediate interrupt on the KeyboardInterrupt (Ctrl-C); see 
    http://stackoverflow.com/questions/1408356/keyboard-interrupts-with-pythons-multiprocessing-pool 

    Further wraps the function in _poolFunctionWrapper to catch exceptions 
    that don't inherit from Exception. 
    """ 
    return pool.map_async(functools.partial(_poolFunctionWrapper, function), iterable).get(timeout) 

def myMap(function, iterable, numProcesses=1, timeout=9999): 
    """Run the function on the iterable, optionally with multiprocessing""" 
    if numProcesses > 1: 
     pool = multiprocessing.Pool(processes=numProcesses, maxtasksperchild=1) 
     mapFunc = functools.partial(_runPool, pool, timeout) 
    else: 
     pool = None 
     mapFunc = map 
    results = mapFunc(function, iterable) 
    if pool is not None: 
     pool.close() 
     pool.join() 
    return results 

Answer 8

我是Python的新手。我到處尋找答案，並偶然發現這個以及其他一些博客和YouTube視頻。我試圖複製粘貼上面的作者的代碼，並在Windows 7 64位我的Python 2.7.13重現它。這接近我想要達到的目標。

我讓我的子進程忽略ControlC並使父進程終止。看起來像繞過子進程確實爲我避免了這個問題。

#!/usr/bin/python 

from multiprocessing import Pool 
from time import sleep 
from sys import exit 


def slowly_square(i): 
    try: 
     print "<slowly_square> Sleeping and later running a square calculation..." 
     sleep(1) 
     return i * i 
    except KeyboardInterrupt: 
     print "<child processor> Don't care if you say CtrlC" 
     pass 


def go(): 
    pool = Pool(8) 

    try: 
     results = pool.map(slowly_square, range(40)) 
    except KeyboardInterrupt: 
     # *** THIS PART NEVER EXECUTES. *** :(
     pool.terminate() 
     pool.close() 
     print "You cancelled the program!" 
     exit(1) 
    print "Finally, here are the results", results 


if __name__ == '__main__': 
    go() 

Answer 9

投票答案沒有解決核心問題，但類似的副作用。

多處理庫的作者Jesse Noller解釋瞭如何在舊的blog post中使用multiprocessing.Pool時正確處理CTRL + C.

import signal 
from multiprocessing import Pool 


def initializer(): 
    """Ignore CTRL+C in the worker process.""" 
    signal.signal(signal.SIGINT, signal.SIG_IGN) 


pool = Pool(initializer=initializer) 

try: 
    pool.map(perform_download, dowloads) 
except KeyboardInterrupt: 
    pool.terminate() 
    pool.join()