如何使用Python的多进程终止进程

Question

如何使用Python的多进程终止进程

pythonmultiprocessingpython-multiprocessing

26

我有一些需要在多个其他系统上运行的代码，这些系统可能会挂起或出现不在我的控制范围内的问题。我想使用Python的多进程来生成子进程独立于主程序运行，然后当它们挂起或出现问题时终止它们，但我不确定最佳方法是什么。

当调用“终止”时，它确实会杀死子进程，但然后它变成一个“僵尸”进程，直到进程对象消失才得以释放。下面的示例代码（其中循环永远不会结束）可以将其终止并在再次调用时允许重新生成，但似乎不是一个好方法（例如，在__init __（）中应该使用multiprocessing.Process()）。

有人有建议吗？

class Process(object):
    def __init__(self):
        self.thing = Thing()
        self.running_flag = multiprocessing.Value("i", 1)

    def run(self):
        self.process = multiprocessing.Process(target=self.thing.worker, args=(self.running_flag,))
        self.process.start()
        print self.process.pid

    def pause_resume(self):
        self.running_flag.value = not self.running_flag.value

    def terminate(self):
        self.process.terminate()

class Thing(object):
    def __init__(self):
        self.count = 1

    def worker(self,running_flag):
        while True:
            if running_flag.value:
                self.do_work()

    def do_work(self):
        print "working {0} ...".format(self.count)
        self.count += 1
        time.sleep(1)

- Dan Littlejohn

我认为直接的答案是，您不希望终止进程 - 因为您已经发现它们会变成僵尸进程，可能会占用内存或其他资源。在我有限的多进程经验中，我只是更加警惕地创建可以优雅地退出的工作进程。这个基本上相同主题的答案是否回答了您的问题？此答案是否解决了您的疑问？ - KobeJohn

不完全正确。我已经阅读了许多类似的线程，它们期望异常发生，然后您可以处理它。我有一个情况，其中某些内容正在运行，但却挂起了。操作员可能会意识到某些内容未正确运行，我希望有一种方法来停止该过程。由于它已经挂起，因此不会响应诸如通过multiprocessing.Value()发送停止标志之类的操作。很可能，他们会杀死整个程序，然后重新启动它，因此我想知道是否有一种好的方法在程序中仅为子进程恢复。 - Dan Littlejohn

我认为我知道你的意思，但我找到的唯一答案就是像上面那样尽力避免“挂起”（例如，用一个捕获所有异常的包装器来包装每个进程）。作为更进一步的证据，在许多情况下，线程可能是更简单的解决方案，并且它有相同的建议。在这一点上，以防您还没有检查 - 确保您真的需要多处理而不是线程，因为它们各自具有优缺点。 - KobeJohn

2个回答

16

Python的多进程处理方式有点令人困惑。

根据多进程指南:

加入僵尸进程

在Unix上，当一个进程完成但没有加入时，它就会成为僵尸进程。由于每次启动新进程（或调用active_children()）时，所有已完成但尚未加入的进程都将被加入，因此不应该有太多僵尸进程。另外，调用已完成进程的Process.is_alive方法将会加入该进程。即便如此，最好还是显式地加入你启动的所有进程。

为了避免进程成为僵尸进程，需要在杀死进程后调用其 join() 方法。

如果您想要一个更简单的处理系统中挂起调用的方法，可以看看 pebble。

- noxdafox

4

那么，调用 process.terminate() 以确保进程终止仍然是必要的吗？或者仅调用 join() 就足以杀死它们了吗？ - weefwefwqg3

6

terminate通过发送SIGTERM信号向目标进程发出终止请求。进程是否遵守请求取决于进程本身(在大多数情况下，它们会遵守)。如果进程已经结束，则它没有任何效果。 join仅仅是指示操作系统在进程结束时回收其资源，否则将一直阻塞到该进程结束。必须始终在已结束的进程上调用join，否则操作系统将积累耗尽的进程的资源。它们通常被称为“僵尸”进程，因为它们实际上是曾经运行过的空壳。 - noxdafox

网页内容由stack overflow 提供, 点击上面的

可以查看英文原文，
原文链接

- Pandemonium · Accepted Answer

你可以将子进程作为后台守护进程来运行。

process.daemon = True

任何守护进程的错误、挂起（或无限循环）都不会影响主进程，只有在主进程退出时才会终止它。

这对于简单问题可以正常工作，直到遇到大量子守护进程，它们将从父进程中不加控制地获取内存。

最好的方法是设置一个队列，让所有子进程与父进程通信，以便我们可以加入它们并进行清理。以下是一些简单的代码，将检查子进程是否挂起（即time.sleep(1000)），并向队列发送消息，以便主进程对其采取行动：

import multiprocessing as mp
import time
import queue

running_flag = mp.Value("i", 1)

def worker(running_flag, q):
    count = 1
    while True:
        if running_flag.value:
            print(f"working {count} ...")
            count += 1
            q.put(count)
            time.sleep(1)
            if count > 3:
                # Simulate hanging with sleep
                print("hanging...")
                time.sleep(1000)

def watchdog(q):
    """
    This check the queue for updates and send a signal to it
    when the child process isn't sending anything for too long
    """
    while True:
        try:
            msg = q.get(timeout=10.0)
        except queue.Empty as e:
            print("[WATCHDOG]: Maybe WORKER is slacking")
            q.put("KILL WORKER")

def main():
    """The main process"""
    q = mp.Queue()

    workr = mp.Process(target=worker, args=(running_flag, q))
    wdog = mp.Process(target=watchdog, args=(q,))

    # run the watchdog as daemon so it terminates with the main process
    wdog.daemon = True

    workr.start()
    print("[MAIN]: starting process P1")
    wdog.start()

    # Poll the queue
    while True:
        msg = q.get()
        if msg == "KILL WORKER":
            print("[MAIN]: Terminating slacking WORKER")
            workr.terminate()
            time.sleep(0.1)
            if not workr.is_alive():
                print("[MAIN]: WORKER is a goner")
                workr.join(timeout=1.0)
                print("[MAIN]: Joined WORKER successfully!")
                q.close()
                break # watchdog process daemon gets terminated

if __name__ == '__main__':
    main()

如果没有终止worker，尝试将其与主进程进行join()操作将会一直阻塞，因为worker从未完成。