一旦其中一个工人满足特定条件，终止Python多处理程序

Question

我正在使用其多处理模块编写Python程序。该程序调用许多辅助函数，每个辅助函数产生一个随机数。
一旦其中一名工人生产的数值大于0.7，我就需要终止程序 。

下面是我的程序，其中“ 如何执行 ”部分尚未填写。任何想法？谢谢。

import time
import numpy as np
import multiprocessing as mp
import time
import sys

def f(i):
    np.random.seed(int(time.time()+i))

    time.sleep(3)
    res=np.random.rand()
    print "From i = ",i, "       res = ",res
    if res>0.7:
        print "find it"
        # terminate  ???? Question: How to do this???


if __name__=='__main__':
    num_workers=mp.cpu_count()
    pool=mp.Pool(num_workers)
    for i in range(num_workers):
        p=mp.Process(target=f,args=(i,))
        p.start()

Answer 1

os.kill()像大锤一样，没有任何方法可以阻止其他短小的暴力。不要去那

为了明智地做到这一点，您需要重新设计基本方法：主流程和工作流程需要相互通信。

我会充实它，但是到目前为止，该示例 太
简单了，无法使用。例如，按照书面规定，最多num_workers只能rand()进行调用，因此没有理由相信其中任何一个必须大于0.7。

一旦worker函数增长了一个循环，它就会变得更加明显。例如，工作人员可以检查mp.Event循环顶部是否设置了，如果退出则直接退出。主要过程将设置Event希望工人停止的时间。

当工人mp.Event发现值>
0.7时，可以设置一个不同的值。主要过程将等待该事件Event，然后设置“停止时间”Event以供工作人员查看，然后执行通常的循环-.join()使工作人员彻底关闭。

假设工人要继续努力，直到至少有人发现一个值>
0.7，这才是一种充实，干净的解决方案。请注意，我已将其删除numpy，因为它与这段代码无关。在任何支持multiprocessing以下平台的库存Python下，此处的代码都可以正常工作：

import random
from time import sleep

def worker(i, quit, foundit):
    print "%d started" % i
    while not quit.is_set():
        x = random.random()
        if x > 0.7:
            print '%d found %g' % (i, x)
            foundit.set()
            break
        sleep(0.1)
    print "%d is done" % i

if __name__ == "__main__":
    import multiprocessing as mp
    quit = mp.Event()
    foundit = mp.Event()
    for i in range(mp.cpu_count()):
        p = mp.Process(target=worker, args=(i, quit, foundit))
        p.start()
    foundit.wait()
    quit.set()

和一些示例输出：

0 started
1 started
2 started
2 found 0.922803
2 is done
3 started
3 is done
4 started
4 is done
5 started
5 is done
6 started
6 is done
7 started
7 is done
0 is done
1 is done

一切都彻底关闭：没有回溯，没有异常终止，没有留下僵尸进程……像哨子一样干净。

杀了它

正如@noxdafox所指出的，有一种Pool.terminate()方法可以在所有平台上尽其所能来杀死工作进程，无论它们在做什么（例如，在Windows上称为平台TerminateProcess()）。我不建议将其用于生产代码，因为突然终止进程可能会使各种共享资源处于不一致状态，或者使它们泄漏。在multiprocessing文档中有关于此的各种警告，您应该在其中添加OS文档。

不过，这可能很方便！这是使用此方法的完整程序。请注意，我将临界值提高到0.95，以使其花费的时间比眨眼的时间更长：

import random
from time import sleep

def worker(i):
    print "%d started" % i
    while True:
        x = random.random()
        print '%d found %g' % (i, x)
        if x > 0.95:
            return x # triggers callback
        sleep(0.5)

# callback running only in __main__
def quit(arg):
    print "quitting with %g" % arg
    # note: p is visible because it's global in __main__
    p.terminate()  # kill all pool workers

if __name__ == "__main__":
    import multiprocessing as mp
    ncpu = mp.cpu_count()
    p = mp.Pool(ncpu)
    for i in range(ncpu):
        p.apply_async(worker, args=(i,), callback=quit)
    p.close()
    p.join()

和一些示例输出：

$ python mptest.py
0 started
0 found 0.391351
1 started
1 found 0.767374
2 started
2 found 0.110969
3 started
3 found 0.611442
4 started
4 found 0.790782
5 started
5 found 0.554611
6 started
6 found 0.0483844
7 started
7 found 0.862496
0 found 0.27175
1 found 0.0398836
2 found 0.884015
3 found 0.988702
quitting with 0.988702
4 found 0.909178
5 found 0.336805
6 found 0.961192
7 found 0.912875
$ [the program ended]