123456789_123456789_123456789_123456789_123456789_

Class: ActiveSupport::Testing::Parallelization::Server

Relationships & Source Files
Super Chains via Extension / Inclusion / Inheritance
Instance Chain:
self, DRb::DRbUndumped
Inherits: Object
Defined in: activesupport/lib/active_support/testing/parallelization/server.rb

Class Method Summary

Instance Attribute Summary

Instance Method Summary

Constructor Details

.new(distributor:) ⇒ Server

[ GitHub ] 

  


# File 'activesupport/lib/active_support/testing/parallelization/server.rb', line 13



def initialize(distributor:)
  @distributor = distributor
  @active_workers = Concurrent::Map.new
  @worker_pids = Concurrent::Map.new
  @in_flight = Concurrent::Map.new
end


  







Instance Attribute Details



  

    #active_workers?  ⇒ Boolean  (readonly)
  

  [ GitHub ]


  

  


# File 'activesupport/lib/active_support/testing/parallelization/server.rb', line 63



def active_workers?
  @active_workers.size > 0
end


  







Instance Method Details



  

    #<<(o)  
  

  [ GitHub ]


  

  


# File 'activesupport/lib/active_support/testing/parallelization/server.rb', line 31



def <<(o)
  o[2] = DRbObject.new(o[2]) if o
  @distributor.add_test(o)
end


  








  

    #interrupt  
  

  [ GitHub ]


  

  


# File 'activesupport/lib/active_support/testing/parallelization/server.rb', line 67



def interrupt
  @distributor.interrupt
end


  








  

    #pop(worker_id)  
  

  [ GitHub ]


  

  


# File 'activesupport/lib/active_support/testing/parallelization/server.rb', line 36



def pop(worker_id)
  if test = @distributor.take(worker_id: worker_id)
    @in_flight[[test[0].to_s, test[1]]] = test
  end

  test
end


  








  

    #record(reporter, result)  
  



  

    
  





  
Raises:


    
  
  • 
    (DRb::DRbConnError)
  
    • 





  [ GitHub ]


  

  


# File 'activesupport/lib/active_support/testing/parallelization/server.rb', line 20



def record(reporter, result)
  raise DRb::DRbConnError if result.is_a?(DRb::DRbUnknown)

  @in_flight.delete([result.klass, result.name])

  reporter.synchronize do
    reporter.prerecord(PrerecordResultClass.new(result.klass), result.name)
    reporter.record(result)
  end
end


  








  

    #remove_dead_workers(dead_pids)  
  

  [ GitHub ]


  

  


# File 'activesupport/lib/active_support/testing/parallelization/server.rb', line 54



def remove_dead_workers(dead_pids)
  dead_pids.each do |dead_pid|
    if worker_id = @worker_pids.key(dead_pid)
      @active_workers.delete(worker_id)
      @worker_pids.delete(worker_id)
    end
  end
end


  








  

    #shutdown  
  

  [ GitHub ]


  

  


# File 'activesupport/lib/active_support/testing/parallelization/server.rb', line 71



def shutdown
  # Wait for initial queue to drain
  while @distributor.pending?
    sleep 0.1
  end

  @distributor.close

  wait_for_active_workers

  @in_flight.values.each do |(klass, name, reporter)|
    result = Minitest::Result.from(klass.new(name))
    error = RuntimeError.new("result not reported")
    error.set_backtrace([""])
    result.failures << Minitest::UnexpectedError.new(error)
    reporter.synchronize do
      reporter.record(result)
    end
  end
rescue Interrupt
  warn "Interrupted. Exiting..."

  @distributor.close

  wait_for_active_workers
end


  








  

    #start_worker(worker_id, worker_pid)  
  

  [ GitHub ]


  

  


# File 'activesupport/lib/active_support/testing/parallelization/server.rb', line 44



def start_worker(worker_id, worker_pid)
  @active_workers[worker_id] = true
  @worker_pids[worker_id] = worker_pid
end


  








  

    #stop_worker(worker_id, worker_pid)  
  

  [ GitHub ]


  

  


# File 'activesupport/lib/active_support/testing/parallelization/server.rb', line 49



def stop_worker(worker_id, worker_pid)
  @active_workers.delete(worker_id)
  @worker_pids.delete(worker_id)
end


  








  

    #wait_for_active_workers   (private)
  

  [ GitHub ]


  

  


# File 'activesupport/lib/active_support/testing/parallelization/server.rb', line 99



def wait_for_active_workers
  while active_workers?
    sleep 0.1
  end
end