Skip to Content Skip to Search

Round-robin distributor - tests are assigned to workers as they arrive.

Tests arrive already shuffled by Minitest based on the seed. Since the arrival order is deterministic for a given seed, round-robin assignment produces reproducible test-to-worker distribution.

This is much simpler than buffering and re-shuffling: tests can start executing immediately as they arrive, and we avoid complex synchronization.

Methods
A
C
I
N
P
T

Constants

WORK_WAIT_TIMEOUT = 0.1
 

Class Public methods

new(worker_count:)

Source: show | on GitHub 

# File activesupport/lib/active_support/testing/parallelization/test_distributor.rb, line 78
def initialize(worker_count:)
  @worker_count = worker_count
  @queues = Array.new(@worker_count) { Queue.new }
  @next_worker = 0
  @mutex = Mutex.new
  @cv = ConditionVariable.new
  @closed = false
end

Instance Public methods

add_test(test)

Source: show | on GitHub 

# File activesupport/lib/active_support/testing/parallelization/test_distributor.rb, line 87
def add_test(test)
  @mutex.synchronize do
    return if @closed || !@queues

    worker_id = @next_worker
    @next_worker = (@next_worker + 1) % @worker_count
    queue = @queues[worker_id]
    queue << test unless queue.closed?
    @cv.signal  # Wake one waiting worker
  end
end

close()

Source: show | on GitHub 

# File activesupport/lib/active_support/testing/parallelization/test_distributor.rb, line 127
def close
  @mutex.synchronize do
    @queues&.each(&:close)
    @closed = true
    @cv.broadcast  # Wake all waiting workers
  end
end

interrupt()

Source: show | on GitHub 

# File activesupport/lib/active_support/testing/parallelization/test_distributor.rb, line 110
def interrupt
  @mutex.synchronize do
    @queues&.each do |q|
      q.clear
      q.close
    end
    @closed = true
    @cv.broadcast  # Wake all waiting workers
  end
end

pending?()

Source: show | on GitHub 

# File activesupport/lib/active_support/testing/parallelization/test_distributor.rb, line 121
def pending?
  @mutex.synchronize do
    @queues&.any? { |q| !q.empty? }
  end
end

take(worker_id:)

Source: show | on GitHub 

# File activesupport/lib/active_support/testing/parallelization/test_distributor.rb, line 99
def take(worker_id:)
  job = nil

  until job || exhausted?(worker_id)
    job = next_job(worker_id)
    wait(worker_id) unless job || exhausted?(worker_id)
  end

  job
end