如何使用 Python 将 stdin/stdout 传输到 Perl 脚本

Question

此 Python 代码通过 Perl 脚本很好地传输数据。

import subprocess
kw = {}
kw['executable'] = None
kw['shell'] = True
kw['stdin'] = None
kw['stdout'] = subprocess.PIPE
kw['stderr'] = subprocess.PIPE
args = ' '.join(['/usr/bin/perl','-w','/path/script.perl','<','/path/mydata'])
subproc = subprocess.Popen(args,**kw)
for line in iter(subproc.stdout.readline, ''):
    print line.rstrip().decode('UTF-8')

但是，它要求我首先将缓冲区保存到磁盘文件（/path/mydata）。在 Python 代码中循环数据并逐行传递到子进程会更干净，如下所示：

import subprocess
kw = {}
kw['executable'] = '/usr/bin/perl'
kw['shell'] = False
kw['stderr'] = subprocess.PIPE
kw['stdin'] = subprocess.PIPE
kw['stdout'] = subprocess.PIPE
args = ['-w','/path/script.perl',]
subproc = subprocess.Popen(args,**kw)
f = codecs.open('/path/mydata','r','UTF-8')
for line in f:
    subproc.stdin.write('%s\n'%(line.strip().encode('UTF-8')))
    print line.strip()  ### code hangs after printing this ###
    for line in iter(subproc.stdout.readline, ''):
        print line.rstrip().decode('UTF-8')
subproc.terminate()
f.close()

在将第一行发送到子进程后，代码会随着 readline 挂起。我还有其他可执行文件完美地使用了完全相同的代码。

我的数据文件可能非常大（1.5 GB） 有没有办法在不保存到文件的情况下完成管道传输数据？我不想为了与其他系统兼容而重新编写perl脚本。

Answer 1

您的代码在该行处阻塞：

for line in iter(subproc.stdout.readline, ''):

因为此迭代终止的唯一方法是到达 EOF（文件结束符），这将在子进程终止时发生。您不想等到进程终止，但是，您只想等到它完成处理发送给它的行。

此外，正如克里斯·摩根已经指出的那样，您遇到了缓冲问题。另一个stackoverflow 上的问题讨论了如何做到不- 使用子进程阻塞读取。我已经从该问题到您的问题对代码进行了快速而肮脏的修改：

def enqueue_output(out, queue):
    for line in iter(out.readline, ''):
        queue.put(line)
    out.close()

kw = {}
kw['executable'] = '/usr/bin/perl'
kw['shell'] = False
kw['stderr'] = subprocess.PIPE
kw['stdin'] = subprocess.PIPE
kw['stdout'] = subprocess.PIPE
args = ['-w','/path/script.perl',]
subproc = subprocess.Popen(args, **kw)
f = codecs.open('/path/mydata','r','UTF-8')
q = Queue.Queue()
t = threading.Thread(target = enqueue_output, args = (subproc.stdout, q))
t.daemon = True
t.start()
for line in f:
    subproc.stdin.write('%s\n'%(line.strip().encode('UTF-8')))
    print "Sent:", line.strip()  ### code hangs after printing this ###
    try:
        line = q.get_nowait()
    except Queue.Empty:
        pass
    else:
        print "Received:", line.rstrip().decode('UTF-8')

subproc.terminate()
f.close()

您很可能需要对此代码进行修改，但至少它不会阻塞。

Answer 2

谢谢斯格格。我也尝试过线程解决方案。然而，仅此解决方案总是挂起。我之前的代码和 srgerg 的代码都缺少最终的解决方案，你的提示给了我最后一个想法。

最终的解决方案写入足够的虚拟数据，迫使缓冲区中的最终有效行。为了支持这一点，我添加了跟踪写入 stdin 的有效行数的代码。线程循环打开输出文件，保存数据，并在读取行等于有效输入行时中断。该解决方案确保它可以逐行读取和写入任何大小的文件。

def std_output(stdout,outfile=''):
    out = 0
    f = codecs.open(outfile,'w','UTF-8')
    for line in iter(stdout.readline, ''):
        f.write('%s\n'%(line.rstrip().decode('UTF-8')))
        out += 1
        if i == out: break
    stdout.close()
    f.close()

outfile = '/path/myout'
infile = '/path/mydata'

subproc = subprocess.Popen(args,**kw)
t = threading.Thread(target=std_output,args=[subproc.stdout,outfile])
t.daemon = True
t.start()

i = 0
f = codecs.open(infile,'r','UTF-8')
for line in f:
    subproc.stdin.write('%s\n'%(line.strip().encode('UTF-8')))
    i += 1
subproc.stdin.write('%s\n'%(' '*4096)) ### push dummy data ###
f.close()
t.join()
subproc.terminate()

Answer 3

请参阅手册中提到的有关使用 Popen.stdin 和 Popen.stdout 的警告（就在 Popen.stdin）：

警告：使用communicate()而不是.stdin.write 、 .stdout.read 或 .stderr.read 以避免由于任何其他操作系统管道缓冲区而导致死锁填充并阻塞子进程。

我意识到一次在内存中拥有一个半千兆字节的字符串并不是很理想，但是使用 communicate() 是一种将工作的方式，正如您所观察到的，一旦操作系统管道缓冲区填满上, 的stdin.write() + stdout.read() 方式可能会陷入僵局。

使用 communicate() 对您来说可行吗？