En nyttig trådmodul

[A_N_K]

I forbindelse med utviklingen av et større program har jeg skrevet en modul oppå den standard threading, kalt mythreading, som jeg har funnet ganske så nyttig underveis. Tanken var at kanskje andre enn meg ville finne koden nyttig, og for den del komme med idéer for forbedringer så jeg legger ut koden her. Hovedforbedringen (slik jeg ser det i alle fall) er at klassen Thread har en metode cancel() som andre tråder kan benytte for å kansellere tråden, og testCancel() som selve tråden kan benytte for å ta en slik forespørsel til følge. testCancel() vil kaste en exception så man kan spre kall til testCancel() utover i koden hvor man venter at ting kan ta tid. Det finnes ingen fasiliteter (det jeg kjenner til i alle fall) i standard Python for å stoppe tråder, så jeg bestemte meg for å implementere det nest beste For eksempelkode se slutten av modulen (under linje 125).

 

import threading

def testCancel():
   thrd = Thread.currentThread()
   thrd.testCancel()

class ThreadError(RuntimeError):
   """ Used to contain an exception that in fact happened in a background thread. """
   def __init__(self, threadName, params):
       self.thread = threadName
       self.excType, self.exc, self.tb = params
       import traceback
       RuntimeError.__init__(self, "Exception occurred in thread %s:\n%s" % (threadName, " ".join(traceback.format_exception(self.exc, \
               self.exc, self.tb))))

def _defHandler(threadName, excInfo):
   print "Default _excHandler"
   raise ThreadError(threadName, excInfo)
_excHandler = _defHandler

def registerExceptionHandler(handler):
   """ Register a handler for exceptions happening in background thread.
   The exception handler will retrieve the name of the thread and a tuple as returned by sys.exc_info() """
   global _excHandler
   _excHandler = handler

def synchronized(func):
   def syncFunc(*args, **kwds):
       func._sync_lock.acquire()
       r = func(*args, **kwds)
       func._sync_lock.release()
       return r
   
   func._sync_lock = threading.Lock()
   return syncFunc

class Cancellation(Exception):
   pass

class TimeoutError(Exception):
   pass

class Thread(object):
   _threadLocal = threading.local()
   _threadLocal.current = None

   class _DummyThread:
       """ Dummy class for objects that get returned by currentThread if no Thread is controlling the current thread. """
       
       def testCancel(self):
           pass
   
   def __init__(self, target=None, args=[], kwds={}, name=None, daemon=False):
       self._thrd = threading.Thread(target=self._run, name=name)
       self._trgt, self._args, self._kwds = target, args, kwds

       for mthd in ("join",):
           setattr(self, mthd, getattr(self._thrd, mthd))
           
       self.__eventCancel = threading.Event()
       if daemon:
           self.daemon = True

   @classmethod
   def currentThread(cls):
       cur = cls._threadLocal.current
       if cur is None:
           return Thread._DummyThread()
       return cur

   def __getName(self):
       return self._thrd.getName()
   def __setName(self, name):
       self._thrd.setName(name)
   name = property(__getName, __setName)

   def __isDaemon(self):
       return self._thrd.isDaemon()
   def __setDaemon(self, daemon):
       self._thrd.setDaemon(daemon)
   daemon = property(__isDaemon, __setDaemon)

   @property
   def alive(self):
       return self._thrd.isAlive()

   def start(self):
       self._thrd.start()

   @synchronized
   def cancel(self, wait=False, timeout=None):
       """ Tell this thread to cancel itself. Will wait till the request is honoured.
       It is also possible that the thread finishes its execution independently of this request,
       this function will return anyway when it notices that the thread is no longer running. """

       assert not (Thread.currentThread() is self)
       self.__eventCancel.set()
       if wait:
           self.join(timeout=timeout)
           if self.alive:
               raise TimeoutError

   def testCancel(self):
       assert Thread.currentThread() is self
       if self.__eventCancel.isSet():
           raise Cancellation

   def run(self):
       if self._trgt is None:
           raise NotImplementedError
       self._trgt(*self._args, **self._kwds)

   def _run(self):
       Thread._threadLocal.current = self

       try:
           self.run()
       except Cancellation:
           pass
       except:
           global _excHandler
           import sys
           _excHandler(self.name, sys.exc_info())

if __name__ == "__main__":
   import time

   def test():
       try:
           while True:
               print "Thread %s running" % Thread.currentThread().name
               time.sleep(1)
               testCancel()
       except Cancellation:
           print "Thread %s asked to cancel" % Thread.currentThread().name

   thrd = Thread(target=test, name="UtenEnTraad")
   thrd.start()
   time.sleep(10)
   thrd.cancel()
   thrd.join()

[zeitgeist]

Kanskje du skal legge den inn i Cheese Shop?

http://cheeseshop.python.org

[A_N_K]

Tja, ikke vurdert det egentlig. Men det er ikke så voldsomt mye jeg har lagt til utover det som finnes fra før i threading når sant skal sies.

 

Edit: Dette minte meg på at jeg bør ta en kikk i ActiveState's "oppskrifter" for å se om det finnes noe for tråding.

 

Edit1: Fant i alle fall en måte å oversette signaler i tråder til events i Qt's eventloop, men jeg likte min måte bedre _)

Endret 23. august 2005 av A_N_K

[A_N_K]

På den annen side har jeg vurdert å sende inn koden til ActiveState's Python Cookbook, men da tror jeg Thread-klassen spesielt bør trimmes litt (mye av grensesnittet er simpelthen wrapping av threading.Thread etter eget forgodtbefinnende). Kom gjerne med innspill til design! Noe jeg glemte å nevne var at med dette rammeverket kan man registrere en callback for å ta imot exceptions i tråder, mye greiere enn at tråden simpelthen dør på seg og dumper feilmeldingen i konsoll.