eventlet/eventlet/hubs/__init__.py

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