在 python 的 concurrent.futures 中查找 BrokenProcessPool 的原因

Posted

技术标签:

【中文标题】在 python 的 concurrent.futures 中查找 BrokenProcessPool 的原因【英文标题】:Finding the cause of a BrokenProcessPool in python's concurrent.futures 【发布时间】:2017-05-18 04:00:58 【问题描述】:

简而言之

当我使用concurrent.futures 并行化我的代码时,我得到一个BrokenProcessPool 异常。不再显示错误。我想找到错误的原因并询问如何做到这一点的想法。

全部问题

我正在使用concurrent.futures 来并行化一些代码。

with ProcessPoolExecutor() as pool:
    mapObj = pool.map(myMethod, args)

我最终得到(并且只有)以下异常:

concurrent.futures.process.BrokenProcessPool: A child process terminated abruptly, the process pool is not usable anymore

不幸的是,该程序很复杂,并且仅在程序运行 30 分钟后才会出现错误。因此,我无法提供一个很好的最小示例。

为了找到问题的原因,我将我并行运行的方法与 try-except-block 包装起来:

def myMethod(*args):
    try:
        ...
    except Exception as e:
        print(e)

问题仍然存在,并且从未输入过 except 块。我得出结论,异常不是来自我的代码。

我的下一步是编写一个自定义 ProcessPoolExecutor 类,它是原始 ProcessPoolExecutor 的子类,并允许我用自定义方法替换一些方法。我复制并粘贴了方法_process_worker的原始代码并添加了一些打印语句。

def _process_worker(call_queue, result_queue):
    """Evaluates calls from call_queue and places the results in result_queue.
        ...
    """
    while True:
        call_item = call_queue.get(block=True)
        if call_item is None:
            # Wake up queue management thread
            result_queue.put(os.getpid())
            return
        try:
            r = call_item.fn(*call_item.args, **call_item.kwargs)
        except BaseException as e:
                print("??? Exception ???")                 # newly added
                print(e)                                   # newly added
            exc = _ExceptionWithTraceback(e, e.__traceback__)
            result_queue.put(_ResultItem(call_item.work_id, exception=exc))
        else:
            result_queue.put(_ResultItem(call_item.work_id,
                                         result=r))

同样,except 块永远不会进入。这是意料之中的,因为我已经确保我的代码不会引发异常(如果一切正常,则应该将异常传递给主进程)。

现在我不知道如何找到错误。此处引发异常:

def submit(self, fn, *args, **kwargs):
    with self._shutdown_lock:
        if self._broken:
            raise BrokenProcessPool('A child process terminated '
                'abruptly, the process pool is not usable anymore')
        if self._shutdown_thread:
            raise RuntimeError('cannot schedule new futures after shutdown')

        f = _base.Future()
        w = _WorkItem(f, fn, args, kwargs)

        self._pending_work_items[self._queue_count] = w
        self._work_ids.put(self._queue_count)
        self._queue_count += 1
        # Wake up queue management thread
        self._result_queue.put(None)

        self._start_queue_management_thread()
        return f

进程池设置在这里被打破:

def _queue_management_worker(executor_reference,
                             processes,
                             pending_work_items,
                             work_ids_queue,
                             call_queue,
                             result_queue):
    """Manages the communication between this process and the worker processes.
        ...
    """
    executor = None

    def shutting_down():
        return _shutdown or executor is None or executor._shutdown_thread

    def shutdown_worker():
        ...

    reader = result_queue._reader

    while True:
        _add_call_item_to_queue(pending_work_items,
                                work_ids_queue,
                                call_queue)

        sentinels = [p.sentinel for p in processes.values()]
        assert sentinels
        ready = wait([reader] + sentinels)
        if reader in ready:
            result_item = reader.recv()
        else:                               #THIS BLOCK IS ENTERED WHEN THE ERROR OCCURS
            # Mark the process pool broken so that submits fail right now.
            executor = executor_reference()
            if executor is not None:
                executor._broken = True
                executor._shutdown_thread = True
                executor = None
            # All futures in flight must be marked failed
            for work_id, work_item in pending_work_items.items():
                work_item.future.set_exception(
                    BrokenProcessPool(
                        "A process in the process pool was "
                        "terminated abruptly while the future was "
                        "running or pending."
                    ))
                # Delete references to object. See issue16284
                del work_item
            pending_work_items.clear()
            # Terminate remaining workers forcibly: the queues or their
            # locks may be in a dirty state and block forever.
            for p in processes.values():
                p.terminate()
            shutdown_worker()
            return
        ...

