python - 多处理线程池未按预期终止

Question

我在 Windows 7 上运行 Python 3.4。

我试图了解更多关于多处理的知识，并通过尝试编写一个函数来对另一个函数调用执行一个干净的小超时。但是我遇到了一个我无法弄清楚的问题。

根据关于多处理的 Python 文档:

Terminate() "Stops the worker processes immediately without completing outstanding work."

但是，当我对此进行测试时，pool.terminate() 似乎在等待工作进程完成而不是杀死它们!

所以当我运行这段代码时:

import multiprocessing.pool
from time import sleep

def timeout(seconds, function, *args, **kwargs):

    pool = multiprocessing.pool.ThreadPool(processes = 1)
    result = pool.apply_async(function, args, kwargs)

    try: result.get(seconds)
    except multiprocessing.context.TimeoutError:
        print("Process timed out")
        pool.terminate()
        pool.join()
        print("Pool terminated")
    finally:
        pool.close()

def worker():
    for n in range(5):
        sleep(1)
        print(n+1)
    print("Process succeeded")

timeout(2.5, worker)

我希望结果是这样的:

1
2
Process timed out
Pool terminated

但我得到的是:

1
2
Process timed out
3
4
5
Process succeeded
Pool terminated

我知道 result.get 引发了 TimeoutError 因为消息“进程超时”打印成功。我知道 pool.terminate() 是出于同样的原因被调用的，它似乎什么也没做!

我觉得这里有些东西我只是不理解。有人可以帮帮我吗？

Answer 1

我不知道为什么，但问题似乎是由 pool.join() 调用引起的，您实际上并不需要它，因为工作进程应该被终止前面的 terminate() 调用。

import multiprocessing.pool
from time import sleep

def timeout(seconds, function, *args, **kwargs):
    pool = multiprocessing.pool.ThreadPool(processes=1)
    result = pool.apply_async(function, args, kwargs)
    try:
        result.get(timeout=seconds)
    except multiprocessing.TimeoutError:
        print("Process timed out")
    pool.terminate()
#    pool.join()  # Don't need this, all worker threads have been stopped.
    print("Pool terminated")

def worker():
    for n in range(5):
        sleep(1)
        print(n+1)
    print("Process succeeded")

timeout(2.5, worker)

输出:

1
2
Process timed out
Pool terminated

无论如何，请注意，自 3.3 版本以来，Pool 对象支持上下文管理协议(protocol)，这意味着 Pool.terminate() 将在使用时自动调用 - 因此函数可以更简洁地写成这样:

def timeout(seconds, function, *args, **kwargs):
    with multiprocessing.pool.ThreadPool(processes=1) as pool:
        result = pool.apply_async(function, args, kwargs)
        try:
            result.get(timeout=seconds)
        except multiprocessing.TimeoutError:
            print("Process timed out")
    print("Pool terminated")