Multiprocessing et classes

Bonjour !

Je vais commencer par présenter ce que j'aimerais savoir :

J'ai essayer ce bout de script et obtenu ceci :

Code:

---

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18

>>> from multiprocessing import Process >>> class Foo: ... def run(self): ... print(self.stop) ... self.stop = False ... print(self.stop) ... def start(self): ... self.stop = True ... p = Process(target = self.run) ... p.start() ... p.join() ... >>> f = Foo() >>> f.start() True # Le self.stop "parallèle" est à True False # Le self.stop "parallèle" est bien passé à False >>> f.stop True # Le self.stop "parallèle" est à False mais celui de ma classe reste à True ...

---

Ma question étant : Comment je pourrait modifier f.stop (et plus globalement f et tous ses attributs) en utilisant les Process ?


En fait mon problème est dût au thread que j'utilise en ce moment dans mon programme. Il y a beaucoups de threads et, pour une raison qui m'échappe, l'ordonnanceur décide parfois d'accorder 1 ou 2 secondes aux threads en laissant freeze mon programme principal (2 secondes pour faire un "l = []" me parait long ...). J'ai chercher une solution et visiblement le multiprocess serait plus performant. Seulement, ces classes qui tournent en parallèle ont besoin de rester actives et surtout de communiquer avec le programme principal (autrement que par un print ...). Et avant d'implémenter le multiprocess, je me suis rendu compte du problème juste au dessus.

Donc si quelqu'un à une idée pour "les threads qui font freeze" je suis preneur, mais j'aimerais quand même comprendre le multiprocess ;)

Salut,

Les "process" de multiprocessing sont des interpréteurs Python différents: chaque process aura sa définition de Foo et les instances sont différentes. Pourquoi ne pas commencer par lire la documentation?
Vous y trouveriez des exemples de cas d'utilisation simples pour "partager" entre plusieurs process.

Citation:

---

En fait mon problème est dût au thread que j'utilise en ce moment dans mon programme. Il y a beaucoups de threads et, pour une raison qui m'échappe, l'ordonnanceur décide parfois d'accorder 1 ou 2 secondes aux threads en laissant freeze mon programme principal (2 secondes pour faire un "l = []" me parait long ...).

---

La question pourrait être est-ce que les threads sont nécessaires? Si oui, est-ce qu'autant de threads sont nécessaires?

- W

Bonjour,

Si tu veux de l'aide sur les threads, il faut dire comment tu les utilises: fais un petit code test (suffisamment complet pour qu'on puisse l'exécuter) avec le problème et montre-le ici.

Par exemple: avec les bibliothèques graphiques (tkinter, pyqt, ...), des "freeze" sont obtenus lorsque qu'on accède au graphique directement à partir des threads.

On peut aussi avoir des conflits en accédant à partir des threads à des variables hors threads avec une mauvaise utilisation des verrous.

J'utilise les threads et je ne rencontre pas tes problèmes. Mais, bien sûr, il faut coder en tenant compte du manuel.

J'ai un générateur de map qui utilise quelque chose comme ça :

Code:

---

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23

>>> from time import sleep >>> from threading import Thread >>> class MapGenerator(Thread): ... def __init__(self): ... Thread.__init__(self) ... self.stop = False ... self.etages = [] ... self.etage_en_cours_de_generation = [] ... self.id_etage_en_cours = 1 ... self.start() # Je n'ai pas mon code sous les yeux donc je sais plus ... def get_etage(self, id): # comment j'apelle start (self ou Thread) mais ça marche ... if id == self.id_etage_en_cours: ... self.id_etage_en_cours += 1 # Derniere étage généré, donc on en prévois un nouveau ... return self.etages[id] ... def run(self): ... while not self.stop: ... if self.id_etage_en_cours == len(self.etages): ... generer_etage(self.id_etage_en_cours) ... self.etages.append(self.etage_en_cours_de_generation) ... sleep(10) ... def generer_etage(self, id_etage): ... # Genère l'étage dans self.etage_en_cours_de_generation ... pass

---

Grosso modo, j'ai une liste dont l'index correspond à chaque étage. Chaque étage est généré dans une liste (etage_en_cours_de_generation) à laquelle je n'accède jamais de l'extérieur. le temps de génération peut être extrèmement long et je ne peux donc pas attendre d'avoir atteint un étage avant de le générer (j'ai déjà atteint les 10 minutes pour un étage). Mais même si je met du temps avant de générer une salle, il me faut du temps pour l'atteindre, et donc cette méthode fonctionne sans problème. J'ai une sécurité qui attend que l'étage soit créer avant de le demander mais il ne m'as jamais servi pour autre chose qu'un test.
Pour l'utilisation de cette classe, je n'ai qu'une méthode publique (en dehors des Thread et Object) qui et get_etage(self, etage)

