Class: Raptor::Cluster

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

Overview

Multi-process web server cluster with advanced concurrency architecture.

Cluster manages multiple worker processes, each running a complete server stack including a ractor pool for HTTP parsing, a thread pool for application processing, plus dedicated reactor and server threads. It handles process forking, signal management, graceful shutdown, and automatic worker restart when a worker process unexpectedly exits.

The architecture provides horizontal scaling through processes while maintaining efficient I/O and CPU utilization within each process through the combination of ractor-based parsing and thread pools on top of NIO reactors.

Flow per worker process:

  1. Server continuously accepts connections but skips acceptance when backlog is high
  2. Reactor manages I/O multiplexing and provides backlog metrics for load control
  3. Ractor pool handles CPU-intensive HTTP parsing in parallel
  4. Thread pool processes Rack applications and handles response writing
  5. Natural load balancing occurs through backpressure-based acceptance control

Examples:

Basic usage

options = {
  workers: 4, ractors: 2, threads: 8,
  binds: ["tcp://0.0.0.0:3000"],
  rackup: "config.ru",
  connection: { first_data_timeout: 30, chunk_data_timeout: 10 }
}
Cluster.run(options)

Constant Summary collapse

INHERITED_FDS_ENV =

Returns:

  • (::String) 
"RAPTOR_INHERITED_FDS"

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(options) ⇒ Cluster

Creates a new Cluster with the specified configuration.

Initializes the cluster with worker, ractor, and thread counts, sets up network binding, loads the Rack application, and prepares for multi-process operation.

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

  • :ractors (Integer) —

    number of ractors per worker process

  • :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



151
152
153
154
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
 
# File 'lib/raptor/cluster.rb', line 151

def initialize(options)
  @drain_accept_queue = options[:drain_accept_queue]
  @worker_count = options[:workers]
  @ractor_count = options[:ractors]
  @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

.run(options) ⇒ void

This method returns an undefined value.

Convenience method to create and run a cluster with the given options.

Parameters:

  • options (Hash) —

    cluster configuration options



59
60
61
 
# File 'lib/raptor/cluster.rb', line 59

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.



448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
 
# File 'lib/raptor/cluster.rb', line 448

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.

Parameters:

  • thread_pool (AtomicThreadPool) —

    the worker's thread pool



919
920
921
922
923
924
925
926
 
# File 'lib/raptor/cluster.rb', line 919

def drain_thread_pool(thread_pool)
  drain = Thread.new { thread_pool.shutdown }
  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.

Parameters:

  • status (Process::Status) —

    the exit status of the process

Returns:

  • (String) —

    a description of the exit reason



934
935
936
937
938
939
940
941
942
 
# File 'lib/raptor/cluster.rb', line 934

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.

Returns:

  • (Integer, nil) —

    the filled slot index, or nil if none was vacated



675
676
677
678
679
680
681
682
683
684
 
# File 'lib/raptor/cluster.rb', line 675

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.



961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
 
# File 'lib/raptor/cluster.rb', line 961

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 "│     ├─ #{@ractor_count} pipeline ractor#{"s" if @ractor_count > 1}"
  Log.info "│     ├─ 1 pipeline collector thread"
  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).

Parameters:

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

    the raw option value

Returns:

  • (Array<Integer>) 


327
328
329
330
331
332
333
334
335
336
 
# File 'lib/raptor/cluster.rb', line 327

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. The activation is skipped (with a warning) when the FD count doesn't match the number of bind URIs.

Parameters:

  • bind_uris (Array<String>) —

    the configured bind URIs

  • filenos (Array<Integer>) —

    file descriptors passed by systemd

Returns:

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


310
311
312
313
314
315
316
317
 
# File 'lib/raptor/cluster.rb', line 310

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.



693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
 
# File 'lib/raptor/cluster.rb', line 693

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.



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
635
636
637
638
639
640
641
642
 
# File 'lib/raptor/cluster.rb', line 607

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.

Returns:

  • (Array<Integer>, nil) 


