Fix handling of implicitly bound objects + add benchmark.

Author: samuel-williams-shopify

benchmark/message_passing.rb

@@ -0,0 +1,72 @@
+# frozen_string_literal: true
+
+# Released under the MIT License.
+# Copyright, 2025, by Samuel Williams.
+
+require "sus/fixtures/benchmark"
+require "async/bus/a_server"
+
+describe "Message Passing Performance" do
+	include Sus::Fixtures::Benchmark
+	include Async::Bus::AServer
+	
+	let(:array) {Array.new}
+	
+	before do
+		start_server do |connection|
+			connection.bind(:array, array)
+			connection.bind(:counter, proc{|value| value + 1})
+		end
+	end
+	
+	measure "simple method call" do |repeats|
+		client.connect do |connection|
+			repeats.times do
+				connection[:array].size
+			end
+		end
+	end
+	
+	measure "method call with arguments" do |repeats|
+		client.connect do |connection|
+			repeats.times do |i|
+				connection[:array] << i
+			end
+		end
+	end
+	
+	measure "method call with return value" do |repeats|
+		client.connect do |connection|
+			repeats.times do |i|
+				result = connection[:array].size
+			end
+		end
+	end
+	
+	measure "proc invocation" do |repeats|
+		client.connect do |connection|
+			repeats.times do
+				connection[:counter].call(1)
+			end
+		end
+	end
+	
+	measure "multiple sequential calls" do |repeats|
+		client.connect do |connection|
+			repeats.times do |i|
+				connection[:array] << i
+				connection[:array].size
+				connection[:array].last
+			end
+		end
+	end
+	
+	measure "property access" do |repeats|
+		client.connect do |connection|
+			repeats.times do
+				connection[:array].length
+			end
+		end
+	end
+end
+

fixtures/async/bus/a_server.rb

@@ -13,7 +13,7 @@
 module Async
 	module Bus
 		AServer = Sus::Shared("a server") do
-			include Sus::Fixtures::Async::ReactorContext
+			include Sus::Fixtures::Async::SchedulerContext
 
 			let(:ipc_path) {File.join(@root, "bus.ipc")}
 			let(:endpoint) {Async::Bus::Protocol.local_endpoint(ipc_path)}
@@ -25,11 +25,11 @@ def around(&block)
 				end
 			end
 
-			def before
+			before do
 				@bound_endpoint = endpoint.bound
 			end
 
-			def after(error = nil)
+			after do
 				@bound_endpoint&.close
 				@server_task&.stop
 			end

gems.rb

@@ -12,6 +12,8 @@
 	gem "bake-gem"
 	gem "bake-releases"
 
+	gem "agent-context"
+	
 	gem "utopia-project"
 end
 
@@ -25,6 +27,7 @@
 	gem "rubocop-socketry"
 
 	gem "sus-fixtures-async"
+	gem "sus-fixtures-benchmark"
 
 	gem "bake-test"
 	gem "bake-test-external"

lib/async/bus/protocol/connection.rb

@@ -84,11 +84,26 @@ def next_id
 
 				attr :transactions
 
+				Explicit = Struct.new(:object) do
+					def temporary?
+						false
+					end
+				end
+				
+				Implicit = Struct.new(:object) do
+					def temporary?
+						true
+					end
+				end
+				
 				# Bind a local object to a name, such that it could be accessed remotely.
 				#
 				# @returns [Proxy] A proxy instance for the bound object.
 				def bind(name, object)
-					@objects[name] = object
+					# Bind the object into the local object store (explicitly bound, not temporary):
+					@objects[name] = Explicit.new(object)
+					
+					# Return the proxy instance for the bound object:
 					return self[name]
 				end
 
@@ -98,7 +113,11 @@ def bind(name, object)
 				def proxy(object)
 					name = "<#{object.class}@#{next_id.to_s(16)}>".freeze
 
-					return bind(name, object)
+					# Bind the object into the local object store (temporary):
+					@objects[name] = Implicit.new(object)
+					
+					# This constructs the Proxy instance:
+					return self[name]
 				end
 
 				# Generate a proxy name for an object and bind it, returning just the name.
@@ -107,12 +126,16 @@ def proxy(object)
 				# @returns [String] The name of the bound object.
 				def proxy_name(object)
 					name = "<#{object.class}@#{next_id.to_s(16)}>".freeze
