Popen : lancer des processus fils

Version imprimable

14/01/2014, 23h13
roipoussiere

Popen : lancer des processus fils
Bonsoir.

Je suis récemment passé de os.system à subprocess.Popen().
La commande est appelée plusieurs centaines de fois à la suite dans mon programme. Avant de lancer un programme, j'ai un petit print qui affiche la progression globale :
Code:

1 2 3 cmd = ... print "traitement fichier n°", i p = subprocess.Popen(shlex.split(cmd))
Ce qui est très bien avec Popen, c'est les processus séparés, j'ai d'énormes gains de perfs par rapport à os.system ! :mrgreen:
Par contre j'ai plusieurs petit soucis :
- Je n'ai plus la possibilité d'afficher la progression générale, tous les process se lancent en même temps.
- Je ne peux plus faire de ctr+C pour interrompre mon script, puisque celui-ci est terminé au bout de quelques secondes, après avoir lancé tous les processus.

Sur la doc, j'ai découvert le flag CREATE_NEW_PROCESS_GROUP qui me semblait être une bonne piste, mais je n'ai pas réussi a le faire fonctionner, visiblement cela ne fonctionne que sous windows.

Merci à vous,

Roipou
15/01/2014, 08h06
emmanuel_dumas
Bonjour

Utilisez-vous la command wait ?
Code:

1 2 p = subprocess.Popen(shlex.split(cmd)) p.wait()
Est-ce volontaire que nous n'en n'avez pas mis ou est-ce un oubli ?

Ne pas en mettre permet de paralléliser les programmes, mais pose les quelques problèmes que vous citez.
En mettre demande un peu de travail pour réintroduire du parallélisme, mais réduit les problèmes.

Cordialement
Emmanuel
15/01/2014, 08h30
VinsS

Salut,

Il faudrait se poser la question de ton intention.

Veux-tu lancer tous les process à la volée ou conserver la main sur leur exécution ?

Les process peuvent-ils, éventuellement, s'exécuter dans des threads ? On ne connait pas la commande exécutée.

En plus du commentaire de A. Dumas, Popen possède une méthode communicate() qui te permet, comme son nom l'indique, de communiquer avec lui et ces process ont un PID aisément récupérable qui te permettrait par exemple d'y mettre fin avec kill().

Éventuellement, poste un exemple utilisable.
15/01/2014, 09h52
roipoussiere

Bonjour, merci pour vos réponses.
Je n'utilise pas la commande wait() car je souhaite conserver le parallélisme. Comme je lai dit, j'ai un gros gain de temps d'execution, de l'ordre de 3 fois plus rapide, ce serait bête de s'en priver.

Je souhaite donc conserver la main sur les process. Comme l'indique le titre, je pensais qu'il était possible d'avoir des processus fils attachés au script, ainsi en tuant le père on tue ses fils, mais je dois être un peu naïf, c'est sans doute-pas aussi facile.

Citation:

En plus du commentaire de A. Dumas, Popen possède une méthode communicate() qui te permet, comme son nom l'indique, de communiquer avec lui et ces process ont un PID aisément récupérable qui te permettrait par exemple d'y mettre fin avec kill().

C'est déjà une info intéressante.
Ce serait une solution de mettre les process dans une file en se débrouillant pour que le nombre de process en parallèle soit égal au nombre de coeurs, et de tuer tous les process d'un coup si on reçoit un signal d'extinction (SIGINT) ?

Sinon je ne pense pas que la commande puisse s'executer dans des threads, ce script lance des commande d'un logiciel de CAO 3D paramétrique (openscad).

Citation:

Éventuellement, poste un exemple utilisable.

Du coup, non, à moins que vous ne téléchargiez le soft je n'ai pas d'exemple utilisable à proposer.

Bonjour

Citation:

Envoyé par roipoussiere

Je n'utilise pas la commande wait() car je souhaite conserver le parallélisme

wait n'est pas équivalent à "pas de parallélisme". Suffit de le placer correctement, si besoin.

somme des temps d'exécution (+ intendance)
Code:

