class Concurrent::RubyThreadPoolExecutor::Worker
@!visibility private
def <<(message)
def <<(message) @queue << message end
def create_worker(queue, pool, idletime)
def create_worker(queue, pool, idletime) Thread.new(queue, pool, idletime) do |my_queue, my_pool, my_idletime| last_message = Concurrent.monotonic_time catch(:stop) do loop do case message = my_queue.pop when :idle_test if (Concurrent.monotonic_time - last_message) > my_idletime my_pool.remove_busy_worker(self) throw :stop else my_pool.worker_not_old_enough(self) end when :stop my_pool.remove_busy_worker(self) throw :stop else task, args = message run_task my_pool, task, args last_message = Concurrent.monotonic_time my_pool.ready_worker(self) end end end end end
def initialize(pool)
def initialize(pool) # 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 end
def kill
def kill @thread.kill end
def run_task(pool, task, args)
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
def stop
def stop @queue << :stop end