event.py 7.3 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220
  1. from __future__ import print_function
  2. from eventlet import hubs
  3. from eventlet.support import greenlets as greenlet
  4. __all__ = ['Event']
  5. class NOT_USED:
  6. def __repr__(self):
  7. return 'NOT_USED'
  8. NOT_USED = NOT_USED()
  9. class Event(object):
  10. """An abstraction where an arbitrary number of coroutines
  11. can wait for one event from another.
  12. Events are similar to a Queue that can only hold one item, but differ
  13. in two important ways:
  14. 1. calling :meth:`send` never unschedules the current greenthread
  15. 2. :meth:`send` can only be called once; create a new event to send again.
  16. They are good for communicating results between coroutines, and
  17. are the basis for how
  18. :meth:`GreenThread.wait() <eventlet.greenthread.GreenThread.wait>`
  19. is implemented.
  20. >>> from eventlet import event
  21. >>> import eventlet
  22. >>> evt = event.Event()
  23. >>> def baz(b):
  24. ... evt.send(b + 1)
  25. ...
  26. >>> _ = eventlet.spawn_n(baz, 3)
  27. >>> evt.wait()
  28. 4
  29. """
  30. _result = None
  31. _exc = None
  32. def __init__(self):
  33. self._waiters = set()
  34. self.reset()
  35. def __str__(self):
  36. params = (self.__class__.__name__, hex(id(self)),
  37. self._result, self._exc, len(self._waiters))
  38. return '<%s at %s result=%r _exc=%r _waiters[%d]>' % params
  39. def reset(self):
  40. # this is kind of a misfeature and doesn't work perfectly well,
  41. # it's better to create a new event rather than reset an old one
  42. # removing documentation so that we don't get new use cases for it
  43. assert self._result is not NOT_USED, 'Trying to re-reset() a fresh event.'
  44. self._result = NOT_USED
  45. self._exc = None
  46. def ready(self):
  47. """ Return true if the :meth:`wait` call will return immediately.
  48. Used to avoid waiting for things that might take a while to time out.
  49. For example, you can put a bunch of events into a list, and then visit
  50. them all repeatedly, calling :meth:`ready` until one returns ``True``,
  51. and then you can :meth:`wait` on that one."""
  52. return self._result is not NOT_USED
  53. def has_exception(self):
  54. return self._exc is not None
  55. def has_result(self):
  56. return self._result is not NOT_USED and self._exc is None
  57. def poll(self, notready=None):
  58. if self.ready():
  59. return self.wait()
  60. return notready
  61. # QQQ make it return tuple (type, value, tb) instead of raising
  62. # because
  63. # 1) "poll" does not imply raising
  64. # 2) it's better not to screw up caller's sys.exc_info() by default
  65. # (e.g. if caller wants to calls the function in except or finally)
  66. def poll_exception(self, notready=None):
  67. if self.has_exception():
  68. return self.wait()
  69. return notready
  70. def poll_result(self, notready=None):
  71. if self.has_result():
  72. return self.wait()
  73. return notready
  74. def wait(self, timeout=None):
  75. """Wait until another coroutine calls :meth:`send`.
  76. Returns the value the other coroutine passed to :meth:`send`.
  77. >>> import eventlet
  78. >>> evt = eventlet.Event()
  79. >>> def wait_on():
  80. ... retval = evt.wait()
  81. ... print("waited for {0}".format(retval))
  82. >>> _ = eventlet.spawn(wait_on)
  83. >>> evt.send('result')
  84. >>> eventlet.sleep(0)
  85. waited for result
  86. Returns immediately if the event has already occurred.
  87. >>> evt.wait()
  88. 'result'
  89. When the timeout argument is present and not None, it should be a floating point number
  90. specifying a timeout for the operation in seconds (or fractions thereof).
  91. """
  92. current = greenlet.getcurrent()
  93. if self._result is NOT_USED:
  94. hub = hubs.get_hub()
  95. self._waiters.add(current)
  96. timer = None
  97. if timeout is not None:
  98. timer = hub.schedule_call_local(timeout, self._do_send, None, None, current)
  99. try:
  100. result = hub.switch()
  101. if timer is not None:
  102. timer.cancel()
  103. return result
  104. finally:
  105. self._waiters.discard(current)
  106. if self._exc is not None:
  107. current.throw(*self._exc)
  108. return self._result
  109. def send(self, result=None, exc=None):
  110. """Makes arrangements for the waiters to be woken with the
  111. result and then returns immediately to the parent.
  112. >>> from eventlet import event
  113. >>> import eventlet
  114. >>> evt = event.Event()
  115. >>> def waiter():
  116. ... print('about to wait')
  117. ... result = evt.wait()
  118. ... print('waited for {0}'.format(result))
  119. >>> _ = eventlet.spawn(waiter)
  120. >>> eventlet.sleep(0)
  121. about to wait
  122. >>> evt.send('a')
  123. >>> eventlet.sleep(0)
  124. waited for a
  125. It is an error to call :meth:`send` multiple times on the same event.
  126. >>> evt.send('whoops')
  127. Traceback (most recent call last):
  128. ...
  129. AssertionError: Trying to re-send() an already-triggered event.
  130. Use :meth:`reset` between :meth:`send` s to reuse an event object.
  131. """
  132. assert self._result is NOT_USED, 'Trying to re-send() an already-triggered event.'
  133. self._result = result
  134. if exc is not None and not isinstance(exc, tuple):
  135. exc = (exc, )
  136. self._exc = exc
  137. hub = hubs.get_hub()
  138. for waiter in self._waiters:
  139. hub.schedule_call_global(
  140. 0, self._do_send, self._result, self._exc, waiter)
  141. def _do_send(self, result, exc, waiter):
  142. if waiter in self._waiters:
  143. if exc is None:
  144. waiter.switch(result)
  145. else:
  146. waiter.throw(*exc)
  147. def send_exception(self, *args):
  148. """Same as :meth:`send`, but sends an exception to waiters.
  149. The arguments to send_exception are the same as the arguments
  150. to ``raise``. If a single exception object is passed in, it
  151. will be re-raised when :meth:`wait` is called, generating a
  152. new stacktrace.
  153. >>> from eventlet import event
  154. >>> evt = event.Event()
  155. >>> evt.send_exception(RuntimeError())
  156. >>> evt.wait()
  157. Traceback (most recent call last):
  158. File "<stdin>", line 1, in <module>
  159. File "eventlet/event.py", line 120, in wait
  160. current.throw(*self._exc)
  161. RuntimeError
  162. If it's important to preserve the entire original stack trace,
  163. you must pass in the entire :func:`sys.exc_info` tuple.
  164. >>> import sys
  165. >>> evt = event.Event()
  166. >>> try:
  167. ... raise RuntimeError()
  168. ... except RuntimeError:
  169. ... evt.send_exception(*sys.exc_info())
  170. ...
  171. >>> evt.wait()
  172. Traceback (most recent call last):
  173. File "<stdin>", line 1, in <module>
  174. File "eventlet/event.py", line 120, in wait
  175. current.throw(*self._exc)
  176. File "<stdin>", line 2, in <module>
  177. RuntimeError
  178. Note that doing so stores a traceback object directly on the
  179. Event object, which may cause reference cycles. See the
  180. :func:`sys.exc_info` documentation.
  181. """
  182. # the arguments and the same as for greenlet.throw
  183. return self.send(None, args)