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 | 453 |
1 files changed, 249 insertions, 204 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 61994d4a47..450ef69744 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,7 @@ require 'thread' require 'monitor' require 'set' -require 'active_support/core_ext/module/synchronization' +require 'active_support/core_ext/module/deprecation' module ActiveRecord # Raised when a connection could not be obtained within the connection @@ -9,6 +9,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 +61,35 @@ 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 # Creates a new ConnectionPool object. +spec+ is a ConnectionSpecification # object which describes database connection information (e.g. adapter, @@ -69,88 +98,39 @@ 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 @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 + # connections available to be checked out + @available = [] - @column_defaults = Hash.new do |h, table_name| - h[table_name] = Hash[columns[table_name].map { |col| - [col.name, col.default] - }] - end + # number of threads waiting to check out a connection + @num_waiting = 0 - @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 + # signal threads waiting + @cond = new_cond 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) + # Hack for tests to be able to add connections. Do not call outside of tests + def insert_connection_for_test!(c) + synchronize do + @connections << c + @available << c 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 @@ -159,21 +139,28 @@ module ActiveRecord # #connection can be called any number of times; the connection is # held in a hash keyed by the thread id. def connection - @reserved_connections[current_connection_id] ||= checkout + synchronize do + @reserved_connections[current_connection_id] ||= checkout + end 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 + synchronize do + @reserved_connections.fetch(current_connection_id) { + return false + }.in_use? + end 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 +168,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 +176,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 = [] + @available = [] 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 + @available.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) + synchronize do + conn = nil - 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 - end + if @num_waiting == 0 + conn = acquire_connection + end + unless conn + conn = wait_until(@timeout) { acquire_connection } end + + conn.lease + checkout_and_verify(conn) end end @@ -287,53 +251,128 @@ 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 + + @available.unshift conn + @cond.signal + end + 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 + @available.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 + + @cond.signal # can make a new connection now end 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 + @cond.broadcast # may violate fairness + end + end private - def new_connection - connection = ActiveRecord::Base.send(spec.adapter_method, spec.config) + # Take an available connection or, if possible, create a new + # one, or nil. + # + # Monitor must be held while calling this method. + # + # Returns: a newly acquired connection. + def acquire_connection + if @available.any? + @available.pop + elsif connections.size < size + checkout_new_connection + end + end + + # Wait on +@cond+ until the block returns non-nil. Note that + # unlike MonitorMixin::ConditionVariable#wait_until, this method + # does not test the block before the first wait period. + # + # Monitor must be held when calling this method. + # + # +timeout+: Integer timeout in seconds + # + # Returns: the result of the block + # + # Raises: + # - PoolFullError: timeout elapsed before +&block+ returned a connection + def wait_until(timeout, &block) + @num_waiting += 1 + begin + t0 = Time.now + loop do + elapsed = Time.now - t0 + if elapsed >= timeout + msg = 'could not obtain a database connection within %0.3f seconds (waited %0.3f seconds)' % + [timeout, elapsed] + raise PoolFullError, msg + end + + @cond.wait(timeout - elapsed) + + conn = yield + return conn if conn + end + ensure + @num_waiting -= 1 + end + 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 + 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 - connection + @reserved_connections.delete thread_id if thread_id + end + + def new_connection + ActiveRecord::Base.send(spec.adapter_method, spec.config) end def current_connection_id #:nodoc: - Thread.current.object_id + ActiveRecord::Base.connection_id ||= Thread.current.object_id end def checkout_new_connection 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 +402,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 +426,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 +459,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[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 +516,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 |