Class: Concurrent::RubyThreadPoolExecutor

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

attr_reader :idletime

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

attr_reader :max_length

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

attr_reader :max_queue

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

attr_reader :min_length

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

def ns_limited_queue?
  @max_queue != 0
end

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

attr_reader :synchronous

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

def completed_task_count
  synchronize { @completed_task_count }
end

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

def largest_length
  synchronize { @largest_length }
end

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

def length
  synchronize { @pool.length }
end

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

def ns_add_busy_worker
  return if @pool.size >= @max_length

  @workers_counter += 1
  @pool << (worker = Worker.new(self, @workers_counter))
  @largest_length = @pool.length if @pool.length > @largest_length
  worker
end

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

def ns_assign_worker(*args, &task)
  # keep growing if the pool is not at the minimum yet
  worker, _ = (@ready.pop if @pool.size >= @min_length) || ns_add_busy_worker
  if worker
    worker << [task, args]
    true
  else
    false
  end
rescue ThreadError
  # Raised when the operating system refuses to create the new thread
  return false
end

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

def ns_enqueue(*args, &task)
  return false if @synchronous

  if !ns_limited_queue? || @queue.size < @max_queue
    @queue << [task, args]
    true
  else
    false
  end
end

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

def ns_execute(*args, &task)
  ns_reset_if_forked

  if ns_assign_worker(*args, &task) || ns_enqueue(*args, &task)
    @scheduled_task_count += 1
    nil
  else
    fallback_action(*args, &task)
  end
end

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

def ns_initialize(opts)
  @min_length      = opts.fetch(:min_threads, DEFAULT_MIN_POOL_SIZE).to_i
  @max_length      = opts.fetch(:max_threads, DEFAULT_MAX_POOL_SIZE).to_i
  @idletime        = opts.fetch(:idletime, DEFAULT_THREAD_IDLETIMEOUT).to_i
  @max_queue       = opts.fetch(:max_queue, DEFAULT_MAX_QUEUE_SIZE).to_i
  @synchronous     = opts.fetch(:synchronous, DEFAULT_SYNCHRONOUS)
  @fallback_policy = opts.fetch(:fallback_policy, :abort)

  raise ArgumentError.new("`synchronous` cannot be set unless `max_queue` is 0") if @synchronous && @max_queue > 0
  raise ArgumentError.new("#{@fallback_policy} is not a valid fallback policy") unless FALLBACK_POLICIES.include?(@fallback_policy)
  raise ArgumentError.new("`max_threads` cannot be less than #{DEFAULT_MIN_POOL_SIZE}") if @max_length < DEFAULT_MIN_POOL_SIZE
  raise ArgumentError.new("`max_threads` cannot be greater than #{DEFAULT_MAX_POOL_SIZE}") if @max_length > DEFAULT_MAX_POOL_SIZE
  raise ArgumentError.new("`min_threads` cannot be less than #{DEFAULT_MIN_POOL_SIZE}") if @min_length < DEFAULT_MIN_POOL_SIZE
  raise ArgumentError.new("`min_threads` cannot be more than `max_threads`") if min_length > max_length

  @pool                 = [] # all workers
  @ready                = [] # used as a stash (most idle worker is at the start)
  @queue                = [] # used as queue
  # @ready or @queue is empty at all times
  @scheduled_task_count = 0
  @completed_task_count = 0
  @largest_length       = 0
  @workers_counter      = 0
  @ruby_pid             = $$ # detects if Ruby has forked
end

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

def ns_kill_execution
  # TODO log out unprocessed tasks in queue
  # TODO try to shutdown first?
  @pool.each(&:kill)
  @pool.clear
  @ready.clear
end

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

def ns_prunable_capacity
  if running?
    [@pool.size - @min_length, @ready.size].min
  else
    @pool.size
  end
end

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

def ns_ready_worker(worker, last_message, success = true)
  task_and_args = @queue.shift
  if task_and_args
    worker << task_and_args
  else
    # stop workers when !running?, do not return them to @ready
    if running?
      raise unless last_message
      @ready.push([worker, last_message])
    else
      worker.stop
    end
  end
end

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

def ns_remove_busy_worker(worker)
  @pool.delete(worker)
  stopped_event.set if @pool.empty? && !running?
  true
end

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

def ns_remove_ready_worker(worker)
  if index = @ready.index { |rw, _| rw == worker }
    @ready.delete_at(index)
  end
  true
end

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

def ns_reset_if_forked
  if $$ != @ruby_pid
    @queue.clear
    @ready.clear
    @pool.clear
    @scheduled_task_count = 0
    @completed_task_count = 0
    @largest_length       = 0
    @workers_counter      = 0
    @ruby_pid             = $$
  end
end

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

def ns_shutdown_execution
  ns_reset_if_forked

  if @pool.empty?
    # nothing to do
    stopped_event.set
  end

  if @queue.empty?
    # no more tasks will be accepted, just stop all workers
    @pool.each(&:stop)
  end
end

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

def ns_worker_died(worker)
  ns_remove_busy_worker worker
  replacement_worker = ns_add_busy_worker
  ns_ready_worker replacement_worker, Concurrent.monotonic_time, false if replacement_worker
end

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

def prune_pool
  deprecated "#prune_pool has no effect and will be removed in next the release, see https://github.com/ruby-concurrency/concurrent-ruby/pull/1082."
end

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

def prune_worker(worker)
  synchronize do
    if ns_prunable_capacity > 0
      remove_worker worker
      true
    else
      false
    end
  end
end

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

def queue_length
  synchronize { @queue.length }
end

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

def ready_worker(worker, last_message)
  synchronize { ns_ready_worker worker, last_message }
end

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

def remaining_capacity
  synchronize do
    if ns_limited_queue?
      @max_queue - @queue.length
    else
      -1
    end
  end
end

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

def remove_worker(worker)
  synchronize do
    ns_remove_ready_worker worker
    ns_remove_busy_worker worker
  end
end

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

def scheduled_task_count
  synchronize { @scheduled_task_count }
end

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

def worker_died(worker)
  synchronize { ns_worker_died worker }
end

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

def worker_task_completed
  synchronize { @completed_task_count += 1 }
end