Repository: itamarst/crochet

Issue:
MemoryErrors when logging handlers raise exceptions
This lets a python 2.7.11 + twisted 14.0 + crochet 1.4 (win32, 32-Bit) grow to 2 GB in size, and be pretty much disfunctional for anything (but not crashing sometimes, only raising a MemoryError if lucky). It also has the same effect for Twisted 17.1.

    import time
    import crochet
    import logging

    class BrokenHandler(logging.Handler):
        def emit(self, record):
             raise ValueError("Kaboom")

    log = logging.getLogger("twisted")
    h = BrokenHandler(logging.DEBUG)
    log.addHandler(h)
    log.setLevel(logging.DEBUG)

    crochet.setup()
    while 1:
         time.sleep(10)

It should not hang like that, even if the logger misbehaves.


Discussion:

Patch:
diff --git a/.travis.yml b/.travis.yml
index c080557..3777fa6 100644
--- a/.travis.yml
+++ b/.travis.yml
@@ -1,14 +1,16 @@
 language: python
 
 env:
-  - TWISTED=Twisted==15.0 RUNTESTS=trial
-  - TWISTED=Twisted RUNTESTS=trial
+  # Oldest supported version:
+  - TWISTED=Twisted==16.0
+  # Latest Twisted:
+  - TWISTED=Twisted
 
 python:
   - 2.7
   - 3.4
   - 3.5
-  - 3.6-dev
+  - 3.6
   - pypy
 
 install:
@@ -17,7 +19,7 @@ install:
   - python setup.py -q install
 
 script:
-  - $RUNTESTS crochet.tests
+  - trial crochet.tests
   - pyflakes crochet
 
 notifications:
diff --git a/crochet/_eventloop.py b/crochet/_eventloop.py
index d6f7baa..9980e99 100644
--- a/crochet/_eventloop.py
+++ b/crochet/_eventloop.py
@@ -298,7 +298,15 @@ class ThreadLogObserver(object):
         """
         A log observer that writes to a queue.
         """
-        self._logWritingReactor.callFromThread(self._observer, msg)
+        def log():
+            try:
+                self._observer(msg)
+            except:
+                # Lower-level logging system blew up, nothing we can do, so
+                # just drop on the floor.
+                pass
+
+        self._logWritingReactor.callFromThread(log)
 
 
 class EventLoop(object):
diff --git a/docs/news.rst b/docs/news.rst
index 7d97d31..1ce83fc 100644
--- a/docs/news.rst
+++ b/docs/news.rst
@@ -1,6 +1,18 @@
 What's New
 ==========
 
+1.7.0
+^^^^^
+
+Bug fixes:
+
+* If the Python `logging.Handler` throws an exception Crochet no longer goes into a death spiral.
+  Thanks to Michael Schlenker for the bug report.
+
+Removed features:
+
+* Versions of Twisted < 16.0 are no longer supported (i.e. no longer tested in CI.)
+
 1.6.0
 ^^^^^
 


