123456789_123456789_123456789_123456789_123456789_

Class: ActiveRecord::ConnectionAdapters::ConnectionPool::Queue

Relationships & Source Files
Extension / Inclusion / Inheritance Descendants
Subclasses:
ConnectionLeasingQueue
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

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

Constructor Details

.new(lock = Monitor.new) ⇒ Queue

[ GitHub ] 

  


# File 'activerecord/lib/active_record/connection_adapters/abstract/connection_pool/queue.rb', line 14



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.

  





  


  [ GitHub ]


  

  


# File 'activerecord/lib/active_record/connection_adapters/abstract/connection_pool/queue.rb', line 103



def any?
  !@queue.empty?
end


  








  

    #any_waiting?  ⇒ Boolean  (readonly)
  



  

    
Test if any threads are currently waiting on the queue.

  





  


  [ GitHub ]


  

  


# File 'activerecord/lib/active_record/connection_adapters/abstract/connection_pool/queue.rb', line 22



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.

  





  


  [ GitHub ]


  

  


# File 'activerecord/lib/active_record/connection_adapters/abstract/connection_pool/queue.rb', line 111



def can_remove_no_wait?
  @queue.size > @num_waiting
end


  







Instance Method Details



  

    #add(element)  
  



  

    
Add element to the queue.  Never blocks.

  



  [ GitHub ]


  

  


# File 'activerecord/lib/active_record/connection_adapters/abstract/connection_pool/queue.rb', line 37



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.

  



  [ GitHub ]


  

  


# File 'activerecord/lib/active_record/connection_adapters/abstract/connection_pool/queue.rb', line 45



def add_back(element)
  synchronize do
    @queue.unshift element
    @cond.signal
  end
end


  








  

    #clear  
  



  

    
Remove all elements from the queue.

  



  [ GitHub ]


  

  


# File 'activerecord/lib/active_record/connection_adapters/abstract/connection_pool/queue.rb', line 60



def clear
  synchronize do
    @queue.clear
  end
end


  








  

    #delete(element)  
  



  

    
If element is in the queue, remove and return it, or nil.

  



  [ GitHub ]


  

  


# File 'activerecord/lib/active_record/connection_adapters/abstract/connection_pool/queue.rb', line 53



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 94



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.

  



  [ GitHub ]


  

  


# File 'activerecord/lib/active_record/connection_adapters/abstract/connection_pool/queue.rb', line 123



def no_wait_poll
  remove if can_remove_no_wait?
end


  








  

    #num_waiting  
  



  

    
Returns the number of threads currently waiting on this
queue.

  



  [ GitHub ]


  

  


# File 'activerecord/lib/active_record/connection_adapters/abstract/connection_pool/queue.rb', line 30



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:



becomes available within timeout seconds,

  



  [ GitHub ]


  

  


# File 'activerecord/lib/active_record/connection_adapters/abstract/connection_pool/queue.rb', line 89



def poll(timeout = nil)
  synchronize { internal_poll(timeout) }
end


  








  

    #remove   (private)
  



  

    
Removes and returns the head of the queue if possible, or nil.

  



  [ GitHub ]


  

  


# File 'activerecord/lib/active_record/connection_adapters/abstract/connection_pool/queue.rb', line 116



def remove
  @queue.pop
end


  








  

    #size  
  



  

    
Number of elements in the queue.

  



  [ GitHub ]


  

  


# File 'activerecord/lib/active_record/connection_adapters/abstract/connection_pool/queue.rb', line 67



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 98



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.

  



  [ GitHub ]


  

  


# File 'activerecord/lib/active_record/connection_adapters/abstract/connection_pool/queue.rb', line 129



def wait_poll(timeout)
  @num_waiting += 1

  t0 = Process.clock_gettime(Process::CLOCK_MONOTONIC)
  elapsed = 0
  loop do
    @cond.wait(timeout - elapsed)

    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