src/KERNEL_PY/kernel/threadhelper.py

   1 # -*- coding: iso-8859-1 -*-
   2 # Copyright (C) 2007-2023  CEA/DEN, EDF R&D, OPEN CASCADE
   3 #
   4 # Copyright (C) 2003-2007  OPEN CASCADE, EADS/CCR, LIP6, CEA/DEN,
   5 # CEDRAT, EDF R&D, LEG, PRINCIPIA R&D, BUREAU VERITAS
   6 #
   7 # This library is free software; you can redistribute it and/or
   8 # modify it under the terms of the GNU Lesser General Public
   9 # License as published by the Free Software Foundation; either
  10 # version 2.1 of the License, or (at your option) any later version.
  11 #
  12 # This library is distributed in the hope that it will be useful,
  13 # but WITHOUT ANY WARRANTY; without even the implied warranty of
  14 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  15 # Lesser General Public License for more details.
  16 #
  17 # You should have received a copy of the GNU Lesser General Public
  18 # License along with this library; if not, write to the Free Software
  19 # Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307 USA
  20 #
  21 # See http://www.salome-platform.org/ or email : webmaster.salome@opencascade.com
  22 #
  23
  24 __author__="gboulant"
  25 __date__ ="$1 avr. 2010 18:12:38$"
  26
  27 import time
  28 import threading
  29
  30 # ===========================================================================
  31 class Runner(threading.Thread):
  32     """
  33     This class provides a tool to run and drive a function in a dedicated thread.
  34     """
  35     def __init__(self, functionToRun=None,*argv):
  36         threading.Thread.__init__( self )
  37         self._stopevent = threading.Event()
  38         self.setFunction(functionToRun)
  39         self.setArguments(*argv)
  40         self._exception = None
  41         self._return    = None
  42         self._observer  = None
  43         self._callbackFunction = None
  44         # We create an id from the name and the time date in milliseconds
  45         self._id = functionToRun.__name__ +"/"+ str(time.time())
  46
  47     def setFunction(self,functionToRun):
  48         """
  49         Positionne la fonction � ex�cuter. La fonction peut �tre la
  50         m�thode d'un objet pass�e sous la forme 'monobjet.mamethode'.
  51         """
  52         self.clear()
  53         self._function = functionToRun
  54
  55     def setArguments(self,*argv):
  56         """
  57         Positionne les arguments � passer � la fonction
  58         """
  59         self.clear()
  60         self._argv = argv
  61
  62     def getReturn(self):
  63         """
  64         Retourne le resultat de la fonction. En cas d'erreur, on
  65         r�cup�rera l'exception lev�e au moyen de la m�thode
  66         getException().
  67         """
  68         return self._return
  69
  70     def setCallbackFunction(self, callbackFunction):
  71         self._callbackFunction = callbackFunction
  72
  73     def setObserver(self, observerToNotify):
  74         """
  75         Permet de sp�cifier un observateur � notifier en fin
  76         d'ex�cution. L'observateur est suppos� �tre un objet qui
  77         impl�mente une m�thode processNotification()
  78         """
  79         try:
  80             observerToNotify.processNotification
  81         except AttributeError:
  82             raise DevelException("The observer should implement the method processNotification()")
  83
  84         self._observer = observerToNotify
  85
  86     def run(self):
  87         """
  88         Ex�cution de la fonction. Impl�mentation de la m�thode run
  89         d�clench�e par l'appel � Thread.start().
  90         """
  91         print("##################### threadhelper.run")
  92         if self._function is None: return
  93         try:
  94             self._return = self._function(*self._argv)
  95         except Exception as e:
  96             print(e)
  97             self._exception = e
  98         self._stopevent.set()
  99         self.notifyObserver()
 100         self.callback()
 101         # _GBO_ Maybe it's no use to have both observers and a callback function.
 102         # One of the two mechanism should be sufficient
 103
 104     def notifyObserver(self):
 105         if self._observer is None:
 106             # Aucune notification pr�vue
 107             return
 108         try:
 109             self._observer.processNotification()
 110         except AttributeError as att:
 111             if str(att) == "processNotification":
 112                 print("L'observateur n'impl�mente pas la m�thode processNotification()")
 113             else:
 114                 print("La fonction processNotification() a lev� une exception:")
 115                 print(att)
 116         except Exception as e:
 117             print("La fonction processNotification() a lev� une exception:")
 118             print(e)
 119
 120     def callback(self):
 121         if self._callbackFunction is None: return
 122         self._callbackFunction()
 123
 124     def isEnded(self):
 125         """
 126         Retourne true si la fonction s'est termin�e naturellement
 127         (correctement ou en erreur). Utile pour les client qui
 128         pratique le pooling (interrogation r�p�t�e � intervalle
 129         r�gulier.
 130         """
 131         return self._stopevent.isSet()
 132         # _GBO_ On n'utilise pas isAlive() pour pouvoir ma�triser
 133         # l'indicateur de stop (exemple de la fonction kill)
 134         # return not self.isAlive()
 135
 136     def getException(self):
 137         return self._exception
 138
 139     def getId(self):
 140         return self._id
 141
 142     def wait(self,timeout=None):
 143         """
 144         Met fin au thread apr�s timeout seconde s'il est encore en
 145         ex�cution.
 146         Si le compte-�-rebours est atteind sans que la fonction
 147         functionToRun soit termin�e, alors le runner est laiss� dans
 148         l'�tat 'not Ended', c'est-�-dire que isEnded retourne
 149         false. On peut ainsi distinguer l'arr�t normal de l'arr�t �
 150         l'issue du timeout.
 151         """
 152         threading.Thread.join(self, timeout)
 153
 154
 155     def kill(self,message="Arr�t demand�"):
 156         """
 157         Cette m�thode doit �tre appeler pour interrombre le
 158         thread. Cet appel d�clare le thread comme en erreur (exception
 159         mise � disposition par la m�thode getException().
 160         """
 161         # On cr�e un exception indiquant la demande d'interruption
 162         self._exception = Exception(message)
 163         self._stopevent.set()
 164         # _GBO_ ATTENTION
 165         # Un thread python ne peut pas en r�alit� �tre int�rrompu.
 166         # La fonction reste donc en ex�cution m�me si on a rejoint le
 167         # thread appelant.
 168
 169     def clear(self):
 170         self._stopevent.clear()
 171         self._exception = None
 172         self._return    = None
 173
 174 # ===========================================================================
 175 CONTINUE=1
 176 STOP=0
 177
 178 class PeriodicTimer( threading.Thread ):
 179     """
 180     Cette classe permet d'amorcer un compte-�-rebours p�riodique pour
 181     op�rer un m�canisme de pooling. On d�finit la fonction appell�e
 182     p�riodiquement et la fonction terminale.
 183     """
 184     def __init__( self,  loopdelay, initdelay=0,
 185                   periodic_action=None,
 186                   ended_action=None,
 187                   *ended_argv ):
 188
 189         if periodic_action is None:
 190             self._periodic_action = self.defaultPeriodicAction
 191         else:
 192             self._periodic_action = periodic_action
 193         self._ended_action = ended_action
 194         self._loopdelay = loopdelay
 195         self._initdelay = initdelay
 196         self._running = CONTINUE
 197         self._ended_argv    = ended_argv
 198         threading.Thread.__init__( self )
 199
 200     def defaultPeriodicAction():
 201         """
 202         Les fonctions 'periodicAction' retournent CONTINU ou STOP
 203         apr�s avoir ex�cut� l'action de fin de boucle. Si STOP est
 204         retourn�, le cycle est interrompu.
 205         """
 206         return CONTINU
 207
 208     def run( self ):
 209         if self._initdelay:
 210             time.sleep( self._initdelay )
 211         self._runtime = time.time()
 212         while self._running == CONTINUE:
 213             start = time.time()
 214             self._running  = self._periodic_action()
 215             self._runtime += self._loopdelay
 216             time.sleep( max( 0, self._runtime - start ) )
 217         if self._ended_action is not None:
 218             self._ended_action( *self._ended_argv )
 219
 220     def stop( self ):
 221         self._running = STOP
 222
 223
 224
 225 #
 226 # ======================================================
 227 # Fonctions de test unitaire
 228 # ======================================================
 229 #
 230
 231 # ======================================================
 232 import os
 233 testfilename="/tmp/threadhelperTestFile"
 234 def testIfContinue():
 235     print("On examine la pr�sence du fichier ", testfilename)
 236     if os.path.exists(testfilename):
 237         return STOP
 238     else:
 239         return CONTINUE
 240
 241 def endedAction():
 242     print("FINI")
 243
 244 def TEST_PeriodicTimer():
 245     periodicTimer=PeriodicTimer(1,0,testIfContinue, endedAction)
 246     periodicTimer.start()
 247
 248
 249 # ======================================================
 250 def function_ok(nbsteps=5):
 251     """
 252     Fonction qui se termine correctement
 253     """
 254     print("D�but")
 255     cnt=0
 256     while ( cnt < nbsteps ):
 257         print("Etape ", cnt)
 258         time.sleep(0.6)
 259         cnt+=1
 260
 261     print("Fin")
 262
 263 def function_with_exception():
 264     """
 265     Fonction qui aboutie � une lev�e d'exception
 266     """
 267     print("D�but")
 268     cnt=0
 269     while ( cnt < 5 ):
 270         print("Etape ", cnt)
 271         time.sleep(1)
 272         cnt+=1
 273
 274     raise Exception("erreur d'ex�cution de la fonction")
 275     print("Fin")
 276
 277 def infinite_function():
 278     """
 279     fonction de dur�e infinie (tant qu'il y a du courant �l�ctrique) pour
 280     le test du timeout.
 281     """
 282     print("D�but")
 283     cnt=0
 284     while ( 1 ):
 285         print("Etape ", cnt)
 286         time.sleep(1)
 287         cnt+=1
 288
 289     raise Exception("erreur")
 290     print("Fin")
 291
 292
 293 def runWithRunner(functionToRun):
 294     """
 295     Ex�cute la fonction avec le runner. On illustre ici la modalit�
 296     d'utilisation du Runner.
 297     """
 298     print("###########")
 299     runner = Runner(functionToRun)
 300     runner.start()
 301
 302     while ( not runner.isEnded() ):
 303         print("La fonction est en cours")
 304         time.sleep(0.2)
 305     e = runner.getException()
 306     if e is not None:
 307         print("La fonction s'est termin�e en erreur")
 308         print(e)
 309         # On peut en fait la relancer
 310         # raise e
 311     else:
 312         print("La fonction s'est termin�e correctement")
 313
 314
 315 def runWithTimeout(functionToRun, timeout=10):
 316     """
 317     Ex�cute la fonction avec le runner. On illustre ici la modalit�
 318     d'utilisation du Runner.
 319     """
 320     print("runWithTimeout : DEBUT")
 321     runner = Runner(functionToRun)
 322     runner.start()
 323
 324     # On se fixe un temps au del� duquel on consid�re que la fonction
 325     # est en erreur => on tue le thread (timeout)
 326     runner.wait(timeout)
 327     print("Apr�s runner.timeout(timeout)")
 328     if not runner.isEnded():
 329         runner.kill()
 330     e = runner.getException()
 331     if e is not None:
 332         print("La fonction s'est termin�e en erreur")
 333         print(e)
 334         # On peut en fait la relancer
 335         # raise e
 336     else:
 337         print("La fonction s'est termin�e correctement")
 338
 339     print("runWithTimeout : FIN")
 340     import sys
 341     sys.exit(0)
 342
 343
 344 def TEST_Timeout():
 345     #runWithTimeout(function_ok)
 346     #runWithTimeout(function_ok,2)
 347     runWithTimeout(function_with_exception)
 348
 349
 350 def TEST_Runner():
 351     runWithRunner(function_ok)
 352     runWithRunner(function_with_exception)
 353     #runWithRunner(infinite_function)
 354
 355
 356 def myCallbackFunction():
 357     print("myCallbackFunction: the job is ended")
 358
 359
 360 def TEST_runWithCallback():
 361     runner = Runner(function_ok,8)
 362     runner.setCallbackFunction(myCallbackFunction)
 363     runner.start()
 364
 365     if runner.getException() is not None:
 366         return False
 367
 368     runnerId = runner.getId()
 369     print("A runner has been started with id="+str(runnerId))
 370     cpt = 0
 371     while ( not runner.isEnded() ):
 372         print("Waiting notification from process "+str(runner.getId())+", step n°"+str(cpt))
 373         time.sleep(0.2)
 374         cpt+=1
 375
 376     return True
 377
 378 if __name__ == "__main__":
 379     from . import unittester
 380     #TEST_PeriodicTimer()
 381     #TEST_Runner()
 382     #TEST_Timeout()
 383     unittester.run("threadhelper","TEST_runWithCallback")
 384
 385