Merge pull request #1425 from mbj/change/to-native-parallel

Change spec suite to native parallel

Gemfile.lock

@@ -70,7 +70,6 @@ PLATFORMS
 DEPENDENCIES
   mutant!
   mutant-license!
-  parallel (~> 1.3)
   rspec (~> 3.10)
   rspec-core (~> 3.10)
   rspec-its (~> 1.3.0)

lib/mutant/parallel.rb

@@ -109,7 +109,7 @@ class Config
     end # Config
 
     class Response
-      include Anima.new(:error, :log, :result)
+      include Anima.new(:error, :job, :log, :result)
     end
 
     # Parallel execution status

lib/mutant/parallel/connection.rb

@@ -41,12 +41,13 @@ def initialize(*)
           @results = []
         end
 
-        def self.read_response(**attributes)
+        def self.read_response(job:, **attributes)
           reader = new(**attributes).read_till_final
 
           Response.new(
-            log:    reader.log,
             error:  reader.error,
+            job:    job,
+            log:    reader.log,
             result: reader.result
           )
         end
@@ -80,7 +81,7 @@ def read_till_final
       private
 
         def timeout
-          @errors << Timeout
+          @errors << Timeout::Error
         end
 
         def advance_result

lib/mutant/parallel/worker.rb

@@ -98,6 +98,7 @@ def call
           response = Connection::Reader.read_response(
             deadline:        config.world.deadline(config.timeout),
             io:              config.world.io,
+            job:             job,
             log_reader:      log_reader,
             marshal:         config.world.marshal,
             response_reader: response_reader

mutant.gemspec

@@ -30,7 +30,6 @@ Gem::Specification.new do |gem|
   gem.add_runtime_dependency('sorbet-runtime', '~> 0.5.0')
   gem.add_runtime_dependency('unparser',       '~> 0.6.9')
 
-  gem.add_development_dependency('parallel',   '~> 1.3')
   gem.add_development_dependency('rspec',      '~> 3.10')
   gem.add_development_dependency('rspec-core', '~> 3.10')
   gem.add_development_dependency('rspec-its',  '~> 1.3.0')

spec/integration/mutant/parallel_spec.rb

@@ -51,16 +51,19 @@ def status
       [
         Mutant::Parallel::Response.new(
           error:  nil,
+          job:    Mutant::Parallel::Source::Job.new(index: 0, payload: 1),
           log:    "Booting: 0\nPayload: 1\n",
           result: 2
         ),
         Mutant::Parallel::Response.new(
           error:  nil,
+          job:    Mutant::Parallel::Source::Job.new(index: 1, payload: 2),
           log:    "Payload: 2\n",
           result: 4
         ),
         Mutant::Parallel::Response.new(
           error:  nil,
+          job:    Mutant::Parallel::Source::Job.new(index: 2, payload: 3),
           log:    "Payload: 3\n",
           result: 6
         )
@@ -98,7 +101,14 @@ def status
 
     response_a, response_b = responses
 
-    expect(response_a).to eql(Mutant::Parallel::Response.new(log: '', result: nil, error: nil))
+    expect(response_a).to eql(
+      Mutant::Parallel::Response.new(
+        error:  nil,
+        job:    Mutant::Parallel::Source::Job.new(index: 0, payload: 1),
+        log:    '',
+        result: nil
+      )
+    )
     expect(response_b.error).to be(EOFError)
     expect(response_b.result).to be(nil)
     expect(response_b.log.match?('<main>')).to be(true)
@@ -134,11 +144,13 @@ def status
       [
         Mutant::Parallel::Response.new(
           error:  nil,
+          job:    Mutant::Parallel::Source::Job.new(index: 0, payload: 1),
           log:    "#{b}\n#{b}\n",
           result: b
         ),
         Mutant::Parallel::Response.new(
           error:  nil,
+          job:    Mutant::Parallel::Source::Job.new(index: 1, payload: 2),
           log:    "#{b}#{b}\n",
           result: b * 2
         )
@@ -233,9 +245,10 @@ def status
     expect(sink.status).to eql(
       [
         Mutant::Parallel::Response.new(
+          error:  Timeout::Error,
+          job:    Mutant::Parallel::Source::Job.new(index: 0, payload: 1),
           log:    '',
-          result: nil,
-          error:  Timeout
+          result: nil
         )
       ]
     )

spec/support/corpus.rb

@@ -1,5 +1,6 @@
 # frozen_string_literal: true
 
+require 'etc'
 require 'mutant'
 require 'parallel'
 
@@ -21,12 +22,9 @@ class Project
       MUTEX = Mutex.new
 
       MUTATION_GENERATION_MESSAGE = 'Total Mutations/Time/Parse-Errors: %s/%0.2fs - %0.2f/s'
-      START_MESSAGE               = 'Starting - %s'
-      FINISH_MESSAGE              = 'Mutations - %4i - %s'
+      FINISH_MESSAGE              = '%4i - %s'
       RUBY_GLOB_PATTERN           = '**/*.rb'
 
-      DEFAULT_MUTATION_COUNT = 0
-
       include Adamantium, Anima.new(
         :mutation_coverage,
         :mutation_generation,
@@ -73,6 +71,33 @@ def concurrency_limits
         end
       end
 
+      class Sink
+        include Mutant::Parallel::Sink
+
+        attr_reader :total
+
+        def initialize
+          @total = 0
+        end
+
+        def stop?
+          false
+        end
+
+        def status
+          @total
+        end
+
+        def response(response)
+          if response.error
+            Mutant::WORLD.stderr.puts(response.log)
+            fail response.error
+          end
+          puts(FINISH_MESSAGE % [response.result, response.job.payload])
+          @total += response.result
+        end
+      end
+
       # Verify mutation generation
       #
       # @return [self]
@@ -82,24 +107,38 @@ def concurrency_limits
       #   otherwise
       def verify_mutation_generation
         checkout
-        timer = Mutant::Timer.new(process: Process)
 
-        start = timer.now
+        sink = Sink.new
 
-        options = {
-          finish:       method(:finish),
-          start:        method(:start),
-          in_processes: Etc.nprocessors
-        }
+        elapsed = Mutant::WORLD.timer.elapsed do
+          driver = Mutant::Parallel.async(
+            config: parallel_config(sink),
+            world:  Mutant::WORLD
+          )
 
-        total = Parallel.map(effective_ruby_paths, options, &method(:check_generation))
-          .reduce(DEFAULT_MUTATION_COUNT, :+)
+          loop do
+            status = driver.wait_timeout(1)
+            break if status.done?
+          end
+        end
 
-        took = timer.now - start
-        puts MUTATION_GENERATION_MESSAGE % [total, took, total / took]
+        puts MUTATION_GENERATION_MESSAGE % [sink.total, elapsed, sink.total / elapsed]
         self
       end
 
+      def parallel_config(sink)
+        Mutant::Parallel::Config.new(
+          block:            method(:check_generation),
+          jobs:             Etc.nprocessors,
+          sink:             sink,
+          timeout:          nil,
+          process_name:     'mutation-generation',
+          source:           Mutant::Parallel::Source::Array.new(jobs: effective_ruby_paths),
+          on_process_start: ->(_) {},
+          thread_name:      'mutation-generation'
+        )
+      end
+
       # Checkout repository
       #
       # @return [self]
@@ -220,33 +259,6 @@ def noinstall?
         ENV.key?('NOINSTALL')
       end
 
-      # Print start progress
-      #
-      # @param [Pathname] path
-      # @param [Integer] _index
-      #
-      # @return [undefined]
-      #
-      def start(path, _index)
-        MUTEX.synchronize do
-          puts START_MESSAGE % path
-        end
-      end
-
-      # Print finish progress
-      #
-      # @param [Pathname] path
-      # @param [Integer] _index
-      # @param [Integer] count
-      #
-      # @return [undefined]
-      #
-      def finish(path, _index, count)
-        MUTEX.synchronize do
-          puts FINISH_MESSAGE % [count, path]
-        end
-      end
-
       # Helper method to execute system commands
       #
       # @param [Array<String>] arguments

spec/unit/mutant/mutation/runner/sink_spec.rb

@@ -5,17 +5,19 @@
 
   let(:mutation_a_index_response) do
     Mutant::Parallel::Response.new(
-      result: mutation_a_index_result,
+      error:  nil,
+      job:    0,
       log:    '',
-      error:  nil
+      result: mutation_a_index_result
     )
   end
 
   let(:mutation_b_index_response) do
     Mutant::Parallel::Response.new(
-      result: mutation_b_index_result,
+      error:  nil,
+      job:    0,
       log:    '',
-      error:  nil
+      result: mutation_b_index_result
     )
   end
 

spec/unit/mutant/parallel/connection/reader_spec.rb

@@ -1,20 +1,22 @@
 # frozen_string_literal: true
 
 RSpec.describe Mutant::Parallel::Connection::Reader do
-  let(:deadline)        { instance_double(Mutant::Timer::Deadline) }
-  let(:header_segment)  { [result_segment.bytesize].pack('N')      }
-  let(:io)              { class_double(IO)                         }
-  let(:marshal)         { class_double(Marshal)                    }
-  let(:log_reader)      { instance_double(IO)                      }
-  let(:response_reader) { instance_double(IO)                      }
-  let(:result)          { double('reader-result')                  }
-  let(:result_segment)  { '<result-segment>'                       }
+  let(:deadline)        { instance_double(Mutant::Timer::Deadline)                  }
+  let(:header_segment)  { [result_segment.bytesize].pack('N')                       }
+  let(:io)              { class_double(IO)                                          }
+  let(:job)             { Mutant::Parallel::Source::Job.new(index: 0, payload: nil) }
+  let(:log_reader)      { instance_double(IO)                                       }
+  let(:marshal)         { class_double(Marshal)                                     }
+  let(:response_reader) { instance_double(IO)                                       }
+  let(:result)          { double('reader-result')                                   }
+  let(:result_segment)  { '<result-segment>'                                        }
 
   describe '.read_response' do
     def apply
       described_class.read_response(
         deadline:        deadline,
         io:              io,
+        job:             job,
         log_reader:      log_reader,
         marshal:         marshal,
         response_reader: response_reader
@@ -90,6 +92,7 @@ def marshal_load
             expect(apply).to eql(
               Mutant::Parallel::Response.new(
                 error:  nil,
+                job:    job,
                 log:    '',
                 result: result
               )
@@ -121,6 +124,7 @@ def marshal_load
             expect(apply).to eql(
               Mutant::Parallel::Response.new(
                 error:  nil,
+                job:    job,
                 log:    '',
                 result: result
               )
@@ -157,6 +161,7 @@ def marshal_load
             expect(apply).to eql(
               Mutant::Parallel::Response.new(
                 error:  nil,
+                job:    job,
                 log:    '',
                 result: result
               )
@@ -186,7 +191,8 @@ def marshal_load
         verify_events do
           expect(apply).to eql(
             Mutant::Parallel::Response.new(
-              error:  Timeout,
+              error:  Timeout::Error,
+              job:    job,
               log:    '<log>',
               result: nil
             )
@@ -214,6 +220,7 @@ def marshal_load
           expect(apply).to eql(
             Mutant::Parallel::Response.new(
               error:  EOFError,
+              job:    job,
               log:    '',
               result: nil
             )
@@ -234,7 +241,8 @@ def marshal_load
         verify_events do
           expect(apply).to eql(
             Mutant::Parallel::Response.new(
-              error:  Timeout,
+              error:  Timeout::Error,
+              job:    job,
               log:    '',
               result: nil
             )
@@ -254,7 +262,8 @@ def marshal_load
         verify_events do
           expect(apply).to eql(
             Mutant::Parallel::Response.new(
-              error:  Timeout,
+              error:  Timeout::Error,
+              job:    job,
               log:    '',
               result: nil
             )