From 336131ec75247accb54ec82f12d4f692e5c327d7 Mon Sep 17 00:00:00 2001 From: Samuel Williams Date: Fri, 21 Aug 2026 09:56:19 +1200 Subject: [PATCH 1/4] Use utilization segment allocator Signed-off-by: Samuel Williams Assisted-By: devx/fec6efe0-6bff-4a9e-a7a1-de183807a581 --- async-service-supervisor.gemspec | 2 +- .../service/supervisor/utilization_monitor.rb | 179 +----------------- releases.md | 4 + test/async/service/utilization_monitor.rb | 31 --- 4 files changed, 9 insertions(+), 207 deletions(-) diff --git a/async-service-supervisor.gemspec b/async-service-supervisor.gemspec index 5391359..7282327 100644 --- a/async-service-supervisor.gemspec +++ b/async-service-supervisor.gemspec @@ -27,7 +27,7 @@ Gem::Specification.new do |spec| spec.add_dependency "async", "~> 2.38" spec.add_dependency "async-bus" spec.add_dependency "async-service", "~> 0.15" - spec.add_dependency "async-utilization", "~> 0.3" + spec.add_dependency "async-utilization", "~> 0.5" spec.add_dependency "io-endpoint" spec.add_dependency "memory", "~> 0.7" spec.add_dependency "memory-leak", "~> 0.10" diff --git a/lib/async/service/supervisor/utilization_monitor.rb b/lib/async/service/supervisor/utilization_monitor.rb index 596ed7d..d860fc9 100644 --- a/lib/async/service/supervisor/utilization_monitor.rb +++ b/lib/async/service/supervisor/utilization_monitor.rb @@ -3,8 +3,6 @@ # Released under the MIT License. # Copyright, 2026, by Samuel Williams. -require "set" - require_relative "monitor" require "async/utilization" @@ -16,178 +14,9 @@ module Supervisor # Uses shared memory to efficiently collect utilization metrics from workers # and aggregates them by service name for monitoring and reporting. class UtilizationMonitor < Monitor - # Allocates and manages shared memory segments for worker utilization data. - # - # Manages a shared memory file that workers can write utilization metrics to. - # Allocates segments to workers and maintains a free list for reuse. - # Each process (supervisor and workers) maps the shared memory file independently. - class SegmentAllocator - # Initialize a new shared memory manager. - # - # Creates and maps the shared memory file. Workers will map the same file - # independently using the provided path. - # - # @parameter path [String] Path to the shared memory file. - # @parameter size [Integer] Total size of the shared memory buffer. - # @parameter segment_size [Integer] Size of each allocation segment (default: 512 bytes). - # @parameter growth_factor [Integer, Float] Factor to grow by when resizing (default: 2, doubles the size). - # Can be less than 2 or a floating point value; the result will be page-aligned to an integer. - def initialize(path, size: IO::Buffer::PAGE_SIZE * 8, segment_size: 512, growth_factor: 2) - @path = path - @size = size - @segment_size = segment_size - @growth_factor = growth_factor - - File.unlink(path) rescue nil - @file = File.open(path, "w+b") - @file.truncate(size) - # Supervisor maps the file for reading worker data - @buffer = IO::Buffer.map(@file, size) - - # Track allocated segments: worker_id => {offset: Integer, schema: Array} - @allocations = {} - - # Free list of segment offsets - @free_list = [] - - # Initialize free list with all segments - (0...(@size / @segment_size)).each do |segment_index| - @free_list << (segment_index * @segment_size) - end - end - - # Allocate a segment for a worker. - # - # Automatically resizes the shared memory file if no segments are available. - # - # @parameter worker_id [Integer] The ID of the worker. - # @parameter schema [Array] Array of [key, type, offset] tuples describing the data layout. - # @returns [Integer] The offset into the shared memory buffer, or nil if allocation fails. - def allocate(worker_id, schema) - # Try to resize if we're out of segments - if @free_list.empty? - unless resize(@size * @growth_factor) - return nil - end - end - - offset = @free_list.shift - @allocations[worker_id] = {offset: offset, schema: schema} - - return offset - end - - # Free a segment allocated to a worker. - # - # @parameter worker_id [Integer] The ID of the worker. - def free(worker_id) - if allocation = @allocations.delete(worker_id) - @free_list << allocation[:offset] - end - end - - # Get the allocation information for a worker. - # - # @parameter worker_id [Integer] The ID of the worker. - # @returns [Hash] Allocation info with :offset and :schema, or nil if not allocated. - def allocation(worker_id) - @allocations[worker_id] - end - - # Get the current size of the shared memory file. - # - # @returns [Integer] The current size of the shared memory file. - def size - @size - end - - # Update the schema for an existing allocation. - # - # @parameter worker_id [Integer] The ID of the worker. - # @parameter schema [Array] Array of [key, type, offset] tuples describing the data layout. - def update_schema(worker_id, schema) - if allocation = @allocations[worker_id] - allocation[:schema] = schema - end - end - - # Read utilization data from a worker's allocated segment. - # - # @parameter worker_id [Integer] The ID of the worker. - # @returns [Hash] Hash mapping keys to their values, or nil if not allocated. - def read(worker_id) - allocation = @allocations[worker_id] - return nil unless allocation - - offset = allocation[:offset] - schema = allocation[:schema] - - result = {} - schema.each do |key, type, field_offset| - absolute_offset = offset + field_offset - - # Use IO::Buffer type symbols directly (i32, u32, i64, u64, f32, f64) - # IO::Buffer accepts both lowercase and uppercase versions - begin - result[key] = @buffer.get_value(type, absolute_offset) - rescue => error - Console.warn(self, "Failed to read value", type: type, key: key, offset: absolute_offset, exception: error) - end - end - - return result - end - - # Resize the shared memory file. - # - # Extends the file to the new size, remaps the buffer, and adds new segments - # to the free list. The new size must be larger than the current size and should - # be page-aligned for optimal performance. - # - # @parameter new_size [Integer] The new size for the shared memory file. - # @returns [Boolean] True if resize was successful, false otherwise. - def resize(new_size) - old_size = @size - return false if new_size <= old_size - - # Ensure new size is page-aligned (rounds up to nearest page boundary) - page_size = IO::Buffer::PAGE_SIZE - new_size = (((new_size + page_size - 1) / page_size) * page_size).to_i - - begin - # Extend the file: - @file.truncate(new_size) - - # Remap the buffer to the new size: - @buffer&.free - @buffer = IO::Buffer.map(@file, new_size) - - # Calculate new segments to add to free list: - old_segment_count = old_size / @segment_size - new_segment_count = new_size / @segment_size - - # Add new segments to free list: - (old_segment_count...new_segment_count).each do |segment_index| - @free_list << (segment_index * @segment_size) - end - - @size = new_size - - Console.info(self, "Resized shared memory", old_size: old_size, new_size: new_size, segments_added: new_segment_count - old_segment_count) - - return true - rescue => error - Console.error(self, "Failed to resize shared memory", old_size: old_size, new_size: new_size, exception: error) - return false - end - end - - # Close the shared memory file. - def close - @file&.close - @buffer = nil - end - end + # @deprecated Use {Async::Utilization::SegmentAllocator} instead. + SegmentAllocator = Async::Utilization::SegmentAllocator + # Initialize a new utilization monitor. # # @parameter path [String] Path to the shared memory file. @@ -199,7 +28,7 @@ def initialize(path: "utilization.shm", interval: 10, size: IO::Buffer::PAGE_SIZ @path = path @segment_size = segment_size - @allocator = SegmentAllocator.new(path, size: size, segment_size: segment_size) + @allocator = Async::Utilization::SegmentAllocator.new(path, size: size, segment_size: segment_size, replace: true) # Track workers: worker_id => supervisor_controller @workers = {} diff --git a/releases.md b/releases.md index 73e40ec..f472cdd 100644 --- a/releases.md +++ b/releases.md @@ -1,5 +1,9 @@ # Releases +## Unreleased + + - Use `Async::Utilization::SegmentAllocator` to manage utilization shared memory. + ## v0.20.0 - Add per-worker snapshots to `Async::Service::Supervisor::UtilizationMonitor`. diff --git a/test/async/service/utilization_monitor.rb b/test/async/service/utilization_monitor.rb index 7f67c6d..0358bce 100644 --- a/test/async/service/utilization_monitor.rb +++ b/test/async/service/utilization_monitor.rb @@ -299,37 +299,6 @@ expect(monitor.status[:data]).to be == {} end - it "does not resize existing file when recreating the utilization monitor" do - # When the supervisor restarts, it recreates the SegmentAllocator. Without unlink, - # File.open(path, "w+b") truncates the existing file. With unlink, we remove the file - # first so the new allocator gets a fresh file; any process with the old file mapped - # keeps a valid mapping to the unlinked inode. - allocator = Async::Service::Supervisor::UtilizationMonitor::SegmentAllocator.new( - shm_path, size: file_size, segment_size: segment_size - ) - - # Resize to make the file larger than initial: - larger_size = file_size * 2 - allocator.resize(larger_size) - - # Open the file and keep a handle; this simulates a worker that has it mapped: - existing_file = File.open(shm_path, "rb") - original_size = existing_file.size - expect(original_size).to be == larger_size - - allocator.close - - # Simulate supervisor restart - recreates allocator at same path: - Async::Service::Supervisor::UtilizationMonitor::SegmentAllocator.new( - shm_path, size: file_size, segment_size: segment_size - ) - - # Our handle still references the original inode; it should not have been resized: - expect(existing_file.size).to be == original_size - ensure - existing_file&.close - end - it "frees segments when workers are removed" do # Register first worker monitor.register(supervisor_controller) From 70ec315212ccbf29dcce8de195bb63fec57861d9 Mon Sep 17 00:00:00 2001 From: Samuel Williams Date: Fri, 21 Aug 2026 10:23:33 +1200 Subject: [PATCH 2/4] Open utilization segment allocator Signed-off-by: Samuel Williams Assisted-By: devx/fec6efe0-6bff-4a9e-a7a1-de183807a581 --- lib/async/service/supervisor/utilization_monitor.rb | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/lib/async/service/supervisor/utilization_monitor.rb b/lib/async/service/supervisor/utilization_monitor.rb index d860fc9..2654c49 100644 --- a/lib/async/service/supervisor/utilization_monitor.rb +++ b/lib/async/service/supervisor/utilization_monitor.rb @@ -28,7 +28,7 @@ def initialize(path: "utilization.shm", interval: 10, size: IO::Buffer::PAGE_SIZ @path = path @segment_size = segment_size - @allocator = Async::Utilization::SegmentAllocator.new(path, size: size, segment_size: segment_size, replace: true) + @allocator = Async::Utilization::SegmentAllocator.open(path, size: size, segment_size: segment_size, replace: true) # Track workers: worker_id => supervisor_controller @workers = {} From e24e66be93f8ac4393e85c2ec05db6c7d8499035 Mon Sep 17 00:00:00 2001 From: Samuel Williams Date: Fri, 21 Aug 2026 11:55:14 +1200 Subject: [PATCH 3/4] Use utilization segment store Signed-off-by: Samuel Williams Assisted-By: devx/fec6efe0-6bff-4a9e-a7a1-de183807a581 --- .../service/supervisor/utilization_monitor.rb | 18 +++++++++--------- releases.md | 2 +- test/async/service/utilization_monitor.rb | 12 ++++++------ 3 files changed, 16 insertions(+), 16 deletions(-) diff --git a/lib/async/service/supervisor/utilization_monitor.rb b/lib/async/service/supervisor/utilization_monitor.rb index 2654c49..40dd888 100644 --- a/lib/async/service/supervisor/utilization_monitor.rb +++ b/lib/async/service/supervisor/utilization_monitor.rb @@ -14,8 +14,8 @@ module Supervisor # Uses shared memory to efficiently collect utilization metrics from workers # and aggregates them by service name for monitoring and reporting. class UtilizationMonitor < Monitor - # @deprecated Use {Async::Utilization::SegmentAllocator} instead. - SegmentAllocator = Async::Utilization::SegmentAllocator + # @deprecated Use {Async::Utilization::SegmentStore} instead. + SegmentAllocator = Async::Utilization::SegmentStore # Initialize a new utilization monitor. # @@ -28,7 +28,7 @@ def initialize(path: "utilization.shm", interval: 10, size: IO::Buffer::PAGE_SIZ @path = path @segment_size = segment_size - @allocator = Async::Utilization::SegmentAllocator.open(path, size: size, segment_size: segment_size, replace: true) + @store = Async::Utilization::SegmentStore.open(path, size: size, segment_size: segment_size, replace: true) # Track workers: worker_id => supervisor_controller @workers = {} @@ -49,7 +49,7 @@ def register(supervisor_controller) return unless worker_id # Allocate a segment first (we'll get schema from worker) - offset = @allocator.allocate(worker_id, []) + offset = @store.allocate(worker_id, []) unless offset Console.warn(self, "Failed to allocate utilization segment", worker_id: worker_id) @@ -67,19 +67,19 @@ def register(supervisor_controller) # Update the allocation with the actual schema if schema && !schema.empty? - @allocator.update_schema(worker_id, schema) + @store.update_schema(worker_id, schema) @workers[worker_id] = supervisor_controller Console.info(self, "Registered worker utilization", worker_id: worker_id, offset: offset, schema: schema) else # Worker didn't provide schema, free the allocation - @allocator.free(worker_id) + @store.free(worker_id) Console.info(self, "Worker did not provide utilization schema", worker_id: worker_id) end end rescue => error Console.error(self, "Error setting up worker utilization", worker_id: worker_id, exception: error) - @allocator.free(worker_id) + @store.free(worker_id) end end end @@ -95,7 +95,7 @@ def remove(supervisor_controller) return unless worker_id @workers.delete(worker_id) - @allocator.free(worker_id) + @store.free(worker_id) Console.debug(self, "Freed utilization segment", worker_id: worker_id) end @@ -145,7 +145,7 @@ def sample def sample_by_worker @guard.synchronize do @workers.each_with_object({}) do |(worker_id, supervisor_controller), workers| - if utilization = @allocator.read(worker_id) + if utilization = @store.read(worker_id) workers[worker_id] = { state: supervisor_controller.state.dup.freeze, utilization: utilization.freeze, diff --git a/releases.md b/releases.md index f472cdd..b420483 100644 --- a/releases.md +++ b/releases.md @@ -2,7 +2,7 @@ ## Unreleased - - Use `Async::Utilization::SegmentAllocator` to manage utilization shared memory. + - Use `Async::Utilization::SegmentStore` to manage utilization shared memory. ## v0.20.0 diff --git a/test/async/service/utilization_monitor.rb b/test/async/service/utilization_monitor.rb index 0358bce..31099b0 100644 --- a/test/async/service/utilization_monitor.rb +++ b/test/async/service/utilization_monitor.rb @@ -359,7 +359,7 @@ ) # Verify initial size - expect(small_monitor.instance_variable_get(:@allocator).size).to be == initial_size + expect(small_monitor.instance_variable_get(:@store).size).to be == initial_size # Create workers to consume all available segments # We need to register enough workers to consume all segments @@ -411,13 +411,13 @@ controller_new.define_singleton_method(:worker){worker_new} # Get size before registering (might trigger resize) - size_before = small_monitor.instance_variable_get(:@allocator).size + size_before = small_monitor.instance_variable_get(:@store).size # This should trigger automatic resize if free list is empty small_monitor.register(controller_new) # Verify the file was resized if it needed to be - final_size = small_monitor.instance_variable_get(:@allocator).size + final_size = small_monitor.instance_variable_get(:@store).size # Size should be >= initial size (might have been resized) expect(final_size).to be >= initial_size @@ -485,7 +485,7 @@ expect(status_before[:data]["test_service"][:connections_total]).to be == 42 # Now trigger a resize: register one more worker — the free list is empty - # so SegmentAllocator#allocate will call resize before handing out a slot. + # so SegmentStore#allocate will call resize before handing out a slot. resize_registry = Async::Utilization::Registry.new resize_worker = Async::Service::Supervisor::Worker.new( process_id: Process.pid, @@ -499,9 +499,9 @@ resize_controller.define_singleton_method(:state){{name: "filler"}} resize_controller.define_singleton_method(:worker){resize_worker} - size_before_resize = small_monitor.instance_variable_get(:@allocator).size + size_before_resize = small_monitor.instance_variable_get(:@store).size small_monitor.register(resize_controller) - size_after_resize = small_monitor.instance_variable_get(:@allocator).size + size_after_resize = small_monitor.instance_variable_get(:@store).size # Confirm the resize actually happened expect(size_after_resize).to be > size_before_resize From e14115680385268401f1e61d108ddfcedaf6aca1 Mon Sep 17 00:00:00 2001 From: Samuel Williams Date: Fri, 21 Aug 2026 12:40:07 +1200 Subject: [PATCH 4/4] Expose utilization segment store Assisted-By: devx/fec6efe0-6bff-4a9e-a7a1-de183807a581 --- .../service/supervisor/utilization_monitor.rb | 3 +++ test/async/service/utilization_monitor.rb | 14 +++++++------- 2 files changed, 10 insertions(+), 7 deletions(-) diff --git a/lib/async/service/supervisor/utilization_monitor.rb b/lib/async/service/supervisor/utilization_monitor.rb index 40dd888..f030219 100644 --- a/lib/async/service/supervisor/utilization_monitor.rb +++ b/lib/async/service/supervisor/utilization_monitor.rb @@ -36,6 +36,9 @@ def initialize(path: "utilization.shm", interval: 10, size: IO::Buffer::PAGE_SIZ @guard = Mutex.new end + # @attribute [Async::Utilization::SegmentStore] The shared utilization segment store. + attr :store + # Register a worker with the utilization monitor. # # Allocates a segment of shared memory and instructs the worker diff --git a/test/async/service/utilization_monitor.rb b/test/async/service/utilization_monitor.rb index 31099b0..6f8bbb0 100644 --- a/test/async/service/utilization_monitor.rb +++ b/test/async/service/utilization_monitor.rb @@ -359,7 +359,7 @@ ) # Verify initial size - expect(small_monitor.instance_variable_get(:@store).size).to be == initial_size + expect(small_monitor.store.size).to be == initial_size # Create workers to consume all available segments # We need to register enough workers to consume all segments @@ -411,15 +411,15 @@ controller_new.define_singleton_method(:worker){worker_new} # Get size before registering (might trigger resize) - size_before = small_monitor.instance_variable_get(:@store).size + size_before = small_monitor.store.size # This should trigger automatic resize if free list is empty small_monitor.register(controller_new) # Verify the file was resized if it needed to be - final_size = small_monitor.instance_variable_get(:@store).size - # Size should be >= initial size (might have been resized) - expect(final_size).to be >= initial_size + final_size = small_monitor.store.size + # Size should be >= the size before registration (might have been resized) + expect(final_size).to be >= size_before # All workers should be registered and readable registry_new.metric(:connections_total).set(100) @@ -499,9 +499,9 @@ resize_controller.define_singleton_method(:state){{name: "filler"}} resize_controller.define_singleton_method(:worker){resize_worker} - size_before_resize = small_monitor.instance_variable_get(:@store).size + size_before_resize = small_monitor.store.size small_monitor.register(resize_controller) - size_after_resize = small_monitor.instance_variable_get(:@store).size + size_after_resize = small_monitor.store.size # Confirm the resize actually happened expect(size_after_resize).to be > size_before_resize