Add concurrency via flay fork and concurrent-ruby

This PR swaps out flay with our own fork of flay that uses
concurrency-ruby classes instead of stdlib `Hash`.

This also introduces a `FileThreadPool` class for running blocks on an
array of files concurrently.

Author: BlakeWilliams

Diff for: Dockerfile

@@ -1,3 +1,4 @@
+
 FROM alpine:edge
 
 WORKDIR /usr/src/app/
@@ -8,8 +9,9 @@ COPY Gemfile.lock /usr/src/app/
 COPY vendor/php-parser/composer.json /usr/src/app/vendor/php-parser/
 COPY vendor/php-parser/composer.lock /usr/src/app/vendor/php-parser/
 
-RUN apk --update add git python nodejs php-cli php-json php-phar php-openssl php-xml curl\
-    ruby ruby-io-console ruby-dev ruby-bundler build-base && \
+RUN apk --update add openssh git python nodejs php-cli php-json php-phar php-openssl php-xml curl\
+    ruby ruby-io-console ruby-dev build-base && \
+    gem install bundler --no-ri --no-rdoc && \
     bundle install -j 4 && \
     apk del build-base && rm -fr /usr/share/ri && \
     curl -sS https://getcomposer.org/installer | php

Diff for: Gemfile

@@ -1,6 +1,8 @@
 source 'https://rubygems.org'
 
-gem 'flay'
+gem 'flay', git: 'https://github.com/codeclimate/flay.git'
+gem 'concurrent-ruby', "~> 1.0.0.pre4"
+gem 'ruby_parser'
 gem 'pry'
 gem 'posix-spawn'
 gem 'sexp_processor'

Diff for: Gemfile.lock

@@ -1,15 +1,19 @@
+GIT
+  remote: https://github.com/codeclimate/flay.git
+  revision: e6e472e064dc459277f8bd57167789f8fed77f56
+  specs:
+    flay (2.6.1)
+
 GEM
   remote: https://rubygems.org/
   specs:
     coderay (1.1.0)
+    concurrent-ruby (1.0.0.pre4)
     diff-lcs (1.2.5)
-    flay (2.6.1)
-      ruby_parser (~> 3.0)
-      sexp_processor (~> 4.0)
     json (1.8.3)
     method_source (0.8.2)
     posix-spawn (0.3.11)
-    pry (0.10.1)
+    pry (0.10.3)
       coderay (~> 1.1.0)
       method_source (~> 0.8.1)
       slop (~> 3.4)
@@ -27,7 +31,7 @@ GEM
       diff-lcs (>= 1.2.0, < 2.0)
       rspec-support (~> 3.3.0)
     rspec-support (3.3.0)
-    ruby_parser (3.7.1)
+    ruby_parser (3.7.2)
       sexp_processor (~> 4.1)
     sexp_processor (4.6.0)
     slop (3.6.0)
@@ -36,12 +40,14 @@ PLATFORMS
   ruby
 
 DEPENDENCIES
-  flay
+  concurrent-ruby (~> 1.0.0.pre4)
+  flay!
   json
   posix-spawn
   pry
   rake
   rspec
+  ruby_parser
   sexp_processor
 
 BUNDLED WITH

Diff for: circle.yml

@@ -5,9 +5,13 @@ machine:
     CLOUDSDK_CORE_DISABLE_PROMPTS: 1
     PRIVATE_REGISTRY: us.gcr.io/code_climate
 
-test:
+dependencies:
   override:
+    - docker info
     - docker build -t=$PRIVATE_REGISTRY/$CIRCLE_PROJECT_REPONAME:b$CIRCLE_BUILD_NUM .
+
+test:
+  override:
     - docker run $PRIVATE_REGISTRY/$CIRCLE_PROJECT_REPONAME:b$CIRCLE_BUILD_NUM bundle exec rake
 
 deployment:

Diff for: lib/cc/engine/analyzers/analyzer_base.rb

@@ -6,15 +6,15 @@ def initialize(engine_config:)
           @engine_config = engine_config
         end
 
-        def run
-          files.map do |file|
-            begin
-              process_file(file)
-            rescue => ex
-              $stderr.puts "Skipping file #{file} due to exception"
-              $stderr.puts "(#{ex.class}) #{ex.message} #{ex.backtrace.join("\n")}"
-            end
-          end
+        def run(file)
+          process_file(file)
+        rescue => ex
+          $stderr.puts "Skipping file #{file} due to exception"
+          $stderr.puts "(#{ex.class}) #{ex.message} #{ex.backtrace.join("\n")}"
+        end
+
+        def files
+          file_list.files
         end
 
         def mass_threshold
