wip

Author: ebarajas

ruby/lib/ci/queue.rb

@@ -42,15 +42,6 @@ def requeueable?(test_result)
       requeueable.nil? || requeueable.call(test_result)
     end
 
-    def shuffle(tests, random, config: nil)
-      if shuffler
-        shuffler.call(tests, random)
-      else
-        strategy = get_strategy(config&.strategy)
-        strategy.order_tests(tests, random: random, config: config)
-      end
-    end
-
     def from_uri(url, config)
       uri = URI(url)
       implementation = case uri.scheme
@@ -66,18 +57,5 @@ def from_uri(url, config)
       end
       implementation.from_uri(uri, config)
     end
-
-    private
-
-    def get_strategy(strategy_name)
-      case strategy_name&.to_sym
-      when :timing_based
-        Strategy::TimingBased.new
-      when :suite_bin_packing
-        Strategy::SuiteBinPacking.new
-      else
-        Strategy::Random.new
-      end
-    end
   end
 end

ruby/lib/ci/queue/common.rb

@@ -36,6 +36,28 @@ def rescue_connection_errors(handler = ->(err) { nil })
       rescue *self::class::CONNECTION_ERRORS => err
         handler.call(err)
       end
+
+      def ordering_strategy
+        case config.strategy.to_sym
+        when :timing_based
+          Strategy::TimingBased.new(config)
+        when :suite_bin_packing
+          # pass redis if available
+          # need to think about a better way to structure queue/strategy interaction
+          redis_instance = if self.respond_to?(:redis, true) # include private methods
+            self.send(:redis)
+          else
+            nil
+          end
+          Strategy::SuiteBinPacking.new(config, redis: redis_instance)
+        else
+          Strategy::Random.new(config)
+        end
+      end
+
+      def reorder_tests(tests, random: Random.new)
+        ordering_strategy.order_tests(tests, random: random)
+      end
     end
   end
 end

ruby/lib/ci/queue/redis.rb

@@ -10,6 +10,7 @@
 require 'ci/queue/redis/supervisor'
 require 'ci/queue/redis/grind_supervisor'
 require 'ci/queue/redis/test_time_record'
+require 'ci/queue/redis/moving_average'
 
 module CI
   module Queue

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

@@ -0,0 +1,65 @@
+# frozen_string_literal: true
+
+module CI
+  module Queue
+    module Redis
+      # Represents a redis hash of moving averages for test durations
+      #
+      # Moving average is calculated using exponential moving average formula
+      class MovingAverage
+        LUA_SCRIPT= <<~LUA
+          local hash_key = KEYS[1]
+          local test_id = ARGV[1]
+          local new_duration = tonumber(ARGV[2])
+          local smoothing = tonumber(ARGV[3])
+          local current_avg = redis.call('HGET', hash_key, test_id)
+          if current_avg then
+            current_avg = tonumber(current_avg)
+            local new_avg = smoothing * new_duration + (1 - smoothing) * current_avg
+            redis.call('HSET', hash_key, test_id, new_avg)
+            return tostring(new_avg)
+          else
+            redis.call('HSET', hash_key, test_id, new_duration)
+            return tostring(new_duration)
+          end
+        LUA
+
+        def initialize(redis, key: "test_duration_moving_averages", smoothing_factor: 0.2)
+          @redis = redis
+          @key = key
+          @smoothing_factor = smoothing_factor
+          @values = {}
+        end
+
+        def [](test_id)
+          load_all if @values.empty?
+          @values[test_id]
+        end
+
+        def update(test_id, duration)
+          new_avg = @redis.eval(LUA_SCRIPT, keys: [@key], argv: [test_id, duration, @smoothing_factor])
+          @values[test_id] = new_avg.to_f
+          new_avg.to_f
+        end
+
+        def load_all
+          batch_size = 1000
+          cursor = '0'
+          @values = {}
+
+          loop do
+            cursor, batch = @redis.hscan(@key, cursor, count: batch_size)
+            batch.each do |test_id, value|
+              @values[test_id] = value.to_f
+            end
+            break if cursor == '0'
+          end
+        end
+
+        def size
+          @redis.hlen(@key)
+        end
+      end
+    end
+  end
+end

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

@@ -5,7 +5,9 @@ module Redis
       class TestTimeRecord < Worker
         def record(test_name, duration)
           record_test_time(test_name, duration)
+          record_test_duration_moving_average(test_name, duration)
           record_test_name(test_name)
+
         end
 
         def fetch
@@ -29,6 +31,10 @@ def record_test_time(test_name, duration)
           nil
         end
 
