Class: Wurk::Swarm

Inherits:
Object
  • Object
show all
Includes:
Component
Defined in:
lib/wurk/swarm.rb,
lib/wurk/swarm/backoff.rb,
lib/wurk/swarm/restart.rb,
lib/wurk/swarm/child_boot.rb,
lib/wurk/swarm/orphan_guard.rb

Overview

Parent supervisor. Forks N children per the worker topology, monitors PIDs, relays signals, respawns crashed children with per-slot exponential backoff, handles rolling restart on SIGUSR1, recycles RSS-bloated children.

The supervise loop never sleeps on behalf of a respawn or a restart: crash backoff is tracked as per-slot due-times (Swarm::Backoff) and rolling restart / recycle run as a non-blocking state machine (Swarm::Restart) advanced one phase per tick. TERM/INT is therefore honored within a tick regardless of restart or backoff state.

Boot ordering (must be exact — see docs/idea/03-process-model.md):

1. Host app boots fully; eager loads done.
2. Railtie `after_initialize` fires.
3. `boot` closes parent-side connections (Redis, ActiveRecord).
4. `boot` forks N children.
5. Each child reconnects DB + opens a fresh Redis pool, then
 installs its own signal handlers and starts the Launcher.
6. Parent calls `supervise` to enter the wait/relay loop.

Signals (see docs/idea/04-signals.md):

TERM/INT  → `shutdown`           (graceful drain; aborts any restart)
TSTP      → relay TSTP           (quiet — stop fetching; one-way, no resume)
USR1      → `rolling_restart`    (zero-downtime cycle)

Defined Under Namespace

Classes: Backoff, ChildBoot, OrphanGuard, Restart

Constant Summary collapse

SUPERVISE_TICK = 
0.2
RESPAWN_BACKOFF = 
1.0
HEARTBEAT_WAIT = 
30
MEMORY_CHECK_INTERVAL = 
10
DEFAULT_SHUTDOWN_TIMEOUT = 
25
SUPERVISOR_POOL_SIZE =

The supervisor's own Redis pool: one connection, disjoint from every capsule pool. Post-boot the parent's only Redis use is the heartbeat probe (heartbeat_seen?) — one SISMEMBER per restart tick — and routing it through the capsule main pool reopened, at capsule size (>= 10), the very sockets step 3 had just closed; the next respawn/restart fork then inherited them. Dedicated and torn down before every fork, so no child ever sees a parent socket.

1
SUPERVISOR_POOL_NAME = 
'swarm-supervisor'
SHUTDOWN_GRACE =

Children each hard_shutdown after their own drain deadline (bulk_requeue

  • a 3s ensure window + heartbeat cleanup); the parent must not SIGKILL them mid-tail, so its own wait always extends past theirs by this much.
5
SWARM_SIGNALS =

USR2 is relayed (log reopen) — without a trap, a logrotate config that signals the master pid would hit USR2's default disposition and kill the whole swarm.

{ 'TERM' => :term, 'INT' => :term, 'TSTP' => :tstp, 'USR1' => :usr1, 'USR2' => :usr2 }.freeze

Constants included from Component

Component::DEFAULT_THREAD_PRIORITY, Component::LEADER_CACHE_TTL_MS, Component::PROCESS_NONCE

Instance Attribute Summary collapse

Attributes included from Component

#config

Instance Method Summary collapse

Methods included from Component

#default_tag, #fire_event, #handle_exception, #hostname, #identity, #leader?, #logger, #mono_ms, #process_nonce, #real_ms, #redis, #safe_thread, #tid, #watchdog

Constructor Details

#initialize(topology:, config: Wurk.configuration, memory_limit: config.memory_limit_kb, shutdown_timeout: DEFAULT_SHUTDOWN_TIMEOUT) ⇒ Swarm

Returns a new instance of Swarm.



69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
 
# File 'lib/wurk/swarm.rb', line 69

def initialize(topology:, config: Wurk.configuration, memory_limit: config.memory_limit_kb,
               shutdown_timeout: DEFAULT_SHUTDOWN_TIMEOUT)
  @topology = topology
  @config = config
  @memory_limit = memory_limit
  @shutdown_timeout = shutdown_timeout
  # One reentrant lock covers the child table AND the restart machine: they
  # are mutually recursive (Restart#advance calls back into `describe` and
  # `spawn`, which read and write `@children`), so separate locks would be
  # taken in opposite orders — enqueue holds children→restart, advance holds
  # restart→children — and deadlock. Monitor's reentrancy is what lets those
  # callbacks re-enter. Never held across a sleep: `wait_for_children` and
  # the supervise tick sleep outside it.
  @lock = ::Monitor.new
  @children = {}
  @assignments = []
  @owner_pid = nil
  @stopping = false
  @shutdown_requested = false
  @quieted = false
  @last_memory_check = 0
  @respawn_backoff = Backoff.new(base: RESPAWN_BACKOFF)
  @restart = build_restart
  init_fork_unsafe_handles
