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

95

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.

MinuitÉclair
la source

Réponses:

80

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))
jfs
la source
7
Je ne vois l'impression qu'après la sortie du code (pas à chaque itération). Avez-vous une suggestion?
Hanan Shteingart
@HananShteingart: Cela fonctionne très bien sur mon système (Ubuntu) avec Python 2 et 3. J'ai utilisé def do_word(*a): time.sleep(.1)comme exemple. Si cela ne fonctionne pas pour vous, créez un exemple de code minimal complet qui illustre votre problème: décrivez à l'aide de mots ce que vous vous attendez à ce qu'il se passe et ce qui se passe à la place, mentionnez comment exécutez votre script Python, quel est votre système d'exploitation, la version Python et postez-le comme une nouvelle question .
jfs
13
J'ai eu le même problème que @HananShteingart: c'est parce que j'essayais d'utiliser Pool.map(). Je ne m'en suis pas rendu compte seulement imap() et je imap_unordered()travaille de cette façon - la documentation dit simplement "Une version plus paresseuse de map ()" mais signifie en réalité "l'itérateur sous-jacent renvoie les résultats au fur et à mesure qu'ils arrivent".
simonmacmullen
@simonmacmullen: la question et ma réponse sont utilisées imap_unordered(). Le problème de Hanan est probablement dû à sys.stderr.write('\r..')(écraser la même ligne pour montrer la progression).
jfs
2
Aussi possible! Je voulais principalement documenter une supposition stupide que j'avais faite - au cas où quelqu'un d'autre le lirait le ferait aussi.
simonmacmullen
94

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
Tim
la source
64
et si pool renvoie une valeur?
Nickpick
11
J'ai créé une liste vide appelée result avant la boucle, puis à l'intérieur de la boucle, faites simplement result.append (x). J'ai essayé cela avec 2 processus et utilisé imap au lieu de map et tout fonctionnait comme je le voulais @nickpick
bs7280
2
donc ma barre de progression est itérée vers de nouvelles lignes au lieu de progresser sur place, une idée pourquoi cela pourrait être?
Austin
2
n'oubliez pas depip install tqdm
M. T
3
@ bs7280 Par result.append (x) vouliez-vous dire result.append (_)? Qu'est-ce que x?
jason
27

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.

reubano
la source
4
Je pense que crée un tas de threads, et chaque thread compte indépendamment
nburn42
1
J'ai des fonctions dans des fonctions qui entraînent une erreur de décapage.
ojunk
21

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)
MinuitÉclair
la source
3
J'ai testé cela pour Python 2.7.6 et rs._number_left semble être le nombre de morceaux restants. Donc, si rs._chunksize n'est pas 1, alors rs._number_left ne sera pas le nombre d'éléments de liste restants.
Allen
Où dois-je mettre ce code? Je veux dire que cela n'est pas exécuté tant que le contenu de rsn'est pas connu et qu'il est un peu tard ou pas?
Wakan Tanka
@WakanTanka: Il va dans le script principal après avoir tourné les fils supplémentaires. Dans mon exemple d'origine, il va dans la boucle "while", où rsa déjà lancé les autres threads.
MidnightLightning
1
Pourriez-vous s'il vous plaît modifier votre question et / ou réponse pour montrer l'exemple de travail minimum. Je ne vois rsdans aucune boucle, je suis un débutant multiprocesseur et cela aiderait. Merci beaucoup.
Wakan Tanka
1
Au moins python 3.5, la solution utilisant _number_leftne fonctionne pas. _number_leftreprésente les morceaux qui restent à traiter. Par exemple, si je veux que 50 éléments soient passés à ma fonction en parallèle, alors pour un pool de threads avec 3 processus _map_async()crée 10 morceaux avec 5 éléments chacun. _number_leftreprésente alors combien de ces morceaux ont été achevés.
mSSM
9

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.

Julien Tourille
la source
je dois changer: [pool.apply_async(my_function, (x,), callback=results.append) for x in dummy_args]pour(pool.apply_async(my_function, (x,), callback=results.append) for x in dummy_args)
David Przybilla
Ce n'est pas vrai. Un objet générateur ne fonctionnera pas ici. Vérifié.
swagatam
9

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.

mrapacz
la source
4

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]
zeawoas
la source
3

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
Aronstef
la source
1

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()
Mott le tuple
la source