+        def record_test_duration_moving_average(test_name, duration)
+          MovingAverage.new(redis).update(test_name, duration)
+        end
+
         def record_test_name(test_name)
           redis.pipelined do |pipeline|
             pipeline.lpush(

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

@@ -30,7 +30,7 @@ def distributed?
 
         def populate(tests, random: Random.new)
           @index = tests.map { |t| [t.id, t] }.to_h
-          executables = Queue.shuffle(tests, random, config: config)
+          executables = reorder_tests(tests, random:)
 
           # Separate chunks from individual tests
           chunks = executables.select { |e| e.is_a?(CI::Queue::TestChunk) }

ruby/lib/ci/queue/strategy/base.rb

@@ -4,10 +4,16 @@ module CI
   module Queue
     module Strategy
       class Base
-        def order_tests(tests, random: Random.new, config: nil)
+        def initialize(config)
+          @config = config
+        end
+
+        attr_reader :config
+
+        def order_tests(tests)
           raise NotImplementedError, "#{self.class} must implement #order_tests"
         end
       end
     end
   end
-end
+end

ruby/lib/ci/queue/strategy/random.rb

@@ -5,10 +5,10 @@ module CI
   module Queue
     module Strategy
       class Random < Base
-        def order_tests(tests, random: Random.new, config: nil)
+        def order_tests(tests, random: Random.new)
           tests.sort.shuffle(random: random)
         end
       end
     end
   end
-end
+end

ruby/lib/ci/queue/strategy/suite_bin_packing.rb

@@ -6,12 +6,36 @@ module CI
   module Queue
     module Strategy
       class SuiteBinPacking < Base
-        def order_tests(tests, random: Random.new, config: nil)
-          timing_data = load_timing_data(config&.timing_file)
-          max_duration = config&.suite_max_duration || 120_000
-          fallback_duration = config&.timing_fallback_duration || 100.0
-          buffer_percent = config&.suite_buffer_percent || 10
+        class << self
+          def load_timing_data(file_path)
+            return {} unless file_path && ::File.exist?(file_path)
+
+            JSON.parse(::File.read(file_path))
+          rescue JSON::ParserError => e
+            warn "Warning: Could not parse timing file #{file_path}: #{e.message}"
+            {}
+          end
+        end
+
+        def initialize(config, redis: nil)
+          super(config)
+
+          if redis
+            @moving_average = CI::Queue::Redis::MovingAverage.new(redis)
+          end
+
+          if config&.timing_file
+            @timing_data = self.class.load_timing_data(config.timing_file)
+          else
+            @timing_data = {}
+          end
 
+          @max_duration = config&.suite_max_duration || 120_000
+          @fallback_duration = config&.timing_fallback_duration || 100.0
+          @buffer_percent = config&.suite_buffer_percent || 10
+        end
+
+        def order_tests(tests, random: ::Random.new, redis: nil)
           # Group tests by suite name
           suites = tests.group_by { |test| extract_suite_name(test.id) }
 
@@ -22,10 +46,6 @@ def order_tests(tests, random: Random.new, config: nil)
               create_chunks_for_suite(
                 suite_name,
                 suite_tests,
-                max_duration,
-                buffer_percent,
-                timing_data,
-                fallback_duration
               )
             )
           end
@@ -40,27 +60,27 @@ def extract_suite_name(test_id)
           test_id.split('#').first
         end
 
-        def load_timing_data(file_path)
-          return {} unless file_path && ::File.exist?(file_path)
-
-          JSON.parse(::File.read(file_path))
-        rescue JSON::ParserError => e
-          warn "Warning: Could not parse timing file #{file_path}: #{e.message}"
-          {}
-        end
+        def get_test_duration(test_id)
+          if @moving_average
+            avg = @moving_average[test_id]
+            return avg if avg
+          end
 
-        def get_test_duration(test_id, timing_data, fallback_duration)
-          timing_data[test_id]&.to_f || fallback_duration
+          if @timing_data.key?(test_id)
+            @timing_data[test_id]
+          else
+            @fallback_duration
+          end
         end
 
-        def create_chunks_for_suite(suite_name, suite_tests, max_duration, buffer_percent, timing_data, fallback_duration)
+        def create_chunks_for_suite(suite_name, suite_tests)
           # Calculate total suite duration
           total_duration = suite_tests.sum do |test|
-            get_test_duration(test.id, timing_data, fallback_duration)
+            get_test_duration(test.id)
           end
 
           # If suite fits in max duration, create full_suite chunk
-          if total_duration <= max_duration
+          if total_duration <= @max_duration
             chunk_id = "#{suite_name}:full_suite"
             # Don't store test_ids in Redis - worker will resolve from index
             # But pass test_count for timeout calculation
@@ -71,20 +91,16 @@ def create_chunks_for_suite(suite_name, suite_tests, max_duration, buffer_percen
           split_suite_into_chunks(
             suite_name,
             suite_tests,
-            max_duration,
-            buffer_percent,
-            timing_data,
-            fallback_duration
           )
         end
 
-        def split_suite_into_chunks(suite_name, suite_tests, max_duration, buffer_percent, timing_data, fallback_duration)
+        def split_suite_into_chunks(suite_name, suite_tests)
           # Apply buffer to max duration
-          effective_max = max_duration * (1 - buffer_percent / 100.0)
+          effective_max = @max_duration * (1 - @buffer_percent / 100.0)
 
           # Sort tests by duration (longest first for better bin packing)
           sorted_tests = suite_tests.sort_by do |test|
-            -get_test_duration(test.id, timing_data, fallback_duration)
+            -get_test_duration(test.id)
           end
 
           # First-fit decreasing bin packing
@@ -94,7 +110,7 @@ def split_suite_into_chunks(suite_name, suite_tests, max_duration, buffer_percen
           chunk_index = 0
 
           sorted_tests.each do |test|
-            test_duration = get_test_duration(test.id, timing_data, fallback_duration)
+            test_duration = get_test_duration(test.id)
 
             if current_chunk_duration + test_duration > effective_max && current_chunk_tests.any?
               # Finalize current chunk and start new one

ruby/lib/ci/queue/strategy/timing_based.rb

@@ -6,7 +6,7 @@ module CI
   module Queue
     module Strategy
       class TimingBased < Base
-        def order_tests(tests, random: Random.new, config: nil)
+        def order_tests(tests, random: Random.new)
           timing_data = load_timing_data(config&.timing_file)
           fallback_duration = config&.timing_fallback_duration || 100.0
 
@@ -20,7 +20,7 @@ def order_tests(tests, random: Random.new, config: nil)
 
         def load_timing_data(file_path)
           return {} unless file_path && ::File.exist?(file_path)
-          
+
           JSON.parse(::File.read(file_path))
         rescue JSON::ParserError => e
           warn "Warning: Could not parse timing file #{file_path}: #{e.message}"
@@ -32,4 +32,4 @@ def load_timing_data(file_path)
       end
     end
   end
-end
+end