Afficher la progression d'un appel de pool multitraitement Python imap_unordered?

J'ai un script qui effectue avec succès un ensemble de tâches de pool multitraitement avec un imap_unordered()appel:

p = multiprocessing.Pool()
rs = p.imap_unordered(do_work, xrange(num_tasks))
p.close() # No more work
p.join() # Wait for completion

Cependant, mon num_tasksest d'environ 250000, et donc le join()verrouillage du thread principal pendant environ 10 secondes, et j'aimerais pouvoir faire écho à la ligne de commande de manière incrémentielle pour montrer que le processus principal n'est pas verrouillé. Quelque chose comme:

p = multiprocessing.Pool()
rs = p.imap_unordered(do_work, xrange(num_tasks))
p.close() # No more work
while (True):
  remaining = rs.tasks_remaining() # How many of the map call haven't been done yet?
  if (remaining == 0): break # Jump out of while loop
  print "Waiting for", remaining, "tasks to complete..."
  time.sleep(2)

Existe-t-il une méthode pour l'objet de résultat ou le pool lui-même qui indique le nombre de tâches restantes? J'ai essayé d'utiliser un multiprocessing.Valueobjet comme compteur ( do_workappelle une counter.value += 1action après avoir effectué sa tâche), mais le compteur n'atteint que ~ 85% de la valeur totale avant d'arrêter l'incrémentation.

Il n'est pas nécessaire d'accéder aux attributs privés de l'ensemble de résultats:

from __future__ import division
import sys

for i, _ in enumerate(p.imap_unordered(do_work, xrange(num_tasks)), 1):
    sys.stderr.write('\rdone {0:%}'.format(i/num_tasks))

Mon préféré - vous donne une jolie petite barre de progression et une ETA d'achèvement pendant que les choses se déroulent et s'engagent en parallèle.

from multiprocessing import Pool
import tqdm

pool = Pool(processes=8)
for _ in tqdm.tqdm(pool.imap_unordered(do_work, tasks), total=len(tasks)):
    pass

J'ai constaté que le travail était déjà fait au moment où j'ai essayé de vérifier sa progression. C'est ce qui a fonctionné pour moi en utilisant tqdm .

pip install tqdm

from multiprocessing import Pool
from tqdm import tqdm

tasks = range(5)
pool = Pool()
pbar = tqdm(total=len(tasks))

def do_work(x):
    # do something with x
    pbar.update(1)

pool.imap_unordered(do_work, tasks)
pool.close()
pool.join()
pbar.close()

Cela devrait fonctionner avec tous les types de multitraitement, qu'ils bloquent ou non.

Trouvé une réponse moi - même avec un peu plus creuser: Jeter un oeil à la __dict__de l' imap_unorderedobjet résultat, je l' ai trouvé a un _indexattribut qui , chaque pas l' exécution des tâches. Donc, cela fonctionne pour la journalisation, enveloppé dans la whileboucle:

p = multiprocessing.Pool()
rs = p.imap_unordered(do_work, xrange(num_tasks))
p.close() # No more work
while (True):
  completed = rs._index
  if (completed == num_tasks): break
  print "Waiting for", num_tasks-completed, "tasks to complete..."
  time.sleep(2)

Cependant, j'ai trouvé que le fait d'échanger le imap_unorderedpour un map_asyncaboutissait à une exécution beaucoup plus rapide, bien que l'objet de résultat soit un peu différent. Au lieu de cela, l'objet de résultat de map_asynca un _number_leftattribut et une ready()méthode:

p = multiprocessing.Pool()
rs = p.map_async(do_work, xrange(num_tasks))
p.close() # No more work
while (True):
  if (rs.ready()): break
  remaining = rs._number_left
  print "Waiting for", remaining, "tasks to complete..."
  time.sleep(0.5)

Je sais que c'est une question assez ancienne, mais voici ce que je fais quand je veux suivre la progression d'un pool de tâches en python.

from progressbar import ProgressBar, SimpleProgress
import multiprocessing as mp
from time import sleep

def my_function(letter):
    sleep(2)
    return letter+letter

dummy_args = ["A", "B", "C", "D"]
pool = mp.Pool(processes=2)

