Threads & Lock

Version imprimable

Voir 40 message(s) de cette discussion en une page

02/04/2011, 17h35
tyrtamos

Désolé, mais je ne sais pas quoi faire avec ça.

Tous mes threads sont sous forme de classes, et j'ai déjà fait des programmes compliqués avec plusieurs dizaines de threads qui se créent et disparaissent en cours d'exécution; qui ont des rôles différents (plusieurs fournisseurs / plusieurs consommateurs); dont l'ordre l'exécution est défini avec une marge de souplesse; qui se partagent des infos en lecture/écriture; dont le thread principal peut tirer périodiquement des statistiques d'activité, etc... Et je n'ai pas vu en quoi cette structure de classe avait un inconvénient quelconque: au contraire!

D'où ma remarque: j'apprends ici que mes threads sont "statiques", mais je leur fait faire tellement de choses que ça ne me dérange pas vraiment...

Tyrtamos
02/04/2011, 19h01
wiztricks

Citation:

Envoyé par tyrtamos

D'où ma remarque: j'apprends ici que mes threads sont "statiques", mais je leur fait faire tellement de choses que ça ne me dérange pas vraiment...

Les difficultés viennent généralement des différences de nature dans les associations entre "état global", "états locaux", "activités", "threads".
Potentiellement, nous avons 1x"état global", Mx"activités", Nx"threads" et jusqu'à MxN "états locaux". Ces choses existent toujours mais, suivant le contexte, il n'est pas toujours nécessaire de les traiter.

Comment matérialiser une "activité" par autre chose qu'un callable qu'il soit fonction d'un module ou méthode d'une classe ou d'une instance?
C'est ce que nous avons dans les GUI: les activités sont postés comme "events" et exécutés via une mainloop et un seul thread.

Il n'est pas interdit de poser que l'activité est le code exécuté via la méthode 'run' d'une classe dérivée de Thread. Tant que les overheads induits par le nombre de threads restent acceptables, pourquoi pas?
Mais dans le cas général, les avantages à paralléliser dépendront du type d'activité: la limite restant la capacité CPU.
Si vous voulez rester "flexible", je ne vois pas comment éviter la séparation entre "activité" et "threads".

Après vous avez les états globaux...
Les classes donnent l'illusion qu'ils sont bien encapsulés dans les instances.
Avec des threads, vous avez un même espace virtuel mais la nécessité de synchroniser d'une façon ou d'une autre les changements d'états associés. Et vous poser la question du contexte d'exécution des différents objets.

Avec le multiprocessing, les espaces virtuels sont différents, les choses peuvent rester simples (modèle Workers ou Client/Servers) ou devenir très compliquées.
- W
PS: Note: le cas général est traité (en partie) par le pattern dit "active object".
02/04/2011, 19h35
tyrtamos

Merci, wiztricks, pour les infos et... la patience.

Tout cela mérite de mûrir doucement...

Tyrtamos

Je dirais ceci :
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 ## from threading import Lock from multiprocessing import Process as Thread from multiprocessing import Lock import time print( 'run globals' ) class CMyConsole: def __init__(self): self._lock = Lock() def __call__( self, txt ): with self._lock: print( txt ) class CMyPrintJob ( Thread ): def __init__( self, MyConsole, txt, nbr ): super( CMyPrintJob, self ).__init__() self.MyConsole = MyConsole self.txt = txt self.nbr = nbr def run( self ): for i in range(self.nbr): self.MyConsole('%s-%d' % (self.txt, i)) time.sleep(0.0) ## force resched if __name__ == '__main__': MyConsole = CMyConsole() # la console... ta = CMyPrintJob( MyConsole, 'aaa', 10 ) ta.start() tb = CMyPrintJob( MyConsole, 'bbb', 10 ) tb.start() ta.join() tb.join()
La ligne "print( 'run globals' )" est très intéressante. Car à l'exécution on a la surprise de voir la chose s'afficher trois fois lorsque l'on utilise Process et une fois quand on utilise Thread.
Wiztricks, qui est malin, n'a pas pu ne pas le faire exprès. Et je crois que je n'en ai pas encore tiré toutes les conclusions que je devrais.

La class MyPrint que j'ai renommée en class CMyConsole est une excellente idée. Le verrou, anciennement nommé lock_stdout est comme l'indique son nom un verrou sur la console. Sa place est naturellement à côté de la fonction print dans un objet console.
De plus, c'est une très bon exemple d'utilisation de __call__, que je ne connaissant pas.

J'ai renommé ce qui s'appelait MyPrinter en MyPrintJob. On aurait pu dire MyPrintWorker. A chacun sa façon de nommer les choses mais on voit mieux que c'est ici que ce trouve la thread ou le process.
C'est redevenu un objet, mais comme dit Tyrtamos, tout cela mérite de mûrir doucement.
Est-ce que le niveau de la classe, avec ses méthodes et ses variables globales a elle, n'offre pas justement un niveau cohérent pour ne pas perdre de vue les états globaux, activités, etc.
Lorsque vous dites que les classes créent des illusion, je suis bien d'accord, mais est-ce que ces illusions ne peuvent pas être utilisées pour se conformer a la réalité les instanciations ?
Est-ce que les règles d'écriture des classes, avec leurs contraintes et leur sémantique (comme Print->Console et Printer->PrintJob ci-cessus), si elles sont utilisées à bon escient, ne pourraient-elles pas justement aider à faire gagner en clarté les programmes parallélisés qui sont forcément compliqué ?

05/04/2011, 20h30
wiztricks

Citation:

Envoyé par FredOoo123

La ligne "print( 'run globals' )" est très intéressante. Car à l'exécution on a la surprise de voir la chose s'afficher trois fois lorsque l'on utilise Process et une fois quand on utilise Thread

