bpo-33078 - Fix queue size on pickling error (GH-6119)

Author: tomMoral

Lib/multiprocessing/queues.py

@@ -161,7 +161,7 @@ def _start_thread(self):
             target=Queue._feed,
             args=(self._buffer, self._notempty, self._send_bytes,
                   self._wlock, self._writer.close, self._ignore_epipe,
-                  self._on_queue_feeder_error),
+                  self._on_queue_feeder_error, self._sem),
             name='QueueFeederThread'
         )
         self._thread.daemon = True
@@ -203,7 +203,7 @@ def _finalize_close(buffer, notempty):
 
     @staticmethod
     def _feed(buffer, notempty, send_bytes, writelock, close, ignore_epipe,
-              onerror):
+              onerror, queue_sem):
         debug('starting thread to feed data to pipe')
         nacquire = notempty.acquire
         nrelease = notempty.release
@@ -255,6 +255,12 @@ def _feed(buffer, notempty, send_bytes, writelock, close, ignore_epipe,
                     info('error in queue thread: %s', e)
                     return
                 else:
+                    # Since the object has not been sent in the queue, we need
+                    # to decrease the size of the queue. The error acts as
+                    # if the object had been silently removed from the queue
+                    # and this step is necessary to have a properly working
+                    # queue.
+                    queue_sem.release()
                     onerror(e, obj)
 
     @staticmethod

Lib/test/_test_multiprocessing.py

@@ -1056,6 +1056,19 @@ def __reduce__(self):
             self.assertTrue(q.get(timeout=1.0))
             close_queue(q)
 
+        with test.support.captured_stderr():
+            # bpo-33078: verify that the queue size is correctly handled
+            # on errors.
+            q = self.Queue(maxsize=1)
+            q.put(NotSerializable())
+            q.put(True)
+            self.assertEqual(q.qsize(), 1)
+            # bpo-30595: use a timeout of 1 second for slow buildbots
+            self.assertTrue(q.get(timeout=1.0))
+            # Check that the size of the queue is correct
+            self.assertEqual(q.qsize(), 0)
+            close_queue(q)
+
     def test_queue_feeder_on_queue_feeder_error(self):
         # bpo-30006: verify feeder handles exceptions using the
         # _on_queue_feeder_error hook.

Misc/NEWS.d/next/Library/2018-03-15-07-38-00.bpo-33078.RmjUF5.rst

@@ -0,0 +1,2 @@
+Fix the size handling in multiprocessing.Queue when a pickling error
+occurs.