windows - `ProcessPoolExecutor` 在 Ubuntu 上工作，但在 Windows 10 上运行带有 Python 3.5.3 的 Jupyter 5.0.0 笔记本时失败并显示 `BrokenProcessPool`

Question

我在 Windows 10 上使用 Python 3.5.3 运行 Jupyter 5.0.0 notebook。以下示例代码无法运行:

from concurrent.futures import as_completed, ProcessPoolExecutor
import time
import numpy as np

def do_work(idx1, idx2):
    time.sleep(0.2)
    return np.mean([idx1, idx2])

with ProcessPoolExecutor(max_workers=4) as executor:
    futures = set()
    for idx in range(32):
        future = winprocess.submit(
            executor, do_work, idx, idx * 2
        )
        futures.add(future)

    for future in as_completed(futures):
        print(future.result())

... 并抛出 BrokenProcessPool:进程池中的进程在未来运行或挂起时突然终止。

代码在 Ubuntu 14.04 上运行良好。

我知道 Windows 没有 os.fork，因此多处理的处理方式不同，并且在交互模式和 Jupyter 中并不总是很好。

在这种情况下，有哪些解决方法可以使 ProcessPoolExecutor 正常工作？

还有一些类似的问题，但它们与multiprocessing.Pool有关:

• multiprocessing.Pool in jupyter notebook works on linux but not windows

Answer 1

仔细检查表明，Jupyter notebook 可以运行使用 ProcessPoolExecutor 并行化的外部 python 模块。因此，一种解决方案是在模块中执行代码的可并行化部分，然后从 Jupyter notebook 调用它。

也就是说，这可以概括为一种实用程序。以下内容可以存储为模块，例如 winprocess.py 并由 jupyter 导入。

import inspect
import types


def execute_source(callback_imports, callback_name, callback_source, args):
    for callback_import in callback_imports:
        exec(callback_import, globals())
    exec('import time' + "\n" + callback_source)
    callback = locals()[callback_name]
    return callback(*args)


def submit(executor, callback, *args):
    callback_source = inspect.getsource(callback)
    callback_imports = list(imports(callback.__globals__))
    callback_name = callback.__name__
    future = executor.submit(
        execute_source,
        callback_imports, callback_name, callback_source, args
    )
    return future


def imports(callback_globals):
    for name, val in list(callback_globals.items()):
        if isinstance(val, types.ModuleType) and val.__name__ != 'builtins' and val.__name__ != __name__:
            import_line = 'import ' + val.__name__
            if val.__name__ != name:
                import_line += ' as ' + name
            yield import_line

以下是您将如何使用它:

from concurrent.futures import as_completed, ProcessPoolExecutor
import time
import numpy as np
import winprocess

def do_work(idx1, idx2):
    time.sleep(0.2)
    return np.mean([idx1, idx2])

with ProcessPoolExecutor(max_workers=4) as executor:
    futures = set()
    for idx in range(32):
        future = winprocess.submit(
            executor, do_work, idx, idx * 2
        )
        futures.add(future)

    for future in as_completed(futures):
        print(future.result())

请注意 executor 已更改为 winprocess 并且原始 executor 被传递给 submit 函数一个参数。

这里发生的是笔记本函数代码和导入被序列化并传递给模块执行。代码只有在安全地进入新进程后才会执行，因此不会因尝试基于 jupyter notebook 本身创建新进程而失败。

以维护别名的方式处理导入。如果您确保导入在函数本身内执行的函数所需的所有内容，则可以删除导入魔法。

此外，此解决方案仅在您将所有必要变量作为参数传递给函数时才有效。可以说该函数应该是静态的，但我认为这也是 ProcessPoolExecutor 的要求。最后，确保您不执行笔记本中其他地方定义的其他功能。仅导入外部模块，因此不会包含其他笔记本功能。