Class: SolidObjects::Supervisor

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

Constant Summary collapse

MAXIMUM_RETENTION_BACKOFF_DOUBLINGS =

Returns:

  • (::Integer) 
16

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(worker_count: SolidObjects.configuration.worker_count, effect_worker_count: SolidObjects.configuration.effect_worker_count, broadcast_worker_count: SolidObjects.configuration.broadcast_worker_count, reminder_scheduler_count: SolidObjects.configuration.reminder_scheduler_count) ⇒ Supervisor

Returns a new instance of Supervisor.

RBS:

  • (?worker_count: Integer, ?effect_worker_count: Integer, ?broadcast_worker_count: Integer, ?reminder_scheduler_count: Integer) -> void

Parameters:

  • worker_count: (Integer) (defaults to: SolidObjects.configuration.worker_count)
  • effect_worker_count: (Integer) (defaults to: SolidObjects.configuration.effect_worker_count)
  • broadcast_worker_count: (Integer) (defaults to: SolidObjects.configuration.broadcast_worker_count)
  • reminder_scheduler_count: (Integer) (defaults to: SolidObjects.configuration.reminder_scheduler_count) 


16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
 
# File 'lib/solid_objects/supervisor.rb', line 16

def initialize(
  worker_count: SolidObjects.configuration.worker_count,
  effect_worker_count: SolidObjects.configuration.effect_worker_count,
  broadcast_worker_count: SolidObjects.configuration.broadcast_worker_count,
  reminder_scheduler_count: SolidObjects.configuration.reminder_scheduler_count
)
  @components = build_components(
    worker_count:,
    effect_worker_count:,
    broadcast_worker_count:,
    reminder_scheduler_count:
  )
  @threads = []
  @monitor = nil
  @started = false
  @cleaned_up_at = nil
  @retention = nil
  @lifecycle = Thread::Mutex.new
end

Instance Attribute Details

#components ⇒ Object (readonly)

Returns the value of attribute components.

Returns:

  • (Object) 


80
81
82
 
# File 'lib/solid_objects/supervisor.rb', line 80

def components
  @components
end

#threads ⇒ Object (readonly)

Returns the value of attribute threads.

Returns:

  • (Object) 


80
81
82
 
# File 'lib/solid_objects/supervisor.rb', line 80

def threads
  @threads
end

Instance Method Details

#build_components(worker_count:, effect_worker_count:, broadcast_worker_count:, reminder_scheduler_count:) ⇒ Array[Worker | EffectExecutor | ReminderScheduler | BroadcastExecutor]

RBS:

  • (worker_count: Integer, effect_worker_count: Integer, broadcast_worker_count: Integer, reminder_scheduler_count: Integer) -> Array[Worker | EffectExecutor | ReminderScheduler | BroadcastExecutor]

Parameters:

  • worker_count: (Integer)
  • effect_worker_count: (Integer)
  • broadcast_worker_count: (Integer)
  • reminder_scheduler_count: (Integer)

Returns:



254
255
256
257
258
259
260
261
262
263
264
 
# File 'lib/solid_objects/supervisor.rb', line 254

def build_components(
  worker_count:,
  effect_worker_count:,
  broadcast_worker_count:,
  reminder_scheduler_count:
)
  Array.new(worker_count) { Worker.new } +
    Array.new(effect_worker_count) { EffectExecutor.new } +
    Array.new(broadcast_worker_count) { BroadcastExecutor.new } +
    Array.new(reminder_scheduler_count) { ReminderScheduler.new }
end

#cleanup_dead_processes ⇒ void

This method returns an undefined value.

RBS:

  • () -> void



213
214
215
216
217
218
219
220
 
# File 'lib/solid_objects/supervisor.rb', line 213

def cleanup_dead_processes
  interval = SolidObjects.configuration.dead_process_cleanup_interval
  return unless interval.positive?
  return if @cleaned_up_at && monotonic_now - @cleaned_up_at < interval

  @cleaned_up_at = monotonic_now
  ProcessRegistry.cleanup_dead
end

#join_until_timeout ⇒ void

This method returns an undefined value.

RBS:

  • () -> void



267
268
269
270
271
272
273
274
275
 
# File 'lib/solid_objects/supervisor.rb', line 267