进程终止是(或似乎是)一个事实,但我不知道为什么。到目前为止,我的想法是否正确? 有哪些可能的原因导致进程在没有消息的情况下终止? (这甚至可能吗?)我可以在哪里应用进一步的诊断?为了更接近解决方案,我应该问自己哪些问题?

我在 64 位 Linux 上使用 python 3.5。

【问题讨论】:

我收到了这个错误,这篇文章解决了我的问题。 ***.com/questions/15900366/… 我得到了同样的错误,多进程退出代码是-11。虽然相同的功能在多线程中工作正常。 【参考方案1】:

我认为我能够做到尽可能:

我在更改后的ProcessPoolExecutor 模块中更改了_queue_management_worker 方法,以便打印失败进程的退出代码:

def _queue_management_worker(executor_reference,
                             processes,
                             pending_work_items,
                             work_ids_queue,
                             call_queue,
                             result_queue):
    """Manages the communication between this process and the worker processes.
        ...
    """
    executor = None

    def shutting_down():
        return _shutdown or executor is None or executor._shutdown_thread

    def shutdown_worker():
        ...

    reader = result_queue._reader

    while True:
        _add_call_item_to_queue(pending_work_items,
                                work_ids_queue,
                                call_queue)

        sentinels = [p.sentinel for p in processes.values()]
        assert sentinels
        ready = wait([reader] + sentinels)
        if reader in ready:
            result_item = reader.recv()
        else:                               

            # BLOCK INSERTED FOR DIAGNOSIS ONLY ---------
            vals = list(processes.values())
            for s in ready:
                j = sentinels.index(s)
                print("is_alive()", vals[j].is_alive())
                print("exitcode", vals[j].exitcode)
            # -------------------------------------------


            # Mark the process pool broken so that submits fail right now.
            executor = executor_reference()
            if executor is not None:
                executor._broken = True
                executor._shutdown_thread = True
                executor = None
            # All futures in flight must be marked failed
            for work_id, work_item in pending_work_items.items():
                work_item.future.set_exception(
                    BrokenProcessPool(
                        "A process in the process pool was "
                        "terminated abruptly while the future was "
                        "running or pending."
                    ))
                # Delete references to object. See issue16284
                del work_item
            pending_work_items.clear()
            # Terminate remaining workers forcibly: the queues or their
            # locks may be in a dirty state and block forever.
            for p in processes.values():
                p.terminate()
            shutdown_worker()
            return
        ...

后来查了一下退出码的含义:

from multiprocessing.process import _exitcode_to_name
print(_exitcode_to_name[my_exit_code])

其中my_exit_code 是在我插入_queue_management_worker 的块中打印的退出代码。在我的情况下,代码是 -11,这意味着我遇到了分段错误。找出这个问题的原因将是一项艰巨的任务,但超出了这个问题的范围。

【讨论】:

【参考方案2】:

如果您使用的是 macOS,则存在一个已知问题,即某些 macOS 版本如何使用分叉,在某些情况下 Python 认为分叉不安全。对我有用的解决方法是使用 no_proxy 环境变量。

编辑 ~/.bash_profile 并包含以下内容(最好在此处指定域或子网列表,而不是 *)

no_proxy='*'

刷新当前上下文

source ~/.bash_profile

我发现并解决了该问题的本地版本是:Python 3.6.0 on macOS 10.14.1 和 10.13.x

来源: Issue 30388 Issue 27126

【讨论】:

MacOS 10.14.6 (18G87) 和 Python 3.7.2 存在同样的问题

以上是关于在 python 的 concurrent.futures 中查找 BrokenProcessPool 的原因的主要内容,如果未能解决你的问题,请参考以下文章

python并发编程之进程池,线程池

Java 中 Future 的 get 方法超时会怎样?

在树莓派上的python3

在CentOS 8上安装Python

如何在命令行里运行python脚本

python编程应用在哪些行业