Python子进程无需等待即可返回代码

Question

希望我的问题足够具体，与我读过的任何其他问题无关。我想使用 subprocess 和 multiprocessing 来连续生成一堆作业并将返回码返回给我。问题是我不想等待（），所以我可以一次生成所有作业，但我确实想知道它何时完成，以便我可以获得返回码。我遇到了这个奇怪的问题，如果我 poll() 它将无法运行。它只是挂在活动监视器中而不运行（我在 Mac 上）。我以为我可以使用观察者线程，但我挂在 q_out.get() 上，这让我相信也许我正在填满缓冲区并陷入僵局。我不知道如何解决这个问题。这基本上就是我的代码的样子。如果有人对如何做到这一点有任何更好的想法，我很乐意完全改变我的方法。

def watchJob(p1,out_q):
    while p1.poll() == None:
        pass
    print "Job is done"
    out_q.put(p1.returncode)

def runJob(out_q):
    LOGFILE = open('job_to_run.log','w')
    p1 = Popen(['../../bin/jobexe','job_to_run'], stdout = LOGFILE)
    t = threading.Thread(target=watchJob, args=(p1,out_q))
    t.start()

out_q= Queue()
outlst=[]
for i in range(len(nprocs)):
    proc = Process(target=runJob, args=(out_q,))
    proc.start()
    outlst.append(out_q.get()) # This hangs indefinitely
    proc.join()

Answer 1

如果您要在线程中运行watchJob，则没有理由使用p1.poll 进行忙循环；只需调用p1.wait() 阻止直到进程完成。使用忙循环需要不断地释放/重新获取 GIL，这会减慢主线程的速度，并且还会占用 CPU，从而更加损害性能。

另外，如果你没有使用子进程的stdout，你不应该把它发送到PIPE，因为如果进程向stdout缓冲区写入足够多的数据，这可能会导致死锁。填满它（这实际上可能是您的情况）。这里也不需要使用multiprocessing；只需在主线程中调用Popen，然后让watchJob 线程等待进程完成。

import threading
from subprocess import Popen
from Queue import Queue

def watchJob(p1, out_q):
    p1.wait()
    out_q.put(p1.returncode)

out_q = Queue()
outlst=[]
p1 = Popen(['../../bin/jobexe','job_to_run'])
t = threading.Thread(target=watchJob, args=(p1,out_q))
t.start()
outlst.append(out_q.get())
t.join()

编辑：

以下是如何以这种方式同时运行多个作业：

out_q = Queue()
outlst = []
threads = []
num_jobs = 3
for _ in range(num_jobs):
    p = Popen(['../../bin/jobexe','job_to_run'])
    t = threading.Thread(target=watchJob, args=(p1, out_q))
    t.start()
    # Don't consume from the queue yet.

# All jobs are running, so now we can start
# consuming results from the queue.
for _ in range(num_jobs):
    outlst.append(out_q.get())
    t.join()

Answer 2

这里不需要多处理也不需要线程。你可以并行运行多个子进程，并在一个线程中收集它们的所有规则：

#!/usr/bin/env python3
from subprocess import Popen

def run(cmd, log_filename):
    with open(log_filename, 'wb', 0) as logfile:
        return Popen(cmd, stdout=logfile)

# start several subprocesses
processes = {run(['echo', c], 'subprocess.%s.log' % c) for c in 'abc'}
# now they all run in parallel
# report as soon as a child process exits
while processes: 
    for p in processes: 
        if p.poll() is not None:
           processes.remove(p) 
           print('{} done, status {}'.format(p.args, p.returncode))
           break

p.args 在 Python 3.3+ 中存储 cmd，在早期的 Python 版本中自己跟踪 cmd。

另见：

• Python threading multiple bash subprocesses?
• Python subprocess in parallel
• Python: execute cat subprocess in parallel
• Using Python's Multiprocessing module to execute simultaneous and separate SEAWAT/MODFLOW model runs

为了限制并行作业的数量，可以使用线程池（如the first link 所示）：

#!/usr/bin/env python3
from multiprocessing.dummy import Pool # use threads
from subprocess import Popen

def run_until_done(args):
    cmd, log_filename = args
    try:
        with open(log_filename, 'wb', 0) as logfile:
            p = Popen(cmd, stdout=logfile)
        return cmd, p.wait(), None
    except Exception as e:
        return cmd, None, str(e)

commands = ((('echo', str(d)), 'subprocess.%03d.log' % d) for d in range(500))
pool = Pool(128) # 128 concurrent commands at a time
for cmd, status, error in pool.imap_unordered(run_until_done, commands):
    if error is None:
       fmt = '{cmd} done, status {status}'
    else:
       fmt = 'failed to run {cmd}, reason: {error}'
    print(fmt.format_map(vars())) # or fmt.format(**vars()) on older versions

示例中的线程池有 128 个线程（不多也不少）。它不能同时执行超过 128 个作业。一旦任何线程释放（完成一个作业），它就会占用另一个，等等。并发执行的作业总数受线程数的限制。新工作不会等待所有 128 个以前的工作完成。当 任何 个旧作业完成后，它就会启动。