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