1 2 3 4 5 6 7 >>> def methode_os_system(combien): ... for i in range(combien): ... _ = os.system("/bin/sleep 1") ... >>> t0=time(); methode_os_system(5); time()-t0 5.024528980255127 >>>
somme des temps des lancements (+ intendance)
Code:

1 2 3 4 5 6 7 8 >>> def methode_popen_sans_wait(combien): ... for i in range(5): ... _ = Popen(("/bin/sleep", "1")) ... >>> >>> t0=time(); methode_popen_sans_wait(5); time()-t0 0.00657200813293457 >>>
temps du plus long (+ intendance)
Code:

1 2 3 4 5 6 7 8 9 10 >>> def methode_popen_wait_final(combien): ... returns = list() ... for i in range(combien): ... returns.append(Popen(("/bin/sleep", "1"))) ... for ret in returns: ... ret.wait() ... >>> >>> t0=time(); methode_popen_wait_final(5); time()-t0 1.0086278915405273
retour à os.system : somme des temps (+ intendance)
Code:

1 2 3 4 5 6 7 8 >>> def methode_popen_wait_individuel(combien): ... for i in range(combien): ... ret = Popen(("/bin/sleep", "1")) ... ret.wait() ... >>> >>> t0=time(); methode_popen_wait_individuel(5); time()-t0 5.019736051559448
Citation:

Envoyé par roipoussiere

Ce serait une solution de mettre les process dans une file en se débrouillant pour que le nombre de process en parallèle soit égal au nombre de coeurs, et de tuer tous les process d'un coup si on reçoit un signal d'extinction (SIGINT) ?

Lancer N fois le même exécutable avec des arguments différents et avec autant de processes en parallèle qu'il y a des coeurs sur la machine : Pool fait ça très bien. Et on peut même encore "causer"

15/01/2014, 11h15
roipoussiere

Merci plxpy pour les extraits de code.

J'imagine donc que parmi les solutions que tu as proposés, le 3ème extrait de code (temps du plus long (+ intendance)) est la seule qui permet de paralléliser les processus sans quitter le script pricipal ?
Par contre je n'ai pas compris un truc : on lance tous les process en les chargeant dans une liste, puis on attends la fin de chaque processus. J'ai testé avec plus de 10 process et j'ai toujours 1sec.
Si on lance plus de process qu'on a de coeurs, ça doit durer plus de 1 sec. non ? Dans ma logique ce serait : arrondi_sup ( nb_de_process / (temps_d'exec_process*nb_de_coeurs) ) -> avec on 4 coeurs, si on lance 4 process : 1sec, 5 process : 2 sec.

Citation:

Lancer N fois le même exécutable avec des arguments différents et avec autant de processes en parallèle qu'il y a des coeurs sur la machine : Pool fait ça très bien. Et on peut même encore "causer"

C'est tout à fait ça. Pool semble intéressant, mais ce qui m'a fait passer à Popen c'est la possibilité de rediriger la sortie standard et d'erreur. Je n'ai pas trouvé comment faire ça dans la doc de Pool.
15/01/2014, 11h38
plxpy

ps : attention - je parle pour un environnement unix-like. C'est peut-être vrai sous Windows mais je ne connais pas bien

Citation:

Envoyé par roipoussiere

J'imagine donc que parmi les solutions que tu as proposés, le 3ème extrait de code (temps du plus long (+ intendance)) est la seule qui permet de paralléliser les processus sans quitter le script pricipal ?

oui

Citation:

Envoyé par roipoussiere

Par contre je n'ai pas compris un truc : on lance tous les process en les chargeant dans une liste, puis on attends la fin de chaque processus.

