Python 子进程挂起命名管道

Question

我挣扎试图模拟这个简单的bash:

$ cat /tmp/fifo.tub &                                                            
[1] 24027
$ gunzip -c /tmp/filedat.dat.gz > /tmp/fifo.tub                                                              
line 01
line 02
line 03
line 04
line 05
line 06
line 07
line 08
line 09
line 10
[1]+  Done                    cat /tmp/fifo.tub

基本上我尝试了这种subprocess方法:

# -*- coding: utf-8 -*-

import os
import sys
import shlex
import pprint
import subprocess

def main():

    fifo = '/tmp/fifo.tub'
    filedat = '/tmp/filedat.dat.gz '
    os.mkfifo(fifo,0777)
    cat  = "cat %s" % fifo
    args_cat = shlex.split(cat)
    pprint.pprint(args_cat)

    cat = subprocess.Popen( args_cat,
                            close_fds=True,
                            preexec_fn=os.setsid)

    print "PID cat: %s" % cat.pid

    f = os.open(fifo ,os.O_WRONLY)

    gunzip = 'gunzip -c  %s' %  (filedat)
    args_gunzip = shlex.split(gunzip)
    pprint.pprint(args_gunzip)

    gunzip = subprocess.Popen( args_gunzip,
                               stdout = f,
                               close_fds=True,
                               preexec_fn=os.setsid)

    print "PID gunzip: %s" % gunzip.pid

    while not cat.poll():
        # hangs for ever
        pass
    return True

if __name__=="__main__":
    main()

cat 进程永不结束。

或者，我尝试用线程绕过这个问题，但我得到了相同的结果。

import os
import sys
import shlex
import pprint
import subprocess
import threading

class Th(threading.Thread):
    def __init__(self,cmd,stdout_h=None):
        self.stdout = None
        self.stderr = None
        self.cmd = cmd
        self.stdout_h = stdout_h

        self.proceso = None
        self.pid = None

        threading.Thread.__init__(self)

    def run(self):
        if self.stdout_h:
            self.proceso = subprocess.Popen(self.cmd,
                                 shell=False,
                                 close_fds=True,
                                 stdout=self.stdout_h)

        else:
            self.proceso = subprocess.Popen( self.cmd,
                                  close_fds=True,
                                  shell=False)
        print "PID: %d" % self.proceso.pid   


def main():

    fifo = '/tmp/fifo.tub'
    filedat = '/tmp/filedat.dat.gz '

    try:
        os.unlink(fifo)
    except:
        pass
    try:
       os.mkfifo(fifo,0777)
    except Exception , err:
       print "Error '%s' tub %s." % (err,fifo)
       sys.exit(5)

    cat  = "cat %s" % fifo
    args_cat = shlex.split(cat)
    pprint.pprint(args_cat)

    cat = Th(cmd=args_cat)
    cat.start()

    try:
        f = os.open(fifo ,os.O_WRONLY)
    except Exception, err:
        print  "Error '%s' when open fifo %s " % (err,fifo)
        sys.exit(5)

    gunzip = 'gunzip -c  %s ' %  (filedat)
    args_gunzip = shlex.split(gunzip)
    pprint.pprint(args_gunzip)

    gunzip = Th(cmd=args_gunzip,stdout_h=f)
    gunzip.start()
    gunzip.join()
    cat.join()

    while  gunzip.proceso.poll() is None:
        pass

    if  cat.proceso.poll() is None:
        print "Why?"
        cat.proceso.terminate() 
    return True

if __name__=="__main__":
    main()

我显然遗漏了一些东西，非常欢迎任何帮助。

Answer 1

您没有关闭 FIFO 文件描述符，所以 cat 只是卡在那里，认为还有更多内容。

我认为您也可以使用 .wait() 方法来执行与 while 循环相同的操作。

# -*- coding: utf-8 -*-

import os
import sys
import shlex
import pprint
import subprocess

def main():

    fifo = '/tmp/fifo.tub'
    filedat = '/tmp/filedat.dat.gz '
    os.mkfifo(fifo,0777)
    cat  = "cat %s" % fifo
    args_cat = shlex.split(cat)
    pprint.pprint(args_cat)

    cat = subprocess.Popen( args_cat,
                            close_fds=True,
                            preexec_fn=os.setsid)

    print "PID cat: %s" % cat.pid

    f = os.open(fifo ,os.O_WRONLY)

    gunzip = 'gunzip -c  %s' %  (filedat)
    args_gunzip = shlex.split(gunzip)
    pprint.pprint(args_gunzip)

    gunzip = subprocess.Popen( args_gunzip,
                               stdout = f,
                               close_fds=True,
                               preexec_fn=os.setsid)

    print "PID gunzip: %s" % gunzip.pid

    gunzip.wait()
    print "gunzip finished"
    os.close(f)
    cat.wait()
    print "cat finished"

    return True

if __name__=="__main__":
    main()