eventlet/eventlet/greenthread.py

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