Merge pull request #1497 from gevent/issue1482

Fix potential crashes in the FFI backends if a watcher was closed and…

CHANGES.rst

@@ -76,6 +76,11 @@
   -m gevent.monkey``. Previously it would use greenlets instead of
   native threads. See :issue:`1484`.
 
+- Fix potential crashes in the FFI backends if a watcher was closed
+  and stopped in the middle of a callback from the event loop and then
+  raised an exception. This could happen if the hub's ``handle_error``
+  function was poorly customized, for example. See :issue:`1482`
+
 1.5a2 (2019-10-21)
 ==================
 

src/gevent/_ffi/loop.py

@@ -100,6 +100,7 @@ class AbstractCallbacks(object):
         """
         #_dbg("Running callback", handle)
         orig_ffi_watcher = None
+        orig_loop = 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
@@ -115,18 +116,16 @@ class AbstractCallbacks(object):
                 return 1
             the_watcher = self.from_handle(handle)
             orig_ffi_watcher = the_watcher._watcher
+            orig_loop = the_watcher.loop
             args = the_watcher.args
-            #_dbg("Running callback", the_watcher, orig_ffi_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:]
-            #_dbg("Calling function", the_watcher.callback, args)
             the_watcher.callback(*args) # None here means we weren't started
         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
@@ -134,15 +133,43 @@ class AbstractCallbacks(object):
                 the_watcher
             except UnboundLocalError:
                 the_watcher = self.from_handle(handle)
+
+            # It may not be safe to do anything with `handle` or `orig_ffi_watcher`
+            # anymore. If the watcher closed or stopped itself *before* throwing the exception,
+            # then the `handle` and `orig_ffi_watcher` may no longer be valid. Attempting to
+            # e.g., dereference the handle is likely to crash the process.
             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.
+
+
+            # If it hasn't been stopped, we need to make sure its
+            # memory stays valid so we can stop it at the native level if needed.
             # If its loop is gone, it has already been stopped,
             # see https://github.com/gevent/gevent/issues/1295 for a case where
-            # that happened
-            if the_watcher.loop is not None:
-                the_watcher.loop._keepaliveset.add(the_watcher)
+            # that happened, as well as issue #1482
+            if (
+                    # The last thing it does. Full successful close.
+                    the_watcher.loop is None
+                    # Only a partial close. We could leak memory and even crash later.
+                    or the_watcher._handle is None
+            ):
+                # Prevent unhandled_onerror from using the invalid handle
+                handle = None
+                exc_info = the_watcher._exc_info
+                del the_watcher._exc_info
+                try:
+                    if orig_loop is not None:
+                        orig_loop.handle_error(the_watcher, *exc_info)
+                    else:
+                        self.unhandled_onerror(*exc_info)
+                except:
+                    print("WARNING: gevent: Error when handling error",
+                          file=sys.stderr)
+                    traceback.print_exc()
+                # Signal that we're closed, no need to do more.
+                return 2
+
+            # Keep it around so we can close it later.
+            the_watcher.loop._keepaliveset.add(the_watcher)
             return -1
         else:
             if (the_watcher.loop is not None

src/gevent/libev/watcher.py

@@ -218,7 +218,7 @@ class async_(_base.AsyncMixin, watcher):
 
     @property
     def pending(self):
-        return bool(libev.ev_async_pending(self._watcher))
+        return self._watcher is not None and bool(libev.ev_async_pending(self._watcher))
 
 # Provide BWC for those that have async
 locals()['async'] = async_

src/gevent/tests/test__hub.py

@@ -337,5 +337,30 @@ class TestLoopInterface(unittest.TestCase):
         verify.verifyObject(ILoop, loop)
 
 
+class TestHandleError(unittest.TestCase):
+
+    def tearDown(self):
+        try:
+            del get_hub().handle_error
+        except AttributeError:
+            pass
+
+    def test_exception_in_custom_handle_error_does_not_crash(self):
+
+        def bad_handle_error(*args):
+            raise AttributeError
+
+        hub = get_hub().handle_error = bad_handle_error
+
+        class MyException(Exception):
+            pass
+
+        def raises():
+            raise MyException
+
+        with self.assertRaises(MyException):
+            gevent.spawn(raises).get()
+
+
 if __name__ == '__main__':
     greentest.main()

src/gevent/tests/test__threadpool.py

@@ -683,5 +683,42 @@ class TestTPE(_AbstractPoolTest):
         del self.pool
 
 
+class TestThreadResult(greentest.TestCase):
+
+    def test_exception_in_on_async_doesnt_crash(self):
+        # Issue 1482. An FFI-based loop could crash the whole process
+        # by dereferencing a handle after it was closed.
+        called = []
+        class MyException(Exception):
+            pass
+
+        def bad_when_ready():
+            called.append(1)
+            raise MyException
+
+        tr = gevent.threadpool.ThreadResult(None, gevent.get_hub(), bad_when_ready)
+
+        def wake():
+            called.append(1)
+            tr.set(42)
+
+        gevent.spawn(wake).get()
+        # Spin the loop a few times to make sure we run the callbacks.
+        # If we neglect to spin, we don't trigger the bug.
+        # If error handling is correct, the exception raised from the callback
+        # will be surfaced in the main greenlet. On windows, it can sometimes take
+        # more than one spin for some reason; if we don't catch it here, then
+        # some other test is likely to die unexpectedly with MyException.
+        with self.assertRaises(MyException):
+            for _ in range(5):
+                gevent.sleep(0.001)
+
+
+        self.assertEqual(called, [1, 1])
+        # But value was cleared in a finally block
+        self.assertIsNone(tr.value)
+        self.assertIsNotNone(tr.receiver)
+
+
 if __name__ == '__main__':
     greentest.main()

src/gevent/threadpool.py

@@ -521,13 +521,11 @@ class ThreadResult(object):
         aw.stop()
         aw.close()
 
-
         # Typically this is pool.semaphore.release and we have to
         # call this in the Hub; if we don't we get the dreaded
         # LoopExit (XXX: Why?)
-        self._call_when_ready()
-
         try:
+            self._call_when_ready()
             if self.exc_info:
                 self.hub.handle_error(self.context, *self.exc_info)
             self.context = None