python-3.x - 找到想要的值后如何打破 joblib 中的并行化？

Question

假设我有这样一个函数:

def f(angle):
    if angle % 5==0:
        return angle 

    return None 

然后我按以下方式使用 joblib.Parallel 和 joblib.delayed:

from joblib import Parallel, delayed
results = Parallel(n_jobs=4)(delayed(f)(angle) for angle in range(360))

这给了我所有我想要的值，只有当角度可以被 5 整除时，它们才会与“无”不同。然后我执行以下操作

for x in results:
   if x is not None:
       return x 

有了这个，我返回第一个不是 None 的结果。但是，如果我们需要做一些花费太多时间的硬计算怎么办？然后我们希望它在并行化期间完成。我不想为了找到我想要的值(value)而强制所有并行化。我想如果我能在并行化过程中立即获得这个值，之后并行化停止并给我我想要的值。使用 joblib 库实现该目标的最佳方法是什么？

Answer 1

两年后，但是...

据我所知，joblib 没有任何方法来停止进程中。所以我相信您将不得不求助于使用多处理库:

from time import sleep
from multiprocessing import Pool


def f(angle):
    sleep(5)  # very expensive computations here

    if angle % 5 == 0:
        return angle

    return None


def main():
    p = Pool(4)

    for x in p.imap_unordered(f, range(360)):
        if x is not None:
            p.terminate()
            return x

不幸的是，imap() 和 imap_unordered() 只能将单个参数映射到函数中。因此，如果 f() 需要多个参数，您需要创建一个可产生多个结果的可迭代对象:

from time import sleep
from multiprocessing import Pool


def f(args):
    angle = args[0]
    angle2 = args[1]

    sleep(5)  # very expensive computations here

    if angle * angle2 % 5 == 0:
        return angle

    return None


class iterable(object):
    def __init__(self, x, y):
        self.i = 0
        self.x = range(x)
        self.y = range(y)
        self.sz1 = len(self.x)
        self.sz2 = len(self.y)

    def __iter__(self):
        return self

    def __next__(self):
        self.i += 1

        if self.i >= self.sz1 or self.i >= self.sz2:
            raise StopIteration
        else:
            return self.x[self.i - 1], self.y[self.i - 1]


def main():
    p = Pool(4)

    for x in p.imap_unordered(f, iterable(360, 360)):
        if x is not None:
            p.terminate()
            return x