aioeventlet.py

import eventlet.hubs.hub
import greenlet
import logging
import signal
import sys
socket = eventlet.patcher.original('socket')
threading = eventlet.patcher.original('threading')

logger = logging.getLogger('aioeventlet')

try:
    import asyncio
    try:
        import selectors
    except ImportError:
        pass
    else:
        asyncio.selectors = selectors

    if sys.platform == 'win32':
        from asyncio.windows_utils import socketpair
    else:
        socketpair = socket.socketpair
except ImportError:
    import trollius as asyncio

    if sys.platform == 'win32':
        from trollius.windows_utils import socketpair
    else:
        socketpair = socket.socketpair

if eventlet.patcher.is_monkey_patched('socket'):
    # trollius must use call original socket and threading functions.
    # Examples: socket.socket(), socket.socketpair(),
    # threading.current_thread().
    asyncio.base_events.socket = socket
    asyncio.base_events.threading = threading
    if hasattr(threading, 'get_ident'):
        asyncio.base_events._get_thread_ident = threading.get_ident
    else:
        # Python 2
        asyncio.base_events._get_thread_ident = threading._get_ident
    asyncio.events.threading = threading
    if sys.platform == 'win32':
        asyncio.windows_events.socket = socket
        asyncio.windows_utils.socket = socket
    else:
        asyncio.unix_events.socket = socket
        asyncio.unix_events.threading = threading
    # FIXME: patch also trollius.py3_ssl

    # BaseDefaultEventLoopPolicy._Local must inherit from threading.local
    # of the original threading module, not the patched threading module
    class _Local(threading.local):
        _loop = None
        _set_called = False

    asyncio.events.BaseDefaultEventLoopPolicy._Local = _Local

_EVENT_READ = asyncio.selectors.EVENT_READ
_EVENT_WRITE = asyncio.selectors.EVENT_WRITE
_HUB_READ = eventlet.hubs.hub.READ
_HUB_WRITE = eventlet.hubs.hub.WRITE

# Eventlet 0.15 or newer?
_EVENTLET15 = hasattr(eventlet.hubs.hub.noop, 'mark_as_closed')


class _TpoolExecutor(object):
    def __init__(self, loop):
        import eventlet.tpool
        self._loop = loop
        self._tpool = eventlet.tpool

    def submit(self, fn, *args, **kwargs):
        f = asyncio.Future(loop=self._loop)
        try:
            res = self._tpool.execute(fn, *args, **kwargs)
        except Exception as exc:
            f.set_exception(exc)
        else:
            f.set_result(res)
        return f

    def shutdown(self, wait=True):
        self._tpool.killall()


class _Selector(asyncio.selectors._BaseSelectorImpl):
    def __init__(self, loop, hub):
        super(_Selector, self).__init__()
        # fd => events
        self._notified = {}
        self._loop = loop
        self._hub = hub
        # eventlet.event.Event() used by FD notifiers to wake up select()
        self._event = None

    def close(self):
        keys = list(self.get_map().values())
        for key in keys:
            self.unregister(key.fd)
        super(_Selector, self).close()

    def _add(self, fd, event):
        if event == _EVENT_READ:
            event_type = _HUB_READ
            func = self._notify_read
        else:
            event_type = _HUB_WRITE
            func = self._notify_write

        if _EVENTLET15:
            self._hub.add(event_type, fd, func, self._throwback, None)
        else:
            self._hub.add(event_type, fd, func)

    def register(self, fileobj, events, data=None):
        key = super(_Selector, self).register(fileobj, events, data)
        if events & _EVENT_READ:
            self._add(key.fd, _EVENT_READ)
        if events & _EVENT_WRITE:
            self._add(key.fd, _EVENT_WRITE)
        return key

    def _remove(self, fd, event):
        if event == _EVENT_READ:
            event_type = _HUB_READ
        else:
            event_type = _HUB_WRITE
        try:
            listener = self._hub.listeners[event_type][fd]
        except KeyError:
            pass
        else:
            self._hub.remove(listener)

    def unregister(self, fileobj):
        key = super(_Selector, self).unregister(fileobj)
        self._remove(key.fd, _EVENT_READ)
        self._remove(key.fd, _EVENT_WRITE)
        return key

    def _notify(self, fd, event):
        if fd in self._notified:
            self._notified[fd] |= event
        else:
            self._notified[fd] = event
        if self._event is not None and not self._event.ready():
            # wakeup the select() method
            self._event.send("ready")

    def _notify_read(self, fd):
        self._notify(fd, _EVENT_READ)

    def _notify_write(self, fd):
        self._notify(fd, _EVENT_WRITE)

    def _throwback(self, fd):
        # FIXME: do something with the FD in this case?
        pass

    def _read_events(self):
        notified = self._notified
        self._notified = {}
        ready = []
        for fd, events in notified.items():
            try:
                key = self.get_key(fd)
            except KeyError:
                continue
            ready.append((key, events & key.events))
        return ready

    def select(self, timeout):
        events = self._read_events()
        if events:
            return events

        self._event = eventlet.event.Event()
        try:
            if timeout is not None:
                def timeout_cb(event):
                    if event.ready():
                        return
                    event.send('timeout')

                eventlet.spawn_after(timeout, timeout_cb, self._event)

                self._event.wait()
                # FIXME: cancel the timeout_cb if wait() returns 'ready'?
            else:
                # blocking call
                self._event.wait()
            return self._read_events()
        finally:
            self._event = None


