Improved RPC mechanism.

Author: ioquatix

async-container-supervisor.gemspec

@@ -24,8 +24,8 @@ Gem::Specification.new do |spec|
 
 	spec.required_ruby_version = ">= 3.1"
 
-	spec.add_dependency "async-container"
+	spec.add_dependency "async-container", "~> 0.22"
 	spec.add_dependency "async-service"
 	spec.add_dependency "io-stream"
-	spec.add_dependency "memory-leak", "~> 0.2"
+	spec.add_dependency "memory-leak", "~> 0.3"
 end

example/simple/simple.rb

@@ -6,20 +6,23 @@ class SleepService < Async::Service::Generic
 	def setup(container)
 		super
 
-		container.run(count: 1, restart: true, health_check_timeout: 2) do |instance|
+		container.run(name: self.class.name, count: 4, restart: true, health_check_timeout: 2) do |instance|
 			Async do
 				client = Async::Container::Supervisor::Client.new(instance, @evaluator.supervisor_endpoint)
 				client.run
 
+				start_time = Time.now
+				
 				instance.ready!
 
 				chunks = []
 				while true
-					Console.info(self, "Allocating memory...")
-					# Allocate 10MB of memory every second:
-					chunks << " " * 1024 * 1024
-					sleep 0.1
+					chunks << " " * 1024 * 1024 * rand(10) 
+					sleep 1
 					instance.ready!
+					
+					uptime = Time.now - start_time
+					instance.name = "Sleeping for #{uptime.to_i} seconds..."
 				end
 			ensure
 				Console.info(self, "Exiting...")
@@ -38,6 +41,6 @@ def setup(container)
 	include Async::Container::Supervisor::Environment
 
 	monitors do
-		[Async::Container::Supervisor::Monitor::MemoryMonitor.new(interval: 1, limit: 1024 * 1024 * 100)]
+		[Async::Container::Supervisor::MemoryMonitor.new(interval: 1, limit: 1024 * 1024 * 100)]
 	end
 end

gems.rb

@@ -7,8 +7,6 @@
 
 gemspec
 
-gem "async-container", path: "../async-container"
-
 group :maintenance, optional: true do
 	gem "bake-gem"
 	gem "bake-modernize"

lib/async/container/supervisor.rb

@@ -7,7 +7,8 @@
 
 require_relative "supervisor/server"
 require_relative "supervisor/client"
-require_relative "supervisor/monitor"
+
+require_relative "supervisor/memory_monitor"
 
 require_relative "supervisor/environment"
 require_relative "supervisor/service"

lib/async/container/supervisor/client.rb

@@ -4,7 +4,7 @@
 # Copyright, 2025, by Samuel Williams.
 
 require "io/stream"
-require_relative "wrapper"
+require_relative "connection"
 
 module Async
 	module Container
@@ -19,47 +19,95 @@ def initialize(instance, endpoint = Supervisor.endpoint)
 					@endpoint = endpoint
 				end
 
+				def dispatch(call)
+					method_name = "do_#{call.message[:do]}"
+					self.public_send(method_name, call)
+				end
+				
 				def connect
-					unless @wrapper
+					unless @connection
 						peer = @endpoint.connect
 						stream = IO::Stream(peer)
-						@wrapper = Wrapper.new(stream)
+						@connection = Connection.new(stream, 0, instance: @instance)
 
-						@wrapper.write(action: "register", instance: @instance)
+						# Register the instance with the server:
+						Async do
+							@connection.call(do: :register, state: @instance)
+						end
 					end
 
-					return @wrapper unless block_given?
+					return @connection unless block_given?
 
 					begin
-						yield @wrapper
+						yield @connection
 					ensure
-						@wrapper.close
+						@connection.close
 					end
 				end
 
 				def close
-					if wrapper = @wrapper
-						@wrapper = nil
-						wrapper.close
+					if connection = @connection
+						@connection = nil
+						connection.close
+					end
+				end
+				
+				private def dump(call)
+					if path = call[:path]
+						File.open(path, "w") do |file|
+							yield file
+						end
+						
+						call.finish(path: path)
+					else
+						buffer = StringIO.new
+						yield buffer
+						
+						call.finish(data: buffer.string)
+					end
+				end
+				
+				def do_scheduler_dump(call)
+					dump(call) do |file|
+						Fiber.scheduler.print_hierarchy(file)
 					end
 				end
 
-				def do_memory_dump(wrapper, message)
-					Console.info(self, "Memory dump:", message)
-					path = message[:path]
+				def do_memory_dump(call)
+					require "objspace"
 
