从 multiprocessing.Pool 调用的函数中退出的正确方法？答案

【问题标题】：Right way to exit from function called by multiprocessing.Pool?从 multiprocessing.Pool 调用的函数中退出的正确方法？
【发布时间】：2018-10-09 19:11:58
【问题描述】：

如何退出名为 mymultiprocessing.Pool 的函数

这是我正在使用的代码示例，当我设置条件退出函数 worker 时，当我在终端中将其用作脚本时，它会停止并且不会退出。

def worker(n):
    if n == 4:
        exit("wrong number")  # tried to use sys.exit(1) did not work
    return n*2

def caller(mylist, n=1):
    n_cores = n if n > 1 else multiprocessing.cpu_count()
    print(n_cores)
    pool = multiprocessing.Pool(processes=n_cores)
    result = pool.map(worker, mylist)
    pool.close()
    pool.join()
    return result

l = [2, 3, 60, 4]
myresult = caller(l, 4)

【问题讨论】：

exit 究竟应该做什么？
简单地退出脚本，错误值为 4 ex: exit("Can not run the script value = 4")
只是来自函数（或return somevalue）的return，而不是调用exit()。您还应该缩进 if __name__ == '__main__': 中的最后几行
你不能从工作进程中退出主脚本。
没有一个适合你的答案吗？

标签： python multiprocessing pool

【解决方案1】：

正如我所说，我认为您不能从工作进程中退出运行主脚本的进程。

你还没有准确解释为什么你想这样做，所以这个答案是一个猜测，但可能会提出一个自定义 Exception 并在显式 except 中处理它，如下所示将是解决此限制的可接受方法。

import multiprocessing
import sys

class WorkerStopException(Exception):
    pass

def worker(n):
    if n == 4:
        raise WorkerStopException()
    return n*2

def caller(mylist, n=1):
    n_cores = n if n > 1 else multiprocessing.cpu_count()
    print(n_cores)
    pool = multiprocessing.Pool(processes=n_cores)
    try:
        result = pool.map(worker, mylist)
    except WorkerStopException:
        sys.exit("wrong number")
    pool.close()
    pool.join()
    return result

if __name__ == '__main__':
    l = [2, 3, 60, 4]
    myresult = caller(l, 4)

运行时显示的输出：

4
wrong number

（4 是我系统的 CPU 数量。）

【讨论】：

主要思想是：我正在处理一些文件并启动一个包含有关此文件的信息的对象：因此，如果文件缺少某些信息，我需要退出执行并通知用户该文件缺少此信息并且我们退出操作更正文件以继续。用户可以输入多个文件，这就是我使用 Pool 的原因，因此它们将被并行处理。希望对您有所帮助。
在这种情况下，我的回答中的方法似乎很有用（如果我理解你所说的正确的话）。
谢谢，我试过了，但结果是：TypeError: catching classes that do not inherit from BaseException is not allowed 我正在使用 python 3.5.4
我使用 Python 3.7.0（在 Windows 上）运行它，所以也许该限制已被删除。
FWIW 错误消息听起来好像您没有使用Exception 作为子类WorkerStopException 的基类，因为Exception 本身是BaseException 的子类，所以自定义子类也是一个（间接）。

【解决方案2】：

pool.map 的问题是，它只会在所有任务完成后引发子进程的异常。但是，一旦在任何进程中检测到错误值，您的 cmets 听起来就需要立即中止所有处理。这将是pool.apply_async 的工作。

pool.apply_async 提供error_callbacks，您可以使用它来让池终止。工人将被逐项提供而不是像pool.map 变体那样逐块提供，因此您有机会在每个已处理的参数上提前退出。

我基本上是在重复使用来自here 的答案：

from time import sleep
from multiprocessing import Pool

def f(x):
    sleep(x)
    print(f"f({x})")
    if x == 4:
        raise ValueError(f'wrong number: {x}')
    return x * 2

def on_error(e):
    if type(e) is ValueError:
        global terminated
        terminated = True
        pool.terminate()
        print(f"oops: {type(e).__name__}('{e}')")


def main():
    global pool
    global terminated

    terminated = False

    pool = Pool(4)
    results = [pool.apply_async(f, (x,), error_callback=on_error)
               for x in range(10)]
    pool.close()
    pool.join()

    if not terminated:
        for r in results:
            print(r.get())


if __name__ == '__main__':
    main()

输出：

f(0)
f(1)
f(2)
f(3)
f(4)
oops: ValueError('wrong number: 4')

Process finished with exit code 0

【讨论】：