Skip to content

Use async-utilization segment store #13

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
samuel-williams-shopify merged 4 commits into from
Aug 21, 2026
Merged

Use async-utilization segment store #13

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
336131e
Use utilization segment allocator
samuel-williams-shopify Aug 20, 2026
70ec315
Open utilization segment allocator
samuel-williams-shopify Aug 20, 2026
e24e66b
Use utilization segment store
samuel-williams-shopify Aug 20, 2026
e141156
Expose utilization segment store
samuel-williams-shopify Aug 21, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion async-service-supervisor.gemspec
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand Down
194 changes: 13 additions & 181 deletions lib/async/service/supervisor/utilization_monitor.rb
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,8 +3,6 @@
# Released under the MIT License.
# Copyright, 2026, by Samuel Williams.

require "set"

require_relative "monitor"
require "async/utilization"

Expand All @@ -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.
Expand All @@ -199,14 +28,17 @@ 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 = {}

@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
Expand All @@ -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)
Expand All @@ -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
Expand All @@ -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
Expand Down Expand Up @@ -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,
Expand Down
4 changes: 4 additions & 0 deletions releases.md
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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`.
Expand Down
47 changes: 8 additions & 39 deletions test/async/service/utilization_monitor.rb
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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,
Expand All @@ -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
Expand Down
Loading