From: Michal Suchanek <hramrach@...>
Date: 2011-08-09T21:57:28+09:00
Subject: problem running pipeline

Hello,

I tried to write a run_pipeline method that would return the exit
status (and stderr) of every process in the pipe.

I am running ruby 1.8.7 (2011-06-30 patchlevel 352) [x86_64-linux]

Inserting a method call by patching in

  somename args
end
def somename args

produces lockup.

I guess I could construct the pipeline in a single method but why
would inserting a method call break otherwise working code is a
mystery.

Any insights welcome.

Thanks

Michal

Code (incomplete but working):

$ cat pipeline.rb
READSIZE = 1024 * 32
def gather_out io, str
    begin
        while true do
            str << (io.readpartial READSIZE)
        end
    rescue EOFError
    end
    str
end

def waiton things
    things.collect{|thing|
        if thing.is_a? Thread then
            thing.join
        else
            Process::waitpid thing
            $?
        end
    }
end

def fork_process cmd, inp_chld, out_chld, err_chld, odd_ends
    pid = fork {
        odd_ends.each{|fd| fd.close}
        STDIN.reopen inp_chld
        STDOUT.reopen out_chld
        STDERR.reopen err_chld
        exec cmd
    }
    inp_chld.close
    out_chld.close
    err_chld.close
    pid
end


def run_pipeline_w cmds
    inp_chld, inp = IO.pipe
    out, out_chld = IO.pipe
    err, err_chld = IO.pipe
    cmd, *cmds = cmds
    if cmds.length == 0 then
        outstr = ""
        errstr = ""
        pid = fork_process cmd, inp_chld, out_chld, err_chld, [inp, out, err]
        threads = [ Thread.new { yield inp ; inp.close},
        Thread.new { gather_out out, outstr },
        Thread.new { gather_out err, errstr } ]
        ret, *_ = waiton [pid] + threads
    else
        raise "FIXME"
    end
    [outstr,errstr,ret]
end

$ cat tc_pipeline.rb
require 'pipeline'
require 'test/unit'
require 'stringio'

class TestPipeline < Test::Unit::TestCase
    def setup
        # none
        @data_lines = 1000
        @data_text = "%i some data"
        @inp = StringIO.new
        output_data @inp
        @inp.rewind
        @inp = @inp.read
    end
    def output_data io
        (1..@data_lines).each {|i| io.puts(@data_text % i) }
    end
    def teardown
        # none
    end

    def test_cat
        out, err, ret = run_pipeline_w( ["cat"] ){|io| output_data io
rescue nil}
        assert_equal ret, 0
        assert_equal @inp, out
        assert_equal err, ""
    end

    def test_doublecat
        out, err1, ret1, err2, ret2 = run_pipeline_w( ["cat","cat"]
){|io| output_data io rescue nil}
        assert_equal ret1, ret2, 0
        assert_equal @inp, out
        assert_equal err1, err2, ""
    end
end

Patch (in my eyes noop but breaking):
diff --git a/pipeline.rb b/pipeline.rb
index 73cb91b..9ef7630 100644
--- a/pipeline.rb
+++ b/pipeline.rb
@@ -38,6 +38,9 @@ end
 def run_pipeline_w cmds
     inp_chld, inp = IO.pipe
     out, out_chld = IO.pipe
+    run_pipeline_w_ios cmds, inp, out, inp_chld, out_chld, [inp, out], []
+end
+def run_pipeline_w_ios cmds, inp, out, inp_chld, out_chld, odd_ends,
threads = []
     err, err_chld = IO.pipe
     cmd, *cmds = cmds
     if cmds.length == 0 then