Simplify worker ordinal API

Assisted-By: devx/c78f867c-4c73-40b2-a763-4a9332e15ef9

Author: samuel-williams-shopify

lib/async/container/context.rb

@@ -7,7 +7,6 @@
 
 require_relative "generic"
 require_relative "channel"
-require_relative "context"
 require_relative "notify/pipe"
 
 module Async
@@ -23,12 +22,10 @@ def self.multiprocess?
 			class Child < Channel
 				# Represents a running child process from the point of view of the child process.
 				class Instance < Notify::Pipe
-					include Context
-					
 					# 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_num: nil)
-						instance = self.new(process.out, num: instance_num)
+					def self.for(process, ordinal: nil)
+						instance = self.new(process.out, ordinal: ordinal)
 
 						# The child process won't be reading from the channel:
 						process.close_read
@@ -41,20 +38,18 @@ def self.for(process, instance_num: nil)
 					# Initialize the child process instance.
 					#
 					# @parameter io [IO] The IO object to use for communication.
-					def initialize(io, num: nil)
+					def initialize(io, ordinal: nil)
 						super(io)
 
 						@name = nil
-						@num = num
+						@ordinal = ordinal
 					end
 
 					# @returns [Integer | Nil] The container-scoped ordinal of this worker.
-					attr :num
+					attr :ordinal
 
-					# @returns [Symbol] The kind of worker this instance represents.
-					def kind
-						:process
-					end
+					# @returns [Object | Nil] The worker this one is nested inside.
+					attr_accessor :parent
 
 					# Generate a hash representation of the process.
 					#
@@ -63,7 +58,7 @@ def as_json(...)
 						{
 							process_id: ::Process.pid,
 							name: @name,
-							num: @num,
+							ordinal: @ordinal,
 						}
 					end
 
@@ -111,9 +106,9 @@ def exec(*arguments, ready: true, **options)
 				# Fork a child process appropriate for a container.
 				#
 				# @returns [Process]
-				def self.fork(instance_num: nil, **options)
+				def self.fork(ordinal: nil, **options)
 					# $stderr.puts fork: caller
-					self.new(instance_num: instance_num, **options) do |process|
+					self.new(ordinal: ordinal, **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)}
@@ -122,7 +117,7 @@ def self.fork(instance_num: nil, **options)
 
 							# This could be a configuration option:
 							::Thread.handle_interrupt(SignalException => :immediate) do
-								yield Instance.for(process, instance_num: instance_num)
+								yield Instance.for(process, ordinal: ordinal)
 							rescue Interrupt
 								# Graceful exit.
 							rescue Exception => error
@@ -151,11 +146,11 @@ def self.spawn(*arguments, name: nil, **options)
 
 				# Initialize the process.
 				# @parameter name [String] The name to use for the child process.
-				def initialize(name: nil, instance_num: nil, **options)
+				def initialize(name: nil, ordinal: nil, **options)
 					super(**options)
 
 					@name = name
-					@instance_num = instance_num
+					@ordinal = ordinal
 					@status = nil
 					@pid = nil
 
@@ -200,7 +195,7 @@ def name= value
 				attr :pid
 
 				# @attribute [Integer | Nil] The container-scoped ordinal of the worker this child represents.
-				attr :instance_num
+				attr :ordinal
 
 				# A human readable representation of the process.
 				# @returns [String]
@@ -282,8 +277,8 @@ def wait(timeout = 0.1)
 			# 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, instance_num: nil, &block)
-				Child.fork(name: name, instance_num: instance_num, &block)
+			def start(name, ordinal: nil, &block)
+				Child.fork(name: name, ordinal: ordinal, &block)
 			end
 		end
 	end

lib/async/container/forked.rb

@@ -54,11 +54,11 @@ def initialize(policy: Policy::DEFAULT, **options)
 				@policy = policy
 				@statistics = @policy.make_statistics
 				@keyed = {}
