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. -
#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.
- #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 85
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 93
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
#clear
Remove all elements from the queue.
# File 'activerecord/lib/active_record/connection_adapters/abstract/connection_pool/queue.rb', line 51
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 44
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 76
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 105
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 71
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 98
def remove @queue.pop end
#synchronize(&block) (private)
[ GitHub ]# File 'activerecord/lib/active_record/connection_adapters/abstract/connection_pool/queue.rb', line 80
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 111
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