使用subprocess.Popen将大量数据传递给stdin

Question

我有点难以理解解决这个简单问题的python方法是什么.

我的问题很简单.如果您使用以下代码,它将挂起.这在子流程模块doc中有详细记载.

import subprocess

proc = subprocess.Popen(['cat','-'],
                        stdin=subprocess.PIPE,
                        stdout=subprocess.PIPE,
                        )
for i in range(100000):
    proc.stdin.write('%d\n' % i)
output = proc.communicate()[0]
print output

寻找一个解决方案(有一个非常有洞察力的线程,但我现在已经丢失了)我发现这个解决方案(以及其他)使用了一个显式的fork:

import os
import sys
from subprocess import Popen, PIPE

def produce(to_sed):
    for i in range(100000):
        to_sed.write("%d\n" % i)
        to_sed.flush()
    #this would happen implicitly, anyway, but is here for the example
    to_sed.close()

def consume(from_sed):
    while 1:
        res = from_sed.readline()
        if not res:
            sys.exit(0)
            #sys.exit(proc.poll())
        print 'received: ', [res]

def main():
    proc = Popen(['cat','-'],stdin=PIPE,stdout=PIPE)
    to_sed = proc.stdin
    from_sed = proc.stdout

    pid = os.fork()
    if pid == 0 :
        from_sed.close()
        produce(to_sed)
        return
    else :
        to_sed.close()
        consume(from_sed)

if __name__ == '__main__':
    main()

虽然这种解决方案在概念上很容易理解,它使用了一个更加过程并坚持太低的水平相比,子模块(即存在只是为了掩盖这样的事情...).

我想知道:是否有一个简单而干净的解决方案使用不会挂起的子进程模块或实现这种模式我必须退后一步并实现旧式选择循环或显式分叉？

谢谢

Answer 1

如果您需要纯Python解决方案,则需要将读取器或编写器放在单独的线程中.该threading软件包是一种轻量级的方法,可以方便地访问常见对象并且不会出现混乱的分叉.

import subprocess
import threading
import sys

proc = subprocess.Popen(['cat','-'],
                        stdin=subprocess.PIPE,
                        stdout=subprocess.PIPE,
                        )
def writer():
    for i in range(100000):
        proc.stdin.write('%d\n' % i)
    proc.stdin.close()
thread = threading.Thread(target=writer)
thread.start()
for line in proc.stdout:
    sys.stdout.write(line)
thread.join()
proc.wait()

看到subprocess模块现代化以支持流和协同程序可能是很好的,这将允许混合Python部件和shell部件的管道构造得更优雅.

Answer 2

如果您不想将所有数据保留在内存中,则必须使用select.例如:

import subprocess
from select import select
import os

proc = subprocess.Popen(['cat'], stdin=subprocess.PIPE, stdout=subprocess.PIPE)

i = 0;
while True:
    rlist, wlist, xlist = [proc.stdout], [], []
    if i < 100000:
        wlist.append(proc.stdin)
    rlist, wlist, xlist = select(rlist, wlist, xlist)
    if proc.stdout in rlist:
        out = os.read(proc.stdout.fileno(), 10)
        print out,
        if not out:
            break
    if proc.stdin in wlist:
        proc.stdin.write('%d\n' % i)
        i += 1
        if i >= 100000:
            proc.stdin.close()