Implement non-blocking read and PID checking

Author: kpaulisse

lib/octocatalog-diff/util/parallel.rb

@@ -11,6 +11,8 @@ module Util
     # If parallel processing has been disabled, this instead executes the tasks serially,
     # but provides the same API as the parallel tasks.
     class Parallel
+      BLOCK_SIZE = 1024 * 16
+
       # This class is called for a task that didn't complete.
       class IncompleteTask < RuntimeError; end
 
@@ -107,32 +109,55 @@ def self.run_tasks_parallel(result, task_array, logger)
           # :nocov:
           this_pid = fork do
             reader.close
-            logger.reopen
+            logger.reopen if logger.respond_to?(:reopen)
             task_result = execute_task(task, logger)
             writer.write YAML.dump(task_result)
+            writer.close
+            logger.close
             exit 0
           end
           # :nocov:
 
-          pidmap[this_pid] = { reader: reader, index: index, start_time: Time.now }
+          pidmap[this_pid] = { reader: reader, index: index, start_time: Time.now, result: [] }
           writer.close
           logger.debug "Launched pid=#{this_pid} for index=#{index}"
           logger.reopen
         end
 
         while pidmap.any?
-          this_pid, exit_obj = Process.wait2
-          next unless pidmap.key?(this_pid)
+          # Read from all pipes
+          pidmap.each do |_this_pid, obj|
+            begin
+              buf = obj[:reader].read_nonblock(BLOCK_SIZE, buf)
+              obj[:result] << buf if buf
+            rescue IO::EAGAINWaitReadable, EOFError, Errno::EAGAIN # rubocop:disable Lint/ShadowedException
+              next
+            end
+          end
+
+          # Any exits?
+          this_pid, exit_obj = Process.wait2(0, Process::WNOHANG)
+          next unless this_pid && pidmap.key?(this_pid)
 
+          index = pidmap[this_pid][:index]
           exitstatus = exit_obj.exitstatus
+          raise "PID=#{this_pid} exited abnormally: #{exit_obj.inspect}" if exitstatus.nil?
           raise "PID=#{this_pid} exited with status #{exitstatus}" unless exitstatus.zero?
-          logger.debug "PID=#{this_pid} completed task in #{Time.now - pidmap[this_pid][:start_time]} seconds"
 
-          index = pidmap[this_pid][:index]
-          result[index] = YAML.load(pidmap[this_pid][:reader].read)
-          pidmap[this_pid][:reader].close
+          begin
+            buf = pidmap[this_pid][:reader].read_nonblock(BLOCK_SIZE, buf)
+            pidmap[this_pid][:result] << buf if buf
+          rescue IO::EAGAINWaitReadable, EOFError, Errno::EAGAIN # rubocop:disable Lint/ShadowedException
+            pidmap[this_pid][:reader].close
+          end
+
+          input = pidmap[this_pid][:result].join('')
+          logger.debug "PID=#{this_pid} completed in #{Time.now - pidmap[this_pid][:start_time]} seconds, #{input.length} bytes"
+
           pidmap.delete(this_pid)
 
+          result[index] = YAML.load(input)
+
           next if result[index].status
           return result[index].exception
         end