Comment calculer le temps d'exécution avec multiprocessing ?

Version imprimable

18/07/2011, 18h25
PauseKawa

Comment calculer le temps d'exécution avec multiprocessing ?
Bonjour,

J'ai louper un passage il me semble : Comment calculer le temps d’exécution avec multiprocessing ?

Avec time le code
Code:

1 2 3 4 t1 = time.clock() testw(workers, repcible, nb) t2 = time.clock() print(t2-t1)
donne un résultat complètement faux.

Merci d'avance
18/07/2011, 20h05
mont29

Euh, un peu plus de code, ce serait possible*?

Parce que si testw() est asynchrone (c-à-d lance les différents processus et retourne immédiatement), c’est normal que le temps soit fantaisistement court. Par contre, si elle attend gentiment la fin de tous les processus de calcul pour retourner, je ne vois vraiment pas pourquoi les résultats seraient faux… :calim2:

Bonsoir,

Citation:

Envoyé par mont29

Parce que si testw() est asynchrone (c-à-d lance les différents processus et retourne immédiatement), c’est normal que le temps soit fantaisistement court. Par contre, si elle attend gentiment la fin de tous les processus de calcul pour retourner, je ne vois vraiment pas pourquoi les résultats seraient faux… :calim2:

C'est bien le cas.
Voici le code qui me pose problème (Et vous remarquerez que je ne connais pas multiprocessing)

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
def testmp(num, nb, n1, n2, rc):
    op = open
    st = str
    while num.value < nb:
        out = op(rc + "couple_" + n1 + "_" + st(num.value) + '_' + n2 + "_" +
                 st(num.value) + ".fasta", 'w')
        src1 = op(n1 + '_' + st(num.value), 'r')
        src2 = op(n2 + '_' + st(num.value), 'r')
        out.write(src1.read() + src2.read())
        src1.close()
        src2.close()
        out.close()
        num.value += 1
 
def inittest(nb, tests):
    print('Grand ménage...')
    print('Création des répertoires cibles')
    for name in tests:
        if os.path.isdir(name):
            shutil.rmtree(name)
        os.mkdir(name)
    print('Suppression des fichiers')
    for item in os.listdir(os.getcwd()):
        if item.startswith('src'):
            os.remove(item)
    print('Création de 2 x ' + str(nb) + ' fichiers')
    letters = 'acgt'
    for index in range(0, nb):
        src1 = ''
        src2 = ''
        for i in range(1, 88):
            src1 += letters[randint(0, 3)]
            src2 += letters[randint(0, 3)]
        with open('src1_' + str(index), 'w') as f1:
            f1.write(src1 + '\n')
        with open('src2_' + str(index), 'w') as f2:
            f2.write(src2 + '\n')
 
if __name__ == "__main__":
    nb = 10000
    tests = ['testopen', 'testmp']
    inittest(nb, tests)
    print('Début des tests')
    print('testopen:')
    repcible = os.path.join(os.getcwd(), 'testopen') + os.sep
    t1 = time.clock()
    testopen(nb, 'src1', 'src2', repcible)
    t2 = time.clock()
    print(t2-t1)
    print('testmp:')
    repcible = os.path.join(os.getcwd(), 'testmp') + os.sep
    workers = 2
    def testw(workers, repcible, nb):
        num = Value('i', 0)
        for w in range(workers):
            p = Process(target=testmp, args=(num, nb, 'src1', 'src2', repcible))
            p.start()
            p.join()
    t1 = time.clock()
    testw(workers, repcible, nb)
    t2 = time.clock()
    print(t2-t1)

Comment attendre que tout soit fini ?

Merci d'avance

18/07/2011, 23h11
mont29

Ben, y a pas encore assez de code pour que je puisse le tester ( ;) ), mais le fait d’utiliser join() casse tout parallélisme, normalement, puisque ça bloque jusqu’à ce que le processus worker retourne (donc ait fini)… Du coup, sauf erreur de ma part, ce code est totalement linéaire (worker 2 est exécuté une fois que worker 1 a fini son boulot)*!

En quoi le résultat donné par time est-il faux, au fait*? Trop long, ou trop court*?

Bonjour,

Citation:

Envoyé par mont29

Ben, y a pas encore assez de code pour que je puisse le tester ( ;) )...
En quoi le résultat donné par time est-il faux, au fait*? Trop long, ou trop court*?

Voici pour le code de test

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
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
#! /usr/bin/env/ python
# -*- coding: utf-8 -*-
# 
#
from random import randint
import os
import time
import shutil
from multiprocessing import Process, Value, Manager
import sys
 
