path: root/python/gevent/hub.py

# Copyright (c) 2009-2015 Denis Bilenko. See LICENSE for details.
"""
Event-loop hub.
"""
from __future__ import absolute_import, print_function
# XXX: FIXME: Refactor to make this smaller
# pylint:disable=too-many-lines
from functools import partial as _functools_partial

import sys
import traceback


from greenlet import greenlet as RawGreenlet
from greenlet import getcurrent
from greenlet import GreenletExit



__all__ = [
    'getcurrent',
    'GreenletExit',
    'spawn_raw',
    'sleep',
    'kill',
    'signal',
    'reinit',
    'get_hub',
    'Hub',
    'Waiter',
]

from gevent._config import config as GEVENT_CONFIG
from gevent._compat import thread_mod_name
from gevent._util import readproperty
from gevent._util import Lazy
from gevent._util import gmctime
from gevent._ident import IdentRegistry

from gevent._hub_local import get_hub
from gevent._hub_local import get_loop
from gevent._hub_local import set_hub
from gevent._hub_local import set_loop
from gevent._hub_local import get_hub_if_exists as _get_hub
from gevent._hub_local import get_hub_noargs as _get_hub_noargs
from gevent._hub_local import set_default_hub_class

from gevent._greenlet_primitives import TrackedRawGreenlet
from gevent._hub_primitives import WaitOperationsGreenlet

# Export
from gevent import _hub_primitives
wait = _hub_primitives.wait_on_objects
iwait = _hub_primitives.iwait_on_objects


from gevent.exceptions import LoopExit

from gevent._waiter import Waiter

# Need the real get_ident. We're imported early enough (by gevent/__init__.py)
# that we can be sure nothing is monkey patched yet.
get_thread_ident = __import__(thread_mod_name).get_ident
MAIN_THREAD_IDENT = get_thread_ident() # XXX: Assuming import is done on the main thread.


def spawn_raw(function, *args, **kwargs):
    """
    Create a new :class:`greenlet.greenlet` object and schedule it to
    run ``function(*args, **kwargs)``.

    This returns a raw :class:`~greenlet.greenlet` which does not have all the useful
    methods that :class:`gevent.Greenlet` has. Typically, applications
    should prefer :func:`~gevent.spawn`, but this method may
    occasionally be useful as an optimization if there are many
    greenlets involved.

    .. versionchanged:: 1.1a3
        Verify that ``function`` is callable, raising a TypeError if not. Previously,
        the spawned greenlet would have failed the first time it was switched to.

    .. versionchanged:: 1.1b1
       If *function* is not callable, immediately raise a :exc:`TypeError`
       instead of spawning a greenlet that will raise an uncaught TypeError.

    .. versionchanged:: 1.1rc2
        Accept keyword arguments for ``function`` as previously (incorrectly)
        documented. Note that this may incur an additional expense.

    .. versionchanged:: 1.3a2
       Populate the ``spawning_greenlet`` and ``spawn_tree_locals``
       attributes of the returned greenlet.

    .. versionchanged:: 1.3b1
       *Only* populate ``spawning_greenlet`` and ``spawn_tree_locals``
       if ``GEVENT_TRACK_GREENLET_TREE`` is enabled (the default). If not enabled,
       those attributes will not be set.

    """
    if not callable(function):
        raise TypeError("function must be callable")

    # The hub is always the parent.
    hub = _get_hub_noargs()

    factory = TrackedRawGreenlet if GEVENT_CONFIG.track_greenlet_tree else RawGreenlet

    # The callback class object that we use to run this doesn't
    # accept kwargs (and those objects are heavily used, as well as being
    # implemented twice in core.ppyx and corecffi.py) so do it with a partial
    if kwargs:
        function = _functools_partial(function, *args, **kwargs)
        g = factory(function, hub)
        hub.loop.run_callback(g.switch)
    else:
        g = factory(function, hub)
        hub.loop.run_callback(g.switch, *args)

    return g


