forking_executor.rb

##
# Parallel test Runner for Rails
#
# This is a spike implementation for multi-process parallel testing with Rails.
# Only works with SQLite3 right now, and doesn't clean up.
#
# Here is an example of how to use it:
#
# ```ruby
#   require 'forking_executor'
#   require 'test_helper'
#   
#   Minitest.parallel_executor = ForkingExecutor.new 2
#   
#   class UserTest < ActiveSupport::TestCase
#     parallelize_me!
#   
#     10.times do |x|
#       define_method :"test_#{x}" do
#         name = "foo-#{rand}"
#         u = User.create! name: name
#         assert_equal name, u.name
#         sleep 1
#       end
#     end
#   end
# ```

require 'drb'
require 'drb/unix'
require 'tempfile'

class ForkingExecutor
  class Server
    include DRb::DRbUndumped

    def initialize
      @queue = Queue.new
    end

    def record reporter, result
      reporter.synchronize { reporter.record result }
    end
    def << o; @queue << o; end
    def pop; @queue.pop; end
  end

  def initialize size
    @size  = size
    @queue = Server.new
    file   = File.join Dir.tmpdir, Dir::Tmpname.make_tmpname('tests', 'fd')
    @url   = "drbunix://#{file}"
    DRb.start_service @url, @queue
    @pool  = []
  end

  def start
    primary_pool = ActiveRecord::Base.connection_handler.retrieve_connection_pool "primary"
    connection_spec = primary_pool.spec.to_hash
    @pool = @size.times.map { |i|
      fork {
        # Setting up new connections is going to be different per app and per
        # database.  Rather than hardcode the block below, we should configure
        # the ForkingExecutor with a strategy object and call that
        connection_spec["database"] = "db/OMGLOL-#{i}.sqlite3"
        ActiveRecord::Base.establish_connection connection_spec
        if ActiveRecord::Migrator.needs_migration?
          old, ENV["VERBOSE"] = ENV["VERBOSE"], "false"
          ActiveRecord::Tasks::DatabaseTasks.migrate
          ENV["VERBOSE"] = old
        end

        DRb.stop_service
        queue = DRbObject.new_with_uri @url
        while job = queue.pop
          klass    = job[0]
          method   = job[1]
          reporter = job[2]
          result = Minitest.run_one_method klass, method
          queue.record reporter, result
        end

        # We should also call the strategy object here for cleanup.
      }
    }
  end

  def << work
    @queue << work
  end

  def shutdown
    @size.times { @queue << nil }
    @pool.each { |pid| Process.waitpid pid }
  end
end