Class: Async::Container::Generic
- Inherits:
-
Object
- Object
- Async::Container::Generic
- Defined in:
- lib/async/container/generic.rb
Overview
A base class for implementing containers.
Constant Summary collapse
- UNNAMED =
"Unnamed"
Instance Attribute Summary collapse
-
#state ⇒ Object
readonly
Returns the value of attribute state.
-
#statistics ⇒ Object
Statistics relating to the behavior of children instances.
Class Method Summary collapse
Instance Method Summary collapse
-
#[](key) ⇒ Object
Look up a child process by key.
-
#async(**options, &block) ⇒ Object
deprecated
Deprecated.
Please use #spawn or Generic.run instead.
-
#failed? ⇒ Boolean
Whether any failures have occurred within the container.
-
#initialize(**options) ⇒ Generic
constructor
A new instance of Generic.
-
#key?(key) ⇒ Boolean
Whether a child instance exists for the given key.
-
#mark?(key) ⇒ Boolean
Mark the container’s keyed instance which ensures that it won’t be discarded.
-
#reload ⇒ Object
Reload the container’s keyed instances.
-
#run(count: Container.processor_count, **options, &block) ⇒ Object
Run multiple instances of the same block in the container.
-
#running? ⇒ Boolean
Whether the container has running children instances.
-
#sleep(duration = nil) ⇒ Object
Sleep until some state change occurs.
-
#spawn(name: nil, restart: false, key: nil, &block) ⇒ Object
Spawn a child instance into the container.
-
#status?(flag) ⇒ Boolean
Returns true if all children instances have the specified status flag set.
-
#stop(timeout = true) ⇒ Object
Stop the children instances.
-
#to_s ⇒ Object
A human readable representation of the container.
-
#wait ⇒ Object
Wait until all spawned tasks are completed.
-
#wait_until_ready ⇒ Object
Wait until all the children instances have indicated that they are ready.
Constructor Details
#initialize(**options) ⇒ Generic
Returns a new instance of Generic.
59 60 61 62 63 64 65 66 67 |
# File 'lib/async/container/generic.rb', line 59 def initialize(**) @group = Group.new @running = true @state = {} @statistics = Statistics.new @keyed = {} end |
Instance Attribute Details
#state ⇒ Object (readonly)
Returns the value of attribute state.
69 70 71 |
# File 'lib/async/container/generic.rb', line 69 def state @state end |
#statistics ⇒ Object
Statistics relating to the behavior of children instances.
85 86 87 |
# File 'lib/async/container/generic.rb', line 85 def statistics @statistics end |
Class Method Details
.run(*arguments, **options, &block) ⇒ Object
53 54 55 |
# File 'lib/async/container/generic.rb', line 53 def self.run(*arguments, **, &block) self.new.run(*arguments, **, &block) end |
Instance Method Details
#[](key) ⇒ Object
Look up a child process by key. A key could be a symbol, a file path, or something else which the child instance represents.
79 80 81 |
# File 'lib/async/container/generic.rb', line 79 def [] key @keyed[key]&.value end |
#async(**options, &block) ⇒ Object
209 210 211 212 213 |
# File 'lib/async/container/generic.rb', line 209 def async(**, &block) spawn(**) do |instance| Async::Reactor.run(instance, &block) end end |
#failed? ⇒ Boolean
Whether any failures have occurred within the container.
89 90 91 |
# File 'lib/async/container/generic.rb', line 89 def failed? @statistics.failed? end |
#key?(key) ⇒ Boolean
Whether a child instance exists for the given key.
244 245 246 247 248 |
# File 'lib/async/container/generic.rb', line 244 def key?(key) if key @keyed.key?(key) end end |
#mark?(key) ⇒ Boolean
Mark the container’s keyed instance which ensures that it won’t be discarded.
231 232 233 234 235 236 237 238 239 240 241 |
# File 'lib/async/container/generic.rb', line 231 def mark?(key) if key if value = @keyed[key] value.mark! return true end end return false end |
#reload ⇒ Object
Reload the container’s keyed instances.
216 217 218 219 220 221 222 223 224 225 226 227 228 |
# File 'lib/async/container/generic.rb', line 216 def reload @keyed.each_value(&:clear!) yield dirty = false @keyed.delete_if do |key, value| value.stop? && (dirty = true) end return dirty end |
#run(count: Container.processor_count, **options, &block) ⇒ Object
Run multiple instances of the same block in the container.
200 201 202 203 204 205 206 |
# File 'lib/async/container/generic.rb', line 200 def run(count: Container.processor_count, **, &block) count.times do spawn(**, &block) end return self end |
#running? ⇒ Boolean
Whether the container has running children instances.
94 95 96 |
# File 'lib/async/container/generic.rb', line 94 def running? @group.running? end |
#sleep(duration = nil) ⇒ Object
Sleep until some state change occurs.
100 101 102 |
# File 'lib/async/container/generic.rb', line 100 def sleep(duration = nil) @group.sleep(duration) end |
#spawn(name: nil, restart: false, key: nil, &block) ⇒ Object
Spawn a child instance into the container.
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 |
# File 'lib/async/container/generic.rb', line 154 def spawn(name: nil, restart: false, key: nil, &block) name ||= UNNAMED if mark?(key) Console.logger.debug(self) {"Reusing existing child for #{key}: #{name}"} return false end @statistics.spawn! fiber do while @running child = self.start(name, &block) state = insert(key, child) begin status = @group.wait_for(child) do || state.update() end ensure delete(key, child) end if status.success? Console.logger.info(self) {"#{child} exited with #{status}"} else @statistics.failure! Console.logger.error(self) {status} end if restart @statistics.restart! else break end end # ensure # Console.logger.error(self) {$!} if $! end.resume return true end |
#status?(flag) ⇒ Boolean
Returns true if all children instances have the specified status flag set. e.g. ‘:ready`. This state is updated by the process readiness protocol mechanism. See Notify::Client for more details.
113 114 115 116 |
# File 'lib/async/container/generic.rb', line 113 def status?(flag) # This also returns true if all processes have exited/failed: @state.all?{|_, state| state[flag]} end |
#stop(timeout = true) ⇒ Object
Stop the children instances.
139 140 141 142 143 144 145 146 147 148 |
# File 'lib/async/container/generic.rb', line 139 def stop(timeout = true) @running = false @group.stop(timeout) if @group.running? Console.logger.warn(self) {"Group is still running after stopping it!"} end ensure @running = true end |
#to_s ⇒ Object
A human readable representation of the container.
73 74 75 |
# File 'lib/async/container/generic.rb', line 73 def to_s "#{self.class} with #{@statistics.spawns} spawns and #{@statistics.failures} failures." end |
#wait ⇒ Object
Wait until all spawned tasks are completed.
105 106 107 |
# File 'lib/async/container/generic.rb', line 105 def wait @group.wait end |
#wait_until_ready ⇒ Object
Wait until all the children instances have indicated that they are ready.
120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 |
# File 'lib/async/container/generic.rb', line 120 def wait_until_ready while true Console.logger.debug(self) do |buffer| buffer.puts "Waiting for ready:" @state.each do |child, state| buffer.puts "\t#{child.class}: #{state.inspect}" end end self.sleep if self.status?(:ready) return true end end end |