C'est ce qui t'est retourné par Popen que tu mémorises dans une liste (donc, pour causer POO, c'est une instance de la classe Popen). Et on mémorise ces instances pour pouvoir, après, appeler leur méthode wait().

Citation:

Envoyé par roipoussiere

J'ai testé avec plus de 10 process et j'ai toujours 1sec. Si on lance plus de process qu'on a de coeurs, ça doit durer plus de 1 sec. non ?

Non. Enfin, pas forcément. Avec les vieilles machines monoprocesseur monocoeur, on faisait déjà du multi-taches/processing (unix, 1970) !
Avec un "sleep 1", je ne suis même pas sur qu'on utilise plusieurs coeurs. De toute façon, c'est l'OS qui décide in fine, pas toi.

Citation:

Envoyé par roipoussiere

C'est tout à fait ça. Pool semble intéressant, mais ce qui m'a fait passer à Popen c'est la possibilité de rediriger la sortie standard et d'erreur. Je n'ai pas trouvé comment faire ça dans la doc de Pool.

Je réfléchis à un exemple simple et parlant. Je reposterai.
15/01/2014, 11h46
wiztricks

Salut,

Si les process se contentent de faire sleep(1s), ils consomment très peu de CPU et vous pouvez en lancer bien plus que le nombre de cœurs: tous se termineront en même temps.

Popen est une fonction Python qui permet de créer des process systèmes. Programmer en Python n'est pas un JOKER permettant de faire l'impasse sur un minimum de savoirs cote administration système, un OS et ses ressources, ...
C'est juste un outil qui vous donne plus facilement les clés pour y accéder avec les bonnes armes et bagages.

- W
15/01/2014, 13h40
roipoussiere

Citation:

ps : attention - je parle pour un environnement unix-like. C'est peut-être vrai sous Windows mais je ne connais pas bien

Je suis sous linux. Mais comme je compte portabiliser le script ça ne change pas grand chose.

Citation:

Non. Enfin, pas forcément. Avec les vieilles machines monoprocesseur monocoeur, on faisait déjà du multi-taches/processing (unix, 1970) !
Avec un "sleep 1", je ne suis même pas sur qu'on utilise plusieurs coeurs. De toute façon, c'est l'OS qui décide in fine, pas toi.

Citation:

Si les process se contentent de faire sleep(1s), ils consomment très peu de CPU et vous pouvez en lancer bien plus que le nombre de cœurs: tous se termineront en même temps.

Je viens d'essayer le code n°3 avec le lancement du vrai programme et pas un simple sleep.
Les process sont en effet tous parallélisés d'un coup, ce qui pose problème puisque j'en lance plus de 100 ; Au bout d'un certain temps ça sature la ram et le swap, du coup l'os ne réponds plus très bien.
Mais l'avantage est qu'on peut faire un ctrl-C :)

http://img4.hostingpics.net/pics/418379cpu.png

J'attends l'exemple avec Pool ;)
15/01/2014, 13h50
plxpy

Coup de coeur !

même sans avoir lu en détail ton dernier post, malgré les réserves de wiztricks, que je partage, quel plaisir de répondre (plus tard) à quelqu'un qui s'investit !!!

Ca nous changera des "et pourquoi quand je fais la somme des indexe de la colonne 3 je ne retrouve pas le déterminant de la matrice transposée si je prends que les indexe 1" qui polluent ce forum depuis fin 2013 !

J'ai testé ceci :
Code:

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 process = list() #[...] # à chaque lancement de process : for i, p in enumerate(process): if p.poll() <= 0: del process[i] print "Un process est terminé. Nettoyage de la liste." p = subprocess.Popen(shlex.split(cmd), stdout = f_out, stderr = f_err) process.append(p) if len(process) >= 8: p.wait() #On attend qu'une place se libère dans la liste avant de lancer un nouveau processus print "La liste contient", len(process), "processus" # toujours à 8 à part au début.
Ça a l'air de fonctionner. Qu'en pensez-vous ? Le soucis est qu'on doit renseigner le nombre de process à paralléliser manuellement, mais bon c'est pas très grave.

EDIT : Merci pour ton dernier message plxpy :) bah, il n'y a plus aucun plaisir à coder si dès qu'on a un problème on attends la solution sur un forum :p
Tu seras sans-doute d'autant plus ravi en remarquant que j'ai trouvé une solution, même si elle n'est peut-être pas encore optimale.

Citation:

"et pourquoi quand je fais la somme des indexe de la colonne 3 je ne retrouve pas le déterminant de la matrice transposée si je prends que les indexe 1"