La ligne est surtout là pour vous montrer les imports de module dans le cas multithreading/Windows et la fragilité de la notion de singleton dans ce contexte.

Citation:

De plus, c'est une très bon exemple d'utilisation de __call__, que je ne connaissant pas.

Juste pour vous montrer qu'on peut transformer classes en fonctions.

Citation:

Est-ce que le niveau de la classe, avec ses méthodes et ses variables globales a elle, n'offre pas justement un niveau cohérent pour ne pas perdre de vue les états globaux, activités, etc.

La classe est juste un constructeur qui peut permettre la création d'un ensemble d'objets qui se ressemblent.
En Python, module, fonctions, classes et instances sont des objets.
Quelque soit le type de l'objet, le contexte "threads", permet l'exécution de plusieurs méthodes "en même temps" et, si on n'y prend garde, des incohérences dans son état. Cela se contrôle avec des verrous, mais vous n'allez pas en coller un à chaque objet... et il va falloir "composer" pour construire des entités cohérentes dont les frontières dépasseront le simple objet de "base" - une instance de classe -.

Citation:

Est-ce que les règles d'écriture des classes, avec leurs contraintes et leur sémantique (comme Print->Console et Printer->PrintJob ci-cessus), si elles sont utilisées à bon escient, ne pourraient-elles pas justement aider à faire gagner en clarté les programmes parallélisés qui sont forcément compliqué ?

Reprenons.
Nos objets "conceptuels" restent:
Console 1 <-----> 0, N activités 0, 1 <-----> 0, M Thread
"conceptuel" = modèle d'une réalité.

Notez que les flèches ne sont pas des objets mais des relations/associations entre eux.

Est-ce que les objets de la réalisation (code) doivent rester fidèles à ces objets conceptuels? Modèle d'une réalité ne nous donne pas trop le choix (sinon c'est que le modèle est à refaire).
D'ailleurs, construire une activité en surchargeant les méthodes __init__ et run de la classe Thread/Process est une façon de traduire l'association/relation entre activité et thread via l'héritage.

Pourquoi je n'aime pas.

Il est quelque peu réducteur de dire qu'une activité "est-un" thread, alors que le modèle dit seulement qu'une activité doit être associée à un thread pour "avancer". En plus avec l'héritage, si on change de modèle de threads, on est embêté car autre API autres comportements...

L'autre aspect est l'impact sur code et tests.
Je ne vois pas comment une activité pourrait être autre chose qu'un "callable", i.e. une fonction ou méthode invoquée avec passage d'arguments.
L'exemple est 'toy' mais l'activité pourra créer des objets, communiquer avec d'autres,... et au plus la chose deviendra compliquée et au plus sa mise au point deviendra ardue. Et pour s'assurer que la logique est "correcte", il sera utile de faire des tests sans thread, en remplaçant les autres activités par des "mocks", par exemple.

Et si activité "est-un" thread, il va falloir "défaire" la liaison pour tester.

Un autre aspect est qu'avec l'héritage, on écrit plus de code alors que nous avons une API (target, args) qui fait le boulot...
note: dans ce cas, le lien n'est plus callable "IS_A" thread mais thread "HAS_A" callable. Le lien est "statique" comme dans le cas "héritage" mais on préserve la "séparation des concerns".

Est-ce que c'est mieux? La réalisation est en ligne avec le modèle et en plus on sait tester indépendamment, donc construire plus sûr.
En tant qu'architecte, je ne peux que préférer.

- W

Voici un exemple de code qui est beaucoup plus proche de ce que je veux vraiment obtenir :
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 # -*- coding: utf_8 -*- import multiprocessing as mult import time class CConsole (object): # Vérrouille les écritures dans la console def __init__(self): self._lock = mult.Lock() def __call__( self, txt ): with self._lock: print( txt ) class CToDoListe (object): # Représente la liste des traitements à faire def __init__( self, ToDo): self.Q = mult.Queue() for item in ToDo: self.Q.put(item) def get_nowait ( self ): return self.Q.get_nowait() class CFullJob (object): # Ecrit n fois chaque élement de la liste à l'aide d'un nombre de Process identique au nombre de CPU def __init__( self, Console, Queue, n ): self.Console = Console self.Queue = Queue self.n = n self.NbrCPU = mult.cpu_count() self.Process = [] def start( self ): for i in range( self.NbrCPU ): self.Process.append ( CProcess( self.Console, self.Queue, self.n ) ) self.Process[i].start() def join( self ): for i in range( self.NbrCPU ): self.Process[i].join() class CProcess ( mult.Process ): # Processus qui fait sa part de boulot tant qu'il en reste dans la liste # ( Dans mon projet réel, l'initialisation du Process peut être lourde, c'est pourquoi # ( il n'est pas initialisé autant de fois qu'il y a d'éléments à traiter mais seulement # ( autant de fois qu'il y a de CPU def __init__( self, Console, Queue, n ): super( CProcess, self ).__init__() self.Console = Console self.Queue = Queue self.n = n def run( self ): QIsNotEmpty = 1 while QIsNotEmpty: try: ItemToDo = self.Queue.get_nowait() for i in range( self.n ): time.sleep(0.2) self.Console('%s-%d' % (ItemToDo, i+1)) except Exception: QIsNotEmpty = 0 if __name__ == '__main__': Console = CConsole() ToDoListe = CToDoListe( ["XXX", "III", "///", ":::", "000", "+++", "###", "%%%", "***", "^^^", "$$$"] ) FullJob = CFullJob( Console, ToDoListe, 3 ) FullJob.start() FullJob.join()
Ce même code fonctionne avec mon projet réel.
Au lieu d'afficher des petits groupes de caractères, il démarre autant de réseaux de neurones que j'ai de processeurs. Et la liste des choses à faire, c'est le traitement des N sorties de mon modèle.
J'ai 4 cœurs dans mon processeur. Pendant l'apprentissage du réseau de neurones, les 4 cœurs tournent à 100%. (L’algorithme du réseau de neurone est dans une DLL en C)
Le résultat effectif, c'est que je vais quasiment 4 fois plus vite. Je dois quand-même recharger toutes les données (entrées, sorties) à chaque initialisation d'un Processus, et ensuite, seulement les nouvelles sorties à traiter.

