我正在使用Python测试子进程流水线。我知道我可以直接在Python中完成以下程序所做的事情，但这不是重点。我只想测试一下流水线，以便了解如何使用它。
我的系统是Linux Ubuntu 9.04，默认使用Python 2.6。
我从这个文档示例开始。
from subprocess import Popen, PIPE
p1 = Popen(["grep", "-v", "not"], stdout=PIPE)
p2 = Popen(["cut", "-c", "1-10"], stdin=p1.stdout, stdout=PIPE)
output = p2.communicate()[0]
print output
那样做是可行的，但由于的标准输入未被重定向，我必须在终端中键入内容以供管道使用。当我键入^D关闭标准输入时，我得到了想要的输出。
然而，我想使用Python字符串变量将数据发送到管道。首先，我尝试在标准输入上写入：
p1 = Popen(["grep", "-v", "not"], stdin=PIPE, stdout=PIPE)
p2 = Popen(["cut", "-c", "1-10"], stdin=p1.stdout, stdout=PIPE)
p1.stdin.write('test\n')
output = p2.communicate()[0] # blocks forever here
没起作用。我尝试在上一行使用p2.stdout.read()，但它也会阻塞。我添加了p1.stdin.flush()和p1.stdin.close()，但它们也没有起作用。然后我转向使用通信：
p1 = Popen(["grep", "-v", "not"], stdin=PIPE, stdout=PIPE)
p2 = Popen(["cut", "-c", "1-10"], stdin=p1.stdout, stdout=PIPE)
p1.communicate('test\n') # blocks forever here
output = p2.communicate()[0]
所以这还不是它。
我注意到运行单个进程（如上面的
p1，去掉p2）可以完美地工作。并且将文件句柄传递给p1（stdin=open(...)）也可以工作。所以问题是：在Python中是否有可能向由2个或更多个子进程组成的管道传递数据，而不会阻塞？为什么不行？
我知道我可以运行一个shell并在shell中运行管道，但那不是我想要的。
更新 1：在遵循Aaron Digulla的提示之后，我现在正在尝试使用线程来使其工作。
首先，我尝试在一个线程上运行p1.communicate。
p1 = Popen(["grep", "-v", "not"], stdin=PIPE, stdout=PIPE)
p2 = Popen(["cut", "-c", "1-10"], stdin=p1.stdout, stdout=PIPE)
t = threading.Thread(target=p1.communicate, args=('some data\n',))
t.start()
output = p2.communicate()[0] # blocks forever here
好的，没有起作用。尝试了其他组合，比如将其更改为.write()和p2.read()。仍然不行。现在让我们尝试相反的方法：
def get_output(subp):
output = subp.communicate()[0] # blocks on thread
print 'GOT:', output
p1 = Popen(["grep", "-v", "not"], stdin=PIPE, stdout=PIPE)
p2 = Popen(["cut", "-c", "1-10"], stdin=p1.stdout, stdout=PIPE)
t = threading.Thread(target=get_output, args=(p2,))
t.start()
p1.communicate('data\n') # blocks here.
t.join()
代码最终会在某个地方阻塞。可能是在生成的线程中，也可能在主线程中，或者两者都存在。所以它没能正常工作。如果您知道如何使其工作，提供工作代码将更加容易。我正在尝试。
更新 2
保罗·杜博伊斯（Paul Du Bois）在下面给出了一些信息，因此我进行了更多测试。
我阅读了整个 subprocess.py 模块并了解了其工作原理。所以我尝试将其应用于代码。
我在Linux上进行测试，但由于我正在测试线程，因此我的第一个方法是复制subprocess.py的communicate()方法中看到的确切的Windows线程代码，但是针对两个进程而不是一个。以下是我尝试过的所有清单：
import os
from subprocess import Popen, PIPE
import threading
def get_output(fobj, buffer):
while True:
chunk = fobj.read() # BLOCKS HERE
if not chunk:
break
buffer.append(chunk)
p1 = Popen(["grep", "-v", "not"], stdin=PIPE, stdout=PIPE)
p2 = Popen(["cut", "-c", "1-10"], stdin=p1.stdout, stdout=PIPE)
b = [] # create a buffer
t = threading.Thread(target=get_output, args=(p2.stdout, b))
t.start() # start reading thread
for x in xrange(100000):
p1.stdin.write('hello world\n') # write data
p1.stdin.flush()
p1.stdin.close() # close input...
t.join()
好吧，它没起作用。即使调用了 p1.stdin.close()，p2.stdout.read() 仍然会阻塞。
然后我尝试了 subprocess.py 上的 posix 代码：
import os
from subprocess import Popen, PIPE
import select
p1 = Popen(["grep", "-v", "not"], stdin=PIPE, stdout=PIPE)
p2 = Popen(["cut", "-c", "1-10"], stdin=p1.stdout, stdout=PIPE)
numwrites = 100000
to_read = [p2.stdout]
to_write = [p1.stdin]
b = [] # create buffer
while to_read or to_write:
read_now, write_now, xlist = select.select(to_read, to_write, [])
if read_now:
data = os.read(p2.stdout.fileno(), 1024)
if not data:
p2.stdout.close()
to_read = []
else:
b.append(data)
if write_now:
if numwrites > 0:
numwrites -= 1
p1.stdin.write('hello world!\n'); p1.stdin.flush()
else:
p1.stdin.close()
to_write = []
print b
还会阻塞
select.select()。通过分散print，我发现了以下内容：
- 读取是有效的。在执行期间代码多次读取。
- 写入也有效。数据被写入
p1.stdin。 - 在
numwrites结束时，调用p1.stdin.close()。 - 当
select()开始阻塞时，只有to_read有内容，即p2.stdout。此时to_write已经为空。 os.read()调用总是返回一些内容，因此从未调用p2.stdout.close()。
从两个测试中得出的结论：关闭管道中第一个进程(grep在这个例子中)的stdin并不能使它将其缓冲输出转储到下一个进程并死亡。
没有办法使它工作吗？
PS: 我不想使用临时文件，我已经测试过文件并知道它可以工作。而且我不想使用Windows。
grep的标准输入必须使其输出。如果这种情况没有发生，那么就真的出了很大的问题。 - Aaron Digulla