-				# Container-scoped allocation of worker ordinals (`instance_num`): a monotonic
-				# counter plus a free set, so a num released by a permanently exited worker is
+				# Container-scoped allocation of worker ordinals: a monotonic
+				# counter plus a free set, so an ordinal released by a permanently exited worker is
 				# recycled, keeping the range compact (e.g. for multiprocess metric files).
-				@next_instance_num = 0
-				@free_instance_nums = Set.new
+				@next_ordinal = 0
+				@free_ordinals = Set.new
 			end
 
 			# @attribute [Group] The group of running children instances.
@@ -218,27 +218,25 @@ def stop(timeout = true)
 			# @parameter key [Symbol] A key used for reloading child instances.
 			# @parameter health_check_timeout [Numeric | Nil] The maximum time a child instance can run without updating its state, before it is terminated as unhealthy.
 			# @parameter startup_timeout [Numeric | Nil] The maximum time a child instance can run without becoming ready, before it is terminated as unhealthy.
-			def spawn(name: nil, restart: false, key: nil, instance_num: nil, health_check_timeout: nil, startup_timeout: nil, &block)
+			def spawn(name: nil, restart: false, key: nil, health_check_timeout: nil, startup_timeout: nil, &block)
 				name ||= UNNAMED
 
 				if mark?(key)
 					Console.debug(self, "Reusing existing child.", child: {key: key, name: name})
 					return false
 				end
 
-				# Allocate before the fiber so the closure captures the num and it stays unchanged
-				# across a restart (which re-enters `start` in the same fiber). Only release a num
-				# we allocated ourselves, and only when the worker permanently exits.
-				owned = instance_num.nil?
-				instance_num ||= acquire_instance_num
+				# Allocate before the fiber so the closure captures the ordinal and it stays
+				# unchanged across a restart (which re-enters `start` in the same fiber).
+				ordinal = acquire_ordinal
 
 				@statistics.spawn!
 
 				fiber do
 					until @stopping
 						Console.debug(self, "Starting child...", child: {key: key, name: name, restart: restart, health_check_timeout: health_check_timeout}, statistics: @statistics)
 
-						child = self.start(name, instance_num: instance_num, &block)
+						child = self.start(name, ordinal: ordinal, &block)
 						state = insert(key, child)
 
 						# Notify policy of spawn