Une différence entre le projet et la démo présentée ici.
Dans le projet, j'attends un résultat à la fin de chaque traitement.
Il me faudrait une liste des résultats accumulée dans CFullJob, pour pouvoir les lire tranquillement après le FullJob.join()

Pour l'instant, j'y arrive avec des grosses astuces, mais quelqu'un a-t-il une idée pour le faire joliment, par exemple en écrivant à partir de la ligne 60/61 de CProcess.run() dans une liste des résultats qui est instanciée dans CFullJob ?

(Voir la discussion précédente)
Le code suivant est plus complet et tourne sous Python 2.6.6, 2.7.1 et 3.2

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
# -*- coding: utf_8 -*-
 
import multiprocessing as mult
import time
import os
import sys
 
if ( sys.version[0]=="3" ):
    from queue import Empty
else:
    from Queue import Empty
 
 
class CConsole (object):
    # Vérrouille les écritures dans la console
 
    def __init__(self):
        self._lock = mult.Lock()
 
    def __call__( self, txt ):
        with self._lock:
            print( txt )
 
 
 
class CToDoQueue (object):
    # Représente la liste des traitements à faire
 
    def __init__( self, ToDo):
        self.Q = mult.Queue()
        for item in ToDo:
            self.Q.put(item)
 
    def get_nowait ( self ):
        return self.Q.get_nowait()
 
 
 
class CFullJob (object):
    # Ecrit n fois chaque élement de la liste à l'aide d'un nomble de Process identique au nombre de CPU
 
    def __init__( self, ToDoListe, n ):
        self.Console     = CConsole()
        self.ToDoQueue   = CToDoQueue( ToDoListe )
        self.n           = n
        self.NbrCPU      = mult.cpu_count()
        self.Process     = []
        self.ResultQueue = mult.Queue()
 
    def __call__( self ):
        self.start()
        self.join()
        ResultList = []
        QIsNotEmpty = True
        while QIsNotEmpty:
            try:
                Result = self.ResultQueue.get_nowait()
                ResultList += Result
            except Empty:
                QIsNotEmpty = False
            except Exception:
                raise
        return ResultList
 
    def start( self ):
        for i in range( self.NbrCPU ):
            self.Process.append ( CProcess( self.ResultQueue , self.Console, self.ToDoQueue, self.n ) )
            self.Process[i].start()
 
    def join( self ):
        for i in range( self.NbrCPU ):
            self.Process[i].join()
 
 
 
class CResult (object):
    # Ceci est un élément de ma liste de résultats
    def __init__( self, txt, pid):
        self.txt = txt
        self.pid = pid
 
 
 
class CProcess ( mult.Process ):
    # Processus qui fait sa part de boulot tant qu'il en reste dans la liste
    # ( Dans mon projet réel, l'initialisation du Process peut être lourde, c'est pourquoi
    # ( il n'est pas initialisé autant de fois qu'il y a d'éléments à traiter mais seulement
    # ( autant de fois qu'il y a de CPU
 
    def __init__( self, ResultQueue, Console, ToDoQueue, n ):
        super( CProcess, self ).__init__()
        self.ResultQueue = ResultQueue
        self.Console     = Console
        self.ToDoQueue   = ToDoQueue
        self.n           = n
 
    def run( self ):
        QIsNotEmpty = True
        while QIsNotEmpty:
            try:
                ItemToDo = self.ToDoQueue.get_nowait()
                for i in range( self.n ):
                    time.sleep(0.2)
                    self.Console('%s-%d' % (ItemToDo, i+1))
                #self.ResultQueue.put(  "[%s] %i" % (ItemToDo[0],os.getpid())  )
                self.ResultQueue.put(  ItemToDo[0]  )
                #ItemResult = CResult( "[%s]" % ItemToDo[0], os.getpid() )
                #self.ResultQueue.put(  ItemResult  )
            except Empty:
                QIsNotEmpty = False
            except Exception:
                raise
 
 
 
if __name__ == '__main__':
 
    # Initialisations
    ToDoListe = ["XXX", "III", "///", ":::", "000", "+++", "###", "%%%", "***", "^^^", "$$$"]
    FullJob   = CFullJob( ToDoListe, 3 )
 
    # Ici je fais le job, dans des Process, avec les start() et les joint()
    ResultList = FullJob()
 
    print('')
 
    # Ici je fais ce que je veux avec mes résultats
    for Result in ResultList:
        print( Result )
        #print( Result.txt, Result.pid )
 
    print('')

1) Question général :
Est-ce que je fais un usage des Queue à bon escient ?

2) Question de détail :
Si j'utilise la ligne 105 plutôt que 106, je n'obtiens pas le "[X] 15246" escompté car il y a un retour à la ligne après chaque caractère. Je trouve cela très bizarre et je ne vois pas d'explication.

3) Question sur mon plus gros problème :
En situation réelle, mon projet retourne des résultats plus compliqué que ça.
Ce sont donc les lignes 107 et 108 que je veux utiliser (plutôt que 105 et 106), avec la ligne 130 plutôt que 129 pour l'affichage.
N'est-il pas possible de placer des instances de CResult dans une Queue ?

Merci pour votre aide.

11/04/2011, 18h57
FredOoo123

.

Les Questions n°2 et n°3 sont résolues en remplaçant simplement la ligne 58

Code:

ResultList += Result

par

Code:

ResultList += [Result]

Les remarques générales sur l'ensemble, et sur l'utilisation des Queue demeurent les bienvenues.

