Integrate controller events with child waits.

Assisted-By: devx/3236e566-7538-432e-a30a-2bdf37265ed4

Author: samuel-williams-shopify

lib/async/container.rb

@@ -3,6 +3,7 @@
 # Released under the MIT License.
 # Copyright, 2017-2025, by Samuel Williams.
 
+require_relative "container/events"
 require_relative "container/signals"
 require_relative "container/controller"
 

lib/async/container/controller.rb

@@ -9,6 +9,8 @@
 require_relative "statistics"
 require_relative "notify"
 require_relative "policy"
+require_relative "events"
+require_relative "signals"
 
 module Async
 	module Container
@@ -41,10 +43,29 @@ def initialize(notify: Notify.open!, container_class: Container, graceful_stop:
 				@graceful_stop = graceful_stop
 
 				@container = nil
-				@signals = {}
+				@events = Events.new
+				@signals = Signals.new(@events)
 
 				self.trap(SIGHUP) do
 					self.restart
+				rescue SetupError => error
+					Console.error(self, error)
+				end
+
+				self.trap(SIGUSR1) do
+					self.reload
+				rescue SetupError => error
+					Console.error(self, error)
+				end
+
+				self.trap(SIGINT) do
+					self.stop
+					:stop
+				end
+
+				self.trap(SIGTERM) do
+					self.stop
+					:stop
 				end
 			end
 
@@ -80,7 +101,7 @@ def to_s
 			# @parameters signal [Symbol] The signal to trap, e.g. `:INT`.
 			# @parameters block [Proc] The signal handler to invoke.
 			def trap(signal, &block)
-				@signals[signal] = block
+				@signals.trap(signal, &block)
 			end
 
 			# Create a policy for managing child lifecycle events.
@@ -117,13 +138,19 @@ def setup(container)
 			end
 
 			# Start the container unless it's already running.
+			# @returns [Generic] The container.
 			def start
 				unless @container
 					Console.info(self, "Controller starting...")
-					self.restart
+					
+					if self.restart(@events) == :event
+						return
+					end
 				end
 
 				Console.info(self, "Controller started.")
+
+				return @container
 			end
 
 			# Stop the container if it's running.
@@ -135,7 +162,7 @@ def stop(graceful = @graceful_stop)
 
 			# Restart the container. A new container is created, and if successful, any old container is terminated gracefully.
 			# This is equivalent to a blue-green deployment.
-			def restart
+			def restart(events = nil)
 				if @container
 					@notify&.restarting!
 
@@ -156,7 +183,11 @@ def restart
 
 				# Wait for all child processes to enter the ready state.
 				Console.info(self, "Waiting for startup...")
-				container.wait_until_ready
+				
+				if container.wait_until_ready(events) == :event
+					return :event
+				end
+				
 				Console.info(self, "Finished startup.")
 
 				if container.failed?
@@ -195,7 +226,7 @@ def reload
 				begin
 					self.setup(@container)
 				rescue
-					raise SetupError, container
+					raise SetupError, @container
 				end
 
 				# Wait for all child processes to enter the ready state.
@@ -212,66 +243,28 @@ def reload
 				end
 			end
 
-			# Enter the controller run loop, trapping `SIGINT` and `SIGTERM`.
+			# Enter the controller run loop.
 			def run
 				@notify&.status!("Initializing controller...")
 
-				with_signal_handlers do
+				@signals.trapped do
 					self.start
 
+					while event = @events.pop(timeout: 0)
+						return if event.call == :stop
+					end
+					
 					while @container&.running?
-						begin
-							@container.wait
-						rescue SignalException => exception
-							if handler = @signals[exception.signo]
-								begin
-									handler.call
-								rescue SetupError => error
-									Console.error(self, error)
-								end
-							else
-								raise
-							end
+						@container.wait(@events)
+						
+						while event = @events.pop(timeout: 0)
+							return if event.call == :stop
 						end
 					end
 				end
-			rescue Interrupt
-				self.stop
-			rescue Terminate
-				self.stop(false)
 			ensure
 				self.stop(false)
 			end
-			
-			private def with_signal_handlers
-				# I thought this was the default... but it doesn't always raise an exception unless you do this explicitly.
-				
-				interrupt_action = Signal.trap(:INT) do
-					# We use `Thread.current.raise(...)` so that exceptions are filtered through `Thread.handle_interrupt` correctly.
-					# $stderr.puts "Received INT signal, interrupting...", caller
-					::Thread.current.raise(Interrupt)
-				end
-				
-				# SIGTERM behaves the same as SIGINT by default.
-				terminate_action = Signal.trap(:TERM) do
-					# $stderr.puts "Received TERM signal, interrupting...", caller
-					::Thread.current.raise(Interrupt)  # Same as SIGINT
-				end
-				
-				hangup_action = Signal.trap(:HUP) do
-					# $stderr.puts "Received HUP signal, restarting...", caller
-					::Thread.current.raise(Restart)
-				end
-				
-				::Thread.handle_interrupt(SignalException => :never) do
-					yield
-				end
-			ensure
-				# Restore the interrupt handler:
-				Signal.trap(:INT, interrupt_action)
-				Signal.trap(:TERM, terminate_action)
-				Signal.trap(:HUP, hangup_action)
-			end
 		end
 	end
 end

lib/async/container/events.rb

@@ -0,0 +1,59 @@
+# frozen_string_literal: true
+
+# Released under the MIT License.
+# Copyright, 2026, by Samuel Williams.
+
+module Async
+	module Container
+		# Represents a queue of events that can wake `IO.select`.
+		class Events
+			# Initialize the event queue.
+			def initialize
+				@queue = ::Thread::Queue.new
+				@input, @output = ::IO.pipe
+				@io = @input
+			end
+			
+			# @attribute [IO] The readable end of the event pipe.
+			attr :input
+			
+			# @attribute [IO] The readable end used to wait for events.
+			attr :io
+			
+			# Enqueue an event and wake any waiter.
+			# @parameter event [Object] The event to enqueue.
+			def <<(event)
+				@queue << event
+				
+				begin
+					@output.write_nonblock(".")
+				rescue ::IO::WaitWritable
+					# The pipe is already full, so any select waiter is already awake:
+				end
+				
+				return self
+			end
+			
+			# Remove and return the next queued event.
+			# @parameter non_block [Boolean] Whether to raise if no event is ready.
+			# @parameter timeout [Numeric | Nil] The maximum time to wait for an event.
+			def pop(non_block = false, timeout: nil)
+				if timeout
+					event = @queue.pop(non_block, timeout: timeout)
+				else
+					event = @queue.pop(non_block)
+				end
+				
+				return unless event
+				
+				@input.read_nonblock(1)
+				
+				return event
+			rescue ::IO::WaitReadable
+				return event if event
+				
+				raise
+			end
+		end
+	end
+end

lib/async/container/generic.rb

@@ -106,13 +106,13 @@ def stopping?
 			# Sleep until some state change occurs or the specified duration elapses.
 			#
 			# @parameter duration [Numeric] the maximum amount of time to sleep for.
-			def sleep(duration = nil)
-				@group.sleep(duration)
+			def sleep(duration = nil, events = nil)
+				@group.sleep(duration, events)
 			end
 
 			# Wait until all spawned tasks are completed.
-			def wait
-				@group.wait
+			def wait(events = nil)
+				@group.wait(events)
 			end
 
 			# Gracefully interrupt all child instances.
@@ -135,7 +135,7 @@ def status?(flag)
 
 			# Wait until all the children instances have indicated that they are ready.
 			# @returns [Boolean] The children all became ready.
-			def wait_until_ready
+			def wait_until_ready(events = nil)
 				while true
 					Console.debug(self) do |buffer|
 						buffer.puts "Waiting for ready:"
@@ -144,7 +144,9 @@ def wait_until_ready
 						end
 					end
 
-					self.sleep
+					if self.sleep(nil, events) == :event
+						return :event
+					end
 
 					if self.status?(:ready)
 						Console.debug(self) do |buffer|

lib/async/container/group.rb

@@ -57,14 +57,16 @@ def empty?
 			end
 
 			# Sleep for at most the specified duration until some state change occurs.
-			def sleep(duration)
-				self.wait_for_children(duration)
+			def sleep(duration, events = nil)
+				self.wait_for_children(duration, events)
 			end
 
 			# Begin any outstanding queued processes and wait for them indefinitely.
-			def wait
+			def wait(events = nil)
 				with_health_checks do |duration|
-					self.wait_for_children(duration)
+					if self.wait_for_children(duration, events) == :event
+						return
+					end
 				end
 			end
 
@@ -211,30 +213,40 @@ def wait_for(channel)
 
 			protected
 
-			def wait_for_children(duration = nil)
+			def wait_for_children(duration = nil, events = nil)
 				# This log is a bit noisy and doesn't really provide a lot of useful information:
 				Console.debug(self, "Waiting for children...", duration: duration, running: @running)
 
 				unless @running.empty?
 					# Maybe consider using a proper event loop here:
-					if ready = self.select(duration)
+					if ready = self.select(duration, events)
+						if events && ready.delete(events.io)
+							result = :event
+						end
+						
 						ready.each do |io|
 							if fiber = @running[io]
 								# This method can be re-entered. While resuming a fiber, a policy hook may be invoked, which may invoke operations on the container. In that case, select may be called again on the same set of waiting fibers. On returning, those fibers may have already completed and removed themselves from @running, so we need to check for that.
 								fiber.resume
 							end
 						end
+						
+						return result
 					end
 				end
 			end
 
 			# Wait for a child process to exit OR a signal to be received.
-			def select(duration)
-				::Thread.handle_interrupt(SignalException => :immediate) do
-					readable, _, _ = ::IO.select(@running.keys, nil, nil, duration)
-					
-					return readable
+			def select(duration, events = nil)
+				waiting = @running.keys
+				
+				if events
+					waiting << events.io
 				end
+				
+				readable, _, _ = ::IO.select(waiting, nil, nil, duration)
+				
+				return readable
 			end
 		end
 	end

lib/async/container/signals.rb

@@ -3,6 +3,8 @@
 # Released under the MIT License.
 # Copyright, 2026, by Samuel Williams.
 
+require_relative "events"
+
 module Async
 	module Container
 		# Represents a collection of process signal handlers which enqueue events.
@@ -27,13 +29,13 @@ def call
 			end
 
 			# Initialize the signal handler collection.
-			# @parameter events [Thread::Queue] The queue used to receive signal events.
-			def initialize(events = ::Thread::Queue.new)
+			# @parameter events [Events] The queue used to receive signal events.
+			def initialize(events = Events.new)
 				@events = events
 				@handlers = {}
 			end
 
-			# @attribute [Thread::Queue] The queue used to receive signal events.
+			# @attribute [Events] The queue used to receive signal events.
 			attr :events
 
 			# Register a signal handler.