Problème de multiprocessing
Bonjour,
Dans le cadre d'un cours, je travaille sur l'implémentation d'algorithmes de parcours de graphes. J'ai implémenté Dijkstra de manière séquentielle et souhaiterai passer en multiprocessing pour optimiser mon temps de calcul.
Je dois exécuter l'algo en partant de chaque noeud de l'arbre. En séquentiel, je le fais grâce à une simple boucle for. Dans mon essai sur le multi-processing, j'ai défini une fonction qui se charge uniquement d'appliquer l'algo à un noeud donné.
Dans mon idée, je voulais ainsi appliquer en parallèle l'algo en partant de plusieurs noeuds. Je le fais actuellement en utilisant Pool du module multiprocessing et la fonction starmap (qui me permet de passer plusieurs arguments à ma fonction de base).
Toutefois, le code ne marche pas et je ne comprends pas du tout pourquoi... (il est identique au code séquentiel contenu dans la boucle for). La fonction est censée écrire dans différents fichiers mais ces fichiers (créés donc pool.starmap fonctionne !) restent vides et les processus restent en attente, bloquant les suivants.
Quelqu'un aurait-il une idée ?
Voici le code correspondant (la fonction main est appelée depuis un fichier main.py ou je ne fais que traiter un fichier .txt de données pour le formater comme il le faut).
Merci beaucoup d'avance !
EDIT : En recopiant mon code, j'ai repéré qu'une ligne n'était pas a sa bonne place. Le tab_dist[node]=0 ne doit pas être dans le if do_print: Dire que ca fait 3 heures que j'essaye de comprendre ce qui bloque...
Tout marche maintenant. Désolé du dérangement.
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
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
|
import multiprocessing as mp
def subprocess_dijkstra(do_print, node, tab_contenu, tab_distances):
tab_dist = dict(tab_distances)
visited_nodes = list()
resultat = ""
filename = "dijkstra"+str(node)+".txt"
if do_print:
dt = open(filename, 'w')
tab_dist[node] = 0
"""Ligne de résultat initiale"""
for valeur in tab_dist.values():
resultat += str(valeur)
resultat += " "
resultat += "\n"
dt.write(resultat)
while len(visited_nodes) != len(tab_contenu):
dmin = -1
""" On se place sur le noeud non visité qui a la distance minimale de notre départ """
for cle, valeur in tab_dist.items():
if cle not in visited_nodes:
if dmin ==-1 or valeur<dmin:
dmin = valeur
node = cle
""" On vérifie que le noeud n'a pas déjà été visité """
if (node not in visited_nodes):
""" On regarde les fils de ce noeud et la longueur des arcs"""
for cle,valeur in tab_contenu[node].items():
tab_dist[cle] = min(tab_dist[cle], tab_dist[node]+valeur)
visited_nodes.append(node)
if do_print:
resultat = ""
""" Ligne de résultat """
for valeur in tab_dist.values():
resultat += str(valeur)
resultat += " "
resultat += "\n"
dt.write(resultat)
if do_print:
dt.close()
def main(do_print,donnees):
tab_contenu = donnees[1]
nb_nodes = int(donnees[0])
tab_distances = {x: float('inf') for x in range(nb_nodes)}
args=[(do_print, x, tab_contenu, tab_distances) for x in range(nb_nodes)]
nb_cpu = mp.cpu_count()
pool = mp.Pool(processes = nb_cpu -1)
pool.starmap(subprocess_dijkstra, args)
pool.close()
pool.join() |