J'y avais pas pensé sur le coups, mais les autres threads servent au path finding et sont effacé quand il n'y a plus d'ennemis (Vérifié à coups de print, il s'arrêtent bel et bien) donc ces threads n'ont peut être pas de problème (en tout cas je n'ai pas plus de freeze avec)


En écrivant ce post je viens de me rendre compte que cette méthode me permettrait de facilement utiliser le multiprocess si je ne partage que la fameuse liste d'étages via Array ou Value, je vais essayer ça.

Salut,

Que mange "generer_etage(self, id_etage)" ?

Dit autrement, quel est la raison technique qui explique:

Citation:

---

le temps de génération peut être extrêmement long et je ne peux donc pas attendre d'avoir atteint un étage avant de le générer (j'ai déjà atteint les 10 minutes pour un étage)

---

L'autre question est par quoi est appelé:

Code:

---

1 2 3 4

... def get_etage(self, id): # comment j'apelle start (self ou Thread) mais ça marche ... if id == self.id_etage_en_cours: ... self.id_etage_en_cours += 1 # Derniere étage généré, donc on en prévois un nouveau ... return self.etages[id]

---

- W

La raison technique c'est le nombre de calcul à faire pour générer l'étage. En gros un étage comprend un nombre de salle variable et leur génération fait énormément appel à random, ce qui la rend plutôt longue ... A niveau de "perception humaine", le temps de chargement ne se sent pas avant 10 salles (et encore, on sent juste que c'est pas instantané, mais on attend pas non plus) mais ayant un peut testé le résultat, j'ai remarqué qu'on se retrouvais vite aux étage à plus de 50 salles et là les temps de chargements commencent vraiment à être long (en sachant que le but est de n'avoir aucune limite sur le nombre de salle, même si c'est impossible).

J'ai une classe qui gère la gestion de la map (son affichage, les case accessibles ou non, les entités, etc.) et quand on change d'étage, je fait un bête "self.grid = self.generateur_de_map.get_etage(self.etage)". La variable "generateur_de_map" est initialisée dans le constructeur et n'est utiliser que pour la fonction "get_etage", c'est le seul et unique accès que je lui donne (en dehors du destructeur)

Salut,

Quand je vous demande ce que mange "generer_etage", c'est parce s'il n'y pas d’entrées sorties (réseau ou disques ou des appels à sleep), on peut supposer que çà calcule... Et pour faire aller plus vite des calculs, il faut soit change l'algo. soit utiliser plus de CPU.

En général, les threads sont le moyen qui permet d'utiliser plus de CPU, avec Python ce n'est pas le cas: il faut passer par multiprocessing, des interpréteurs séparés. Les "mémoires" étant alors privées (avec les threads, un interpréteur, la mémoire est commune) les échanges de données se font via messages.

Tout cela est compliqué et çà ne se raconte pas en quelques phrases. Pas la peine d'entrer dans les détails, il y a déjà plein d'écrits, documentations, exemples,... que vous pourrez lire plus tard.

Le vrai sujet est "comment vous permettre de répartir le boulot entre plusieurs CPU (via multiprocessing) sans trop vous prendre les pieds dans le tapis et qu'on y passe/perde trop de temps.

J'ai compris que ce qu'il faut exécuter "à part" est la fonction "generer_etage".
Lorsque le calcul d'un étage est terminé, on récupère le résultat dans le thread qui a lancé cela.
Ce modèle de distribution du boulot est assez bien pris en charge par le module concurrent.futures.
note: Si votre version de Python est >= 2.6 et <= 3.1, il faut l'installer via PyPI, il ne vient avec Python qu'à partir de 3.2+
Regardez comment fonctionnent les exemples de la documentations.
Vous devriez pouvoir adapter votre code assez facilement.

- W

Ha oui, j'ai oublié de préciser que je suis en 3.3

Citation:

---

Envoyé par wiztricks

Quand je vous demande ce que mange "generer_etage", c'est parce s'il n'y pas d’entrées sorties (réseau ou disques ou des appels à sleep), on peut supposer que çà calcule... Et pour faire aller plus vite des calculs, il faut soit change l'algo. soit utiliser plus de CPU.

---

Le temps de calcul ne me gène absolument pas, du moment qu'il est fait en parrallèle.

Citation:

---

Envoyé par wiztricks

J'ai compris que ce qu'il faut exécuter "à part" est la fonction "generer_etage".
Lorsque le calcul d'un étage est terminé, on récupère le résultat dans le thread qui a lancé cela.
Ce modèle de distribution du boulot est assez bien pris en charge par le module concurrent.futures.

---

J'ai essayer de l'imbriquer mais je me retrouve avec un petit soucis :s
J'ai essayer 2 version, la première c'est de faire un "submit", mais il prend du temps :

Code:

---

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18

>>> from concurrent.futures import ProcessPoolExecutor >>> class Foo: ... def __init__(self, x): ... self.x = x # Valeur d'exemple ... self.add = 4 # Valeur du prochain Foo à ajouter ... def run(self): ... # Création de l'executeur ... with ProcessPoolExecutor(max_workers = 1) as executor: ... # Ajout de l'action ... executor.submit(self.new_foo, self.add) ... def new_foo(self, x): # Fonction qui attend un moment avant de créer un Foo ... for i in range(100000000): ... pass ... return Foo(x) ... >>> foo = Foo(1) >>> foo.run() # J'attend un moment ici, ce que je cherche à éviter

---

Et la 2e c'est de faire le submit dans un Thread (de 2 manière différentes : héritage avec Foo(Thread) et appel via Thread(target=self.run) )

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

from concurrent.futures import ProcessPoolExecutor from threading import Thread class SubFoo: def __init__(self): self.x = 5   class Foo(Thread): def __init__(self, x): Thread.__init__(self) self.x = x self.subs = [] def run(self): self.stop = False self.add = 4 while not self.stop: if self.add: with ProcessPoolExecutor(max_workers = 1) as executor: foo = executor.submit(self.new_foo, ) self.subs.append(foo.result()) self.add = 0 def new_foo(self, x): return SubFoo(x)   f = Foo(1) f.start()

---

Ce qui génère cette erreur :

Citation:

---

Traceback (most recent call last):
File "/usr/lib/python3.3/multiprocessing/queues.py", line 249, in _feed
send(obj)
File "/usr/lib/python3.3/multiprocessing/connection.py", line 206, in send
ForkingPickler(buf, pickle.HIGHEST_PROTOCOL).dump(obj)
_pickle.PicklingError: Can't pickle <class '_thread.lock'>: attribute lookup _thread.lock failed

---

Personnellement, je pensais que la 1er solution fonctionnerait, mais je doit attendre que la fonction finisse pour reprendre la main ...
Pour la 2e j'ai pas trouvé de solution, le Pickle à l'air d'être incapable de faire passe un trhead.lock :s

Je viens aussi de revérifier et j'ai bien des lag avec ce seul thread et pas de lag quand je le met en classe 'non-parrallèle' (en revanche j'ai des temps de chargement que je cherche à éviter)

Salut,
Pour apprendre à utiliser concurrent.futures, il sera plus simple de commencer par oublier classes et threading: vous distribuez du boulot. Et matérialiser ce boulot par une fonction évitera bien des soucis pour que le module puisse bosser pour vous en vous permettant de ne pas trop vous prendre le chou.

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

import concurrent.futures from concurrent.futures import ProcessPoolExecutor import threading import time import os   class Bidon: def __init__(self, value): self.value = value self._pid = os.getpid()   def __str__(self): return 'bidon: value=%d, pid=%s' % (self.value, self._pid)   def create_bidon(value): time.sleep(0.05) return Bidon(value)   if __name__ == '__main__': def do_work(): with ProcessPoolExecutor(max_workers = 2) as executor: futures = { executor.submit(create_bidon, x): x for x in range(8) } for fs in concurrent.futures.as_completed(futures): b = futures[fs] r = fs.result() print ("1 * %d = %s" % (b, r))   p = threading.Thread(target=do_work) p.start() p.join()

---

- W

Ça fonctionne, merci ^^
Pour ceux qui voudrait récupéré les résultats de ces actions parallèles :

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

from concurrent.futures import ProcessPoolExecutor import threading import time   class Foo: def __init__(self, x): self.x = x # Value example self.thread_results = {} # Results of your parallel function self.threads = {} # Thread with ProcessPoolExecutor # Kill thread (but waiting the end of this thread) def __del__(self): for t in self.threads.values(): t.join() # Return the value of "self.thread_results[v]" and change it def get(self, v = 0): if v in self.threads: self.threads[v].join() # Wait the end of the actual thread g = self.thread_results # Launch the parallel function with your result-dict and your value (this change next value of self.thread_results[v]) self.threads[v] = threading.Thread(target=do_work, args=(self.thread_results, v)) self.threads[v].start() # If you haven't value in self.thread_results[v], relaunch the function if v not in g: return self.get(v) # Return the value return g[v]   # Launch function "new_foo" with argument 'v' in a different process def do_work(result_dict, v): with ProcessPoolExecutor(max_workers = 2) as executor: fs = executor.submit(new_foo, v) # Launch r = fs.result() # Get result #print ("1 * %d = %s" % (b, r)) result_dict[v] = r # Save result   # Wait 2 seconds and return a new instance of Foo def new_foo(x): time.sleep(2) return Foo(x)   if __name__ == '__main__': f = Foo(1) print(f.get()) print(f.get(2)) print(f.get(2)) print(f.get(1)) # Try to do other "f.get(X)" in terminal, you haven't waiting time if thread is finish ;)

---

Il reste encore des soucis avec le pickling mais cette fois à cause des classes et pas du thread, donc quand j'aurais le temps je l'enlèverais par moi même avec get et set state (si j'enlève cette classe tout fonctionne bien).

Merci pour votre aide ^^