def testopen(nb, n1, n2, rc):
    op = open
    st = str
    for i in range(0, nb):
        out = op(rc + "couple_" + n1 + "_" + st(i) + '_' + n2 + "_" + st(i) +
                 ".fasta", 'w')
        src1 = op(n1 + '_' + st(i), 'r')
        src2 = op(n2 + '_' + st(i), 'r')
        out.write(src1.read() + src2.read())
        src1.close()
        src2.close()
        out.close()
 
def testmp(lproxy, nb, n1, n2, rc):
    op = open
    st = str
    while True:
        try:
            val = lproxy.pop()
        except IndexError:
            break
        out = op(rc + "couple_" + n1 + "_" + val + '_' + n2 + "_" + val +
                 ".fasta", 'w')
        src1 = op(n1 + '_' + val, 'r')
        src2 = op(n2 + '_' + val, 'r')
        out.write(src1.read() + src2.read())
        src1.close()
        src2.close()
        out.close()
 
def testmpmulti(lproxy, endtime, tclock, nb, n1, n2, rc):
    op = open
    st = str
    while True:
        try:
            val = lproxy.pop()
        except IndexError:
            endtime.append(tclock())
            break
        out = op(rc + "couple_" + n1 + "_" + val + '_' + n2 + "_" + val +
                 ".fasta", 'w')
        src1 = op(n1 + '_' + val, 'r')
        src2 = op(n2 + '_' + val, 'r')
        out.write(src1.read() + src2.read())
        src1.close()
        src2.close()
        out.close()
 
def inittest(nb, tests):
    print('Grand ménage...')
    print('Création des répertoires cibles')
    for name in tests:
        if os.path.isdir(name):
            shutil.rmtree(name)
        os.mkdir(name)
    print('Suppression des fichiers')
    for item in os.listdir(os.getcwd()):
        if item.startswith('src'):
            os.remove(item)
    print('Création de 2 x ' + str(nb) + ' fichiers')
    letters = 'acgt'
    for index in range(0, nb):
        src1 = ''
        src2 = ''
        for i in range(1, 88):
            src1 += letters[randint(0, 3)]
            src2 += letters[randint(0, 3)]
        with open('src1_' + str(index), 'w') as f1:
            f1.write(src1 + '\n')
        with open('src2_' + str(index), 'w') as f2:
            f2.write(src2 + '\n')
 
if __name__ == "__main__":
    nb = 10000
    tests = ['testopen', 'testmp', 'testmpmulti']
    inittest(nb, tests)
    print('Début des tests')
    print('testopen:')
    t1 = time.clock()
    repcible = os.path.join(os.getcwd(), 'testopen') + os.sep
    testopen(nb, 'src1', 'src2', repcible)
    t2 = time.clock()
    print(t2-t1)
    print(len(os.listdir(repcible)))
    print('testmp:')
    t1 = time.clock()
    lworkers = []
    workers = 4
    repcible = os.path.join(os.getcwd(), 'testmp') + os.sep
    manager = Manager()
    lproxy = manager.list()
    for i in range(nb):
        lproxy.append(str(i))
    for w in range(workers):
        lworkers.append(Process(target=testmp, args=(lproxy, nb, 'src1', 'src2',
                        repcible)))
    for w in lworkers:
        w.start()
    while len(lproxy) > 0:
        pass
    t2 = time.clock()
    print(t2-t1)
    time.sleep(2)
    print(len(os.listdir(repcible)))
    print('testmpmulti:')
    repcible = os.path.join(os.getcwd(), 'testmpmulti') + os.sep
    manager = Manager()
    endtime = manager.list()
    lproxy = manager.list()
    tclock = time.clock
    for i in range(nb):
        lproxy.append(str(i))
    p1 = Process(target=testmpmulti, args=(lproxy, endtime, tclock, nb, 'src1',
                 'src2', repcible))
    p2 = Process(target=testmpmulti, args=(lproxy, endtime, tclock, nb, 'src1',
                 'src2', repcible))
    p3 = Process(target=testmpmulti, args=(lproxy, endtime, tclock, nb, 'src1',
                 'src2', repcible))
    p4 = Process(target=testmpmulti, args=(lproxy, endtime, tclock, nb, 'src1',
                 'src2', repcible))
    p1.start()
    p2.start()
    p3.start()
    p4.start()
    while any((p1.is_alive(), p2.is_alive(), p3.is_alive(), p4.is_alive())):
        pass
    print(max(endtime)) # A revoir
    print(len(os.listdir(repcible)))

Le souci c'est que

Code:

