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