[Conseil] Diviser des tâches - Module multiprocessing

Bonsoir tout le monde.


J'ai actuellement une fonction lourde et mal optimisée. Mais je bosse déjà là dessus et je pense que malgré tous mes efforts le temps d'exécution restera conséquent (ça se chiffre en minutes...au moins). Mais c'est une autre histoire. Cette fonction doit être appelée un nombre conséquent de fois. Je pense que diviser les tâches en plusieurs processus indépendants est particulièrement indiqué dans ce cas (vous pouvez me détromper, il n'y a pas de soucis).

J'ai fait quelques recherches sur le module multiprocessing de python. Mais je ne suis jamais tombé exactement sur ce que je cherche. Voici mon besoin: admettons que j'ai une liste de tâches à effectuer, un nombre inconnus nécessitant une durée d'exécution inconnue. J'aimerais pouvoir simplement lancer en parallèle un nombre défini de processus et ainsi traiter l'ensemble des tâches. Dans l'idéal, une fois une tâche de finie une autre se lance même si les autres tournent toujours. Mais il faudrait qu'il n'y jamais plus de n processus en même temps.

Je ne crois pas au père noël, ça doit très sûrement existé. ;) Mon petit doigt me dit d'utiliser pool(), mais les exemples que j'ai pu lire ne m'ont pas convaincus.


Pour le moment je suis parvenu à bricoler quelque chose qui traite mes tâches par blocs de n processus. C'est déjà pas mal même si pas très académique:

Code:

---

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23

from multiprocessing import Process, Manager import time import random   def heavy_func(case_id, jobs_done): """Dummy heavy function""" time.sleep(random.randint(1,10)) jobs_done[case_id] = 'Done'   if __name__ == '__main__':   jobs_to_do = ['Case%i' % (i,) for i in range(10)] jobs_done = Manager().dict()   n_procs = 2 procs = [Process(target=heavy_func, args=(c, jobs_done)) for c in jobs_to_do] for i in range(0, len(jobs_to_do), n_procs): for p in procs[i:i+n_procs]: p.start() for p in procs[i:i+n_procs]: p.join()   print jobs_done

---

J'utilise ici Manager() pour partager entre les processus un même dictionnaire. Actuellement mes résultats sont sortis sous cette forme, donc ça m'arrange bien.


Ju

Salut,

Compte tenu de la syntaxe de vos print, vous travaillez avec Python2.
Python 3.2 introduit le module concurrent.futures qui encapsule les fonctionnalités des modules threading et multiprocessing pour réaliser facilement le mapping de N taches sur M process, N > M.
Plein d'exemples traînent sur Internet et dans la doc.
Il existe aussi un backport de ce module pour Python 2.

Bon courage,
- W

Salut,

Merci Wiztricks, je ne connaissais pas concurrent.futures. Je vais étudier la doc pour voir si je trouve mon bonheur.

Effectivement, j'aurais du mentionné que je bosse avec Python 2.7, c'est important. Je ne peux malheureusement pas encore migrer vers une version 3.* pour causes professionnelles. En faisant des recherches je suis tombé sur des modules comme asyncore, mais j'en ai pas lu que du bien.

J'ai vu dernièrement qu'avec la version 3.4 vient asyncio qui doit rendre la programmation asynchrone plus aisée. Mais là encore je n'y ai pas droit...

Mon besoin est tout de même simple, je pense m'en sortir avec uniquement les libs standards. Je ne cherche pas à faire des calculs distribués sur plusieurs machines en réseau.

Je vais essayer de pondre un truc moi-même, mais si quelqu'un à un exemple de prêt je suis preneur.


Ju

Salut,

