在多处理中启动嵌套进程

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模块创建子进程时,我不确定如何设置守护进程属性.您可以尝试通过初始化列表传递此属性.