.

Le Module MultiCore.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
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
# -*- coding: utf_8 -*-
 
import multiprocessing as multi
import sys
 
__all__ = ["MultiProcess", "CoreProcess"]
 
 
if ( sys.version[0]=="3" ):
    from queue import Empty
else:
    from Queue import Empty
 
 
class CConsole (object):
    # Vérrouille les écritures dans la console
 
    def __init__(self):
        self._lock = multi.Lock()
 
    def __call__( self, txt ):
        with self._lock:
            print( txt )
 
 
 
class CToDoQueue (object):
    # Représente la liste des traitements à faire
 
    def __init__( self, ToDo):
        self.Q = multi.Queue()
        for item in ToDo:
            self.Q.put(item)
 
    def get_nowait ( self ):
        return self.Q.get_nowait()
 
 
 
class MultiProcess (object):
    # Lance un nombre de Process identique au nombre de CPU sur la machine pour effectuer tous les traitements dans la liste
 
 
    def __init__( self, CProcess, ToDoListe ):
        self.CProcess    = CProcess
        self.ToDoQueue   = CToDoQueue( ToDoListe )
        self.Console     = CConsole()
        self.NbrCPU      = multi.cpu_count()
        self.Process     = []
        self.ResultQueue = multi.Queue()
 
    def __call__( self ):
        self.start()
        self.join()
        ResultList = []
        QIsNotEmpty = True
        while QIsNotEmpty:
            try:
                Result = self.ResultQueue.get_nowait()
                ResultList += [Result]
            except Empty:
                QIsNotEmpty = False
            except Exception:
                raise
        return ResultList # Dans n'importe quel ordre
 
    def start( self ):
        for i in range( self.NbrCPU ):
            self.Process.append ( self.CProcess( self.ResultQueue , self.Console, self.ToDoQueue ) )
            self.Process[i].start()
 
    def join( self ):
        for i in range( self.NbrCPU ):
            self.Process[i].join()
 
 
 
 
class CoreProcess ( multi.Process ):
    # Processus qui fait sa part de boulot tant qu'il en reste dans la liste
    # ( Dans mon projet réel, l'initialisation du Process peut être lourde, c'est pourquoi
    # ( il n'est pas initialisé autant de fois qu'il y a d'éléments à traiter mais seulement
    # ( autant de fois qu'il y a de CPU
 
    def __init__( self, ResultQueue, Console, ToDoQueue ):
        super( CoreProcess, self ).__init__()
        self.ResultQueue = ResultQueue
        self.Console     = Console
        self.ToDoQueue   = ToDoQueue
        self.Empty       = Empty
 
    def run( self ):  # Devra être surchargé pour faire vraiment ce que l'on veut
        import time, os
        QIsNotEmpty = True
        while QIsNotEmpty:
            try:
                # ////////////////////////////////////////////////
                ItemToDo = self.getItemToDo()
                time.sleep(0.2)
                self.Console( ItemToDo )
                self.addItemResult( ItemToDo*ItemToDo )
                # \\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\
            except self.Empty:
                QIsNotEmpty = False
            except Exception:
                raise
 
    def getItemToDo( self ):
        return self.ToDoQueue.get_nowait()
 
    def addItemResult( self, ItemResult ):
        self.ResultQueue.put( ItemResult )
 
 
 
if __name__ == '__main__':
 
    # Initialisations
    ToDoListe = [ 1, 2, 3, 4, 5, 6, 7, 8, 9, 10 ]
    MyMultiProcess = MultiProcess( CoreProcess, ToDoListe )
 
    # Ici je fais le job, dans des Process, avec les start() et les joint()
    ResultList = MyMultiProcess()
 
    print('')
 
    # Ici je fais ce que je veux avec mes résultats
    for Result in ResultList:
        print( Result )
 
    print('')

Sortie dans la console en cas d'exécution du module :

Code:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
1
2
3
4
5
6
7
8
9
10
 
1
4
9
16
25
36
49
64
81
100
 
Appuyez sur une touche pour continuer...

Voici un premier exemple d'utilisation :

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
# -*- coding: utf_8 -*-
 
import MultiCore
 
 
class CResult (object):
    # Ceci est un élément de ma liste de résultats
    def __init__( self, value, result, pid):
        self.value = value
        self.result = result
        self.pid = pid
    def show( self ):
        return "%d*%d = %d - %i" % ( self.value, self.value, self.result, self.pid )
 
 
class CMyProcess ( MultiCore.CoreProcess ):
    # Processus qui fait sa part de boulot tant qu'il en reste dans la liste
    # Interface :
    #    run()           : Doit être surchargé
    #    getItemToDo()   : Obtenir une tâche à exécuter
    #    addItemResult() : Ajouter un résultat
    #    Empty           : Exception quand la liste des tâches est vide
    #    Console( txt )  : Pour faire un print verrouillé
    def run( self ):
        import time
        QIsNotEmpty = True
        while QIsNotEmpty:
            try:
                # ////////////////////////////////////////////////
                ItemToDo = self.getItemToDo()
                self.Console('%d' % ( ItemToDo*ItemToDo ))
                ItemResult = CResult( ItemToDo, ItemToDo*ItemToDo, self.pid )
                time.sleep(0.1)
                self.addItemResult( ItemResult )
                # \\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\
            except self.Empty:
                QIsNotEmpty = False
            except Exception:
                raise
 
 
if __name__ == '__main__':
 
 
    # Initialisations
    ToDoListe = [ 1, 2, 3, 4, 5, 6, 7, 8, 9, 10 ]
    MyMultiProcess = MultiCore.MultiProcess( CMyProcess, ToDoListe )
 
    # Ici je fais le job, dans des Process, avec les start() et les joint()
    ResultList = MyMultiProcess()
 
    print('')
 
    # Ici je fais ce que je veux avec mes résultats
    for Result in ResultList:
        print( Result.show() )
 
    print('')

