python multiprocessing.Pool kill *特定*长时间运行或挂起的进程

Question

我需要执行一个包含许多并行数据库连接和查询的池。我想使用 multiprocessing.Pool 或 concurrent.futures ProcessPoolExecutor。 Python 2.7.5

在某些情况下，查询请求花费的时间太长或永远不会完成(挂起/僵尸进程)。我想从超时的 multiprocessing.Pool 或 concurrent.futures ProcessPoolExecutor 中终止特定进程。

这是一个如何终止/重新生成整个进程池的示例，但理想情况下我会尽量减少 CPU 抖动，因为我只想终止在超时秒后未返回数据的特定长时间运行的进程。

由于某些原因，在返回并完成所有结果后，下面的代码似乎无法终止/加入进程池。它可能与发生超时时杀死工作进程有关，但是当他们被杀死时池会创建新的工作进程并且结果符合预期。

from multiprocessing import Pool
import time
import numpy as np
from threading import Timer
import thread, time, sys

def f(x):
    time.sleep(x)
    return x

if __name__ == '__main__':
    pool = Pool(processes=4, maxtasksperchild=4)

    results = [(x, pool.apply_async(f, (x,))) for x in np.random.randint(10, size=10).tolist()]

    while results:
        try:
            x, result = results.pop(0)
            start = time.time()
            print result.get(timeout=5), '%d done in %f Seconds!' % (x, time.time()-start)

        except Exception as e:
            print str(e)
            print '%d Timeout Exception! in %f' % (x, time.time()-start)
            for p in pool._pool:
                if p.exitcode is None:
                    p.terminate()

    pool.terminate()
    pool.join()

Answer 1

我没有完全理解你的问题。你说你想停止一个特定的进程，但是在你的异常处理阶段，你正在调用所有作业的终止。不确定你为什么这样做。另外，我很确定使用 multiprocessing.Pool 中的内部变量不是很安全。说了这么多，我想你的问题是为什么这个程序在超时发生时没有完成。如果那是问题所在，那么以下方法就可以解决问题:

from multiprocessing import Pool
import time
import numpy as np
from threading import Timer
import thread, time, sys

def f(x):
    time.sleep(x)
    return x

if __name__ == '__main__':
    pool = Pool(processes=4, maxtasksperchild=4)

    results = [(x, pool.apply_async(f, (x,))) for x in np.random.randint(10, size=10).tolist()]

    result = None
    start = time.time()
    while results:
        try:
            x, result = results.pop(0)
            print result.get(timeout=5), '%d done in %f Seconds!' % (x, time.time()-start)
        except Exception as e:
            print str(e)
            print '%d Timeout Exception! in %f' % (x, time.time()-start)
            for i in reversed(range(len(pool._pool))):
                p = pool._pool[i]
                if p.exitcode is None:
                    p.terminate()
                del pool._pool[i]

    pool.terminate()
    pool.join()

重点是您需要从池中移除项目；仅仅调用终止它们是不够的。