def sleep(seconds=0, ref=True):
    """
    Put the current greenlet to sleep for at least *seconds*.

    *seconds* may be specified as an integer, or a float if fractional
    seconds are desired.

    .. tip:: In the current implementation, a value of 0 (the default)
       means to yield execution to any other runnable greenlets, but
       this greenlet may be scheduled again before the event loop
       cycles (in an extreme case, a greenlet that repeatedly sleeps
       with 0 can prevent greenlets that are ready to do I/O from
       being scheduled for some (small) period of time); a value greater than
       0, on the other hand, will delay running this greenlet until
       the next iteration of the loop.

    If *ref* is False, the greenlet running ``sleep()`` will not prevent :func:`gevent.wait`
    from exiting.

    .. versionchanged:: 1.3a1
       Sleeping with a value of 0 will now be bounded to approximately block the
       loop for no longer than :func:`gevent.getswitchinterval`.

    .. seealso:: :func:`idle`
    """
    hub = _get_hub_noargs()
    loop = hub.loop
    if seconds <= 0:
        waiter = Waiter(hub)
        loop.run_callback(waiter.switch, None)
        waiter.get()
    else:
        with loop.timer(seconds, ref=ref) as t:
            # Sleeping is expected to be an "absolute" measure with
            # respect to time.time(), not a relative measure, so it's
            # important to update the loop's notion of now before we start
            loop.update_now()
            hub.wait(t)


def idle(priority=0):
    """
    Cause the calling greenlet to wait until the event loop is idle.

    Idle is defined as having no other events of the same or higher
    *priority* pending. That is, as long as sockets, timeouts or even
    signals of the same or higher priority are being processed, the loop
    is not idle.

    .. seealso:: :func:`sleep`
    """
    hub = _get_hub_noargs()
    watcher = hub.loop.idle()
    if priority:
        watcher.priority = priority
    hub.wait(watcher)


def kill(greenlet, exception=GreenletExit):
    """
    Kill greenlet asynchronously. The current greenlet is not unscheduled.

    .. note::

        The method :meth:`Greenlet.kill` method does the same and
        more (and the same caveats listed there apply here). However, the MAIN
        greenlet - the one that exists initially - does not have a
        ``kill()`` method, and neither do any created with :func:`spawn_raw`,
        so you have to use this function.

    .. caution:: Use care when killing greenlets. If they are not prepared for
       exceptions, this could result in corrupted state.

    .. versionchanged:: 1.1a2
        If the ``greenlet`` has a :meth:`kill <Greenlet.kill>` method, calls it. This prevents a
        greenlet from being switched to for the first time after it's been
        killed but not yet executed.
    """
    if not greenlet.dead:
        if hasattr(greenlet, 'kill'):
            # dealing with gevent.greenlet.Greenlet. Use it, especially
            # to avoid allowing one to be switched to for the first time
            # after it's been killed
            greenlet.kill(exception=exception, block=False)
        else:
            _get_hub_noargs().loop.run_callback(greenlet.throw, exception)