results = []

pbar = ProgressBar(widgets=[SimpleProgress()], maxval=len(dummy_args)).start()

r = [pool.apply_async(my_function, (x,), callback=results.append) for x in dummy_args]

while len(results) != len(dummy_args):
    pbar.update(len(results))
    sleep(0.5)
pbar.finish()

print results

Fondamentalement, vous utilisez apply_async avec un callbak (dans ce cas, il s'agit d'ajouter la valeur renvoyée à une liste), vous n'avez donc pas à attendre pour faire autre chose. Ensuite, dans une boucle while, vous vérifiez la progression du travail. Dans ce cas, j'ai ajouté un widget pour le rendre plus joli.

Le résultat:

4 of 4                                                                         
['AA', 'BB', 'CC', 'DD']

J'espère que ça aide.

Comme suggéré par Tim, vous pouvez utiliser tqdmet imappour résoudre ce problème. Je viens de tomber sur ce problème et de peaufiner la imap_unorderedsolution, afin que je puisse accéder aux résultats du mappage. Voici comment ça fonctionne:

from multiprocessing import Pool
import tqdm

pool = multiprocessing.Pool(processes=4)
mapped_values = list(tqdm.tqdm(pool.imap_unordered(do_work, range(num_tasks)), total=len(values)))

Si vous ne vous souciez pas des valeurs renvoyées par vos travaux, vous n'avez pas besoin d'affecter la liste à une variable.

pour tous ceux qui recherchent une solution simple fonctionnant avec Pool.apply_async():

from multiprocessing import Pool
from tqdm import tqdm
from time import sleep


def work(x):
    sleep(0.5)
    return x**2

n = 10

p = Pool(4)
pbar = tqdm(total=n)
res = [p.apply_async(work, args=(
    i,), callback=lambda _: pbar.update(1)) for i in range(n)]
results = [p.get() for p in res]

J'ai créé une classe personnalisée pour créer une impression de progression. Maby cela aide:

from multiprocessing import Pool, cpu_count


class ParallelSim(object):
    def __init__(self, processes=cpu_count()):
        self.pool = Pool(processes=processes)
        self.total_processes = 0
        self.completed_processes = 0
        self.results = []

    def add(self, func, args):
        self.pool.apply_async(func=func, args=args, callback=self.complete)
        self.total_processes += 1

    def complete(self, result):
        self.results.extend(result)
        self.completed_processes += 1
        print('Progress: {:.2f}%'.format((self.completed_processes/self.total_processes)*100))

    def run(self):
        self.pool.close()
        self.pool.join()

    def get_results(self):
        return self.results

Essayez cette approche simple basée sur les files d'attente, qui peut également être utilisée avec la mise en commun. N'oubliez pas que l'impression de quoi que ce soit après le lancement de la barre de progression entraînera son déplacement, au moins pour cette barre de progression particulière. (Progrès de PyPI 1.5)

import time
from progress.bar import Bar

def status_bar( queue_stat, n_groups, n ):

    bar = Bar('progress', max = n)  

    finished = 0
    while finished < n_groups:

        while queue_stat.empty():
            time.sleep(0.01)

        gotten = queue_stat.get()
        if gotten == 'finished':
            finished += 1
        else:
            bar.next()
    bar.finish()


def process_data( queue_data, queue_stat, group):

    for i in group:

        ... do stuff resulting in new_data

        queue_stat.put(1)

    queue_stat.put('finished')  
    queue_data.put(new_data)

def multiprocess():

    new_data = []

    groups = [[1,2,3],[4,5,6],[7,8,9]]
    combined = sum(groups,[])

    queue_data = multiprocessing.Queue()
    queue_stat = multiprocessing.Queue()

    for i, group in enumerate(groups): 

        if i == 0:

            p = multiprocessing.Process(target = status_bar,
                args=(queue_stat,len(groups),len(combined)))
                processes.append(p)
                p.start()

        p = multiprocessing.Process(target = process_data,
        args=(queue_data, queue_stat, group))
        processes.append(p)
        p.start()

    for i in range(len(groups)):
        data = queue_data.get() 
        new_data += data

    for p in processes:
        p.join()