Class: Raptor::Cluster

Inherits:
Object
  • Object
show all
Defined in:
lib/raptor/cluster.rb,
sig/generated/raptor/cluster.rbs

Overview

Forks and supervises worker processes. Handles graceful shutdown on INT and TERM, phased restart on USR1, hot restart on USR2, and (on Linux) refork on URG. Restarts workers that exit unexpectedly or stop checking in.

Constant Summary collapse

INHERITED_FDS_ENV =

Returns:

  • (::String) 
"RAPTOR_INHERITED_FDS"
HTTP1_RACTOR_COUNT_CAP =

Returns:

  • (::Integer) 
3
HTTP2_RACTOR_COUNT_CAP =

Returns:

  • (::Integer) 
2

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(options) ⇒ Cluster

Creates a new Cluster with the given options.

RBS:

  • (Hash[Symbol, untyped] options) -> void

Parameters:

  • options (Hash) —

    cluster configuration options

Options Hash (options):

  • :binds (Array<String>) —

    array of bind URIs

  • :socket_backlog (Integer) —

    kernel listen() queue depth for TCP/SSL listeners

  • :drain_accept_queue (Boolean) —

    whether to drain the kernel accept queue on shutdown

  • :workers (Integer) —

    number of worker processes

  • :threads (Integer) —

    number of threads per worker process

  • :app (#call) —

    pre-built Rack application

  • :rackup (String) —

    path to Rack configuration file

  • :chdir (String, nil) —

    directory to change to before loading the Rack application, or nil to leave the working directory unchanged

  • :environment (String, nil) —

    Raptor's application environment label; falls back to $RAILS_ENV, then $RACK_ENV, then "development"

  • :connection (Hash) —

    per-connection settings shared across protocols

  • :http1 (Hash) —

    HTTP/1.1-specific settings

  • :http2 (Hash) —

    HTTP/2-specific settings

  • :worker_boot_timeout (Integer) —

    seconds to wait for a worker to finish booting before killing it

  • :worker_timeout (Integer) —

    seconds to wait for a booted worker to check in before killing it

  • :worker_drain_timeout (Integer) —

    seconds a worker waits for in-flight requests during shutdown before force-killing app threads

  • :worker_shutdown_timeout (Integer) —

    seconds to wait for graceful worker exit before force-killing

  • :refork_after (Integer, Array<Integer>, nil) —

    request-count threshold(s) at which a warm worker is promoted to a fork source for phased refork; nil or 0 disables. Requires PR_SET_CHILD_SUBREAPER (Linux)

  • :before_fork (Array<Proc>) —

    procs called in the master before every worker fork

  • :before_worker_boot (Array<Proc>) —

    procs called in each worker before it begins serving

  • :before_worker_shutdown (Array<Proc>) —

    procs called in each worker before its graceful shutdown

  • :before_refork (Array<Proc>) —

    procs called in a worker before it transitions to a seed

  • :stats_file (String, nil) —

    path to write per-worker stats JSON, or nil to disable

  • :pid_file (String, nil) —

    path to write the master PID to, or nil to disable

  • :stdout_file (String, nil) —

    path to redirect stdout to, reopened on SIGHUP, or nil to disable

  • :stderr_file (String, nil) —

    path to redirect stderr to, reopened on SIGHUP, or nil to disable

  • :access_log_file (String, nil) —

    path to write Common Log Format access logs to, reopened on SIGHUP, or nil to disable

  • :launch_command (String, nil) —

    path of the program to re-exec on hot restart, or nil to disable

  • :launch_argv (Array<String>, nil) —

    command-line arguments for the hot-restart exec, or nil to disable

  • :on_error (#call, nil) —

    callback invoked with (env, exception) when the Rack app raises



155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
 
# File 'lib/raptor/cluster.rb', line 155

def initialize(options)
  @drain_accept_queue = options[:drain_accept_queue]
  @worker_count = options[:workers]
  @http1_ractor_count = options[:http1][:ractors] || self.class.default_http1_ractor_count(@worker_count)
  @http2_ractor_count = options[:http2][:ractors] || self.class.default_http2_ractor_count(@worker_count)
  @thread_count = options[:threads]
  @environment = options[:environment] || ENV["RAILS_ENV"] || ENV["RACK_ENV"] || "development"
  @connection_options = options[:connection]
  @http1_options = options[:http1]
  @http2_options = options[:http2]
  @worker_boot_timeout = options[:worker_boot_timeout]
  @worker_timeout = options[:worker_timeout]
  @worker_drain_timeout = options[:worker_drain_timeout]
  @worker_shutdown_timeout = options[:worker_shutdown_timeout]
  @before_fork = Array(options[:before_fork])
  @before_worker_boot = Array(options[:before_worker_boot])
  @before_worker_shutdown = Array(options[:before_worker_shutdown])
  @before_refork = Array(options[:before_refork])
  @stats_file = options[:stats_file]
  @pid_file = options[:pid_file]
  @stdout_file = options[:stdout_file]
  @stderr_file = options[:stderr_file]
  @access_log_file = options[:access_log_file]
  @access_log_io = nil
  @launch_command = options[:launch_command]
  @launch_argv = options[:launch_argv]
  @on_error = options[:on_error]

  Dir.chdir(options[:chdir]) if options[:chdir]

  inherited_fds = if raw = ENV.delete(INHERITED_FDS_ENV)
    JSON.parse(raw)
  elsif (systemd_fds = Systemd.listen_fds).any?
    Systemd.clear_listen_env
    pair_systemd_fds(options[:binds], systemd_fds)
  else
    {}
  end
  @binder = Binder.new(options[:binds], socket_backlog: options[:socket_backlog], inherited_fds: inherited_fds)
  @server_port = @binder.server_port
  @app = options[:app] || Rack::Builder.parse_file(options[:rackup])
  log_initialization

  @shutdown = false
  @workers = {}
  @timed_out = Set.new
  @stats = Stats.new(@worker_count)
  @phase = 0
  @phased_restart_requested = false
  @phased_restarting = false
  @hot_restart_requested = false

  @refork_thresholds = normalize_refork_thresholds(options[:refork_after])
  @refork_requested = false
  @refork_threshold_idx = 0
  @seed_pid = nil
  @seed_ready = false
  @seed_vacated_index = nil
end

Class Method Details

.default_http1_ractor_count(worker_count, cores: Integer(Concurrent.available_processor_count)) ⇒ Integer

Returns the default number of HTTP/1.1 pipeline ractors per worker for the given worker count, rounded from the available processor count per worker and clamped to [1, HTTP1_RACTOR_COUNT_CAP].

RBS:

  • (Integer worker_count, ?cores: Integer) -> Integer

Parameters:

  • worker_count (Integer) —

    the configured worker count

  • cores (Integer) (defaults to: Integer(Concurrent.available_processor_count)) —

    the available processor count

  • cores: (Integer) (defaults to: Integer(Concurrent.available_processor_count))

Returns:

  • (Integer) 


53
54
55
 
# File 'lib/raptor/cluster.rb', line 53

def self.default_http1_ractor_count(worker_count, cores: Integer(Concurrent.available_processor_count))
  (cores.to_f / worker_count).round.clamp(1, HTTP1_RACTOR_COUNT_CAP)
end

.default_http2_ractor_count(worker_count, cores: Integer(Concurrent.available_processor_count)) ⇒ Integer

Returns the default number of HTTP/2 pipeline ractors per worker for the given worker count, rounded from the available processor count per worker and clamped to [1, HTTP2_RACTOR_COUNT_CAP].

RBS:

  • (Integer worker_count, ?cores: Integer) -> Integer

Parameters:

  • worker_count (Integer) —

    the configured worker count

  • cores (Integer) (defaults to: Integer(Concurrent.available_processor_count)) —

    the available processor count

  • cores: (Integer) (defaults to: Integer(Concurrent.available_processor_count))

Returns:

  • (Integer) 


67
68
69
 
# File 'lib/raptor/cluster.rb', line 67

def self.default_http2_ractor_count(worker_count, cores: Integer(Concurrent.available_processor_count))
  (cores.to_f / worker_count).round.clamp(1, HTTP2_RACTOR_COUNT_CAP)
end

.run(options) ⇒ void

This method returns an undefined value.

Creates and runs a cluster with the given options.

RBS:

  • (Hash[Symbol, untyped] options) -> void

Parameters:

  • options (Hash) —

    cluster configuration options



39
40
41
 
# File 'lib/raptor/cluster.rb', line 39

def self.run(options)
  new(options).run
end

Instance Method Details

#check_refork_trigger ⇒ void

This method returns an undefined value.

Promotes the most-experienced worker to a seed and starts a phased refork when the next refork_after threshold is crossed or a manual refork was requested via SIGURG.

RBS:

  • () -> void



440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
 
# File 'lib/raptor/cluster.rb', line 440

def check_refork_trigger
  candidate = pick_refork_candidate
  return unless candidate

  candidate_index, candidate_requests = candidate
  threshold = @refork_thresholds[@refork_threshold_idx]

  if @refork_requested
    @refork_requested = false
  elsif !threshold || candidate_requests < threshold
    return
  else
    @refork_threshold_idx += 1
  end

  promote_worker_to_seed(candidate_index)
end

#drain_thread_pool(thread_pool) ⇒ void

This method returns an undefined value.

Shuts down the worker's application thread pool, force-killing the underlying threads if in-flight requests have not finished within worker_drain_timeout seconds.

RBS:

  • (AtomicThreadPool thread_pool) -> void

Parameters:

  • thread_pool (AtomicThreadPool) —

    the worker's thread pool



923
924
925
926
927
928
929
930
931
932
933
934
 
# File 'lib/raptor/cluster.rb', line 923

def drain_thread_pool(thread_pool)
  drain = Thread.new do
    Thread.current.name = "Thread Pool Drain"

    thread_pool.shutdown
  end
  return if drain.join(@worker_drain_timeout)

  Log.warn "Force-killing in-flight app threads after #{@worker_drain_timeout}s drain timeout"
  thread_pool.instance_variable_get(:@threads).each(&:kill)
  drain.join
end

#exit_description(status) ⇒ String

Returns a human-readable description of how a process exited.

RBS:

  • (Process::Status status) -> String

Parameters:

  • status (Process::Status) —

    the exit status of the process

Returns:

  • (String) —

    a description of the exit reason



942
943
944
945
946
947
948
949
950
 
# File 'lib/raptor/cluster.rb', line 942

def exit_description(status)
  if status.exited?
    "exited with code #{status.exitstatus}"
  elsif status.signaled?
    "killed by SIG#{Signal.signame(status.termsig)}"
  else
    "exited"
  end
end

#fill_vacated_seed_slot ⇒ Integer?

Spawns a replacement worker for the slot the seed vacated when it was promoted, returning the slot index it filled.

RBS:

  • () -> Integer?

Returns:

  • (Integer, nil) —

    the filled slot index, or nil if none was vacated



667
668
669
670
671
672
673
674
675
676
 
# File 'lib/raptor/cluster.rb', line 667

def fill_vacated_seed_slot
  index = @seed_vacated_index
  return unless index

  @seed_vacated_index = nil
  return if @shutdown

  spawn_worker(index)
  index
end

#log_initialization ⇒ void

This method returns an undefined value.

Prints the cluster's startup banner showing process structure and bind addresses.

RBS:

  • () -> void



969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
 
# File 'lib/raptor/cluster.rb', line 969

def log_initialization
  Log.info "Cluster initializing:"
  Log.info "├─ Version: #{VERSION}"
  Log.info "├─ Ruby Version: #{RUBY_DESCRIPTION}"
  Log.info "├─ Environment: #{@environment}"
  Log.info "├─ Master PID: #{Process.pid}"
  Log.info "│  └─ #{@worker_count} worker process#{"es" if @worker_count > 1}"
  Log.info "│     ├─ 1 server thread"
  Log.info "│     ├─ 1 reactor thread"
  Log.info "│     ├─ #{@http1_ractor_count} HTTP/1.1 pipeline ractor#{"s" if @http1_ractor_count > 1}"
  Log.info "│     ├─ #{@http2_ractor_count} HTTP/2 pipeline ractor#{"s" if @http2_ractor_count > 1}"
  Log.info "│     ├─ 2 pipeline collector threads"
  Log.info "│     ├─ #{@thread_count} worker thread#{"s" if @thread_count > 1}"
  Log.info "│     └─ 1 stats thread"
  Log.info "└─ Listening on #{@binder.addresses.join(", ")}"
end

#normalize_refork_thresholds(value) ⇒ Array<Integer>

Normalises the refork_after option into a sorted array of positive thresholds. Accepts nil, 0, an Integer, or an Array; anything else falls back to an empty array (feature disabled).

RBS:

  • (untyped value) -> Array[Integer]

Parameters:

  • value (Integer, Array<Integer>, nil) —

    the raw option value

Returns:

  • (Array<Integer>) 


319
320
321
322
323
324
325
326
327
328
 
# File 'lib/raptor/cluster.rb', line 319

def normalize_refork_thresholds(value)
  case value
  when Integer
    value.positive? ? [value].freeze : [].freeze
  when Array
    value.select { |threshold| threshold.is_a?(Integer) && threshold.positive? }.sort.freeze
  else
    [].freeze
  end
end

#pair_systemd_fds(bind_uris, filenos) ⇒ Hash{String => Array<Integer>}

Returns the inherited-FDs hash for a systemd socket-activation handoff, pairing each bind URI with the FD systemd passed at the same index. Returns an empty hash when the FD count doesn't match the bind count.

RBS:

  • (Array[String] bind_uris, Array[Integer] filenos) -> Hash[String, Array[Integer]]

Parameters:

  • bind_uris (Array<String>) —

    the configured bind URIs

  • filenos (Array<Integer>) —

    file descriptors passed by systemd

Returns:

  • (Hash{String => Array<Integer>}) 


302
303
304
305
306
307
308
309
 
# File 'lib/raptor/cluster.rb', line 302

def pair_systemd_fds(bind_uris, filenos)
  if bind_uris.length != filenos.length
    Log.warn "Ignoring socket activation: #{filenos.length} fd(s) from systemd, #{bind_uris.length} bind(s) configured"
    return {}
  end

  bind_uris.zip(filenos).to_h { |bind_uri, fileno| [bind_uri, [fileno]] }
end

#perform_hot_restart ⇒ void

This method returns an undefined value.

Re-execs the master process with a fresh boot of the same Raptor invocation, handing the new master its listening sockets so accepted connections continue to be served across the swap.

RBS:

  • () -> void



685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
 
# File 'lib/raptor/cluster.rb', line 685

def perform_hot_restart
  @hot_restart_requested = false

  unless @launch_command && @launch_argv
    Log.warn "Hot restart unavailable: launch command not captured"
    return
  end

  Log.info "Hot restart starting"
  monotonic_usec = (Process.clock_gettime(Process::CLOCK_MONOTONIC) * 1_000_000).to_i
  Systemd.notify("RELOADING=1\nMONOTONIC_USEC=#{monotonic_usec}")
  @shutdown = true
  stop_workers
  @binder.clear_close_on_exec
  ENV[INHERITED_FDS_ENV] = JSON.generate(@binder.inheritable_fds)
  File.delete(@stats_file) rescue nil if @stats_file
  File.delete(@pid_file) rescue nil if @pid_file
  @stats.unmap
  $stdout.flush
  $stderr.flush
  exec(@launch_command, *@launch_argv)
end

#perform_phased_restart ⇒ void

This method returns an undefined value.

Replaces each worker process one at a time, waiting for the new worker to boot before moving on to the next.

RBS:

  • () -> void



599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
 
# File 'lib/raptor/cluster.rb', line 599

def perform_phased_restart
  @phased_restart_requested = false
  @phased_restarting = true
  @phase += 1
  Log.info "Phased restart starting"

  begin
    wait_for_seed_ready
    filled_index = fill_vacated_seed_slot

    @workers.keys.sort.each do |index|
      return if @shutdown
      next if index == filled_index

      target_pid = @workers[index]
      next unless target_pid

      Process.kill("TERM", target_pid) rescue nil

      deadline = Process.clock_gettime(Process::CLOCK_MONOTONIC) + 60
      until @shutdown
        reap_workers
        current = @workers[index]
        stat = @stats.all[index]
        break if current && current != target_pid && stat[:pid] == current && stat[:booted]
        break if Process.clock_gettime(Process::CLOCK_MONOTONIC) > deadline

        sleep 0.1
      end
    end

    Log.info "Phased restart complete"
  ensure
    @phased_restarting = false
  end
end

#pick_refork_candidate ⇒ Array<Integer>?

Picks the most-experienced booted worker in the current phase, returning its slot index and its request count. Returns nil when no worker qualifies.

RBS:

  • () -> Array[Integer]?

Returns:

  • (Array<Integer>, nil) 


465
466
467
468
469
470
471
472
473
474
475
476
477
478
 
# File 'lib/raptor/cluster.rb', line 465

def pick_refork_candidate
  best_index = nil
  best_requests = -1
  @stats.all.each_with_index do |stat, index|
    next unless @workers[index] == stat[:pid]
    next unless stat[:booted]
    next unless stat[:phase] == @phase
    next unless stat[:requests] > best_requests

    best_index = index
    best_requests = stat[:requests]
  end
  [best_index, best_requests] if best_index
end

#pid_alive?(pid) ⇒ Boolean

Checks whether a process with the given pid is currently alive.

RBS:

  • (Integer pid) -> bool

Parameters:

  • pid (Integer) —

    the pid to probe

Returns:

  • (Boolean) —

    true if the process exists



378
379
380
381
382
383
384
385
 
# File 'lib/raptor/cluster.rb', line 378

def pid_alive?(pid)
  Process.kill(0, pid)
  true
rescue Errno::ESRCH, Errno::ECHILD
  false
rescue Errno::EPERM
  true
end

#poll_seed_ready ⇒ void

This method returns an undefined value.

Records the seed's readiness when its ready marker has arrived. Non-blocking.

RBS:

  • () -> void



530
531
532
533
534
535
536
537
538
 
# File 'lib/raptor/cluster.rb', line 530

def poll_seed_ready
  return unless @resp_r && @seed_pid && !@seed_ready
  return unless @resp_r.wait_readable(0)

  bytes = @resp_r.read_nonblock(4, exception: false)
  return unless bytes.is_a?(String) && bytes.bytesize == 4

  @seed_ready = true if bytes.unpack1("L").zero?
end

#promote_worker_to_seed(index) ⇒ void

This method returns an undefined value.

Retires the current seed and promotes the given worker into its place, queueing a phased refork for the remaining workers.

RBS:

  • (Integer index) -> void

Parameters:

  • index (Integer) —

    slot index of the worker to promote



487
488
489
490
491
492
493
494
495
496
497
498
499
500
 
# File 'lib/raptor/cluster.rb', line 487

def promote_worker_to_seed(index)
  pid = @workers[index]
  return unless pid

  retire_current_seed
  Log.info "Promoting worker #{index} to seed for phased refork"
  ReuseportBPF.mark_unavailable(index) if @bpf_active
  Process.kill("URG", pid) rescue nil
  @workers.delete(index)
  @seed_pid = pid
  @seed_ready = false
  @seed_vacated_index = index
  @phased_restart_requested = true
end

#reap_pid(pid, status) ⇒ void

This method returns an undefined value.

Records a reaped pid, respawning its worker slot unless the cluster is shutting down. Clears the seed reference when the seed exits.

RBS:

  • (Integer pid, Process::Status status) -> void

Parameters:

  • pid (Integer) —

    the reaped pid

  • status (Process::Status) —

    the exit status



414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
 
# File 'lib/raptor/cluster.rb', line 414

def reap_pid(pid, status)
  if pid == @seed_pid
    Log.info "Seed (#{pid}) exited, #{exit_description(status)}"
    @seed_pid = nil
    return
  end

  index = @workers.key(pid)
  return unless index

  @workers.delete(index)
  @timed_out.delete(pid)

  unless @shutdown
    Log.warn "Restarting worker #{index} (#{pid}), #{exit_description(status)}"
    spawn_worker(index)
  end
end

#reap_workers ⇒ Symbol

Reaps any worker processes that have exited, respawning each one unless the cluster is shutting down.

RBS:

  • () -> Symbol

Returns:

  • (Symbol) —

    :no_children when nothing is left to supervise, otherwise :reaped



393
394
395
396
397
398
399
400
401
402
403
404
 
# File 'lib/raptor/cluster.rb', line 393

def reap_workers
  loop do
    pid, status = Process.wait2(-1, Process::WNOHANG)
    break unless pid

    reap_pid(pid, status)
  end

  @workers.empty? && !@seed_pid ? :no_children : :reaped
rescue Errno::ECHILD
  @workers.empty? && !@seed_pid ? :no_children : :reaped
end

#reopen_logs ⇒ void

This method returns an undefined value.

Redirects $stdout, $stderr, and the access log to their configured paths. No-op for any stream whose target path is nil.

RBS:

  • () -> void



992
993
994
995
996
997
998
999
1000
 
# File 'lib/raptor/cluster.rb', line 992

def reopen_logs
  $stdout.reopen(@stdout_file, "a").sync = true if @stdout_file
  $stderr.reopen(@stderr_file, "a").sync = true if @stderr_file
  return unless @access_log_file

  @access_log_io ||= File.open(@access_log_file, "a")
  @access_log_io.reopen(@access_log_file, "a")
  @access_log_io.sync = true
end

#reopen_logs_and_signal_workers ⇒ void

This method returns an undefined value.

Reopens the master's log files and forwards SIGHUP to each worker so they reopen their own inherited file descriptors.

RBS:

  • () -> void



1008
1009
1010
1011
 
# File 'lib/raptor/cluster.rb', line 1008

def reopen_logs_and_signal_workers
  reopen_logs
  @workers.values.each { |pid| Process.kill("HUP", pid) rescue nil }
end

#retire_current_seed ⇒ void

This method returns an undefined value.

Terminates the currently-active seed process, if any, and waits for it to exit. Its seed-forked workers stay attached to the master and keep serving.

RBS:

  • () -> void



509
510
511
512
513
514
515
516
517
518
519
520
521
522
 
# File 'lib/raptor/cluster.rb', line 509

def retire_current_seed
  return unless @seed_pid && pid_alive?(@seed_pid)

  Log.info "Retiring seed (#{@seed_pid})"
  Process.kill("TERM", @seed_pid) rescue nil

  deadline = Process.clock_gettime(Process::CLOCK_MONOTONIC) + @worker_shutdown_timeout
  while @seed_pid && pid_alive?(@seed_pid)
    reap_workers
    break if Process.clock_gettime(Process::CLOCK_MONOTONIC) > deadline

    sleep 0.05
  end
end

#run ⇒ void

This method returns an undefined value.

Runs the cluster until a graceful shutdown is signalled.

RBS:

  • () -> void



220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
 
# File 'lib/raptor/cluster.rb', line 220

def run
  $stdout.sync = true
  $stderr.sync = true

  reopen_logs

  trap("INT") { shutdown }
  trap("TERM") { shutdown }
  trap("HUP") { reopen_logs_and_signal_workers }
  trap("USR1") { @phased_restart_requested = true }
  trap("USR2") { @hot_restart_requested = true }

  if @refork_thresholds.any?
    if Subreaper.enable
      @fork_r, @fork_w = IO.pipe
      @resp_r, @resp_w = IO.pipe
      trap("URG") { @refork_requested = true }
    else
      Log.warn "Ignoring refork_after: PR_SET_CHILD_SUBREAPER not supported on this platform"
      @refork_thresholds = [].freeze
    end
  end

  File.open(@pid_file, File::CREAT | File::EXCL | File::WRONLY) { |file| file.write(Process.pid.to_s) } if @pid_file

  @bpf_active = ReuseportBPF.setup(@worker_count)

  @worker_count.times { |index| spawn_worker(index) }

  stats_file_thread = if @stats_file
    Thread.new do
      Thread.current.name = "Stats File Writer"

      write_stats_file_loop
    end
  end

  Systemd.notify("READY=1\nMAINPID=#{Process.pid}")

  until @shutdown
    break if reap_workers == :no_children

    perform_hot_restart if @hot_restart_requested
    poll_seed_ready if @seed_pid && !@seed_ready
    perform_phased_restart if @phased_restart_requested && !@phased_restarting
    check_refork_trigger if @refork_thresholds.any? && !@phased_restarting
    timeout_hung_workers

    sleep 0.1
  end

  Systemd.notify("STOPPING=1")
  stop_workers
  stats_file_thread&.join
  File.delete(@stats_file) rescue nil if @stats_file
  File.delete(@pid_file) rescue nil if @pid_file
  @stats.unmap
  @binder.close
end

#run_seed_loop(index) ⇒ void

This method returns an undefined value.

Runs the seed's fork loop, forking a replacement worker for each slot index the master asks for.

RBS:

  • (Integer index) -> void

Parameters:

  • index (Integer) —

    the seed's original slot index



886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
 
# File 'lib/raptor/cluster.rb', line 886

def run_seed_loop(index)
  Log.info "Worker #{index} promoted to seed"

  seed_shutdown = false
  trap("INT") { seed_shutdown = true }
  trap("TERM") { seed_shutdown = true }
  trap("URG", "IGNORE")

  child_pids = []
  trap("CHLD") do
    child_pids.reject! { Process.wait(_1, Process::WNOHANG) rescue true }
  end

  @resp_w.write([0].pack("L"))

  until seed_shutdown
    next unless @fork_r.wait_readable(1)

    bytes = @fork_r.read_nonblock(8, exception: false)
    break unless bytes.is_a?(String) && bytes.bytesize == 8

    slot_index, child_phase = bytes.unpack("LL")
    pid = fork { run_worker(slot_index, child_phase) }
    child_pids << pid
    @resp_w.write([pid].pack("L")) rescue nil
  end
rescue Errno::EPIPE, IOError
end

#run_worker(index, phase) ⇒ void

This method returns an undefined value.

Runs a worker process's full server stack until a shutdown signal is received or a critical component fails. On SIGURG the worker drains and transitions into a seed loop that forks replacement workers on master's request.

RBS:

  • (Integer index, Integer phase) -> void

Parameters:

  • index (Integer) —

    slot index for this worker in the stats region

  • phase (Integer) —

    the cluster phase this worker was forked at



718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
 
# File 'lib/raptor/cluster.rb', line 718

def run_worker(index, phase)
  @fork_w.close if @fork_w && !@fork_w.closed?
  @resp_r.close if @resp_r && !@resp_r.closed?

  shutdown_requested = false
  promote_to_seed = false
  trap("INT") { shutdown_requested = true }
  trap("TERM") { shutdown_requested = true }
  trap("HUP") { reopen_logs }
  trap("USR1", "IGNORE")
  trap("USR2", "IGNORE")
  trap("URG") { promote_to_seed = true } if @fork_r

  Raptor::CPU.pin(index) if Raptor::CPU.count >= @worker_count

  started_at = Process.clock_gettime(Process::CLOCK_REALTIME)
  request_count = 0

  @stats.write(
    index,
    pid: Process.pid,
    phase: phase,
    requests: 0,
    backlog: 0,
    busy_threads: 0,
    thread_capacity: @thread_count,
    started_at:,
    last_checkin: started_at,
    booted: false
  )

  reactor = nil

  counting_app = ->(env) {
    request_count += 1
    @app.call(env)
  }
  thread_pool = AtomicThreadPool.new(size: @thread_count)
  http1 = Http1.new(
    counting_app,
    @server_port,
    connection_options: @connection_options,
    http1_options: @http1_options,
    access_log_io: @access_log_io,
    on_error: @on_error
  )
  http2 = Http2.new(
    counting_app,
    @server_port,
    connection_options: @connection_options,
    http2_options: @http2_options,
    access_log_io: @access_log_io,
    on_error: @on_error
  )
  http1_ractor_pool = RactorPool.new(
    size: @http1_ractor_count,
    worker: http1.parser_worker,
    name: "HTTP/1.1"
  ) do |parsed_result|
    begin
      http1.handle_parsed_request(parsed_result, reactor, thread_pool)
    rescue => error
      Log.rescued_error(error)
    end
  end

  http2_ractor_pool = RactorPool.new(
    size: @http2_ractor_count,
    worker: http2.parser_worker,
    name: "HTTP/2"
  ) do |parsed_result|
    begin
      http2.handle_parsed_request(parsed_result, reactor, thread_pool)
    rescue => error
      Log.rescued_error(error)
    end
  end

  reactor = Reactor.new(
    http1_ractor_pool,
    http2_ractor_pool,
    thread_pool,
    connection_options: @connection_options,
    http1_options: @http1_options
  )
  reactor_thread = reactor.run

  worker_listeners = if @bpf_active
    bpf_listeners = ReuseportBPF.create_worker_listeners(@binder.bind_uris, index, @binder.socket_backlog)
    ReuseportBPF.enable_load_reporting(index)
    non_tcp_listeners = @binder.listeners.reject { |listener| listener.is_a?(TCPServer) }
    bpf_listeners + non_tcp_listeners
  else
    @binder.listeners
  end

  server = Server.new(
    @binder,
    reactor,
    thread_pool,
    http1,
    http2,
    connection_options: @connection_options,
    drain_accept_queue: @drain_accept_queue,
    listeners: worker_listeners,
    worker_index: (index if @bpf_active)
  )
  server_thread = server.run

  @before_worker_boot.each(&:call)

  Log.info "Worker #{index} booted"

  stats_thread = Thread.new do
    Thread.current.name = "Stats Writer"

    loop do
      @stats.write(
        index,
        pid: Process.pid,
        phase: phase,
        requests: request_count,
        backlog: reactor.backlog,
        busy_threads: thread_pool.active_count,
        thread_capacity: @thread_count,
        started_at:,
        last_checkin: Process.clock_gettime(Process::CLOCK_REALTIME),
        booted: true
      )
      break if shutdown_requested || promote_to_seed

      sleep 1
    end
  end

  until shutdown_requested || promote_to_seed
    break unless server_thread.alive? && reactor_thread.alive?

    sleep 0.5
  end

  if promote_to_seed
    @before_refork.each(&:call)
    server.stop_accepting
  else
    @before_worker_shutdown.each(&:call)
    server.shutdown
  end

  server_thread.join
  reactor.shutdown
  reactor_thread.join
  http1_ractor_pool.shutdown
  http2_ractor_pool.shutdown
  http1.shutdown
  drain_thread_pool(thread_pool)
  stats_thread.join

  run_seed_loop(index) if promote_to_seed
end

#shutdown ⇒ void

This method returns an undefined value.

Initiates graceful shutdown of the cluster.

RBS:

  • () -> void



957
958
959
960
961
 
# File 'lib/raptor/cluster.rb', line 957

def shutdown
  return if @shutdown

  @shutdown = true
end

#spawn_worker(index) ⇒ void

This method returns an undefined value.

Forks a new worker process and registers it at the given index, forking from the seed when one is active and off the master otherwise.

RBS:

  • (Integer index) -> void

Parameters:

  • index (Integer) —

    slot index for this worker in the stats region



337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
 
# File 'lib/raptor/cluster.rb', line 337

def spawn_worker(index)
  if @seed_pid && pid_alive?(@seed_pid)
    pid = spawn_worker_via_seed(index)
    if pid
      @workers[index] = pid
      return
    end
    Log.warn "Seed (#{@seed_pid}) failed to fork worker #{index}, falling back to direct fork"
    @seed_pid = nil
  end

  @before_fork.each(&:call)
  pid = fork { run_worker(index, @phase) }
  @workers[index] = pid
end

#spawn_worker_via_seed(index) ⇒ Integer?

Asks the seed to fork a new worker at the given index, returning the child pid, or nil when the seed doesn't respond in time.

RBS:

  • (Integer index) -> Integer?

Parameters:

  • index (Integer) —

    slot index for the new worker

Returns:

  • (Integer, nil) —

    the forked worker's pid, or nil on failure



360
361
362
363
364
365
366
367
368
369
370
 
# File 'lib/raptor/cluster.rb', line 360

def spawn_worker_via_seed(index)
  @fork_w.write([index, @phase].pack("LL"))
  return unless @resp_r.wait_readable(5)

  bytes = @resp_r.read_nonblock(4, exception: false)
  return unless bytes.is_a?(String) && bytes.bytesize == 4

  bytes.unpack1("L")
rescue Errno::EPIPE, IOError
  nil
end

#stats ⇒ Array<Hash>

Returns stats for all worker processes.

RBS:

  • () -> Array[Hash[Symbol, untyped]]

Returns:

  • (Array<Hash>) —

    array of per-worker stat hashes, each containing :pid, :index, :phase, :requests, :backlog, :busy_threads, :thread_capacity, :started_at, :last_checkin, and :booted



287
288
289
 
# File 'lib/raptor/cluster.rb', line 287

def stats
  @stats.all
end

#stop_workers ⇒ void

This method returns an undefined value.

Stops every worker (and the seed if one is active), escalating from TERM to KILL if any fail to exit within worker_shutdown_timeout.

RBS:

  • () -> void



546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
 
# File 'lib/raptor/cluster.rb', line 546

def stop_workers
  @workers.values.each { |pid| Process.kill("TERM", pid) rescue nil }
  Process.kill("TERM", @seed_pid) rescue nil if @seed_pid

  deadline = Process.clock_gettime(Process::CLOCK_MONOTONIC) + @worker_shutdown_timeout
  until (@workers.empty? && !@seed_pid) || Process.clock_gettime(Process::CLOCK_MONOTONIC) > deadline
    reap_workers
    sleep 0.05
  end
  return if @workers.empty? && !@seed_pid

  pids = @workers.values + [@seed_pid].compact
  Log.warn "Force-killing #{pids.size} process(es) after #{@worker_shutdown_timeout}s"
  pids.each { |pid| Process.kill("KILL", pid) rescue nil }
  pids.each { |pid| Process.wait(pid) rescue nil }
end

#timeout_hung_workers ⇒ void

This method returns an undefined value.

Kills workers that have stopped checking in. A booted worker that fails to update its stats slot within worker_timeout seconds is assumed to be hung (deadlocked app, runaway loop, blocked syscall); a worker still in startup is held to worker_boot_timeout. Killed workers are then restarted by reap_workers.

RBS:

  • () -> void



572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
 
# File 'lib/raptor/cluster.rb', line 572

def timeout_hung_workers
  now = Process.clock_gettime(Process::CLOCK_REALTIME)
  stats = @stats.all

  @workers.each do |index, pid|
    next if @timed_out.include?(pid)

    stat = stats[index]
    next unless stat[:pid] == pid

    timeout = stat[:booted] ? @worker_timeout : @worker_boot_timeout
    elapsed = now - stat[:last_checkin]
    next if elapsed <= timeout

    action = stat[:booted] ? "check in" : "boot"
    Log.warn "Killing worker #{index} (#{pid}), failed to #{action} within #{timeout}s"
    Process.kill("KILL", pid) rescue nil
    @timed_out << pid
  end
end

#wait_for_seed_ready ⇒ void

This method returns an undefined value.

Blocks until the freshly-promoted seed has signalled readiness, or times out and clears the seed reference. No-op when no seed is being promoted.

RBS:

  • () -> void



643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
 
# File 'lib/raptor/cluster.rb', line 643

def wait_for_seed_ready
  return unless @seed_pid && !@seed_ready

  deadline = Process.clock_gettime(Process::CLOCK_MONOTONIC) + @worker_drain_timeout + 5
  until @seed_ready || @shutdown
    break unless pid_alive?(@seed_pid)
    break if Process.clock_gettime(Process::CLOCK_MONOTONIC) > deadline

    poll_seed_ready
    sleep 0.1
  end

  return if @seed_ready

  Log.warn "Seed (#{@seed_pid}) didn't signal ready in time, falling back to direct forks"
  @seed_pid = nil
end

#write_stats_file_loop ⇒ void

This method returns an undefined value.

Writes the stats file on a 1-second interval until shutdown.

RBS:

  • () -> void



1018
1019
1020
1021
1022
1023
1024
1025
1026
 
# File 'lib/raptor/cluster.rb', line 1018

def write_stats_file_loop
  loop do
    File.write(@stats_file, JSON.generate({ master_pid: Process.pid, workers: @stats.all }))
    break if @shutdown

    sleep 1
  end
rescue SystemCallError
end