diff options
Diffstat (limited to 'activerecord/lib/active_record/connection_adapters/abstract/connection_pool.rb')
-rw-r--r-- | activerecord/lib/active_record/connection_adapters/abstract/connection_pool.rb | 405 |
1 files changed, 197 insertions, 208 deletions
diff --git a/activerecord/lib/active_record/connection_adapters/abstract/connection_pool.rb b/activerecord/lib/active_record/connection_adapters/abstract/connection_pool.rb index 77a5fe1efb..46c7fc71ac 100644 --- a/activerecord/lib/active_record/connection_adapters/abstract/connection_pool.rb +++ b/activerecord/lib/active_record/connection_adapters/abstract/connection_pool.rb @@ -1,7 +1,8 @@ require 'thread' require 'monitor' require 'set' -require 'active_support/core_ext/module/synchronization' +require 'active_support/core_ext/module/deprecation' +require 'timeout' module ActiveRecord # Raised when a connection could not be obtained within the connection @@ -9,6 +10,10 @@ module ActiveRecord class ConnectionTimeoutError < ConnectionNotEstablished end + # Raised when a connection pool is full and another connection is requested + class PoolFullError < ConnectionNotEstablished + end + module ConnectionAdapters # Connection pool base class for managing Active Record database # connections. @@ -57,10 +62,50 @@ module ActiveRecord # * +wait_timeout+: number of seconds to block and wait for a connection # before giving up and raising a timeout error (default 5 seconds). class ConnectionPool - attr_accessor :automatic_reconnect - attr_reader :spec, :connections - attr_reader :columns, :columns_hash, :primary_keys, :tables - attr_reader :column_defaults + # Every +frequency+ seconds, the reaper will call +reap+ on +pool+. + # A reaper instantiated with a nil frequency will never reap the + # connection pool. + # + # Configure the frequency by setting "reaping_frequency" in your + # database yaml file. + class Reaper + attr_reader :pool, :frequency + + def initialize(pool, frequency) + @pool = pool + @frequency = frequency + end + + def run + return unless frequency + Thread.new(frequency, pool) { |t, p| + while true + sleep t + p.reap + end + } + end + end + + include MonitorMixin + + attr_accessor :automatic_reconnect, :timeout + attr_reader :spec, :connections, :size, :reaper + + class Latch # :nodoc: + def initialize + @mutex = Mutex.new + @cond = ConditionVariable.new + end + + def release + @mutex.synchronize { @cond.broadcast } + end + + def await + @mutex.synchronize { @cond.wait @mutex } + end + end # Creates a new ConnectionPool object. +spec+ is a ConnectionSpecification # object which describes database connection information (e.g. adapter, @@ -69,88 +114,23 @@ module ActiveRecord # # The default ConnectionPool maximum size is 5. def initialize(spec) + super() + @spec = spec # The cache of reserved connections mapped to threads @reserved_connections = {} - # The mutex used to synchronize pool access - @connection_mutex = Monitor.new - @queue = @connection_mutex.new_cond @timeout = spec.config[:wait_timeout] || 5 + @reaper = Reaper.new self, spec.config[:reaping_frequency] + @reaper.run # default max pool size to 5 @size = (spec.config[:pool] && spec.config[:pool].to_i) || 5 + @latch = Latch.new @connections = [] - @checked_out = [] @automatic_reconnect = true - @tables = {} - @visitor = nil - - @columns = Hash.new do |h, table_name| - h[table_name] = with_connection do |conn| - - # Fetch a list of columns - conn.columns(table_name, "#{table_name} Columns").tap do |columns| - - # set primary key information - columns.each do |column| - column.primary = column.name == primary_keys[table_name] - end - end - end - end - - @columns_hash = Hash.new do |h, table_name| - h[table_name] = Hash[columns[table_name].map { |col| - [col.name, col] - }] - end - - @column_defaults = Hash.new do |h, table_name| - h[table_name] = Hash[columns[table_name].map { |col| - [col.name, col.default] - }] - end - - @primary_keys = Hash.new do |h, table_name| - h[table_name] = with_connection do |conn| - table_exists?(table_name) ? conn.primary_key(table_name) : 'id' - end - end - end - - # A cached lookup for table existence. - def table_exists?(name) - return true if @tables.key? name - - with_connection do |conn| - conn.tables.each { |table| @tables[table] = true } - @tables[name] = true if !@tables.key?(name) && conn.table_exists?(name) - end - - @tables.key? name - end - - # Clears out internal caches: - # - # * columns - # * columns_hash - # * tables - def clear_cache! - @columns.clear - @columns_hash.clear - @column_defaults.clear - @tables.clear - end - - # Clear out internal caches for table with +table_name+. - def clear_table_cache!(table_name) - @columns.delete table_name - @columns_hash.delete table_name - @column_defaults.delete table_name - @primary_keys.delete table_name end # Retrieve the connection associated with the current thread, or call @@ -162,18 +142,21 @@ module ActiveRecord @reserved_connections[current_connection_id] ||= checkout end - # Check to see if there is an active connection in this connection - # pool. + # Is there an open connection that is being used for the current thread? def active_connection? - @reserved_connections.key? current_connection_id + @reserved_connections.fetch(current_connection_id) { + return false + }.in_use? end # Signal that the thread is finished with the current connection. # #release_connection releases the connection-thread association # and returns the connection to the pool. def release_connection(with_id = current_connection_id) - conn = @reserved_connections.delete(with_id) - checkin conn if conn + synchronize do + conn = @reserved_connections.delete(with_id) + checkin conn if conn + end end # If a connection already exists yield it to the block. If no connection @@ -181,7 +164,7 @@ module ActiveRecord # connection when finished. def with_connection connection_id = current_connection_id - fresh_connection = true unless @reserved_connections[connection_id] + fresh_connection = true unless active_connection? yield connection ensure release_connection(connection_id) if fresh_connection @@ -189,95 +172,72 @@ module ActiveRecord # Returns true if a connection has already been opened. def connected? - !@connections.empty? + synchronize { @connections.any? } end # Disconnects all connections in the pool, and clears the pool. def disconnect! - @reserved_connections.each do |name,conn| - checkin conn - end - @reserved_connections = {} - @connections.each do |conn| - conn.disconnect! + synchronize do + @reserved_connections = {} + @connections.each do |conn| + checkin conn + conn.disconnect! + end + @connections = [] end - @connections = [] end # Clears the cache which maps classes. def clear_reloadable_connections! - @reserved_connections.each do |name, conn| - checkin conn - end - @reserved_connections = {} - @connections.each do |conn| - conn.disconnect! if conn.requires_reloading? - end - @connections.delete_if do |conn| - conn.requires_reloading? - end - end - - # Verify active connections and remove and disconnect connections - # associated with stale threads. - def verify_active_connections! #:nodoc: - clear_stale_cached_connections! - @connections.each do |connection| - connection.verify! + synchronize do + @reserved_connections = {} + @connections.each do |conn| + checkin conn + conn.disconnect! if conn.requires_reloading? + end + @connections.delete_if do |conn| + conn.requires_reloading? + end end end - # Return any checked-out connections back to the pool by threads that - # are no longer alive. - def clear_stale_cached_connections! - keys = @reserved_connections.keys - Thread.list.find_all { |t| - t.alive? - }.map { |thread| thread.object_id } - keys.each do |key| - checkin @reserved_connections[key] - @reserved_connections.delete(key) - end + def clear_stale_cached_connections! # :nodoc: + reap end + deprecate :clear_stale_cached_connections! => "Please use #reap instead" # Check-out a database connection from the pool, indicating that you want # to use it. You should call #checkin when you no longer need this. # - # This is done by either returning an existing connection, or by creating - # a new connection. If the maximum number of connections for this pool has - # already been reached, but the pool is empty (i.e. they're all being used), - # then this method will wait until a thread has checked in a connection. - # The wait time is bounded however: if no connection can be checked out - # within the timeout specified for this pool, then a ConnectionTimeoutError - # exception will be raised. + # This is done by either returning and leasing existing connection, or by + # creating a new connection and leasing it. + # + # If all connections are leased and the pool is at capacity (meaning the + # number of currently leased connections is greater than or equal to the + # size limit set), an ActiveRecord::PoolFullError exception will be raised. # # Returns: an AbstractAdapter object. # # Raises: - # - ConnectionTimeoutError: no connection can be obtained from the pool - # within the timeout period. + # - PoolFullError: no connection can be obtained from the pool. def checkout - # Checkout an available connection - @connection_mutex.synchronize do - loop do - conn = if @checked_out.size < @connections.size - checkout_existing_connection - elsif @connections.size < @size - checkout_new_connection - end - return conn if conn - - @queue.wait(@timeout) - - if(@checked_out.size < @connections.size) - next - else - clear_stale_cached_connections! - if @size == @checked_out.size - raise ConnectionTimeoutError, "could not obtain a database connection#{" within #{@timeout} seconds" if @timeout}. The max pool size is currently #{@size}; consider increasing it." - end + loop do + # Checkout an available connection + synchronize do + # Try to find a connection that hasn't been leased, and lease it + conn = connections.find { |c| c.lease } + + # If all connections were leased, and we have room to expand, + # create a new connection and lease it. + if !conn && connections.size < size + conn = checkout_new_connection + conn.lease end + return checkout_and_verify(conn) if conn end + + Timeout.timeout(@timeout, PoolFullError) { @latch.await } end end @@ -287,30 +247,58 @@ module ActiveRecord # +conn+: an AbstractAdapter object, which was obtained by earlier by # calling +checkout+ on this pool. def checkin(conn) - @connection_mutex.synchronize do + synchronize do conn.run_callbacks :checkin do - @checked_out.delete conn - @queue.signal + conn.expire end + + release conn + end + @latch.release + end + + # Remove a connection from the connection pool. The connection will + # remain open and active but will no longer be managed by this pool. + def remove(conn) + synchronize do + @connections.delete conn + + # FIXME: we might want to store the key on the connection so that removing + # from the reserved hash will be a little easier. + release conn end + @latch.release end - synchronize :clear_reloadable_connections!, :verify_active_connections!, - :connected?, :disconnect!, :with => :@connection_mutex + # Removes dead connections from the pool. A dead connection can occur + # if a programmer forgets to close a connection at the end of a thread + # or a thread dies unexpectedly. + def reap + synchronize do + stale = Time.now - @timeout + connections.dup.each do |conn| + remove conn if conn.in_use? && stale > conn.last_use && !conn.active? + end + end + @latch.release + end private - def new_connection - connection = ActiveRecord::Base.send(spec.adapter_method, spec.config) + def release(conn) + thread_id = if @reserved_connections[current_connection_id] == conn + current_connection_id + else + @reserved_connections.keys.find { |k| + @reserved_connections[k] == conn + } + end - # TODO: This is a bit icky, and in the long term we may want to change the method - # signature for connections. Also, if we switch to have one visitor per - # connection (and therefore per thread), we can get rid of the thread-local - # variable in Arel::Visitors::ToSql. - @visitor ||= connection.class.visitor_for(self) - connection.visitor = @visitor + @reserved_connections.delete thread_id if thread_id + end - connection + def new_connection + ActiveRecord::Base.send(spec.adapter_method, spec.config) end def current_connection_id #:nodoc: @@ -321,19 +309,14 @@ module ActiveRecord raise ConnectionNotEstablished unless @automatic_reconnect c = new_connection + c.pool = self @connections << c - checkout_and_verify(c) - end - - def checkout_existing_connection - c = (@connections - @checked_out).first - checkout_and_verify(c) + c end def checkout_and_verify(c) c.run_callbacks :checkout do c.verify! - @checked_out << c end c end @@ -363,14 +346,18 @@ module ActiveRecord # ActiveRecord::Base.connection_handler. Active Record models use this to # determine that connection pool that they should use. class ConnectionHandler - attr_reader :connection_pools - - def initialize(pools = {}) + def initialize(pools = Hash.new { |h,k| h[k] = {} }) @connection_pools = pools + @class_to_pool = Hash.new { |h,k| h[k] = {} } + end + + def connection_pools + @connection_pools[Process.pid] end def establish_connection(name, spec) - @connection_pools[name] = ConnectionAdapters::ConnectionPool.new(spec) + set_pool_for_spec spec, ConnectionAdapters::ConnectionPool.new(spec) + set_class_to_pool name, connection_pools[spec] end # Returns true if there are any active connections among the connection @@ -383,21 +370,16 @@ module ActiveRecord # and also returns connections to the pool cached by threads that are no # longer alive. def clear_active_connections! - @connection_pools.each_value {|pool| pool.release_connection } + connection_pools.each_value {|pool| pool.release_connection } end # Clears the cache which maps classes. def clear_reloadable_connections! - @connection_pools.each_value {|pool| pool.clear_reloadable_connections! } + connection_pools.each_value {|pool| pool.clear_reloadable_connections! } end def clear_all_connections! - @connection_pools.each_value {|pool| pool.disconnect! } - end - - # Verify active connections. - def verify_active_connections! #:nodoc: - @connection_pools.each_value {|pool| pool.verify_active_connections! } + connection_pools.each_value {|pool| pool.disconnect! } end # Locate the connection of the nearest super class. This can be an @@ -421,52 +403,56 @@ module ActiveRecord # can be used as an argument for establish_connection, for easily # re-establishing the connection. def remove_connection(klass) - pool = @connection_pools.delete(klass.name) + pool = class_to_pool.delete(klass.name) return nil unless pool + connection_pools.delete pool.spec pool.automatic_reconnect = false pool.disconnect! pool.spec.config end def retrieve_connection_pool(klass) - pool = @connection_pools[klass.name] + pool = get_pool_for_class klass.name return pool if pool - return nil if ActiveRecord::Base == klass - retrieve_connection_pool klass.superclass + return nil if ActiveRecord::Model == klass + retrieve_connection_pool klass.active_record_super end - end - class ConnectionManagement - class Proxy # :nodoc: - attr_reader :body, :testing - - def initialize(body, testing = false) - @body = body - @testing = testing - end + private - def method_missing(method_sym, *arguments, &block) - @body.send(method_sym, *arguments, &block) - end + def class_to_pool + @class_to_pool[Process.pid] + end - def respond_to?(method_sym, include_private = false) - super || @body.respond_to?(method_sym) - end + def set_pool_for_spec(spec, pool) + @connection_pools[Process.pid][spec] = pool + end - def each(&block) - body.each(&block) - end + def set_class_to_pool(name, pool) + @class_to_pool[Process.pid][name] = pool + pool + end - def close - body.close if body.respond_to?(:close) + def get_pool_for_class(klass) + @class_to_pool[Process.pid].fetch(klass) { + c_to_p = @class_to_pool.values.find { |class_to_pool| + class_to_pool[klass] + } - # Don't return connection (and perform implicit rollback) if - # this request is a part of integration test - ActiveRecord::Base.clear_active_connections! unless testing - end + if c_to_p + pool = c_to_p[klass] + pool = ConnectionAdapters::ConnectionPool.new pool.spec + set_pool_for_spec pool.spec, pool + set_class_to_pool klass, pool + else + set_class_to_pool klass, nil + end + } end + end + class ConnectionManagement def initialize(app) @app = app end @@ -474,9 +460,12 @@ module ActiveRecord def call(env) testing = env.key?('rack.test') - status, headers, body = @app.call(env) + response = @app.call(env) + response[2] = ::Rack::BodyProxy.new(response[2]) do + ActiveRecord::Base.clear_active_connections! unless testing + end - [status, headers, Proxy.new(body, testing)] + response rescue ActiveRecord::Base.clear_active_connections! unless testing raise |