end

Instance Attribute Details

#topology ⇒ Object (readonly)

Returns the value of attribute topology.



67
68
69
 
# File 'lib/wurk/swarm.rb', line 67

def topology
  @topology
end

Instance Method Details

#boot(install_signals: true) ⇒ Object

install_signals: is false in tests so the integration suite can drive shutdown / rolling_restart directly without poisoning the test process's signal handlers.

Traps go in BEFORE fork_children: a TERM landing in the (previously post-fork) window between fork and trap installation left the parent on its default disposition — it died instantly and orphaned live, fetching children. Installed first, the trap queues the TERM and the supervise loop drains it (relaying to children) even if it arrives mid-boot.

Raises:

  • (ArgumentError) 


104
105
106
107
108
109
110
111
112
113
114
 
# File 'lib/wurk/swarm.rb', line 104

def boot(install_signals: true)
  raise 'Wurk::Swarm already booted' unless @assignments.empty?
  raise ArgumentError, 'Topology has no slots' if @topology.empty?

  @assignments = @topology.assignments.freeze
  @owner_pid = ::Process.pid
  install_signal_handlers if install_signals
  close_parent_sockets
  fork_children
  child_pids
end

#children ⇒ Object

A snapshot, not the live table. Callers read this off the supervise thread, which inserts (respawn, restart) and deletes (reap) on every tick; handing out the live Hash lets them iterate it mid-mutation.



119
120
121
 
# File 'lib/wurk/swarm.rb', line 119

def children
  @lock.synchronize { @children.dup }
end

#owner? ⇒ Boolean

Only the process that forked the children may supervise or signal them. A forked child inherits this object along with the host's at_exit hooks — and rails_boot registers one that drains the swarm — so ChildBoot's exit reaches a full drain inside the child. There @children holds the child's SIBLINGS, not its own children: unguarded, that TERMs them, stalls the whole shutdown timeout waiting on PIDs it can never reap, then SIGKILLs whatever survived.

Returns:

  • (Boolean) 


166
167
168
 
# File 'lib/wurk/swarm.rb', line 166

def owner?
  ::Process.pid == @owner_pid
end

#quiet_swarm ⇒ Object

TSTP quiet is one-way and GLOBAL (spec §21.3): it must survive respawns and memory recycles, or a quieted-but-crashed child's replacement would resume fetching mid-maintenance. The flag makes every future fork boot already-quieted (see ChildBoot start_quiet).



174
175
176
177
 
# File 'lib/wurk/swarm.rb', line 174

def quiet_swarm
  @quieted = true
  relay_signal('TSTP')
end

#request_shutdown ⇒ Object

Cross-thread drain request: raise a flag the supervise loop observes on its next tick instead of draining on the caller's thread. shutdown walks and clears the child table that the supervise thread is concurrently mutating (reap, respawn, restart), so two threads inside it race. Every caller that isn't the supervise thread — rails_boot's at_exit, which fires on the host's main thread — comes through here.



155
156
157
 
# File 'lib/wurk/swarm.rb', line 155

def request_shutdown
  @shutdown_requested = true
end

#rolling_restart ⇒ Object

SIGUSR1: queue every live child for the rolling-restart state machine, which replaces one slot at a time (spawn replacement → await its heartbeat → TERM the old child → await its drain) without blocking the supervise thread, so TERM stays responsive throughout the cycle.



183
184
185
 
# File 'lib/wurk/swarm.rb', line 183

def rolling_restart
  @lock.synchronize { @restart.enqueue(@children.keys) }
end

#shutdown(timeout: @shutdown_timeout) ⇒ Object 



137
138
139
140
141
142
143
144
145
146
147
 
# File 'lib/wurk/swarm.rb', line 137

def shutdown(timeout: @shutdown_timeout)
  return unless owner?

  @stopping = true
  @lock.synchronize { @restart.abort }
  relay_signal('TERM')
  wait_for_children(timeout + SHUTDOWN_GRACE)
  hard_kill_stragglers
  close_supervisor_pool
  close_signal_pipe
end

#supervise ⇒ Object 



123
124
125
126
127
128
129
130
131
132
133
134
135
 
# File 'lib/wurk/swarm.rb', line 123

def supervise
  return unless owner?

  until done?
    drain_signals
    shutdown if @shutdown_requested && !@stopping
    reap_children
    spawn_due_respawns
    advance_restart unless @stopping
    check_memory_pressure
    sleep SUPERVISE_TICK
  end
end