threading et écouteur d'événements [Résolu]

[zerros]

Bonjour,

Je suis en train de tester le threading sur python, et je rencontre quelque difficultés. Mon test est simple:

Je voudrai qu'un thread fasse des insertions dans une base pendant que mon programme principal affiche ce qui a été introduit.

Voilà en gros le code que je teste (j'ai viré toute la partie bdd car ça fonctionne bien sur cette partie).

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
#!/usr/bin/env python
 
import sys
import os
import subprocess
import re
import signal
import MySQLdb
import threading
import multiprocessing
import time
 
class Affiche2(threading.Thread):
        def __init__(self, nom = ''):
                threading.Thread.__init__(self)
                self.nom = nom
                self._stopevent = threading.Event( )
        def run(self):
                ''' FAIS DES INSERTIONS EN BASE '''
        def stop(self):
                self._stopevent.set( )
 
''' main '''
''' appelle la fonction de connexion '''
conn = connectDB()
 
print "Daemon ready to work"
 
c = Affiche2('Thread C')
c.start()
time.sleep(6.5)
c.stop()
 
while (1):
     ''' FAIS DES SELECT ET AFFICHE LE RESULTAT '''

Voilà ce que ça donne en gros. Le prob est que quand le thread démarre, il ne rend pas la main tant que les insertions ne sont pas terminés. or je voudrai le faire tourner en parallèle du programme principal.

Auriez-vous une piste à me filer ?

[VinsS]

Salut,

C'est fait pour tourner conjointement avec ton code principal, justement.

Et ton code fonctionne très bien d'ailleurs:

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
 
import threading
import time
 
class Affiche2(threading.Thread):
    def __init__(self, nom=''):
        threading.Thread.__init__(self)
        self.nom = nom
        self.is_alive = True
 
    def run(self):
        while self.is_alive:
            time.sleep(1)
            print 'Et pourtant je tourne'
 
 
print "Daemon ready to work"
 
c = Affiche2('Thread C')
c.start()
count = 5
while count:
    print count
    time.sleep(1)
    count -= 1
 
c.is_alive = False
time.sleep(2)

Où est le problème ?

[wiztricks]

Salut,
Le partage d'une même connexion à la BDD entre plusieurs threads est spécifique à l'API (MySQLDB) tant côté support que mise en œuvre.
- W

[zerros]

En fait, pour la partie bdd, je ne m'en fais pas. à la limite, je re-crée une nouvelle connexion dans le thread, et zou ...

Ce que je ne comprends pas, c'est qu'au début du programme, il me dit bien que les threads ont été démarrés. Mais le reste du programme n'est pas exécuté tant que les threads ne sont pas terminés.

Par exemple, si dans mon __main__ j'ai 2 start() pour démarrer 2 threads et que juste après je fais un print, le print ne s'affiche pas tant que les threads ne sont pas terminés.

Par contre, c'est vrai que les 2 threads s'exécutent simultanément. Si je comprends bien, il faut donc que mon thread qui insère et celui qui fait les select tournent en parallèle et non, juste un thread pour l'insert et le select dans le __main__ après le start() du thread d'insertion ?

Sinon, il va falloir que je buche sur le partage de ressource entre 2 threads car les 2 threads, à un moment donné, devront utiliser le même port série. Ne pouvant pas ouvrir 2 fois le port, je serai obligé de l'ouvrir dans le programme principal, et de partager mon objet sur les 2 threads. C'est possible ça ?

[VinsS]

Dans la question de départ il s'agit d'un seul thread.

Ce thread ne fonctionne pas comme attendu parce qu'il est mal utilisé.

1. Un thread doit être conçu comme une tâche unique.
2. Le code principal ne doit rien en attendre.
3. Il devrait (je ne dis pas il doit) disposer de toutes ses resources avant d'être lancé.
4. il faut le laisser mourir de sa mort naturelle.

Le code exemple de mon post précédent remplit exactement la tâche décrite dans la question initiale.

J'invite zerros à relire sa question.

[zerros]

ok, après avoir fait quelques tests avec le code de VincsS (merci), le fonctionnement me convient, sauf que mes 2 threads doivent tourner en boucle. Du coup voici le code que j'ai fait, mais je suis encore confronté à plusieurs soucis:

1 - Le CPU monte d'un coup dès que je lance le démon
2 - Impossible d'arrêter les threads sans killer le process 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
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
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
#!/usr/bin/env python2.7
 
import sys
import os
import subprocess
import re
import signal
import MySQLdb
import threading
import time
 
##########################################################################
# class
##########################################################################
 
class watchSerial(threading.Thread):
	''' thread init '''
	def __init__(self, name = ''):
		threading.Thread.__init__(self)
		self.name = name
		self._stopevent = threading.Event()
 
	''' main process of the thread '''
	def run(self):
		while not self._stopevent.isSet():
			conn = connectDB()
			results = execSQL(conn,'SELECT * FROM evt WHERE type="Receive" ORDER BY tstamp')
			for row in results:
				idEvt = row["idEvt"]
				action = row["action"]
				hardName = row["hardName"]
				idHard = row["idHard"]
				hardKey = row["hardKey"]
				serial = row["serial"]
				cmd = row["cmd"]
 
				print "[" + str(serial) + "] found evt " + action
				print "[" + str(serial) + "] Deleting evt " + str(idEvt)
				execSQL(conn,'DELETE FROM evt WHERE idEvt=' + str(idEvt))
 
			closeDB(conn)
 
	''' stop the thread '''
	def stop(self):
		self._stopevent.set()
 
class watchEvt(threading.Thread):
	''' thread init '''
	def __init__(self, name = ''):
		threading.Thread.__init__(self)
		self.name = name
		self._stopevent = threading.Event()
 
	''' main process of the thread '''
	def run(self):
		while not self._stopevent.isSet():
			conn = connectDB()
 
			results = execSQL(conn,'SELECT * FROM evt WHERE type="Send" OR type="Ring" OR type="Sys" ORDER BY tstamp')
			for row in results:
				idEvt = row["idEvt"]
				action = row["action"]
				hardName = row["hardName"]
				idHard = row["idHard"]
				hardKey = row["hardKey"]
				serial = row["serial"]
				cmd = row["cmd"]
 
				if (action == "ring"):
					print "[" + str(serial) + "] found evt " + action
					res = execSQL(conn,'SELECT * FROM conf WHERE idConf=1')
					for conf in res:
						appPath = conf['appPath']
						ringFile = conf['ringFile']
 
					ring(appPath,ringFile,serial)
 
				if (action == "sys"):
					print "[" + str(serial) + "] found evt " + action
					execSys(cmd,serial)
 
				print "[" + str(serial) + "] Deleting evt " + str(idEvt)
				execSQL(conn,'DELETE FROM evt WHERE idEvt=' + str(idEvt))
 
			closeDB(conn)
			''' self._stopevent.wait(1.0) '''
 
	''' stop the thread '''
	def stop(self):
		self._stopevent.set()
 
##########################################################################
# functions
##########################################################################
 
''' execute a system command on host '''
def execSys(sysCmd,serial):
	print "[" + str(serial) + "] Executing sys cmd:  " + str(sysCmd)
	os.system(sysCmd)
 
''' execute ring bell  '''
def ring(appPath,ringFile,serial):
	conf = appPath + "/config/mplayer.conf"
	file = appPath + "/ringtones/" + ringFile;
	ret = subprocess.Popen("ps -aef | grep mplayer | grep -v grep | awk '{print $2}'", shell=True, stdout=subprocess.PIPE)
	ps = ret.stdout.read() . rstrip('\n')
	ret.stdout.close()
	val = re.search(r"(\d+)", ps)
	if val:
		print "[" + str(serial) + "] Already ringing ..."
	else:
		os.system("mplayer -include " + conf + " " + file + " &")
		print "[" + str(serial) + "] Ringing"
 
''' connect to DB '''
def connectDB():
	try:
		conn = MySQLdb.connect(
			host = "127.0.0.1",
			user = "xxxx",
			passwd = "xxxx",
			db = "xxxx"
		)
		return conn
	except MySQLdb.Error, e:
		print "Error %d: %s" % (e.args[0], e.args[1])
		sys.exit(1)
 
''' execute sql queries '''
def execSQL(conn,query):
	try:
		cursor = conn.cursor(MySQLdb.cursors.DictCursor)
		cursor.execute(query)
		rows = cursor.fetchall()
		cursor.close()
		return rows
	except MySQLdb.Error, e:
		print "Error %d: %s" % (e.args[0], e.args[1])
		sys.exit(1)
 
''' disconnect from DB '''
def closeDB(conn):
	conn.close()
 
##########################################################################
# main program
##########################################################################
 
if __name__=='__main__':
	try:
		watchEvtT = watchEvt('watchEvt')
		watchSerialT = watchSerial('watchSerial')
		watchEvtT.start()
		watchSerialT.start()
	except KeyboardInterrupt:
		print "Ctrl-C pressed, exiting"
		watchEvtT.stop()
		watchSerialT.stop()
		sys.exit()

Le watchEvt récupère des événements de type 'Sys', 'Ring' et 'Serial' et les exécute (pour le moment je n'ai fait que le ring et sys) et le watchSerial devra au final vérifier ce qui arrive sur une liaison serial et faire des insertions de type "Receive" dans la table des événements. N'ayant pas encore préparer la partie serial, je fais juste un insert de type "Receive" dans watchEvt et un select puis un drop dans watchSerial.

Pourquoi est-ce que lorsque je fais CTR-C le programme ne s'arrête pas ? Pourquoi est-ce que le programme prend autant de ressources lorsqu'il tourne ? Il tourne aux alentours de 60% et monte même parfois à 99% de load cpu.

Ce démon python tourne sur une raspberry pi doté d'un CPU arm.

J'espère que vous pourrez m'aiguiller sur ce que je fais de mal ...

[wiztricks]

Envoyé par zerros

Pourquoi est-ce que lorsque je fais CTR-C le programme ne s'arrête pas ? Pourquoi est-ce que le programme prend autant de ressources lorsqu'il tourne ? Il tourne aux alentours de 60% et monte même parfois à 99% de load cpu.

Tel que vous avez écrit cela les "control-C" sont pris en compte jusqu'au retour des thread.start... après, le programme attend que les threads se termine mais il n'y a rien pour attraper "control-C".

Exemple:

Code :

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
import threading
import time
import sys
 
def waitforever():
    while True:
        print "waitforever"
        time.sleep(1)
 
if __name__ == '__main__':
    th = threading.Thread(target=waitforever)
    th.daemon = True  
    th.start()
 
    while True:
        try:
            time.sleep(0.5)
        except KeyboardInterrupt:
            print 'KeyboardInterrupt'
            break
 
    print ('done')

Le programme principal attend un eventuel control-C.
th.daemon = True permet de sortir du programme: il n'y a pas de mécanique pour arrêter la thread.

Sans pause entre les différents passages, çà bouffe du CPU inutilement: on cherche dans la base de donnée sans rien trouver et on boucle.

Le vrai sujet est pourquoi utiliser des threads:
- çà ne sert à rien dans votre cas: les activités sont synchrones(*),
- comme vous ne savez pas les utiliser vous accumulez les dépendances entre des difficultés: interroger le SGDB, lancer des taches, mettre cela en threads,...

(*) watchSerial n'est qu'un cas particulier de watchEvt: plutôt que lancer une commande on détruit les event de type "Receive".

- W

[zerros]

En fait le démon sera asynchrone au final. Actuellement, pour faire les tests j'envoi un insert depuis le watchEvt pour que le select de watchSerial puisse le "poper" et le détruire. Je fais ça car je n'ai pas encore fait la partie serial.

Au final, quand un événement sera récupérer dans watchEvt, notamment si c'est un événement de type serial, une donnée sera écrite dans un serial (/dev/ttyUSB0). L'appareil connecté sur ce serial va exécuter une action, puis renvoyer un status qui sera "traper" par le watchSerial et inséré en base avec le type Receive. Mais le matériel peut recevoir des ordres d'une autre source également, et le status de ces ordres doivent être insérés en base également.

Les threads doivent tourner en boucle pour traiter les événements aussitôt qu'ils arrivent. Dans ce cas là, existe t il un autre moyen d'arriver à mes fins qu'utiliser les threads ?

[zerros]

j'ai ajouté une tempo de 0.5s et le cpu tombe à 3. oufff. 0.5, ça ira très bien pour commencer.

Il me reste plus qu'à trouver une solution pour la partie arrêt des threads, et/ou carrément changer la manière de tout gérer en utilisant autre chose que les threads. Mais quoi ?

[wiztricks]

Envoyé par zerros

Les threads doivent tourner en boucle pour traiter les événements aussitôt qu'ils arrivent. Dans ce cas là, existe t il un autre moyen d'arriver à mes fins qu'utiliser les threads ?

Votre "Aussitot" passe par:

1. mise en base,
2. attente de la fin du traitement des taches lancées par watchEvt (ce qui a été lu 'avant'),
3. la lecture de ce qui est nouveau,

Le délai entre un événement et son traitement est difficile à borner => ce n'est pas du temps réel. Vu que çà passe par des accès disques le temps de réponse (délai) sera toujours supérieur à quelques millisecondes.

Si votre "fin" est de bouffer du temps CPU pour rien, vous y arriverez sûrement avec des threads. Si vous voulez que les délais de traitements soient bornés et inférieurs à 1/10 de milliseconde, il faut revoir le design.

- W

[wiztricks]

Envoyé par zerros

Il me reste plus qu'à trouver une solution pour la partie arrêt des threads, et/ou carrément changer la manière de tout gérer en utilisant autre chose que les threads.

La mécanique que vous avez déjà fonctionne.
Mais vous ne savez pas l'utiliser.
Relisez ce que je vous ai déjà indiqué plus haut.

- W

[tyrtamos]

Bonjour,

Avec les threads, on fait à peu près ce qu'on veut, sauf à les arrêter "sauvagement". Mais la solution proposée ici (avec le while et le stop) fonctionne très bien.

Il y a plusieurs choses qui me gènent dans le dernier code de zerros.

- chaque thread cherche des données (select) et en efface (delete). Il n'est donc pas impossible qu'un thread essaie d'effacer des données qui viennent d'être effacées par un autre thread. Ce n'est pas logique. Même si le serveur accepte les accès concurrents, il faudrait que chaque thread commence et finisse son travail sans qu'un autre thread s'interpose sur les mêmes données: on fait ça avec un verrou!

- les threads peuvent effectivement avoir un fonctionnement permanent au sein d'un programme, mais le fait que la seule façon de les arrêter est un Ctrl-C me choque un peu. Il faudrait, par exemple, que lorsqu'un thread ne trouve plus de données dans son "select", il génère lui-même son arrêt (fin de sa méthode "run").

- dernier point mineur, il ne devrait pas y avoir de "print" dans les threads, parce qu'il n'y a qu'un seul canal d'affichage (sys.stdout) que les threads et le "thread pricipal" doivent se partager. On devrait donc utiliser un verrou.

Voilà un petit programme de test qui prend en compte le 1er point (le verrou sur la base de données):

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
#!/usr/bin/python
# -*- coding: utf-8 -*-
from __future__ import division
# Python 2.7
 
import sys, os, threading, time
 
##########################################################################
class Monthread(threading.Thread):
 
    def __init__(self, name = ''):
        threading.Thread.__init__(self)
        self.name = name
        self.encore = True
 
    def run(self):
        global verroudb
 
        while self.encore:
            verroudb.acquire() # bloque jusqu'à ce que le verrou soit libéré
            #conn = connectDB()
            print
            print "thead %s instruction SELECT" % (self.name,)
            time.sleep(0.1) # simul temps d'execution
            print "thead %s instruction DELETE" % (self.name,)
            #closeDB(conn)
            verroudb.release() # libère le verrou
            time.sleep(0.1) # pour que while ne consomme pas trop de ressources
 
    def stop(self):
        self.encore = False
 
##########################################################################
if __name__=='__main__':
 
    # création d'un verrou pour que chaque thread intervienne seul sur la base
    verroudb = threading.Lock()
 
    # ouverture des threads
    thread1 = Monthread("1")
    thread2 = Monthread("2")
    thread3 = Monthread("3")
    thread4 = Monthread("4")
    thread5 = Monthread("5")
    thread1.start()
    thread2.start()
    thread3.start()
    thread4.start()
    thread5.start()
 
    # boucle d'attente interruptible par Ctrl-C
    try:
        while True:
            time.sleep(0.1)
    except KeyboardInterrupt:
        pass
 
    # fermeture des threads
    thread1.stop()
    thread2.stop()
    thread3.stop()
    thread4.stop()
    thread5.stop()
 
    # on attend que tous les threads soient effectivement arrêtés
    while thread1.isAlive() or thread2.isAlive() or thread3.isAlive() or thread4.isAlive() or thread5.isAlive():
        time.sleep(0.1)
 
    print
    print "fin du programme!"

A l'exécution, il est affiché:

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
...
 
thead 1 instruction SELECT
thead 1 instruction DELETE
 
thead 2 instruction SELECT
thead 2 instruction DELETE
 
thead 3 instruction SELECT
thead 3 instruction DELETE
 
thead 4 instruction SELECT
thead 4 instruction DELETE
 
thead 5 instruction SELECT
thead 5 instruction DELETE
 
thead 1 instruction SELECT
thead 1 instruction DELETE
 
thead 2 instruction SELECT
thead 2 instruction DELETE
 
...

On voit bien que lorsqu'un thread commence par un select, il termine bien par le delete sans qu'un autre thread n'intervienne entre les 2.

L'ordre d'intervention des thread dépend du caractère aléatoire des tâches de la boucle while qui ne se trouverait pas soumis au verrou sur la base de données. Si par contre il n'y a aucune tâche hors du bloc "acquire() ...release()", les threads interviendont toujours dans le même ordre. Dans ce cas, on se demande si les threads sont bien utilies ici...

Et avec un Ctrl-C, le programme arrête proprement les 5 threads, et affiche à la fin que le programme est terminé.

[wiztricks]

Salut,

Envoyé par tyrtamos

Même si le serveur accepte les accès concurrents, il faudrait que chaque thread commence et finisse son travail sans qu'un autre thread s'interpose sur les mêmes données: on fait ça avec un verrou!

Comment fait-on lorsque il s'agit de process indépendants qui tournent sur des machines différentes? Avec les SGDB, on passe par des transactions.

Ceci dit le code du thread watchSerial récupère les entrées de la table evt de type "Receive" alors que watchEvt s'occupe des types "Send", "Ring", "Sys": ce ne sont pas les mêmes données.

Le traitement de watchSerial pourrait être intégré dans watchEvt qui s'occuperait en plus de ce type là.

In fine, on a un thread qui fait tout de façon séquentielle et le programme principal qui attend qu'on tape "control-C".

- les threads peuvent effectivement avoir un fonctionnement permanent au sein d'un programme, mais le fait que la seule façon de les arrêter est un Ctrl-C me choque un peu. Il faudrait, par exemple, que lorsqu'un thread ne trouve plus de données dans son "select", il génère lui-même son arrêt (fin de sa méthode "run").

Le code de zerros arrête les threads proprement: il a associé un Event à chaque thread qu'il teste dans le run avant d'exécuter l'itération suivante de ses traitements.
Ce qui est en défaut, c'est le traitement des "control-C".
J'espère que votre exemple lui permettra d'y voir plus clair.

- W

[zerros]

mouarff. je suis en train de décortiquer avec le manuel python lol.

Je reviens vous dire quand j'aurai compris et appliquer vos conseils

merciii en tout cas. a très vite

[zerros]

merci à vous !!! C'est impeccable. ç marche au poil. J'ai mis 0.5 pour libérer encore un peu plus de ressource, et j'ai utilisé les Event en lisant le tuto de developpez: http://python.developpez.com/faq/?page=Thread

Je reviendrai surement poser des questions sur le partage de ressources entre les threads.

Merccciiiiiii