Class: ActiveRecord::ConnectionAdapters::ConnectionPool::Queue
Relationships & Source Files | |
Extension / Inclusion / Inheritance Descendants | |
Subclasses:
|
|
Inherits: | Object |
Defined in: | activerecord/lib/active_record/connection_adapters/abstract/connection_pool/queue.rb |
Overview
Active Record Connection Pool Queue
Threadsafe, fair, LIFO queue. Meant to be used by ::ActiveRecord::ConnectionAdapters::ConnectionPool
with which it shares a Monitor.
Class Method Summary
- .new(lock = Monitor.new) ⇒ Queue constructor
Instance Attribute Summary
-
#any_waiting? ⇒ Boolean
readonly
Test if any threads are currently waiting on the queue.
-
#any? ⇒ Boolean
readonly
private
Test if the queue currently contains any elements.
-
#can_remove_no_wait? ⇒ Boolean
readonly
private
A thread can remove an element from the queue without waiting if and only if the number of currently available connections is strictly greater than the number of waiting threads.
Instance Method Summary
-
#add(element)
Add
element
to the queue. -
#add_back(element)
Add
element
to the back of the queue. -
#clear
Remove all elements from the queue.
-
#delete(element)
If
element
is in the queue, remove and return it, ornil
. -
#num_waiting
Returns the number of threads currently waiting on this queue.
-
#poll(timeout = nil)
Remove the head of the queue.
-
#size
Number of elements in the queue.
- #internal_poll(timeout) private
-
#no_wait_poll
private
Remove and return the head of the queue if the number of available elements is strictly greater than the number of threads currently waiting.
-
#remove
private
Removes and returns the head of the queue if possible, or
nil
. - #synchronize(&block) private
-
#wait_poll(timeout)
private
Waits on the queue up to
timeout
seconds, then removes and returns the head of the queue.
Constructor Details
.new(lock = Monitor.new) ⇒ Queue
# File 'activerecord/lib/active_record/connection_adapters/abstract/connection_pool/queue.rb', line 13
def initialize(lock = Monitor.new) @lock = lock @cond = @lock.new_cond @num_waiting = 0 @queue = [] end
Instance Attribute Details
#any? ⇒ Boolean
(readonly, private)
Test if the queue currently contains any elements.
# File 'activerecord/lib/active_record/connection_adapters/abstract/connection_pool/queue.rb', line 100
def any? !@queue.empty? end
#any_waiting? ⇒ Boolean
(readonly)
Test if any threads are currently waiting on the queue.
# File 'activerecord/lib/active_record/connection_adapters/abstract/connection_pool/queue.rb', line 21
def any_waiting? synchronize do @num_waiting > 0 end end
#can_remove_no_wait? ⇒ Boolean
(readonly, private)
A thread can remove an element from the queue without waiting if and only if the number of currently available connections is strictly greater than the number of waiting threads.
# File 'activerecord/lib/active_record/connection_adapters/abstract/connection_pool/queue.rb', line 108
def can_remove_no_wait? @queue.size > @num_waiting end
Instance Method Details
#add(element)
Add element
to the queue. Never blocks.
# File 'activerecord/lib/active_record/connection_adapters/abstract/connection_pool/queue.rb', line 36
def add(element) synchronize do @queue.push element @cond.signal end end
#add_back(element)
Add element
to the back of the queue. Never blocks.
# File 'activerecord/lib/active_record/connection_adapters/abstract/connection_pool/queue.rb', line 44
def add_back(element) synchronize do @queue.unshift element @cond.signal end end
#clear
Remove all elements from the queue.
# File 'activerecord/lib/active_record/connection_adapters/abstract/connection_pool/queue.rb', line 59
def clear synchronize do @queue.clear end end
#delete(element)
If element
is in the queue, remove and return it, or nil
.
# File 'activerecord/lib/active_record/connection_adapters/abstract/connection_pool/queue.rb', line 52
def delete(element) synchronize do @queue.delete(element) end end
#internal_poll(timeout) (private)
[ GitHub ]# File 'activerecord/lib/active_record/connection_adapters/abstract/connection_pool/queue.rb', line 91
def internal_poll(timeout) no_wait_poll || (timeout && wait_poll(timeout)) end
#no_wait_poll (private)
Remove and return the head of the queue if the number of available elements is strictly greater than the number of threads currently waiting. Otherwise, return nil
.
# File 'activerecord/lib/active_record/connection_adapters/abstract/connection_pool/queue.rb', line 120
def no_wait_poll remove if can_remove_no_wait? end
#num_waiting
Returns the number of threads currently waiting on this queue.
# File 'activerecord/lib/active_record/connection_adapters/abstract/connection_pool/queue.rb', line 29
def num_waiting synchronize do @num_waiting end end
#poll(timeout = nil)
Remove the head of the queue.
If timeout
is not given, remove and return the head of the queue if the number of available elements is strictly greater than the number of threads currently waiting (that is, don’t jump ahead in line). Otherwise, return nil
.
If timeout
is given, block if there is no element available, waiting up to timeout
seconds for an element to become available.
Raises:
-
::ActiveRecord::ConnectionTimeoutError
iftimeout
is given and no element
becomes available within timeout
seconds,
# File 'activerecord/lib/active_record/connection_adapters/abstract/connection_pool/queue.rb', line 86
def poll(timeout = nil) synchronize { internal_poll(timeout) } end
#remove (private)
Removes and returns the head of the queue if possible, or nil
.
# File 'activerecord/lib/active_record/connection_adapters/abstract/connection_pool/queue.rb', line 113
def remove @queue.pop end
#size
Number of elements in the queue.
# File 'activerecord/lib/active_record/connection_adapters/abstract/connection_pool/queue.rb', line 66
def size synchronize do @queue.size end end
#synchronize(&block) (private)
[ GitHub ]# File 'activerecord/lib/active_record/connection_adapters/abstract/connection_pool/queue.rb', line 95
def synchronize(&block) @lock.synchronize(&block) end
#wait_poll(timeout) (private)
Waits on the queue up to timeout
seconds, then removes and returns the head of the queue.
# File 'activerecord/lib/active_record/connection_adapters/abstract/connection_pool/queue.rb', line 126
def wait_poll(timeout) @num_waiting += 1 t0 = Process.clock_gettime(Process::CLOCK_MONOTONIC) elapsed = 0 loop do ActiveSupport::Dependencies.interlock.permit_concurrent_loads do @cond.wait(timeout - elapsed) end return remove if any? elapsed = Process.clock_gettime(Process::CLOCK_MONOTONIC) - t0 if elapsed >= timeout msg = "could not obtain a connection from the pool within %0.3f seconds (waited %0.3f seconds); all pooled connections were in use" % [timeout, elapsed] raise ConnectionTimeoutError, msg end end ensure @num_waiting -= 1 end