python - 具有共享队列和结束条件的多处理

Question

我有这个原始函数，我想切换到多进程:

def optimal(t0, tf, frequences, delay, ratio = 0):

    First = True                            # First        
    for s in delay:
        delay = 0                           # delay between signals,
        timelines = list()

        for i in range(len(frequences)):
            timelines.append(time_builder(frequences[i], t0+delay, tf))
            delay += s

       trio_overlap = trio_combination(timelines, ratio)

        valid = True
        for items in trio_overlap.values():
            if len(list(set(items))) == len(items):
                continue
            else:
                valid = False

        if not valid:
            continue

        overlap = duo_combination(timelines)

    optimal = ... depending of conditions        
    return optimal

如果valid = True测试后，它将计算一个名为 optim_param 的优化参数并尽量减少它。如果低于某个阈值，optim_param < 0.3 , 我跳出循环并以此值作为我的答案。

我的问题是，随着我开发模型，复杂性开始上升，单线程计算耗时过长。我想并行处理计算。由于每个进程都必须将使用 s 值获得的结果与当前最优值进行比较，因此我尝试实现一个队列。

这是我第一次做多处理，即使我认为我在正确的轨道上，我也觉得我的代码有点困惑和不完整。我能得到一些帮助吗？

谢谢:D

Answer 1

考虑使用 Pool.imap_unordered，而不是为每个案例手动创建流程。诀窍是如何在获得可通过的结果时干净地关闭:您可以通过传递一个提前退出的生成器来实现这一点，以防设置了它检查每个周期的标志。主程序从迭代器中读取，保持看到的最佳结果，并在足够好时设置标志。最后一个技巧是减慢从生成器读取的(内部)线程，以防止在获得良好结果后必须等待(或不干净地杀死)的计划任务大量积压。给定池中的进程数，可以使用信号量实现该步调。

这是一个示例(带有简单的分析)来演示:

import multiprocessing,threading,os

def interrupted(data,sem,interrupt):
  for x in data:
    yield x
    sem.acquire()
    if interrupt: break

def analyze(x): return x**2

np=os.cpu_count()
pool=multiprocessing.Pool(np)
sem=threading.Semaphore(np-1)
token=[]                        # mutable

vals=pool.imap_unordered(analyze,interrupted(range(-10,10),sem,token))
pool.close()                    # optional: to let processes exit faster

best=None
for res in vals:
  if best is None or res<best:
    best=res
    if best<5: token.append(None) # make it truthy
  sem.release()
pool.join()

print(best)

当然还有其他方法可以与生成器共享信号量和中断标志；这种方式使用丑陋的数据类型，但具有不使用全局变量(甚至闭包)的优点。