Merge pull request #334 from Shopify/shioyama/parallel_ci_queue

Remove rate limit on reserved tests

Author: shioyama

ruby/lib/ci/queue/redis/worker.rb

@@ -16,7 +16,7 @@ class Worker < Base
         attr_reader :total
 
         def initialize(redis, config)
-          @reserved_test = nil
+          @reserved_tests = Set.new
           @shutdown_required = false
           super(redis, config)
         end
@@ -135,7 +135,7 @@ def requeue(test, offset: Redis.requeue_offset)
             argv: [config.max_requeues, global_max_requeues, test_key, offset],
           ) == 1
 
-          @reserved_test = test_key unless requeued
+          reserved_tests << test_key unless requeued
           requeued
         end
 
@@ -152,25 +152,24 @@ def release!
 
         attr_reader :index
 
+        def reserved_tests
+          @reserved_tests ||= Set.new
+        end
+
         def worker_id
           config.worker_id
         end
 
-        def raise_on_mismatching_test(test_key)
-          if @reserved_test == test_key
-            @reserved_test = nil
-          else
-            raise ReservationError, "Acknowledged #{test_key.inspect} but #{@reserved_test.inspect} was reserved"
+        def raise_on_mismatching_test(test)
+          unless reserved_tests.delete?(test)
+            raise ReservationError, "Acknowledged #{test.inspect} but only #{reserved_tests.map(&:inspect).join(", ")} reserved"
           end
         end
 
         def reserve
-          if @reserved_test
-            raise ReservationError, "#{@reserved_test.inspect} is already reserved. " \
-              "You have to acknowledge it before you can reserve another one"
-          end
-
-          @reserved_test = (try_to_reserve_lost_test || try_to_reserve_test)
+          test = (try_to_reserve_lost_test || try_to_reserve_test)
+          reserved_tests << test
+          test
         end
 
         def try_to_reserve_test

ruby/lib/ci/queue/static.rb

@@ -89,14 +89,15 @@ def remaining
       end
 
       def running
-        @reserved_test ? 1 : 0
+        reserved_tests.empty? ? 0 : 1
       end
 
       def poll
-        while !@shutdown && config.circuit_breakers.none?(&:open?) && !max_test_failed? && @reserved_test = @queue.shift
-          yield index.fetch(@reserved_test)
+        while !@shutdown && config.circuit_breakers.none?(&:open?) && !max_test_failed? && reserved_test = @queue.shift
+          reserved_tests << reserved_test
+          yield index.fetch(reserved_test)
         end
-        @reserved_test = nil
+        reserved_tests.clear
       end
 
       def exhausted?
@@ -142,6 +143,10 @@ def should_requeue?(key)
       def requeues
         @requeues ||= Hash.new(0)
       end
+
+      def reserved_tests
+        @reserved_tests ||= Set.new
+      end
     end
   end
 end

ruby/test/minitest/queue/build_status_recorder_test.rb

@@ -79,7 +79,7 @@ def test_retrying_test
     private
 
     def reserve(queue, method_name)
-      queue.instance_variable_set(:@reserved_test, Minitest::Queue::SingleExample.new("Minitest::Test", method_name).id)
+      queue.instance_variable_set(:@reserved_tests, Set.new([Minitest::Queue::SingleExample.new("Minitest::Test", method_name).id]))
     end
 
     def worker(id)