具有共享队列和结束准则的多处理

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)

当然，还有其他方法可以与生成器共享信号量和中断标志；这种方法使用难看的数据类型，但其优点是不使用全局变量（甚至闭包）。在