Concurrent processing assets [3.x backport]

lib/sprockets/manifest.rb

@@ -77,6 +77,7 @@ def initialize(*args)
       end
 
       @data = data
+      @executor = Concurrent::FixedThreadPool.new([8, Concurrent.processor_count].min)
     end
 
     # Returns String path to manifest.json file.
@@ -115,6 +116,8 @@ def files
       @data['files'] ||= {}
     end
 
+    attr_accessor :executor
+
     # Public: Find all assets matching pattern set in environment.
     #
     # Returns Enumerator of Assets.
@@ -130,22 +133,27 @@ def find(*args)
 
       environment = self.environment.cached
 
-      paths.each do |path|
-        environment.find_all_linked_assets(path) do |asset|
-          yield asset
+      promises = paths.map do |path|
+        Concurrent::Promise.execute(executor: executor) do
+          environment.find_all_linked_assets(path) do |asset|
+            yield asset
+          end
         end
       end
 
       if filters.any?
-        environment.logical_paths do |logical_path, filename|
-          if filters.any? { |f| f.call(logical_path, filename) }
-            environment.find_all_linked_assets(filename) do |asset|
-              yield asset
+        promises += environment.logical_paths.map do |logical_path, filename|
+          Concurrent::Promise.execute(executor: executor) do
+            if filters.any? { |f| f.call(logical_path, filename) }
+              environment.find_all_linked_assets(filename) do |asset|
+                yield asset
+              end
             end
           end
         end
       end
 
+      Concurrent::Promise.zip(*promises).wait!
       nil
     end
 

test/test_manifest.rb

@@ -701,4 +701,34 @@ def teardown
       assert_raises('kaboom') { manifest.compile('application.js') }
     end
   end
+
+  # Record Processor sequence with thread context-switching to test concurrency.
+  class SlowProcessor
+    attr_reader :seq, :threads, :total
+
+    def initialize(total)
+      @total = total
+      @seq = []
+      @threads = []
+    end
+
+    def call(_)
+      seq << '0'
+      threads << Thread.current
+      if threads.length == total
+        threads.map(&:run)
+      else
+        Thread.stop
+      end
+      seq << '1'
+      nil
+    end
+  end
+
+  test 'concurrent processing' do
+    processor = SlowProcessor.new(2)
+    @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
 end