| 572 | |
| 573 | @staticmethod |
| 574 | def _handle_results(outqueue, get, cache): |
| 575 | thread = threading.current_thread() |
| 576 | |
| 577 | while 1: |
| 578 | try: |
| 579 | task = get() |
| 580 | except (OSError, EOFError): |
| 581 | util.debug('result handler got EOFError/OSError -- exiting') |
| 582 | return |
| 583 | |
| 584 | if thread._state != RUN: |
| 585 | assert thread._state == TERMINATE, "Thread not in TERMINATE" |
| 586 | util.debug('result handler found thread._state=TERMINATE') |
| 587 | break |
| 588 | |
| 589 | if task is None: |
| 590 | util.debug('result handler got sentinel') |
| 591 | break |
| 592 | |
| 593 | job, i, obj = task |
| 594 | try: |
| 595 | cache[job]._set(i, obj) |
| 596 | except KeyError: |
| 597 | pass |
| 598 | task = job = obj = None |
| 599 | |
| 600 | while cache and thread._state != TERMINATE: |
| 601 | try: |
| 602 | task = get() |
| 603 | except (OSError, EOFError): |
| 604 | util.debug('result handler got EOFError/OSError -- exiting') |
| 605 | return |
| 606 | |
| 607 | if task is None: |
| 608 | util.debug('result handler ignoring extra sentinel') |
| 609 | continue |
| 610 | job, i, obj = task |
| 611 | try: |
| 612 | cache[job]._set(i, obj) |
| 613 | except KeyError: |
| 614 | pass |
| 615 | task = job = obj = None |
| 616 | |
| 617 | if hasattr(outqueue, '_reader'): |
| 618 | util.debug('ensuring that outqueue is not full') |
| 619 | # If we don't make room available in outqueue then |
| 620 | # attempts to add the sentinel (None) to outqueue may |
| 621 | # block. There is guaranteed to be no more than 2 sentinels. |
| 622 | try: |
| 623 | for i in range(10): |
| 624 | if not outqueue._reader.poll(): |
| 625 | break |
| 626 | get() |
| 627 | except (OSError, EOFError): |
| 628 | pass |
| 629 | |
| 630 | util.debug('result handler exiting: len(cache)=%s, thread._state=%s', |
| 631 | len(cache), thread._state) |