一个目录中的多个子进程答案

【问题标题】：Multiple subprocesses in a directory一个目录中的多个子进程
【发布时间】：2017-05-05 20:38:38
【问题描述】：

我正在尝试在 shell 上使用 python 自动进行手动模拟测试。我在父目录中的子文件夹下设置了我的数据，如下所示：

/vol/parent/child1
/vol/parent/child2
/vol/parent/child3
...
/vol/parent/childn

我有单独的脚本，它们在每个子目录中运行一个测试作业（也是一个 python 程序）。脚本使用 subprocess.Popen ，示例如下：

folder_list = next(os.walk(parent))[1]

for child in folder_list:
 f = open("/vol/testjob1/nohup.out", "w")
 p = subprocess.Popen([<...testjob1.py...test parameters>], cwd = '/vol01/testjob1/', stdout=f)

这会在每个子文件夹上运行作业，每个线程 1 个作业。

通常，我等待所有 n 个作业完成，然后运行我的下一个测试（比如 testjob2），这也是一个类似的子进程类型程序。

我正在尝试将所有这些子流程测试脚本组合成 1 个主脚本。主要目标是当一个作业在 1 个线程上完成时，下一个子进程在该线程上开始。我尝试在末尾附加 p.wait() ，但这意味着它将等待每个线程完成，然后再启动下一个线程（这会破坏并行运行这些作业的目的）。

如何结合使用 subprocess.popen、p.wait() 和 for 循环来确保一个进程依次启动，如下所示：

(child1)| thread1: Subprocess1 (done)--> start subprocess2 (done)--> start subprocess3...
(child2)| thread2: Subprocess1 (done)--> start subprocess2 (done)--> start subprocess3...
...
(childn)| threadn: Subprocess1 (done)--> start subprocess2 (done)--> start subprocess3...

【问题讨论】：

我不明白。你真的创建线程吗？因为如果你这样做，你可以在每个线程中循环和等待（这就是我会做的，我什至会在你的情况下使用check_output 而不是Popen（顺便说一句，因为f 在你的循环中是打开的，每个新结果覆盖前一个）
如果您创建线程，我们看不到代码。您是否将Popen 与线程混淆？因为Popen 创建了进程。

标签： python multithreading subprocess

【解决方案1】：

现在您有一个线程启动所有文件夹的作业。您需要将其反转为每个文件夹启动其所有作业的一个线程。使用 ThreadPool 很容易做到这一点，它会为您处理线程启动/排队操作。

import multiprocessing.pool

jobs = [<...testjob1.py...test parameters>,
    <...testjob2.py...test parameters>,
    <...testjob3.py...test parameters>]

def worker(folder, jobs, outfp):
    for job in jobs:
        subprocess.call(job, cwd=folder, stdout=outfp))

folder_list = next(os.walk(parent))[1]
pool = multiprocessing.pool.ThreadPool(len(folder_list))
for result in pool.map(worker, ((folder, jobs, f) for folder in folder_list))):
    pass
pool.close()
pool.join()

如果可以同时在同一个文件夹中运行多个作业，您可以更精细地安排它们。假设一个目录中的作业本质上比其他目录长。在上面的示例中，您将等待“高投票”。在此示例中，工作会更均匀地分布。

import multiprocessing.pool

jobs = [<...testjob1.py...test parameters>,
    <...testjob2.py...test parameters>,
    <...testjob3.py...test parameters>]

def worker(folder, jobs, outfp):
    subprocess.call(job, cwd=folder, stdout=outfp))

folder_list = next(os.walk(parent))[1]
pool = multiprocessing.pool.ThreadPool(len(folder_list))
for result in pool.map(worker, ((folder, job, f) 
        for folder , job in itertools.product(folder_list, jobs))),
        chunksize=1):
    pass
pool.close()
pool.join()

【讨论】：