Class: Async::Limiter::Queued

Inherits:
Generic
  • Object
show all
Defined in:
lib/async/limiter/queued.rb

Overview

Queue-based limiter that distributes pre-existing resources with priority/timeout support.

This limiter manages a finite set of resources (connections, API keys, etc.) that are pre-populated in a queue. It provides priority-based allocation and timeout support for resource acquisition.

Unlike Limited which counts abstract permits, Queued distributes actual resource objects and supports priority queues for fair allocation.

Instance Attribute Summary collapse

Attributes inherited from Generic

#Registry-like object for utilization metrics., #Tags associated with this limiter for identification or categorization., #tags, #utilization

Class Method Summary collapse

Instance Method Summary collapse

Methods inherited from Generic

#acquire, #as_json, #async, #release, #sync, #to_json

Constructor Details

#initialize(queue = self.class.default_queue, **options) ⇒ Queued

Initialize a queue-based limiter.



32
33
34
35
36
37
38
39
40
41
42
 
# File 'lib/async/limiter/queued.rb', line 32

def initialize(queue = self.class.default_queue, **options)
	super(**options)
	@queue = queue
	
	@acquired_count_metric = @utilization.metric(:acquired_count)
	@available_count_metric = @utilization.metric(:available_count)
	@waiting_count_metric = @utilization.metric(:waiting_count)
	@reacquire_waiting_count_metric = @utilization.metric(:reacquire_waiting_count)
	
	update_utilization_metrics
end

Instance Attribute Details

#queue ⇒ Object (readonly)

Returns the value of attribute queue.



45
46
47
 
# File 'lib/async/limiter/queued.rb', line 45

def queue
  @queue
end

Class Method Details

.default_queue ⇒ Object 



23
24
25
26
27
 
# File 'lib/async/limiter/queued.rb', line 23

def self.default_queue
	Queue.new.tap do |queue|
		queue.push(true)
	end
end

Instance Method Details

#acquired_count ⇒ Object 



48
49
50
 
# File 'lib/async/limiter/queued.rb', line 48

def acquired_count
	@acquired_count_metric.value
end

#available_count ⇒ Object 



53
54
55
 
# File 'lib/async/limiter/queued.rb', line 53

def available_count
	@queue.size
end

#expand(count, value = true) ⇒ Object

Expand the queue with additional resources.



76
77
78
79
80
81
82
 
# File 'lib/async/limiter/queued.rb', line 76

def expand(count, value = true)
	count.times do
		@queue.push(value)
	end
	
	update_utilization_metrics
end

#limited? ⇒ Boolean

Check if a new task can be acquired.

Returns:

  • (Boolean) 


69
70
71
 
# File 'lib/async/limiter/queued.rb', line 69

def limited?
	@queue.empty?
end

#reacquire_waiting_count ⇒ Object 



63
64
65
 
# File 'lib/async/limiter/queued.rb', line 63

def reacquire_waiting_count
	@reacquire_waiting_count_metric.value
end

#statistics ⇒ Object

Get current limiter statistics.



86
87
88
89
90
91
92
93
94
95
96
97
98
 
# File 'lib/async/limiter/queued.rb', line 86

def statistics
	@mutex.synchronize do
		{
			waiting: @queue.waiting_count,
			available: @queue.size,
			acquired_count: @acquired_count_metric.value,
			available_count: @queue.size,
			waiting_count: @queue.waiting_count,
			reacquire_waiting_count: @reacquire_waiting_count_metric.value,
			timing: @timing.statistics
		}
	end
end

#The queue managing resources.=(queuemanagingresources. = (value)) ⇒ Object 



45
 
# File 'lib/async/limiter/queued.rb', line 45

attr_reader :queue

#waiting_count ⇒ Object 



58
59
60
 
# File 'lib/async/limiter/queued.rb', line 58

def waiting_count
	@queue.waiting_count
end