1
2
3
4
5
    t1 = time.clock()
    p = (Process(...))
    ...
    t2 = time.clock()
    print(t2-t1)

Mesure le temps du bloc entre t1 et t2, pas le temps des process.
Ce que je recherche c'est le calcul du temps des process sans passer par un while (là je calcule le temps du while en fait) et plus 'propre' que dans testmpmulti.

@+

19/07/2011, 16h59
mont29
Eh bien, il suffit d’encadrer le contenu de tes target avec le motif
Code:

1 2 3 4 t1 = time.clock() ... t2 = time.clock() endtime.append(t2-t1)
… Non*?

Ou alors, tu utilises os.times(), qui renvoie un tupple dont les deux premiers membres sont les temps utilisateur et noyau du processus courant, en mettant par exemple à la fin de ta fonction target*:
Code:

1 2 tms = os.times() endtime.append(tms[0] + tms[1])

Bonsoir,

Citation:

Envoyé par mont29

Ou alors, tu utilises os.times()

C'est bien ce que je me suis cet aprem : A force d'utiliser le tournevis plat on en oubli le cruciforme.

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
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
#! /usr/bin/env/ python
# -*- coding: utf-8 -*-
# 
#
from random import randint
import os
import time
import shutil
from multiprocessing import Process, Value, Manager
import sys
 
def testopen(nb, n1, n2, rc):
    op = open
    st = str
    for i in range(0, nb):
        out = op(rc + "couple_" + n1 + "_" + st(i) + '_' + n2 + "_" + st(i) +
                 ".fasta", 'w')
        src1 = op(n1 + '_' + st(i), 'r')
        src2 = op(n2 + '_' + st(i), 'r')
        out.write(src1.read() + src2.read())
        src1.close()
        src2.close()
        out.close()
 
def testmp(lproxy, nb, n1, n2, rc):
    op = open
    st = str
    while True:
        try:
            val = lproxy.pop()
        except IndexError:
            break
        out = op(rc + "couple_" + n1 + "_" + val + '_' + n2 + "_" + val +
                 ".fasta", 'w')
        src1 = op(n1 + '_' + val, 'r')
        src2 = op(n2 + '_' + val, 'r')
        out.write(src1.read() + src2.read())
        src1.close()
        src2.close()
        out.close()
 
def inittest(nb, tests):
    print('Grand ménage...')
    print('Création des répertoires cibles')
    for name in tests:
        if os.path.isdir(name):
            shutil.rmtree(name)
        os.mkdir(name)
    print('Suppression des fichiers')
    for item in os.listdir(os.getcwd()):
        if item.startswith('src'):
            os.remove(item)
    print('Création de 2 x ' + str(nb) + ' fichiers')
    letters = 'acgt'
    for index in range(0, nb):
        src1 = ''
        src2 = ''
        for i in range(1, 88):
            src1 += letters[randint(0, 3)]
            src2 += letters[randint(0, 3)]
        with open('src1_' + str(index), 'w') as f1:
            f1.write(src1 + '\n')
        with open('src2_' + str(index), 'w') as f2:
            f2.write(src2 + '\n')
 
if __name__ == "__main__":
    nb = 100000
    tests = ['testopen', 'testmp', 'testmpmulti']
    inittest(nb, tests)
    print('Début des tests')
    print('testopen:')
    t1 = time.clock()
    repcible = os.path.join(os.getcwd(), 'testopen') + os.sep
    testopen(nb, 'src1', 'src2', repcible)
    t2 = time.clock()
    print(t2-t1)
    print(len(os.listdir(repcible)))
    print('testmp:')
    t1 = os.times()[0]
    lworkers = []
    workers = 4
    repcible = os.path.join(os.getcwd(), 'testmp') + os.sep
    manager = Manager()
    lproxy = manager.list()
    for i in range(nb):
        lproxy.append(str(i))
    for w in range(workers):
        lworkers.append(Process(target=testmp, args=(lproxy, nb, 'src1', 'src2',
                        repcible)))
    for w in lworkers:
        w.start()
    while len(lproxy) > 0:
        pass
    t2 = os.times()[0]
    print(t2-t1)
    time.sleep(2)
    print(len(os.listdir(repcible)))
    print('testmpmulti:')
    t1 = os.times()[0]
    repcible = os.path.join(os.getcwd(), 'testmpmulti') + os.sep
    manager = Manager()
    lproxy = manager.list()
    for i in range(nb):
        lproxy.append(str(i))
    p1 = Process(target=testmp, args=(lproxy, nb, 'src1', 'src2', repcible))
    p2 = Process(target=testmp, args=(lproxy, nb, 'src1', 'src2', repcible))
    p3 = Process(target=testmp, args=(lproxy, nb, 'src1', 'src2', repcible))
    p4 = Process(target=testmp, args=(lproxy, nb, 'src1', 'src2', repcible))
    p1.start()
    p2.start()
    p3.start()
    p4.start()
    while any((p1.is_alive(), p2.is_alive(), p3.is_alive(), p4.is_alive())):
        pass
    t2 = os.times()[0]
    print(t2-t1)
    print(len(os.listdir(repcible)))