-					bind(name, object)
+					
+					# Bind the object into the local object store (temporary):
+					@objects[name] = Implicit.new(object)
+					
+					# Return the name:
 					return name
 				end
 
 				def object(name)
-					@objects[name]
+					@objects[name]&.object
 				end
 
 				private def finalize(name)
@@ -122,7 +145,7 @@ def object(name)
 				end
 
 				def []=(name, object)
-					@objects[name] = object
+					@objects[name] = Explicit.new(object)
 				end
 
 				def [](name)
@@ -164,11 +187,9 @@ def run(parent: Task.current)
 
 					@unpacker.each do |message|
 						case message
-						when Release
-							@objects.delete(message.name)
 						when Invoke
 							# If the object is not found, send an error response and skip the transaction:
-							if object = @objects[message.name]
+							if object = @objects[message.name]&.object
 								transaction = self.transaction!(message.id)
 
 								parent.async(annotation: "Invoke #{message.name}") do
@@ -188,6 +209,12 @@ def run(parent: Task.current)
 							else
 								# Stale message - transaction already closed (e.g. timeout) or never existed (ignore silently).
 							end
+						when Release
+							name = message.name
+							if @objects[name]&.temporary?
+								# Only delete temporary objects, not explicitly bound ones:
+								@objects.delete(name)
+							end
 						else
 							Console.error(self, "Unexpected message:", message)
 						end

lib/async/bus/protocol/wrapper.rb

@@ -25,27 +25,27 @@ def initialize(bus, reference_types: [Controller])
 					# The order here matters.
 
 					self.register_type(0x00, Invoke, recursive: true,
-						packer: ->(invoke, packer) {invoke.pack(packer)},
-						unpacker: ->(unpacker) {Invoke.unpack(unpacker)},
+						packer: ->(invoke, packer){invoke.pack(packer)},
+						unpacker: ->(unpacker){Invoke.unpack(unpacker)},
 					)
 
 					[Return, Yield, Error, Next, Throw, Close].each_with_index do |klass, index|
 						self.register_type(0x01 + index, klass, recursive: true,
-							packer: ->(value, packer) {value.pack(packer)},
-							unpacker: ->(unpacker) {klass.unpack(unpacker)},
+							packer: ->(value, packer){value.pack(packer)},
+							unpacker: ->(unpacker){klass.unpack(unpacker)},
 						)
 					end
 
 					# Reverse serialize proxies back into proxies:
 					# When a Proxy is received, create a proxy pointing back
 					self.register_type(0x10, Proxy,
-						packer: ->(proxy) {proxy.__name__},
+						packer: ->(proxy){proxy.__name__},
 						unpacker: @bus.method(:[]),
 					)
 
 					self.register_type(0x11, Release, recursive: true,
-						packer: ->(release, packer) {release.pack(packer)},
-						unpacker: ->(unpacker) {Release.unpack(unpacker)},
+						packer: ->(release, packer){release.pack(packer)},
+						unpacker: ->(unpacker){Release.unpack(unpacker)},
 					)
 
 					self.register_type(0x20, Symbol)
@@ -56,8 +56,8 @@ def initialize(bus, reference_types: [Controller])
 					)
 
 					self.register_type(0x22, Class,
-						packer: ->(klass) {klass.name},
-						unpacker: ->(name) {Object.const_get(name)},
+						packer: ->(klass){klass.name},
+						unpacker: ->(name){Object.const_get(name)},
 					)
 
 					# Serialize objects into proxies:

releases.md

@@ -1,3 +1,5 @@
 # Releases
 
 ## Unreleased
+
+  - Fix handling of temporary objects.

test/async/bus/protocol/transaction.rb

@@ -113,7 +113,7 @@ def service.yielding_method
 				def service.yielding_method
 					yield 1
 				end
-					
+				
 				connection.bind(:service, service)
 			end
 
@@ -136,7 +136,7 @@ def service.yielding_method
 				def service.error_method
 					raise ArgumentError, "Invalid argument"
 				end
-					
+				
 				connection.bind(:service, service)
 			end
 

test/async/bus/server.rb

@@ -125,7 +125,7 @@ def before
 			@server_task = Async do
 				server.accept do |connection|
 					connection.bind(:hash, hash)
-					connection.bind(:sum_key, proc {|key| hash[key].sum})
+					connection.bind(:sum_key, proc{|key| hash[key].sum})
 				end
 			end
 		end