退出多处理脚本

Question

当目标函数抛出错误时，我试图退出多处理脚本，但父进程并没有退出，而是挂起。

这是我用来复制问题的测试脚本：

#!/usr/bin/python3.5

import time, multiprocessing as mp

def myWait(wait, resultQueue):
    startedAt = time.strftime("%H:%M:%S", time.localtime())
    time.sleep(wait)
    endedAt = time.strftime("%H:%M:%S", time.localtime())
    name = mp.current_process().name
    resultQueue.put((name, wait, startedAt, endedAt))

# queue initialisation
resultQueue = mp.Queue()

# process creation arg: (process number, sleep time, queue)
proc =  [
    mp.Process(target=myWait, name = ' _One_', args=(2, resultQueue,)),
    mp.Process(target=myWait, name = ' _Two_', args=(2, resultQueue,))
    ]

# starting processes
for p in proc:
    p.start()

for p in proc:
    p.join()

# print results
results = {}
for p in proc:
    name, wait, startedAt, endedAt = resultQueue.get()
    print('Process %s started at %s wait %s ended at %s' % (name, startedAt, wait, endedAt))

这很完美，我可以看到父脚本在htop 中生成了两个子进程，但是当我想强制父脚本退出时，如果在myWait 目标函数中抛出错误，父进程就会挂起并且甚至不会产生任何子进程。我必须ctrl-c 才能杀死它。

def myWait(wait, resultQueue):
    try:
        # do something wrong
    except:
        raise SystemExit

我已经尝试了各种退出函数的方法（例如exit()、sys.exit()、os._exit()...），但无济于事。

Answer 1

首先，您的代码存在一个主要问题：您试图在刷新队列内容（如果有）之前加入进程，这可能会导致死锁。请参阅此处标题为“加入使用队列的进程”的部分：https://docs.python.org/3/library/multiprocessing.html#multiprocessing-programming

其次，对resultQueue.get() 的调用将被阻塞，直到它接收到一些数据，这永远不会发生 如果myWait 函数引发异常并且在此之前没有数据被推入队列。所以让它成为非阻塞的，让它在一个循环中检查任何数据，直到它最终收到一些东西或者有什么问题。

这里有一个快速的'n'dirty修复给你的想法：

#!/usr/bin/python3.5

import multiprocessing as mp
import queue
import time

def myWait(wait, resultQueue):
    raise Exception("error!")

# queue initialisation
resultQueue = mp.Queue()

# process creation arg: (process number, sleep time, queue)
proc =  [
    mp.Process(target=myWait, name = ' _One_', args=(2, resultQueue,)),
    mp.Process(target=myWait, name = ' _Two_', args=(2, resultQueue,))
    ]

# starting processes
for p in proc:
    p.start()

# print results
results = {}
for p in proc:
    while True:
        if not p.is_alive():
            break

        try:
            name, wait, startedAt, endedAt = resultQueue.get(block=False)
            print('Process %s started at %s wait %s ended at %s'
                  % (name, startedAt, wait, endedAt))
            break
        except queue.Empty:
            pass

for p in proc:
    p.join()

函数myWait 会抛出异常，但两个进程仍会加入，程序会顺利退出。

Answer 2

您应该使用multiprocessing.Pool 为您管理流程。然后使用Pool.imap_unordered 按完成的顺序迭代结果。一旦您遇到第一个异常，您就可以停止池及其子进程（当您退出 with Pool() as pool 块时，这会自动完成）。例如

from multiprocessing import Pool
import time

def my_wait(args):
    name, wait = args
    if wait == 2:
        raise ValueError("error!")
    else:
        startedAt = time.strftime("%H:%M:%S", time.localtime())
        time.sleep(wait)
        endedAt = time.strftime("%H:%M:%S", time.localtime())
        return name, wait, startedAt, endedAt

if __name__ == "__main__":
    try:
        with Pool() as pool:
            args = [["_One_", 2], ["_Two_", 3]]
            for name, wait, startedAt, endedAt in pool.imap_unordered(my_wait, args):     
                print('Task %s started at %s wait %s ended at %s' % (name,
                    startedAt, wait, endedAt))
    except ValueError as e:
        print(e)

此方法不适合长时间、低工作负载的任务，因为它只会并行运行与其管理的子进程数量一样多的任务（但这是您可以设置的）。如果您需要运行不同的功能，这也不是很好。