From 13ae3ed41a2bae74976e815cdef77210ef96a20f Mon Sep 17 00:00:00 2001
From: Grzegorz Grzywacz <grzgrzgrz3@gmail.com>
Date: Sat, 13 May 2017 12:37:08 +0200
Subject: [PATCH 1/2] bpo-30414: multiprocesing.Queue._feed do not break from
 main loop on exc

Queue background running thread was not handling exceptions correctly.
Any exception occurred inside thread (putting unpickable object) cause
feeder to finish running. After that every message put into queue is
silently ignored.
---
 Lib/multiprocessing/queues.py     | 22 ++++++++++------------
 Lib/test/_test_multiprocessing.py | 10 ++++++++++
 2 files changed, 20 insertions(+), 12 deletions(-)

diff --git a/Lib/multiprocessing/queues.py b/Lib/multiprocessing/queues.py
index a4f4ef8b7b5334..7f77837a74acb6 100644
--- 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
index 3eb83c59ab17d7..9d98aad1fce911 100644
--- a/Lib/test/_test_multiprocessing.py
+++ b/Lib/test/_test_multiprocessing.py
@@ -752,6 +752,16 @@ def test_timeout(self):
         # Windows (usually 15.6 ms)
         self.assertGreaterEqual(delta, 0.170)
 
+    def test_queue_feder_donot_stop_onexc(self):
+        class NotSerializable(object):
+            def __reduce__(self):
+                raise AttributeError
+        with test.support.captured_stderr():
+            q = multiprocessing.Queue()
+            q.put(NotSerializable())
+            q.put(True)
+            self.assertTrue(q.get(timeout=0.1))
+
 #
 #
 #

From af2fb35f88aea5fc2d2ea0a87eef7ce70b1820e0 Mon Sep 17 00:00:00 2001
From: Grzegorz Grzywacz <grzgrzgrz3@gmail.com>
Date: Thu, 25 May 2017 11:36:19 +0200
Subject: [PATCH 2/2] bpo-30414: multiprocesing.Queue._feed do not break from
 main loop on exc

Queue background running thread was not handling exceptions correctly.
Any exception occurred inside thread (putting unpickable object) cause
feeder to finish running. After that every message put into queue is
silently ignored.
---
 Lib/test/_test_multiprocessing.py | 8 ++++++--
 Misc/NEWS                         | 3 +++
 2 files changed, 9 insertions(+), 2 deletions(-)

diff --git a/Lib/test/_test_multiprocessing.py b/Lib/test/_test_multiprocessing.py
index 9d98aad1fce911..f1f93674935e7a 100644
--- a/Lib/test/_test_multiprocessing.py
+++ b/Lib/test/_test_multiprocessing.py
@@ -752,12 +752,16 @@ def test_timeout(self):
         # Windows (usually 15.6 ms)
         self.assertGreaterEqual(delta, 0.170)
 
-    def test_queue_feder_donot_stop_onexc(self):
+    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 = multiprocessing.Queue()
+            q = self.Queue()
             q.put(NotSerializable())
             q.put(True)
             self.assertTrue(q.get(timeout=0.1))
diff --git a/Misc/NEWS b/Misc/NEWS
index 74f7922934d90c..18b2ef294b2588 100644
--- a/Misc/NEWS
+++ b/Misc/NEWS
@@ -334,6 +334,9 @@ Extension Modules
 Library
 -------
 
+- bpo-30414: multiprocessing.Queue._feed background running
+  thread do not break from main loop on exception.
+
 - bpo-30149: inspect.signature() now supports callables with
   variable-argument parameters wrapped with partialmethod.
   Patch by Dong-hee Na.