Factor out IO-selection code, prepare for PR #2466

claui · claui · commit cae2fdbcda87 · 2017-06-13T12:53:37.000+02:00
This commit sets the stage for an upcoming fix for #2466. In a nutshell, this is what it does: - In `cask/system_command.rb`, factor out existing code (“IO selection”) that we’re going to need later in order to fix PR #2466; - move said code into its own class `Utils::IOSelector`; - split the class into smaller methods to make the code more expressive than before; and - add unit tests for `Utils::IOSelector` (they’re a bit bloated because edge cases.)
diff --git a/Library/Homebrew/cask/lib/hbc/system_command.rb b/Library/Homebrew/cask/lib/hbc/system_command.rb
@@ -81,31 +81,16 @@ def each_output_line(&b)
 
       write_input_to(raw_stdin)
       raw_stdin.close_write
-      each_line_from [raw_stdout, raw_stderr], &b
 
+      ::Utils::IOSelector
+        .each_line_from(stdout: raw_stdout, stderr: raw_stderr, &b)
       @processed_status = raw_wait_thr.value
     end
 
     def write_input_to(raw_stdin)
       [*options[:input]].each { |line| raw_stdin.print line }
     end
 
-    def each_line_from(sources)
-      loop do
-        readable_sources = IO.select(sources)[0]
-        readable_sources.delete_if(&:eof?).first(1).each do |source|
-          type = ((source == sources[0]) ? :stdout : :stderr)
-          begin
-            yield(type, source.readline_nonblock || "")
-          rescue IO::WaitReadable, EOFError
-            next
-          end
-        end
-        break if readable_sources.empty?
-      end
-      sources.each(&:close_read)
-    end
-
     def result
       Result.new(command,
                  processed_output[:stdout],
diff --git a/Library/Homebrew/test/spec_helper.rb b/Library/Homebrew/test/spec_helper.rb
@@ -3,6 +3,7 @@
 require "rspec/its"
 require "rspec/wait"
 require "set"
+require "English"
 
 if ENV["HOMEBREW_TESTS_COVERAGE"]
   require "simplecov"
diff --git a/Library/Homebrew/test/utils/io_selector_spec.rb b/Library/Homebrew/test/utils/io_selector_spec.rb
@@ -0,0 +1,176 @@
+require "utils/io_selector"
+
+describe Utils::IOSelector do
+  describe "given text streams" do
+    let(:first_pipe) { IO.pipe }
+    let(:second_pipe) { IO.pipe }
+
+    let(:first_reader) { first_pipe[0] }
+    let(:second_reader) { second_pipe[0] }
+
+    let(:first_writer) { first_pipe[1] }
+    let(:second_writer) { second_pipe[1] }
+
+    let(:queues) { { first: Queue.new, second: Queue.new } }
+
+    let(:write_first!) do
+      thread = Thread.new(first_writer, queues[:second]) do |io, queue|
+        io.puts "Lorem"
+        wait(1).for { queue.pop }.to end_with("\n")
+        io.puts "dolor"
+        io.close
+      end
+      thread.abort_on_exception = true
+      thread
+    end
+
+    let(:write_second!) do
+      thread = Thread.new(second_writer, queues[:first]) do |io, queue|
+        wait(1).for { queue.pop }.to end_with("\n")
+        io.puts "ipsum"
+        wait(1).for { queue.pop }.to end_with("\n")
+        io.puts "sit"
+        io.puts "amet"
+        io.close
+      end
+      thread.abort_on_exception = true
+      thread
+    end
+
+    let(:queue_feeder) do
+      lambda do |proc_under_test, *args, &block|
+        proc_under_test.call(*args) do |tag, string_received|
+          queues[tag] << string_received
+          block.call(tag, string_received)
+        end
+      end
+    end
+
+    before do
+      allow_any_instance_of(Utils::IOSelector)
+        .to receive(:each_line_nonblock)
+        .and_wrap_original do |*args, &block|
+        queue_feeder.call(*args, &block)
+      end
+
+      write_first!
+      write_second!
+    end
+
+    after do
+      write_first!.exit
+      write_second!.exit
+    end
+
+    describe "::each_line_from" do
+      subject do
+        line_hash = { first: "", second: "", full_text: "" }
+
+        Utils::IOSelector.each_line_from(
+          first: first_reader,
+          second: second_reader,
+        ) do |tag, string_received|
+          line_hash[tag] << string_received
+          line_hash[:full_text] << string_received
+        end
+        line_hash
+      end
+
+      before { wait(1).for(subject) }
+
+      its([:first]) { is_expected.to eq("Lorem\ndolor\n") }
+      its([:second]) { is_expected.to eq("ipsum\nsit\namet\n") }
+
+      its([:full_text]) {
+        is_expected.to eq("Lorem\nipsum\ndolor\nsit\namet\n")
+      }
+    end
+
+    describe "::new" do
+      let(:selector) do
+        Utils::IOSelector.new(
+          first: first_reader,
+          second: second_reader,
+        )
+      end
+      subject { selector }
+
+      describe "pre-read" do
+        its(:pending_streams) {
+          are_expected.to eq([first_reader, second_reader])
+        }
+
+        its(:separator) {
+          is_expected.to eq($INPUT_RECORD_SEPARATOR)
+        }
+      end
+
+      describe "post-read" do
+        before do
+          wait(1).for {
+            subject.each_line_nonblock {}
+            true
+          }.to be true
+        end
+
+        after { expect(selector.all_streams).to all be_closed }
+
+        its(:pending_streams) { are_expected.to be_empty }
+        its(:separator) {
+          is_expected.to eq($INPUT_RECORD_SEPARATOR)
+        }
+      end
+
+      describe "#each_line_nonblock" do
+        subject do
+          line_hash = { first: "", second: "", full_text: "" }
+          super().each_line_nonblock do |tag, string_received|
+            line_hash[tag] << string_received
+            line_hash[:full_text] << string_received
+          end
+          line_hash
+        end
+
+        before { wait(1).for(subject) }
+        after { expect(selector.all_streams).to all be_closed }
+
+        its([:first]) { is_expected.to eq("Lorem\ndolor\n") }
+        its([:second]) { is_expected.to eq("ipsum\nsit\namet\n") }
+
+        its([:full_text]) {
+          is_expected.to eq("Lorem\nipsum\ndolor\nsit\namet\n")
+        }
+      end
+
+      its(:all_streams) {
+        are_expected.to eq([first_reader, second_reader])
+      }
+
+      its(:all_tags) { are_expected.to eq([:first, :second]) }
+
+      describe "#tag_of" do
+        subject do
+          {
+            "first tag" => super().tag_of(first_reader),
+            "second tag" => super().tag_of(second_reader),
+          }
+        end
+
+        its(["first tag"]) { is_expected.to eq(:first) }
+        its(["second tag"]) { is_expected.to eq(:second) }
+      end
+    end
+
+    describe "::new with a custom separator" do
+      subject do
+        tagged_streams = {
+          first: first_reader,
+          second: second_reader,
+        }
+        Utils::IOSelector.new(tagged_streams, ",")
+      end
+
+      its(:separator) { is_expected.to eq(",") }
+    end
+  end
+end
diff --git a/Library/Homebrew/utils.rb b/Library/Homebrew/utils.rb
@@ -9,6 +9,7 @@
 require "utils/github"
 require "utils/hash"
 require "utils/inreplace"
+require "utils/io_selector"
 require "utils/link"
 require "utils/popen"
 require "utils/svn"
diff --git a/Library/Homebrew/utils/io_selector.rb b/Library/Homebrew/utils/io_selector.rb
@@ -0,0 +1,79 @@
+require "delegate"
+require "English"
+
+require "extend/io"
+
+module Utils
+  #
+  # The class `IOSelector` is a wrapper for `IO::select` with the
+  # added benefit that it spans the streams' lifetimes.
+  #
+  # The class accepts multiple IOs which must be open for reading.
+  # It then notifies the client as data becomes available
+  # per-stream.
+  #
+  # Its main use is to allow a client to read both `stdout` and
+  # `stderr` of a subprocess in a way that avoids buffer-related
+  # deadlocks.
+  #
+  # For a more in-depth explanation, see:
+  #   https://github.com/Homebrew/brew/pull/2466
+  #
+  class IOSelector < DelegateClass(Hash)
+    attr_reader :separator
+
+    alias all_streams keys
+    alias all_tags values
+    alias tag_of fetch
+
+    def self.each_line_from(streams = {},
+      separator = $INPUT_RECORD_SEPARATOR, &block)
+      new(streams, separator).each_line_nonblock(&block)
+    end
+
+    def initialize(streams = {},
+      separator = $INPUT_RECORD_SEPARATOR)
+      super(streams.invert.compare_by_identity)
+      @separator = separator
+    end
+
+    def each_line_nonblock
+      each_readable_stream_until_eof do |stream|
+        line = stream.readline_nonblock(separator) || ""
+        yield(tag_of(stream), line)
+      end
+      close_streams
+    end
+
+    def pending_streams
+      @pending_streams ||= all_streams.dup
+    end
+
+    private
+
+    def each_readable_stream_until_eof(&block)
+      loop do
+        readable_streams.each do |stream|
+          pending_streams.delete(stream) if stream.eof?
+          yield_gracefully(stream, &block)
+        end
+        break if pending_streams.empty?
+      end
+    end
+
+    def readable_streams
+      IO.select(pending_streams)[0]
+    end
+
+    def yield_gracefully(stream)
+      yield(stream)
+    rescue IO::WaitReadable, IO::WaitWritable, EOFError
+      # We'll be back until/unless EOF
+      return
+    end
+
+    def close_streams
+      all_streams.each(&:close_read)
+    end
+  end
+end
