greenthread.py 11 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299
  1. from collections import deque
  2. import sys
  3. from eventlet import event
  4. from eventlet import hubs
  5. from eventlet import support
  6. from eventlet import timeout
  7. from eventlet.hubs import timer
  8. from eventlet.support import greenlets as greenlet
  9. import six
  10. import warnings
  11. __all__ = ['getcurrent', 'sleep', 'spawn', 'spawn_n',
  12. 'kill',
  13. 'spawn_after', 'spawn_after_local', 'GreenThread']
  14. getcurrent = greenlet.getcurrent
  15. def sleep(seconds=0):
  16. """Yield control to another eligible coroutine until at least *seconds* have
  17. elapsed.
  18. *seconds* may be specified as an integer, or a float if fractional seconds
  19. are desired. Calling :func:`~greenthread.sleep` with *seconds* of 0 is the
  20. canonical way of expressing a cooperative yield. For example, if one is
  21. looping over a large list performing an expensive calculation without
  22. calling any socket methods, it's a good idea to call ``sleep(0)``
  23. occasionally; otherwise nothing else will run.
  24. """
  25. hub = hubs.get_hub()
  26. current = getcurrent()
  27. assert hub.greenlet is not current, 'do not call blocking functions from the mainloop'
  28. timer = hub.schedule_call_global(seconds, current.switch)
  29. try:
  30. hub.switch()
  31. finally:
  32. timer.cancel()
  33. def spawn(func, *args, **kwargs):
  34. """Create a greenthread to run ``func(*args, **kwargs)``. Returns a
  35. :class:`GreenThread` object which you can use to get the results of the
  36. call.
  37. Execution control returns immediately to the caller; the created greenthread
  38. is merely scheduled to be run at the next available opportunity.
  39. Use :func:`spawn_after` to arrange for greenthreads to be spawned
  40. after a finite delay.
  41. """
  42. hub = hubs.get_hub()
  43. g = GreenThread(hub.greenlet)
  44. hub.schedule_call_global(0, g.switch, func, args, kwargs)
  45. return g
  46. def spawn_n(func, *args, **kwargs):
  47. """Same as :func:`spawn`, but returns a ``greenlet`` object from
  48. which it is not possible to retrieve either a return value or
  49. whether it raised any exceptions. This is faster than
  50. :func:`spawn`; it is fastest if there are no keyword arguments.
  51. If an exception is raised in the function, spawn_n prints a stack
  52. trace; the print can be disabled by calling
  53. :func:`eventlet.debug.hub_exceptions` with False.
  54. """
  55. return _spawn_n(0, func, args, kwargs)[1]
  56. def spawn_after(seconds, func, *args, **kwargs):
  57. """Spawns *func* after *seconds* have elapsed. It runs as scheduled even if
  58. the current greenthread has completed.
  59. *seconds* may be specified as an integer, or a float if fractional seconds
  60. are desired. The *func* will be called with the given *args* and
  61. keyword arguments *kwargs*, and will be executed within its own greenthread.
  62. The return value of :func:`spawn_after` is a :class:`GreenThread` object,
  63. which can be used to retrieve the results of the call.
  64. To cancel the spawn and prevent *func* from being called,
  65. call :meth:`GreenThread.cancel` on the return value of :func:`spawn_after`.
  66. This will not abort the function if it's already started running, which is
  67. generally the desired behavior. If terminating *func* regardless of whether
  68. it's started or not is the desired behavior, call :meth:`GreenThread.kill`.
  69. """
  70. hub = hubs.get_hub()
  71. g = GreenThread(hub.greenlet)
  72. hub.schedule_call_global(seconds, g.switch, func, args, kwargs)
  73. return g
  74. def spawn_after_local(seconds, func, *args, **kwargs):
  75. """Spawns *func* after *seconds* have elapsed. The function will NOT be
  76. called if the current greenthread has exited.
  77. *seconds* may be specified as an integer, or a float if fractional seconds
  78. are desired. The *func* will be called with the given *args* and
  79. keyword arguments *kwargs*, and will be executed within its own greenthread.
  80. The return value of :func:`spawn_after` is a :class:`GreenThread` object,
  81. which can be used to retrieve the results of the call.
  82. To cancel the spawn and prevent *func* from being called,
  83. call :meth:`GreenThread.cancel` on the return value. This will not abort the
  84. function if it's already started running. If terminating *func* regardless
  85. of whether it's started or not is the desired behavior, call
  86. :meth:`GreenThread.kill`.
  87. """
  88. hub = hubs.get_hub()
  89. g = GreenThread(hub.greenlet)
  90. hub.schedule_call_local(seconds, g.switch, func, args, kwargs)
  91. return g
  92. def call_after_global(seconds, func, *args, **kwargs):
  93. warnings.warn(
  94. "call_after_global is renamed to spawn_after, which"
  95. "has the same signature and semantics (plus a bit extra). Please do a"
  96. " quick search-and-replace on your codebase, thanks!",
  97. DeprecationWarning, stacklevel=2)
  98. return _spawn_n(seconds, func, args, kwargs)[0]
  99. def call_after_local(seconds, function, *args, **kwargs):
  100. warnings.warn(
  101. "call_after_local is renamed to spawn_after_local, which"
  102. "has the same signature and semantics (plus a bit extra).",
  103. DeprecationWarning, stacklevel=2)
  104. hub = hubs.get_hub()
  105. g = greenlet.greenlet(function, parent=hub.greenlet)
  106. t = hub.schedule_call_local(seconds, g.switch, *args, **kwargs)
  107. return t
  108. call_after = call_after_local
  109. def exc_after(seconds, *throw_args):
  110. warnings.warn("Instead of exc_after, which is deprecated, use "
  111. "Timeout(seconds, exception)",
  112. DeprecationWarning, stacklevel=2)
  113. if seconds is None: # dummy argument, do nothing
  114. return timer.Timer(seconds, lambda: None)
  115. hub = hubs.get_hub()
  116. return hub.schedule_call_local(seconds, getcurrent().throw, *throw_args)
  117. # deprecate, remove
  118. TimeoutError, with_timeout = (
  119. support.wrap_deprecated(old, new)(fun) for old, new, fun in (
  120. ('greenthread.TimeoutError', 'Timeout', timeout.Timeout),
  121. ('greenthread.with_timeout', 'with_timeout', timeout.with_timeout),
  122. ))
  123. def _spawn_n(seconds, func, args, kwargs):
  124. hub = hubs.get_hub()
  125. g = greenlet.greenlet(func, parent=hub.greenlet)
  126. t = hub.schedule_call_global(seconds, g.switch, *args, **kwargs)
  127. return t, g
  128. class GreenThread(greenlet.greenlet):
  129. """The GreenThread class is a type of Greenlet which has the additional
  130. property of being able to retrieve the return value of the main function.
  131. Do not construct GreenThread objects directly; call :func:`spawn` to get one.
  132. """
  133. def __init__(self, parent):
  134. greenlet.greenlet.__init__(self, self.main, parent)
  135. self._exit_event = event.Event()
  136. self._resolving_links = False
  137. def wait(self):
  138. """ Returns the result of the main function of this GreenThread. If the
  139. result is a normal return value, :meth:`wait` returns it. If it raised
  140. an exception, :meth:`wait` will raise the same exception (though the
  141. stack trace will unavoidably contain some frames from within the
  142. greenthread module)."""
  143. return self._exit_event.wait()
  144. def link(self, func, *curried_args, **curried_kwargs):
  145. """ Set up a function to be called with the results of the GreenThread.
  146. The function must have the following signature::
  147. def func(gt, [curried args/kwargs]):
  148. When the GreenThread finishes its run, it calls *func* with itself
  149. and with the `curried arguments <http://en.wikipedia.org/wiki/Currying>`_ supplied
  150. at link-time. If the function wants to retrieve the result of the GreenThread,
  151. it should call wait() on its first argument.
  152. Note that *func* is called within execution context of
  153. the GreenThread, so it is possible to interfere with other linked
  154. functions by doing things like switching explicitly to another
  155. greenthread.
  156. """
  157. self._exit_funcs = getattr(self, '_exit_funcs', deque())
  158. self._exit_funcs.append((func, curried_args, curried_kwargs))
  159. if self._exit_event.ready():
  160. self._resolve_links()
  161. def unlink(self, func, *curried_args, **curried_kwargs):
  162. """ remove linked function set by :meth:`link`
  163. Remove successfully return True, otherwise False
  164. """
  165. if not getattr(self, '_exit_funcs', None):
  166. return False
  167. try:
  168. self._exit_funcs.remove((func, curried_args, curried_kwargs))
  169. return True
  170. except ValueError:
  171. return False
  172. def main(self, function, args, kwargs):
  173. try:
  174. result = function(*args, **kwargs)
  175. except:
  176. self._exit_event.send_exception(*sys.exc_info())
  177. self._resolve_links()
  178. raise
  179. else:
  180. self._exit_event.send(result)
  181. self._resolve_links()
  182. def _resolve_links(self):
  183. # ca and ckw are the curried function arguments
  184. if self._resolving_links:
  185. return
  186. self._resolving_links = True
  187. try:
  188. exit_funcs = getattr(self, '_exit_funcs', deque())
  189. while exit_funcs:
  190. f, ca, ckw = exit_funcs.popleft()
  191. f(self, *ca, **ckw)
  192. finally:
  193. self._resolving_links = False
  194. def kill(self, *throw_args):
  195. """Kills the greenthread using :func:`kill`. After being killed
  196. all calls to :meth:`wait` will raise *throw_args* (which default
  197. to :class:`greenlet.GreenletExit`)."""
  198. return kill(self, *throw_args)
  199. def cancel(self, *throw_args):
  200. """Kills the greenthread using :func:`kill`, but only if it hasn't
  201. already started running. After being canceled,
  202. all calls to :meth:`wait` will raise *throw_args* (which default
  203. to :class:`greenlet.GreenletExit`)."""
  204. return cancel(self, *throw_args)
  205. def cancel(g, *throw_args):
  206. """Like :func:`kill`, but only terminates the greenthread if it hasn't
  207. already started execution. If the grenthread has already started
  208. execution, :func:`cancel` has no effect."""
  209. if not g:
  210. kill(g, *throw_args)
  211. def kill(g, *throw_args):
  212. """Terminates the target greenthread by raising an exception into it.
  213. Whatever that greenthread might be doing; be it waiting for I/O or another
  214. primitive, it sees an exception right away.
  215. By default, this exception is GreenletExit, but a specific exception
  216. may be specified. *throw_args* should be the same as the arguments to
  217. raise; either an exception instance or an exc_info tuple.
  218. Calling :func:`kill` causes the calling greenthread to cooperatively yield.
  219. """
  220. if g.dead:
  221. return
  222. hub = hubs.get_hub()
  223. if not g:
  224. # greenlet hasn't started yet and therefore throw won't work
  225. # on its own; semantically we want it to be as though the main
  226. # method never got called
  227. def just_raise(*a, **kw):
  228. if throw_args:
  229. six.reraise(throw_args[0], throw_args[1], throw_args[2])
  230. else:
  231. raise greenlet.GreenletExit()
  232. g.run = just_raise
  233. if isinstance(g, GreenThread):
  234. # it's a GreenThread object, so we want to call its main
  235. # method to take advantage of the notification
  236. try:
  237. g.main(just_raise, (), {})
  238. except:
  239. pass
  240. current = getcurrent()
  241. if current is not hub.greenlet:
  242. # arrange to wake the caller back up immediately
  243. hub.ensure_greenlet()
  244. hub.schedule_call_global(0, current.switch)
  245. g.throw(*throw_args)