home *** CD-ROM | disk | FTP | other *** search
/ Chip 2003 January / Chip_2003-01_cd2.bin / convert / eJayMp3Pro / mp3pro_demo.exe / THREADING.PY < prev    next >
Encoding:
Python Source  |  1998-06-14  |  18.3 KB  |  639 lines

  1. # threading.py:
  2. # Proposed new threading module, emulating a subset of Java's threading model
  3.  
  4. import sys
  5. import time
  6. import thread
  7. import traceback
  8. import StringIO
  9.  
  10. # Rename some stuff so "from threading import *" is safe
  11.  
  12. _sys = sys
  13. del sys
  14.  
  15. _time = time.time
  16. _sleep = time.sleep
  17. del time
  18.  
  19. _start_new_thread = thread.start_new_thread
  20. _allocate_lock = thread.allocate_lock
  21. _get_ident = thread.get_ident
  22. del thread
  23.  
  24. _print_exc = traceback.print_exc
  25. del traceback
  26.  
  27. _StringIO = StringIO.StringIO
  28. del StringIO
  29.  
  30.  
  31. # Debug support (adapted from ihooks.py)
  32.  
  33. _VERBOSE = 0
  34.  
  35. if __debug__:
  36.  
  37.     class _Verbose:
  38.  
  39.         def __init__(self, verbose=None):
  40.             if verbose is None:
  41.                 verbose = _VERBOSE
  42.             self.__verbose = verbose
  43.  
  44.         def _note(self, format, *args):
  45.             if self.__verbose:
  46.                 format = format % args
  47.                 format = "%s: %s\n" % (
  48.                     currentThread().getName(), format)
  49.                 _sys.stderr.write(format)
  50.  
  51. else:
  52.     # Disable this when using "python -O"
  53.     class _Verbose:
  54.         def __init__(self, verbose=None):
  55.             pass
  56.         def _note(self, *args):
  57.             pass
  58.  
  59.  
  60. # Synchronization classes
  61.  
  62. Lock = _allocate_lock
  63.  
  64. def RLock(*args, **kwargs):
  65.     return apply(_RLock, args, kwargs)
  66.  
  67. class _RLock(_Verbose):
  68.     
  69.     def __init__(self, verbose=None):
  70.         _Verbose.__init__(self, verbose)
  71.         self.__block = _allocate_lock()
  72.         self.__owner = None
  73.         self.__count = 0
  74.  
  75.     def __repr__(self):
  76.         return "<%s(%s, %d)>" % (
  77.                 self.__class__.__name__,
  78.                 self.__owner and self.__owner.getName(),
  79.                 self.__count)
  80.  
  81.     def acquire(self, blocking=1):
  82.         me = currentThread()
  83.         if self.__owner is me:
  84.             self.__count = self.__count + 1
  85.             if __debug__:
  86.                 self._note("%s.acquire(%s): recursive success", self, blocking)
  87.             return 1
  88.         rc = self.__block.acquire(blocking)
  89.         if rc:
  90.             self.__owner = me
  91.             self.__count = 1
  92.             if __debug__:
  93.                 self._note("%s.acquire(%s): initial succes", self, blocking)
  94.         else:
  95.             if __debug__:
  96.                 self._note("%s.acquire(%s): failure", self, blocking)
  97.         return rc
  98.  
  99.     def release(self):
  100.         me = currentThread()
  101.         assert self.__owner is me, "release() of un-acquire()d lock"
  102.         self.__count = count = self.__count - 1
  103.         if not count:
  104.             self.__owner = None
  105.             self.__block.release()
  106.             if __debug__:
  107.                 self._note("%s.release(): final release", self)
  108.         else:
  109.             if __debug__:
  110.                 self._note("%s.release(): non-final release", self)
  111.  
  112.     # Internal methods used by condition variables
  113.  
  114.     def _acquire_restore(self, (count, owner)):
  115.         self.__block.acquire()
  116.         self.__count = count
  117.         self.__owner = owner
  118.         if __debug__:
  119.             self._note("%s._acquire_restore()", self)
  120.  
  121.     def _release_save(self):
  122.         if __debug__:
  123.             self._note("%s._release_save()", self)
  124.         count = self.__count
  125.         self.__count = 0
  126.         owner = self.__owner
  127.         self.__owner = None
  128.         self.__block.release()
  129.         return (count, owner)
  130.  
  131.     def _is_owned(self):
  132.         return self.__owner is currentThread()
  133.  
  134.  
  135. def Condition(*args, **kwargs):
  136.     return apply(_Condition, args, kwargs)
  137.  
  138. class _Condition(_Verbose):
  139.  
  140.     def __init__(self, lock=None, verbose=None):
  141.         _Verbose.__init__(self, verbose)
  142.         if lock is None:
  143.             lock = RLock()
  144.         self.__lock = lock
  145.         # Export the lock's acquire() and release() methods
  146.         self.acquire = lock.acquire
  147.         self.release = lock.release
  148.         # If the lock defines _release_save() and/or _acquire_restore(),
  149.         # these override the default implementations (which just call
  150.         # release() and acquire() on the lock).  Ditto for _is_owned().
  151.         try:
  152.             self._release_save = lock._release_save
  153.         except AttributeError:
  154.             pass
  155.         try:
  156.             self._acquire_restore = lock._acquire_restore
  157.         except AttributeError:
  158.             pass
  159.         try:
  160.             self._is_owned = lock._is_owned
  161.         except AttributeError:
  162.             pass
  163.         self.__waiters = []
  164.  
  165.     def __repr__(self):
  166.         return "<Condition(%s, %d)>" % (self.__lock, len(self.__waiters))
  167.  
  168.     def _release_save(self):
  169.         self.__lock.release()           # No state to save
  170.  
  171.     def _acquire_restore(self, x):
  172.         self.__lock.acquire()           # Ignore saved state
  173.  
  174.     def _is_owned(self):
  175.         if self.__lock.acquire(0):
  176.             self.__lock.release()
  177.             return 0
  178.         else:
  179.             return 1
  180.  
  181.     def wait(self, timeout=None):
  182.         me = currentThread()
  183.         assert self._is_owned(), "wait() of un-acquire()d lock"
  184.         waiter = _allocate_lock()
  185.         waiter.acquire()
  186.         self.__waiters.append(waiter)
  187.         saved_state = self._release_save()
  188.         if timeout is None:
  189.             waiter.acquire()
  190.             if __debug__:
  191.                 self._note("%s.wait(): got it", self)
  192.         else:
  193.             endtime = _time() + timeout
  194.             delay = 0.000001 # 1 usec
  195.             while 1:
  196.                 gotit = waiter.acquire(0)
  197.                 if gotit or _time() >= endtime:
  198.                     break
  199.                 _sleep(delay)
  200.                 if delay < 1.0:
  201.                     delay = delay * 2.0
  202.             if not gotit:
  203.                 if __debug__:
  204.                     self._note("%s.wait(%s): timed out", self, timeout)
  205.                 try:
  206.                     self.__waiters.remove(waiter)
  207.                 except ValueError:
  208.                     pass
  209.             else:
  210.                 if __debug__:
  211.                     self._note("%s.wait(%s): got it", self, timeout)
  212.         self._acquire_restore(saved_state)
  213.  
  214.     def notify(self, n=1):
  215.         me = currentThread()
  216.         assert self._is_owned(), "notify() of un-acquire()d lock"
  217.         __waiters = self.__waiters
  218.         waiters = __waiters[:n]
  219.         if not waiters:
  220.             if __debug__:
  221.                 self._note("%s.notify(): no waiters", self)
  222.             return
  223.         self._note("%s.notify(): notifying %d waiter%s", self, n,
  224.                    n!=1 and "s" or "")
  225.         for waiter in waiters:
  226.             waiter.release()
  227.             try:
  228.                 __waiters.remove(waiter)
  229.             except ValueError:
  230.                 pass
  231.  
  232.     def notifyAll(self):
  233.         self.notify(len(self.__waiters))
  234.  
  235.  
  236. def Semaphore(*args, **kwargs):
  237.     return apply(_Semaphore, args, kwargs)
  238.  
  239. class _Semaphore(_Verbose):
  240.  
  241.     # After Tim Peters' semaphore class, but bnot quite the same (no maximum)
  242.  
  243.     def __init__(self, value=1, verbose=None):
  244.         assert value >= 0, "Semaphore initial value must be >= 0"
  245.         _Verbose.__init__(self, verbose)
  246.         self.__cond = Condition(Lock())
  247.         self.__value = value
  248.  
  249.     def acquire(self, blocking=1):
  250.         rc = 0
  251.         self.__cond.acquire()
  252.         while self.__value == 0:
  253.             if not blocking:
  254.                 break
  255.             self.__cond.wait()
  256.         else:
  257.             self.__value = self.__value - 1
  258.             rc = 1
  259.         self.__cond.release()
  260.         return rc
  261.  
  262.     def release(self):
  263.         self.__cond.acquire()
  264.         self.__value = self.__value + 1
  265.         self.__cond.notify()
  266.         self.__cond.release()
  267.  
  268.  
  269. def Event(*args, **kwargs):
  270.     return apply(_Event, args, kwargs)
  271.  
  272. class _Event(_Verbose):
  273.  
  274.     # After Tim Peters' event class (without is_posted())
  275.  
  276.     def __init__(self, verbose=None):
  277.         _Verbose.__init__(self, verbose)
  278.         self.__cond = Condition(Lock())
  279.         self.__flag = 0
  280.  
  281.     def isSet(self):
  282.         return self.__flag
  283.  
  284.     def set(self):
  285.         self.__cond.acquire()
  286.         self.__flag = 1
  287.         self.__cond.notifyAll()
  288.         self.__cond.release()
  289.  
  290.     def clear(self):
  291.         self.__cond.acquire()
  292.         self.__flag = 0
  293.         self.__cond.release()
  294.  
  295.     def wait(self, timeout=None):
  296.         self.__cond.acquire()
  297.         if not self.__flag:
  298.             self.__cond.wait(timeout)
  299.         self.__cond.release()
  300.  
  301.  
  302. # Helper to generate new thread names
  303. _counter = 0
  304. def _newname(template="Thread-%d"):
  305.     global _counter
  306.     _counter = _counter + 1
  307.     return template % _counter
  308.  
  309. # Active thread administration
  310. _active_limbo_lock = _allocate_lock()
  311. _active = {}
  312. _limbo = {}
  313.  
  314.  
  315. # Main class for threads
  316.  
  317. class Thread(_Verbose):
  318.  
  319.     __initialized = 0
  320.  
  321.     def __init__(self, group=None, target=None, name=None,
  322.                  args=(), kwargs={}, verbose=None):
  323.         assert group is None, "group argument must be None for now"
  324.         _Verbose.__init__(self, verbose)
  325.         self.__target = target
  326.         self.__name = str(name or _newname())
  327.         self.__args = args
  328.         self.__kwargs = kwargs
  329.         self.__daemonic = self._set_daemon()
  330.         self.__started = 0
  331.         self.__stopped = 0
  332.         self.__block = Condition(Lock())
  333.         self.__initialized = 1
  334.  
  335.     def _set_daemon(self):
  336.         # Overridden in _MainThread and _DummyThread
  337.         return currentThread().isDaemon()
  338.  
  339.     def __repr__(self):
  340.         assert self.__initialized, "Thread.__init__() was not called"
  341.         status = "initial"
  342.         if self.__started:
  343.             status = "started"
  344.         if self.__stopped:
  345.             status = "stopped"
  346.         if self.__daemonic:
  347.             status = status + " daemon"
  348.         return "<%s(%s, %s)>" % (self.__class__.__name__, self.__name, status)
  349.  
  350.     def start(self):
  351.         assert self.__initialized, "Thread.__init__() not called"
  352.         assert not self.__started, "thread already started"
  353.         if __debug__:
  354.             self._note("%s.start(): starting thread", self)
  355.         _active_limbo_lock.acquire()
  356.         _limbo[self] = self
  357.         _active_limbo_lock.release()
  358.         _start_new_thread(self.__bootstrap, ())
  359.         self.__started = 1
  360.         _sleep(0.000001)    # 1 usec, to let the thread run (Solaris hack)
  361.  
  362.     def run(self):
  363.         if self.__target:
  364.             apply(self.__target, self.__args, self.__kwargs)
  365.  
  366.     def __bootstrap(self):
  367.         try:
  368.             self.__started = 1
  369.             _active_limbo_lock.acquire()
  370.             _active[_get_ident()] = self
  371.             del _limbo[self]
  372.             _active_limbo_lock.release()
  373.             if __debug__:
  374.                 self._note("%s.__bootstrap(): thread started", self)
  375.             try:
  376.                 self.run()
  377.             except SystemExit:
  378.                 if __debug__:
  379.                     self._note("%s.__bootstrap(): raised SystemExit", self)
  380.             except:
  381.                 if __debug__:
  382.                     self._note("%s.__bootstrap(): unhandled exception", self)
  383.                 s = _StringIO()
  384.                 _print_exc(file=s)
  385.                 _sys.stderr.write("Exception in thread %s:\n%s\n" %
  386.                                  (self.getName(), s.getvalue()))
  387.             else:
  388.                 if __debug__:
  389.                     self._note("%s.__bootstrap(): normal return", self)
  390.         finally:
  391.             self.__stop()
  392.             self.__delete()
  393.  
  394.     def __stop(self):
  395.         self.__block.acquire()
  396.         self.__stopped = 1
  397.         self.__block.notifyAll()
  398.         self.__block.release()
  399.  
  400.     def __delete(self):
  401.         _active_limbo_lock.acquire()
  402.         del _active[_get_ident()]
  403.         _active_limbo_lock.release()
  404.  
  405.     def join(self, timeout=None):
  406.         assert self.__initialized, "Thread.__init__() not called"
  407.         assert self.__started, "cannot join thread before it is started"
  408.         assert self is not currentThread(), "cannot join current thread"
  409.         if __debug__:
  410.             if not self.__stopped:
  411.                 self._note("%s.join(): waiting until thread stops", self)
  412.         self.__block.acquire()
  413.         if timeout is None:
  414.             while not self.__stopped:
  415.                 self.__block.wait()
  416.             if __debug__:
  417.                 self._note("%s.join(): thread stopped", self)
  418.         else:
  419.             deadline = _time() + timeout
  420.             while not self.__stopped:
  421.                 delay = deadline - _time()
  422.                 if delay <= 0:
  423.                     if __debug__:
  424.                         self._note("%s.join(): timed out", self)
  425.                     break
  426.                 self.__block.wait(delay)
  427.             else:
  428.                 if __debug__:
  429.                     self._note("%s.join(): thread stopped", self)
  430.         self.__block.release()
  431.  
  432.     def getName(self):
  433.         assert self.__initialized, "Thread.__init__() not called"
  434.         return self.__name
  435.  
  436.     def setName(self, name):
  437.         assert self.__initialized, "Thread.__init__() not called"
  438.         self.__name = str(name)
  439.  
  440.     def isAlive(self):
  441.         assert self.__initialized, "Thread.__init__() not called"
  442.         return self.__started and not self.__stopped
  443.     
  444.     def isDaemon(self):
  445.         assert self.__initialized, "Thread.__init__() not called"
  446.         return self.__daemonic
  447.  
  448.     def setDaemon(self, daemonic):
  449.         assert self.__initialized, "Thread.__init__() not called"
  450.         assert not self.__started, "cannot set daemon status of active thread"
  451.         self.__daemonic = daemonic
  452.  
  453.  
  454. # Special thread class to represent the main thread
  455. # This is garbage collected through an exit handler
  456.  
  457. class _MainThread(Thread):
  458.  
  459.     def __init__(self):
  460.         Thread.__init__(self, name="MainThread")
  461.         self._Thread__started = 1
  462.         _active_limbo_lock.acquire()
  463.         _active[_get_ident()] = self
  464.         _active_limbo_lock.release()
  465.         try:
  466.             self.__oldexitfunc = _sys.exitfunc
  467.         except AttributeError:
  468.             self.__oldexitfunc = None
  469.         _sys.exitfunc = self.__exitfunc
  470.  
  471.     def _set_daemon(self):
  472.         return 0
  473.  
  474.     def __exitfunc(self):
  475.         self._Thread__stop()
  476.         t = _pickSomeNonDaemonThread()
  477.         if t:
  478.             if __debug__:
  479.                 self._note("%s: waiting for other threads", self)
  480.         while t:
  481.             t.join()
  482.             t = _pickSomeNonDaemonThread()
  483.         if self.__oldexitfunc:
  484.             if __debug__:
  485.                 self._note("%s: calling exit handler", self)
  486.             self.__oldexitfunc()
  487.         if __debug__:
  488.             self._note("%s: exiting", self)
  489.         self._Thread__delete()
  490.  
  491. def _pickSomeNonDaemonThread():
  492.     for t in enumerate():
  493.         if not t.isDaemon() and t.isAlive():
  494.             return t
  495.     return None
  496.  
  497.  
  498. # Dummy thread class to represent threads not started here.
  499. # These aren't garbage collected when they die,
  500. # nor can they be waited for.
  501. # Their purpose is to return *something* from currentThread().
  502. # They are marked as daemon threads so we won't wait for them
  503. # when we exit (conform previous semantics).
  504.  
  505. class _DummyThread(Thread):
  506.     
  507.     def __init__(self):
  508.         Thread.__init__(self, name=_newname("Dummy-%d"))
  509.         self.__Thread_started = 1
  510.         _active_limbo_lock.acquire()
  511.         _active[_get_ident()] = self
  512.         _active_limbo_lock.release()
  513.  
  514.     def _set_daemon(self):
  515.         return 1
  516.  
  517.     def join(self):
  518.         assert 0, "cannot join a dummy thread"
  519.  
  520.  
  521. # Global API functions
  522.  
  523. def currentThread():
  524.     try:
  525.         return _active[_get_ident()]
  526.     except KeyError:
  527.         print "currentThread(): no current thread for", _get_ident()
  528.         return _DummyThread()
  529.  
  530. def activeCount():
  531.     _active_limbo_lock.acquire()
  532.     count = len(_active) + len(_limbo)
  533.     _active_limbo_lock.release()
  534.     return count
  535.  
  536. def enumerate():
  537.     _active_limbo_lock.acquire()
  538.     active = _active.values() + _limbo.values()
  539.     _active_limbo_lock.release()
  540.     return active
  541.  
  542.  
  543. # Create the main thread object
  544.  
  545. _MainThread()
  546.  
  547.  
  548. # Self-test code
  549.  
  550. def _test():
  551.  
  552.     import random
  553.  
  554.     class BoundedQueue(_Verbose):
  555.  
  556.         def __init__(self, limit):
  557.             _Verbose.__init__(self)
  558.             self.mon = RLock()
  559.             self.rc = Condition(self.mon)
  560.             self.wc = Condition(self.mon)
  561.             self.limit = limit
  562.             self.queue = []
  563.  
  564.         def put(self, item):
  565.             self.mon.acquire()
  566.             while len(self.queue) >= self.limit:
  567.                 self._note("put(%s): queue full", item)
  568.                 self.wc.wait()
  569.             self.queue.append(item)
  570.             self._note("put(%s): appended, length now %d",
  571.                        item, len(self.queue))
  572.             self.rc.notify()
  573.             self.mon.release()
  574.  
  575.         def get(self):
  576.             self.mon.acquire()
  577.             while not self.queue:
  578.                 self._note("get(): queue empty")
  579.                 self.rc.wait()
  580.             item = self.queue[0]
  581.             del self.queue[0]
  582.             self._note("get(): got %s, %d left", item, len(self.queue))
  583.             self.wc.notify()
  584.             self.mon.release()
  585.             return item
  586.  
  587.     class ProducerThread(Thread):
  588.  
  589.         def __init__(self, queue, quota):
  590.             Thread.__init__(self, name="Producer")
  591.             self.queue = queue
  592.             self.quota = quota
  593.  
  594.         def run(self):
  595.             from random import random
  596.             counter = 0
  597.             while counter < self.quota:
  598.                 counter = counter + 1
  599.                 self.queue.put("%s.%d" % (self.getName(), counter))
  600.                 _sleep(random() * 0.00001)
  601.  
  602.  
  603.     class ConsumerThread(Thread):
  604.  
  605.         def __init__(self, queue, count):
  606.             Thread.__init__(self, name="Consumer")
  607.             self.queue = queue
  608.             self.count = count
  609.  
  610.         def run(self):
  611.             while self.count > 0:
  612.                 item = self.queue.get()
  613.                 print item
  614.                 self.count = self.count - 1
  615.  
  616.     import time
  617.  
  618.     NP = 3
  619.     QL = 4
  620.     NI = 5
  621.  
  622.     Q = BoundedQueue(QL)
  623.     P = []
  624.     for i in range(NP):
  625.         t = ProducerThread(Q, NI)
  626.         t.setName("Producer-%d" % (i+1))
  627.         P.append(t)
  628.     C = ConsumerThread(Q, NI*NP)
  629.     for t in P:
  630.         t.start()
  631.         _sleep(0.000001)
  632.     C.start()
  633.     for t in P:
  634.         t.join()
  635.     C.join()
  636.  
  637. if __name__ == '__main__':
  638.     _test()
  639.