Python multiprocessing - best way to understand progress for each process

user_6394941 0
Viewed 300

I would like to know the progress of my processes. At the moment what I am using is not very effective. This is a mwe:

import time
from multiprocessing import Pool as ProcessPool
import progressbar
import random

def some_random_calculation(n):
    with progressbar.ProgressBar(max_value=n) as bar:
        for i in range(0,n):
            time.sleep(1)
            bar.update(i)

if __name__=='__main__':

    arguments = [random.randint(4,10) for i in range(4)]

    pool = ProcessPool(4)
    results = pool.map_async(some_random_calculation, arguments)
    print(results.get())
    pool.close() 
    pool.join()

In this case, I am using progressbar2, however, the output is continuously updated on the same line when there is more than 1 process:

You see from the image that the bars are in sorted order just because after the first bar is ended a new one is created by other processes. When there are multiple processes a single bar is updated on the same line.

I am looking for a fix to my problem, it would be cool to have n bars dynamically updated. However, probably there is a smarter way to get a sense of the progress of different processes. Any advice?

progress-bar multiprocessing python
1 Answers

So this is by far not perfect, the subject is pretty complex if you want to get everything right. But one thing is sure, you should monitor the progress from outside the subprocesses.

The fastest and probably the easiest way to do it would be to have a call-function that returns the status, and the governor outside can keep the user updated on the progress. That would look something like this:

import os, signal
from threading import Thread, enumerate as t_enumerate
from time import time, sleep
from random import random

clear = lambda: os.system('cls' if os.name=='nt' else 'clear')

def sig_handler(signal, frame):
    for t in t_enumerate():
        if t.getName() != 'MainThread':
            t.stop()
    exit(0)
signal.signal(signal.SIGINT, sig_handler)

class worker(Thread):
    def __init__(self, init_value=0):
        Thread.__init__(self)
        self.init_value = init_value
        self.progress = 0
        self.run_state = True
        self.start() # Start ourselves instead of from outside.

    def poll(self):
        return self.progress

    def stop(self):
        self.run_state = False

    def run(self):
        main_thread = None
        for t in t_enumerate():
            if t.getName() == 'MainThread':
                main_thread = t
                break

        while main_thread and self.run_state and main_thread.isAlive():
            for i in range(0, 100):
                self.init_value *= i
                self.progress = i
                sleep(random())
            break # Yea kinda unessecary while loop. meh..

workers = [worker(0) for i in range(4)]

while len(t_enumerate()) > 1:
    clear()
    for index, worker_handle in enumerate(workers):
        progress = worker_handle.poll()
        print(f'Thread {index} is at {progress}/100.')
    sleep(1)

The other approach would be for each thread to acquire a lock on the thread pool before printing. But this adds complexity, for starters, they would all need to sync when it's time to print, so that they don't arbitrarily acquire the lock to print, but you're in some other part of the output process where something else is being printed. Or they would print in the wrong order, or you would need to keep track of which row you should backtrack to re-write..

There's probably going to be a Threading guru here with a better answer, but this is my two cents. Just add a poller function, do a combined status update and live with the very limited processing power it takes to call each thread. Unless you have thousands of them, you won't have any performance impact by calling multiple times.

user_9299990
Related