Class: SolidObjects::Worker

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

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(process_registry: ProcessRegistry.new) ⇒ Worker

Returns a new instance of Worker.

RBS:

  • (?process_registry: ProcessRegistry) -> void

Parameters:



20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
 
# File 'lib/solid_objects/worker.rb', line 20

def initialize(process_registry: ProcessRegistry.new)
  @process_registry = process_registry
  process_record = process_registry.register
  @activation_manager = ActivationManager.new(owner_id: process_record.id)
  @activations = {}
  @stopped = false
  @shutdown_requested = false
  @polling_backoff = PollingBackoff.new(
    minimum_interval: [
      SolidObjects.configuration.polling_interval,
      SolidObjects.configuration.lease_renewal_interval
    ].min,
    maximum_interval: [
      SolidObjects.configuration.idle_polling_interval,
      SolidObjects.configuration.lease_renewal_interval
    ].min,
    on_change: ->(transition) do
      SolidObjects.instrument(
        :"polling.interval_changed",
        role: "actors",
        **transition
      )
    end
  )
end

Instance Attribute Details

#activation_manager ⇒ Object (readonly)

Returns the value of attribute activation_manager.

Returns:

  • (Object) 


148
149
150
 
# File 'lib/solid_objects/worker.rb', line 148

def activation_manager
  @activation_manager
end

#activations ⇒ Object (readonly)

Returns the value of attribute activations.

Returns:

  • (Object) 


148
149
150
 
# File 'lib/solid_objects/worker.rb', line 148

def activations
  @activations
end

#polling_backoff ⇒ Object (readonly)

Returns the value of attribute polling_backoff.

Returns:

  • (Object) 


148
149
150
 
# File 'lib/solid_objects/worker.rb', line 148

def polling_backoff
  @polling_backoff
end

#process_registry ⇒ Object (readonly)

Returns the value of attribute process_registry.

Returns:

  • (Object) 


148
149
150
 
# File 'lib/solid_objects/worker.rb', line 148

def process_registry
  @process_registry
end

Instance Method Details

#cached_ready_activation ⇒ Activation?

RBS:

  • () -> Activation?

Returns:



151
152
153
 
# File 'lib/solid_objects/worker.rb', line 151

def cached_ready_activation
  activations.each_value.find(&:ready?)
end

#current_polling_interval ⇒ Float

RBS:

  • () -> Float

Returns:

  • (Float) 


142
143
144
 
# File 'lib/solid_objects/worker.rb', line 142

def current_polling_interval
  polling_backoff.current_interval
end

#maintain_cached_activations ⇒ void

This method returns an undefined value.

RBS:

  • () -> void



156
157
158
159
160
161
162
163
164
165
166
 
# File 'lib/solid_objects/worker.rb', line 156

def maintain_cached_activations
  activations.each_value do |activation|
    if activation.idle?
      release_activation(activation)
    elsif activation.lease_renewal_due?
      activation.renew_lease
    end
  rescue LostActivation
    release_activation(activation)
  end
end

#release_activation(activation) ⇒ void

This method returns an undefined value.

RBS:

  • (Activation) -> void

Parameters:



169
170
171
172
173
 
# File 'lib/solid_objects/worker.rb', line 169

def release_activation(activation)
  activations.delete(activation.lease.instance_id)
  activation.yield_ready_messages if activation.pass_exhausted?
  activation.deactivate
end

#request_shutdown ⇒ void

This method returns an undefined value.

RBS:

  • () -> void



115
116
117
118
 
# File 'lib/solid_objects/worker.rb', line 115

def request_shutdown
  @shutdown_requested = true
  SolidObjects.wake_up.signal
end

#run ⇒ void

This method returns an undefined value.

RBS:

  • () -> void



91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
 
# File 'lib/solid_objects/worker.rb', line 91

def run
  ProcessRegistry.warn_if_polling_is_only_cross_process_wake_up

  until shutdown_requested?
    wake_up = SolidObjects.wake_up
    watch = wake_up.respond_to?(:watch) ? wake_up.watch : wake_up
    processed = run_once
    if processed.positive?
      polling_backoff.reset(:work)
      next
    end

    notified = watch.wait(timeout: current_polling_interval)
    if notified == false
      polling_backoff.record_idle
    else
      polling_backoff.reset(:wake_up)
    end
  end
ensure
  stop
end

#run_once ⇒ Integer

RBS:

  • () -> Integer

Returns:

  • (Integer) 


47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
 
# File 'lib/solid_objects/worker.rb', line 47

def run_once
  return 0 if stopped?

  process_registry.heartbeat
  maintain_cached_activations
  activation = cached_ready_activation || activation_manager.claim_next
  return 0 unless activation

  activations[activation.lease.instance_id] = activation
  processed = LeaseRenewer.new(
    activation:,
    process_registry:
  ).around { activation.drain }
  release_activation(activation) if activation.pass_exhausted?
  processed
rescue ActorDestroyed
  release_activation(activation) if activation
  0
rescue StateMigrationError, ApplicationWriteForbidden, LostActivation => error
  release_activation(activation) if activation
  SolidObjects.configuration.logger.error(
    event: "solid_objects.worker.error",
    process_id: process_registry.process_record&.id,
    error_class: error.class.name,
    error_message: error.message
  )
  0
end

#run_until_idle(max_passes: 1_000) ⇒ Integer

RBS:

  • (?max_passes: Integer) -> Integer

Parameters:

  • max_passes: (Integer) (defaults to: 1_000)

Returns:

  • (Integer) 


77
78
79
80
81
82
83
84
85
86
87
88
 
# File 'lib/solid_objects/worker.rb', line 77

def run_until_idle(max_passes: 1_000)
  total = 0

  max_passes.times do
    processed = run_once
    break if processed.zero?

    total += processed
  end

  total
end

#shutdown_requested? ⇒ Boolean

RBS:

  • () -> bool

Returns:

  • (Boolean) 


137
138
139
 
# File 'lib/solid_objects/worker.rb', line 137

def shutdown_requested?
  @shutdown_requested
end

#stop ⇒ void

This method returns an undefined value.

RBS:

  • () -> void



121
122
123
124
125
126
127
128
129
 
# File 'lib/solid_objects/worker.rb', line 121

def stop
  return if stopped?

  @stopped = true
  process_registry.start_draining
  activations.each_value(&:deactivate)
  activations.clear
  process_registry.stop
end

#stopped? ⇒ Boolean

RBS:

  • () -> bool

Returns:

  • (Boolean) 


132
133
134
 
# File 'lib/solid_objects/worker.rb', line 132

def stopped?
  @stopped
end