summaryrefslogtreecommitdiff
path: root/Lib/multiprocessing
diff options
context:
space:
mode:
authorThomas Moreau <thomas.moreau.2010@gmail.com>2018-03-21 16:50:28 +0100
committerAntoine Pitrou <pitrou@free.fr>2018-03-21 16:50:28 +0100
commite2f33add635df4fde81be9960bab367e010c19bf (patch)
tree26f61daafe01350703a9f0d40f962e5789b086a7 /Lib/multiprocessing
parent9308dea3e1fd565d50a76a667e4e8ef0568b7053 (diff)
downloadcpython-git-e2f33add635df4fde81be9960bab367e010c19bf.tar.gz
bpo-33078 - Fix queue size on pickling error (GH-6119)
Diffstat (limited to 'Lib/multiprocessing')
-rw-r--r--Lib/multiprocessing/queues.py10
1 files changed, 8 insertions, 2 deletions
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