asyncore est plutôt oriente protocole réseau. En dessous ca multiplexe select ou equivalent. C'est mono-threads.
Je ne vois pas trop le rapport avec (ce que j'ai compris de) votre besoin.

- W

Hello,

Deux idées comme ça en passant, qui pourraient peut-être faire le taf (sans garantie, je suis pas très au point sur ce genre de probleme).

Option 1: Un sémaphore, que chaque sous-process doit acquérir pour lancer son job. Tous les sous process sont lancés en même temps, mais seulement le nombre voulu sont en fonctionnement à un instant T

Option 2: Même principe que 1, sauf que chaque lancement process est encapsulé dans un simple thread, qui lui lancera le sous-process lorsqu'il aura acquis le sémaphore. Ça devrait permettre une meilleure répartition sur les différents processeurs.

Salut,

Citation:

---

Envoyé par Petibidon

Option 1: Un sémaphore, que chaque sous-process doit acquérir pour lancer son job. Tous les sous process sont lancés en même temps, mais seulement le nombre voulu sont en fonctionnement à un instant T

Option 2: Même principe que 1, sauf que chaque lancement process est encapsulé dans un simple thread, qui lui lancera le sous-process lorsqu'il aura acquis le sémaphore. Ça devrait permettre une meilleure répartition sur les différents processeurs.

---

Sur le papier ca fonctionne mais je trouve que c'est un peu lourd car (1) cree N process qui ne foutent rien (mais qui consomment de la ressource) en attendant d’acquérir le sémaphore. (2) est encore plus lourd puisqu'on encapsule
cela avec des threads.
concurrent.futures (voir les sources) crée un seul thread qui gère une queue.Queue (un objet Python derrière lequel se cache un sémaphore) des taches en attente et s'occupe de démarrer la tache suivante des qu'un slot (process ou thread se libère).
L'avantage est qu'on n'a pas a coder l'infrastructure qui gère le dispatch des N taches sur M process ou threads, ni la réalisation des interfaces qui vont bien.

- W

Merci à tous les deux.

Je ne peux malheureusement pas m'orienter vers concurrent.futures car je serais obligé de lancer la lourde machinerie administrative de ma boite afin de valider le module... Mais je vais voir dans les sources si je ne peux pas soit m'inspirer soit honteusement copier une partie. J'ai pas trop d'espoir quand même.

Pour ce qui est des sémaphores, ça m'était totalement passé à côté. Je viens de faire un essai assez concluant:

Code:

---

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34

from multiprocessing import Process, Manager, Semaphore import time import random   def heavy_func(case_id): """Dummy heavy function""" print '%s is running' % (case_id,) time.sleep(random.randint(1,20))   def heavy_func_wrapper(case_id, jobs_done, pool_sema): pool_sema.acquire() heavy_func(case_id) pool_sema.release() jobs_done[case_id] = 'Done'     if __name__ == '__main__':   n_procs = 4   jobs_to_do = ['Case%i' % (i,) for i in range(10)] jobs_done = Manager().dict() pool_sema = Semaphore(n_procs)   procs = [Process(target=heavy_func_wrapper, args=(job, jobs_done, pool_sema)) for job in jobs_to_do]   for p in procs: p.start()   for p in procs: p.join()   print jobs_done

---

ça à l'air de fonctionner. Mais je dois encore bosser pour mieux comprendre ces outils. Je pensais placer le pool_sema.acquire() avant p.start(). Mais tous les processus se lancent tout de même à la suite. J'ai alors pensé faire ceci sans succès:

Code:

---

1 2 3 4 5

for p in procs: pool_sema.acquire() p.start() p.join() pool_sema.re()

---

Les processus se lancent à la suite, et chacun attend que le précèdent ai fini. En même temps c'est un peu le but de la fonction de join().. Bref, j'ai écrit un wrapper vite fait. ça a au moins l'avantage de ne pas toucher du tout à la fonction lourde comme c'était le cas avant. Il faut que je fasse un essai grandeur nature pour évaluer si trop de mémoire est bouffée avec cette méthode.

Ju

Citation:

---

Envoyé par Julien N

Je ne peux malheureusement pas m'orienter vers concurrent.futures car je serais obligé de lancer la lourde machinerie administrative de ma boite afin de valider le module... Mais je vais voir dans les sources si je ne peux pas soit m'inspirer soit honteusement copier une partie. J'ai pas trop d'espoir quand même.

---

C'est du "pure Python".
Rien ne vous empêche d'inclure tout le module dans votre code sans qu'il soit installe "officiellement" dans site-package.
La seule différence sera que vous serez (aussi) responsable de ce code la.
Mais comme il est documente, teste,... ce sera toujours "moins lourd" que de re-développer.

- W

Citation:

---

Envoyé par wiztricks

Rien ne vous empêche d'inclure tout le module dans votre code sans qu'il soit installe "officiellement" dans site-package.

---

Je viens de faire un test plutôt concluant. Le module est léger (4 fichiers, ça va) et j'ai trouvé très facilement de la doc et des exemples sur internet. Que demander de plus?

J'ai produit ceci en m'inspirant de l'exemple fourni dans la doc:

Code:

---

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22

import time import random from process import ProcessPoolExecutor   def heavy_func(case_id, a): """Dummy heavy function""" print '%s is running with a=%i' % (case_id, a) time.sleep(random.randint(1,10)) return 'Done'   def multiprocessed_func(func, jobs_to_do, nprocs=None): # Let the executor divide the work among processes by using 'map'. jobs_done = {} with ProcessPoolExecutor(max_workers=nprocs) as executor: for job, res in [(job, executor.submit(heavy_func, *job)) for job in jobs_to_do]: jobs_done[job] = res.result() return jobs_done   if __name__ == '__main__':   jobs_to_do = [('Case%i' % (i,), i) for i in range(6)] print multiprocessed_func(heavy_func, jobs_to_do, 4)

---

J'en suis satisfait, c'est à peu de chose près ce à quoi je pensais initialement. Si on me pose problème avec ce module, je pourrais toujours utiliser ma version bricolée.

Je mets en résolu, mais je suis ouvert aux commentaires. Merci à vous deux pour vos réponses.

Ju

PS: Si en lieu et place de from concurrent.futures import ProcessPoolExecutor j'ai from process import ProcessPoolExecutor c'est parce que j'ai copié directement les fichiers sources dans mon dossier de travail (le fichier process.py). D'ailleurs j'ai du également modifié cette ligne import _base dans process.py