Want more features on Pastebin? Sign Up, it's FREE!
Guest

threading.py

By: a guest on Mar 26th, 2011  |  syntax: Python  |  size: 18.32 KB  |  views: 88  |  expires: Never
download  |  raw  |  embed  |  report abuse  |  print
Text below is selected. Please press Ctrl+C to copy to your clipboard. (⌘+C on Mac)
  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. ThreadError = thread.error
  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.         try:    # restore state no matter what (e.g., KeyboardInterrupt)
  189.             if timeout is None:
  190.                 waiter.acquire()
  191.                 if __debug__:
  192.                     self._note("%s.wait(): got it", self)
  193.             else:
  194.                 endtime = _time() + timeout
  195.                 delay = 0.000001 # 1 usec
  196.                 while 1:
  197.                     gotit = waiter.acquire(0)
  198.                     if gotit or _time() >= endtime:
  199.                         break
  200.                     _sleep(delay)
  201.                     if delay < 1.0:
  202.                         delay = delay * 2.0
  203.                 if not gotit:
  204.                     if __debug__:
  205.                         self._note("%s.wait(%s): timed out", self, timeout)
  206.                     try:
  207.                         self.__waiters.remove(waiter)
  208.                     except ValueError:
  209.                         pass
  210.                 else:
  211.                     if __debug__:
  212.                         self._note("%s.wait(%s): got it", self, timeout)
  213.         finally:
  214.             self._acquire_restore(saved_state)
  215.  
  216.     def notify(self, n=1):
  217.         me = currentThread()
  218.         assert self._is_owned(), "notify() of un-acquire()d lock"
  219.         __waiters = self.__waiters
  220.         waiters = __waiters[:n]
  221.         if not waiters:
  222.             if __debug__:
  223.                 self._note("%s.notify(): no waiters", self)
  224.             return
  225.         self._note("%s.notify(): notifying %d waiter%s", self, n,
  226.                    n!=1 and "s" or "")
  227.         for waiter in waiters:
  228.             waiter.release()
  229.             try:
  230.                 __waiters.remove(waiter)
  231.             except ValueError:
  232.                 pass
  233.  
  234.     def notifyAll(self):
  235.         self.notify(len(self.__waiters))
  236.  
  237.  
  238. def Semaphore(*args, **kwargs):
  239.     return apply(_Semaphore, args, kwargs)
  240.  
  241. class _Semaphore(_Verbose):
  242.  
  243.     # After Tim Peters' semaphore class, but not quite the same (no maximum)
  244.  
  245.     def __init__(self, value=1, verbose=None):
  246.         assert value >= 0, "Semaphore initial value must be >= 0"
  247.         _Verbose.__init__(self, verbose)
  248.         self.__cond = Condition(Lock())
  249.         self.__value = value
  250.  
  251.     def acquire(self, blocking=1):
  252.         rc = 0
  253.         self.__cond.acquire()
  254.         while self.__value == 0:
  255.             if not blocking:
  256.                 break
  257.             self.__cond.wait()
  258.         else:
  259.             self.__value = self.__value - 1
  260.             rc = 1
  261.         self.__cond.release()
  262.         return rc
  263.  
  264.     def release(self):
  265.         self.__cond.acquire()
  266.         self.__value = self.__value + 1
  267.         self.__cond.notify()
  268.         self.__cond.release()
  269.  
  270.  
  271. def Event(*args, **kwargs):
  272.     return apply(_Event, args, kwargs)
  273.  
  274. class _Event(_Verbose):
  275.  
  276.     # After Tim Peters' event class (without is_posted())
  277.  
  278.     def __init__(self, verbose=None):
  279.         _Verbose.__init__(self, verbose)
  280.         self.__cond = Condition(Lock())
  281.         self.__flag = 0
  282.  
  283.     def isSet(self):
  284.         return self.__flag
  285.  
  286.     def set(self):
  287.         self.__cond.acquire()
  288.         self.__flag = 1
  289.         self.__cond.notifyAll()
  290.         self.__cond.release()
  291.  
  292.     def clear(self):
  293.         self.__cond.acquire()
  294.         self.__flag = 0
  295.         self.__cond.release()
  296.  
  297.     def wait(self, timeout=None):
  298.         self.__cond.acquire()
  299.         if not self.__flag:
  300.             self.__cond.wait(timeout)
  301.         self.__cond.release()
  302.  
  303.  
  304. # Helper to generate new thread names
  305. _counter = 0
  306. def _newname(template="Thread-%d"):
  307.     global _counter
  308.     _counter = _counter + 1
  309.     return template % _counter
  310.  
  311. # Active thread administration
  312. _active_limbo_lock = _allocate_lock()
  313. _active = {}
  314. _limbo = {}
  315.  
  316.  
  317. # Main class for threads
  318.  
  319. class Thread(_Verbose):
  320.  
  321.     __initialized = 0
  322.  
  323.     def __init__(self, group=None, target=None, name=None,
  324.                  args=(), kwargs={}, verbose=None):
  325.         assert group is None, "group argument must be None for now"
  326.         _Verbose.__init__(self, verbose)
  327.         self.__target = target
  328.         self.__name = str(name or _newname())
  329.         self.__args = args
  330.         self.__kwargs = kwargs
  331.         self.__daemonic = self._set_daemon()
  332.         self.__started = 0
  333.         self.__stopped = 0
  334.         self.__block = Condition(Lock())
  335.         self.__initialized = 1
  336.  
  337.     def _set_daemon(self):
  338.         # Overridden in _MainThread and _DummyThread
  339.         return currentThread().isDaemon()
  340.  
  341.     def __repr__(self):
  342.         assert self.__initialized, "Thread.__init__() was not called"
  343.         status = "initial"
  344.         if self.__started:
  345.             status = "started"
  346.         if self.__stopped:
  347.             status = "stopped"
  348.         if self.__daemonic:
  349.             status = status + " daemon"
  350.         return "<%s(%s, %s)>" % (self.__class__.__name__, self.__name, status)
  351.  
  352.     def start(self):
  353.         assert self.__initialized, "Thread.__init__() not called"
  354.         assert not self.__started, "thread already started"
  355.         if __debug__:
  356.             self._note("%s.start(): starting thread", self)
  357.         _active_limbo_lock.acquire()
  358.         _limbo[self] = self
  359.         _active_limbo_lock.release()
  360.         _start_new_thread(self.__bootstrap, ())
  361.         self.__started = 1
  362.         _sleep(0.000001)    # 1 usec, to let the thread run (Solaris hack)
  363.  
  364.     def run(self):
  365.         if self.__target:
  366.             apply(self.__target, self.__args, self.__kwargs)
  367.  
  368.     def __bootstrap(self):
  369.         try:
  370.             self.__started = 1
  371.             _active_limbo_lock.acquire()
  372.             _active[_get_ident()] = self
  373.             del _limbo[self]
  374.             _active_limbo_lock.release()
  375.             if __debug__:
  376.                 self._note("%s.__bootstrap(): thread started", self)
  377.             try:
  378.                 self.run()
  379.             except SystemExit:
  380.                 if __debug__:
  381.                     self._note("%s.__bootstrap(): raised SystemExit", self)
  382.             except:
  383.                 if __debug__:
  384.                     self._note("%s.__bootstrap(): unhandled exception", self)
  385.                 s = _StringIO()
  386.                 _print_exc(file=s)
  387.                 _sys.stderr.write("Exception in thread %s:\n%s\n" %
  388.                                  (self.getName(), s.getvalue()))
  389.             else:
  390.                 if __debug__:
  391.                     self._note("%s.__bootstrap(): normal return", self)
  392.         finally:
  393.             self.__stop()
  394.             try:
  395.                 self.__delete()
  396.             except:
  397.                 pass
  398.  
  399.     def __stop(self):
  400.         self.__block.acquire()
  401.         self.__stopped = 1
  402.         self.__block.notifyAll()
  403.         self.__block.release()
  404.  
  405.     def __delete(self):
  406.         _active_limbo_lock.acquire()
  407.         del _active[_get_ident()]
  408.         _active_limbo_lock.release()
  409.  
  410.     def join(self, timeout=None):
  411.         assert self.__initialized, "Thread.__init__() not called"
  412.         assert self.__started, "cannot join thread before it is started"
  413.         assert self is not currentThread(), "cannot join current thread"
  414.         if __debug__:
  415.             if not self.__stopped:
  416.                 self._note("%s.join(): waiting until thread stops", self)
  417.         self.__block.acquire()
  418.         if timeout is None:
  419.             while not self.__stopped:
  420.                 self.__block.wait()
  421.             if __debug__:
  422.                 self._note("%s.join(): thread stopped", self)
  423.         else:
  424.             deadline = _time() + timeout
  425.             while not self.__stopped:
  426.                 delay = deadline - _time()
  427.                 if delay <= 0:
  428.                     if __debug__:
  429.                         self._note("%s.join(): timed out", self)
  430.                     break
  431.                 self.__block.wait(delay)
  432.             else:
  433.                 if __debug__:
  434.                     self._note("%s.join(): thread stopped", self)
  435.         self.__block.release()
  436.  
  437.     def getName(self):
  438.         assert self.__initialized, "Thread.__init__() not called"
  439.         return self.__name
  440.  
  441.     def setName(self, name):
  442.         assert self.__initialized, "Thread.__init__() not called"
  443.         self.__name = str(name)
  444.  
  445.     def isAlive(self):
  446.         assert self.__initialized, "Thread.__init__() not called"
  447.         return self.__started and not self.__stopped
  448.  
  449.     def isDaemon(self):
  450.         assert self.__initialized, "Thread.__init__() not called"
  451.         return self.__daemonic
  452.  
  453.     def setDaemon(self, daemonic):
  454.         assert self.__initialized, "Thread.__init__() not called"
  455.         assert not self.__started, "cannot set daemon status of active thread"
  456.         self.__daemonic = daemonic
  457.  
  458.  
  459. # Special thread class to represent the main thread
  460. # This is garbage collected through an exit handler
  461.  
  462. class _MainThread(Thread):
  463.  
  464.     def __init__(self):
  465.         Thread.__init__(self, name="MainThread")
  466.         self._Thread__started = 1
  467.         _active_limbo_lock.acquire()
  468.         _active[_get_ident()] = self
  469.         _active_limbo_lock.release()
  470.         import atexit
  471.         atexit.register(self.__exitfunc)
  472.  
  473.     def _set_daemon(self):
  474.         return 0
  475.  
  476.     def __exitfunc(self):
  477.         self._Thread__stop()
  478.         t = _pickSomeNonDaemonThread()
  479.         if t:
  480.             if __debug__:
  481.                 self._note("%s: waiting for other threads", self)
  482.         while t:
  483.             t.join()
  484.             t = _pickSomeNonDaemonThread()
  485.         if __debug__:
  486.             self._note("%s: exiting", self)
  487.         self._Thread__delete()
  488.  
  489. def _pickSomeNonDaemonThread():
  490.     for t in enumerate():
  491.         if not t.isDaemon() and t.isAlive():
  492.             return t
  493.     return None
  494.  
  495.  
  496. # Dummy thread class to represent threads not started here.
  497. # These aren't garbage collected when they die,
  498. # nor can they be waited for.
  499. # Their purpose is to return *something* from currentThread().
  500. # They are marked as daemon threads so we won't wait for them
  501. # when we exit (conform previous semantics).
  502.  
  503. class _DummyThread(Thread):
  504.  
  505.     def __init__(self):
  506.         Thread.__init__(self, name=_newname("Dummy-%d"))
  507.         self._Thread__started = 1
  508.         _active_limbo_lock.acquire()
  509.         _active[_get_ident()] = self
  510.         _active_limbo_lock.release()
  511.  
  512.     def _set_daemon(self):
  513.         return 1
  514.  
  515.     def join(self):
  516.         assert 0, "cannot join a dummy thread"
  517.  
  518.  
  519. # Global API functions
  520.  
  521. def currentThread():
  522.     try:
  523.         return _active[_get_ident()]
  524.     except KeyError:
  525.         ##print "currentThread(): no current thread for", _get_ident()
  526.         return _DummyThread()
  527.  
  528. def activeCount():
  529.     _active_limbo_lock.acquire()
  530.     count = len(_active) + len(_limbo)
  531.     _active_limbo_lock.release()
  532.     return count
  533.  
  534. def enumerate():
  535.     _active_limbo_lock.acquire()
  536.     active = _active.values() + _limbo.values()
  537.     _active_limbo_lock.release()
  538.     return active
  539.  
  540.  
  541. # Create the main thread object
  542.  
  543. _MainThread()
  544.  
  545.  
  546. # Self-test code
  547.  
  548. def _test():
  549.  
  550.     import random
  551.  
  552.     class BoundedQueue(_Verbose):
  553.  
  554.         def __init__(self, limit):
  555.             _Verbose.__init__(self)
  556.             self.mon = RLock()
  557.             self.rc = Condition(self.mon)
  558.             self.wc = Condition(self.mon)
  559.             self.limit = limit
  560.             self.queue = []
  561.  
  562.         def put(self, item):
  563.             self.mon.acquire()
  564.             while len(self.queue) >= self.limit:
  565.                 self._note("put(%s): queue full", item)
  566.                 self.wc.wait()
  567.             self.queue.append(item)
  568.             self._note("put(%s): appended, length now %d",
  569.                        item, len(self.queue))
  570.             self.rc.notify()
  571.             self.mon.release()
  572.  
  573.         def get(self):
  574.             self.mon.acquire()
  575.             while not self.queue:
  576.                 self._note("get(): queue empty")
  577.                 self.rc.wait()
  578.             item = self.queue[0]
  579.             del self.queue[0]
  580.             self._note("get(): got %s, %d left", item, len(self.queue))
  581.             self.wc.notify()
  582.             self.mon.release()
  583.             return item
  584.  
  585.     class ProducerThread(Thread):
  586.  
  587.         def __init__(self, queue, quota):
  588.             Thread.__init__(self, name="Producer")
  589.             self.queue = queue
  590.             self.quota = quota
  591.  
  592.         def run(self):
  593.             from random import random
  594.             counter = 0
  595.             while counter < self.quota:
  596.                 counter = counter + 1
  597.                 self.queue.put("%s.%d" % (self.getName(), counter))
  598.                 _sleep(random() * 0.00001)
  599.  
  600.  
  601.     class ConsumerThread(Thread):
  602.  
  603.         def __init__(self, queue, count):
  604.             Thread.__init__(self, name="Consumer")
  605.             self.queue = queue
  606.             self.count = count
  607.  
  608.         def run(self):
  609.             while self.count > 0:
  610.                 item = self.queue.get()
  611.                 print item
  612.                 self.count = self.count - 1
  613.  
  614.     import time
  615.  
  616.     NP = 3
  617.     QL = 4
  618.     NI = 5
  619.  
  620.     Q = BoundedQueue(QL)
  621.     P = []
  622.     for i in range(NP):
  623.         t = ProducerThread(Q, NI)
  624.         t.setName("Producer-%d" % (i+1))
  625.         P.append(t)
  626.     C = ConsumerThread(Q, NI*NP)
  627.     for t in P:
  628.         t.start()
  629.         _sleep(0.000001)
  630.     C.start()
  631.     for t in P:
  632.         t.join()
  633.     C.join()
  634.  
  635. if __name__ == '__main__':
  636.     _test()
clone this paste RAW Paste Data