eventlet/eventlet/hubs/__init__.py

   1 import sys
   2 import os
   3 from eventlet.support import greenlets as greenlet, six
   4 from eventlet import patcher
   5
   6 try:
   7     # try and import pkg_resources ...
   8     import pkg_resources
   9 except ImportError:
  10     # ... but do not depend on it
  11     pkg_resources = None
  12
  13
  14 __all__ = ["use_hub", "get_hub", "get_default_hub", "trampoline"]
  15
  16 threading = patcher.original('threading')
  17 _threadlocal = threading.local()
  18
  19
  20 def get_default_hub():
  21     """Select the default hub implementation based on what multiplexing
  22     libraries are installed.  The order that the hubs are tried is:
  23
  24     * epoll
  25     * kqueue
  26     * poll
  27     * select
  28
  29     It won't automatically select the pyevent hub, because it's not
  30     python-thread-safe.
  31
  32     .. include:: ../doc/common.txt
  33     .. note :: |internal|
  34     """
  35
  36     # pyevent hub disabled for now because it is not thread-safe
  37     # try:
  38     #    import eventlet.hubs.pyevent
  39     #    return eventlet.hubs.pyevent
  40     # except:
  41     #    pass
  42
  43     select = patcher.original('select')
  44     try:
  45         import eventlet.hubs.epolls
  46         return eventlet.hubs.epolls
  47     except ImportError:
  48         try:
  49             import eventlet.hubs.kqueue
  50             return eventlet.hubs.kqueue
  51         except ImportError:
  52             if hasattr(select, 'poll'):
  53                 import eventlet.hubs.poll
  54                 return eventlet.hubs.poll
  55             else:
  56                 import eventlet.hubs.selects
  57                 return eventlet.hubs.selects
  58
  59
  60 def use_hub(mod=None):
  61     """Use the module *mod*, containing a class called Hub, as the
  62     event hub. Usually not required; the default hub is usually fine.
  63
  64     Mod can be an actual module, a string, or None.  If *mod* is a module,
  65     it uses it directly.   If *mod* is a string and contains either '.' or ':'
  66     use_hub tries to import the hub using the 'package.subpackage.module:Class'
  67     convention, otherwise use_hub looks for a matching setuptools entry point
  68     in the 'eventlet.hubs' group to load or finally tries to import
  69     `eventlet.hubs.mod` and use that as the hub module.  If *mod* is None,
  70     use_hub uses the default hub.  Only call use_hub during application
  71     initialization,  because it resets the hub's state and any existing
  72     timers or listeners will never be resumed.
  73     """
  74     if mod is None:
  75         mod = os.environ.get('EVENTLET_HUB', None)
  76     if mod is None:
  77         mod = get_default_hub()
  78     if hasattr(_threadlocal, 'hub'):
  79         del _threadlocal.hub
  80     if isinstance(mod, six.string_types):
  81         assert mod.strip(), "Need to specify a hub"
  82         if '.' in mod or ':' in mod:
  83             modulename, _, classname = mod.strip().partition(':')
  84             mod = __import__(modulename, globals(), locals(), [classname])
  85             if classname:
  86                 mod = getattr(mod, classname)
  87         else:
  88             found = False
  89             if pkg_resources is not None:
  90                 for entry in pkg_resources.iter_entry_points(
  91                         group='eventlet.hubs', name=mod):
  92                     mod, found = entry.load(), True
  93                     break
  94             if not found:
  95                 mod = __import__(
  96                     'eventlet.hubs.' + mod, globals(), locals(), ['Hub'])
  97     if hasattr(mod, 'Hub'):
  98         _threadlocal.Hub = mod.Hub
  99     else:
 100         _threadlocal.Hub = mod
 101
 102
 103 def get_hub():
 104     """Get the current event hub singleton object.
 105
 106     .. note :: |internal|
 107     """
 108     try:
 109         hub = _threadlocal.hub
 110     except AttributeError:
 111         try:
 112             _threadlocal.Hub
 113         except AttributeError:
 114             use_hub()
 115         hub = _threadlocal.hub = _threadlocal.Hub()
 116     return hub
 117
 118 from eventlet import timeout
 119
 120
 121 def trampoline(fd, read=None, write=None, timeout=None,
 122                timeout_exc=timeout.Timeout,
 123                mark_as_closed = None):
 124     """Suspend the current coroutine until the given socket object or file
 125     descriptor is ready to *read*, ready to *write*, or the specified
 126     *timeout* elapses, depending on arguments specified.
 127
 128     To wait for *fd* to be ready to read, pass *read* ``=True``; ready to
 129     write, pass *write* ``=True``. To specify a timeout, pass the *timeout*
 130     argument in seconds.
 131
 132     If the specified *timeout* elapses before the socket is ready to read or
 133     write, *timeout_exc* will be raised instead of ``trampoline()``
 134     returning normally.
 135
 136     .. note :: |internal|
 137     """
 138     t = None
 139     hub = get_hub()
 140     current = greenlet.getcurrent()
 141     assert hub.greenlet is not current, 'do not call blocking functions from the mainloop'
 142     assert not (
 143         read and write), 'not allowed to trampoline for reading and writing'
 144     try:
 145         fileno = fd.fileno()
 146     except AttributeError:
 147         fileno = fd
 148     if timeout is not None:
 149         def _timeout(exc):
 150             # This is only useful to insert debugging
 151             current.throw(exc)
 152         t = hub.schedule_call_global(timeout, _timeout, timeout_exc)
 153     try:
 154         if read:
 155             listener = hub.add(hub.READ, fileno, current.switch, current.throw, mark_as_closed)
 156         elif write:
 157             listener = hub.add(hub.WRITE, fileno, current.switch, current.throw, mark_as_closed)
 158         try:
 159             return hub.switch()
 160         finally:
 161             hub.remove(listener)
 162     finally:
 163         if t is not None:
 164             t.cancel()
 165
 166 def notify_close(fd):
 167     """
 168     A particular file descriptor has been explicitly closed. Register for any
 169     waiting listeners to be notified on the next run loop.
 170     """
 171     hub = get_hub()
 172     hub.notify_close(fd)
 173
 174 def notify_opened(fd):
 175     """
 176     Some file descriptors may be closed 'silently' - that is, by the garbage
 177     collector, by an external library, etc. When the OS returns a file descriptor
 178     from an open call (or something similar), this may be the only indication we
 179     have that the FD has been closed and then recycled.
 180     We let the hub know that the old file descriptor is dead; any stuck listeners
 181     will be disabled and notified in turn.
 182     """
 183     hub = get_hub()
 184     hub.mark_as_reopened(fd)
 185
 186
 187 class IOClosed(IOError):
 188     pass