使用 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"

在 wait() 返回之前不会传递 KeyboardInterrupt 异常，并且它永远不会返回，因此中断永远不会发生。 KeyboardInterrupt 几乎肯定应该中断条件等待。

请注意，如果指定了超时，则不会发生这种情况； cond.wait(1) 将立即接收中断。因此，解决方法是指定超时。为此，请替换

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

为

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

或类似的。

Answer 2

根据我最近的发现，最好的解决方案是将工作进程设置为完全忽略 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 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

投票的答案并没有解决核心问题，而是解决了类似的副作用。

Jesse Noller，multiprocessing 库的作者，解释了在旧的 multiprocessing.Pool 时如何正确处理 CTRL+C /01/08/multiprocessingpool-and-keyboardinterrupt" rel="noreferrer">博客文章。

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()

Answer 5

其中许多答案都是旧的，并且/或者如果您正在执行诸如 Pool.map更高版本的 Python（我正在运行 3.8.5） >，它会阻塞，直到所有提交的任务完成。以下是我的解决方案。

1. 在主进程中调用 signal.signal(signal.SIGINT, signal.SIG_IGN) 以完全忽略 Ctrl-C。
2. 处理池将使用池初始值设定项进行初始化，该池初始值设定项将因此初始化每个处理器：全局变量 ctrl_c_entered 将设置为 False 并调用 signal.signal( signal.SIGINT, signal.SIG_IGN) 将被发出以最初忽略 Ctrl-C。该调用的返回值将被保存；这是原始的默认处理程序，重新建立后允许处理 KyboardInterrupt 异常。
3. 装饰器 handle_ctrl_c 可用于装饰多处理函数和方法，这些函数和方法应在输入 Ctrl-C 时立即退出。该装饰器将测试是否设置了全局 ctrl_c_entered 标志，如果是，则甚至不运行该函数/方法，而是返回一个 KeyboardInterrupt 异常实例。否则，将建立 KeyboardInterrupt 的 try/catch 处理程序，并调用修饰的函数/方法。如果输入 Ctrl-C，则全局 ctrl_c_entered 将设置为 True 并返回 KeyboardInterrupt 异常实例。无论如何，在返回之前装饰器都会重新建立 SIG_IGN 处理程序。

本质上，所有提交的任务都将被允许启动，但一旦输入 Ctrl-C，将立即终止并返回 KeyBoardInterrupt 异常。主进程可以测试返回值是否存在这样的返回值，以检测是否输入了Ctrl-C。

from multiprocessing import Pool
import signal
from time import sleep
from functools import wraps

def handle_ctrl_c(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        global ctrl_c_entered
        if not ctrl_c_entered:
            signal.signal(signal.SIGINT, default_sigint_handler) # the default
            try:
                return func(*args, **kwargs)
            except KeyboardInterrupt:
                ctrl_c_entered = True
                return KeyboardInterrupt()
            finally:
                signal.signal(signal.SIGINT, pool_ctrl_c_handler)
        else:
            return KeyboardInterrupt()
    return wrapper

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

def pool_ctrl_c_handler(*args, **kwargs):
    global ctrl_c_entered
    ctrl_c_entered = True

def init_pool():
    # set global variable for each process in the pool:
    global ctrl_c_entered
    global default_sigint_handler
    ctrl_c_entered = False
    default_sigint_handler = signal.signal(signal.SIGINT, pool_ctrl_c_handler)

def main():
    signal.signal(signal.SIGINT, signal.SIG_IGN)
    pool = Pool(initializer=init_pool)
    results = pool.map(slowly_square, range(10))
    if any(map(lambda x: isinstance(x, KeyboardInterrupt), results)):
        print('Ctrl-C was entered.')
    print(results)
    pool.close()
    pool.join()

if __name__ == '__main__':
    main()

印刷：

Ctrl-C was entered.
[0, 1, 4, 9, 16, 25, 36, 49, KeyboardInterrupt(), KeyboardInterrupt()]

Answer 6

通常，这个简单的结构适用于 Pool 上的 Ctrl-C ：

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

signal.signal(signal.SIGINT, signal_handle)

正如一些类似帖子中所述：

在 Python 中捕获键盘中断而不使用 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:
        pool.terminate()
        pool.close()
        print "You cancelled the program!"
        exit(1)
    print "Finally, here are the results", results


if __name__ == '__main__':
    go()

从 pool.terminate() 开始的部分似乎永远不会执行。

Answer 9

我发现，目前最好的解决方案是不使用 multiprocessing.pool 功能，而是推出自己的池功能。我提供了一个示例来演示 apply_async 的错误，以及一个示例来展示如何完全避免使用池功能。

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

Answer 10

您可以尝试使用 Pool 对象的 apply_async 方法，如下所示：

import multiprocessing
import time
from datetime import datetime


def test_func(x):
    time.sleep(2)
    return x**2


def apply_multiprocessing(input_list, input_function):
    pool_size = 5
    pool = multiprocessing.Pool(processes=pool_size, maxtasksperchild=10)

    try:
        jobs = {}
        for value in input_list:
            jobs[value] = pool.apply_async(input_function, [value])

        results = {}
        for value, result in jobs.items():
            try:
                results[value] = result.get()
            except KeyboardInterrupt:
                print "Interrupted by user"
                pool.terminate()
                break
            except Exception as e:
                results[value] = e
        return results
    except Exception:
        raise
    finally:
        pool.close()
        pool.join()


if __name__ == "__main__":
    iterations = range(100)
    t0 = datetime.now()
    results1 = apply_multiprocessing(iterations, test_func)
    t1 = datetime.now()
    print results1
    print "Multi: {}".format(t1 - t0)

    t2 = datetime.now()
    results2 = {i: test_func(i) for i in iterations}
    t3 = datetime.now()
    print results2
    print "Non-multi: {}".format(t3 - t2)

输出：

100
Multiprocessing run time: 0:00:41.131000
100
Non-multiprocessing run time: 0:03:20.688000

此方法的优点是中断前处理的结果将在结果字典中返回：

>>> apply_multiprocessing(range(100), test_func)
Interrupted by user
{0: 0, 1: 1, 2: 4, 3: 9, 4: 16, 5: 25}

Answer 11

奇怪的是，看起来您还必须处理子级中的 KeyboardInterrupt 。我本希望这能按书面方式工作...尝试将 slowly_square 更改为：

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

这应该按您的预期工作。