Sortie console :

Code:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
1
4
9
16
25
36
49
64
81
100
 
1*1 = 1 - 13092
2*2 = 4 - 9016
3*3 = 9 - 11056
4*4 = 16 - 11392
5*5 = 25 - 13092
6*6 = 36 - 9016
7*7 = 49 - 11056
8*8 = 64 - 11392
9*9 = 81 - 13092
10*10 = 100 - 9016
 
Appuyez sur une touche pour continuer...

Voici un second exemple d'utilisation :

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
# -*- coding: utf_8 -*-
 
import MultiCore
 
 
class CResult (object):
    # Ceci est un élément de ma liste de résultats
    def __init__( self, txt, pid):
        self.txt = txt
        self.pid = pid
 
 
class CMyProcess ( MultiCore.CoreProcess ):
    # Processus qui fait sa part de boulot tant qu'il en reste dans la liste
    # Interface :
    #    run()           : Doit être surchargé
    #    getItemToDo()   : Obtenir une tâche à exécuter
    #    addItemResult() : Ajouter un résultat
    #    Empty           : Exception quand la liste des tâches est vide
    #    Console( txt )  : Pour faire un print verrouillé
    def run( self ):
        import time
        QIsNotEmpty = True
        while QIsNotEmpty:
            try:
                # ////////////////////////////////////////////////
                ItemToDo = self.getItemToDo()
                for i in range( 3 ):
                    time.sleep(0.2)
                    self.Console('%s-%d' % (ItemToDo, i+1))
                ItemResult = CResult( "[%s]" % ItemToDo[0], self.pid )
                self.addItemResult( ItemResult )
                # \\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\
            except self.Empty:
                QIsNotEmpty = False
            except Exception:
                raise
 
 
if __name__ == '__main__':
 
 
    # Initialisations
    ToDoListe = ["XXX", "III", "///", ":::", "000", "+++", "###", "%%%", "***", "^^^", "$$$"]
    MyMultiProcess = MultiCore.MultiProcess( CMyProcess, ToDoListe )
 
    # Ici je fais le job, dans des Process, avec les start() et les joint()
    ResultList = MyMultiProcess()
 
    print('')
 
    # Ici je fais ce que je veux avec mes résultats
    for Result in ResultList:
        print( Result.txt, Result.pid )
 
    print('')

Sortie console :

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
XXX-1
III-1
///-1
:::-1
XXX-2
III-2
///-2
:::-2
XXX-3
III-3
///-3
:::-3
+++-1
000-1
###-1
%%%-1
+++-2
000-2
###-2
%%%-2
+++-3
000-3
###-3
%%%-3
***-1
^^^-1
$$$-1
***-2
^^^-2
$$$-2
***-3
^^^-3
$$$-3
 
('[X]', 10184)
('[I]', 10960)
('[/]', 9660)
('[:]', 10004)
('[+]', 10960)
('[0]', 10184)
('[#]', 9660)
('[%]', 10004)
('[*]', 10960)
('[^]', 10184)
('[$]', 9660)
 
Appuyez sur une touche pour continuer...

Je suis intéressé par toute critique ou suggestion.
Merci

12/04/2011, 08h05
tyrtamos

Bonjour,

C'est de la belle programmation, et tout fonctionne: j'obtiens les mêmes résultats. Je mettrai un certain temps à éplucher ton code pour mieux le comprendre.

Juste un point que je ne comprends pas bien (mais j'ai de grosses lacunes sur le sujet): tu crées autant de processus qu'il y a de cores dans le CPU, mais quelle assurance as-tu qu'il ne seront pas tous traités par le même? Y a-t-il une méthode Python pour équilibrer les charges entre les cores?

Tyrtamos
12/04/2011, 09h47
mont29

Ben, normalement, c’est le boulot de l’OS de répartir les processus entre les différents cores/CPUs disponibles, non ?
12/04/2011, 10h12
FredOoo123

Citation:

Envoyé par tyrtamos

Tu crées autant de processus qu'il y a de cores dans le CPU, mais quelle assurance as-tu qu'il ne seront pas tous traités par le même? Y a-t-il une méthode Python pour équilibrer les charges entre les cores?

D'après ce que j'ai compris, il n'y a aucun moyen de placer tel Process ou tel Thread dans le core de son choix. Mais ce n'est pas l'application ou Python qui équilibre les charges, c'est le système. Ça c'est la première chose à comprendre.
La seconde, c'est qu'il faut faire confiance au système. J'ai lu quelque part quelqu'un qui disait qu' "il est illusoire de croire que l'on va faire intuitivement mieux que ce pourquoi le système a déjà été optimisé", et cela me rappel quelques cours de fac. Donc le fais confiance.

Un exemple que je trouve convainquant : Tu ne peux pas savoir, et ton application non plus, même quand elle tourne, ce qui se passe dans le système. Qui te dit que l'utilisateur de ton programme n'a pas déjà lancé 10 grosses cochonnes d'applications qui se battent elles aussi pour les core ?
Si ton application essaye de regarder les temps d'occupation des cores pour définir une stratégie optimale, il fait le boulot du système et donc il perd du temps.

Ensuite, lorsque je fait tourner mon exemple avec mon vrai projet, les calculs sont beaucoup plus lourds. Et dans le gestionnaire de tâches je constate que pendant l'exécution de mon application, mes 4 cores sont utilisés à 100%. J'insiste, c'est pas 99% mais bien 100%.
J'ai un Intel i5, dont le premier core est spécialisé dans la gestion des 4 autres. Et la fonction cpu_count() de Python retourne 4, non pas 5.

Donc, ma stratégie à moi c'est simplement de démarrer autant de Process que j'ai de core. Le système fera le reste.

Remarque: Quand j'utilise des Thread, je suis très loin des 100% même sur le premier core, et l'équilibre entre les cores est très mauvais, sans doute parce que Python est un langage interprété et qu'il démarre tous les Thread dans le Process de l’interpréteur (...j'ai lu). Il faudrait tester en C avec les Thread, ce que je n'ai pas fait.
En revanche, en C, dans une utilisation normale de mon application, sans Thread ni multi Process, la répartition des charges entre les cores est aussi très mauvaise. L'algorithme n'a pas été pensé pour les processeurs multi core.

