Discussion :

[Alliaël]

Bonjour,

Je développe actuellement un programme utilisant le multithreading et je me heurte a un gros problème.
Lors de la veille de la machine il semble qu'une partie du programme continue, tandis que l'autre ne s'exécute plus.
Cependant ca n'arrive pas a chaque mise en veille, donc j'ai beaucoup de mal a comprendre d'ou ca vient.

Le main est les suivant :

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
 
#!/usr/bin/env python
# encoding: utf-8
 
from DB_Reader_Writer import *
from WL_threading import *
from UsefullFuction import *
from Downloader import *
import threading, Queue
import events_db
import sys, os
import WL_threading
from WL_web_services import INGV_requestor
from events_db import DBEventList
from WL_DB_stuff import *
from velmod import VelocityModel
 
 
#writer
Writer=DB_Writer("waveloc")
#Writer.Create_All_Tables()
 
 
####################  create workers ###########################################
#download
requestsQueue_Download = Queue.Queue()
resultsQueue_Download = Queue.Queue()
workRequests_Download={}
for i in range(1):
  worker_Download = Worker(requestsQueue_Download, resultsQueue_Download)
 
#untar
requestsQueue_Untar = Queue.Queue()
resultsQueue_Untar = Queue.Queue()
workRequests_Untar={}
for i in range(1):
  worker_Untar = Worker(requestsQueue_Untar, resultsQueue_Untar)
 
 
#reader
Reader=DB_Reader("waveloc")
Writer.Create_All_Tables()
####################  create work list ################################
 
Liste_Download_data=Reader.Extract_Waveloc_for_Download(1000)
print "Dowload list : ",Liste_Download_data
 
liste_ID_and_number=[]
liste_fseed_name=[]
################### work ###############################
 
#dowload data
Requestor=Downloader('user@domain.com', 'passeword', data_dir,"waveloc")
 
#launch the download request
for Waveloc_ID in Liste_Download_data :
  id = worker_Download.performWork( Requestor.run_circular_query_and_return_path, Waveloc_ID)
  workRequests_Download[id] = str(Waveloc_ID)
 
 
#wait resultats about download request to untar data and complet the waveloc_data table and station_data table
while workRequests_Download:
  time.sleep(1)
  showResults(resultsQueue_Download,workRequests_Download,liste_ID_and_number)
  ###untar data
  for ID_and_number in liste_ID_and_number :
    if ID_and_number[1] !='False' :
      id = worker_Untar.performWork(Untar_data,data_dir,ID_and_number[1],ID_and_number[0],'True')
      workRequests_Untar[id] = ID_and_number
    liste_ID_and_number.remove(ID_and_number)
    while workRequests_Untar:
      time.sleep(1)
      showResults(resultsQueue_Untar,workRequests_Untar,liste_fseed_name)
    for ID_Fseed in liste_fseed_name :
      #complet waveloc_data table
      fseed_full_filename= ID_Fseed[1][0]
      fseed_filename="%s%s%s"%(fseed_full_filename.split(os.sep)[-2], os.sep, fseed_full_filename.split(os.sep)[-1])
      Writer.put_fseed_data_into_waveloc_db(fseed_filename,ID_Fseed[0][0])
      #complet station_data table
      station_data_list=read_stations_file(ID_Fseed[1][1])
      for station_data in station_data_list:
        Writer.put_station_data_into_waveloc_db(station_data)
      #complet station and waveloc table (table obsolete)
        #station_name_and_net=[(station_data[0],station_data[1])]
	#Writer.populate_waveloc_and_stations_db(ID_Fseed[0][0],station_name_and_net)
      #remove ID_Fseed from the list
      liste_fseed_name.remove(ID_Fseed)

