Comment Modifier un membre via des Process

Version imprimable

Bonjour à tous,
En partant d'exemples tirés de "Python essential reference" de Beazley, je tache de modifier un dictionnaire, membre d'une classe via des Process lancés depuis cette même classe et...une usine à gaz.
Le Trace back est tout à fait cryptique pour moi et j'aurais besoin d'une traduction "pour les nuls".
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 #!/usr/bin/python # *-* coding UTF-8 *-* from multiprocessing import Process, JoinableQueue, Lock , Manager class QTest(): def __init__(self): self.m=Manager() self.l1=Lock() self.l2=Lock() self.l3=Lock() self.dl={'a':self.l1,'b':self.l2,'c':self.l3,'d':self.l1,'e':self.l2,'f':self.l3,'g':self.l1,'h':self.l2,'i':self.l3,'j':self.l1} self.d=self.m.dict() self.q=JoinableQueue() self.cons1=Process(target=self.consumer,args=(self.q,)) self.cons1.deamon=True self.cons2=Process(target=self.consumer,args=(self.q,)) self.cons2.deamon=True self.prod=Process(target=self.producer,args=(['a','b','c','d','e','f','g','h','i','j'],self.q)) self.prod.deamon=True self.cons1.start() self.cons2.start() self.prod.start() self.q.join() def consumer(self,iq): while True: item=iq.get() if item is None: break self.dl[item].acquire() self.d[item]="done" self.dl[item].release() iq.task_done() print "consumer done" print self.d def producer(self,seq,oq): for item in seq: oq.put(item) oq.put(None) oq.put(None) if __name__=='__main__': qt=QTest()
Dans le cas improbable où le code ne produirait pas d'erreur, voici un extrait du trace back:

Citation:

Process Process-3:
Traceback (most recent call last):
File "/usr/lib/python2.6/multiprocessing/process.py", line 232, in _bootstrap
self.run()
File "/usr/lib/python2.6/multiprocessing/process.py", line 88, in run
self._target(*self._args, **self._kwargs)
File "./T3.py", line 34, in consumer
self.d[item]="done"
File "<string>", line 2, in __setitem__
File "/usr/lib/python2.6/multiprocessing/managers.py", line 722, in _callmethod
self._connect()
File "/usr/lib/python2.6/multiprocessing/managers.py", line 709, in _connect
conn = self._Client(self._token.address, authkey=self._authkey)
File "/usr/lib/python2.6/multiprocessing/connection.py", line 143, in Client
c = SocketClient(address)
File "/usr/lib/python2.6/multiprocessing/connection.py", line 263, in SocketClient
s.connect(address)
File "<string>", line 1, in connect
error: [Errno 2] No such file or directory

En modifiant le code comme suit ça va déjà mieux mais les modifications du dictionnaire ne sont plus prisent en compte en dehors des Process. Pourquoi?

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
#!/usr/bin/python
 
# *-* coding UTF-8 *-*
 
 
from multiprocessing import Process, JoinableQueue, Lock#suppression du Manager
 
class QTest():
	def __init__(self):
		self.l1=Lock()
		self.l2=Lock()
		self.l3=Lock()
		self.dl={'a':self.l1,'b':self.l2,'c':self.l3,'d':self.l1,'e':self.l2,'f':self.l3,'g':self.l1,'h':self.l2,'i':self.l3,'j':self.l1}
		self.d={}#modif
		self.q=JoinableQueue()
		self.cons1=Process(target=self.consumer,args=(self.q,))
		self.cons1.deamon=True
		self.cons2=Process(target=self.consumer,args=(self.q,))
		self.cons2.deamon=True
		self.prod=Process(target=self.producer,args=(['a','b','c','d','e','f','g','h','i','j'],self.q))
		self.prod.deamon=True
		self.cons1.start()
		self.cons2.start()
		self.prod.start()
		self.q.join()
 
	def consumer(self,iq):
		while True:
			item=iq.get()
			if item is None:
				break	
			self.dl[item].acquire()
			self.d[item]="done"
			self.dl[item].release()
			iq.task_done()
		print "consumer done"
		print self.d
 
	def producer(self,seq,oq):
		for item in seq:
			oq.put(item)
		oq.put(None)
		oq.put(None)
 
if __name__=='__main__':
	qt=QTest()
	print qt.d #--->{}

22/07/2010, 10h27
wiztricks

Salut,

Citation:

En partant d'exemples tirés de "Python essential reference" de Beazley, je tache de modifier un dictionnaire, membre d'une classe via des Process lancés depuis cette même classe et...une usine à gaz.

Si ce design n'est pas imposé dans le cadre d'un exercice scolaire, vos difficultés à le décrire montre qu'il n'est pas achevé/fini/stable...

Comme le code ne sera que le reflet de vos pensées: ce sera GiGO(*)!
Si on dit: "Comment synchroniser les mises à jour d'un "dict" partagé par pkusieurs threads/process?"
L'objet partagé est une sous classe d'un dict.
Que faut-il connaitre des accédant pour que çà fonctionne?
- W
(*) garbage in, garbage out

