python · pitrou · May 25, 2017 · May 13, 2017 · May 25, 2017 · May 25, 2017
diff --git a/Lib/multiprocessing/queues.py b/Lib/multiprocessing/queues.py
@@ -221,8 +221,8 @@ def _feed(buffer, notempty, send_bytes, writelock, close, ignore_epipe):
        else:
            wacquire = None

-        try:
-            while 1:
+        while 1:
+            try:
                nacquire()
                try:
                    if not buffer:
@@ -249,21 +249,19 @@ def _feed(buffer, notempty, send_bytes, writelock, close, ignore_epipe):
                                wrelease()
                except IndexError:
                    pass
-        except Exception as e:
-            if ignore_epipe and getattr(e, 'errno', 0) == errno.EPIPE:
-                return
-            # Since this runs in a daemon thread the resources it uses
-            # may be become unusable while the process is cleaning up.
-            # We ignore errors which happen after the process has
-            # started to cleanup.
-            try:
+            except Exception as e:
+                if ignore_epipe and getattr(e, 'errno', 0) == errno.EPIPE:
+                    return
+                # Since this runs in a daemon thread the resources it uses
+                # may be become unusable while the process is cleaning up.
+                # We ignore errors which happen after the process has
+                # started to cleanup.
                if is_exiting():
                    info('error in queue thread: %s', e)
+                    return
                else:
                    import traceback
                    traceback.print_exc()
-            except Exception:
-                pass

 _sentinel = object()


diff --git a/Lib/test/_test_multiprocessing.py b/Lib/test/_test_multiprocessing.py
@@ -752,6 +752,20 @@ def test_timeout(self):
        # Windows (usually 15.6 ms)
        self.assertGreaterEqual(delta, 0.170)

+    def test_queue_feeder_donot_stop_onexc(self):
+        # bpo-30414: verify feeder handles exceptions correctly
+        if self.TYPE != 'processes':
+            self.skipTest('test not appropriate for {}'.format(self.TYPE))
+
+        class NotSerializable(object):
+            def __reduce__(self):
+                raise AttributeError
+        with test.support.captured_stderr():
+            q = self.Queue()
+            q.put(NotSerializable())
+            q.put(True)
+            self.assertTrue(q.get(timeout=0.1))
+
 #
 #
 #

diff --git a/Misc/NEWS b/Misc/NEWS
@@ -341,6 +341,9 @@ Extension Modules
 Library
 -------

+- bpo-30414: multiprocessing.Queue._feed background running
+  thread do not break from main loop on exception.
+
 - bpo-30003: Fix handling escape characters in HZ codec.  Based on patch
  by Ma Lin.