Discussion :

[newelfik]

Bonjour à tous,

Je m'essaie au calcul parallèle depuis quelques jours et je rencontre un comportement inattendu. J'utilise le package multiprocessing standard sur python 2.7, et je tente d'appeler une méthode en parallèle. Cette méthode devra remplir un dictionnaire de données.

Un code simpliste dans un fichier à part fonctionne très bien:

Code :

Sélectionner tout - Visualiser dans une fenêtre à part

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
 
def push_in_array(self,array,i,val):
        print 'push %s in %s' %(i,val) 
 
        n = 9999
        while n > 0:
            r = sqrt(n)
            n = n-1
 
        array[i] = val
 
    def Run(self) :
 
 
        manager = Manager()
        d = manager.dict()    
        t0 = time()     
 
 
        if self.multi:
            # Multiprocessing Mode
            print "Multiprocessing computation..." 
 
            p = {}      
            for i in range(self.nb):
 
                p[i] = Process(target = self.push_in_array, args=(d,i,5,))
                p[i].start()
 
            for i in range(self.nb):
                p[i].join()
 
            tp = round(time() - t0, 3) 
            print "Time elapsed in multiprocessing computation: %.2f\n" % tp
 
        else:
            # Serial Mode
            print "Serial computation..." 
 
            for i in range(self.nb):
 
                push_in_array(d,i,5)            
 
            tp = round(time() - t0, 3) 
            print "Time elapsed in serial computation: %.2f\n" % tp
 
        print d

Code :

Sélectionner tout - Visualiser dans une fenêtre à part

1
2
3
4
5
6
7
8
9
Multiprocessing computation...
push 0 in 5
push 1 in 5
push 2 in 5
push 4 in 5
push 3 in 5
Time elapsed in multiprocessing computation: 0.89
 
{0: 5, 1: 5, 2: 5, 3: 5, 4: 5}

l'intégration de ce code dans ma librairie entraine par contre un comportement étrange, le premier proc reprend depuis le début du programme à son start(), puis revenu à l'endroit du fork(), le nouveau fils recommence encore, et encore ... et le code n'avance pas finalement.

Avez vous déja rencontrer ce genre de problème ?

[VinsS]

Salut,

Est-ce que tu es sûr de ton code ?

L'indentation laisse à penser que Run() est sous-fonction de push_in_array, dans un cas tu utilises self.push_in_array et dans le deuxième cas push_in_array d'autre part, push_in_array ne retourne rien ...

[newelfik]

Autant pour moi, voici le code complet :

Code :

Sélectionner tout - Visualiser dans une fenêtre à part

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
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
# Main
#=====
 
class RunMultiProc(object) :
 
    def __init__(self, nb, multi):
        self.nb = nb
        self.multi = multi
 
    def push_in_array2(self,array,i,j,val):
        print 'push %s in [%s,%s]' %(val,i,j) 
        array[i,j] = val
 
    def push_in_array(self,array,i,val):
        print 'push %s in %s' %(val,i) 
 
        n = 9999
        while n > 0:
            r = sqrt(n)
            n = n-1
 
        array[i] = val
        for n in range(5):
            self.push_in_array2(array,i,n, n)
 
    def Run(self) :
 
 
        manager = Manager()
        d = manager.dict()    
        t0 = time()     
 
 
        if self.multi:
            # Multiprocessing Mode
            print "Multiprocessing computation..." 
 
            p = {}      
            for i in range(self.nb):
 
                p[i] = Process(target = self.push_in_array, args=(d,i,5,))
                p[i].start()
 
            for i in range(self.nb):
                p[i].join()
 
            tp = round(time() - t0, 3) 
            print "Time elapsed in multiprocessing computation: %.2f\n" % tp
 
        else:
            # Serial Mode
            print "Serial computation..." 
 
            for i in range(self.nb):
 
                push_in_array(d,i,5)            
 
            tp = round(time() - t0, 3) 
            print "Time elapsed in serial computation: %.2f\n" % tp
 
        print d
 
if __name__ == "__main__":
 
    multi = False
 
    if len(sys.argv) > 1:
        if sys.argv[1] == 'Multi' or sys.argv[1] == 'multi':
            multi = True
 
    # Init environment
    nbProc = 5
 
    run = RunMultiProc( nbProc, multi )
    run.Run()
 
    # Compare
    #print "Speed-up: %sx \n" % round(ts/tp, 2) 
 
    # Exit
    sys.exit(0)
 
# End
#====

Est il nécessaire d'avoir un return pour ce genre de méthode ?

[newelfik]

De ce fait, lorsque j'essaie de translater ce code à ma lib :

Code :

Sélectionner tout - Visualiser dans une fenêtre à part

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
 
        # Multiprocessing
        print "Multiprocessing computation..." 
        manager = Manager()
        print '1'
        d = manager.dict()
        print '2'
        proc = {}
        print '3'
 
        for (i, process_s) in enumerate(self.processList):
            proc[i] = Process(target = self.parallelCall, args=(array,i,))
            proc[i].start()
 
        for i in range(nbProcess):
            proc[i].join()

Avec la méthode dans la meme classe :

Code :

Sélectionner tout - Visualiser dans une fenêtre à part

1
2
3
 
    def parallelCall(self, array, i):
        print 'TRACE : %s (%s,%s)' %('parallelCall',i, 'process')

j'ai une boucle infinie d'appels..

Code :

Sélectionner tout - Visualiser dans une fenêtre à part

1
2
3
4
5
6
7
8
9
10
11
12
13
 
Multiprocessing computation...
1
2
3
Multiprocessing computation...
1
2
3
Multiprocessing computation...
1
2
3