Bonsoir,
J'ai beau méditer sur votre réponse, je n'avance pas d'un iota...
J'ai donc écrit un code plus simple pour cerner les problèmes un par un:
Il s'agit de deux classes: miniF instancie une liste l de miniQ qui par leurs méthodes send() poussent dans un JoinableQueue leurs index dans l.
Par ces index, miniF modifie leurs valeurs self.__state.
La valeur __state semble bien modifiée durant toute l'opération mais aussitôt l'appelle de miniF.stop(), une relecture de l montre les états des miniQ inchangés.
Est-ce que tout les passages de valeur ou d'argument se font bien par référence en python?
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 #!/usr/bin/python # *-* coding UTF-8 *-* from multiprocessing import JoinableQueue, Process class miniQ(): def __init__(self,idx,q,state='undone'): self.__idx=idx self.q=q self.__state=state def __repr__(self): ret="miniQ("+str(self.idx)+";"+self.state+")" return ret def send(self): self.q.put(self.__idx) @property def state(self): return self.__state @property def idx(self): return self.__idx @state.setter def state(self,sym): self.__state=sym print self.__state class miniF(): def __init__(self): self.q=JoinableQueue() self.p=Process(target=self.recv) self.p.deamon=True self.l=[] for i in xrange(10): self.l.append(miniQ(i,self.q)) def recv(self): while True: idx=self.q.get() self.l[idx].state='done' print self.l[idx] self.q.task_done() def launch(self): self.p.start() for mQ in self.l: mQ.send() def read(self): for mQ in self.l: print mQ def stop(self): self.q.join() print "miniF stop..." for mQ in self.l: print mQ self.p.terminate() if __name__=='__main__': mf=miniF() mf.launch() mf.stop()
Je crois avoir compris que les modifications apportées au membre self.l "n'existent" qu'à l'intérieur du process qui les fait.
D'où votre réponse en forme de question:

Citation:

Que faut-il connaitre des accédant pour que çà fonctionne?

Suis-je sur la bonne voie? le pid du process peut-il m'aider et dans ce cas comment?

22/07/2010, 22h19
wiztricks

Salut
Vous avez changé le sujet: disparu le dictionnaire partagé, nous voilà avec des process qui consomment une liste de taches.
Pourquoi pas mais si l'objectif n'est pas défini, il sera difficile à atteindre.
1. Décrivez succinctement ce que vous cherchez à faire
2. Utilisez des threads à la place de Process. Ce n'est peut être pas ce que vous voulez, mais en attendant cela réduit le nombre de difficultés empilées.
Note: Les problèmes sont:
- synchronisation des accès,
- partage des objets,
Threading rend 2 transparent. C'est déjà çà.
- W
22/07/2010, 23h37
Krishna

Bonsoir,
En fait j'aimerais créer une classe A() qui instancie des objets B() dans un conteneur permettant leur indexation (dictionnaire,list...). Et permettre à A() de modifier ses objets B() en fonction des messages qu'ils envoient vers A().
Mon problème est la sauvegarde des ces modifications. J'aimerais comprendre pourquoi, en utilisant un JoinableQueue dans mes différents essais, les objets B() "reviennent" toujours à leur état initial.

Salut,

Si vous ne pouvez être plus précis, on peut raisonnablement faire ce qu'on veut...

Ci dessous, vous avez un dict partagé ou les clés sont associées à des objets "partagés". J'utilise la fonctionnalité Manager de Multiprocessing,
elle est "plus simple" que la mémoire partagée...
Le processus principal crée le manager et des workers.
Les workers se contentent d'afficher le contenu du dict pendant que le processus principal le met à jour de façon aléatoire.
Les mises à jour devrait pouvoir être asymétriques mais, pas le temps de construire un scénario de test.
- W

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 multiprocessing import Process, Manager, current_process
import time
import random
 
import sys
def debug(s):
    print >>sys.stderr, s
 
def do_work(ns, shared_dict, pollint):
    p = current_process()
    while not ns.all_done:
        debug('%d: sd= %s' % (p.pid, ', '.join([
                '%s: %s' % (k, v) for k, v in shared_dict.items() ])))
        time.sleep(pollint)
 
class SharedObject(object):
    def __init__(self, value):
        self._value = value
    def __str__(self):
        return 'so: %s' % self._value
 
if __name__ == '__main__':
    COUNT = 5
    WORKERS_COUNT = 2
    manager = Manager()
 
    ns = manager.Namespace()
    ns.all_done = False
    shared_dict = manager.dict()
 
    workers = [ Process(target=do_work, args=(ns, shared_dict, 0.05, )) \
                    for x in range(WORKERS_COUNT) ]
    [ w.start() for w in workers ]
    time.sleep(1.5) # let starts workers
 
    for x in range(COUNT):  # applies random update on dict
        k, v= random.choice(('a', 'b', 'c')), random.randint(0, 9)
        shared_dict[k] = SharedObject(v)
        time.sleep(0.1)
 
    debug('setting ns.all_done to True')
    ns.all_done = True
    debug ('--- ns: %s' % ns)
 
    [ w.join() for w in workers ]
    manager.shutdown()
    debug('done')

24/07/2010, 14h35
Krishna

Waou....
C'est vraiment du beau code.
Désolé pour le manque de précision de ma question mais je ne peux envisager un problème qu'à hauteur de mes compétences...
J'ai maintenant beaucoup de choses sur lesquelles réfléchir et travailler!!
A bientôt peut-être avec une question (je l'espère) plus précise.
:merci: