Class: Async::Utilization::SegmentStore

Inherits:
Object
  • Object
show all
Defined in:
lib/async/utilization/segment_store.rb

Overview

Represents a shared memory segment store for utilization data.

Stores fixed-size segments in a shared memory file, associates each segment with a utilization schema, and reads the resulting values.

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(file, buffer, size:, segment_size:, growth_factor:) ⇒ SegmentStore

Initialize the shared memory segment store.



70
71
72
73
74
75
76
77
78
79
80
81
82
83
 
# File 'lib/async/utilization/segment_store.rb', line 70

def initialize(file, buffer, size:, segment_size:, growth_factor:)
	@file = file
	@buffer = buffer
	@size = size
	@segment_size = segment_size
	@growth_factor = growth_factor
	
	@allocations = {}
	@free_list = []
	
	(0...(@size / @segment_size)).each do |segment_index|
		@free_list << (segment_index * @segment_size)
	end
end

Instance Attribute Details

#size ⇒ Object (readonly)

Returns the value of attribute size.



123
124
125
 
# File 'lib/async/utilization/segment_store.rb', line 123

def size
  @size
end

#The current size of the shared memory file.(currentsizeofthesharedmemoryfile.) ⇒ Object (readonly) 



123
 
# File 'lib/async/utilization/segment_store.rb', line 123

attr :size

Class Method Details

.open(path, size: IO::Buffer::PAGE_SIZE * 8, segment_size: 512, growth_factor: 2, replace: false) ⇒ Object

Open a shared memory segment store.

Raises:

  • (ArgumentError) 


27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
 
# File 'lib/async/utilization/segment_store.rb', line 27

def self.open(path, size: IO::Buffer::PAGE_SIZE * 8, segment_size: 512, growth_factor: 2, replace: false)
	raise ArgumentError, "Size must be a positive integer!" unless size.is_a?(Integer) && size > 0
	raise ArgumentError, "Segment size must be a positive integer!" unless segment_size.is_a?(Integer) && segment_size > 0
	raise ArgumentError, "Segment size must not exceed size!" if segment_size > size
	raise ArgumentError, "Growth factor must be greater than 1!" unless growth_factor.is_a?(Numeric) && growth_factor.real? && growth_factor > 1
	
	if replace
		begin
			File.unlink(path)
		rescue Errno::ENOENT
			# The file does not need to be replaced:
		end
	end
	
	file = File.open(path, "w+bx")
	buffer = nil
	
	begin
		file.truncate(size)
		buffer = IO::Buffer.map(file, size)
		store = new(file, buffer, size: size, segment_size: segment_size, growth_factor: growth_factor)
	rescue
		buffer&.free
		file.close
		raise
	end
	
	return store unless block_given?
	
	begin
		yield store
	ensure
		store.close
	end
end

Instance Method Details

#allocate(key, schema) ⇒ Object

Allocate a segment for the given key.

The shared memory file is automatically resized if no segments are available.



92
93
94
95
96
97
98
99
100
101
102
103
 
# File 'lib/async/utilization/segment_store.rb', line 92

def allocate(key, schema)
	if @free_list.empty?
		unless resize(@size * @growth_factor)
			return nil
		end
	end
	
	offset = @free_list.shift
	@allocations[key] = {offset: offset, schema: schema}
	
	return offset
end

#allocation(key) ⇒ Object

Get the allocation information for the given key.



118
119
120
 
# File 'lib/async/utilization/segment_store.rb', line 118

def allocation(key)
	@allocations[key]
end

#close ⇒ Object

Close the shared memory file.



199
200
201
202
203
204
205
 
# File 'lib/async/utilization/segment_store.rb', line 199

def close
	@buffer&.free
	@buffer = nil
	
	@file&.close
	@file = nil
end

#free(key) ⇒ Object

Free the segment allocated to the given key.



108
109
110
111
112
 
# File 'lib/async/utilization/segment_store.rb', line 108

def free(key)
	if allocation = @allocations.delete(key)
		@free_list << allocation[:offset]
	end
end

#read(key) ⇒ Object

Read utilization data from an allocated segment.



139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
 
# File 'lib/async/utilization/segment_store.rb', line 139

def read(key)
	allocation = @allocations[key]
	return nil unless allocation
	
	offset = allocation[:offset]
	schema = allocation[:schema]
	
	result = {}
	schema.each do |field_key, type, field_offset|
		absolute_offset = offset + field_offset
		
		begin
			result[field_key] = @buffer.get_value(type, absolute_offset)
		rescue => error
			Console.warn(self, "Failed to read value", type: type, key: field_key, offset: absolute_offset, exception: error)
		end
	end
	
	return result
end

#resize(new_size) ⇒ Object

Resize the shared memory file.

The new size is rounded up to the nearest page boundary.



166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
 
# File 'lib/async/utilization/segment_store.rb', line 166

def resize(new_size)
	old_size = @size
	return false if new_size <= old_size
	
	page_size = IO::Buffer::PAGE_SIZE
	new_size = (((new_size + page_size - 1) / page_size) * page_size).to_i
	
	begin
		@file.truncate(new_size)
		buffer = IO::Buffer.map(@file, new_size)
		
		@buffer&.free
		@buffer = buffer
		
		old_segment_count = old_size / @segment_size
		new_segment_count = new_size / @segment_size
		
		(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

#update_schema(key, schema) ⇒ Object

Update the schema for an existing allocation.



129
130
131
132
133
 
# File 'lib/async/utilization/segment_store.rb', line 129

def update_schema(key, schema)
	if allocation = @allocations[key]
		allocation[:schema] = schema
	end
end