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