Python 多处理模块：加入超时的进程

Python multiprocessing module: join processes with timeout

提问by brp

im doing an optimization of parameters of a complex simulation. Im using the multiprocessing module for enhancing the performance of the optimization algorithm. The basics of multiprocessing I learned at http://pymotw.com/2/multiprocessing/basics.html. The complex simulation lasts different times depending on the given parameters from the optimization algorithm, around 1 to 5 minutes. If the parameters are chosen very badly, the simulation can last 30 minutes or more and the results are not useful. So I was thinking about build in a timeout to the multiprocessing, that terminates all simulations that last more than a defined time. Heres an abstracted version of the problem:

我正在对复杂模拟的参数进行优化。我使用多处理模块来提高优化算法的性能。我在http://pymotw.com/2/multiprocessing/basics.html学到了多处理的基础知识。根据优化算法的给定参数，复杂模拟持续不同的时间，大约 1 到 5 分钟。如果参数选择得非常糟糕，模拟可能会持续 30 分钟或更长时间，并且结果没有用。所以我在考虑建立多处理超时，终止所有持续超过定义时间的模拟。这是问题的抽象版本：

import numpy as np
import time
import multiprocessing

def worker(num):

    time.sleep(np.random.random()*20)

def main():

    pnum = 10    

    procs = []
    for i in range(pnum):
        p = multiprocessing.Process(target=worker, args=(i,), name = ('process_' + str(i+1)))
        procs.append(p)
        p.start()
        print 'starting', p.name

    for p in procs:
        p.join(5)
        print 'stopping', p.name

if __name__ == "__main__":
    main()

The line p.join(5)defines the timeout of 5 seconds. Because of the for-loop for p in procs:the program waits 5 seconds until the first process is finished and then again 5 seconds until the second process is finished and so on, but i want the program to terminate all processes that last more than 5 seconds. Additionally, if none of the processes last longer than 5 seconds the program must not wait this 5 seconds.

该行p.join(5)定义了 5 秒的超时。由于 for 循环for p in procs:，程序等待 5 秒直到第一个进程完成，然后再等待 5 秒直到第二个进程完成，依此类推，但我希望程序终止持续超过 5 秒的所有进程。此外，如果所有进程的持续时间都不超过 5 秒，则程序不得等待这 5 秒。

采纳答案by dano

You can do this by creating a loop that will wait for some timeout amount of seconds, frequently checking to see if all processes are finished. If they don't all finish in the allotted amount of time, then terminate all of the processes:

您可以通过创建一个循环来执行此操作，该循环将等待一些超时秒数，并经常检查所有进程是否已完成。如果它们没有在分配的时间内全部完成，则终止所有进程：

TIMEOUT = 5 
start = time.time()
while time.time() - start <= TIMEOUT:
    if not any(p.is_alive() for p in procs):
        # All the processes are done, break now.
        break

    time.sleep(.1)  # Just to avoid hogging the CPU
else:
    # We only enter this if we didn't 'break' above.
    print("timed out, killing all processes")
    for p in procs:
        p.terminate()
        p.join()

回答by brp

Thanks to the help of dano I found a solution:

感谢 dano 的帮助，我找到了一个解决方案：

import numpy as np
import time
import multiprocessing

def worker(num):

    time.sleep(np.random.random()*20)

def main():

    pnum = 10    
    TIMEOUT = 5 
    procs = []
    bool_list = [True]*pnum

    for i in range(pnum):
        p = multiprocessing.Process(target=worker, args=(i,), name = ('process_' + str(i+1)))
        procs.append(p)
        p.start()
        print 'starting', p.name

    start = time.time()
    while time.time() - start <= TIMEOUT:
        for i in range(pnum):
            bool_list[i] = procs[i].is_alive()

        print bool_list

        if np.any(bool_list):  
            time.sleep(.1)  
        else:
            break
    else:
        print("timed out, killing all processes")
        for p in procs:
            p.terminate()

    for p in procs:
        print 'stopping', p.name,'=', p.is_alive()
        p.join()

if __name__ == "__main__":
    main()

Its not the most elegant way, Im sure there is a better way than using bool_list. Processes that are still alive after the timeout of 5 seconds will be killed. If you are setting shorter times in the worker function than the timeout, you will see that the program stops before the timeout of 5 seconds is reached. Im still open for more elegant solutions if there are :)

这不是最优雅的方式，我相信有比使用bool_list. 超时 5 秒后仍然存活的进程将被杀死。如果您在工作函数中设置的时间比超时时间短，您将看到程序在达到 5 秒超时之前停止。如果有的话，我仍然愿意接受更优雅的解决方案:)

回答by Raiden Drake

If you want to kill all the processes you could use the Pool from multiprocessing you'll need to define a general timeout for all the execution as opposed of individual timeouts.

如果你想杀死所有可以使用多处理池的进程，你需要为所有执行定义一个通用超时，而不是单个超时。

import numpy as np
import time
from multiprocessing import Pool

def worker(num):
    xtime = np.random.random()*20
    time.sleep(xtime)
    return xtime

def main():

    pnum = 10
    pool = Pool()
    args = range(pnum)
    pool_result = pool.map_async(worker, args)

    # wait 5 minutes for every worker to finish
    pool_result.wait(timeout=300)

    # once the timeout has finished we can try to get the results
    if pool_result.ready():
        print pool_result.get(timeout=1)

if __name__ == "__main__":
    main()

This will get you a list with the return values for all your workers in order.
More information here: https://docs.python.org/2/library/multiprocessing.html#module-multiprocessing.pool

这将为您提供一个按顺序列出所有工人的返回值的列表。
更多信息：https: //docs.python.org/2/library/multiprocessing.html#module-multiprocessing.pool