Class: Async::Service::Supervisor::MemoryMonitor

Inherits:
Monitor
  • Object
show all
Defined in:
lib/async/service/supervisor/memory_monitor.rb

Overview

Monitors worker memory usage and restarts workers that exceed limits.

Uses the ‘memory` gem to track process memory and detect leaks.

Instance Attribute Summary collapse

Instance Method Summary collapse

Methods inherited from Monitor

#run, #status, #to_json

Constructor Details

#initialize(interval: 10, total_size_limit: nil, free_size_minimum: nil, **options) ⇒ MemoryMonitor

Create a new memory monitor.



24
25
26
27
28
29
30
31
32
33
34
35
 
# File 'lib/async/service/supervisor/memory_monitor.rb', line 24

def initialize(interval: 10, total_size_limit: nil, free_size_minimum: nil, **options)
	super(interval: interval)
	@cluster = Memory::Leak::Cluster.new(total_size_limit: total_size_limit, free_size_minimum: free_size_minimum)
	
	# We use these options when adding processes to the cluster:
	@options = options
	
	@processes = Hash.new{|hash, key| hash[key] = Set.new.compare_by_identity}
	
	# Queue to serialize cluster modifications to prevent race conditions:
	@guard = Mutex.new
end

Instance Attribute Details

#cluster ⇒ Object (readonly)

Returns the value of attribute cluster.



38
39
40
 
# File 'lib/async/service/supervisor/memory_monitor.rb', line 38

def cluster
  @cluster
end

#The cluster of processes being monitored.(clusterofprocessesbeingmonitored.) ⇒ Object (readonly) 



38
 
# File 'lib/async/service/supervisor/memory_monitor.rb', line 38

attr_reader :cluster

Instance Method Details

#add(process_id) ⇒ Object

Add a process to the memory monitor. You may override this to control how processes are added to the cluster.



43
44
45
 
# File 'lib/async/service/supervisor/memory_monitor.rb', line 43

def add(process_id)
	@cluster.add(process_id, **@options)
end

#as_json ⇒ Object

Serialize memory cluster data for JSON.



89
90
91
 
# File 'lib/async/service/supervisor/memory_monitor.rb', line 89

def as_json
	@cluster.as_json
end

#memory_leak_detected(process_id, monitor) ⇒ Object

Invoked when a memory leak is detected.



98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
 
# File 'lib/async/service/supervisor/memory_monitor.rb', line 98

def memory_leak_detected(process_id, monitor)
	Console.warn(self, "Memory leak detected!", child: {process_id: process_id}, monitor: monitor)
	
	# Kill the process gently:
	begin
		Console.info(self, "Killing process!", child: {process_id: process_id})
		Process.kill(:INT, process_id)
	rescue Errno::ESRCH
		# No such process - he's dead Jim.
	rescue => error
		Console.warn(self, "Failed to kill process!", child: {process_id: process_id}, exception: error)
	end
	
	true
end

#register(supervisor_controller) ⇒ Object

Register a worker with the memory monitor.



50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
 
# File 'lib/async/service/supervisor/memory_monitor.rb', line 50

def register(supervisor_controller)
	process_id = supervisor_controller.process_id
	return unless process_id
	
	Console.debug(self, "Registering worker.", supervisor_controller: supervisor_controller, process_id: process_id)
	
	@guard.synchronize do
		controllers = @processes[process_id]
		
		if controllers.empty?
			Console.debug(self, "Registering process.", child: {process_id: process_id})
			self.add(process_id)
		end
		
		controllers.add(supervisor_controller)
	end
end

#remove(supervisor_controller) ⇒ Object

Remove a worker from the memory monitor.



71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
 
# File 'lib/async/service/supervisor/memory_monitor.rb', line 71

def remove(supervisor_controller)
	process_id = supervisor_controller.process_id
	return unless process_id
	
	@guard.synchronize do
		controllers = @processes[process_id]
		
		controllers.delete(supervisor_controller)
		
		if controllers.empty?
			Console.debug(self, "Removing process.", child: {process_id: process_id})
			@cluster.remove(process_id)
			@processes.delete(process_id)
		end
	end
end

#run_once ⇒ Object

Run one iteration of the memory monitor.



115
116
117
118
119
120
121
122
123
124
125
126
 
# File 'lib/async/service/supervisor/memory_monitor.rb', line 115

def run_once
	@guard.synchronize do
		# This block must return true if the process was killed.
		@cluster.check! do |process_id, monitor|
			begin
				memory_leak_detected(process_id, monitor)
			rescue => error
				Console.error(self, "Failed to handle memory leak!", child: {process_id: process_id}, exception: error)
			end
		end
	end
end