From fe575dd4a9f0fa0e71a89fae9f4a951a9fb36058 Mon Sep 17 00:00:00 2001 From: Nick Sieger Date: Thu, 7 Aug 2008 01:27:40 -0700 Subject: Nearing the finish line. Initial fixed-size connection pool implemented, more docs --- .../abstract/connection_pool.rb | 138 +++++++++++++++++---- 1 file changed, 116 insertions(+), 22 deletions(-) (limited to 'activerecord/lib/active_record/connection_adapters/abstract/connection_pool.rb') 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 28446fd64d..97c6e3e886 100644 --- a/activerecord/lib/active_record/connection_adapters/abstract/connection_pool.rb +++ b/activerecord/lib/active_record/connection_adapters/abstract/connection_pool.rb @@ -2,14 +2,43 @@ require 'monitor' require 'set' module ActiveRecord + # Raised when a connection could not be obtained within the connection + # acquisition timeout period. + class ConnectionTimeoutError < ConnectionNotEstablished + end + module ConnectionAdapters - # Connection pool base class and ActiveRecord database connections. - class ConnectionPool + # Connection pool base class for managing ActiveRecord database + # connections. + # + # Connections can be obtained and used from a connection pool in several + # ways: + # + # 1. Simply use ActiveRecord::Base.connection as in pre-connection-pooled + # ActiveRecord. Eventually, when you're done with the connection and + # wish it to be returned to the pool, you call + # ActiveRecord::Base.connection_pool.release_thread_connection. + # 2. Manually check out a connection from the pool with + # ActiveRecord::Base.connection_pool.checkout. You are responsible for + # returning this connection to the pool when finished by calling + # ActiveRecord::Base.connection_pool.checkin(connection). + # 3. Use ActiveRecord::Base.connection_pool.with_connection(&block), which + # obtains a connection, yields it as the sole argument to the block, + # and returns it to the pool after the block completes. + class AbstractConnectionPool # Factory method for connection pools. - # Determines pool type to use based on contents of connection specification. - # FIXME: specification configuration TBD. + # Determines pool type to use based on contents of connection + # specification. Additional options for connection specification: + # + # * +pool+: number indicating size of fixed connection pool to use + # * +wait_timeout+ (optional): number of seconds to block and wait + # for a connection before giving up and raising a timeout error. def self.create(spec) - ConnectionPerThread.new(spec) + if spec.config[:pool] && spec.config[:pool].to_i > 0 + ConnectionPool.new(spec) + else + ConnectionPerThread.new(spec) + end end delegate :verification_timeout, :to => "::ActiveRecord::Base" @@ -115,11 +144,16 @@ module ActiveRecord end private :connections - synchronize :connection, :release_thread_connection, :checkout, :checkin, + synchronize :connection, :release_thread_connection, :clear_reloadable_connections!, :verify_active_connections!, :connected?, :disconnect!, :with => :@connection_mutex private + def new_connection + config = spec.config.reverse_merge(:allow_concurrency => ActiveRecord::Base.allow_concurrency) + ActiveRecord::Base.send(spec.adapter_method, config) + end + def active_connection_name #:nodoc: Thread.current.object_id end @@ -139,7 +173,10 @@ module ActiveRecord end end - class ConnectionPerThread < ConnectionPool + # ConnectionPerThread is a simple implementation: always create/disconnect + # on checkout/checkin, and use the base class reserved connections hash to + # manage the per-thread connections. + class ConnectionPerThread < AbstractConnectionPool def active_connection @reserved_connections[active_connection_name] end @@ -155,12 +192,6 @@ module ActiveRecord end private - # Set the connection for the class. - def new_connection - config = spec.config.reverse_merge(:allow_concurrency => ActiveRecord::Base.allow_concurrency) - ActiveRecord::Base.send(spec.adapter_method, config) - end - def connections @reserved_connections.values end @@ -170,6 +201,70 @@ module ActiveRecord end end + # ConnectionPool provides a full, fixed-size connection pool with timed + # waits when the pool is exhausted. + class ConnectionPool < AbstractConnectionPool + def initialize(spec) + super + # default 5 second timeout + @timeout = spec.config[:wait_timeout] || 5 + @size = spec.config[:pool].to_i + @queue = @connection_mutex.new_cond + @connections = [] + @checked_out = [] + end + + def checkout + # Checkout an available connection + conn = @connection_mutex.synchronize do + if @connections.length < @size + checkout_new_connection + elsif @checked_out.size < @connections.size + checkout_existing_connection + end + end + return conn if conn + + # No connections available; wait for one + @connection_mutex.synchronize do + if @queue.wait(@timeout) + checkout_existing_connection + else + raise ConnectionTimeoutError, "could not obtain a database connection in a timely fashion" + end + end + end + + def checkin(conn) + @connection_mutex.synchronize do + @checked_out -= conn + @queue.signal + end + end + + private + def checkout_new_connection + c = new_connection + @connections << c + @checked_out << c + c + end + + def checkout_existing_connection + c = [@connections - @checked_out].first + @checked_out << c + c + end + + def connections + @connections + end + + def remove_connection(conn) + @connections.delete conn + end + end + module ConnectionHandlerMethods def initialize(pools = {}) @connection_pools = pools @@ -180,7 +275,7 @@ module ActiveRecord end def establish_connection(name, spec) - @connection_pools[name] = ConnectionAdapters::ConnectionPool.create(spec) + @connection_pools[name] = ConnectionAdapters::AbstractConnectionPool.create(spec) end # for internal use only and for testing; @@ -238,15 +333,14 @@ module ActiveRecord pool.spec.config if pool end - private - def retrieve_connection_pool(klass) - loop do - pool = @connection_pools[klass.name] - return pool if pool - return nil if ActiveRecord::Base == klass - klass = klass.superclass - end + def retrieve_connection_pool(klass) + loop do + pool = @connection_pools[klass.name] + return pool if pool + return nil if ActiveRecord::Base == klass + klass = klass.superclass end + end end # This connection handler is not thread-safe, as it does not protect access -- cgit v1.2.3