class signal(object):
    """
    Call the *handler* with the *args* and *kwargs* when the process
    receives the signal *signalnum*.

    The *handler* will be run in a new greenlet when the signal is delivered.

    This returns an object with the useful method ``cancel``, which, when called,
    will prevent future deliveries of *signalnum* from calling *handler*.

    .. note::

        This may not operate correctly with SIGCHLD if libev child watchers
        are used (as they are by default with os.fork).

    .. versionchanged:: 1.2a1
       The ``handler`` argument is required to be callable at construction time.
    """

    # XXX: This is manually documented in gevent.rst while it is aliased in
    # the gevent module.

    greenlet_class = None

    def __init__(self, signalnum, handler, *args, **kwargs):
        if not callable(handler):
            raise TypeError("signal handler must be callable.")

        self.hub = _get_hub_noargs()
        self.watcher = self.hub.loop.signal(signalnum, ref=False)
        self.watcher.start(self._start)
        self.handler = handler
        self.args = args
        self.kwargs = kwargs
        if self.greenlet_class is None:
            from gevent import Greenlet
            self.greenlet_class = Greenlet

    def _get_ref(self):
        return self.watcher.ref

    def _set_ref(self, value):
        self.watcher.ref = value

    ref = property(_get_ref, _set_ref)
    del _get_ref, _set_ref

    def cancel(self):
        self.watcher.stop()

    def _start(self):
        try:
            greenlet = self.greenlet_class(self.handle)
            greenlet.switch()
        except: # pylint:disable=bare-except
            self.hub.handle_error(None, *sys._exc_info()) # pylint:disable=no-member

    def handle(self):
        try:
            self.handler(*self.args, **self.kwargs)
        except: # pylint:disable=bare-except
            self.hub.handle_error(None, *sys.exc_info())


def reinit(hub=None):
    """
    reinit() -> None

    Prepare the gevent hub to run in a new (forked) process.

    This should be called *immediately* after :func:`os.fork` in the
    child process. This is done automatically by
    :func:`gevent.os.fork` or if the :mod:`os` module has been
    monkey-patched. If this function is not called in a forked
    process, symptoms may include hanging of functions like
    :func:`socket.getaddrinfo`, and the hub's threadpool is unlikely
    to work.

    .. note:: Registered fork watchers may or may not run before
       this function (and thus ``gevent.os.fork``) return. If they have
       not run, they will run "soon", after an iteration of the event loop.
       You can force this by inserting a few small (but non-zero) calls to :func:`sleep`
       after fork returns. (As of gevent 1.1 and before, fork watchers will
       not have run, but this may change in the future.)

    .. note:: This function may be removed in a future major release
       if the fork process can be more smoothly managed.

    .. warning:: See remarks in :func:`gevent.os.fork` about greenlets
       and event loop watchers in the child process.
    """
    # Note the signature line in the docstring: hub is not a public param.

    # The loop reinit function in turn calls libev's ev_loop_fork
    # function.
    hub = _get_hub() if hub is None else hub
    if hub is None:
        return

    # Note that we reinit the existing loop, not destroy it.
    # See https://github.com/gevent/gevent/issues/200.
    hub.loop.reinit()
    # libev's fork watchers are slow to fire because the only fire
    # at the beginning of a loop; due to our use of callbacks that
    # run at the end of the loop, that may be too late. The
    # threadpool and resolvers depend on the fork handlers being
    # run (specifically, the threadpool will fail in the forked
    # child if there were any threads in it, which there will be
    # if the resolver_thread was in use (the default) before the
    # fork.)
    #
    # If the forked process wants to use the threadpool or
    # resolver immediately (in a queued callback), it would hang.
    #
    # The below is a workaround. Fortunately, all of these
    # methods are idempotent and can be called multiple times
    # following a fork if the suddenly started working, or were
    # already working on some platforms. Other threadpools and fork handlers
    # will be called at an arbitrary time later ('soon')
    for obj in (hub._threadpool, hub._resolver, hub.periodic_monitoring_thread):
        getattr(obj, '_on_fork', lambda: None)()

    # TODO: We'd like to sleep for a non-zero amount of time to force the loop to make a
    # pass around before returning to this greenlet. That will allow any
    # user-provided fork watchers to run. (Two calls are necessary.) HOWEVER, if
    # we do this, certain tests that heavily mix threads and forking,
    # like 2.7/test_threading:test_reinit_tls_after_fork, fail. It's not immediately clear
    # why.
    #sleep(0.00001)
    #sleep(0.00001)


hub_ident_registry = IdentRegistry()

