Class: SolidObjects::Supervisor

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

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) 


13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
 
# File 'lib/solid_objects/supervisor.rb', line 13

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
  @lifecycle = Thread::Mutex.new
end

Instance Attribute Details

#components ⇒ Object (readonly)

Returns the value of attribute components.

Returns:

  • (Object) 


74
75
76
 
# File 'lib/solid_objects/supervisor.rb', line 74

def components
  @components
end

#threads ⇒ Object (readonly)

Returns the value of attribute threads.

Returns:

  • (Object) 


74
75
76
 
# File 'lib/solid_objects/supervisor.rb', line 74

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:



178
179
180
181
182
183
184
185
186
187
188
 
# File 'lib/solid_objects/supervisor.rb', line 178

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



137
138
139
140
141
142
143
144
 
# File 'lib/solid_objects/supervisor.rb', line 137

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



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

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



82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
 
# File 'lib/solid_objects/supervisor.rb', line 82

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) 


202
203
204
 
# File 'lib/solid_objects/supervisor.rb', line 202

def monotonic_now
  ::Process.clock_gettime(::Process::CLOCK_MONOTONIC)
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



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

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



105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
 
# File 'lib/solid_objects/supervisor.rb', line 105

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

#run ⇒ void

This method returns an undefined value.

RBS:

  • () -> void



33
34
35
36
37
38
 
# File 'lib/solid_objects/supervisor.rb', line 33

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

#start ⇒ void

This method returns an undefined value.

RBS:

  • () -> void



41
42
43
44
45
46
47
48
 
# File 'lib/solid_objects/supervisor.rb', line 41

def start
  return if @started

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

#stop ⇒ void

This method returns an undefined value.

RBS:

  • () -> void



51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
 
# File 'lib/solid_objects/supervisor.rb', line 51

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



155
156
157
158
159
160
161
162
163
 
# File 'lib/solid_objects/supervisor.rb', line 155

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

#supervise(component) ⇒ Thread

RBS:

  • (untyped) -> Thread

Parameters:

  • (Object)

Returns:

  • (Thread) 


147
148
149
 
# File 'lib/solid_objects/supervisor.rb', line 147

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

#thread_error(thread) ⇒ String?

RBS:

  • (Thread?) -> String?

Parameters:

  • (Thread, nil)

Returns:

  • (String, nil) 


129
130
131
132
133
134
 
# File 'lib/solid_objects/supervisor.rb', line 129

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