diff options
Diffstat (limited to 'python/gevent/_ffi/loop.py')
| -rw-r--r-- | python/gevent/_ffi/loop.py | 709 |
1 files changed, 709 insertions, 0 deletions
diff --git a/python/gevent/_ffi/loop.py b/python/gevent/_ffi/loop.py new file mode 100644 index 0000000..efe4462 --- /dev/null +++ b/python/gevent/_ffi/loop.py @@ -0,0 +1,709 @@ +""" +Basic loop implementation for ffi-based cores. +""" +# pylint: disable=too-many-lines, protected-access, redefined-outer-name, not-callable +from __future__ import absolute_import, print_function + +from collections import deque +import sys +import os +import traceback + +from gevent._ffi import _dbg +from gevent._ffi import GEVENT_DEBUG_LEVEL +from gevent._ffi import TRACE +from gevent._ffi.callback import callback +from gevent._compat import PYPY + +from gevent import getswitchinterval + +__all__ = [ + 'AbstractLoop', + 'assign_standard_callbacks', +] + + +class _EVENTSType(object): + def __repr__(self): + return 'gevent.core.EVENTS' + +EVENTS = GEVENT_CORE_EVENTS = _EVENTSType() + + +##### +## Note on CFFI objects, callbacks and the lifecycle of watcher objects +# +# Each subclass of `watcher` allocates a C structure of the +# appropriate type e.g., struct gevent_ev_io and holds this pointer in +# its `_gwatcher` attribute. When that watcher instance is garbage +# collected, then the C structure is also freed. The C structure is +# passed to libev from the watcher's start() method and then to the +# appropriate C callback function, e.g., _gevent_ev_io_callback, which +# passes it back to python's _python_callback where we need the +# watcher instance. Therefore, as long as that callback is active (the +# watcher is started), the watcher instance must not be allowed to get +# GC'd---any access at the C level or even the FFI level to the freed +# memory could crash the process. +# +# However, the typical idiom calls for writing something like this: +# loop.io(fd, python_cb).start() +# thus forgetting the newly created watcher subclass and allowing it to be immediately +# GC'd. To combat this, when the watcher is started, it places itself into the loop's +# `_keepaliveset`, and it only removes itself when the watcher's `stop()` method is called. +# Often, this is the *only* reference keeping the watcher object, and hence its C structure, +# alive. +# +# This is slightly complicated by the fact that the python-level +# callback, called from the C callback, could choose to manually stop +# the watcher. When we return to the C level callback, we now have an +# invalid pointer, and attempting to pass it back to Python (e.g., to +# handle an error) could crash. Hence, _python_callback, +# _gevent_io_callback, and _python_handle_error cooperate to make sure +# that the watcher instance stays in the loops `_keepaliveset` while +# the C code could be running---and if it gets removed, to not call back +# to Python again. +# See also https://github.com/gevent/gevent/issues/676 +#### +class AbstractCallbacks(object): + + + def __init__(self, ffi): + self.ffi = ffi + self.callbacks = [] + if GEVENT_DEBUG_LEVEL < TRACE: + self.from_handle = ffi.from_handle + + def from_handle(self, handle): # pylint:disable=method-hidden + x = self.ffi.from_handle(handle) + return x + + def python_callback(self, handle, revents): + """ + Returns an integer having one of three values: + + - -1 + An exception occurred during the callback and you must call + :func:`_python_handle_error` to deal with it. The Python watcher + object will have the exception tuple saved in ``_exc_info``. + - 1 + Everything went according to plan. You should check to see if the libev + watcher is still active, and call :func:`python_stop` if it is not. This will + clean up the memory. Finding the watcher still active at the event loop level, + but not having stopped itself at the gevent level is a buggy scenario and + shouldn't happen. + - 2 + Everything went according to plan, but the watcher has already + been stopped. Its memory may no longer be valid. + + This function should never return 0, as that's the default value that + Python exceptions will produce. + """ + #print("Running callback", handle) + orig_ffi_watcher = None + try: + # Even dereferencing the handle needs to be inside the try/except; + # if we don't return normally (e.g., a signal) then we wind up going + # to the 'onerror' handler (unhandled_onerror), which + # is not what we want; that can permanently wedge the loop depending + # on which callback was executing. + # XXX: See comments in that function. We may be able to restart and do better? + if not handle: + # Hmm, a NULL handle. That's not supposed to happen. + # We can easily get into a loop if we deref it and allow that + # to raise. + _dbg("python_callback got null handle") + return 1 + the_watcher = self.from_handle(handle) + orig_ffi_watcher = the_watcher._watcher + args = the_watcher.args + if args is None: + # Legacy behaviour from corecext: convert None into () + # See test__core_watcher.py + args = _NOARGS + if args and args[0] == GEVENT_CORE_EVENTS: + args = (revents, ) + args[1:] + #print("Calling function", the_watcher.callback, args) + the_watcher.callback(*args) + except: # pylint:disable=bare-except + _dbg("Got exception servicing watcher with handle", handle, sys.exc_info()) + # It's possible for ``the_watcher`` to be undefined (UnboundLocalError) + # if we threw an exception (signal) on the line that created that variable. + # This is typically the case with a signal under libuv + try: + the_watcher + except UnboundLocalError: + the_watcher = self.from_handle(handle) + the_watcher._exc_info = sys.exc_info() + # Depending on when the exception happened, the watcher + # may or may not have been stopped. We need to make sure its + # memory stays valid so we can stop it at the ev level if needed. + the_watcher.loop._keepaliveset.add(the_watcher) + return -1 + else: + if (the_watcher.loop is not None + and the_watcher in the_watcher.loop._keepaliveset + and the_watcher._watcher is orig_ffi_watcher): + # It didn't stop itself, *and* it didn't stop itself, reset + # its watcher, and start itself again. libuv's io watchers MAY + # do that. + # The normal, expected scenario when we find the watcher still + # in the keepaliveset is that it is still active at the event loop + # level, so we don't expect that python_stop gets called. + #_dbg("The watcher has not stopped itself, possibly still active", the_watcher) + return 1 + return 2 # it stopped itself + + def python_handle_error(self, handle, _revents): + _dbg("Handling error for handle", handle) + if not handle: + return + try: + watcher = self.from_handle(handle) + exc_info = watcher._exc_info + del watcher._exc_info + # In the past, we passed the ``watcher`` itself as the context, + # which typically meant that the Hub would just print + # the exception. This is a problem because sometimes we can't + # detect signals until late in ``python_callback``; specifically, + # test_selectors.py:DefaultSelectorTest.test_select_interrupt_exc + # installs a SIGALRM handler that raises an exception. That exception can happen + # before we enter ``python_callback`` or at any point within it because of the way + # libuv swallows signals. By passing None, we get the exception prapagated into + # the main greenlet (which is probably *also* not what we always want, but + # I see no way to distinguish the cases). + watcher.loop.handle_error(None, *exc_info) + finally: + # XXX Since we're here on an error condition, and we + # made sure that the watcher object was put in loop._keepaliveset, + # what about not stopping the watcher? Looks like a possible + # memory leak? + # XXX: This used to do "if revents & (libev.EV_READ | libev.EV_WRITE)" + # before stopping. Why? + try: + watcher.stop() + except: # pylint:disable=bare-except + watcher.loop.handle_error(watcher, *sys.exc_info()) + return # pylint:disable=lost-exception + + def unhandled_onerror(self, t, v, tb): + # This is supposed to be called for signals, etc. + # This is the onerror= value for CFFI. + # If we return None, C will get a value of 0/NULL; + # if we raise, CFFI will print the exception and then + # return 0/NULL; (unless error= was configured) + # If things go as planned, we return the value that asks + # C to call back and check on if the watcher needs to be closed or + # not. + + # XXX: TODO: Could this cause events to be lost? Maybe we need to return + # a value that causes the C loop to try the callback again? + # at least for signals under libuv, which are delivered at very odd times. + # Hopefully the event still shows up when we poll the next time. + watcher = None + handle = tb.tb_frame.f_locals['handle'] if tb is not None else None + if handle: # handle could be NULL + watcher = self.from_handle(handle) + if watcher is not None: + watcher.loop.handle_error(None, t, v, tb) + return 1 + + # Raising it causes a lot of noise from CFFI + print("WARNING: gevent: Unhandled error with no watcher", + file=sys.stderr) + traceback.print_exception(t, v, tb) + + def python_stop(self, handle): + if not handle: # pragma: no cover + print( + "WARNING: gevent: Unable to dereference handle; not stopping watcher. " + "Native resources may leak. This is most likely a bug in gevent.", + file=sys.stderr) + # The alternative is to crash with no helpful information + # NOTE: Raising exceptions here does nothing, they're swallowed by CFFI. + # Since the C level passed in a null pointer, even dereferencing the handle + # will just produce some exceptions. + return + watcher = self.from_handle(handle) + watcher.stop() + + if not PYPY: + def python_check_callback(self, watcher_ptr): # pylint:disable=unused-argument + # If we have the onerror callback, this is a no-op; all the real + # work to rethrow the exception is done by the onerror callback + + # NOTE: Unlike the rest of the functions, this is called with a pointer + # to the C level structure, *not* a pointer to the void* that represents a + # <cdata> for the Python Watcher object. + pass + else: # PyPy + # On PyPy, we need the function to have some sort of body, otherwise + # the signal exceptions don't always get caught, *especially* with + # libuv (however, there's no reason to expect this to only be a libuv + # issue; it's just that we don't depend on the periodic signal timer + # under libev, so the issue is much more pronounced under libuv) + # test_socket's test_sendall_interrupted can hang. + # See https://github.com/gevent/gevent/issues/1112 + + def python_check_callback(self, watcher_ptr): # pylint:disable=unused-argument + # Things we've tried that *don't* work: + # greenlet.getcurrent() + # 1 + 1 + try: + raise MemoryError() + except MemoryError: + pass + + def python_prepare_callback(self, watcher_ptr): + loop = self._find_loop_from_c_watcher(watcher_ptr) + if loop is None: # pragma: no cover + print("WARNING: gevent: running prepare callbacks from a destroyed handle: ", + watcher_ptr) + return + loop._run_callbacks() + + def check_callback_onerror(self, t, v, tb): + watcher_ptr = tb.tb_frame.f_locals['watcher_ptr'] if tb is not None else None + if watcher_ptr: + loop = self._find_loop_from_c_watcher(watcher_ptr) + if loop is not None: + # None as the context argument causes the exception to be raised + # in the main greenlet. + loop.handle_error(None, t, v, tb) + return None + raise v # Let CFFI print + + def _find_loop_from_c_watcher(self, watcher_ptr): + raise NotImplementedError() + + + +def assign_standard_callbacks(ffi, lib, callbacks_class, extras=()): # pylint:disable=unused-argument + # callbacks keeps these cdata objects alive at the python level + callbacks = callbacks_class(ffi) + extras = tuple([(getattr(callbacks, name), error) for name, error in extras]) + for (func, error_func) in ((callbacks.python_callback, None), + (callbacks.python_handle_error, None), + (callbacks.python_stop, None), + (callbacks.python_check_callback, + callbacks.check_callback_onerror), + (callbacks.python_prepare_callback, + callbacks.check_callback_onerror)) + extras: + # The name of the callback function matches the 'extern Python' declaration. + error_func = error_func or callbacks.unhandled_onerror + callback = ffi.def_extern(onerror=error_func)(func) + # keep alive the cdata + # (def_extern returns the original function, and it requests that + # the function be "global", so maybe it keeps a hard reference to it somewhere now + # unlike ffi.callback(), and we don't need to do this?) + callbacks.callbacks.append(callback) + + # At this point, the library C variable (static function, actually) + # is filled in. + + return callbacks + + +if sys.version_info[0] >= 3: + basestring = (bytes, str) + integer_types = (int,) +else: + import __builtin__ # pylint:disable=import-error + basestring = (__builtin__.basestring,) + integer_types = (int, __builtin__.long) + + + + +_NOARGS = () + +CALLBACK_CHECK_COUNT = 50 + +class AbstractLoop(object): + # pylint:disable=too-many-public-methods,too-many-instance-attributes + + error_handler = None + + _CHECK_POINTER = None + + _TIMER_POINTER = None + _TIMER_CALLBACK_SIG = None + + _PREPARE_POINTER = None + + starting_timer_may_update_loop_time = False + + # Subclasses should set this in __init__ to reflect + # whether they were the default loop. + _default = None + + def __init__(self, ffi, lib, watchers, flags=None, default=None): + self._ffi = ffi + self._lib = lib + self._ptr = None + self._handle_to_self = self._ffi.new_handle(self) # XXX: Reference cycle? + self._watchers = watchers + self._in_callback = False + self._callbacks = deque() + # Stores python watcher objects while they are started + self._keepaliveset = set() + self._init_loop_and_aux_watchers(flags, default) + + + def _init_loop_and_aux_watchers(self, flags=None, default=None): + + self._ptr = self._init_loop(flags, default) + + + # self._check is a watcher that runs in each iteration of the + # mainloop, just after the blocking call. It's point is to handle + # signals. It doesn't run watchers or callbacks, it just exists to give + # CFFI a chance to raise signal exceptions so we can handle them. + self._check = self._ffi.new(self._CHECK_POINTER) + self._check.data = self._handle_to_self + self._init_and_start_check() + + # self._prepare is a watcher that runs in each iteration of the mainloop, + # just before the blocking call. It's where we run deferred callbacks + # from self.run_callback. This cooperates with _setup_for_run_callback() + # to schedule self._timer0 if needed. + self._prepare = self._ffi.new(self._PREPARE_POINTER) + self._prepare.data = self._handle_to_self + self._init_and_start_prepare() + + # A timer we start and stop on demand. If we have callbacks, + # too many to run in one iteration of _run_callbacks, we turn this + # on so as to have the next iteration of the run loop return to us + # as quickly as possible. + # TODO: There may be a more efficient way to do this using ev_timer_again; + # see the "ev_timer" section of the ev manpage (http://linux.die.net/man/3/ev) + # Alternatively, setting the ev maximum block time may also work. + self._timer0 = self._ffi.new(self._TIMER_POINTER) + self._timer0.data = self._handle_to_self + self._init_callback_timer() + + # TODO: We may be able to do something nicer and use the existing python_callback + # combined with onerror and the class check/timer/prepare to simplify things + # and unify our handling + + def _init_loop(self, flags, default): + """ + Called by __init__ to create or find the loop. The return value + is assigned to self._ptr. + """ + raise NotImplementedError() + + def _init_and_start_check(self): + raise NotImplementedError() + + def _init_and_start_prepare(self): + raise NotImplementedError() + + def _init_callback_timer(self): + raise NotImplementedError() + + def _stop_callback_timer(self): + raise NotImplementedError() + + def _start_callback_timer(self): + raise NotImplementedError() + + def _check_callback_handle_error(self, t, v, tb): + self.handle_error(None, t, v, tb) + + def _run_callbacks(self): # pylint:disable=too-many-branches + # When we're running callbacks, its safe for timers to + # update the notion of the current time (because if we're here, + # we're not running in a timer callback that may let other timers + # run; this is mostly an issue for libuv). + + # That's actually a bit of a lie: on libev, self._timer0 really is + # a timer, and so sometimes this is running in a timer callback, not + # a prepare callback. But that's OK, libev doesn't suffer from cascading + # timer expiration and its safe to update the loop time at any + # moment there. + self.starting_timer_may_update_loop_time = True + try: + count = CALLBACK_CHECK_COUNT + now = self.now() + expiration = now + getswitchinterval() + self._stop_callback_timer() + while self._callbacks: + cb = self._callbacks.popleft() # pylint:disable=assignment-from-no-return + count -= 1 + self.unref() # XXX: libuv doesn't have a global ref count! + callback = cb.callback + cb.callback = None + args = cb.args + if callback is None or args is None: + # it's been stopped + continue + + try: + callback(*args) + except: # pylint:disable=bare-except + # If we allow an exception to escape this method (while we are running the ev callback), + # then CFFI will print the error and libev will continue executing. + # There are two problems with this. The first is that the code after + # the loop won't run. The second is that any remaining callbacks scheduled + # for this loop iteration will be silently dropped; they won't run, but they'll + # also not be *stopped* (which is not a huge deal unless you're looking for + # consistency or checking the boolean/pending status; the loop doesn't keep + # a reference to them like it does to watchers...*UNLESS* the callback itself had + # a reference to a watcher; then I don't know what would happen, it depends on + # the state of the watcher---a leak or crash is not totally inconceivable). + # The Cython implementation in core.ppyx uses gevent_call from callbacks.c + # to run the callback, which uses gevent_handle_error to handle any errors the + # Python callback raises...it unconditionally simply prints any error raised + # by loop.handle_error and clears it, so callback handling continues. + # We take a similar approach (but are extra careful about printing) + try: + self.handle_error(cb, *sys.exc_info()) + except: # pylint:disable=bare-except + try: + print("Exception while handling another error", file=sys.stderr) + traceback.print_exc() + except: # pylint:disable=bare-except + pass # Nothing we can do here + finally: + # NOTE: this must be reset here, because cb.args is used as a flag in + # the callback class so that bool(cb) of a callback that has been run + # becomes False + cb.args = None + + # We've finished running one group of callbacks + # but we may have more, so before looping check our + # switch interval. + if count == 0 and self._callbacks: + count = CALLBACK_CHECK_COUNT + self.update_now() + if self.now() >= expiration: + now = 0 + break + + # Update the time before we start going again, if we didn't + # just do so. + if now != 0: + self.update_now() + + if self._callbacks: + self._start_callback_timer() + finally: + self.starting_timer_may_update_loop_time = False + + def _stop_aux_watchers(self): + raise NotImplementedError() + + def destroy(self): + if self._ptr: + try: + if not self._can_destroy_loop(self._ptr): + return False + self._stop_aux_watchers() + self._destroy_loop(self._ptr) + finally: + # not ffi.NULL, we don't want something that can be + # passed to C and crash later. This will create nice friendly + # TypeError from CFFI. + self._ptr = None + del self._handle_to_self + del self._callbacks + del self._keepaliveset + + return True + + def _can_destroy_loop(self, ptr): + raise NotImplementedError() + + def _destroy_loop(self, ptr): + raise NotImplementedError() + + @property + def ptr(self): + return self._ptr + + @property + def WatcherType(self): + return self._watchers.watcher + + @property + def MAXPRI(self): + return 1 + + @property + def MINPRI(self): + return 1 + + def _handle_syserr(self, message, errno): + try: + errno = os.strerror(errno) + except: # pylint:disable=bare-except + traceback.print_exc() + try: + message = '%s: %s' % (message, errno) + except: # pylint:disable=bare-except + traceback.print_exc() + self.handle_error(None, SystemError, SystemError(message), None) + + def handle_error(self, context, type, value, tb): + handle_error = None + error_handler = self.error_handler + if error_handler is not None: + # we do want to do getattr every time so that setting Hub.handle_error property just works + handle_error = getattr(error_handler, 'handle_error', error_handler) + handle_error(context, type, value, tb) + else: + self._default_handle_error(context, type, value, tb) + + def _default_handle_error(self, context, type, value, tb): # pylint:disable=unused-argument + # note: Hub sets its own error handler so this is not used by gevent + # this is here to make core.loop usable without the rest of gevent + # Should cause the loop to stop running. + traceback.print_exception(type, value, tb) + + + def run(self, nowait=False, once=False): + raise NotImplementedError() + + def reinit(self): + raise NotImplementedError() + + def ref(self): + # XXX: libuv doesn't do it this way + raise NotImplementedError() + + def unref(self): + raise NotImplementedError() + + def break_(self, how=None): + raise NotImplementedError() + + def verify(self): + pass + + def now(self): + raise NotImplementedError() + + def update_now(self): + raise NotImplementedError() + + def update(self): + import warnings + warnings.warn("'update' is deprecated; use 'update_now'", + DeprecationWarning, + stacklevel=2) + self.update_now() + + def __repr__(self): + return '<%s at 0x%x %s>' % (self.__class__.__name__, id(self), self._format()) + + @property + def default(self): + return self._default if self._ptr else False + + @property + def iteration(self): + return -1 + + @property + def depth(self): + return -1 + + @property + def backend_int(self): + return 0 + + @property + def backend(self): + return "default" + + @property + def pendingcnt(self): + return 0 + + def io(self, fd, events, ref=True, priority=None): + return self._watchers.io(self, fd, events, ref, priority) + + def timer(self, after, repeat=0.0, ref=True, priority=None): + return self._watchers.timer(self, after, repeat, ref, priority) + + def signal(self, signum, ref=True, priority=None): + return self._watchers.signal(self, signum, ref, priority) + + def idle(self, ref=True, priority=None): + return self._watchers.idle(self, ref, priority) + + def prepare(self, ref=True, priority=None): + return self._watchers.prepare(self, ref, priority) + + def check(self, ref=True, priority=None): + return self._watchers.check(self, ref, priority) + + def fork(self, ref=True, priority=None): + return self._watchers.fork(self, ref, priority) + + def async_(self, ref=True, priority=None): + return self._watchers.async_(self, ref, priority) + + # Provide BWC for those that can use 'async' as is + locals()['async'] = async_ + + if sys.platform != "win32": + + def child(self, pid, trace=0, ref=True): + return self._watchers.child(self, pid, trace, ref) + + def install_sigchld(self): + pass + + def stat(self, path, interval=0.0, ref=True, priority=None): + return self._watchers.stat(self, path, interval, ref, priority) + + def callback(self, priority=None): + return callback(self, priority) + + def _setup_for_run_callback(self): + raise NotImplementedError() + + def run_callback(self, func, *args): + # If we happen to already be running callbacks (inside + # _run_callbacks), this could happen almost immediately, + # without the loop cycling. + cb = callback(func, args) + self._callbacks.append(cb) + self._setup_for_run_callback() + + return cb + + def _format(self): + if not self._ptr: + return 'destroyed' + msg = self.backend + if self.default: + msg += ' default' + msg += ' pending=%s' % self.pendingcnt + msg += self._format_details() + return msg + + def _format_details(self): + msg = '' + fileno = self.fileno() # pylint:disable=assignment-from-none + try: + activecnt = self.activecnt + except AttributeError: + activecnt = None + if activecnt is not None: + msg += ' ref=' + repr(activecnt) + if fileno is not None: + msg += ' fileno=' + repr(fileno) + #if sigfd is not None and sigfd != -1: + # msg += ' sigfd=' + repr(sigfd) + return msg + + def fileno(self): + return None + + @property + def activecnt(self): + if not self._ptr: + raise ValueError('operation on destroyed loop') + return 0 |