Concurrent asset compile

lib/sprockets/manifest.rb

@@ -120,11 +120,14 @@ def find(*args)
       return to_enum(__method__, *args) unless block_given?
 
       environment = self.environment.cached
-      args.flatten.each do |path|
-        environment.find_all_linked_assets(path) do |asset|
-          yield asset
+      promises = args.flatten.map do |path|
+        Concurrent::Promise.execute(executor: executor) do
+          environment.find_all_linked_assets(path) do |asset|
+            yield asset
+          end
         end
       end
+      promises.each(&:wait!)
 
       nil
     end
@@ -160,7 +163,6 @@ def compile(*args)
 
       filenames            = []
       concurrent_exporters = []
-      executor             = Concurrent::FixedThreadPool.new(Concurrent.processor_count)
 
       find(*args) do |asset|
         mtime = Time.now.iso8601
@@ -183,11 +185,6 @@ def compile(*args)
         exporters_for_asset(asset) do |exporter|
           next if exporter.skip?(logger)
 
-          if !environment.export_concurrent
-            exporter.call
-            next
-          end
-
           if promise.nil?
             promise = Concurrent::Promise.new(executor: executor) { exporter.call }
             concurrent_exporters << promise.execute
@@ -327,5 +324,9 @@ def logger
           logger
         end
       end
+
+      def executor
+        @executor ||= environment.export_concurrent ? :fast : :immediate
+      end
   end
 end

test/test_manifest.rb

@@ -505,7 +505,7 @@ def teardown
     manifest = Sprockets::Manifest.new(@env, @dir)
 
     result = manifest.find_sources("mobile/a.js", "mobile/b.js")
-    assert_equal ["var A;\n", "var B;\n"], result.to_a
+    assert_equal ["var A;\n", "var B;\n"], result.to_a.sort
   end
 
   test "find_sources without environment" do
@@ -571,4 +571,73 @@ def teardown
       assert_equal 'kaboom', ex.message
     end
   end
+
+  # Sleep duration to context switch between concurrent threads.
+  CONTEXT_SWITCH_DURATION = 0.1
+
+  # Record Exporter sequence with a delay to test concurrency.
+  class SlowExporter < Sprockets::Exporters::Base
+    class << self
+      attr_accessor :seq
+    end
+
+    def call
+      SlowExporter.seq << '0'
+      sleep CONTEXT_SWITCH_DURATION
+      SlowExporter.seq << '1'
+    end
+  end
+
+  class SlowExporter2 < SlowExporter
+  end
+
+  test 'concurrent exporting' do
+    # Register 2 exporters and compile 2 files to ensure that
+    # all 4 exporting tasks run concurrently.
+    SlowExporter.seq = []
+    @env.register_exporter 'image/png',SlowExporter
+    @env.register_exporter 'image/png',SlowExporter2
+    Sprockets::Manifest.new(@env, @dir).compile('logo.png', 'troll.png')
+    assert_equal %w(0 0 0 0 1 1 1 1), SlowExporter.seq
+  end
+
+  test 'sequential exporting' do
+    @env.export_concurrent = false
+    SlowExporter.seq = []
+    @env.register_exporter 'image/png',SlowExporter
+    @env.register_exporter 'image/png',SlowExporter2
+    Sprockets::Manifest.new(@env, @dir).compile('logo.png', 'troll.png')
+    assert_equal %w(0 1 0 1 0 1 0 1), SlowExporter.seq
+  end
+
+  # Record Processor sequence with a delay to test concurrency.
+  class SlowProcessor
+    attr_reader :seq
+
+    def initialize
+      @seq = []
+    end
+
+    def call(_)
+      seq << '0'
+      sleep CONTEXT_SWITCH_DURATION
+      seq << '1'
+      nil
+    end
+  end
+
+  test 'concurrent processing' do
+    processor = SlowProcessor.new
+    @env.register_postprocessor 'image/png', processor
+    Sprockets::Manifest.new(@env, @dir).compile('logo.png', 'troll.png')
+    assert_equal %w(0 0 1 1), processor.seq
+  end
+
+  test 'sequential processing' do
+    @env.export_concurrent = false
+    processor = SlowProcessor.new
+    @env.register_postprocessor 'image/png', processor
+    Sprockets::Manifest.new(@env, @dir).compile('logo.png', 'troll.png')
+    assert_equal %w(0 1 0 1), processor.seq
+  end
 end