我正在使用multiprocessing库编写一个程序来计算一些东西。需要计算大约10K个输入,每个输入的计算时间介于0.2秒到10秒之间。
我目前使用的方法是使用一个进程池:
# Inputs
signals = [list(s) for s in itertools.combinations_with_replacement(possible_inputs, 3)]
# Compute
with mp.Pool(processes = N) as p:
p.starmap(compute_solutions, [(s, t0, tf, folder) for s in signals])
print (" | Computation done.")
我注意到在检查最后300/400个输入时,程序变得非常缓慢。我的问题是: Pool
和 starmap()
如何运作?
根据我的观察,如果我有10K个输入和N = 4
(4个进程),那么前2,500个输入分配给第一个进程,接下来的2,500个分配给第二个进程,...每个进程按顺序处理其输入。
这意味着,如果有些进程在其他进程之前清除了队列,它们就不会获得要执行的新任务。
这正确吗?
如果是这样,我怎样才能拥有一个更智能的系统,可以用这个伪代码表示:
workers = Initialize N workers
tasks = A list of the tasks to perform
for task in tasks:
if a worker is free:
submit task to this worker
else:
wait
感谢您的帮助 :)
N.B:不同的map函数有什么区别。我相信存在map()
、imap_unordered()
、imap
和starmap
。
它们之间有什么区别?我们应该在什么情况下使用哪一个?