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