python - 多处理 - 映射到列表，终止超过超时限制的进程

Question

我有一个要使用多处理修改的元素列表。问题是对于某些特定输入(在尝试之前无法观察到)，my 函数的一部分会停止。我用下面的代码从概念上展示了这一点，其中函数 sometimes_stalling_processing() 偶尔会无限期地停止。

为了说明这一点，我正在使用网络抓取工具处理一堆链接，其中一些链接即使在请求模块中使用超时也会停止。我尝试过不同的方法(例如使用 eventlet)，但得出的结论是在多处理级别处理它可能更容易。

def stable_processing(obs):
    ...
    return processed_obs

def sometimes_stalling_processing(obs):
    ...
    return processed_obs

def extract_info(obs):
    new_obs = stable_processing(obs)
    try:
        new_obs = sometimes_stalling_processing(obs)
    except MyTimedOutError: # error doesn't exist, just here for conceptual purposes
        pass
    return new_obs

pool = Pool(processes=n_threads)
processed_dataset = pool.map(extract_info, dataset)
pool.close()
pool.join()

这个问题 ( How can I abort a task in a multiprocessing.Pool after a timeout? ) 看起来非常相似，但我无法将其转换为使用 map 而不是 apply。我也尝试过使用 eventlet 包，但是 doesn't work .请注意，我使用的是 Python 2.7。

如何使 pool.map() 对个别观察超时并终止 sometimes_stalling_processing？

Answer 1

你可以看看pebble图书馆。

from pebble import ProcessPool
from concurrent.futures import TimeoutError

def sometimes_stalling_processing(obs):
    ...
    return processed_obs

with ProcessPool() as pool:
    future = pool.map(sometimes_stalling_processing, dataset, timeout=10)

    iterator = future.result()

    while True:
        try:
            result = next(iterator)
        except StopIteration:
            break
        except TimeoutError as error:
            print("function took longer than %d seconds" % error.args[1])

documentaion 中的更多示例.