Python concurrent.futures:处理子进程中的异常

Question

我有一个非常简单的vanilla实现concurrent.futures.ProcessPoolExecutor- 类似于(使用Python 3.6):

files = get_files()
processor = get_processor_instance()
with concurrent.futures.ProcessPoolExecutor() as executor:
    list(executor.map(processor.process, files))

虽然它processor是许多可用处理器类中的任何一个的实例,但它们都共享该process方法,其大致如下所示:

def process(self, file):
    log.debug(f"Processing source file {file.name}.")
    with DBConnection(self.db_url) as session:
        file = session.merge(file)
        session.refresh(file)
        self._set_file(file)
        timer = perf_counter()
        try:
            self.records = self._get_records()
            self._save_output()
        except Exception as ex:
            log.warning(f"Failed to process source file {file.ORIGINAL_NAME}: {ex}")
            self.error_time = time.time()
            self.records = None
        else:
            process_duration = perf_counter() - timer
            log.info(f'File {file.name} processed in {process_duration:.6f} seconds.')
            file.process_duration = process_duration
        session.commit()

每个类的实现_get_records和_save_output方法各不相同,但我的问题是处理错误.我故意测试它,以便这两个方法中的一个耗尽内存,但我希望except上面的块能够捕获它并移动下一个文件 - 这正是我在单个代码中运行代码时会发生的情况处理.

如果我ProcessPoolExecutor如上所述使用它,它会引发BrokenProcessPool异常并杀死所有执行:

Traceback (most recent call last):
  File "/vagrant/myapp/myapp.py", line 94, in _process
    list(executor.map(processor.process, files))
  File "/home/ubuntu/.pyenv/versions/3.6.3/lib/python3.6/concurrent/futures/process.py", line 366, in _chain_from_iterable_of_lists
    for element in iterable:
  File "/home/ubuntu/.pyenv/versions/3.6.3/lib/python3.6/concurrent/futures/_base.py", line 586, in result_iterator
    yield fs.pop().result()
  File "/home/ubuntu/.pyenv/versions/3.6.3/lib/python3.6/concurrent/futures/_base.py", line 432, in result
    return self.__get_result()
  File "/home/ubuntu/.pyenv/versions/3.6.3/lib/python3.6/concurrent/futures/_base.py", line 384, in __get_result
    raise self._exception
concurrent.futures.process.BrokenProcessPool: A process in the process pool was terminated abruptly while the future was running or pending.

我当然可以BrokenProcessPool在调用代码中捕获,但我更喜欢在内部处理错误并继续下一个文件.

我也尝试过使用标准multiprocessing.Pool对象,如下所示:

with multiprocessing.Pool() as pool:
    pool.map(processor.process, files)

在这种情况下,行为甚至更奇怪:在开始处理前两个文件(引发内存不足错误)之后,它继续处理后面的文件,这些文件较小,因此完全处理.但是,该except块显然永远不会被触发(没有日志消息,没有error_time),并且应用程序只是挂起,既不完成也不做任何事情,直到手动杀死.

我希望该try..except块能够使每个进程自包含,处理自己的错误而不影响主应用程序.任何想法如何实现？

Answer 1

所以,经过大量的调试(以及@ RomanPerekhrest建议检查executor对象的应有的信誉),我已经找到了原因.如问题中所述,测试数据由许多文件组成,其中两个文件非常大(每个文件超过100万行).这两个导致我的测试机器(2GB VM)窒息,但以不同的方式 - 而第一个,更大,导致定期内存错误,这将由处理except,第二个只是导致一个sigkill.在没有太多探索的情况下,我怀疑较大的文件根本无法在读取内存(在_get_records方法中完成),而较小的文件可以,但随后对它进行操作(完成_save_output)会导致溢出并终止进程.

我的解决方案是简单地捕获BrokenProcessPool异常并告知用户该问题; 我还添加了一个选项,它在一个进程中运行处理任务,在这种情况下,任何太大的文件都只是标记为有错误:

files = get_files()
processor = get_processor_instance()
results = []
if args.nonconcurrent:
    results = list(map(processor.process, files))
else:
    with concurrent.futures.ProcessPoolExecutor() as executor:
        try:
            results = list(executor.map(processor.process, files))
        except concurrent.futures.process.BrokenProcessPool as ex:
            raise MyCustomProcessingError(
                f"{ex} This might be caused by limited system resources. "
                "Try increasing system memory or disable concurrent processing "
                "using the --nonconcurrent option."
            )