如何在多处理中组合超时错误和 tqdm 进度条

Question

我想同时使用超时错误和tqdm进度条执行多处理。

我已经成功地单独尝试了它们。 我应该如何组合逻辑？

目标：

• 进度条应随每次imap_unordered调用而更新

• 应检查每个进程是否存在超时错误

我已经尝试了一百万种方法来组合它们(未显示(。 每次我用 tqdm 包装imap_unordered调用时，我都无法访问"res.next"方法进行超时。

from multiprocessing import Pool, TimeoutError
from tqdm import tqdm
def runner(obj):
    obj.go()
    return obj
def dispatch(objs):
    with Pool() as pool:
        newObjs = list(tqdm(pool.imap_unordered(runner, objs), total=len(objs)))
    # need to find a way to integrate TimeoutError into progress bar
    # I've tried this a million ways using multiprocessing
    # try:
    #     res.next(timeout=10)
    # except TimeoutError:
    #     raise
    return newObjs

代码非常适合进度条。 需要跟踪是否有任何进程超过超时。

Answer 1

您可以在没有迭代器的情况下分配进度条，并使用 update() 手动更新它。

from multiprocessing import Pool, TimeoutError as mpTimeoutError
from tqdm import tqdm

def runner(obj):
    obj.go()
    return obj

def dispatch(objs):
    with Pool() as pool:
        it = pool.imap_unordered(runner, objs)
        pbar = tqdm(total=len(objs))
        new_objs = []
        while True:
            try:
                new_objs.append(it.next(timeout=10))
                pbar.update()
            except mpTimeoutError:
                raise
            except StopIteration:
                # signal that the iterator is exhausted
                pbar.close()
                break
    return new_objs