@@ -33,12 +33,12 @@ def process_file(path)
           raise NoMethodError.new("Subclass must implement `process_file`")
         end
 
-        def files
-          ::CC::Engine::Analyzers::FileList.new(
+        def file_list
+          @_file_list ||= ::CC::Engine::Analyzers::FileList.new(
             engine_config: engine_config,
             default_paths: self.class::DEFAULT_PATHS,
             language: self.class::LANGUAGE
-          ).files
+          )
         end
       end
     end

Diff for: lib/cc/engine/analyzers/engine_config.rb

@@ -14,6 +14,10 @@ def languages
           config.fetch("languages", {})
         end
 
+        def concurrency
+          config.fetch("concurrency", 2)
+        end
+
         def mass_threshold_for(language)
           threshold = fetch_language(language).fetch("mass_threshold", nil)
 

Diff for: lib/cc/engine/analyzers/file_thread_pool.rb

@@ -0,0 +1,58 @@
+require "thread"
+
+module CC
+  module Engine
+    module Analyzers
+      class FileThreadPool
+        DEFAULT_CONCURRENCY = 2
+        MAX_CONCURRENCY = 2
+
+        def initialize(files, concurrency: DEFAULT_CONCURRENCY)
+          @files = files
+          @concurrency = concurrency
+        end
+
+        def run(&block)
+          queue = build_queue
+
+          @workers = thread_count.times.map do
+            Thread.new do
+              begin 
+                while item = queue.pop(true)
+                  yield item
+                end
+              rescue ThreadError
+              end
+            end
+          end
+        end
+
+        def join
+          workers.map(&:join)
+        end
+
+        private
+
+        attr_reader :files, :concurrency, :workers
+
+        def build_queue
+          Queue.new.tap do |queue|
+            files.each do |file|
+              queue.push(file)
+            end
+          end
+        end
+
+        def thread_count
+          if (1..MAX_CONCURRENCY) === concurrency
+            concurrency
+          elsif concurrency < 1
+            DEFAULT_CONCURRENCY
+          else
+            DEFAULT_CONCURRENCY
+          end
+        end
+      end
+    end
+  end
+end

Diff for: lib/cc/engine/analyzers/javascript/main.rb

@@ -18,7 +18,6 @@ class Main < CC::Engine::Analyzers::Base
           DEFAULT_MASS_THRESHOLD = 40
           BASE_POINTS = 3000
 
-
           private
 
           def process_file(path)

Diff for: lib/cc/engine/analyzers/python/main.rb

@@ -15,7 +15,7 @@ class Main < CC::Engine::Analyzers::Base
           DEFAULT_MASS_THRESHOLD = 40
           BASE_POINTS = 1000
 
-          attr_reader :directory, :engine_config
+          private
 
           def process_file(path)
             Node.new(::CC::Engine::Analyzers::Python::Parser.new(File.binread(path), path).parse.syntax_tree, path).format

Diff for: lib/cc/engine/analyzers/reporter.rb

@@ -1,23 +1,36 @@
 require 'cc/engine/analyzers/violation'
+require 'cc/engine/analyzers/file_thread_pool'
+require 'thread'
 
 module CC
   module Engine
     module Analyzers
       class Reporter
         TIMEOUT = 10
-        def initialize(language_strategy, io)
+
+        def initialize(engine_config, language_strategy, io)
+          @engine_config = engine_config
           @language_strategy = language_strategy
           @io = io
         end
 
         def run
-          sexps = language_strategy.run
+          process_files
+          report
+        end
 
-          sexps.each do |sexp|
+        def process_files
+          pool = FileThreadPool.new(
+            language_strategy.files,
+            concurrency: engine_config.concurrency
+          )
+
+          pool.run do |file|
+            sexp = language_strategy.run(file)
             process_sexp(sexp)
           end
 
-          report
+          pool.join
         end
 
         def report
@@ -37,7 +50,7 @@ def flay
           @flay ||= Flay.new(flay_options)
         end
 
-        attr_reader :language_strategy, :io
+        attr_reader :engine_config, :language_strategy, :io
 
         def mass_threshold
           @mass_threshold ||= language_strategy.mass_threshold

Diff for: lib/cc/engine/analyzers/ruby/main.rb

@@ -23,8 +23,6 @@ class Main < CC::Engine::Analyzers::Base
 
           private
 
-          attr_reader :directory, :engine_config
-
           def process_file(file)
             RubyParser.new.process(File.binread(file), file, TIMEOUT)
           end
@@ -33,4 +31,3 @@ def process_file(file)
     end
   end
 end
-

Diff for: lib/cc/engine/duplication.rb

