python - 在多处理中启动嵌套进程

Question

我有一个启动多个进程的主文件，其中一个进程再次启动多个进程。我在启动嵌套进程集时遇到问题。

我在一个文件中有以下代码:

# parallel_test.py
import Queue
import multiprocessing
import time
import threading


def worker(q):
    while not q.empty():
        try:
            row = q.get(False)
            print row

            time.sleep(1)

        except Queue.Empty:
            break


def main():
    print 'creating queue'
    q = multiprocessing.Queue()

    print 'enqueuing'
    for i in range(100):
        q.put(i)

    num_processes = 15
    pool = []

    for i in range(num_processes):
        print 'launching process {0}'.format(i)
        p = multiprocessing.Process(target=worker, args=(q,))
        p.start()
        pool.append(p)

    for p in pool:
        p.join()

if __name__ == '__main__':
    main()

单独运行此文件 python parallel_test.py 工作正常并按预期打印数字。但是从另一个文件作为另一个进程启动它会导致问题。我的主文件:

# main_loop_test.py
import parallel_test
from multiprocessing import Pool
import time


def main():
    targets = [parallel_test.main]

    running = True

    while running:
        try:
            p = Pool(12)

            for target in targets:
                p.apply_async(target)

            p.close()  # For some reason you need to run close() before join()
            p.join()  # What for all the steps to be done

            print 'All steps done'

            time.sleep(2)

        except KeyboardInterrupt as e:
            print "<<<<<<<<<<<<<<<<<<CAUGHT KEYBOARD INTERRUPT FROM USER>>>>>>>>>>>>>>>>>>>"
            running = False


if __name__ == '__main__':
    main()

parallel_test.py 似乎尝试启动一个进程(什么都不做)然后退出该函数，main_loop_test.py 打印“所有步骤已完成”。没有数字被打印出来。输出:

creating queue
enqueuing
launching process 0
All steps done
creating queue
enqueuing
launching process 0
All steps done

怎么了？我在使用 Pool 而不是在 parallel_test.py 中自己管理进程时遇到了同样的问题。不过，用线程代替多处理是可行的。

Answer 1

当您从另一个程序将它作为子进程调用时，您无法从 parallel_test 创建子进程，因为该进程被创建为守护进程，正如链接 https://docs.python.org/2/library/multiprocessing.html 中提到的那样, 守护进程不允许创建子进程。您必须通过将进程的守护进程属性设置为 false 来将进程创建为非守护进程，如下所示。

p = multiprocessing.Process(target=test.main)
p.daemon = False
p.start()
p.join()

当您通过 Pool 模块创建子进程时，我不确定如何设置守护进程属性。您可以尝试通过初始化列表传递此属性。