def join_until_timeout
  deadline = monotonic_now + SolidObjects.configuration.shutdown_timeout
  threads.each do |thread|
    remaining = deadline - monotonic_now
    break unless remaining.positive?

    thread.join(remaining)
  end
end

#monitor_loop ⇒ void

This method returns an undefined value.

A role that raises leaves its thread dead. Without replacement the process keeps running while quietly doing less work, so the supervisor watches its threads and restarts any that stopped before shutdown. A failing pass must not stop supervision, and must not retry without pacing either: a persistently failing database would otherwise spin.

RBS:

  • () -> void



88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
 
# File 'lib/solid_objects/supervisor.rb', line 88

def monitor_loop
  while @started
    begin
      replace_dead_roles
      cleanup_dead_processes
    rescue => error
      SolidObjects.instrument(
        :"supervisor.monitor_failed",
        error_class: error.class.name,
        error_message: error.message
      )
    end
    sleep SolidObjects.configuration.supervisor_monitor_interval
  end
end

#monotonic_now ⇒ Float

RBS:

  • () -> Float

Returns:

  • (Float) 


278
279
280
 
# File 'lib/solid_objects/supervisor.rb', line 278

def monotonic_now
  ::Process.clock_gettime(::Process::CLOCK_MONOTONIC)
end

#prune_expired_records ⇒ void

This method returns an undefined value.

Every actor call writes a durable message row, so retention that is only configured and never run leaves those rows to grow without bound. The supervisor runs it rather than requiring every application to schedule its own job.

RBS:

  • () -> void



180
181
182
183
184
185
 
# File 'lib/solid_objects/supervisor.rb', line 180

def prune_expired_records
  return unless SolidObjects.configuration.retention_interval.positive?

  MessagePruner.new.prune
  ProcessPruner.new.prune
end

#release_wake_up ⇒ void

This method returns an undefined value.

A wake-up adapter may hold connections outside the pool, which would otherwise accumulate across restarts in one process.

RBS:

  • () -> void



244
245
246
247
248
249
250
251
 
# File 'lib/solid_objects/supervisor.rb', line 244

def release_wake_up
  wake_up = SolidObjects.wake_up
  return unless wake_up.respond_to?(:stop)

  wake_up.stop
rescue
  nil
end

#replace_dead_roles ⇒ void

This method returns an undefined value.

A role that raises runs its own shutdown cleanup on the way out, so a crashed component reports itself stopped exactly like one that was asked to stop. While the supervisor is still running, a dead thread can only mean a crash, so replacement keys on the supervisor rather than on the component. The crashed instance has already released its process record, so a fresh one takes its place.

RBS:

  • () -> void



111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
 
# File 'lib/solid_objects/supervisor.rb', line 111

def replace_dead_roles
  components.each_with_index do |component, index|
    thread = threads[index]
    next if thread&.alive?

    replaced = @lifecycle.synchronize do
      next false unless @started

      replacement = component.class.new
      components[index] = replacement
      threads[index] = supervise(replacement)
      replacement
    end
    break unless replaced

    SolidObjects.instrument(
      :"supervisor.role_replaced",
      role: replaced.class.name,
      error_class: thread_error(thread)
    )
  end
end

#retention_loop ⇒ void

This method returns an undefined value.

Retention gets its own thread rather than sharing the monitor's. A large backlog or a lock wait can make a pass slow, and role replacement must not wait behind housekeeping.

RBS:

  • () -> void



146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
 
# File 'lib/solid_objects/supervisor.rb', line 146

def retention_loop
  failures = 0
  while @started
    begin
      prune_expired_records
      failures = 0
    rescue => error
      failures += 1
      SolidObjects.instrument(
        :"supervisor.retention_failed",
        error_class: error.class.name,
        error_message: error.message
      )
    end
    wait_for_next_retention(failures)
  end
end

#retention_pause(failures) ⇒ Float

A transient lock or connection error must not defer retention for the whole interval, so a failed pass retries at monitor cadence. The pause then doubles per consecutive failure, capped by the interval, so a database that stays down is not polled once a second forever.

RBS:

  • (Integer) -> Float

Parameters:

  • (Integer)

Returns:

  • (Float) 


192
193
194
195
196
197
198
199
200
 
