From e2f33add635df4fde81be9960bab367e010c19bf Mon Sep 17 00:00:00 2001 From: Thomas Moreau Date: Wed, 21 Mar 2018 16:50:28 +0100 Subject: bpo-33078 - Fix queue size on pickling error (GH-6119) --- Lib/multiprocessing/queues.py | 10 ++++++++-- 1 file changed, 8 insertions(+), 2 deletions(-) (limited to 'Lib/multiprocessing') diff --git a/Lib/multiprocessing/queues.py b/Lib/multiprocessing/queues.py index d66d37a5c3..715a9b0e12 100644 --- a/Lib/multiprocessing/queues.py +++ b/Lib/multiprocessing/queues.py @@ -161,7 +161,7 @@ class Queue(object): 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 @@ class Queue(object): @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 @@ class Queue(object): 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 -- cgit v1.2.1