不能 stop\kill 一次由 multiprocessing.Pool 产生的所有进程

Question

当有 error\exception 时，我需要 stop\kill 所有进程。我在 StackOwerflow 解决方案中发现使用 psutil 杀死所有进程，但有时我会遇到一个问题 - 当 psutil 杀死子进程和主进程时，新进程可能会启动并且代码继续执行。

import psutil

class MyClass:
    parent_pid = 0
    ids_list = range(300)

    def main(self):
        self.parent_pid = os.getpid()
        pool = multiprocessing.Pool(3)

        for osm_id in self.ids_list:
            pool.apply_async(self.handle_country_or_region,  
                             kwds=dict(country_id=osm_id),
                             error_callback=self.kill_proc_tree)

        pool.close()
        pool.join()

    def kill_proc_tree(self, including_parent=True):
        parent = psutil.Process(self.parent_pid)
        children = parent.children(recursive=True)

        for child in children:
            child.kill()
        psutil.wait_procs(children, timeout=5)

        if including_parent:
            parent.kill()
            parent.wait(5)

    def handle_country_or_region(self, country_id=None, queue=None):
        pass
        # here I do some task

看来我需要终止池而不是终止进程，但在这种情况下，如果我这样做

pool.close()
pool.terminate()
pool.join()

我的终端停止执行任何操作，新行完全为空（即没有“>>>”）并且没有任何反应。

理想情况下，我希望有下一个流程：如果有任何 error\exception，stop\kill 所有代码执行并返回到终端中的交互式提示。

谁能帮我让它正常工作？我使用 Python 3.5 和 Ubuntu 15.10

Answer 1

解决方案非常简单 - 将 'killer'-函数放在 'main'.

中

完整代码如下：

class MyClass:
    ids_list = range(300)

    def main(self):
        pool = multiprocessing.Pool(3)

        def kill_pool(err_msg):
            print(err_msg)
            pool.terminate()

        for osm_id in self.ids_list:
            pool.apply_async(self.handle_country_or_region,     
                             kwds=dict(country_id=osm_id),
                             error_callback=kill_pool)

        pool.close()
        pool.join()

    def handle_country_or_region(self, country_id=None, queue=None):
        pass  # here I do some task

如果有人需要使用 queue，下面是代码的扩展变体，它展示了如何以正确的方式完成 queue，避免出现僵尸进程：

import pickle
import os
import multiprocessing

class MyClass:
    ids_list = range(300)
    folder = os.path.join(os.getcwd(), 'app_geo')
    STOP_TOKEN = 'stop queue'

    def main(self):

        # >>> Queue part shared between processes <<<
        manager = multiprocessing.Manager()
        remove_id_queue = manager.Queue()

        remove_id_process = multiprocessing.Process(target=self.remove_id_from_file,
                                                    args=(remove_id_queue,))
        remove_id_process.start()
        # >>> End of queue part <<<

        pool = multiprocessing.Pool(3)

        def kill_pool(err_msg):
            print(err_msg)
            pool.terminate()

        for osm_id in self.ids_list:
            pool.apply_async(self.handle_country_or_region,     
                             kwds=dict(country_id=osm_id),
                             error_callback=kill_pool)

        pool.close()
        pool.join()

        # >>> Anti-zombie processes queue part <<<
        remove_id_queue.put(self.STOP_TOKEN)
        remove_id_process.join()
        manager.shutdown()
        # >>> End

    def handle_country_or_region(self, country_id=None, queue=None):
        # here I do some task
        queue.put(country_id)

    def remove_id_from_file(self, some_queue):
        while True:
            osm_id = some_queue.get()
            if osm_id == self.STOP_TOKEN:
                return
            self.ids_list.remove(osm_id)
            with open(self.folder + '/ids_list.pickle', 'wb') as f:
                pickle.dump(self.ids_list, f)

不能 stop\kill 一次由 multiprocessing.Pool 产生的所有进程

Cant stop\kill all processes at once produced by multiprocessing.Pool

python

multiprocessing

python-3.x