""" 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 # 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