@@ -29,7 +29,7 @@ def initialize(directory:, engine_config:, io:)
       def run
         languages_to_analyze.each do |language|
           engine = LANGUAGES[language].new(engine_config: engine_config)
-          reporter = CC::Engine::Analyzers::Reporter.new(engine, io)
+          reporter = CC::Engine::Analyzers::Reporter.new(engine_config, engine, io)
           reporter.run
         end
       end

Diff for: spec/cc/engine/analyzers/file_thread_pool_spec.rb

@@ -0,0 +1,69 @@
+require "spec_helper"
+require "cc/engine/analyzers/file_thread_pool"
+
+RSpec.describe CC::Engine::Analyzers::FileThreadPool do
+  describe "#run" do
+    it "uses default count of threads when concurrency is not specified" do
+      allow(Thread).to receive(:new)
+
+      pool = CC::Engine::Analyzers::FileThreadPool.new([])
+      pool.run  {}
+
+      expect(Thread).to have_received(:new).exactly(
+        CC::Engine::Analyzers::FileThreadPool::DEFAULT_CONCURRENCY
+      ).times
+    end
+
+    it "uses default concurrency when concurrency is over max" do
+      allow(Thread).to receive(:new)
+
+      run_pool_with_concurrency(
+        CC::Engine::Analyzers::FileThreadPool::DEFAULT_CONCURRENCY + 2
+      )
+
+      expect(Thread).to have_received(:new).exactly(
+        CC::Engine::Analyzers::FileThreadPool::DEFAULT_CONCURRENCY
+      ).times
+    end
+
+    it "uses default concucurrency when concucurrency is under 1" do
+      allow(Thread).to receive(:new)
+
+      run_pool_with_concurrency(-2)
+
+      expect(Thread).to have_received(:new).exactly(
+        CC::Engine::Analyzers::FileThreadPool::DEFAULT_CONCURRENCY
+      ).times
+    end
+
+    it "uses supplied concurrency when valid" do
+      allow(Thread).to receive(:new)
+
+      run_pool_with_concurrency(1)
+
+      expect(Thread).to have_received(:new).exactly(1).times
+    end
+
+    it "calls block for each file" do
+      pool = CC::Engine::Analyzers::FileThreadPool.new(["abc", "123", "xyz"])
+
+      results = []
+      pool.run do |f|
+        results.push f.reverse
+      end
+      pool.join
+
+      expect(results).to include("cba")
+      expect(results).to include("321")
+      expect(results).to include("zyx")
+    end
+  end
+
+  def run_pool_with_concurrency(concurrency)
+      pool = CC::Engine::Analyzers::FileThreadPool.new(
+        [],
+        concurrency: concurrency
+      )
+      pool.run  {}
+  end
+end

Diff for: spec/cc/engine/analyzers/javascript/main_spec.rb

@@ -66,7 +66,7 @@ def run_engine(config = nil)
     io = StringIO.new
 
     engine = ::CC::Engine::Analyzers::Javascript::Main.new(engine_config: config)
-    reporter = ::CC::Engine::Analyzers::Reporter.new(engine, io)
+    reporter = ::CC::Engine::Analyzers::Reporter.new(double(concurrency: 2), engine, io)
 
     reporter.run
 

Diff for: spec/cc/engine/analyzers/php/main_spec.rb

@@ -66,7 +66,7 @@ def run_engine(config = nil)
     io = StringIO.new
 
     engine = ::CC::Engine::Analyzers::Php::Main.new(engine_config: config)
-    reporter = ::CC::Engine::Analyzers::Reporter.new(engine, io)
+    reporter = ::CC::Engine::Analyzers::Reporter.new(double(concurrency: 2), engine, io)
 
     reporter.run
 

Diff for: spec/cc/engine/analyzers/python/main_spec.rb

@@ -54,7 +54,7 @@ def run_engine(config = nil)
     io = StringIO.new
 
     engine = ::CC::Engine::Analyzers::Python::Main.new(engine_config: config)
-    reporter = ::CC::Engine::Analyzers::Reporter.new(engine, io)
+    reporter = ::CC::Engine::Analyzers::Reporter.new(double(concurrency: 2), engine, io)
 
     reporter.run
 

Diff for: spec/cc/engine/analyzers/ruby/main_spec.rb

@@ -74,7 +74,7 @@ def run_engine(config = {})
 
     config = CC::Engine::Analyzers::EngineConfig.new(config)
     engine = ::CC::Engine::Analyzers::Ruby::Main.new(engine_config: config)
-    reporter = ::CC::Engine::Analyzers::Reporter.new(engine, io)
+    reporter = ::CC::Engine::Analyzers::Reporter.new(double(concurrency: 2), engine, io)
 
     reporter.run