issue #456: core: raise error during defer() if Broker shutdown

docs/changelog.rst

@@ -294,6 +294,11 @@ Core Library
   accidental reconfiguration of the :mod:`logging` package in a child from
   setting up a feedback loop.
 
+* `#456 <https://github.com/dw/mitogen/issues/456>`_: a descriptive error is
+  logged when :meth:`mitogen.core.Broker.defer` is called after the broker has
+  shut down, preventing new messages being enqueued that will never be sent,
+  and subsequently producing a program hang.
+
 * `16ca111e <https://github.com/dw/mitogen/commit/16ca111e>`_: handle OpenSSH
   7.5 permission denied prompts when ``~/.ssh/config`` rewrites are present.
 
@@ -325,6 +330,7 @@ bug reports, testing, features and fixes in this release contributed by
 `Johan Beisser <https://github.com/jbeisser>`_,
 `Josh Smift <https://github.com/jbscare>`_,
 `Mehdi <https://github.com/mehdisat7>`_,
+`Michael DeHaan <https://github.com/mpdehaan>`_,
 `Mohammed Naser <https://github.com/mnaser/>`_,
 `@syntonym <https://github.com/syntonym/>`_, and
 `@yodatak <https://github.com/yodatak/>`_.

mitogen/core.py

@@ -2010,10 +2010,18 @@ class Waker(BasicStream):
             if e.args[0] != errno.EBADF:
                 raise
 
+    dead_msg = (
+        "An attempt was made to enqueue a message with a Broker that has "
+        "already begun shut down. If you are receiving this message, it is "
+        "likely your program indirectly called Broker.shutdown() too early."
+    )
+
     def defer(self, func, *args, **kwargs):
         if threading.currentThread().ident == self.broker_ident:
             _vv and IOLOG.debug('%r.defer() [immediate]', self)
             return func(*args, **kwargs)
+        if not self._broker._alive:
+            raise Error(self.dead_msg)
 
         _vv and IOLOG.debug('%r.defer() [fd=%r]', self, self.transmit_side.fd)
         self._lock.acquire()

tests/broker_test.py

@@ -22,6 +22,30 @@ class ShutdownTest(testlib.TestCase):
         actual_close()
 
 
+class DeferTest(testlib.TestCase):
+    klass = mitogen.core.Broker
+
+    def test_defer(self):
+        latch = mitogen.core.Latch()
+        broker = self.klass()
+        try:
+            broker.defer(lambda: latch.put(123))
+            self.assertEquals(123, latch.get())
+        finally:
+            broker.shutdown()
+            broker.join()
+
+    def test_defer_after_shutdown(self):
+        latch = mitogen.core.Latch()
+        broker = self.klass()
+        broker.shutdown()
+        broker.join()
+
+        e = self.assertRaises(mitogen.core.Error,
+            lambda: broker.defer(lambda: latch.put(123)))
+        self.assertEquals(e.args[0], mitogen.core.Waker.dead_msg)
+
+
 class DeferSyncTest(testlib.TestCase):
     klass = mitogen.core.Broker