Par contre il me semble que je n'ai pas bien compris dont les deux premiers membres sont les temps utilisateur et noyau du processus courant

Code:

1
2
    tms = os.times()
    endtime.append(tms[0] + tms[1])

Pourquoi temps utilisateur + noyau du processus ?

Merci d'avance

Eh bien, un processus passe une partie de son temps d’exécution en espace utilisateur (c’est là qu’il effectue la majeure partie –*si pas la totalité*– de ses calculs, là que s’exécute son propre code).

Cependant, l’espace utilisateur a des possibilités et permissions restreintes (il n’a notamment aucun accès au matériel), donc quand le processus veut faire certaine choses qu’il n’a pas le droit de faire en espace utilisateur (comme accéder au disque dur, par exemple), il va devoir passer en espace noyau (ou système).

Évidemment, ce n’est pas son propre code qui s’exécute alors, mais un code intégré au noyau, appelé par le biais d’une fonction système. Et tout aussi évidemment, tout ça est caché au programmeur lambda par des bibliothèques comme la libc pour C (sur laquelle s’appuie python).

Donc, quand on fait un open() ou un read() en python, une partie du code sous-jacent sera exécuté en espace noyau, avec les privilèges associés… Il y a donc deux “durées processeur” pour un processus, le temps qu’il passe en espace utilisateur, et celui qu’il passe en espace noyau (ou plutôt, que le noyau passe à son service).

En tout cas, c’est valable pour Linux, mais j’imagine que même windaube, qui n’aime pourtant rien tant que de ne rien faire comme tout le monde, a un comprtement similaire.

Bref, pour en revenir à os.times(), au vu de ton code, je crois que tu n’as pas compris ce que c’est. ;) Au contraire de time.time(), qui renvoie l’instant présent, en quelque sorte, os.times() renvoie des durées, c’est-à-dire le temps qu’a passé, jusque ici, le processus qui appelle cette fonction, que ce soit en espace noyau ou en espace utilisateur.

Autrement dit*:
* Il ne faut pas appeler os.times() dans le processus principal, mais à la fin des fonctions qui servent de target à ton multiprocessing. De cette façon, tu auras le temps utilisé par les sous-processus, juste avant qu’ils ne retournent.
* os.times() renvoyant une durée, pas besoin de l’appeler deux fois (une au départ, une à la fin), la valeur “brute” est la bonne*!
* Enfin, la somme des deux durées (espaces utilisateur et noyau) ne donne pas la durée réelle qu’à pris le processus à s’exécuter, mais uniquement le temps processeur qu’il a consommé*!

