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..f030219 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::SegmentStore} instead. + SegmentAllocator = Async::Utilization::SegmentStore + # 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) + @store = Async::Utilization::SegmentStore.open(path, size: size, segment_size: segment_size, replace: true) # Track workers: worker_id => supervisor_controller @workers = {} @@ -207,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 @@ -220,7 +52,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) @@ -238,19 +70,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 @@ -266,7 +98,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 @@ -316,7 +148,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 73e40ec..b420483 100644 --- a/releases.md +++ b/releases.md @@ -1,5 +1,9 @@ # Releases +## Unreleased + + - Use `Async::Utilization::SegmentStore` 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..6f8bbb0 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) @@ -390,7 +359,7 @@ ) # Verify initial size - expect(small_monitor.instance_variable_get(:@allocator).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 @@ -442,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(:@allocator).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(:@allocator).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) @@ -516,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, @@ -530,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.store.size small_monitor.register(resize_controller) - size_after_resize = small_monitor.instance_variable_get(:@allocator).size + size_after_resize = small_monitor.store.size # Confirm the resize actually happened expect(size_after_resize).to be > size_before_resize