Class: Puma::Cluster

# File 'lib/puma/cluster.rb', line 165

def all_workers_idle_timed_out?
  (@workers.map(&:pid) - idle_timed_out_worker_pids).empty?
end

# File 'lib/puma/cluster.rb', line 161

def all_workers_in_phase?
  @workers.all? { |w| w.phase == @phase }
end

# File 'lib/puma/cluster.rb', line 146

def next_worker_index
  occupied_positions = @workers.map(&:index)
  idx = 0
  idx += 1 until !occupied_positions.include?(idx)
  idx
end

# File 'lib/puma/cluster.rb', line 287

def preload?
  @options[:preload_app]
end

# File 'lib/puma/cluster.rb', line 263

def stats
  old_worker_count = @workers.count { |w| w.phase != @phase }
  worker_status = @workers.map do |w|
    {
      started_at: utc_iso8601(w.started_at),
      pid: w.pid,
      index: w.index,
      phase: w.phase,
      booted: w.booted?,
      last_checkin: utc_iso8601(w.last_checkin),
      last_status: w.last_status,
    }
  end

  {
    started_at: utc_iso8601(@started_at),
    workers: @workers.size,
    phase: @phase,
    booted_workers: worker_status.count { |w| w[:booted] },
    old_workers: old_worker_count,
    worker_status: worker_status,
  }.merge(super)
end

# File 'lib/puma/cluster.rb', line 30

attr_reader :workers

# File 'lib/puma/cluster.rb', line 136

def cull_start_index(diff)
  case @options[:worker_culling_strategy]
  when :oldest
    0
  else # :youngest
    -diff
  end
end

# File 'lib/puma/cluster.rb', line 112

def cull_workers
  diff = @workers.size - @options[:workers]
  return if diff < 1
  debug "Culling #{diff} workers"

  workers = workers_to_cull(diff)
  debug "Workers to cull: #{workers.inspect}"

  workers.each do |worker|
    log "- Worker #{worker.index} (PID: #{worker.pid}) terminating"
    worker.term
  end
end

# File 'lib/puma/cluster.rb', line 292

def fork_worker!
  if (worker = worker_at 0)
    worker.phase += 1
  end
  phased_restart(true)
end

# File 'lib/puma/cluster.rb', line 249

def halt
  @status = :halt
  wakeup!
end

# File 'lib/puma/cluster.rb', line 612

def idle_timed_out_worker_pids
  idle_workers.keys
end

# File 'lib/puma/cluster.rb', line 616

def idle_workers
  @idle_workers ||= {}
end

# File 'lib/puma/cluster.rb', line 228

def phased_restart(refork = false)
  return false if @options[:preload_app] && !refork

  @phased_restart = true
  wakeup!

  true
end

# File 'lib/puma/cluster.rb', line 59

def redirect_io
  super

  @workers.each { |x| x.hup }
end

# File 'lib/puma/cluster.rb', line 254

def reload_worker_directory
  dir = @launcher.restart_dir
  log "+ Changing to #{dir}"
  Dir.chdir dir
end

# File 'lib/puma/cluster.rb', line 223

def restart
  @restart = true
  stop
end

# File 'lib/puma/cluster.rb', line 352