-					File.open(path, "w") do |file|
+					dump(call) do |file|
 						ObjectSpace.dump_all(output: file)
 					end
 				end
 
+				def do_thread_dump(call)
+					dump(call) do |file|
+						Thread.list.each do |thread|
+							file.puts(thread.inspect)
+							file.puts(thread.backtrace)
+						end
+					end
+				end
+				
+				def do_garbage_profile_start(call)
+					GC::Profiler.enable
+					call.finish(started: true)
+				end
+				
+				def do_garbage_profile_stop(call)
+					GC::Profiler.disable
+					
+					dump(connection, message) do |file|
+						file.puts GC::Profiler.result
+					end
+				end
+				
 				def run
 					Async do |task|
 						loop do
-							Console.info(self, "Connecting to supervisor...")
-							connect do |wrapper|
-								Console.info(self, "Connected to supervisor.")
-								wrapper.run(self)
+							connect do |connection|
+								connection.run(self)
 							end
 						rescue => error
 							Console.error(self, "Unexpected error while running client!", exception: error)

lib/async/container/supervisor/connection.rb

@@ -0,0 +1,165 @@
+# frozen_string_literal: true
+
+# Released under the MIT License.
+# Copyright, 2025, by Samuel Williams.
+
+require "json"
+
+module Async
+	module Container
+		module Supervisor
+			class Connection
+				class Call
+					def initialize(connection, id, message)
+						@connection = connection
+						@id = id
+						@message = message
+						
+						@queue = ::Thread::Queue.new
+					end
+					
+					# @attribute [Connection] The connection that initiated the call.
+					attr :connection
+					
+					# @attribute [Hash] The message that initiated the call.
+					attr :message
+					
+					def [] key
+						@message[key]
+					end
+					
+					def push(**response)
+						@queue.push(response)
+					end
+					
+					def pop(...)
+						@queue.pop(...)
+					end
+					
+					def each(&block)
+						while response = self.pop
+							yield response
+						end
+					end
+					
+					def finish(**response)
+						self.push(id: @id, finished: true, **response)
+						@queue.close
+					end
+					
+					def closed?
+						@queue.closed?
+					end
+					
+					def self.dispatch(connection, target, id, message)
+						Async do
+							call = self.new(connection, id, message)
+							connection.calls[id] = call
+							
+							target.dispatch(call)
+							
+							while response = call.pop
+								connection.write(id: id, **response)
+							end
+						ensure
+							# If the queue is closed, we don't need to send a finished message.
+							unless call.closed?
+								connection.write(id: id, finished: true)
+							end
+							
+							connection.calls.delete(id)
+						end
+					end
+					
+					def self.call(connection, **message, &block)
+						id = connection.next_id
+						call = self.new(connection, id, message)
+						
+						connection.calls[id] = call
+						connection.write(id: id, **message)
+						
+						if block_given?
+							call.each(&block)
+						else
+							return call.pop
+						end
+					end
+				end
+				
+				def initialize(stream, id, **state)
+					@stream = stream
+					@state = state
+					
+					@calls = {}
+					
+					@id = id
+				end
+				
+				# @attribute [Hash(Integer, Call)] Calls in progress.
+				attr :calls
+				
+				# @attribute [Hash(Symbol, Object)] State associated with this connection, for example the process ID, etc.
+				attr_accessor :state
+				
+				def next_id
+					@id += 2
+				end
+				
+				def write(**message)
+					@stream.write(JSON.dump(message) << "\n")
+					@stream.flush
+				end
+				
+				def call(timeout: nil, **message)
+					id = next_id
+					calls[id] = ::Thread::Queue.new
+					
+					write(id: id, **message)
+					
+					return calls[id].pop(timeout: timeout)
+				ensure
+					calls.delete(id)
+				end
+				
+				def read
+					if line = @stream&.gets
+						JSON.parse(line, symbolize_names: true)
+					end
+				end
+				
+				def each
+					while message = self.read
+						yield message
+					end
+				end
+				
+				def call(...)
+					Call.call(self, ...)
+				end
+				
+				def run(target)
+					self.each do |message|
+						if id = message[:id]
+							if call = @calls[id]
+								# Response to a call:
+								call.push(**message)
+							else
+								# Incoming call:
+								Call.dispatch(self, target, id, message)
+							end
+						else
+							Console.error(self, "Unknown message:", message)
+						end
+					end
+				end
+				
+				def close
+					if stream = @stream
+						@stream = nil
+						stream.close
+					end
+				end
+			end
+		end
+	end
+end