Move implementation of Thread and Process into Threaded::Child and Process::Child respectively.

Author: ioquatix

fixtures/async/container/a_container.rb

@@ -3,6 +3,8 @@
 # Released under the MIT License.
 # Copyright, 2019-2024, by Samuel Williams.
 
+require "async"
+
 module Async
 	module Container
 		AContainer = Sus::Shared("a container") do
@@ -67,10 +69,15 @@ module Container
 				it "can stop the child process" do
 					container.spawn do
 						sleep(1)
+					rescue Interrupt
+						# Ignore.
 					end
 
 					expect(container).to be(:running?)
 
+					# TODO Investigate why without this, the interrupt can occur before the process is sleeping...
+					sleep 0.001
+					
 					container.stop
 
 					expect(container).not.to be(:running?)

lib/async/container/forked.rb

@@ -3,8 +3,11 @@
 # Released under the MIT License.
 # Copyright, 2017-2024, by Samuel Williams.
 
+require_relative "error"
+
 require_relative "generic"
-require_relative "process"
+require_relative "channel"
+require_relative "notify/pipe"
 
 module Async
 	module Container
@@ -15,11 +18,190 @@ def self.multiprocess?
 				true
 			end
 
+			# Represents a running child process from the point of view of the parent container.
+			class Child < Channel
+				# Represents a running child process from the point of view of the child process.
+				class Instance < Notify::Pipe
+					# Wrap an instance around the {Process} instance from within the forked child.
+					# @parameter process [Process] The process intance to wrap.
+					def self.for(process)
+						instance = self.new(process.out)
+						
+						# The child process won't be reading from the channel:
+						process.close_read
+						
+						instance.name = process.name
+						
+						return instance
+					end
+					
+					def initialize(io)
+						super
+						
+						@name = nil
+					end
+					
+					# Set the process title to the specified value.
+					# @parameter value [String] The name of the process.
+					def name= value
+						if @name = value
+							::Process.setproctitle(@name)
+						end
+					end
+					
+					# The name of the process.
+					# @returns [String]
+					def name
+						@name
+					end
+					
+					# Replace the current child process with a different one. Forwards arguments and options to {::Process.exec}.
+					# This method replaces the child process with the new executable, thus this method never returns.
+					def exec(*arguments, ready: true, **options)
+						if ready
+							self.ready!(status: "(exec)")
+						else
+							self.before_spawn(arguments, options)
+						end
+						
+						::Process.exec(*arguments, **options)
+					end
+				end
+				
+				# Fork a child process appropriate for a container.
+				# @returns [Process]
+				def self.fork(**options)
+					# $stderr.puts fork: caller
+					self.new(**options) do |process|
+						::Process.fork do
+							# We use `Thread.current.raise(...)` so that exceptions are filtered through `Thread.handle_interrupt` correctly.
+							Signal.trap(:INT) {::Thread.current.raise(Interrupt)}
+							Signal.trap(:TERM) {::Thread.current.raise(Terminate)}
+							Signal.trap(:HUP) {::Thread.current.raise(Restart)}
+							
+							# This could be a configuration option:
+							::Thread.handle_interrupt(SignalException => :immediate) do
+								yield Instance.for(process)
+							rescue Interrupt
+								# Graceful exit.
+							rescue Exception => error
+								Console.error(self, error)
+								
+								exit!(1)
+							end
+						end
+					end
+				end
+				
+				def self.spawn(*arguments, name: nil, **options)
+					self.new(name: name) do |process|
+						Notify::Pipe.new(process.out).before_spawn(arguments, options)
+						
+						::Process.spawn(*arguments, **options)
+					end
+				end
+				
+				# Initialize the process.
+				# @parameter name [String] The name to use for the child process.
+				def initialize(name: nil)
+					super()
+					
+					@name = name
+					@status = nil
+					@pid = nil
+					
+					@pid = yield(self)
+					
+					# The parent process won't be writing to the channel:
+					self.close_write
+				end
+				
+				# Set the name of the process.
+				# Invokes {::Process.setproctitle} if invoked in the child process.
+				def name= value
+					@name = value
+					
+					# If we are the child process:
+					::Process.setproctitle(@name) if @pid.nil?
+				end
+				
+				# The name of the process.
+				# @attribute [String]
+				attr :name
+				
+				# @attribute [Integer] The process identifier.
+				attr :pid
+				
+				# A human readable representation of the process.
+				# @returns [String]
+				def inspect
+					"\#<#{self.class} name=#{@name.inspect} status=#{@status.inspect} pid=#{@pid.inspect}>"
+				end
+				
+				alias to_s inspect
+				
+				# Invoke {#terminate!} and then {#wait} for the child process to exit.
+				def close
+					self.terminate!
+					self.wait
+				ensure
+					super
+				end
+				
+				# Send `SIGINT` to the child process.
+				def interrupt!
+					unless @status
+						::Process.kill(:INT, @pid)
+					end
+				end
+				
+				# Send `SIGTERM` to the child process.
+				def terminate!
+					unless @status
+						::Process.kill(:TERM, @pid)
+					end
+				end
+				
+				# Send `SIGHUP` to the child process.
+				def restart!
+					unless @status
+						::Process.kill(:HUP, @pid)
+					end
+				end
+				
+				# Wait for the child process to exit.
+				# @asynchronous This method may block.
+				#
+				# @returns [::Process::Status] The process exit status.
+				def wait
+					if @pid && @status.nil?
+						Console.debug(self, "Waiting for process to exit...", pid: @pid)
+						
+						_, @status = ::Process.wait2(@pid, ::Process::WNOHANG)
+
+						while @status.nil?
+							sleep(0.1)
+							
+							_, @status = ::Process.wait2(@pid, ::Process::WNOHANG)
+							
+							if @status.nil?
+								Console.warn(self) {"Process #{@pid} is blocking, has it exited?"}
+							end
+						end
+					end
+					
+					Console.debug(self, "Process exited.", pid: @pid, status: @status)
+					
+					return @status
+				end
+			end
+			
+			
 			# Start a named child process and execute the provided block in it.
 			# @parameter name [String] The name (title) of the child process.
 			# @parameter block [Proc] The block to execute in the child process.
 			def start(name, &block)
-				Process.fork(name: name, &block)
+				Child.fork(name: name, &block)
 			end
 		end
 	end

lib/async/container/generic.rb

@@ -3,8 +3,6 @@
 # Released under the MIT License.
 # Copyright, 2019-2024, by Samuel Williams.
 
-require "async"
-
 require "etc"
 
 require_relative "group"
@@ -197,8 +195,12 @@ def run(count: Container.processor_count, **options, &block)
 
 			# @deprecated Please use {spawn} or {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::Reactor.run(instance, &block)
+					Async(instance, &block)
 				end
 			end