Discussion :

[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 :

Sélectionner tout - Visualiser dans une fenêtre à part

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 :

Sélectionner tout - Visualiser dans une fenêtre à part

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 :

Sélectionner tout - Visualiser dans une fenêtre à part

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]

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 :

Sélectionner tout - Visualiser dans une fenêtre à part

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