Load testing example (producer->consumer pattern)

load_test.rb

class LoadTest
  def initialize(time:, rate:, pool:)
    @time = time
    @rate = rate
    @pool = pool
    @requests = []
    @started = false
    @shutdown = false
    @lock = Mutex.new
    @cond = ConditionVariable.new
    @queue = Queue.new
  end

  def start!(producer, consumer)
    return if @started
    @started = true
    @producer = Thread.new { run_producer(producer) }
    @consumers = @pool.times.map { |i| Thread.new(i) { |x| run_consumer(x, consumer) } }
  end

  def rate=(rate)
    @lock.synchronize { @rate = rate }
  end
  
  def time=(time)
    @lock.synchronize { @time = time }
  end

  def shutdown!
    @shutdown = true
    @pool.times { @queue << :shutdown }
  end

  def run_producer(producer)
    loop do
      if @shutdown
        puts "[producer] shutting down"
        break
      end

      @lock.synchronize do
        @requests.reject! { |v| v < Time.now - @time }
        if @requests.size < @rate
          @queue << producer.call
          @cond.wait(@lock)
        else
          sleep(0.1)
        end
      end
    end
  end

  def run_consumer(tname, consumer)
    loop do
      job = @queue.pop
      if job == :shutdown
        puts "[consumer:#{tname}] shutting down"
        break
      end
      consumer.call(job)
      @lock.synchronize do
        now = Time.now
        puts "[consumer:#{tname}] #{Time.now}"
        @requests << now
        @cond.signal
      end
    end
  end
end

runner.rb

test = LoadTest.new(time: 1, rate: 21, pool: 5)

data = Customer.limit(100).pluck(:id)
producer = -> { data.sample }
consumer = ->(id) { puts "Got customer: #{id}" }

test.start!(producer, consumer)
test.rate = 30 # increase the rate
test.shutdown!