123456789_123456789_123456789_123456789_123456789_

Class: Concurrent::RubyThreadPoolExecutor::Worker

Relationships & Source Files
Super Chains via Extension / Inclusion / Inheritance
Instance Chain:
self, Concern::Logging
Inherits: Object
Defined in: lib/concurrent-ruby/concurrent/executor/ruby_thread_pool_executor.rb

Constant Summary

Concern::Logging - Included

SEV_LABEL

Class Method Summary

Instance Method Summary

Concern::Logging - Included

#log

Logs through Concurrent.global_logger, it can be overridden by setting @logger.

Constructor Details

.new(pool, id) ⇒ Worker

[ GitHub ] 

  


# File 'lib/concurrent-ruby/concurrent/executor/ruby_thread_pool_executor.rb', line 313



def initialize(pool, id)
  # instance variables accessed only under pool's lock so no need to sync here again
  @queue  = Queue.new
  @pool   = pool
  @thread = create_worker @queue, pool, pool.idletime

  if @thread.respond_to?(:name=)
    @thread.name = [pool.name, 'worker', id].compact.join('-')
  end
end


  







Instance Method Details



  

    #<<(message)  
  

  [ GitHub ]


  

  


# File 'lib/concurrent-ruby/concurrent/executor/ruby_thread_pool_executor.rb', line 324



def <<(message)
  @queue << message
end


  








  

    #create_worker(queue, pool, idletime)   (private)
  

  [ GitHub ]


  

  


# File 'lib/concurrent-ruby/concurrent/executor/ruby_thread_pool_executor.rb', line 338



def create_worker(queue, pool, idletime)
  Thread.new(queue, pool, idletime) do |my_queue, my_pool, my_idletime|
    catch(:stop) do
      loop do

        case message = my_queue.pop
        when :stop
          my_pool.remove_busy_worker(self)
          throw :stop

        else
          task, args = message
          run_task my_pool, task, args
          my_pool.ready_worker(self, Concurrent.monotonic_time)
        end
      end
    end
  end
end


  








  

    #kill  
  

  [ GitHub ]


  

  


# File 'lib/concurrent-ruby/concurrent/executor/ruby_thread_pool_executor.rb', line 332



def kill
  @thread.kill
end


  








  

    #run_task(pool, task, args)   (private)
  

  [ GitHub ]


  

  


# File 'lib/concurrent-ruby/concurrent/executor/ruby_thread_pool_executor.rb', line 358



def run_task(pool, task, args)
  task.call(*args)
  pool.worker_task_completed
rescue => ex
  # let it fail
  log DEBUG, ex
rescue Exception => ex
  log ERROR, ex
  pool.worker_died(self)
  throw :stop
end


  








  

    #stop  
  

  [ GitHub ]


  

  


# File 'lib/concurrent-ruby/concurrent/executor/ruby_thread_pool_executor.rb', line 328



def stop
  @queue << :stop
end