Test patch:
diff --git a/crochet/tests/test_logging.py b/crochet/tests/test_logging.py
new file mode 100644
index 0000000..9dc351b
--- /dev/null
+++ b/crochet/tests/test_logging.py
@@ -0,0 +1,86 @@
+"""Tests for the logging bridge."""
+
+from __future__ import absolute_import
+
+from twisted.trial.unittest import SynchronousTestCase
+import threading
+
+from twisted.python import threadable
+
+from .._eventloop import ThreadLogObserver
+
+
+class ThreadLogObserverTest(SynchronousTestCase):
+    """
+    Tests for ThreadLogObserver.
+
+    We use Twisted's SyncTestCase to ensure that unhandled logged errors get
+    reported as errors, in particular for test_error.
+    """
+    def test_stop(self):
+        """
+        ThreadLogObserver.stop() stops the thread started in __init__.
+        """
+        threadLog = ThreadLogObserver(None)
+        self.assertTrue(threadLog._thread.is_alive())
+        threadLog.stop()
+        threadLog._thread.join()
+        self.assertFalse(threadLog._thread.is_alive())
+
+    def test_emit(self):
+        """
+        ThreadLogObserver.emit runs the wrapped observer's in its thread, with
+        the given message.
+        """
+        messages = []
+        def observer(msg):
+            messages.append((threading.current_thread().ident, msg))
+
+        threadLog = ThreadLogObserver(observer)
+        ident = threadLog._thread.ident
+        msg1 = {}
+        msg2 = {"a": "b"}
+        threadLog(msg1)
+        threadLog(msg2)
+        threadLog.stop()
+        # Wait for writing to finish:
+        threadLog._thread.join()
+        self.assertEqual(messages, [(ident, msg1), (ident, msg2)])
+
+    def test_errors(self):
+        """
+        ThreadLogObserver.emit catches and silently drops exceptions from its
+        observer.
+        """
+        messages = []
+        counter = []
+        def observer(msg):
+            counter.append(1)
+            if len(counter) == 2:
+                raise RuntimeError("ono a bug")
+            messages.append(msg)
+
+        threadLog = ThreadLogObserver(observer)
+        msg1 = {"m": "1"}
+        msg2 = {"m": "2"}
+        msg3 = {"m": "3"}
+        threadLog(msg1)
+        threadLog(msg2)
+        threadLog(msg3)
+        threadLog.stop()
+        # Wait for writing to finish:
+        threadLog._thread.join()
+        self.assertEqual(messages, [msg1, msg3])
+
+    def test_ioThreadUnchanged(self):
+        """
+        ThreadLogObserver does not change the Twisted I/O thread (which is
+        supposed to match the thread the main reactor is running in.)
+        """
+        threadLog = ThreadLogObserver(None)
+        threadLog.stop()
+        threadLog._thread.join()
+        self.assertIn(threadable.ioThread,
+                      # Either reactor was never run, or run in thread running
+                      # the tests:
+                      (None, threading.current_thread().ident))
diff --git a/crochet/tests/test_setup.py b/crochet/tests/test_setup.py
index f60f99e..5bc7850 100644
--- a/crochet/tests/test_setup.py
+++ b/crochet/tests/test_setup.py
@@ -14,7 +14,6 @@ import twisted
 from twisted.python.log import PythonLoggingObserver
 from twisted.python import log
 from twisted.python.runtime import platform
-from twisted.python import threadable
 from twisted.internet.task import Clock
 
 from .._eventloop import EventLoop, ThreadLogObserver, _store
@@ -265,56 +264,6 @@ class ProcessSetupTests(TestCase):
         self.assertFalse(reactor.getDelayedCalls())
 
 
-class ThreadLogObserverTest(TestCase):
-    """
-    Tests for ThreadLogObserver.
-    """
-    def test_stop(self):
-        """
-        ThreadLogObserver.stop() stops the thread started in __init__.
-        """
-        threadLog = ThreadLogObserver(None)
-        self.assertTrue(threadLog._thread.is_alive())
-        threadLog.stop()
-        threadLog._thread.join()
-        self.assertFalse(threadLog._thread.is_alive())
-
-    def test_emit(self):
-        """
-        ThreadLogObserver.emit runs the wrapped observer's in its thread, with
-        the given message.
-        """
-        messages = []
-        def observer(msg):
-            messages.append((threading.current_thread().ident, msg))
-
-        threadLog = ThreadLogObserver(observer)
-        ident = threadLog._thread.ident
-        msg1 = {}
-        msg2 = {"a": "b"}
-        threadLog(msg1)
-        threadLog(msg2)
-        threadLog.stop()
-        # Wait for writing to finish:
-        threadLog._thread.join()
-        self.assertEqual(messages, [(ident, msg1), (ident, msg2)])
-
-
-    def test_ioThreadUnchanged(self):
-        """
-        ThreadLogObserver does not change the Twisted I/O thread (which is
-        supposed to match the thread the main reactor is running in.)
-        """
-        threadLog = ThreadLogObserver(None)
-        threadLog.stop()
-        threadLog._thread.join()
-        self.assertIn(threadable.ioThread,
-                      # Either reactor was never run, or run in thread running
-                      # the tests:
-                      (None, threading.current_thread().ident))
-
-
-
 class ReactorImportTests(TestCase):
     """
     Tests for when the reactor gets imported.