干净的Python多进程终止取决于出口标志 [英] Clean Python multiprocess termination dependant on an exit flag

问题描述

我正在尝试使用多个进程创建一个程序，如果发生错误，我想干净地终止所有产生的进程.下面，我为我想做的事情写了一些伪类型代码，但是我不知道与所有发生错误的进程进行通信的最佳方法是什么，它们应该终止.

I am attempting to create a program using multiple processes and I would like to cleanly terminate all the spawned processes if errors occur. below I've wrote out some pseudo type code for what I think I need to do but I don't know what the best way is to communicate to all the processes that an error has occured and they should terminate.

我认为我应该在此类中使用类，但是我对Python还是很陌生，所以我只是想首先了解一些基础知识.

I think I should be using classes for this sort of thing but I'm quite new to Python so I'm just trying to get my head around the basics first.

#imports

exitFlag = True

# Function for threads to process   
def url_thread_worker( ):   
# while exitFlag:
    try:
        # do something
    except:
        # we've ran into a problem, we need to kill all the spawned processes and cleanly exit the program
        exitFlag = False

def processStarter( ):

    process_1 = multiprocessing.Process( name="Process-1", target=url_thread_worker, args=( ) ) 
    process_2 = multiprocessing.Process( name="Process-2", target=url_thread_worker, args=( ) ) 

    process_1.start()
    process_2.start()


if __name__ == '__main__':
     processStarter( )

预先感谢

推荐答案

这是我的建议:

import multiprocessing
import threading
import time

def good_worker():   
    print "[GoodWorker] Starting"
    time.sleep(4)
    print "[GoodWorker] all good"

def bad_worker():
    print "[BadWorker] Starting"
    time.sleep(2)
    raise Exception("ups!")

class MyProcManager(object):
    def __init__(self):
        self.procs = []
        self.errors_flag = False
        self._threads = []
        self._lock = threading.Lock()

    def terminate_all(self):
        with self._lock:
            for p in self.procs:
                if p.is_alive():
                    print "Terminating %s" % p
                    p.terminate()

    def launch_proc(self, func, args=(), kwargs= {}):
        t = threading.Thread(target=self._proc_thread_runner,
                             args=(func, args, kwargs))
        self._threads.append(t)
        t.start()

    def _proc_thread_runner(self, func, args, kwargs):
        p = multiprocessing.Process(target=func, args=args, kwargs=kwargs)
        self.procs.append(p)
        p.start()
        while p.exitcode is None:
            p.join()
        if p.exitcode > 0:
            self.errors_flag = True
            self.terminate_all()

    def wait(self):
        for t in self._threads:
            t.join()

if __name__ == '__main__':
    proc_manager = MyProcManager()
    proc_manager.launch_proc(good_worker) 
    proc_manager.launch_proc(good_worker) 
    proc_manager.launch_proc(bad_worker) 
    proc_manager.wait()
    if proc_manager.errors_flag:
        print "Errors flag is set: some process crashed"
    else:
        print "Everything closed cleanly"

您需要为每个进程运行一个包装线程，等待其结束. 进程结束时，检查退出代码:如果> 0，则表示它引发了一些未处理的异常.现在调用Terminate_all()关闭所有剩余的活动进程. 包装线程也将完成，因为它们取决于进程的运行.

You need to have a wrapper thread for each process run, that waits for its end. When a process ends, check for the exitcode: if > 0, means it raised some unhandled exception. Now call terminate_all() to close all remaining active processes. The wrapper threads will also finish as they are dependent on the process run.

此外，在您的代码中，您可以随时随意调用proc_manager.terminate_all().您可以在其他线程或类似的线程中检查某些标志.

Also, in your code you're completely free to call proc_manager.terminate_all() whenever you want. You can be checking for some flags in a different thread or something like that..

希望对您的情况有益.

PS:顺便说一句..在您的原始代码中，您执行了类似于全局exit_flag的操作:多重处理中永远不会有全局" exit_flag，因为当您使用具有独立内存空间的独立进程时，它根本就不是全局的.这仅在可以共享状态的线程环境中有效.如果您需要在多处理中使用它，则必须在进程之间进行明确的通信( Pipe and Queue完成此操作)或类似共享内存的操作对象

PS: btw.. in your original code you did something like an global exit_flag: you can never have a "global" exit_flag in multiprocessing because it simply ain't global as you are using separated processes with separated memory spaces. That only works in threaded environments where state can be shared. If you need it in multiprocessing then you must have explicit communication between processes (Pipe and Queue accomplish that) or something like shared memory objects

这篇关于干净的Python多进程终止取决于出口标志的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！