def run
  @status = :run

  output_header "cluster"

  # This is aligned with the output from Runner, see Runner#output_header
  log "*      Workers: #{@options[:workers]}"

  if preload?
    # Threads explicitly marked as fork safe will be ignored. Used in Rails,
    # but may be used by anyone. Note that we need to explicit
    # Process::Waiter check here because there's a bug in Ruby 2.6 and below
    # where calling thread_variable_get on a Process::Waiter will segfault.
    # We can drop that clause once those versions of Ruby are no longer
    # supported.
    fork_safe = ->(t) { !t.is_a?(Process::Waiter) && t.thread_variable_get(:fork_safe) }

    before = Thread.list.reject(&fork_safe)

    log "*     Restarts: (\u2714) hot (\u2716) phased"
    log "* Preloading application"
    load_and_bind

    after = Thread.list.reject(&fork_safe)

    if after.size > before.size
      threads = (after - before)
      if threads.first.respond_to? :backtrace
        log "! WARNING: Detected #{after.size-before.size} Thread(s) started in app boot:"
        threads.each do |t|
          log "! #{t.inspect} - #{t.backtrace ? t.backtrace.first : ''}"
        end
      else
        log "! WARNING: Detected #{after.size-before.size} Thread(s) started in app boot"
      end
    end
  else
    log "*     Restarts: (\u2714) hot (\u2714) phased"

    unless @config.app_configured?
      error "No application configured, nothing to run"
      exit 1
    end

    @launcher.binder.parse @options[:binds]
  end

  read, @wakeup = Puma::Util.pipe

  setup_signals

  # Used by the workers to detect if the master process dies.
  # If select says that @check_pipe is ready, it's because the
  # master has exited and @suicide_pipe has been automatically
  # closed.
  #
  @check_pipe, @suicide_pipe = Puma::Util.pipe

  # Separate pipe used by worker 0 to receive commands to
  # fork new worker processes.
  @fork_pipe, @fork_writer = Puma::Util.pipe

  log "Use Ctrl-C to stop"

  single_worker_warning

  redirect_io

  Plugins.fire_background

  @launcher.write_state

  start_control

  @master_read, @worker_write = read, @wakeup

  @options[:worker_write] = @worker_write

  @config.run_hooks(:before_fork, nil, @log_writer)

  spawn_workers

  Signal.trap "SIGINT" do
    stop
  end

  begin
    booted = false
    in_phased_restart = false
    workers_not_booted = @options[:workers]

    while @status == :run
      begin
        if @options[:idle_timeout] && all_workers_idle_timed_out?
          log "- All workers reached idle timeout"
          break
        end

        if @phased_restart
          start_phased_restart
          @phased_restart = false
          in_phased_restart = true
          workers_not_booted = @options[:workers]
        end

        check_workers

        if read.wait_readable([0, @next_check - Time.now].max)
          req = read.read_nonblock(1)
          next unless req

          if req == PIPE_WAKEUP
            @next_check = Time.now
            next
          end

          result = read.gets
          pid = result.to_i

          if req == PIPE_BOOT || req == PIPE_FORK
            pid, idx = result.split(':').map(&:to_i)
            w = worker_at idx
            w.pid = pid if w.pid.nil?
          end

          if w = @workers.find { |x| x.pid == pid }
            case req
            when PIPE_BOOT
              w.boot!
              log "- Worker #{w.index} (PID: #{pid}) booted in #{w.uptime.round(2)}s, phase: #{w.phase}"
              @next_check = Time.now
              workers_not_booted -= 1
            when PIPE_EXTERNAL_TERM
              # external term, see worker method, Signal.trap "SIGTERM"
              w.term!
            when PIPE_TERM
              w.term unless w.term?
            when PIPE_PING
              status = result.sub(/^\d+/,'').chomp
              w.ping!(status)
              @events.fire(:ping!, w)

              if in_phased_restart && @options[:fork_worker] && workers_not_booted.positive? && w0 = worker_at(0)
                w0.ping!(status)
                @events.fire(:ping!, w0)
              end

              if !booted && @workers.none? {|worker| worker.last_status.empty?}
                @events.fire_on_booted!
                debug_loaded_extensions("Loaded Extensions - master:") if @log_writer.debug?
                booted = true
              end
            when PIPE_IDLE
              if idle_workers[pid]
                idle_workers.delete pid
              else
                idle_workers[pid] = true
              end
            end
          else
            log "! Out-of-sync worker list, no #{pid} worker"
          end
        end

        if in_phased_restart && workers_not_booted.zero?
          @events.fire_on_booted!
          debug_loaded_extensions("Loaded Extensions - master:") if @log_writer.debug?
          in_phased_restart = false
        end
      rescue Interrupt
        @status = :stop
      end
    end

    stop_workers unless @status == :halt
  ensure
    @check_pipe.close
    @suicide_pipe.close
    read.close
    @wakeup.close
  end
end

# File 'lib/puma/cluster.rb', line 301

def setup_signals
  if @options[:fork_worker]
    Signal.trap "SIGURG" do
      fork_worker!
    end

    # Auto-fork after the specified number of requests.
    if (fork_requests = @options[:fork_worker].to_i) > 0
      @events.register(:ping!) do |w|
        fork_worker! if w.index == 0 &&
          w.phase == 0 &&
          w.last_status[:requests_count] >= fork_requests
      end
    end
  end

  Signal.trap "SIGCHLD" do
    wakeup!
  end

  Signal.trap "TTIN" do
    @options[:workers] += 1
    wakeup!
  end

  Signal.trap "TTOU" do
    @options[:workers] -= 1 if @options[:workers] >= 2
    wakeup!
  end

  master_pid = Process.pid

  Signal.trap "SIGTERM" do
    # The worker installs their own SIGTERM when booted.
    # Until then, this is run by the worker and the worker
    # should just exit if they get it.
    if Process.pid != master_pid
      log "Early termination of worker"
      exit! 0
    else
      @launcher.close_binder_listeners

      stop_workers
      stop
      @events.fire_on_stopped!
      raise(SignalException, "SIGTERM") if @options[:raise_exception_on_sigterm]
      exit 0 # Clean exit, workers were stopped
    end
  end
end

# File 'lib/puma/cluster.rb', line 537

def single_worker_warning
  return if @options[:workers] != 1 || @options[:silence_single_worker_warning]

  log "! WARNING: Detected running cluster mode with 1 worker."
  log "! Running Puma in cluster mode with a single worker is often a misconfiguration."
  log "! Consider running Puma in single-mode (workers = 0) in order to reduce memory overhead."
  log "! Set the `silence_single_worker_warning` option to silence this warning message."
end

# File 'lib/puma/cluster.rb', line 98

def spawn_worker(idx, master)
  @config.run_hooks(:before_worker_fork, idx, @log_writer)

  pid = fork { worker(idx, master) }
  if !pid
    log "! Complete inability to spawn new workers detected"
    log "! Seppuku is the only choice."
    exit! 1
  end

  @config.run_hooks(:after_worker_fork, idx, @log_writer)
  pid