class EventLoop(asyncio.SelectorEventLoop):
    def __init__(self):
        self._greenthread = None

        # Store a reference to the hub to ensure
        # that we always use the same hub
        self._hub = eventlet.hubs.get_hub()

        selector = _Selector(self, self._hub)

        super(EventLoop, self).__init__(selector=selector)

        # Force a call to set_debug() to set hub.debug_blocking
        self.set_debug(self.get_debug())

        if eventlet.patcher.is_monkey_patched('thread'):
            self._default_executor = _TpoolExecutor(self)

    def stop(self):
        super(EventLoop, self).stop()
        # selector.select() is running: write into the self-pipe to wake up
        # the selector
        self._write_to_self()

    def call_soon(self, callback, *args, **kwargs):
        handle = super(EventLoop, self).call_soon(callback, *args, **kwargs)
        if self._selector is not None and self._selector._event:
            # selector.select() is running: write into the self-pipe to wake up
            # the selector
            self._write_to_self()
        return handle

    def call_at(self, when, callback, *args, **kwargs):
        handle = super(EventLoop, self).call_at(when, callback, *args, **kwargs)
        if self._selector is not None and self._selector._event:
            # selector.select() is running: write into the self-pipe to wake up
            # the selector
            self._write_to_self()
        return handle

    def set_debug(self, debug):
        super(EventLoop, self).set_debug(debug)

        self._hub.debug_exceptions = debug

        # Detect blocking eventlet functions. The feature is implemented with
        # signal.alarm() which is is not available on Windows. Signal handlers
        # can only be set from the main loop. So detecting blocking functions
        # cannot be used on Windows nor from a thread different than the main
        # thread.
        self._hub.debug_blocking = (
            debug
            and (sys.platform != 'win32')
            and isinstance(threading.current_thread(), threading._MainThread))

        if (self._hub.debug_blocking
        and hasattr(self, 'slow_callback_duration')):
            self._hub.debug_blocking_resolution = self.slow_callback_duration

    def run_forever(self):
        self._greenthread = eventlet.getcurrent()
        try:
            super(EventLoop, self).run_forever()
        finally:
            if self._hub.debug_blocking:
                # eventlet event loop is still running: cancel the current
                # detection of blocking tasks
                signal.alarm(0)
            self._greenthread = None

    def time(self):
        return self._hub.clock()


class EventLoopPolicy(asyncio.DefaultEventLoopPolicy):
    _loop_factory = EventLoop


def wrap_greenthread(gt, loop=None):
    """Wrap an eventlet GreenThread, or a greenlet, into a Future object.

    The Future object waits for the completion of a greenthread. The result
    or the exception of the greenthread will be stored in the Future object.

    The greenthread must be wrapped before its execution starts. If the
    greenthread is running or already finished, an exception is raised.

    For greenlets, the run attribute must be set.
    """
    if loop is None:
        loop = asyncio.get_event_loop()
    fut = asyncio.Future(loop=loop)

    if not isinstance(gt, greenlet.greenlet):
        raise TypeError("greenthread or greenlet request, not %s"
                        % type(gt))

    if gt:
        raise RuntimeError("wrap_greenthread: the greenthread is running")
    if gt.dead:
        raise RuntimeError("wrap_greenthread: the greenthread already finished")

    if isinstance(gt, eventlet.greenthread.GreenThread):
        orig_main = gt.run
        def wrap_func(*args, **kw):
            try:
                orig_main(*args, **kw)
            except Exception as exc:
                fut.set_exception(exc)
            else:
                result = gt.wait()
                fut.set_result(result)
        gt.run = wrap_func
    else:
        try:
            orig_func = gt.run
        except AttributeError:
            raise RuntimeError("wrap_greenthread: the run attribute "
                               "of the greenlet is not set")
        def wrap_func(*args, **kw):
            try:
                result = orig_func(*args, **kw)
            except Exception as exc:
                fut.set_exception(exc)
            else:
                fut.set_result(result)
        gt.run = wrap_func
    return fut


def yield_future(future, loop=None):
    """Wait for a future, a task, or a coroutine object from a greenthread.

    Yield control other eligible eventlet coroutines until the future is done
    (finished successfully or failed with an exception).

    Return the result or raise the exception of the future.

    The function must not be called from the greenthread
    running the aioeventlet event loop.
    """
    if hasattr(asyncio, "ensure_future"):
        ensure_future = asyncio.ensure_future
    else:  # use of async keyword has been Deprecated since Python 3.4.4
        ensure_future =  getattr(asyncio, "async")
    future = ensure_future(future, loop=loop)
    if future._loop._greenthread == eventlet.getcurrent():
        raise RuntimeError("yield_future() must not be called from "
                           "the greenthread of the aioeventlet event loop")

    event = eventlet.event.Event()
    def done(fut):
        try:
            result = fut.result()
        except Exception as exc:
            event.send_exception(exc)
        else:
            event.send(result)

    future.add_done_callback(done)
    return event.wait()