@@ -312,31 +310,31 @@ def spawn(name: nil, restart: false, key: nil, instance_num: nil, health_check_t
 						end
 					end
 				ensure
-					release_instance_num(instance_num) if owned
+					release_ordinal(ordinal)
 				end.resume
 
 				return true
 			end
 
-			# Allocate a container-scoped worker ordinal, recycling the lowest released num.
+			# Allocate a container-scoped worker ordinal, recycling the lowest released ordinal.
 			# Allocation runs on the single cooperative reactor thread (acquire in the run loop,
 			# release in the fiber's ensure), so no synchronisation is required.
-			protected def acquire_instance_num
-				unless @free_instance_nums.empty?
-					num = @free_instance_nums.min
-					@free_instance_nums.delete(num)
-					return num
+			protected def acquire_ordinal
+				unless @free_ordinals.empty?
+					ordinal = @free_ordinals.min
+					@free_ordinals.delete(ordinal)
+					return ordinal
 				end
 
-				num = @next_instance_num
-				@next_instance_num += 1
-				num
+				ordinal = @next_ordinal
+				@next_ordinal += 1
+				ordinal
 			end
 
 			# Return a worker ordinal to the free set once its worker has permanently exited. Using a
-			# Set makes release idempotent, so a double release can't hand the same num to two workers.
-			protected def release_instance_num(instance_num)
-				@free_instance_nums.add(instance_num)
+			# Set makes release idempotent, so a double release can't hand the same ordinal to two workers.
+			protected def release_ordinal(ordinal)
+				@free_ordinals.add(ordinal)
 			end
 
 			# Run multiple instances of the same block in the container.

lib/async/container/generic.rb

@@ -27,7 +27,7 @@ def run(count: nil, forks: nil, threads: nil, health_check_timeout: nil, **optio
 						container = Threaded.new
 
 						# Link each inner thread worker to this fork, so the thread instance can reach
-						# the durable process num via `instance.parent` / `instance.context`.
+						# the durable process ordinal via `instance.parent`.
 						container.run(count: threads, health_check_timeout: health_check_timeout, **options) do |worker|
 							worker.parent = instance
 							block.call(worker)

lib/async/container/hybrid.rb

@@ -5,7 +5,6 @@
 
 require_relative "generic"
 require_relative "channel"
-require_relative "context"
 require_relative "notify/pipe"
 
 module Async
@@ -42,33 +41,29 @@ def error
 
 				# Represents a running child thread from the point of view of the child thread.
 				class Instance < Notify::Pipe
-					include Context
-					
 					# Wrap an instance around the {Thread} instance from within the threaded child.
 					# @parameter thread [Thread] The thread intance to wrap.
-					def self.for(thread, instance_num: nil)
-						instance = self.new(thread.out, num: instance_num)
+					def self.for(thread, ordinal: nil)
+						instance = self.new(thread.out, ordinal: ordinal)
 
 						return instance
 					end
 
 					# Initialize the child thread instance.
 					#
 					# @parameter io [IO] The IO object to use for communication with the parent.
-					def initialize(io, num: nil)
+					def initialize(io, ordinal: nil)
 						@thread = ::Thread.current
-						@num = num
+						@ordinal = ordinal
 
 						super(io)
 					end
 
 					# @returns [Integer | Nil] The container-scoped ordinal of this worker.
-					attr :num
+					attr :ordinal
 
-					# @returns [Symbol] The kind of worker this instance represents.
-					def kind
-						:thread
-					end
+					# @returns [Object | Nil] The worker this one is nested inside.
+					attr_accessor :parent
 
 					# Generate a hash representation of the thread.
 					#
@@ -78,7 +73,7 @@ def as_json(...)
 							process_id: ::Process.pid,
 							thread_id: @thread.object_id,
 							name: @thread.name,
-							num: @num,
+							ordinal: @ordinal,
 						}
 					end
 
@@ -124,12 +119,12 @@ def exec(*arguments, ready: true, **options)
 				# Start a new child thread and execute the provided block in it.
 				#
 				# @parameter options [Hash] Additional options to to the new child instance.
-				def self.fork(instance_num: nil, **options)
-					self.new(instance_num: instance_num, **options) do |thread|
+				def self.fork(ordinal: nil, **options)
+					self.new(ordinal: ordinal, **options) do |thread|
 						::Thread.new do
 							# This could be a configuration option (see forked implementation too):
 							::Thread.handle_interrupt(SignalException => :immediate) do
-								yield Instance.for(thread, instance_num: instance_num)
+								yield Instance.for(thread, ordinal: ordinal)
 							end
 						end
 					end
@@ -138,10 +133,10 @@ def self.fork(instance_num: nil, **options)
 				# Initialize the thread.
 				#
 				# @parameter name [String] The name to use for the child thread.
-				def initialize(name: nil, instance_num: nil, **options)
+				def initialize(name: nil, ordinal: nil, **options)
 					super(**options)
 
-					@instance_num = instance_num
+					@ordinal = ordinal
 					@status = nil
 
 					@thread = yield(self)
@@ -165,7 +160,7 @@ def initialize(name: nil, instance_num: nil, **options)
 				end
 
 				# @attribute [Integer | Nil] The container-scoped ordinal of the worker this child represents.
-				attr :instance_num
+				attr :ordinal
 
 				# Convert the child process to a hash, suitable for serialization.
 				#
@@ -303,8 +298,8 @@ def finished(error = nil)
 			# Start a named child thread 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, instance_num: nil, &block)
-				Child.fork(name: name, instance_num: instance_num, &block)
+			def start(name, ordinal: nil, &block)
+				Child.fork(name: name, ordinal: ordinal, &block)
 			end
 		end
 	end