C'est vrai ça, pourquoi ? :mouarf:

Voici l'exemple avec Pool. A toi de l'adapter à ton cas de figure ou de t'en inspirer, notamment en fonction de la communication entre le "lanceur" et tes processes "openscad".

Pour ton code du post précédent, je vois l'idée (tu tentes de maintenir "à la main", 8 processes en parallèle) mais on ne peut pas en dire plus car tu ne donnes pas les infos nécessaires.

Sinon, pour la parallélisation, le nombre de coeurs (multiprocessing.cpu_count()), c'est une chose, mais il faut aussi faire attention :

à la mémoire qu'utilise chaque process lancé,
aux accés disque, nombreux ? pas nombreux ? volume ? (ça peut bouchonner pour aller écrire sur le même disque...)

J'ai en tête des traitements (d'images satellite, taille importante) où le point de blocage est la mémoire (même à 8Go), et pas les 12 coeurs.

fichier executable.py (j'ai pris soin de faire un chmod +x pour le rendre exécutable)

Code:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
#!/usr/bin/env python
# -*- coding:utf-8 -*-
 
import os
import sys
 
from time import sleep, time
 
noiter = int(sys.argv[1])               # no iteration
quand  = time() - float(sys.argv[2])    # t0 fourni par l'appelant
pid = os.getpid()
 
# traces sur stdout et stderr (bufferisation sur stdout donc flush)
print >> sys.stderr, "(%.2f) démarrage process %d, iter %d (stderr)" % (quand, pid, noiter)
print >> sys.stdout, "(%.2f) démarrage process %d, iter %d (stdout)" % (quand, pid, noiter)
sys.stdout.flush()
 
sleep(2)
 
# pour vérifier que l'exécution a bien eu lieu
open('temoin_%d' % pid, 'w').close()

fichier lanceur.py

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
#!/usr/bin/env python
# -*- coding:utf-8 -*-
 
import os
import sys
from subprocess import call
from multiprocessing import Pool
from time import time
 
 
def lanceur(noiter, t0):
    call('./executable.py %d %f' % (noiter, t0), shell=True)
 
 
if __name__ == '__main__':
 
    nbtasks = int(sys.argv[1])  # nombre d'executions
    workers = int(sys.argv[2])  # nombre de processes en parallèle
    t0 = time()
 
    print >> sys.stderr, "%d taches sur %d files" % (nbtasks, workers)
 
    pool = Pool(processes=workers)
 
    for i in range(nbtasks): 
        _ = pool.apply_async(lanceur, (i,t0))
 
    pool.close()
    pool.join()

une exécution

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
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
plx@sony:~/Bureau/repertoire$ ls
executable.py  lanceur.py
plx@sony:~/Bureau/repertoire$ ./lanceur.py 20 3
20 taches sur 3 files
(0.06) démarrage process 5257, iter 1 (stderr)
(0.06) démarrage process 5257, iter 1 (stdout)
(0.07) démarrage process 5258, iter 2 (stderr)
(0.07) démarrage process 5258, iter 2 (stdout)
(0.08) démarrage process 5259, iter 0 (stderr)
(0.08) démarrage process 5259, iter 0 (stdout)
(2.11) démarrage process 5261, iter 3 (stderr)
(2.11) démarrage process 5261, iter 3 (stdout)
(2.13) démarrage process 5265, iter 5 (stderr)
(2.13) démarrage process 5265, iter 5 (stdout)
(2.13) démarrage process 5264, iter 4 (stderr)
(2.13) démarrage process 5264, iter 4 (stdout)
(4.17) démarrage process 5267, iter 6 (stderr)
(4.17) démarrage process 5267, iter 6 (stdout)
(4.18) démarrage process 5271, iter 8 (stderr)
(4.18) démarrage process 5271, iter 8 (stdout)
(4.20) démarrage process 5270, iter 7 (stderr)
(4.20) démarrage process 5270, iter 7 (stdout)
(6.23) démarrage process 5275, iter 10 (stderr)
(6.23) démarrage process 5275, iter 10 (stdout)
(6.23) démarrage process 5274, iter 9 (stderr)
(6.23) démarrage process 5274, iter 9 (stdout)
(6.26) démarrage process 5277, iter 11 (stderr)
(6.26) démarrage process 5277, iter 11 (stdout)
(8.28) démarrage process 5279, iter 12 (stderr)
(8.28) démarrage process 5279, iter 12 (stdout)
(8.28) démarrage process 5281, iter 13 (stderr)
(8.28) démarrage process 5281, iter 13 (stdout)
(8.31) démarrage process 5283, iter 14 (stderr)
(8.31) démarrage process 5283, iter 14 (stdout)
(10.32) démarrage process 5287, iter 16 (stderr)
(10.32) démarrage process 5287, iter 16 (stdout)
(10.32) démarrage process 5285, iter 15 (stderr)
(10.32) démarrage process 5285, iter 15 (stdout)
(10.36) démarrage process 5289, iter 17 (stderr)
(10.36) démarrage process 5289, iter 17 (stdout)
(12.37) démarrage process 5292, iter 18 (stderr)
(12.37) démarrage process 5292, iter 18 (stdout)
(12.37) démarrage process 5293, iter 19 (stderr)
(12.37) démarrage process 5293, iter 19 (stdout)
plx@sony:~/Bureau/repertoire$ 
plx@sony:~/Bureau/repertoire$ ls
executable.py  temoin_5259  temoin_5267  temoin_5275  temoin_5283  temoin_5292
lanceur.py     temoin_5261  temoin_5270  temoin_5277  temoin_5285  temoin_5293
temoin_5257    temoin_5264  temoin_5271  temoin_5279  temoin_5287
temoin_5258    temoin_5265  temoin_5274  temoin_5281  temoin_5289
plx@sony:~/Bureau/repertoire$

Merci pour le code plxpy. J'ai testé et ça marche.

Dans mon cas, à priori, ni la ram ni les accès disques sont un poblème :
- En parallélisant 8 processus, j'ai une augmentation de l'usage de la ram de seulement 20% sur 4Go, donc à la louche 100Mo/processus.
- Open-scad c'est de la CAO 3D donc surtout du calcul graphique et peu d'accès disque.

À moins d'avoir fait un truc qui n'est pas du tout conseillé, ma solution me convient.
Il y a un avantage à utiliser la tienne ?

Je poste détail du code pour info :
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 def _start_processes(self, scad_path, nb_files, f_table): t_init = time.time() self.nb_created = 0 for line in f_table: options = ... output_file = ... self.openscad(scad_path, options, output_file) self._end_of_process(nb_files, t_init) while self.process: # attend d'avoir fini tous les process self._end_of_process(nb_files, t_init) def _end_of_process(self, nb_files, t_init): spent = time.strftime("%Hh%Mm%Ss", time.gmtime(time.time()-t_init)) for i, p in enumerate(self.process): if p.poll() == 0: del self.process[i] self.nb_created += 1 print spent + ": Created file " + str(self.nb_created) + "/" + str(nb_files) + "." def openscad(self, scad_file_path, options, output_file): cmd = self.openscad_path + " " + scad_file_path + " -o " + output_file + " " + options out = None if self.verbose else open(os.devnull, 'w') p = subprocess.Popen(shlex.split(cmd), stdout = out) self.process.append(p) if len(self.process) >= self.nb_job_slots: p.wait()

Il y a deux ou trois choses qui me chagrinent dans ton code.

1 - Tu fais tout ça en passant par une classe. OK, pas de problème.
Pourquoi implémenter une méthode openscad ? Pourquoi spécialiser ta classe au lancement de la seule commande openscad ?

Quand tu auras arrêté, validé ta gestion de N processes en parallèle, il y a de grandes chances que tu réutilises tout ça pour lancer d'autres choses. Tu écriras une nouvelle methode "mon_autre_commande" à chaque fois ? En modifiant, à chaque fois, ta classe ?

C'est à l'appelant (qui sait ce qu'il veut faire et lancer) de fournir les commandes à lancer à ta classe générique. Donc en gros, une méthode "add_command(self, command)" et une autre "go(self)" qui démarre tout ça

2 - lié à ça, c'est une très mauvaise idée de lier ta gestion de N processes en parallèle à ta fonction openscad proprement-dite (lignes 26 à 29). Et si, rongé par le remord, tu décides de changer ta gestion et d'utiliser Pool : tu re-tritures la méthode openscad ?

Non. openscad lance openscad, point barre. Elle retourne p, et c'est tout.

3 - ligne 29 : quand tu atteins le nombre maximum de processes en parallèle que tu t'es fixé, tu attends que ... le dernier lancé ait fini ... dans certains cas de figure, ça peut être équivalent à un os.system(). Et si le dernier lancé est le dernier à se terminer ? pendant tout ce temps, tu as 8-1 coeurs "qui branlent rien" ! C'est ballot. Globalement faut être à l'écoute de tous les processes qui ont été lancés, pas uniquement le dernier.

4 - (du détail) tu ouvres (peux ouvrir) os.devnull à chaque lancement. Je n'ai pas plus gratté que ça, mais ça peut peut-être poser problème (nombre max de fichiers ouverts, même si c'est le même). Pourquoi ne pas l'ouvrir une bonne fois pour toutes et t'en resservir ?

Voici, pour illustrer, un script écrit il y a 4 ans (quand on était sous python 2.5 et que multiprocessing n'existait pas), que j'ai un peu retouché (en 4 ans, on fait des progrès ...) et qui est utilisé quotidiennement en production depuis. Il se présente sous la forme d'un exécutable car cela permet de l'appeler aussi depuis un exe en C, shell, awk, ...

Il lit les commandes à lancer sur son entrée standard et, quand il a fini, commence à gérer, "à la main", n processes en parallèle. Il attend, en argument, le nombre de processes en parallèle à maintenir et un temps de temporisation (pour ne pas boucler comme un malade à faire des poll).
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 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 #!/usr/bin/env python # -*- coding:utf-8 -*- import sys from subprocess import Popen from time import sleep # nombre de processe en parallèle et durée de temporisation workers = int(sys.argv[1]) delay = float(sys.argv[2]) # liste de mes "files d'attente" slots = [ None ] * workers # sondage des processes en cours - maj de ma liste def update_slots(): for s, slot in enumerate(slots): if slot is not None: if slot.poll() is not None: slots[s] = None # récupération d'un emplacement (avec attente si nécessaire) def get_slot(): while True: update_slots() try: free = slots.index(None) return free except: pass sleep(delay) # lecture des commandes à lancer sur l'entrée standard commands = sys.stdin.readlines() commands = map(lambda l: l.rstrip(), commands) # lancement de toutes les commandes while commands: # je récupère un "emplacement" libre free_slot = get_slot() # je lance la commande et "l'insère" à cet emplacement slots[free_slot] = Popen(commands.pop(0), shell=True) # toutes les commandes ont été lancées # on attend que toutes soient terminés while slots.count(None) != workers: update_slots() sleep(delay) print "That's all folks !"
A l'exécution (toujours avec sleep qui a quand même l'avantage d'être très prévisible sur la durée du traitement)
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 Python 2.7.6 (v2.7.6:3a1db0d2747e, Nov 10 2013, 00:42:54) [GCC 4.2.1 (Apple Inc. build 5666) (dot 3)] on darwin Type "help", "copyright", "credits" or "license" for more information. >>> from subprocess import Popen, PIPE >>> >>> commands = [ "sleep 2; echo reveil %d" % i for i in range(10) ] >>> # 3 files, temporisation de 0.1 seconde ... p = Popen('./lanceur.py 3 0.1', shell=True, stdin=PIPE) >>> >>> for cmd in commands: ... print >> p.stdin, cmd ... >>> # je ferme le "robinet" ... p.stdin.close() >>> reveil 0 reveil 1 reveil 2 reveil 3 reveil 4 reveil 5 reveil 6 reveil 7 reveil 8 reveil 9 That's all folks ! >>>
Faut le lancer pour voir la "dynamique" du machin.