Python 多处理 - 池中的进程数量是否会因错误而减少？

Question

代码:

import multiprocessing
print(f'num cpus {multiprocessing.cpu_count():d}')
import sys; print(f'Python {sys.version} on {sys.platform}')

def _process(m):
    print(m) #; return m
    raise ValueError(m)

args_list = [[i] for i in range(1, 20)]

if __name__ == '__main__':
    with multiprocessing.Pool(2) as p:
        print([r for r in p.starmap(_process, args_list)])

打印:

num cpus 8
Python 3.7.1 (v3.7.1:260ec2c36a, Oct 20 2018, 03:13:28) 
[Clang 6.0 (clang-600.0.57)] on darwin
1
7
4
10
13
16
19
multiprocessing.pool.RemoteTraceback: 
"""
Traceback (most recent call last):
  File "/Library/Frameworks/Python.framework/Versions/3.7/lib/python3.7/multiprocessing/pool.py", line 121, in worker
    result = (True, func(*args, **kwds))
  File "/Library/Frameworks/Python.framework/Versions/3.7/lib/python3.7/multiprocessing/pool.py", line 47, in starmapstar
    return list(itertools.starmap(args[0], args[1]))
  File "/Users/ubik-mac13/Library/Preferences/PyCharm2018.3/scratches/multiprocess_error.py", line 8, in _process
    raise ValueError(m)
ValueError: 1
"""

The above exception was the direct cause of the following exception:

Traceback (most recent call last):
  File "/Users/ubik-mac13/Library/Preferences/PyCharm2018.3/scratches/multiprocess_error.py", line 18, in <module>
    print([r for r in p.starmap(_process, args_list)])
  File "/Library/Frameworks/Python.framework/Versions/3.7/lib/python3.7/multiprocessing/pool.py", line 298, in starmap
    return self._map_async(func, iterable, starmapstar, chunksize).get()
  File "/Library/Frameworks/Python.framework/Versions/3.7/lib/python3.7/multiprocessing/pool.py", line 683, in get
    raise self._value
ValueError: 1

Process finished with exit code 1

将池中的进程数增加到 3 或 4 会打印所有奇数(可能是乱序的):

1
3
5
9
11
7
13
15
17
19

从 5 及以上开始，它会打印 1-19 的所有范围。那么这里会发生什么呢？进程是否会在多次失败后崩溃？

这当然是一个玩具示例，但它来 self 遇到的一个真实代码问题 - 让多处理池稳定运行几天后，CPU 使用率下降，就好像某些进程被杀死一样(请注意，CPU 利用率在下降) 03/04 和 03/06，当时还有很多任务需要运行):

当代码终止时，它向我展示了一个(只有一个像这里一样，而进程还有更多)multiprocessing.pool.RemoteTraceback - 额外的问题是这个回溯是随机的吗？在此玩具示例中，通常为 ValueError: 1 但有时也会出现其他数字。多处理是否保留第一个崩溃进程的第一个回溯？

Answer 1

不，只是整个任务崩溃了，而不是进程本身。您在玩具示例中观察到的行为可以通过工作人员数量和可迭代长度的组合产生的 block 大小来解释。当您从here获取函数calc_chunksize_info时您可以看到生成的 block 大小的差异:

calc_chunksize_info(n_workers=2, len_iterable=20)
# Chunkinfo(n_workers=2, len_iterable=20, n_chunks=7, chunksize=3, last_chunk=2)

calc_chunksize_info(n_workers=5, len_iterable=20)
# Chunkinfo(n_workers=5, len_iterable=20, n_chunks=20, chunksize=1, last_chunk=1) 

如果 block 大小 > 1，则全部保持不变 "taskels" (1. Definitions: Taskel)一旦第一个任务引发异常，任务中的数据也会丢失。直接在目标函数中处理预期的异常，或者编写一个额外的错误处理包装器来防止这种情况发生。

When the code terminated it presented me with one (and one only as here, while the processes were many more) multiprocessing.pool.RemoteTraceback - bonus question is this traceback random? In this toy example, it is usually ValueError: 1 but sometimes also other numbers appear. Does multiprocessing keep the first traceback from the first process that crashes?

工作进程从共享队列中获取任务。从队列中读取是按顺序进行的，因此任务 1 始终会在任务 2 之前读取。不过，无法预测结果在工作线程中准备就绪的顺序。有很多硬件和操作系统相关的因素在起作用，所以是的，回溯是随机的，因为结果的顺序是随机的，因为(字符串化的)回溯是发送回父级的结果的一部分。结果也会通过共享队列发回，并且 Pool 在内部处理返回任务 JIT。如果任务返回不成功，整个作业将被标记为不成功，并且进一步到达的任务将被丢弃。一旦作业中的所有任务都返回，只有第一个检索到的异常才会在父级中重新引发。