La stratégie finalement, c'est donc que l'application donne les moyens au système d'utiliser la machine le mieux possible.
12/04/2011, 10h42
mont29

Le problème des threads dans Python n’est pas tant qu ’ils s’exécutent dans un seul et même processus (c’est le principe des threads !), mais que l’interpréteur Python utilise un verrou global (le GIL, je crois…), ce qui a pour effet de sérialiser les threads (puisqu’un seul d’entre eux peut s’exécuter à la fois) !

Sinon, l’OS est capable de répartir les threads exactement comme les processus normaux.

Une version simplifiée, améliorée et commentée de MultoCore.py (pour Python 2.6.6, 2.7.1 et 3.2)

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
# -*- coding: utf_8 -*-
 
import multiprocessing as multi
import sys
 
__all__ = ["MultiProcess", "CoreProcess"]
 
 
if ( sys.version[0]=="3" ):   #: En fait, j'utilise partout (enfin, seulement deux fois) des multiprocessing.Queue()
    from queue import Empty   #: mais l'exception qui dit que la Queue est vide est celle de la Queue standard,
else:                         #: laquelle se trouve dans le module 'Queue', ou 'queue' dans les versions 3.x de Python
    from Queue import Empty
 
 
 
class CConsole (object):
    # Vérrouille les écritures dans la console
 
    def __init__(self):
        self._lock = multi.Lock()
 
    def __call__( self, txt ):
        with self._lock:
            print( txt )
 
 
 
class MultiProcess (object):
    # Lance un nombre de Process identique au nombre de CPU sur la machine pour effectuer tous les traitements dans la liste
 
    def __init__( self, CProcess, ToDoList, **DArgs ):
        self.CProcess    = CProcess
        self.ToDoQueue   = multi.Queue()  #: Et non plus = CToDoQueue( ToDoListe )
        for item in ToDoList:             #: Ces deux lignes remplacent la class CToDoQueue présentes dans
            self.ToDoQueue.put(item)      #:   les versions précédentes du module
        self.Console     = CConsole()
        self.NbrCPU      = multi.cpu_count()
        self.Process     = []
        self.ResultQueue = multi.Queue()
        self.DArgs = DArgs
 
    def __call__( self ):
        self.start()
        self.join()
        #: Ici, tous les travaux sont fini, on s'occupe de ranger les résultats
        ResultList = []
        QIsNotEmpty = True
        while QIsNotEmpty:
            try:
                Result = self.ResultQueue.get_nowait()
                ResultList += [Result]
            except Empty:
                QIsNotEmpty = False
            except Exception:
                raise
        return ResultList # Dans n'importe quel ordre
 
    def start( self ):
        for i in range( self.NbrCPU ):
            self.Process.append ( self.CProcess( self.ResultQueue , self.Console, self.ToDoQueue, **self.DArgs ) )
            self.Process[i].start()
 
    def join( self ):
        for i in range( self.NbrCPU ):
            self.Process[i].join()
 
 
 
class CoreProcess ( multi.Process ):
    # Processus qui fait sa part de boulot tant qu'il en reste dans la liste
    # ( Dans mon projet réel, l'initialisation du Process peut être lourde, c'est pourquoi
    # ( il n'est pas initialisé autant de fois qu'il y a d'éléments à traiter mais seulement
    # ( autant de fois qu'il y a de CPU
    # Interface :
    #    run()           : Doit être surchargé
    #    getItemToDo()   : Obtenir une tâche à exécuter
    #    addItemResult() : Ajouter un résultat
    #    Empty           : Exception quand la liste des tâches est vide
    #    Console( txt )  : Pour faire un print verrouillé
    #    Et aussi tout argument nommé passé à MultiProcess
 
    def __init__( self, ResultQueue, Console, ToDoQueue, **DArgs ):
        super( CoreProcess, self ).__init__()
        self.ResultQueue = ResultQueue
        self.Console     = Console
        self.ToDoQueue   = ToDoQueue
        self.Empty       = Empty               #: Comme ça, l'utilisateur qui hérite de CoreProcess n'est pas emmerdé avec
        for attr, value in DArgs.items():      #: la déclaration Queue.Empty ou queue.Empty
            dict.__setattr__( self, attr, value )     #: Permet à tous les arguments nommés de DArgs, par exemple toto=33
                                                      #: d'être accessibles dans run() par self.toto
    def run( self ): # Devra être surchargé pour faire vraiment ce que l'on veut
        import time, os
        QIsNotEmpty = True
        while QIsNotEmpty:
            try:
                # ////////////////////////////////////////////////
                ItemToDo = self.getItemToDo()
                time.sleep(0.2)
                self.Console( "%s %s" % (ItemToDo, self.toto) )
                self.addItemResult( ItemToDo*ItemToDo )
                # \\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\
            except self.Empty:
                QIsNotEmpty = False
            except Exception:
                raise
 
    def getItemToDo( self ):
        return self.ToDoQueue.get_nowait()
 
    def addItemResult( self, ItemResult ):
        self.ResultQueue.put( ItemResult )
 
 
 
