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

Question

我的问题希望足够具体，不会与我读过的任何其他问题相关。我想使用子进程和多重处理来连续生成一堆作业并将返回代码返回给我。问题是我不想 wait() 这样我可以一次生成所有作业，但我确实想知道它何时完成，以便我可以获得返回代码。我遇到了这个奇怪的问题，如果我 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

这里既不需要多处理也不需要线程。您可以并行运行多个子进程并在单个线程中收集它们的法规:

#!/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 个先前作业完成。当任何旧作业完成时它就会启动。