python - `multiprocessing` 与 `concurrent.futures` 中的最大 worker 数

Question

在 Python 3.8 中，concurrent.futures.ProcessPoolExecutor已更新以将能够在 Windows 上使用的工作程序(进程)的最大数量限制为 61。有关原因，请参阅 this和 this ，但据我了解:

在 Windows 上，multiprocessing调用 Windows API 函数 WaitForMultipleObjects ，用于等待进程完成。它最多可以等待 63 个对象，减去结果队列读取器和线程唤醒读取器，因此有 61 个限制。 (即 Windows 使用每个进程的线程来跟踪进程)。

(另见 this SO issue) multiprocessing ，但是，仍然使用 os.cpu_count() .它抛出一个 Value Error起初，但随后继续使用我 100% 的 CPU 内核。例如，

Exception in thread Thread-N:
Traceback (most recent call last):
  File "C:\Users\username\AppData\Local\Programs\Python\Python38\lib\threading.py", line 932, in _bootstrap_inner
    self.run()
  File "C:\Users\username\AppData\Local\Programs\Python\Python38\lib\threading.py", line 870, in run
    self._target(*self._args, **self._kwargs)
  File "C:\Users\username\AppData\Local\Programs\Python\Python38\lib\multiprocessing\pool.py", line 519, in _handle_workers       
    cls._wait_for_updates(current_sentinels, change_notifier)
  File "C:\Users\username\AppData\Local\Programs\Python\Python38\lib\multiprocessing\pool.py", line 499, in _wait_for_updates     
    wait(sentinels, timeout=timeout)
  File "C:\Users\username\AppData\Local\Programs\Python\Python38\lib\multiprocessing\connection.py", line 879, in wait
    ready_handles = _exhaustive_wait(waithandle_to_obj.keys(), timeout)
  File "C:\Users\username\AppData\Local\Programs\Python\Python38\lib\multiprocessing\connection.py", line 811, in _exhaustive_wait
    res = _winapi.WaitForMultipleObjects(L, False, timeout)
ValueError: need at most 63 handles, got a sequence of length 98

我的机器有 96 个内核。这个“错误”真的是错误吗？如果没有，我应该使用 multiprocessing模块而不是 concurrent.futures模块，它将我的 CPU 使用率限制为 61 个内核？
编辑:我怀疑这是一个错误，因为我假设 multiprocess将继续等待引发错误的进程完成。如果我不限制内核数量，这似乎会发生(程序在 CPU 使用率下降后挂起)。但是，我不确定是否真的如此。

Answer 1

你的问题很好。查看代码，这似乎是一个不可恢复的错误。但在我看来，ThreadPoolExecutor 中会有代码，这让我无法理解。在 Windows 下将池大小限制为 61，并且不对 multiprocessing.Pool 强制执行此操作类(class)。无论如何，使用以下程序进行检查应该很容易。如果不打印 完毕! 并挂起，我会说肯定有问题，如果您使用 multiprocessing.Pool，您应该明确限制池大小:

import multiprocessing

def worker(x):
    return x ** 2

def main():
    pool = multiprocessing.Pool(96)
    results = pool.map(worker, range(96))
    assert len(results) == 96
    pool.close()
    pool.join()
    print('Done!')

if __name__ == '__main__':
    main()

但是你的程序挂起的事实是相当确定的，上面的程序会挂起，我怀疑你甚至不会到达assert。陈述。无论哪种方式，使用大于 61 的池大小都不可靠。