if __name__ == '__main__':
 
    # Initialisations
    ToDoList = [ 1, 2, 3, 4, 5, 6, 7, 8, 9, 10 ]
    MyMultiProcess = MultiProcess( CoreProcess, ToDoList, toto=33 )
 
    # Ici je fais le job, dans des Process, avec les start() et les joint()
    ResultList = MyMultiProcess()
 
    print('')
 
    # Ici je fais ce que je veux avec mes résultats
    for Result in ResultList:
        print( Result )
 
    print('')

La simplification c'est qu'il n'y a plus de class CToDoQueue.

L'amélioration c'est que l'on peut passer autant de d'arguments nommés au constructeur de MultiProcess (voir **DArgs, par exemple toto=33), on les retrouvera sous forme d'attributs dans CoreProcess (exemple self.toto a pour valeur 33).

12/04/2011, 10h48
FredOoo123

Citation:

Envoyé par mont29

l’interpréteur Python utilise un verrou global (le GIL, je crois…), ce qui a pour effet de sérialiser les threads (puisqu’un seul d’entre eux peut s’exécuter à la fois) !

Oui, c'est ça que j'avais lu.

Citation:

Envoyé par mont29

Sinon, l’OS est capable de répartir les threads exactement comme les processus normaux.

C'est donc le GIL qui empêche le système de répartir les threads.
12/04/2011, 11h00
mont29

Oui, puisque chaque thread doit acquérir le GIL pour pouvoir s’exécuter, et qu’un seul d’entre eux peut le faire à la fois, du point de vue de l’OS, il n’y a toujours qu’un seul thread en activité, les autres étant en idle en attente du fameux GIL…

En clair, il est impossible de faire de la programmation vraiment parallèle avec python, sans passer par des processus indépendants, comme tu le fais dans ton code (si j’ai bien suivi :mouarf: ).

Autre exemple d'utilisation, et problème :

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
# -*- coding: utf_8 -*-
 
import MultiCore
 
import ctypes  #: Pour pouvoir loader une DLL
import os
os.environ['PATH'] = "D:\\CODE\\IM\\gcc" + ';' + os.environ['PATH'] #: Pour trouver le chemin de la DLL
 
 
 
class CResult (object):
    # Ceci est un élément de ma liste de résultats
    def __init__( self, txt, pid):
        self.txt = txt
        self.pid = pid
 
 
class CMyProcess ( MultiCore.CoreProcess ):
    # Processus qui fait sa part de boulot tant qu'il en reste dans la liste
    def run( self ):
        QIsNotEmpty = True
        while QIsNotEmpty:
            try:
        # ////////////////////////////////////////////////////////////////////
                ItemToDo = self.getItemToDo()
                for i in range( self.nbr ):
                    self.Console( '%s' % ItemToDo )
                    #:dll.MyPrint( ItemToDo, 10 )                  #: dll.MyPrint écrit 10 fois ItemToTo
                    #:for i in range(10): dll.MySPrint( ItemToDo ) #: dll.MySPrint écrit simplement 1 fois ItemToTo
                ItemResult = CResult( "[%s]" % ItemToDo[0], self.pid )
                self.addItemResult( ItemResult )
        # \\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\\
            except self.Empty:
                QIsNotEmpty = False
            except Exception:
                raise
 
 
 
if __name__ == '__main__':
 
    # Initialisations
    ToDoList = ["XXX", "III", "///", ":::", "000", "+++", "###", "%%%", "***", "^^^", "$$$"]
    #:dll = ctypes.CDLL( "MyPrinter.dll" )
    MyMultiProcess = MultiCore.MultiProcess( CMyProcess, ToDoList, nbr=2 ) # Si je passe dll=dll, ça ne marche pas
 
    # Ici je fais le job, dans des Process, avec les start() et les joint()
    ResultList = MyMultiProcess()
 
    print('')
 
    # Ici je fais ce que je veux avec mes résultats
    for Result in ResultList:
        print( Result.txt, Result.pid )
 
    print('')

Mon objectif est d'utiliser une DLL dans les Process. (C'est mon droit, non ? Et de toutes façons je n'ai pas le choix)
Tout ce qui concerne la DLL est en commentaire dans l'exemple ci-dessus pour fournir un code en état de marche.

Maintenant je modifie les lignes 44 et 45 de la manière suivante :

Code:

1
2
    dll = ctypes.CDLL( "MyPrinter.dll" )
    MyMultiProcess = MultiCore.MultiProcess( CMyProcess, ToDoList, nbr=2, dll=dll ) # Si je passe dll=dll, ça ne marche pas

Et voici la grosse erreur que j'obtiens : (thread_04_10_M.py = MultiCore.py chez moi, thread_04_10_03.py c'est l'exemple ci-dessus)

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
Traceback (most recent call last):
  File "thread_04_10_03.py", line 48, in <module>
    ResultList = MyMultiProcess()
  File "D:\CODE\IM\Python\thread_04_10_M.py", line 43, in __call__
    self.start()
  File "D:\CODE\IM\Python\thread_04_10_M.py", line 61, in start
    self.Process[i].start()
  File "C:\MySys\Python26\lib\multiprocessing\process.py", line 104, in start
    self._popen = Popen(self)
  File "C:\MySys\Python26\lib\multiprocessing\forking.py", line 239, in __init__
    dump(process_obj, to_child, HIGHEST_PROTOCOL)
  File "C:\MySys\Python26\lib\multiprocessing\forking.py", line 162, in dump
    ForkingPickler(file, protocol).dump(obj)
  File "C:\MySys\Python26\lib\pickle.py", line 224, in dump
    self.save(obj)
  File "C:\MySys\Python26\lib\pickle.py", line 331, in save
    self.save_reduce(obj=obj, *rv)
  File "C:\MySys\Python26\lib\pickle.py", line 419, in save_reduce
    save(state)
  File "C:\MySys\Python26\lib\pickle.py", line 286, in save
    f(self, obj) # Call unbound method with explicit self
  File "C:\MySys\Python26\lib\pickle.py", line 649, in save_dict
    self._batch_setitems(obj.iteritems())
  File "C:\MySys\Python26\lib\pickle.py", line 681, in _batch_setitems
    save(v)
  File "C:\MySys\Python26\lib\pickle.py", line 331, in save
    self.save_reduce(obj=obj, *rv)
  File "C:\MySys\Python26\lib\pickle.py", line 419, in save_reduce
    save(state)
  File "C:\MySys\Python26\lib\pickle.py", line 286, in save
    f(self, obj) # Call unbound method with explicit self
  File "C:\MySys\Python26\lib\pickle.py", line 649, in save_dict
    self._batch_setitems(obj.iteritems())
  File "C:\MySys\Python26\lib\pickle.py", line 681, in _batch_setitems
    save(v)
  File "C:\MySys\Python26\lib\pickle.py", line 295, in save
    self.save_global(obj)
  File "C:\MySys\Python26\lib\pickle.py", line 748, in save_global
    (obj, module, name))
