| 228 | |
| 229 | @staticmethod |
| 230 | def _feed(buffer, notempty, send_bytes, writelock, reader_close, |
| 231 | writer_close, ignore_epipe, onerror, queue_sem): |
| 232 | debug('starting thread to feed data to pipe') |
| 233 | nacquire = notempty.acquire |
| 234 | nrelease = notempty.release |
| 235 | nwait = notempty.wait |
| 236 | bpopleft = buffer.popleft |
| 237 | sentinel = _sentinel |
| 238 | if sys.platform != 'win32': |
| 239 | wacquire = writelock.acquire |
| 240 | wrelease = writelock.release |
| 241 | else: |
| 242 | wacquire = None |
| 243 | |
| 244 | while 1: |
| 245 | try: |
| 246 | nacquire() |
| 247 | try: |
| 248 | if not buffer: |
| 249 | nwait() |
| 250 | finally: |
| 251 | nrelease() |
| 252 | try: |
| 253 | while 1: |
| 254 | obj = bpopleft() |
| 255 | if obj is sentinel: |
| 256 | debug('feeder thread got sentinel -- exiting') |
| 257 | reader_close() |
| 258 | writer_close() |
| 259 | return |
| 260 | |
| 261 | # serialize the data before acquiring the lock |
| 262 | obj = _ForkingPickler.dumps(obj) |
| 263 | if wacquire is None: |
| 264 | send_bytes(obj) |
| 265 | else: |
| 266 | wacquire() |
| 267 | try: |
| 268 | send_bytes(obj) |
| 269 | finally: |
| 270 | wrelease() |
| 271 | except IndexError: |
| 272 | pass |
| 273 | except Exception as e: |
| 274 | if ignore_epipe and getattr(e, 'errno', 0) == errno.EPIPE: |
| 275 | return |
| 276 | # Since this runs in a daemon thread the resources it uses |
| 277 | # may be become unusable while the process is cleaning up. |
| 278 | # We ignore errors which happen after the process has |
| 279 | # started to cleanup. |
| 280 | if is_exiting(): |
| 281 | info('error in queue thread: %s', e) |
| 282 | return |
| 283 | else: |
| 284 | # Since the object has not been sent in the queue, we need |
| 285 | # to decrease the size of the queue. The error acts as |
| 286 | # if the object had been silently removed from the queue |
| 287 | # and this step is necessary to have a properly working |