Use our own parallel implementation

Author: kpaulisse

lib/octocatalog-diff/util/parallel.rb

@@ -1,16 +1,19 @@
 # frozen_string_literal: true
 
-# Helper to use the 'parallel' gem to perform tasks
+# Parallelize process executation.
 
-require 'parallel'
 require 'stringio'
+require 'yaml'
 
 module OctocatalogDiff
   module Util
     # This is a utility class to execute tasks in parallel, using the 'parallel' gem.
     # If parallel processing has been disabled, this instead executes the tasks serially,
     # but provides the same API as the parallel tasks.
     class Parallel
+      # This class is called for a task that didn't complete.
+      class IncompleteTask < RuntimeError; end
+
       # This class represents a parallel task. It requires a method reference, which will be executed with
       # any supplied arguments. It can optionally take a text description and a validator function.
       class Task
@@ -60,7 +63,7 @@ def initialize(opts = {})
       #
       # Note: Parallelization throws intermittent errors under travis CI, so it will be disabled by
       # default for integration tests.
-      def self.run_tasks(task_array, logger = nil, parallelized = !ENV.key?('OCTOCATALOG_DIFF_TRAVIS_CI_DISABLE_PARALLEL'))
+      def self.run_tasks(task_array, logger = nil, parallelized = true)
         # Create a throwaway logger object if one is not given
         logger ||= Logger.new(StringIO.new)
 
@@ -75,7 +78,7 @@ def self.run_tasks(task_array, logger = nil, parallelized = !ENV.key?('OCTOCATAL
         end
 
         result = task_array.map do |x|
-          Result.new(exception: ::Parallel::Kill.new('Killed'), args: x.args)
+          Result.new(exception: IncompleteTask.new('Killed'), args: x.args)
         end
         logger.debug "Initialized parallel task result array: size=#{result.size}"
 
@@ -94,42 +97,45 @@ def self.run_tasks(task_array, logger = nil, parallelized = !ENV.key?('OCTOCATAL
       # @param task_array [Array<OctocatalogDiff::Util::Parallel::Task>] Tasks to perform
       # @param logger [Logger] Logger
       def self.run_tasks_parallel(result, task_array, logger)
-        opts = {
-          isolation: true,
-          finish: lambda do |item, i, parallel_result|
-            # Set the result array element to the result
-            result[i] = parallel_result
-
-            # Kill all other parallel tasks if this task failed by throwing an exception
-            raise ::Parallel::Kill unless parallel_result.exception.nil?
-
-            # Run the validator to determine if the result is in fact valid. The validator
-            # returns true or false. If true, set the 'valid' attribute in the result. If
-            # false, kill all other parallel tasks.
-            if item.validate(parallel_result.output, logger)
-              logger.debug("Success #{item.description}")
-            else
-              logger.warn("Failed #{item.description}")
-              result[i].status = false
-              raise ::Parallel::Kill
-            end
+        pidmap = {}
+
+        task_array.each_with_index do |task, index|
+          reader, writer = IO.pipe
+          this_pid = fork do
+            reader.close
+            logger.reopen
+            task_result = execute_task(task, logger)
+            writer.write YAML.dump(task_result)
+            exit 0
           end
-        }
+          pidmap[this_pid] = { reader: reader, index: index, start_time: Time.now }
+          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)
 
-        ::Parallel.each(task_array, opts) do |ele|
-          # simplecov does not detect that this code runs because it's forked, but this is
-          # tested extensively in the parallel_spec.rb spec file.
-          # :nocov:
+          exitstatus = exit_obj.exitstatus
+          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
+          pidmap.delete(this_pid)
+          break unless result[index].status
+        end
+      ensure
+        pidmap.each do |pid, pid_data|
+          pid_data[:reader].close
           begin
-            logger.debug("Begin #{ele.description}")
-            output = ele.execute(logger)
-            logger.debug("Success #{ele.description}")
-            Result.new(output: output, status: true, args: ele.args)
-          rescue => exc
-            logger.debug("Failed #{ele.description}: #{exc.class} #{exc.message}")
-            Result.new(exception: exc, status: false, args: ele.args)
+            Process.kill('TERM', pid)
+          rescue Errno::ESRCH # rubocop:disable Lint/HandleExceptions
+            # If the process doesn't exist, that's fine.
           end
-          # :nocov:
         end
       end
 

octocatalog-diff.gemspec

@@ -26,7 +26,6 @@ EOF
   s.add_runtime_dependency 'diffy', '>= 3.1.0'
   s.add_runtime_dependency 'httparty', '>= 0.11.0'
   s.add_runtime_dependency 'hashdiff', '>= 0.3.0'
-  s.add_runtime_dependency 'parallel', '>= 1.11.1'
   s.add_runtime_dependency 'rugged', '>= 0.25.0b2'
 
   s.add_development_dependency 'rspec', '~> 3.4.0'

spec/octocatalog-diff/integration/integration_helper.rb

@@ -115,9 +115,6 @@ def self.integration(options = {})
       options[:parallel] = false if ENV['COVERAGE']
       options[:INTEGRATION] = true
 
-      # Disable parallel under travis CI
-      ENV['OCTOCATALOG_DIFF_TRAVIS_CI_DISABLE_PARALLEL'] = 'true' if ENV['TRAVIS']
-
       # Run octocatalog-diff CLI method. Capture stdout and stderr using 'strio'.
       logger, logger_string = OctocatalogDiff::Spec.setup_logger
       begin

spec/octocatalog-diff/tests/cli_spec.rb

@@ -80,7 +80,8 @@
           '--to-catalog', OctocatalogDiff::Spec.fixture_path('catalogs/tiny-catalog-2.json'),
           '--fact-file', OctocatalogDiff::Spec.fixture_path('facts/valid-facts.yaml'),
           '--cached-master-dir', @cached_master_directory,
-          '-o', File.join(@cached_master_directory, 'trashfile.txt')
+          '-o', File.join(@cached_master_directory, 'trashfile.txt'),
+          '--no-parallel'
         ]
         opts = {
           debug: true,

spec/octocatalog-diff/tests/util/parallel_spec.rb

@@ -59,6 +59,9 @@ def two(_arg, _logger = nil)
             break if File.file?(File.join($octocatalog_diff_util_parallel_spec_tempdir, 'one'))
             sleep 0.1
           end
+          # Sometimes the system will still handle the second process if it's near-simultaneous
+          # so sleep for a bit before exiting.
+          sleep 0.5
           raise 'Two failed'
         end
       end
@@ -106,7 +109,7 @@ def two(_arg, _logger = nil)
       one_result = result[0]
       expect(one_result).to be_a_kind_of(OctocatalogDiff::Util::Parallel::Result)
       expect(one_result.status).to eq(nil)
-      expect(one_result.exception).to be_a_kind_of(::Parallel::Kill)
+      expect(one_result.exception).to be_a_kind_of(OctocatalogDiff::Util::Parallel::IncompleteTask)
       expect(one_result.exception.message).to eq('Killed')
       expect(one_result.output).to eq(nil)
 
@@ -345,7 +348,7 @@ def two(arg, _logger = nil)
       two_result = result[1]
       expect(two_result).to be_a_kind_of(OctocatalogDiff::Util::Parallel::Result)
       expect(two_result.status).to eq(nil)
-      expect(two_result.exception).to be_a_kind_of(::Parallel::Kill)
+      expect(two_result.exception).to be_a_kind_of(OctocatalogDiff::Util::Parallel::IncompleteTask)
       expect(two_result.exception.message).to eq('Killed')
     end