管理和终止任何进程的强大方法

Question

我正在编写代码以并行运行实验。我无法控制实验的作用，他们可能会打开使用 subprocess.Popen 或 check_output 来运行一个或多个额外的子进程。

我有两个条件：我希望能够终止超过超时的实验，并且我想终止 KeyboardInterrupt 上的实验。

大多数终止进程的方法并不能确保所有子进程等都被杀死。如果 100 次实验一个接一个地运行，但它们都生成了在超时发生后仍然存活的子进程并且实验被认为被终止，这显然是一个问题。

我现在处理这个问题的方式是包含将实验配置存储在数据库中的代码，生成从命令行加载和运行实验的代码，然后通过subprocess.Popen(cmd, shell=True, start_new_session=True) 调用这些命令并使用os.killpg 杀死它们超时。

我的主要问题然后是：通过命令行调用这些实验感觉很麻烦，那么有没有办法直接通过multiprocessing.Process(target=fn)调用代码并达到start_new_session=True + @987654329相同的效果@ 超时和KeyboardInterrupt?

<file1>
def run_exp(config):
    do work
    return result

if __name__ == "__main__":
    save_exp(run_exp(load_config(sys.args)))

<file2>
def monitor(queue):
    active = set()  # active process ids
    while True:
        msg = queue.get()
        if msg == "sentinel":
             <loop over active ids and kill them with os.killpg>
        else:
            <add or remove id from active set>


def worker(args):
    id, queue = args
    command = f"python <file1> {id}"
    with subprocess.Popen(command, shell=True, ..., start_new_session=True) as process:
        try:
            queue.put(f"start {process.pid}")
            process.communicate(timeout=timeout)
        except TimeoutExpired:
            os.killpg(process.pid, signal.SIGINT)  # send signal to the process group
            process.communicate()
        finally:
            queue.put(f"done {process.pid}")

def main():
    <save configs => c_ids>
    queue = manager.Queue()
    process = Process(target=monitor, args=(queue,))
    process.start()

    def clean_exit():
        queue.put("sentinel")
        <terminate pool and monitor process>

    r = pool.map_async(worker, [(c_id, queue) for c_id in c_ids])
    atexit.register(clean_exit)
    r.wait()
    <terminate pool and monitor process>

我发布了一个代码框架，详​​细说明了通过命令行启动进程并终止它们的方法。我的方法的那个版本的另一个复杂之处是，当KeyboardInterrupt 到达时，队列已经终止（因为缺少更好的词）并且不可能与监控进程通信（哨兵消息永远不会到达）。相反，我不得不求助于将进程 ID 写入文件并将文件读回主进程以终止仍在运行的进程。如果您知道解决此队列问题的方法，我很想了解它。

Answer 1

我认为问题在于您正在存储子进程 pid 来杀死它，您需要主机进程 pid，并且您使用了 signal.SIGINT，我认为应该是 signal.SIGTERM。试试这个，而不是这一行：

os.killpg(process.pid, signal.SIGINT)

使用这一行：

os.killpg(os.getpgid(process.pid), signal.SIGTERM) 

Answer 2

我想有一种方法可以避免这种情况，那就是使用 Try catch 块。
说如果KeyboardInterrupt 到达main() 那么你可以试试这个：

def main():
    try:
        <save configs => c_ids>
        queue = manager.Queue()
        process = Process(target=monitor, args=(queue,))
        process.start()

        def clean_exit():
            queue.put("sentinel")
            <terminate pool and monitor process>

        r = pool.map_async(worker, [(c_id, queue) for c_id in c_ids])
        atexit.register(clean_exit)
        r.wait()
        <terminate pool and monitor process>
    except KeyboardInterrupt as e:
        pass
        #write the process you want to keep continuing. 

猜猜这会有所帮助。