Add health_check_timeout for detecting hung processes.

Author: ioquatix

async-container.gemspec

@@ -24,5 +24,5 @@ Gem::Specification.new do |spec|
 
 	spec.required_ruby_version = ">= 3.1"
 
-	spec.add_dependency "async", "~> 2.10"
+	spec.add_dependency "async", "~> 2.22"
 end

examples/health_check/test.rb

@@ -0,0 +1,39 @@
+#!/usr/bin/env ruby
+# frozen_string_literal: true
+
+# Released under the MIT License.
+# Copyright, 2022, by Anton Sozontov.
+# Copyright, 2024, by Samuel Williams.
+
+require "../../lib/async/container/controller"
+
+NAMES = [
+	"Cupcake", "Donut", "Eclair", "Froyo", "Gingerbread", "Honeycomb", "Ice Cream Sandwich", "Jelly Bean", "KitKat", "Lollipop", "Marshmallow", "Nougat", "Oreo", "Pie", "Apple Tart"
+]
+
+class Controller < Async::Container::Controller
+	def setup(container)
+		container.run(count: 10, restart: true, health_check_timeout: 1) do |instance|
+			if container.statistics.failed?
+				Console.debug(self, "Child process restarted #{container.statistics.restarts} times.")
+			else
+				Console.debug(self, "Child process started.")
+			end
+			
+			instance.name = NAMES.sample
+			
+			instance.ready!
+			
+			while true
+				# Must update status more frequently than health check timeout...
+				sleep(rand*1.2)
+				
+				instance.ready!
+			end
+		end
+	end
+end
+
+controller = Controller.new # (container_class: Async::Container::Threaded)
+
+controller.run

examples/test.rb

@@ -116,6 +116,24 @@ def initialize(name: nil)
 					self.close_write
 				end
 
+				# Convert the child process to a hash, suitable for serialization.
+				#
+				# @returns [Hash] The request as a hash.
+				def as_json(...)
+					{
+						name: @name,
+						pid: @pid,
+						status: @status&.to_i,
+					}
+				end
+				
+				# Convert the request to JSON.
+				#
+				# @returns [String] The request as JSON.
+				def to_json(...)
+					as_json.to_json(...)
+				end
+				
 				# Set the name of the process.
 				# Invokes {::Process.setproctitle} if invoked in the child process.
 				def name= value

lib/async/container/forked.rb

@@ -4,6 +4,7 @@
 # Copyright, 2019-2024, by Samuel Williams.
 
 require "etc"
+require "async/clock"
 
 require_relative "group"
 require_relative "keyed"
@@ -141,7 +142,8 @@ def stop(timeout = true)
 			# @parameter name [String] The name of the child instance.
 			# @parameter restart [Boolean] Whether to restart the child instance if it fails.
 			# @parameter key [Symbol] A key used for reloading child instances.
-			def spawn(name: nil, restart: false, key: nil, &block)
+			# @parameter health_check_timeout [Numeric | Nil] The maximum time a child instance can run without updating its state, before it is terminated as unhealthy.
+			def spawn(name: nil, restart: false, key: nil, health_check_timeout: nil, &block)
 				name ||= UNNAMED
 
 				if mark?(key)
@@ -157,9 +159,24 @@ def spawn(name: nil, restart: false, key: nil, &block)
 
 						state = insert(key, child)
 
+						# If a health check 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
+							age_clock = state[:age] = Clock.start
+						end
+						
 						begin
 							status = @group.wait_for(child) do |message|
-								state.update(message)
+								case message
+								when :health_check!
+									if health_check_timeout&.<(age_clock.total)
+										Console.warn(self, "Child failed health check!", child: child, age: age_clock.total, health_check_timeout: health_check_timeout)
+										# If the child has failed the health check, we assume the worst and terminate it (SIGTERM).
+										child.terminate!
+									end
+								else
+									state.update(message)
+									age_clock&.reset!
+								end
 							end
 						ensure
 							delete(key, child)

lib/async/container/generic.rb

@@ -13,7 +13,12 @@ module Container
 		# Manages a group of running processes.
 		class Group
 			# Initialize an empty group.
-			def initialize
+			#
+			# @parameter health_check_interval [Numeric | Nil] The (biggest) interval at which health checks are performed.
+			def initialize(health_check_interval: 1.0)
+				@health_check_interval = health_check_interval
+				
+				# The running fibers, indexed by IO:
 				@running = {}
 
 				# This queue allows us to wait for processes to complete, without spawning new processes as a result.
@@ -57,8 +62,36 @@ def sleep(duration)
 			def wait
 				self.resume
 
-				while self.running?
-					self.wait_for_children
+				with_health_checks do |duration|
+					self.wait_for_children(duration)
+				end
+			end
+			
+			private def with_health_checks
+				if @health_check_interval
+					health_check_clock = Clock.start
+					
+					while self.running?
+						duration = [@health_check_interval - health_check_clock.total, 0].max
+						
+						yield duration
+						
+						if health_check_clock.total > @health_check_interval
+							self.health_check!
+							health_check_clock.reset!
+						end
+					end
+				else
+					while self.running?
+						yield nil
+					end
+				end
+			end
+			
+			# Perform a health check on all running processes.
+			def health_check!
+				@running.each_value do |fiber|
+					fiber.resume(:health_check!)
 				end
 			end
 
@@ -119,15 +152,19 @@ def wait_for(channel)
 				@running[io] = Fiber.current
 
 				while @running.key?(io)
+					# Wait for some event on the channel:
 					result = Fiber.yield
 
 					if result == Interrupt
 						channel.interrupt!
 					elsif result == Terminate
 						channel.terminate!
+					elsif result
+						yield result
 					elsif message = channel.receive
 						yield message
 					else
+						# Wait for the channel to exit:
 						return channel.wait
 					end
 				end

lib/async/container/group.rb

@@ -125,6 +125,23 @@ def initialize(name: nil)
 					end
 				end
 
+				# Convert the child process to a hash, suitable for serialization.
+				#
+				# @returns [Hash] The request as a hash.
+				def as_json(...)
+					{
+						name: @thread.name,
+						status: @status&.as_json,
+					}
+				end
+				
+				# Convert the request to JSON.
+				#
+				# @returns [String] The request as JSON.
+				def to_json(...)
+					as_json.to_json(...)
+				end
+				
 				# Set the name of the thread.
 				# @parameter value [String] The name to set.
 				def name= value
@@ -191,6 +208,14 @@ def success?
 						@error.nil?
 					end
 
+					def as_json(...)
+						if @error
+							@error.inspect
+						else
+							true
+						end
+					end
+					
 					# A human readable representation of the status.
 					def to_s
 						"\#<#{self.class} #{success? ? "success" : "failure"}>"