| 679 | |
| 680 | @classmethod |
| 681 | def _terminate_pool(cls, taskqueue, inqueue, outqueue, pool, change_notifier, |
| 682 | worker_handler, task_handler, result_handler, cache): |
| 683 | # this is guaranteed to only be called once |
| 684 | util.debug('finalizing pool') |
| 685 | |
| 686 | # Notify that the worker_handler state has been changed so the |
| 687 | # _handle_workers loop can be unblocked (and exited) in order to |
| 688 | # send the finalization sentinel all the workers. |
| 689 | worker_handler._state = TERMINATE |
| 690 | change_notifier.put(None) |
| 691 | |
| 692 | task_handler._state = TERMINATE |
| 693 | |
| 694 | util.debug('helping task handler/workers to finish') |
| 695 | cls._help_stuff_finish(inqueue, task_handler, len(pool)) |
| 696 | |
| 697 | if (not result_handler.is_alive()) and (len(cache) != 0): |
| 698 | raise AssertionError( |
| 699 | "Cannot have cache with result_handler not alive") |
| 700 | |
| 701 | result_handler._state = TERMINATE |
| 702 | change_notifier.put(None) |
| 703 | outqueue.put(None) # sentinel |
| 704 | |
| 705 | # We must wait for the worker handler to exit before terminating |
| 706 | # workers because we don't want workers to be restarted behind our back. |
| 707 | util.debug('joining worker handler') |
| 708 | if threading.current_thread() is not worker_handler: |
| 709 | worker_handler.join() |
| 710 | |
| 711 | # Terminate workers which haven't already finished. |
| 712 | if pool and hasattr(pool[0], 'terminate'): |
| 713 | util.debug('terminating workers') |
| 714 | for p in pool: |
| 715 | if p.exitcode is None: |
| 716 | p.terminate() |
| 717 | |
| 718 | util.debug('joining task handler') |
| 719 | if threading.current_thread() is not task_handler: |
| 720 | task_handler.join() |
| 721 | |
| 722 | util.debug('joining result handler') |
| 723 | if threading.current_thread() is not result_handler: |
| 724 | result_handler.join() |
| 725 | |
| 726 | if pool and hasattr(pool[0], 'terminate'): |
| 727 | util.debug('joining pool workers') |
| 728 | for p in pool: |
| 729 | if p.is_alive(): |
| 730 | # worker has not yet exited |
| 731 | util.debug('cleaning up worker %d' % p.pid) |
| 732 | p.join() |
| 733 | |
| 734 | def __enter__(self): |
| 735 | self._check_running() |