python - 如何在多处理中结合 TimeoutError 和 tq​​dm 进度条？

Question

我想使用 TimeoutError 和 tq​​dm 进度条执行多重处理。

我已经成功地分别尝试了它们。我该如何结合逻辑？

目标:

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

• 每个进程都应该检查 TimeoutError

我已经尝试了一百万种方法来组合它们(未显示)。每次我用 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