class Hub(WaitOperationsGreenlet):
    """
    A greenlet that runs the event loop.

    It is created automatically by :func:`get_hub`.

    .. rubric:: Switching

    Every time this greenlet (i.e., the event loop) is switched *to*,
    if the current greenlet has a ``switch_out`` method, it will be
    called. This allows a greenlet to take some cleanup actions before
    yielding control. This method should not call any gevent blocking
    functions.
    """

    #: If instances of these classes are raised into the event loop,
    #: they will be propagated out to the main greenlet (where they will
    #: usually be caught by Python itself)
    SYSTEM_ERROR = (KeyboardInterrupt, SystemExit, SystemError)

    #: Instances of these classes are not considered to be errors and
    #: do not get logged/printed when raised by the event loop.
    NOT_ERROR = (GreenletExit, SystemExit)

    #: The size we use for our threadpool. Either use a subclass
    #: for this, or change it immediately after creating the hub.
    threadpool_size = 10

    # An instance of PeriodicMonitoringThread, if started.
    periodic_monitoring_thread = None

    # The ident of the thread we were created in, which should be the
    # thread that we run in.
    thread_ident = None

    #: A string giving the name of this hub. Useful for associating hubs
    #: with particular threads. Printed as part of the default repr.
    #:
    #: .. versionadded:: 1.3b1
    name = ''

    # NOTE: We cannot define a class-level 'loop' attribute
    # because that conflicts with the slot we inherit from the
    # Cythonized-bases.

    def __init__(self, loop=None, default=None):
        WaitOperationsGreenlet.__init__(self, None, None)
        self.thread_ident = get_thread_ident()
        if hasattr(loop, 'run'):
            if default is not None:
                raise TypeError("Unexpected argument: default")
            self.loop = loop
        elif get_loop() is not None:
            # Reuse a loop instance previously set by
            # destroying a hub without destroying the associated
            # loop. See #237 and #238.
            self.loop = get_loop()
        else:
            if default is None and self.thread_ident != MAIN_THREAD_IDENT:
                default = False

            if loop is None:
                loop = self.backend
            self.loop = self.loop_class(flags=loop, default=default) # pylint:disable=not-callable
        self._resolver = None
        self._threadpool = None
        self.format_context = GEVENT_CONFIG.format_context
        self.minimal_ident = hub_ident_registry.get_ident(self)

    @Lazy
    def ident_registry(self):
        return IdentRegistry()

    @property
    def loop_class(self):
        return GEVENT_CONFIG.loop

    @property
    def backend(self):
        return GEVENT_CONFIG.libev_backend

    @property
    def main_hub(self):
        """
        Is this the hub for the main thread?

        .. versionadded:: 1.3b1
        """
        return self.thread_ident == MAIN_THREAD_IDENT

    def __repr__(self):
        if self.loop is None:
            info = 'destroyed'
        else:
            try:
                info = self.loop._format()
            except Exception as ex: # pylint:disable=broad-except
                info = str(ex) or repr(ex) or 'error'
        result = '<%s %r at 0x%x %s' % (
            self.__class__.__name__,
            self.name,
            id(self),
            info)
        if self._resolver is not None:
            result += ' resolver=%r' % self._resolver
        if self._threadpool is not None:
            result += ' threadpool=%r' % self._threadpool
        result += ' thread_ident=%s' % (hex(self.thread_ident), )
        return result + '>'

    def handle_error(self, context, type, value, tb):
        """
        Called by the event loop when an error occurs. The arguments
        type, value, and tb are the standard tuple returned by :func:`sys.exc_info`.

        Applications can set a property on the hub with this same signature
        to override the error handling provided by this class.

        Errors that are :attr:`system errors <SYSTEM_ERROR>` are passed
        to :meth:`handle_system_error`.

        :param context: If this is ``None``, indicates a system error that
            should generally result in exiting the loop and being thrown to the
            parent greenlet.
        """
        if isinstance(value, str):
            # Cython can raise errors where the value is a plain string
            # e.g., AttributeError, "_semaphore.Semaphore has no attr", <traceback>
            value = type(value)
        if not issubclass(type, self.NOT_ERROR):
            self.print_exception(context, type, value, tb)
        if context is None or issubclass(type, self.SYSTEM_ERROR):
            self.handle_system_error(type, value)

    def handle_system_error(self, type, value):
        """
        Called from `handle_error` when the exception type is determined
        to be a :attr:`system error <SYSTEM_ERROR>`.

        System errors cause the exception to be raised in the main
        greenlet (the parent of this hub).
        """
        current = getcurrent()
        if current is self or current is self.parent or self.loop is None:
            self.parent.throw(type, value)
        else:
            # in case system error was handled and life goes on
            # switch back to this greenlet as well
            cb = None
            try:
                cb = self.loop.run_callback(current.switch)
            except: # pylint:disable=bare-except
                traceback.print_exc(file=self.exception_stream)
            try:
                self.parent.throw(type, value)
            finally:
                if cb is not None:
                    cb.stop()

    @readproperty
    def exception_stream(self):
        """
        The stream to which exceptions will be written.
        Defaults to ``sys.stderr`` unless assigned to.

        .. versionadded:: 1.2a1
        """
        # Unwrap any FileObjectThread we have thrown around sys.stderr
        # (because it can't be used in the hub). Tricky because we are
        # called in error situations when it's not safe to import.
        stderr = sys.stderr
        if type(stderr).__name__ == 'FileObjectThread':
            stderr = stderr.io # pylint:disable=no-member
        return stderr

    def print_exception(self, context, type, value, tb):
        # Python 3 does not gracefully handle None value or tb in
        # traceback.print_exception() as previous versions did.
        # pylint:disable=no-member
        errstream = self.exception_stream

        if value is None:
            errstream.write('%s\n' % type.__name__)
        else:
            traceback.print_exception(type, value, tb, file=errstream)
        del tb

        try:
            errstream.write(gmctime())
            errstream.write(' ' if context is not None else '\n')
        except: # pylint:disable=bare-except
            # Possible not safe to import under certain
            # error conditions in Python 2
            pass

        if context is not None:
            if not isinstance(context, str):
                try:
                    context = self.format_context(context)
                except: # pylint:disable=bare-except
                    traceback.print_exc(file=self.exception_stream)
                    context = repr(context)
            errstream.write('%s failed with %s\n\n' % (context, getattr(type, '__name__', 'exception'), ))


    def run(self):
        """
        Entry-point to running the loop. This method is called automatically
        when the hub greenlet is scheduled; do not call it directly.

        :raises gevent.exceptions.LoopExit: If the loop finishes running. This means
           that there are no other scheduled greenlets, and no active
           watchers or servers. In some situations, this indicates a
           programming error.
        """
        assert self is getcurrent(), 'Do not call Hub.run() directly'
        self.start_periodic_monitoring_thread()
        while 1:
            loop = self.loop
            loop.error_handler = self
            try:
                loop.run()
            finally:
                loop.error_handler = None  # break the refcount cycle
            debug = []
            if hasattr(loop, 'debug'):
                debug = loop.debug()
            self.parent.throw(LoopExit('This operation would block forever', self, debug))
        # this function must never return, as it will cause switch() in the parent greenlet
        # to return an unexpected value
        # It is still possible to kill this greenlet with throw. However, in that case
        # switching to it is no longer safe, as switch will return immediately

    def start_periodic_monitoring_thread(self):
        if self.periodic_monitoring_thread is None and GEVENT_CONFIG.monitor_thread:
            # Note that it is possible for one real thread to
            # (temporarily) wind up with multiple monitoring threads,
            # if hubs are started and stopped within the thread. This shows up
            # in the threadpool tests. The monitoring threads will eventually notice their
            # hub object is gone.
            from gevent._monitor import PeriodicMonitoringThread
            from gevent.events import PeriodicMonitorThreadStartedEvent
            from gevent.events import notify_and_call_entry_points
            self.periodic_monitoring_thread = PeriodicMonitoringThread(self)

            if self.main_hub:
                self.periodic_monitoring_thread.install_monitor_memory_usage()

            notify_and_call_entry_points(PeriodicMonitorThreadStartedEvent(
                self.periodic_monitoring_thread))

        return self.periodic_monitoring_thread

    def join(self, timeout=None):
        """Wait for the event loop to finish. Exits only when there are
        no more spawned greenlets, started servers, active timeouts or watchers.

        If *timeout* is provided, wait no longer for the specified number of seconds.

        Returns True if exited because the loop finished execution.
        Returns False if exited because of timeout expired.
        """
        assert getcurrent() is self.parent, "only possible from the MAIN greenlet"
        if self.dead:
            return True

        waiter = Waiter(self)

        if timeout is not None:
            timeout = self.loop.timer(timeout, ref=False)
            timeout.start(waiter.switch, None)

        try:
            try:
                waiter.get()
            except LoopExit:
                return True
        finally:
            if timeout is not None:
                timeout.stop()
                timeout.close()
        return False

    def destroy(self, destroy_loop=None):
        """
        Destroy this hub and clean up its resources.

        If you manually create hubs, you *should* call this
        method before disposing of the hub object reference.
        """
        if self.periodic_monitoring_thread is not None:
            self.periodic_monitoring_thread.kill()
            self.periodic_monitoring_thread = None
        if self._resolver is not None:
            self._resolver.close()
            del self._resolver
        if self._threadpool is not None:
            self._threadpool.kill()
            del self._threadpool
        if destroy_loop is None:
            destroy_loop = not self.loop.default
        if destroy_loop:
            if get_loop() is self.loop:
                # Don't let anyone try to reuse this
                set_loop(None)
            self.loop.destroy()
        else:
            # Store in case another hub is created for this
            # thread.
            set_loop(self.loop)


        self.loop = None
        if _get_hub() is self:
            set_hub(None)


    # XXX: We can probably simplify the resolver and threadpool properties.

    @property
    def resolver_class(self):
        return GEVENT_CONFIG.resolver

    def _get_resolver(self):
        if self._resolver is None:
            self._resolver = self.resolver_class(hub=self) # pylint:disable=not-callable
        return self._resolver

    def _set_resolver(self, value):
        self._resolver = value

    def _del_resolver(self):
        self._resolver = None

    resolver = property(_get_resolver, _set_resolver, _del_resolver,
                        """
                        The DNS resolver that the socket functions will use.

                        .. seealso:: :doc:`/dns`
                        """)


    @property
    def threadpool_class(self):
        return GEVENT_CONFIG.threadpool

    def _get_threadpool(self):
        if self._threadpool is None:
            # pylint:disable=not-callable
            self._threadpool = self.threadpool_class(self.threadpool_size, hub=self)
        return self._threadpool

    def _set_threadpool(self, value):
        self._threadpool = value

    def _del_threadpool(self):
        self._threadpool = None

    threadpool = property(_get_threadpool, _set_threadpool, _del_threadpool,
                          """
                          The threadpool associated with this hub.

                          Usually this is a
                          :class:`gevent.threadpool.ThreadPool`, but
                          you :attr:`can customize that
                          <gevent._config.Config.threadpool>`.

                          Use this object to schedule blocking
                          (non-cooperative) operations in a different
                          thread to prevent them from halting the event loop.
                          """)


set_default_hub_class(Hub)



class linkproxy(object):
    __slots__ = ['callback', 'obj']

    def __init__(self, callback, obj):
        self.callback = callback
        self.obj = obj

    def __call__(self, *args):
        callback = self.callback
        obj = self.obj
        self.callback = None
        self.obj = None
        callback(obj)