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

  • :max_threads (Integer, Float) —

    maximum number of threads per worker process, or Float::INFINITY for no limit

  • :cpu_affinity (Boolean) —

    whether to pin each worker to a CPU

  • :clean_thread_locals (Boolean) —

    whether to clear application thread locals after each request

  • :clean_fiber_locals (Boolean) —

    whether to process each request in a fresh Fiber

  • :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 with the worker index before it begins serving

  • :before_worker_shutdown (Array<Proc>) —

    procs called with the worker index 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

  • :control_url (String, nil) —

    unix:// URL serving cluster stats, 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



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
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
 
# File 'lib/raptor/cluster.rb', line 168

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]
  @max_thread_count = options[:max_threads]
  @cpu_affinity = options[:cpu_affinity]
  @clean_thread_locals = options[:clean_thread_locals]
  @clean_fiber_locals = options[:clean_fiber_locals]
  @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]
  @control_server = ControlServer.new(options[:control_url]) { stats_hash } if options[:control_url]
  @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)
  @started_at = Process.clock_gettime(Process::CLOCK_REALTIME)
  @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) 


55
56
57
 
# File 'lib/raptor/cluster.rb', line 55

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) 


69
70
71
 
# File 'lib/raptor/cluster.rb', line 69

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



41
42
43
 
# File 'lib/raptor/cluster.rb', line 41

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



511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
 
# File 'lib/raptor/cluster.rb', line 511

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



1001
1002
1003
1004
1005
1006
1007
1008
1009
1010
1011
1012
 
# File 'lib/raptor/cluster.rb', line 1001

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



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

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



738
739
740
741
742
743
744
745
746
747
 
# File 'lib/raptor/cluster.rb', line 738

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



1047
1048
1049
1050
1051
1052
1053
1054
1055
1056
1057
1058
1059
1060
1061
1062
1063
1064
1065
1066
1067
1068
1069
1070
1071
 
# File 'lib/raptor/cluster.rb', line 1047

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}"
  http2_enabled = @binder.http2?
  Log.info "│     ├─ #{@http2_ractor_count} HTTP/2 pipeline ractor#{"s" if @http2_ractor_count > 1}" if http2_enabled
  collector_count = http2_enabled ? 2 : 1
  Log.info "│     ├─ #{collector_count} pipeline collector thread#{"s" if collector_count > 1}"
  thread_limit = if @max_thread_count == Float::INFINITY
    " (scaling, no limit)"
  elsif @max_thread_count
    " (scaling up to #{@max_thread_count})"
  else
    ""
  end
  Log.info "│     ├─ #{@thread_count} worker thread#{"s" if @thread_count > 1}#{thread_limit}"
  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>) 


390
391
392
393
394
395
396
397
398
399
 
# File 'lib/raptor/cluster.rb', line 390

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>}) 


373
374
375
376
377
378
379
380
 
# File 'lib/raptor/cluster.rb', line 373

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



756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
 
# File 'lib/raptor/cluster.rb', line 756

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)
  @control_server&.shutdown
  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



670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
 
# File 'lib/raptor/cluster.rb', line 670

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) 


536
537
538
539
540
541
542
543
544
545
546
547
548
549
 
# File 'lib/raptor/cluster.rb', line 536

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



449
450
451
452
453
454
455
456
 
# File 'lib/raptor/cluster.rb', line 449

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



601
602
603
604
605
606
607
608
609
 
# File 'lib/raptor/cluster.rb', line 601

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



558
559
560
561
562
563
564
565
566
567
568
569
570
571
 
# File 'lib/raptor/cluster.rb', line 558

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



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

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



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

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



1079
1080
1081
1082
1083
1084
1085
1086
1087
 
# File 'lib/raptor/cluster.rb', line 1079

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



1095
1096
1097
1098
 
# File 'lib/raptor/cluster.rb', line 1095

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



580
581
582
583
584
585
586
587
588
589
590
591
592
593
 
# File 'lib/raptor/cluster.rb', line 580

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



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
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
 
# File 'lib/raptor/cluster.rb', line 239

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)

  @control_server&.bind
  @worker_count.times { |index| spawn_worker(index) }
  @control_server&.start

  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
  @control_server&.shutdown
  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



964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
 
# File 'lib/raptor/cluster.rb', line 964

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



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
878
879
880
881
882
883
884
885
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
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
 
# File 'lib/raptor/cluster.rb', line 790

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 @cpu_affinity && 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, max_size: @max_thread_count)
  http1 = Http1.new(
    counting_app,
    @server_port,
    connection_options: @connection_options,
    http1_options: @http1_options,
    access_log_io: @access_log_io,
    clean_thread_locals: @clean_thread_locals,
    clean_fiber_locals: @clean_fiber_locals,
    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,
    clean_thread_locals: @clean_thread_locals,
    clean_fiber_locals: @clean_fiber_locals,
    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 = if @binder.http2?
    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
  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 { |hook| hook.parameters.empty? ? hook.call : hook.call(index) }

  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_pool.size,
        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 { |hook| hook.parameters.empty? ? hook.call : hook.call(index) }
    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



1035
1036
1037
1038
1039
 
# File 'lib/raptor/cluster.rb', line 1035

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



408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
 
# File 'lib/raptor/cluster.rb', line 408

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



431
432
433
434
435
436
437
438
439
440
441
 
# File 'lib/raptor/cluster.rb', line 431

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



309
310
311
 
# File 'lib/raptor/cluster.rb', line 309

def stats
  @stats.all
end

#stats_hash ⇒ Hash

Returns cluster statistics for the control server. For an adaptive pool, max_threads is the number of threads currently running so utilization reflects the capacity available at that moment.

RBS:

  • () -> Hash[Symbol, untyped]

Returns:

  • (Hash) —

    cluster statistics



320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
 
# File 'lib/raptor/cluster.rb', line 320

def stats_hash
  worker_stats = @stats.all
  worker_status = @workers.sort.map do |index, pid|
    stat = worker_stats[index]
    stat = {} unless stat && stat[:pid] == pid
    capacity = stat.fetch(:thread_capacity, 0)
    active = stat.fetch(:busy_threads, 0)
    total_work = stat.fetch(:backlog, 0)

    {
      started_at: timestamp(stat.fetch(:started_at, 0)),
      pid: pid,
      index: index,
      phase: stat.fetch(:phase, @phase),
      booted: stat.fetch(:booted, false),
      last_checkin: timestamp(stat.fetch(:last_checkin, 0)),
      last_status: {
        backlog: [total_work - active, 0].max,
        running: capacity,
        pool_capacity: [capacity - total_work, 0].max,
        busy_threads: active,
        max_threads: capacity,
        requests_count: stat.fetch(:requests, 0),
      },
    }
  end

  {
    started_at: timestamp(@started_at),
    workers: worker_status.length,
    phase: @phase,
    booted_workers: worker_status.count { |worker| worker[:booted] },
    old_workers: worker_status.count { |worker| worker[:phase] != @phase },
    worker_status: worker_status,
  }
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



617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
 
# File 'lib/raptor/cluster.rb', line 617

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



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

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

#timestamp(timestamp) ⇒ String

RBS:

  • (Float timestamp) -> String

Parameters:

  • timestamp (Float)

Returns:

  • (String) 


360
361
362
 
# File 'lib/raptor/cluster.rb', line 360

def timestamp(timestamp)
  Time.at(timestamp).utc.iso8601(6)
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



714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
 
# File 'lib/raptor/cluster.rb', line 714

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



1105
1106
1107
1108
1109
1110
1111
1112
1113
 
# File 'lib/raptor/cluster.rb', line 1105

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