synchronize.py 8.8 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296
  1. #
  2. # Module implementing synchronization primitives
  3. #
  4. # processing/synchronize.py
  5. #
  6. # Copyright (c) 2006-2008, R Oudkerk --- see COPYING.txt
  7. #
  8. __all__ = [ 'Lock', 'RLock', 'Semaphore', 'BoundedSemaphore', 'Condition',
  9. 'Event']
  10. import threading
  11. import os
  12. import sys
  13. from struct import pack as _pack, unpack as _unpack, calcsize as _calcsize
  14. from time import time as _time, sleep as _sleep
  15. from processing import _processing
  16. from processing.process import currentProcess, _registerAfterFork
  17. from processing.logger import debug
  18. from processing.finalize import Finalize
  19. from processing.forking import assertSpawning
  20. #
  21. # Constants to describe the kind of blocker (normal lock is a bounded sem)
  22. #
  23. RECURSIVE_MUTEX, SEMAPHORE, BOUNDED_SEMAPHORE = range(3)
  24. #
  25. # Base class for semaphores and mutexes; wraps `_processing.SemLock`
  26. #
  27. class SemLock(object):
  28. def __init__(self, kind, value):
  29. sl = self._semlock = _processing.SemLock(kind, value)
  30. debug('created semlock with handle %s' % sl.handle)
  31. self.__setstate__((sl.handle, sl.kind, sl.maxvalue))
  32. if sys.platform != 'win32':
  33. def _afterFork(obj):
  34. obj._semlock._afterFork()
  35. _registerAfterFork(self, _afterFork)
  36. def __getstate__(self):
  37. assertSpawning(self)
  38. return self._state
  39. def __setstate__(self, state):
  40. self._state = state
  41. if not hasattr(self, '_semlock'):
  42. self._semlock = _processing.SemLock._rebuild(*state)
  43. debug('recreated blocker with handle %r' % state[0])
  44. self.acquire = self._semlock.acquire
  45. self.release = self._semlock.release
  46. self.__enter__ = self._semlock.__enter__
  47. self.__exit__ = self._semlock.__exit__
  48. #
  49. # Semaphore
  50. #
  51. class Semaphore(SemLock):
  52. def __init__(self, value=1):
  53. SemLock.__init__(self, SEMAPHORE, value)
  54. def getValue(self):
  55. return self._semlock._getValue()
  56. def __repr__(self):
  57. try:
  58. value = self._semlock._getValue()
  59. except (KeyboardInterrupt, SystemExit):
  60. raise
  61. except:
  62. value = 'unknown'
  63. return '<Semaphore(value=%s)>' % value
  64. #
  65. # Bounded semaphore
  66. #
  67. class BoundedSemaphore(Semaphore):
  68. def __init__(self, value=1):
  69. SemLock.__init__(self, BOUNDED_SEMAPHORE, value)
  70. def __repr__(self):
  71. try:
  72. value = self._semlock._getValue()
  73. except (KeyboardInterrupt, SystemExit):
  74. raise
  75. except:
  76. value = 'unknown'
  77. return '<BoundedSemaphore(value=%s, maxvalue=%s)>' % \
  78. (value, self._semlock.maxvalue)
  79. #
  80. # Non-recursive lock
  81. #
  82. class Lock(SemLock):
  83. def __init__(self):
  84. SemLock.__init__(self, BOUNDED_SEMAPHORE, 1)
  85. def __repr__(self):
  86. try:
  87. if self._semlock._isMine():
  88. name = currentProcess().getName()
  89. if threading.currentThread().getName() != 'MainThread':
  90. name += '|' + threading.currentThread().getName()
  91. elif self._semlock._getValue() == 1:
  92. name = 'None'
  93. elif self._semlock._count() > 0:
  94. name = 'SomeOtherThread'
  95. else:
  96. name = 'SomeOtherProcess'
  97. except (KeyboardInterrupt, SystemExit):
  98. raise
  99. except Exception:
  100. name = 'unknown'
  101. return '<Lock(owner=%s)>' % name
  102. #
  103. # Recursive lock
  104. #
  105. class RLock(SemLock):
  106. def __init__(self):
  107. SemLock.__init__(self, RECURSIVE_MUTEX, 1)
  108. def __repr__(self):
  109. try:
  110. if self._semlock._isMine():
  111. name = currentProcess().getName()
  112. if threading.currentThread().getName() != 'MainThread':
  113. name += '|' + threading.currentThread().getName()
  114. count = self._semlock._count()
  115. elif self._semlock._getValue() == 1:
  116. name, count = 'None', 0
  117. elif self._semlock._count() > 0:
  118. name, count = 'SomeOtherThread', 'nonzero'
  119. else:
  120. name, count = 'SomeOtherProcess', 'nonzero'
  121. except (KeyboardInterrupt, SystemExit):
  122. raise
  123. except Exception:
  124. name, count = 'unknown', 'unknown'
  125. return '<RLock(%s, %s)>' % (name, count)
  126. #
  127. # Condition variable
  128. #
  129. class Condition(object):
  130. def __init__(self, lock=None):
  131. state = (lock or RLock(), Semaphore(0), Semaphore(0), Semaphore(0))
  132. self.__setstate__(state)
  133. def __getstate__(self):
  134. assertSpawning(self)
  135. return self._state
  136. def __setstate__(self, state):
  137. (self._lock, self._sleeping_count,
  138. self._woken_count, self._wait_semaphore) = self._state = state
  139. self.acquire = self._lock.acquire
  140. self.release = self._lock.release
  141. self.__enter__ = self._lock.__enter__
  142. self.__exit__ = self._lock.__exit__
  143. def __repr__(self):
  144. try:
  145. num_waiters = (self._sleeping_count._semlock._getValue() -
  146. self._woken_count._semlock._getValue())
  147. except (KeyboardInterrupt, SystemExit):
  148. raise
  149. except Exception:
  150. num_waiters = 'unkown'
  151. return '<Condition(%s, %s)>' % (self._lock, num_waiters)
  152. def wait(self, timeout=None):
  153. assert self._lock._semlock._isMine(), \
  154. 'must acquire() condition before using wait()'
  155. # indicate that this thread is going to sleep
  156. self._sleeping_count.release()
  157. # release lock
  158. count = self._lock._semlock._count()
  159. for i in xrange(count):
  160. self._lock.release()
  161. try:
  162. # wait for notification or timeout
  163. self._wait_semaphore.acquire(True, timeout)
  164. finally:
  165. # indicate that this thread has woken
  166. self._woken_count.release()
  167. # reacquire lock
  168. for i in xrange(count):
  169. self._lock.acquire()
  170. def notify(self):
  171. assert self._lock._semlock._isMine(), 'lock is not owned'
  172. assert not self._wait_semaphore.acquire(False)
  173. # to take account of timeouts since last notify() we subtract
  174. # woken_count from sleeping_count and rezero woken_count
  175. while self._woken_count.acquire(False):
  176. res = self._sleeping_count.acquire(False)
  177. assert res
  178. if self._sleeping_count.acquire(False): # try grabbing a sleeper
  179. self._wait_semaphore.release() # wake up one sleeper
  180. self._woken_count.acquire() # wait for the sleeper to wake
  181. # rezero _wait_semaphore in case a timeout just happened
  182. self._wait_semaphore.acquire(False)
  183. def notifyAll(self):
  184. assert self._lock._semlock._isMine(), 'lock is not owned'
  185. assert not self._wait_semaphore.acquire(False)
  186. # to take account of timeouts since last notify*() we subtract
  187. # woken_count from sleeping_count and rezero woken_count
  188. while self._woken_count.acquire(False):
  189. res = self._sleeping_count.acquire(False)
  190. assert res
  191. sleepers = 0
  192. while self._sleeping_count.acquire(False):
  193. self._wait_semaphore.release() # wake up one sleeper
  194. sleepers += 1
  195. if sleepers:
  196. for i in xrange(sleepers):
  197. self._woken_count.acquire() # wait for a sleeper to wake
  198. # rezero wait_semaphore in case some timeouts just happened
  199. while self._wait_semaphore.acquire(False):
  200. pass
  201. #
  202. # Event
  203. #
  204. class Event(object):
  205. def __init__(self):
  206. self._cond = Condition(Lock())
  207. self._flag = Semaphore(0)
  208. def isSet(self):
  209. self._cond.acquire()
  210. try:
  211. if self._flag.acquire(False):
  212. self._flag.release()
  213. return True
  214. return False
  215. finally:
  216. self._cond.release()
  217. def set(self):
  218. self._cond.acquire()
  219. try:
  220. self._flag.acquire(False)
  221. self._flag.release()
  222. self._cond.notifyAll()
  223. finally:
  224. self._cond.release()
  225. def clear(self):
  226. self._cond.acquire()
  227. try:
  228. self._flag.acquire(False)
  229. finally:
  230. self._cond.release()
  231. def wait(self, timeout=None):
  232. self._cond.acquire()
  233. try:
  234. if self._flag.acquire(False):
  235. self._flag.release()
  236. else:
  237. self._cond.wait(timeout)
  238. finally:
  239. self._cond.release()