et les travailleur en parallèles suivent ce code (trouvé dans le manuel "python in a nutshell"

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
class Worker(threading.Thread):
  requestID=0
  def __init__(self,requestsQueue,resultsQueue,**kwds):
    threading.Thread.__init__(self, **kwds)
    self.setDaemon(1)
    self.workRequestQueue = requestsQueue
    self.resultQueue = resultsQueue
    self.start()
  def performWork(self,callable,*args,**kwds):
    # called by the main thread as callable would be, but w/o return
    Worker.requestID += 1
    self.workRequestQueue.put((Worker.requestID, callable, args, kwds))
    return Worker.requestID
  def run(self):
    while 1:
      (requestID, callable, args, kwds) = self.workRequestQueue.get()
      self.resultQueue.put((requestID,callable(*args, **kwds)))

Le main va télécharger des données puis en parallèle du prochain téléchargement vas les extraire et les insérer dans une base de donnée.

Cependant il arrive, après une mise en veille, que le téléchargement se poursuive sans que le reste du travail soit effectué. (il n'y a cependant plus de messages envoyés sur la sortie standard)

Ce problème a eut lieu sur une machine sous debian, tout comme sur une machine sous MacOs

[mont29]

Ce problème m’a l’air assez bizarre –*surtout quand on sait que les threads sont «*faux*» en python (ils sont exécuté séquentiellement, à cause du Global Lock…).

Cependant, le “while workRequests_Download:” me semble problématique –*en effet, cela implique que la boucle principale est finie dès que cette queue est vide, or le fait qu’elle soit vide ne signifie pas que tous les téléchargements sont finis, certains peuvent être encore en cours…

Encore une fois, je ne suis pas sûr du tout que ce soit lié, mais à première vue, c’est le truc qui me choque le plus…

[Alliaël]

Merci

La liste workRequests_Download contient un élément par téléchargement et on lui retire un élément a chaque fois qu'un téléchargement se termine (avec la fonction showResults) donc la logique voudrait qu'elle ne soit vide quand il n'y a plus de téléchargement a effectuer.

Est il possible que l'écriture dans la base de donnée soit impossible pendant que la machine est en veille ?

[mont29]

Je me trompe peut-être, mais il me semble que l’élément à télécharger est enlevé de la queue (première instruction de la fonction run() de tes workers) avant le lancement du téléchargement (via l’utilisation du callable, deuxième instruction de la fonction run() ), non*?

Donc workRequests_Download peut me semble-t-il se retrouver vide alors qu’il y a encore des téléchargements en cours… Mais encore une fois, je ne suis pas sûr que ce soit la source de ton problème Il est assez spécifique et difficile à cerner…

[Alliaël]

Oui ils sont retirés de la queue de travail pour être mit dans la queue de résultat. Et c'est seulement s'ils sont dans la queue de résultat et en même temps que l'on renvois le résultat que l'on enlève l'élément de workRequest_Download.

la fonction showResults étant celle ci

Code :

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

1
2
3
4
5
6
7
def showResults(QueueT,workRequests,liste_resultat):
    while 1:
      try : id,results = QueueT.get_nowait()
      except Queue.Empty: return
     # print 'Result %d : %s %s' %(id, workRequests[id], results)
      liste_resultat.append([workRequests[id],results])
      del workRequests[id]

[mont29]

My mistake, c’est bien ce que je craignais… J’avais confondu requestsQueue_Download et workRequests_Download…

Du coup, je ne vois pas trop… Mon “best guess” serait que la fonction de téléchargement (Requestor.run_circular_query_and_return_path, si je ne me plante pas, cette fois ) ne supporte pas la mise en veille, autrement dit que le téléchargement se produise, mais sans renvoyer/capter le signal qu’il est fini… Enfin, je nage, quoi

As-tu placé des print un peu partout, pour essayer de voir ce qui se passe quand ton erreur se produit*? Tu dis que les téléchargements se poursuivent, mais est-ce toujours visible dans les queues concernées*?

[kango]

surtout quand on sait que les threads sont «*faux*» en python (ils sont exécuté séquentiellement, à cause du Global Lock…)

hello,

un cpu donne l'illusion de traiter des opérations en parallèle en leur donnant la main de manière séquentielle.

la particularité des thread Python (et ce à cause du Global Lock), c'est qu'ils ne sont exécutés que par le CPU en charge du process python parent et ne tire donc pas avantage des autres CPU dans le cadre d'une machine multi-coeurs.

une architecture multi-thread en Python ne permet donc pas de gagner en performances brutes mais essentiellement en "réactivité".

l'utilisation de process (via la librairie standard multiprocesing) permet de s'affranchir de cette limitation, au prix... d'autres problèmes :p

[wiztricks]

Salut,

une architecture multi-thread en Python ne permet donc pas de gagner en performances brutes mais essentiellement en "réactivité".

ce n'est pas si simple.
En fait l’interpréteur Python vole au dessus de bibliothèques C/C++ ou autres qui peuvent "bypasser" cette limite dans la "pratique".
Essayez de faire du multithreading avec une BDD de type PostgreSQL par exemple.

Lors de la veille de la machine il semble qu'une partie du programme continue, tandis que l'autre ne s'exécute plus.
Cependant ca n'arrive pas a chaque mise en veille, donc j'ai beaucoup de mal a comprendre d'ou ca vient.

Passer "en veille" est un état système que subira votre code Python comme toutes les autres 'activités'... Reste au système à gérer la transition et là çà peut dépendre des IO en cours, et autres interruptions qui infligeront un délai variable au passage 'en pause'.

- W

[Alliaël]

Merci pour vos réponses.

Je suis encore entrain de débug ce programme mais visiblement la mise en veille n'était qu'un leurre. L'erreur arrive une fois de temps en temps et comme le programme tournait surtout pendant mon absence.

Pour le multi-thread en python, on m'a conseillé de faire plutôt du multi-process est-ce plus intéressant en performance ?(sur une machine a plusieurs coeurs et avec Mysql en BDD)