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

#Tags associated with this limiter for identification or categorization., #tags

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, timing: Timing::None, parent: nil) ⇒ Queued

Initialize a queue-based limiter.



33
34
35
36
37
38
 
# File 'lib/async/limiter/queued.rb', line 33

def initialize(queue = self.class.default_queue, timing: Timing::None, parent: nil)
	super(timing: timing, parent: parent)
	@queue = queue
	@acquired_count = 0
	@reacquire_waiting_count = 0
end

Instance Attribute Details

#queue ⇒ Object (readonly)

Returns the value of attribute queue.



41
42
43
 
# File 'lib/async/limiter/queued.rb', line 41

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 



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

def acquired_count
	@mutex.synchronize{@acquired_count}
end

#available_count ⇒ Object 



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

def available_count
	@queue.size
end

#expand(count, value = true) ⇒ Object

Expand the queue with additional resources.



72
73
74
75
76
 
# File 'lib/async/limiter/queued.rb', line 72

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

#limited? ⇒ Boolean

Check if a new task can be acquired.

Returns:

  • (Boolean) 


65
66
67
 
# File 'lib/async/limiter/queued.rb', line 65

def limited?
	@queue.empty?
end

#reacquire_waiting_count ⇒ Object 



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

def reacquire_waiting_count
	@mutex.synchronize{@reacquire_waiting_count}
end

#statistics ⇒ Object

Get current limiter statistics.



80
81
82
83
84
85
86
87
88
89
90
91
92
 
# File 'lib/async/limiter/queued.rb', line 80

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

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



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

attr_reader :queue

#waiting_count ⇒ Object 



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

def waiting_count
	@queue.waiting_count
end