Use IO#readline_nonblock instead of IO#readline when reading output f…

…rom CPI sub-processes.

IO.#readline, being a blocking method, can cause a dealock when the cpi
sends output without a line break. We had such issues with azure cpi.

src/bosh-director/lib/cloud/external_cpi.rb

@@ -1,6 +1,28 @@
 require 'membrane'
 require 'open3'
 
+# Robbed from:
+# https://github.com/Homebrew/brew/blob/master/Library/Homebrew/extend/io.rb
+class IO
+  def readline_nonblock(sep = $INPUT_RECORD_SEPARATOR)
+    line = +""
+    buffer = +""
+
+    loop do
+      break if buffer == sep
+
+      read_nonblock(1, buffer)
+      line.concat(buffer)
+    end
+
+    line.freeze
+  rescue IO::WaitReadable, EOFError => e
+    raise e if line.empty?
+
+    line.freeze
+  end
+end
+
 module Bosh::Clouds
   class ExternalCpi
     # Raised when the external CPI executable returns an error unknown to director
@@ -110,7 +132,7 @@ def invoke_cpi_method(method_name, *arguments)
 
           readable = ready[0]
           readable.each do |f|
-            line = f.readline
+            line = f.readline_nonblock
             case f.fileno
             when stdout.fileno
               cpi_response.write(line)

src/bosh-director/spec/assets/bin/dummy_cpi

@@ -0,0 +1,13 @@
+#!/usr/bin/env ruby
+
+# Dummy cpi to test for a dealock scenario when a cpi sends an inclomplete line (missing line break) to STDOUT
+
+# Force the reading end to see incomplete line so that it blocks on 'readline' method
+# by $stdout.sync = true
+$stdout.sync = true
+$stdout.print('{"result":"OK","log":"LogMessage","error":null}')
+
+sleep 1
+
+# send a massive amount of output to fill the pipe buffer (64 KB) so that this script blocks on the next line
+$stderr.print('abcdefghijklmnopqrstuwxyz' * 100000)

src/bosh-director/spec/unit/cloud/external_cpi_response_wrapper_spec.rb

@@ -30,12 +30,12 @@
     allow(stdout).to receive(:fileno).and_return(1)
 
     stdout_reponse_values = [cpi_response, nil, cpi_response, nil, cpi_response, nil]
-    allow(stdout).to receive(:readline) { stdout_reponse_values.shift || raise(EOFError) }
+    allow(stdout).to receive(:readline_nonblock) { stdout_reponse_values.shift || raise(EOFError) }
 
     allow(stderr).to receive(:fileno).and_return(2)
 
     stderr_reponse_values = [cpi_error, nil, cpi_error, nil, cpi_error, nil]
-    allow(stderr).to receive(:readline) { stderr_reponse_values.shift || raise(EOFError) }
+    allow(stderr).to receive(:readline_nonblock) { stderr_reponse_values.shift || raise(EOFError) }
   end
 
   before(:each) do

src/bosh-director/spec/unit/cloud/external_cpi_spec.rb

@@ -38,12 +38,12 @@ def self.it_calls_cpi_method(method, *arguments, api_version: 1)
       allow(stdout).to receive(:fileno).and_return(1)
 
       stdout_reponse_values = [cpi_response, nil, cpi_response, nil, cpi_response, nil]
-      allow(stdout).to receive(:readline) { stdout_reponse_values.shift || raise(EOFError) }
+      allow(stdout).to receive(:readline_nonblock) { stdout_reponse_values.shift || raise(EOFError) }
 
       allow(stderr).to receive(:fileno).and_return(2)
 
       stderr_reponse_values = [cpi_error, nil, cpi_error, nil, cpi_error, nil]
-      allow(stderr).to receive(:readline) { stderr_reponse_values.shift || raise(EOFError) }
+      allow(stderr).to receive(:readline_nonblock) { stderr_reponse_values.shift || raise(EOFError) }
 
       allow(Open3).to receive(:popen3).and_yield(stdin, stdout, stderr, wait_thread)
 
@@ -644,4 +644,20 @@ def self.it_raises_an_error_with_ok_to_retry(error_class)
   describe '#info' do
     it_calls_cpi_method(:info)
   end
+
+  context 'Fix for a dealock scenario when a cpi sub-process sends an inclomplete line (missing "\n") to STDOUT' do
+    let(:logger) { Logging::Logger.new('ExternalCpi') }
+    let(:cpi_log_path) { '/var/vcap/task/5/cpi' }
+    let(:config) { double('Bosh::Director::Config', logger: logger, cpi_task_log: cpi_log_path) }
+    let(:external_cpi) { described_class.new(asset("bin/dummy_cpi"), 'fake-director-uuid', logger) }
+
+    before { stub_const('Bosh::Director::Config', config) }
+    before { FileUtils.mkdir_p('/var/vcap/task/5') }
+    before { allow(File).to receive(:executable?).with(asset("bin/dummy_cpi")).and_return(true) }
+
+    it 'does not deadlock' do
+      result = external_cpi.info
+      expect(result).to eq 'OK'
+    end
+  end
 end