我正在使用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