Voici à quoi devrait ressembler ton code, je pense*:
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 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 #! /usr/bin/env/ python # -*- coding: utf-8 -*- # # from random import randint import os import time import shutil from multiprocessing import Process, Value, Manager import sys def testopen(nb, n1, n2, rc): op = open st = str for i in range(0, nb): out = op(rc + "couple_" + n1 + "_" + st(i) + '_' + n2 + "_" + st(i) + ".fasta", 'w') src1 = op(n1 + '_' + st(i), 'r') src2 = op(n2 + '_' + st(i), 'r') out.write(src1.read() + src2.read()) src1.close() src2.close() out.close() def testmp(lproxy, nb, n1, n2, rc): op = open st = str while True: try: val = lproxy.pop() except IndexError: break out = op(rc + "couple_" + n1 + "_" + val + '_' + n2 + "_" + val + ".fasta", 'w') src1 = op(n1 + '_' + val, 'r') src2 = op(n2 + '_' + val, 'r') out.write(src1.read() + src2.read()) src1.close() src2.close() out.close() def testmpmulti(lproxy, proctime, nb, n1, n2, rc): op = open st = str while True: try: val = lproxy.pop() except IndexError: break out = op(rc + "couple_" + n1 + "_" + val + '_' + n2 + "_" + val + ".fasta", 'w') src1 = op(n1 + '_' + val, 'r') src2 = op(n2 + '_' + val, 'r') out.write(src1.read() + src2.read()) src1.close() src2.close() out.close() tms = os.times() proctime.append(tms[0] + tms[1]) def inittest(nb, tests): print('Grand ménage...') print('Création des répertoires cibles') for name in tests: if os.path.isdir(name): shutil.rmtree(name) os.mkdir(name) print('Suppression des fichiers') for item in os.listdir(os.getcwd()): if item.startswith('src'): os.remove(item) print('Création de 2 x ' + str(nb) + ' fichiers') letters = 'acgt' for index in range(0, nb): src1 = '' src2 = '' for i in range(1, 88): src1 += letters[randint(0, 3)] src2 += letters[randint(0, 3)] with open('src1_' + str(index), 'w') as f1: f1.write(src1 + '\n') with open('src2_' + str(index), 'w') as f2: f2.write(src2 + '\n') if __name__ == "__main__": nb = 10000 tests = ['testopen', 'testmp', 'testmpmulti'] inittest(nb, tests) print('Début des tests') print('testopen:') t1 = time.clock() repcible = os.path.join(os.getcwd(), 'testopen') + os.sep testopen(nb, 'src1', 'src2', repcible) t2 = time.clock() print(t2-t1) print(len(os.listdir(repcible))) print('testmp:') t1 = time.clock() lworkers = [] workers = 4 repcible = os.path.join(os.getcwd(), 'testmp') + os.sep manager = Manager() lproxy = manager.list() for i in range(nb): lproxy.append(str(i)) for w in range(workers): lworkers.append(Process(target=testmp, args=(lproxy, nb, 'src1', 'src2', repcible))) for w in lworkers: w.start() while len(lproxy) > 0: pass t2 = time.clock() print(t2-t1) time.sleep(2) print(len(os.listdir(repcible))) print('testmpmulti:') repcible = os.path.join(os.getcwd(), 'testmpmulti') + os.sep manager = Manager() proctime = manager.list() lproxy = manager.list() for i in range(nb): lproxy.append(str(i)) p1 = Process(target=testmpmulti, args=(lproxy, proctime, nb, 'src1', 'src2', repcible)) p2 = Process(target=testmpmulti, args=(lproxy, proctime, nb, 'src1', 'src2', repcible)) p3 = Process(target=testmpmulti, args=(lproxy, proctime, nb, 'src1', 'src2', repcible)) p4 = Process(target=testmpmulti, args=(lproxy, proctime, nb, 'src1', 'src2', repcible)) p1.start() p2.start() p3.start() p4.start() while any((p1.is_alive(), p2.is_alive(), p3.is_alive(), p4.is_alive())): pass print(proctime) print(len(os.listdir(repcible)))
Pour autant que os.times() soit bien ce que tu recherches…

20/07/2011, 20h23
wiztricks

.join

Salut,

La fonction qui permet au lanceur d'attendre la fin des processus/threads fils serait plutôt .join(timeout).

Fonctionnellement .isalive fait la même chose mais utiliser cette méthode attendre que... avec un while signifie "pooling" et gâchis de temps CPU.

Le .join passe l'appelant en "pause" jusqu'à "timeout"...

C'est préférable dans tous les cas et à fortiori lorsqu'on mesure des performances, isn't it?

Cordialement
- W
20/07/2011, 20h44
PauseKawa

Il me semble aussi.

Pour le moment je suis en vacance donc pas trop de temps pour tester tout cela comme je le souhaite.
Dans l'état j'en suis plus sur un test externe (python testit.py lescript.py avec argv/subprocess.call/time.time par exemple : à voir)

Je reviens dès que plus dispo.

@+
20/07/2011, 20h52
wiztricks

Salut,
Si c'est des tests de créations de fichiers de la discussion de l'autre jour...
Je n'ai pas compris pourquoi tu utilisais l'artillerie "process" avant de passer par la case "threads": la chose ne mange pas assez de CPU pour le mériter.
Note: Concurrent est "better" pour passer de l'un à l'autre.
Bonnes vacances!
- W
20/07/2011, 21h17
PauseKawa

Citation:

Envoyé par wiztricks

Bonnes vacances!

Sans rire, en vacances et encore sur le forum :mouarf:. Bientôt la cure de désintoxication ;).
Je suis passé par la case threads, c'est pour mon info perso.
20/07/2011, 21h33
wiztricks

Citation:

Envoyé par PauseKawa

Sans rire, en vacances est encore sur le forum :mouarf:. Bientôt la cure de désintoxication ;).
Je suis passer par la case threads, c'est pour mon info perso.

Je ne suis pas en vacances, moi - snif.
- W