# File 'lib/solid_objects/supervisor.rb', line 192

def retention_pause(failures)
  interval = SolidObjects.configuration.retention_interval
  interval = SolidObjects.configuration.supervisor_monitor_interval unless interval.positive?
  return interval if failures.zero?

  backoff = SolidObjects.configuration.supervisor_monitor_interval *
    (2**[ failures - 1, MAXIMUM_RETENTION_BACKOFF_DOUBLINGS ].min)
  [ backoff, interval ].min
end

#run ⇒ void

This method returns an undefined value.

RBS:

  • () -> void



37
38
39
40
41
42
 
# File 'lib/solid_objects/supervisor.rb', line 37

def run
  start
  threads.each(&:join)
ensure
  stop
end

#start ⇒ void

This method returns an undefined value.

RBS:

  • () -> void



45
46
47
48
49
50
51
52
53
 
# File 'lib/solid_objects/supervisor.rb', line 45

def start
  return if @started

  @started = true
  @threads = components.map { |component| supervise(component) }
  @monitor = Thread.new { monitor_loop }
  @retention = Thread.new { retention_loop }
  SolidObjects.instrument(:"supervisor.started", component_count: components.length)
end

#stop ⇒ void

This method returns an undefined value.

RBS:

  • () -> void



56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
 
# File 'lib/solid_objects/supervisor.rb', line 56

def stop
  return unless @started

  begin
    # Flipping the flag under the same lock replacement takes means a
    # replacement either completes before shutdown reads the component
    # list, or never starts.
    @lifecycle.synchronize { @started = false }
    stop_monitor
    stop_retention
    components.each(&:request_shutdown)
    join_until_timeout
    components.reject(&:stopped?).each(&:stop)
  ensure
    # Connections held outside the pool must be released even when a
    # component fails to stop, or they accumulate across restarts.
    release_wake_up
    @monitor = nil
    SolidObjects.instrument(:"supervisor.stopped", component_count: components.length)
  end
end

#stop_monitor ⇒ void

This method returns an undefined value.

The monitor only performs maintenance, so shutdown must never return while it is still alive: a pass blocked on the database would otherwise outlive the supervisor that owns it.

RBS:

  • () -> void



231
232
233
234
235
236
237
238
239
 
# File 'lib/solid_objects/supervisor.rb', line 231

def stop_monitor
  monitor = @monitor
  @monitor = nil
  return unless monitor

  monitor.join(SolidObjects.configuration.shutdown_timeout)
  monitor.kill if monitor.alive?
  monitor.join(SolidObjects.configuration.supervisor_monitor_interval)
end

#stop_retention ⇒ void

This method returns an undefined value.

RBS:

  • () -> void



203
204
205
206
207
208
209
210
 
# File 'lib/solid_objects/supervisor.rb', line 203

def stop_retention
  retention = @retention
  @retention = nil
  return unless retention

  retention.join(SolidObjects.configuration.shutdown_timeout)
  retention.kill if retention.alive?
end

#supervise(component) ⇒ Thread

RBS:

  • (untyped) -> Thread

Parameters:

  • (Object)

Returns:

  • (Thread) 


223
224
225
 
# File 'lib/solid_objects/supervisor.rb', line 223

def supervise(component)
  Thread.new { component.run }
end

#thread_error(thread) ⇒ String?

RBS:

  • (Thread?) -> String?

Parameters:

  • (Thread, nil)

Returns:

  • (String, nil) 


135
136
137
138
139
140
 
# File 'lib/solid_objects/supervisor.rb', line 135

def thread_error(thread)
  thread&.join
  nil
rescue => error
  error.class.name
end

#wait_for_next_retention(failures) ⇒ void

This method returns an undefined value.

Sleeping the whole interval would make shutdown wait out an hour-long nap, so the pause is taken in short steps that notice a stop request.

RBS:

  • (Integer) -> void

Parameters:

  • (Integer) 


167
168
169
170
171
172
173
 
# File 'lib/solid_objects/supervisor.rb', line 167

def wait_for_next_retention(failures)
  deadline = monotonic_now + retention_pause(failures)
  step = SolidObjects.configuration.supervisor_monitor_interval
  while @started && monotonic_now < deadline
    sleep [ step, deadline - monotonic_now ].min
  end
end