class Generic

def self.run(...)
	self.new.run(...)
end

parameter　count　Integer

The number of instances to start.

def run(count: Container.processor_count, **options, &block)
	count.times do
		spawn(**options, &block)
	end
	
	return self
end

parameter　policy　Policy

The policy to use for managing child lifecycle events.

parameter　options　Hash

Options passed to the class Async::Container::Group instance.

def initialize(policy: Policy::DEFAULT, **options)
	@group = Group.new(**options)
	@stopping = false
	
	@state = {}
	
	@policy = policy
	@statistics = @policy.make_statistics
	@keyed = {}
end

attribute　Group

The group of running children instances.

returns　Integer

The number of running children instances.

def size
	@group.size
end

attribute　Hash(Child, Hash)

The state of each child instance.

attribute　Policy

The policy for managing child lifecycle events.

returns　String

def to_s
	"#{self.class} with #{@statistics.spawns} spawns and #{@statistics.failures} failures."
end

def [] key
	@keyed[key]
end

attribute　Statistics

returns　Boolean

def failed?
	@statistics.failed?
end

def running?
	@group.running?
end

returns　Boolean

def stopping?
	@stopping
end

parameter　duration　Numeric

the maximum amount of time to sleep for.

def sleep(duration = nil)
	@group.sleep(duration)
end

def wait
	@group.wait
end

def interrupt
	# We must enter the stopping state before signalling the children. Interrupting a child causes it to drain and exit, but the main run loop will respawn any child that exits while `restart: true` and the container is not stopping (see the `restart && !@stopping` gate in `#run`). Without setting this flag, an interrupted child immediately respawns, so the container never drains and `#wait` never returns.
	#
	# This matters most for `Hybrid` containers: a `SIGINT`/`SIGTERM` delivered to a fork is translated into a call to `#interrupt` on the inner threaded container, which typically runs with `restart: true` (the default for `async-service` managed services). If `#interrupt` did not set this flag, the inner threads would drain, exit, and respawn in a loop, so a single signal would never terminate the fork. Setting `@stopping = true` here makes `#interrupt` behave as the start of a graceful shutdown: children drain and exit, are not respawned, and the fork terminates - consistent with how `Forked` and `Threaded` containers handle a single interrupt.
	@stopping = true
	@group.interrupt
end

returns　Boolean

def status?(flag)
	# This also returns true if all processes have exited/failed:
	@state.all?{|_, state| state[flag]}
end

returns　Boolean

The children all became ready.

def wait_until_ready
	while true
		Console.debug(self) do |buffer|
			buffer.puts "Waiting for ready:"
			@state.each do |child, state|
				buffer.puts "\t#{child.inspect}: #{state}"
			end
		end
		
		self.sleep
		
		if self.status?(:ready)
			Console.debug(self) do |buffer|
				buffer.puts "All ready:"
				@state.each do |child, state|
					buffer.puts "\t#{child.inspect}: #{state}"
				end
			end
			
			return true
		end
	end
end

parameter　timeout　Boolean | Numeric

Whether to stop gracefully, or a specific timeout.

def stop(timeout = true)
	if @stopping
		Console.warn(self, "Container is already stopping!")
		return
	end
	
	Console.info(self, "Stopping container...", timeout: timeout)
	@stopping = true
	@group.stop(timeout)
	
	if @group.running?
		Console.warn(self, "Group is still running after stopping it!")
	else
		Console.info(self, "Group has stopped.")
	end
rescue => error
	Console.error(self, "Error while stopping container!", exception: error)
	raise
end

parameter　name　String

The name of the child instance.

parameter　restart　Boolean

Whether to restart the child instance if it fails.

parameter　key　Symbol

An optional key used to look up (via Async::Container::Generic#[]) and reuse the child instance.

parameter　health_check_timeout　Numeric | Nil

The maximum time a child instance can run without updating its state, before it is terminated as unhealthy.

parameter　startup_timeout　Numeric | Nil

The maximum time a child instance can run without becoming ready, before it is terminated as unhealthy.

def spawn(name: nil, restart: false, key: nil, health_check_timeout: nil, startup_timeout: nil, &block)
	name ||= UNNAMED
	
	if reuse?(key)
		Console.debug(self, "Reusing existing child.", child: {key: key, name: name})
		return false
	end
	
	@statistics.spawn!
	
	fiber do
		until @stopping
			Console.debug(self, "Starting child...", child: {key: key, name: name, restart: restart, health_check_timeout: health_check_timeout}, statistics: @statistics)
			
			child = self.start(name, &block)
			state = insert(key, child)
			
			# Notify policy of spawn
			begin
				@policy.child_spawn(self, child, name: name, key: key)
			rescue => error
				Console.error(self, "Policy error in child_spawn!", exception: error)
			end
			
			Console.debug(self, "Started child.", child: child, spawn: {key: key, restart: restart, health_check_timeout: health_check_timeout}, statistics: @statistics)
			
			# If a health check or startup timeout is specified, we will monitor the child process and terminate it if it does not update its state within the specified time.
			if health_check_timeout || startup_timeout
				age_clock = state[:age] = Clock.start
			end
			
			status = nil
			
			begin
				status = @group.wait_for(child) do |message|
					case message
					when :health_check!
						if state[:ready]
							# If a health check timeout is specified, we will monitor the child process and terminate it if it does not update its state within the specified time.
							if health_check_timeout
								if health_check_timeout < age_clock.total
									health_check_failed(child, age_clock, health_check_timeout)
								end
							end
						else
							# If a startup timeout is specified, we will monitor the child process and terminate it if it does not become ready within the specified time.
							if startup_timeout
								if startup_timeout < age_clock.total
									startup_failed(child, age_clock, startup_timeout)
								end
							end
						end
					else
						state.update(message)
						
						# Reset the age clock if the child has become ready:
						if state[:ready]
							age_clock&.reset!
						end
					end
				end
			rescue => error
				Console.error(self, "Error during child process management!", exception: error, stopping: @stopping)
			ensure
				delete(key, child)
			end
			
			if status&.success?
				Console.debug(self, "Child exited successfully.", status: status, stopping: @stopping)
			else
				@statistics.failure!
				Console.error(self, "Child exited with error!", status: status, stopping: @stopping)
			end
			
			# Notify policy of exit (after statistics are updated):
			begin
				@policy.child_exit(self, child, status, name: name, key: key)
			rescue => error
				Console.error(self, "Policy error in child_exit!", exception: error)
			end
			
			if restart && !@stopping
				@statistics.restart!
			else
				break
			end
		end
	end.resume
	
	return true
end

deprecated

Please use Async::Container::Generic#spawn or Async::Container::Generic.run instead.

def async(**options, &block)
	# warn "#{self.class}##{__method__} is deprecated, please use `spawn` or `run` instead.", uplevel: 1
	
	require "async"
	
	spawn(**options) do |instance|
		Async(instance, &block)
	end
end

def reload
	yield
end

def reuse?(key)
	if key
		@keyed.key?(key)
	else
		false
	end
end

def key?(key)
	if key
		@keyed.key?(key)
	end
end

def insert(key, child)
	if key
		@keyed[key] = child
	end
	
	state = {}
	
	@state[child] = state
	
	return state
end

def delete(key, child)
	if key
		@keyed.delete(key)
	end
	
	@state.delete(child)
end