pickle.PicklingError: Can't pickle <class 'ctypes._FuncPtr'>: it's not found as ctypes._FuncPtr
Appuyez sur une touche pour continuer... Traceback (most recent call last):
  File "<string>", line 1, in <module>
  File "C:\MySys\Python26\lib\multiprocessing\forking.py", line 342, in main
    self = load(from_parent)
  File "C:\MySys\Python26\lib\pickle.py", line 1370, in load
    return Unpickler(file).load()
  File "C:\MySys\Python26\lib\pickle.py", line 858, in load
    dispatch[key](self)
  File "C:\MySys\Python26\lib\pickle.py", line 880, in load_eof
    raise EOFError
EOFError

Maintenant voici la fausse solution que j'utilise pour le moment :
Je place la ligne

Code:

dll = ctypes.CDLL( "MyPrinter.dll" )

au dessus de

Code:

if __name__ == '__main__':

et à la ligne 45 je ne passe pas le paramètre

Code:

dll=dll

Ce qui veut dire que j'ai autant d'instances de la DLL que de Process. Disons 4 fois.
Ça veut dire que dans la situation réelle de mon projet, je dois refaire tout un paquet de réinitialisations 4 fois, recharger un fichier qui peut être gros, alors que ce n'est pas vraiment utile.

En fait, admettons dans un premier temps que je doive tout recharger 4 fois, ma DLL c'est toujours la même !

Là, je veux bien un peu d'aide.

Il y a deux sujets que je comprends mal:
- les modes de chargement des DLL (mode "threadable" en VC++ ..?)
- la notion de pickle en Python (la sérialisation...?)

.

12/04/2011, 12h09
mont29

Bon, je propose mes pistes, mais là je ne peux vraiment pas prétendre à l’expertise, hein :mrgreen:

De ce que je déduis de ces traces, à la création du processus, toutes les données qui lui sont transmises sont sérialisées – et la classe ctype.CDLL n’est manifestement pas sérialisable (ce qui me paraît assez logique ;) ). De toute façon, je ne pense pas que plusieurs processus puissent partager un même objet (sauf peut-être s’il y a partage mémoire, mais là je suis largué :aie: ).

Maintenant, le chargement d’une dll (ou d’un so sous unix) revient à l’OS, qui normalement ne le fait qu’une fois, avant d’y relier tous les processus qui en ont besoin. Donc je ne pense pas que ta dll soit chargée plusieurs fois en mémoire, la seule chose que tu auras en de multiples exemplaires, c’est la “glue” python/ctype – et je pense que c’est tout à fait normal !

Voilà, maintenant, il te faudrait l’avis d’un expert python/ctype/OS pour valider mes modestes déductions… :P
12/04/2011, 15h46
FredOoo123

Citation:

Envoyé par mont29

classe ctype.CDLL n’est manifestement pas sérialisable (ce qui me paraît assez logique ;) )

Mais peut-être que d'autres class permettent de charger une DLL d'une autre manière ?
(Par exemple Numpy, mais enfin, celle-là il ne marche pas non plus)

Et pourquoi ça paraît logique ?

Citation:

Envoyé par mont29

De toute façon, je ne pense pas que plusieurs processus puissent partager un même objet (sauf peut-être s’il y a partage mémoire, mais là je suis largué :aie: ).

Maintenant, le chargement d’une dll (ou d’un so sous unix) revient à l’OS, qui normalement ne le fait qu’une fois, avant d’y relier tous les processus qui en ont besoin. Donc je ne pense pas que ta dll soit chargée plusieurs fois en mémoire, la seule chose que tu auras en de multiples exemplaires, c’est la “glue” python/ctype – et je pense que c’est tout à fait normal !

Oui, il s'agit bien de partager quelque chose en mémoire.

Oui, c'est exactement pareille sous Windows (un seul chargement par l'OS, et un paquet de “glue” pour chaque programme qui utilise la DLL)

Dans le dernier exemple, avec la DLL qui fait les print dans la console, j'ai placé un petit compteur qui numérote les items traités de la ToDoListe.
Tous ceux qui on été traités dans le même Process ont un compteur commun, global entre eux. (les numéros commencent à 1 et se suivent, mais dans un autre Process, ça recommence à 1)
Disons que si je pouvais avoir un compteur global, ça serait vraiment bien pour mon projet réel.

Aussi, il faut savoir qu'en ce qui concerne la DLL du projet réel, je n'ai pas directement la main dessus, mais je connais l'auteur. Donc quand j'aurai bien compris ce qui se passe, je pourrais lui demander quelques petites choses
12/04/2011, 16h30
tyrtamos

Merci pour les réponses à mes questions plus haut.

Sinon, il me semble que le module multiprocessing permet aux différents processus de partager des données. Mais je ne sais pas si ça permettrait de partager l'appel à des dll.

Tyrtamos

Voir 40 message(s) de cette discussion en une page