end

# File 'lib/puma/cluster.rb', line 65

def spawn_workers
  diff = @options[:workers] - @workers.size
  return if diff < 1

  master = Process.pid
  if @options[:fork_worker]
    @fork_writer << "-1\n"
  end

  diff.times do
    idx = next_worker_index

    if @options[:fork_worker] && idx != 0
      @fork_writer << "#{idx}\n"
      pid = nil
    else
      pid = spawn_worker(idx, master)
    end

    debug "Spawned worker: #{pid}"
    @workers << WorkerHandle.new(idx, pid, @phase, @options)
  end

  if @options[:fork_worker] && all_workers_in_phase?
    @fork_writer << "0\n"

    if worker_at(0).phase > 0
      @fork_writer << "-2\n"
    end
  end
end

# File 'lib/puma/cluster.rb', line 47

def start_phased_restart
  @events.fire_on_restart!
  @phase += 1
  log "- Starting phased worker restart, phase: #{@phase}"

  # Be sure to change the directory again before loading
  # the app. This way we can pick up new code.
  dir = @launcher.restart_dir
  log "+ Changing to #{dir}"
  Dir.chdir dir
end

# File 'lib/puma/cluster.rb', line 237

def stop
  @status = :stop
  wakeup!
end

# File 'lib/puma/cluster.rb', line 242

def stop_blocked
  @status = :stop if @status == :run
  wakeup!
  @control&.stop true
  Process.waitall
end

# File 'lib/puma/cluster.rb', line 32

def stop_workers
  log "- Gracefully shutting down workers..."
  @workers.each { |x| x.term }

  begin
    loop do
      wait_workers
      break if @workers.reject {|w| w.pid.nil?}.empty?
      sleep 0.2
    end
  rescue Interrupt
    log "! Cancelled waiting for workers"
  end
end

# File 'lib/puma/cluster.rb', line 598

def timeout_workers
  @workers.each do |w|
    if !w.term? && w.ping_timeout <= Time.now
      details = if w.booted?
                  "(Worker #{w.index} failed to check in within #{@options[:worker_timeout]} seconds)"
                else
                  "(Worker #{w.index} failed to boot within #{@options[:worker_boot_timeout]} seconds)"
                end
      log "! Terminating timed out worker #{details}: #{w.pid}"
      w.kill
    end
  end
end

# File 'lib/puma/cluster.rb', line 548

def wait_workers
  # Reap all children, known workers or otherwise.
  # If puma has PID 1, as it's common in containerized environments,
  # then it's responsible for reaping orphaned processes, so we must reap
  # all our dead children, regardless of whether they are workers we spawned
  # or some reattached processes.
  reaped_children = {}
  loop do
    begin
      pid, status = Process.wait2(-1, Process::WNOHANG)
      break unless pid
      reaped_children[pid] = status
    rescue Errno::ECHILD
      break
    end
  end

  @workers.reject! do |w|
    next false if w.pid.nil?
    begin
      # We may need to check the PID individually because:
      # 1. From Ruby versions 2.6 to 3.2, `Process.detach` can prevent or delay
      #    `Process.wait2(-1)` from detecting a terminated process: https://bugs.ruby-lang.org/issues/19837.
      # 2. When `fork_worker` is enabled, some worker may not be direct children,
      #    but grand children.  Because of this they won't be reaped by `Process.wait2(-1)`.
      if reaped_children.delete(w.pid) || Process.wait(w.pid, Process::WNOHANG)
        true
      else
        w.term if w.term?
        nil
      end
    rescue Errno::ECHILD
      begin
        Process.kill(0, w.pid)
        # child still alive but has another parent (e.g., using fork_worker)
        w.term if w.term?
        false
      rescue Errno::ESRCH, Errno::EPERM
        true # child is already terminated
      end
    end
  end

  # Log unknown children
  reaped_children.each do |pid, status|
    log "! reaped unknown child process pid=#{pid} status=#{status}"
  end
end

# File 'lib/puma/cluster.rb', line 201

def worker(index, master)
  @workers = []

  @master_read.close
  @suicide_pipe.close
  @fork_writer.close

  pipes = { check_pipe: @check_pipe, worker_write: @worker_write }
  if @options[:fork_worker]
    pipes[:fork_pipe] = @fork_pipe
    pipes[:wakeup] = @wakeup
  end

  server = start_server if preload?
  new_worker = Worker.new index: index,
                          master: master,
                          launcher: @launcher,
                          pipes: pipes,
                          server: server
  new_worker.run
end

# File 'lib/puma/cluster.rb', line 153

def worker_at(idx)
  @workers.find { |w| w.index == idx }
end

# File 'lib/puma/cluster.rb', line 126

def workers_to_cull(diff)
  workers = @workers.sort_by(&:started_at)

  # In fork_worker mode, worker 0 acts as our master process.
  # We should avoid culling it to preserve copy-on-write memory gains.
  workers.reject! { |w| w.index == 0 } if @options[:fork_worker]

  workers[cull_start_index(diff), diff]
end