Python 多处理提前终止

Question

当我的脚本运行时，可能会在某个时候发生错误。在这种情况下，所有进程都应该正确终止，应该返回一条错误消息，并且脚本应该退出。

我现在的代码似乎还不能满足这些要求。当发生错误时，它被发送到 report_error()，脚本最终卡在终端中，Activity Monitor 显示许多 Python 进程仍在运行。

环境

• Mac OS X 10.8.5
• python 3.3.3

从脚本中的任意点终止所有进程的正确方法是什么？

#!/usr/bin/env python3
# -*- coding: utf-8 -*-


import sys
from multiprocessing import Pool


# Global variables.

input_files = [
    'test_data_0.csv',
    'test_data_1.csv'
]


def report_error(error):

    # Reports errors then exits script.
    print("Error: {0}".format(error), file=sys.stderr)
    sys.exit(1)

    # What I really want is to report the error, properly terminate all processes,
    # and then exit the script.


def read_file(file):

    try:
        # Read file into list.
    except Exception as error:
        report_error(error)


def check_file(file):

    # Do some error checking on file.
    if error:
        report_error(error)


def job(file):

    # Executed on each item in input_files.

    check_file(file)
    read_file(file)


def main():

    # Sets up a process pool. Defaults to number of cores.
    # Each input gets passed to job and processed in a separate process.
    p = Pool()
    p.map(job, input_files)

    # Closing and joining a pool is important to ensure all resources are freed properly.
    p.close()
    p.join()


if __name__ == '__main__':
    main()

Answer 1

首先，使用 sys.exit() 杀死子工作进程实际上会破坏池，并使 map 命令永远挂起。当前，multiprocessing 无法在工作进程处理作业时从工作进程的崩溃中正确恢复(有一个错误报告和一个解决此问题的补丁 here，不管它的值(value))。

您可以通过多种方式做您实际上想做的事。由于您似乎并不关心辅助函数返回的值，最简单的方法是使用 imap_unordered代替 map，在出现故障时从 worker 引发异常，然后简单地迭代 imap_unordered 返回的迭代器:

def report_error(error):

    # Reports errors then exits script.
    print("Error: {0}".format(error), file=sys.stderr)
    raise error # Raise the exception

...

def main():
    p = Pool()
    try:
        list(p.imap_unordered(job, input_files))
    except Exception:
        print("a worker failed, aborting...")
        p.close()
        p.terminate()
    else:
        p.close()
        p.join()

if __name__ == '__main__':
    main()

使用 imap_unordered，结果将在子级发送后立即返回给父级。因此，如果子进程将异常发送回父进程，它将立即在父进程中重新引发。我们捕获该异常，打印一条消息，然后终止该池。