473
474
475
476
477
478
479
480
481
482
483
484
485
486
 
# File 'lib/raptor/cluster.rb', line 473

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.

Parameters:

  • pid (Integer) —

    the pid to probe

Returns:

  • (Boolean) —

    true if the process exists



386
387
388
389
390
391
392
393
 
# File 'lib/raptor/cluster.rb', line 386

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.



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

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") == 0
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.

Parameters:

  • index (Integer) —

    slot index of the worker to promote



495
496
497
498
499
500
501
502
503
504
505
506
507
508
 
# File 'lib/raptor/cluster.rb', line 495

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.

Parameters:

  • pid (Integer) —

    the reaped pid

  • status (Process::Status) —

    the exit status



422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
 
# File 'lib/raptor/cluster.rb', line 422

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.

Returns:

  • (Symbol) —

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



401
402
403
404
405
406
407
408
409
410
411
412
 
# File 'lib/raptor/cluster.rb', line 401

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.



983
984
985
986
987
988
989
990
991
 
# File 'lib/raptor/cluster.rb', line 983

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.



999
1000
1001
1002
 
# File 'lib/raptor/cluster.rb', line 999

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.



517
518
519
520
521
522
523
524
525
526
527
528
529
530
 
# File 'lib/raptor/cluster.rb', line 517

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.

Starts the multi-process cluster and manages worker processes.

Forks the configured number of worker processes and monitors them, restarting any that exit unexpectedly or stop checking in. Handles graceful shutdown via INT or TERM signals, phased restart via USR1, and hot restart via USR2.

Each worker process includes:

  • 1 server thread (continuously accepts connections with backpressure control)
  • 1 reactor thread (I/O multiplexing, timeout handling, backlog monitoring)
  • N pipeline ractors (parallel HTTP parsing)
  • 1 pipeline collector thread (coordinates parsing results)
  • M worker threads (Rack application processing and response writing)
  • 1 stats thread (writes per-worker metrics to shared memory every second)


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
279
280
281
282
283
284
285
286
 
# File 'lib/raptor/cluster.rb', line 228

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.

Parameters:

  • index (Integer) —

    the seed's original slot index



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
 
# File 'lib/raptor/cluster.rb', line 882

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.

Parameters:

  • index (Integer) —

    slot index for this worker in the stats region

  • phase (Integer) —

    the cluster phase this worker was forked at



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
 
# File 'lib/raptor/cluster.rb', line 726

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
  )
  ractor_pool = RactorPool.new(
    size: @ractor_count,
    worker: http1.http_parser_worker
  ) do |parsed_result|
    begin
      if parsed_result[:protocol] == :http2
        http2.handle_parsed_request(parsed_result, reactor, thread_pool)
      else
        http1.handle_parsed_request(parsed_result, reactor, thread_pool)
      end
    rescue => error
      Log.rescued_error(error)
    end
  end

  reactor = Reactor.new(
    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)
    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
  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.



949
950
951
952
953
 
# File 'lib/raptor/cluster.rb', line 949

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.

Parameters:

  • index (Integer) —

    slot index for this worker in the stats region



345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
 
# File 'lib/raptor/cluster.rb', line 345

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.

Parameters:

  • index (Integer) —

    slot index for the new worker

Returns:

  • (Integer, nil) —

    the forked worker's pid, or nil on failure



368
369
370
371
372
373
374
375
376
377
378
 
# File 'lib/raptor/cluster.rb', line 368

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

  bytes = @resp_r.read_nonblock(4, exception: false)
  return nil 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.

Returns:

  • (Array<Hash>) —

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



294
295
296
 
# File 'lib/raptor/cluster.rb', line 294

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.



554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
 
# File 'lib/raptor/cluster.rb', line 554

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.



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

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.



651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
 
# File 'lib/raptor/cluster.rb', line 651

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.



1009
1010
1011
1012
1013
1014
1015
1016
1017
 
# File 'lib/raptor/cluster.rb', line 1009

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