python - 在 python 多处理中更新共享 tqdm 进度条

Question

我想从生成的进程内部更新进度条，如下所示:

import multiprocessing as mp
import random
import time
from tqdm import tqdm

def test(queue, pbar, lock):
    while True:
        x = queue.get()
        if x is None:
            break
        for i in range(x):
            time.sleep(1)
            lock.acquire()
            pbar.update(1)
            lock.release()

queue = mp.Queue()
lock = mp.Lock()
processes = []
pbar = tqdm(total=5050)
for rank in range(4):
    p = mp.Process(target=test, args=(queue, pbar, lock))
    p.start()
    processes.append(p)
pbar.close()

for idx in range(100):
    queue.put(idx)

for _ in range(4):
    queue.put(None)  # sentinel values to signal subprocesses to exit

for p in processes:
        p.join()  # wait for all subprocesses to finish

上面给出了不一致的更新(进度上下波动)。 我找到了这个answer ，但它们都不适合我，因为我想更新 test 函数内的进度条。我怎样才能做到这一点？

Answer 1

我会稍微调整一下程序:

1.) 创建 update_bar 进程，该进程创建进度条并从另一个队列读取值并使用这些值更新进度条

2.) 此更新进程有 daemon=True 参数，因此退出时不会阻塞

3.) test 进程在启动时接收 bar_queue 并在其中放置值(如果它们想要更新进度条)。

import time
from tqdm import tqdm
import multiprocessing as mp


def test(queue, bar_queue):
    while True:
        x = queue.get()
        if x is None:
            break
        for _ in range(x):
            time.sleep(0.05)
            bar_queue.put_nowait(1)


def update_bar(q):
    pbar = tqdm(total=188)

    while True:
        x = q.get()
        pbar.update(x)


if __name__ == "__main__":
    queue = mp.Queue()
    bar_queue = mp.Queue()

    processes = [
        mp.Process(target=test, args=(queue, bar_queue)) for _ in range(4)
    ]

    # start update progress bar process
    # daemon= parameter is set to True so this process won't block us upon exit
    bar_process = mp.Process(target=update_bar, args=(bar_queue,), daemon=True)
    bar_process.start()

    for p in processes:
        p.start()

    for idx in range(20):
        queue.put(idx)

    for _ in range(4):
        queue.put(None)  # sentinel values to signal subprocesses to exit

    for p in processes:
        p.join()  # wait for all subprocesses to finish

    time.sleep(0.5)