Multiprocessin, thread repart depuis le main

Version imprimable

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:

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:

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 ?

08/02/2013, 12h42
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 ...

Autant pour moi, voici le code complet :

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
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 ?

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

Code:

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:

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

j'ai une boucle infinie d'appels..

Code:

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