diff options
Diffstat (limited to 'Lib/multiprocessing/queues.py')
| -rw-r--r-- | Lib/multiprocessing/queues.py | 10 |
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 |