我正在构建一个并行 python 应用程序,它本质上调用了一个围绕外部库的 C 包装器。需要并行性才能在所有 CPU 内核上同时运行计算。
我最终使用了pathos.multiprocessing.ProcessPool
,但这些池缺少maxtaskperchild
标准multiprocessing.Pool
类构造函数的参数(请参阅此处的参考资料)。我需要这个功能,因为 C 库依赖于进程时钟来定义一些执行时间限制,这些时间限制最终会在任务堆积起来时达到。
有没有办法让ProcessPool
经理在给定数量的任务后更新工作流程?
阐明我的意图的示例代码:
from pathos.pools import ProcessPool
from os import getpid
import collections
def print_pid(task_id):
pid = getpid()
return pid
if __name__ == "__main__":
NUM_TASKS = 50
MAX_PER_CHILD = 2
# limit each process to maximum MAX_PER_CHILD tasks
# we would like the pool to exit the process and spawn a new one
# when a task counter reaches the limit
# below argument 'maxtasksperchild' would work with standard 'multiprocessing'
pool = ProcessPool(ncpu=2, maxtasksperchild=MAX_PER_CHILD)
results = pool.map(print_pid, range(NUM_TASKS), chunksize=1)
tasks_per_pid = dict(collections.Counter(results))
print(tasks_per_pid)
# printed result
# {918: 8, 919: 6, 920: 6, 921: 6, 922: 6, 923: 6, 924: 6, 925: 6}
# observe that all processes did more than MAX_PER_CHILD tasks
我试过的
maxtasksperchild
在构造函数中设置ProcessPool
(参见上面的朴素示例)似乎没有做任何事情- 调用
sys.exit()
worker函数使程序挂起 - 我在深入研究源代码时发现了一些提示