Ticket #2162: active_record_connection_pooling4.diff

activerecord/test/fixtures/db_definitions/mysql.sql

@@ -188 +188 @@
-CREATE TABLE `fk_test_has_fk` (
-  `id`    INTEGER NOT NULL PRIMARY KEY,
-  `fk_id` INTEGER NOT NULL,
-
+
-  FOREIGN KEY (`fk_id`) REFERENCES `fk_test_has_pk`(`id`)
-) TYPE=InnoDB;
-

activerecord/test/pool_connections_test.rb

@@ +1 @@
+require 'abstract_unit'
+require 'fixtures/topic'
+
+class PoolConnectionsTest < Test::Unit::TestCase
+  self.use_transactional_fixtures = false
+  
+  fixtures :topics
+    
+  def setup
+    super
+    ActiveRecord::Base.pool_connections = true
+    ActiveRecord::Base.pool_connections_min = 2
+    ActiveRecord::Base.pool_connections_max = 5    
+    ActiveRecord::Base.establish_connection(ActiveRecord::Base.remove_connection)    
+  end
+    
+  def test_with_pooling_on
+    # assert that pool was created by establish_connection
+    assert_not_nil(ActiveRecord::Base.connection.pool)
+    
+    # assert connection pooling works propertly with subclasses
+    assert_not_nil(Topic.connection.pool)
+    assert_equal(Topic.connection, ActiveRecord::Base.connection)
+    
+    # Test that pool_connections_min connections initially established
+    assert_equal(2, ActiveRecord::Base.connection.pool_size)
+    
+    # Test that additional connections are added to pool as needed, and threads wait
+    # when all connections are out
+    ActiveRecord::Base.pool_connections_wait = true
+    ActiveRecord::Base.connection.borrow do
+      Thread.new do
+        Topic.find :first
+        # should be 2 connections because we created 2 at startup (pool_connections_min=2)
+        assert_equal(2, ActiveRecord::Base.connection.pool_size)
+        
+        # borrow connection then start new thread while holding connection, each new thread
+        # should increase size of pool as the connections are all in use
+        ActiveRecord::Base.connection.borrow do
+          Thread.new do
+            Topic.transaction do
+              assert_equal(3, ActiveRecord::Base.connection.pool_size)
+              ActiveRecord::Base.connection.borrow do
+                Thread.new do
+                  Topic.find :first
+                  assert_equal(4, ActiveRecord::Base.connection.pool_size)                
+                  ActiveRecord::Base.connection.borrow do
+                    Thread.new do
+                      Topic.find :first
+                      
+                      # We should have 5 in there now which is what we set pool_connections_max to
+                      assert_equal(5, ActiveRecord::Base.connection.pool_size)                    
+                      5.times do
+                        # Make these 5 threads share the one connection that is available in the pool
+                        # instead of creating more connections when an empty pool is found
+                        Thread.new do
+                          assert_not_nil(Topic.find(:first))
+                          assert_not_nil(Topic.find(:first))
+                          assert_equal(5, ActiveRecord::Base.connection.pool_size)
+                        end.join
+                      end                                                                
+                    end.join
+                  end                                
+                end.join
+              end
+            end
+          end.join
+        end          
+      end.join      
+    end
+  end
+  
+  def test_no_wait    
+    # Try with pool_connections_wait set to false
+    ActiveRecord::Base.pool_connections_wait = false
+    
+    # borrow connection then start new thread while holding connection, each new thread
+    # should increase size of pool as the connections are all in use
+    ActiveRecord::Base.connection.borrow do
+      Thread.new do
+        ActiveRecord::Base.connection.borrow do
+          Thread.new do
+            ActiveRecord::Base.connection.borrow do
+              Thread.new do
+                ActiveRecord::Base.connection.borrow do
+                  Thread.new do
+                    ActiveRecord::Base.connection.borrow do
+                      Thread.new do
+                        # No more left in pool (5 are in use by threads) and no more can
+                        # be created, should throw ConnectionNotEstablished
+                        assert_raise(ActiveRecord::ConnectionNotEstablished) {Topic.find :first}
+                        assert_equal(5, ActiveRecord::Base.connection.pool_size)
+                      end.join
+                    end
+                  end.join
+                end
+              end.join
+            end
+          end.join
+        end
+      end.join
+    end            
+    assert_equal(5, ActiveRecord::Base.connection.pool_size)                
+  end
+  
+  def test_reaper
+    # Test that connections are released properly
+    
+    # set reaper period to 3 seconds (default is 30) so we don't take forever to run the test
+    ActiveRecord::Base.pool_connections_reaper_period = 3
+    ActiveRecord::Base.connection.pool.reset
+    ActiveRecord::Base.connection.borrow do
+      Thread.new do
+        ActiveRecord::Base.connection.borrow do
+          Thread.new do
+            ActiveRecord::Base.connection.borrow do
+            end
+          end.join
+        end
+      end.join
+    end
+    assert_equal(3, ActiveRecord::Base.connection.pool_size)
+    sleep(3)
+    
+    # No connection should have been closed on last run of reaper
+    assert_equal(3, ActiveRecord::Base.connection.pool_size)
+    ActiveRecord::Base.connection.borrow do
+      ActiveRecord::Base.connection.borrow do
+      end
+    end
+    sleep(3)
+    
+    # One should have been closed on last run of reaper
+    assert_equal(2, ActiveRecord::Base.connection.pool_size)
+    ActiveRecord::Base.connection.borrow do
+    end
+    sleep(4)
+    
+    # None should have been closed on last run of reaper
+    assert_equal(2,ActiveRecord::Base.connection.pool_size)
+  end
+  
+  def test_transactions
+    # Test transactions
+    Topic.transaction do
+      t = Topic.find :first
+      t.content = "I changed the content!"
+      t.save
+      
+      # If another thread accesses the Topic before the transaction was committed
+      # it should get the old value
+      Thread.new do
+        t2 = Topic.find :first
+        assert_equal(t2.content, "Have a nice day")
+      end.join      
+    end        
+  end  
+    
+end

activerecord/lib/active_record/connection_adapters/abstract/connection_specification.rb

@@ -106 +106 @@
-    # can be used as argument for establish_connection, for easy
-    # re-establishing of the connection.
-    def self.remove_connection(klass=self)
-
+      
-      @@defined_connections.delete(klass)
-      active_connections[klass] = nil
-      @connection = nil
@@ -115 +115 @@
-
-    # Set the connection for the class.
-    def self.connection=(spec)
-
+ or spec.kind_of?(ActiveRecord::ConnectionPool::Wrapper)
-        active_connections[self] = spec
-      elsif spec.kind_of?(ConnectionSpecification)
-        self.connection = self.send(spec.adapter_method, spec.config)

activerecord/lib/active_record/connection_adapters/abstract_adapter.rb

@@ -44 +44 @@
-        @runtime = 0
-        return rt
-      end
+      
+      def close
+        begin
+          @connection.close
+        rescue Exception => e
+          log_info("error while closing real connection: #{e.message}")
+          raise ActiveRecord::ActiveRecordError, "error while closing real connection: #{e.message}"
+        end
+      end
-
-      protected  
-        def log(sql, name)

activerecord/lib/active_record/connection_pool.rb

@@ +1 @@
+require 'monitor'
+
+module ActiveRecord
+
+  class Base
+  
+    # Determines whether or not to use a connection pool to manage the actual connections.  
+    # Is tested and applied on a class by class basis.  When set to true, @@allow_concurrency is 
+    # effectively disregarded
+    cattr_accessor :pool_connections
+    @@pool_connections = false
+
+    cattr_accessor :connection_pool
+    @@connection_pool=nil
+    
+    # Initial and minimum size of the connection pool.  Meaningful only when @@pool_connections is true
+    cattr_accessor :pool_connections_min
+    @@pool_connections_min = 0
+
+    # Maximum size of the connection pool.  Meaningful only when @@pool_connections is true
+    cattr_accessor :pool_connections_max
+    @@pool_connections_max = 0
+
+    # Whether or not a thread will wait for a pool connection to become available, 
+    # or if ConnectionNotEstablished is thrown
+    cattr_accessor :pool_connections_wait
+    @@pool_connections_wait = true
+   
+    # Period in seconds between sweeping for unused connections in the pool
+    cattr_accessor :pool_connections_reaper_period
+    @@pool_connections_reaper_period = 30
+    
+    # alias adapter methods to get a ConnectionPool::Wrapper when pooling is enabled    
+    class << self
+      #alias_method :mysql_connection_raw, :mysql_connection
+      adapter_methods = RAILS_CONNECTION_ADAPTERS.collect {|a| "#{a}_connection" }
+      adapter_methods.each do |am|
+        if method_defined?(am.to_sym)
+          alias_method "#{am}_raw".to_sym, am.to_sym
+          module_eval <<-EOM
+            def #{am}(*args)
+              if pool_connections
+                if @@connection_pool.nil?
+                  @@connection_pool = ConnectionPool.new(Proc.new {#{am}_raw(*args)})
+                  @@connection_pool.startup
+                end                
+                return @@connection_pool.wrapper
+              else
+                return #{am}_raw(*args)
+              end              
+            end
+          EOM
+        end
+      end
+      
+      alias_method :remove_connection_raw, :remove_connection
+      def remove_connection(klass=self)      
+        if @@connection_pool
+          @@connection_pool.shutdown
+          @@connection_pool = nil
+        end
+        remove_connection_raw(klass)
+      end
+    end
+  end
+
+  class ConnectionPool
+    include MonitorMixin
+    
+    def initialize(prc)
+      super()
+      @prc = prc
+    end
+    
+    def wrapper
+      Wrapper.new(self)
+    end
+    
+    def startup
+      @min_size,@max_size = ActiveRecord::Base.pool_connections_min,ActiveRecord::Base.pool_connections_max
+      @available_connections = []
+      @acquired_connections = []
+      @min_size.times { @available_connections << @prc.call }
+      @cond_var ||= new_cond
+      @use_check = 0
+      @use_thread = Thread.new { reap_connections }
+    end
+    
+    def shutdown
+      synchronize do
+        @cond_var.wait_until { @acquired_connections.size==0 }
+        # TODO - connection_adapters should have a close method to close the physical connection
+        #@available_connections.each do |c|
+        #  c.close          
+        #end
+        @available_connections = nil        
+      end
+      @use_thread.terminate
+    end    
+    
+    def reset
+      shutdown
+      startup
+    end
+    
+    def current_size
+      return @available_connections.size + @acquired_connections.size
+    end
+    
+    def reap_connections
+      Base.logger.debug("starting reaper thread")
+      loop do
+        sleep(ActiveRecord::Base.pool_connections_reaper_period)
+        synchronize do
+          Base.logger.debug("use_check is #{@use_check}")
+          if @use_check < current_size
+            num_to_free = current_size - @use_check
+            max_to_free = current_size-@min_size
+            num_to_free = max_to_free if num_to_free > max_to_free
+            Base.logger.debug("freeing #{num_to_free} connections")            
+            @available_connections[0...num_to_free].each do |conn|              
+              conn.close
+              @available_connections.delete(conn)
+            end
+          end
+        end
+        @use_check=0
+      end
+    end
+        
+    def acquire_connection
+      conn = nil
+      synchronize do
+        loop do
+          if @available_connections.size > 0
+            conn = @available_connections.shift
+            break
+          elsif (@max_size > 0 and current_size < @max_size) or @max_size==0
+            conn = @prc.call
+            break
+          else
+            if ActiveRecord::Base.pool_connections_wait
+              @cond_var.wait
+            else
+              raise ActiveRecord::ConnectionNotEstablished
+            end
+          end  
+        end
+        @acquired_connections << conn
+        @use_check=@acquired_connections.size if @acquired_connections.size > @use_check
+        Base.logger.debug("current_size=" + current_size.to_s + " max_size=" + @max_size.to_s)       
+      end
+      conn
+    end
+    
+    def release_connection(conn)
+      synchronize do
+        @acquired_connections.delete(conn)
+        @available_connections << conn
+        @cond_var.signal
+      end
+    end
+    
+    # The Wrapper class is used as a proxy for the real connection adapter.
+    # All methods are forwarded to the underlying connections in the pool via
+    # the borrow_connection method, which ensures connections go back to the pool
+    # after the method is executed on the connection
+    class Wrapper
+      attr_reader :pool
+      
+      # copy definition of transaction method from database_statements.rb so that internal transaction 
+      # methods (begin_db_transaction, etc.) go through the wrapper
+      def transaction(start_db_transaction = true)
+        transaction_open = false
+        begin
+          if block_given?
+            if start_db_transaction
+              begin_db_transaction 
+              transaction_open = true
+            end
+            yield
+          end
+        rescue Exception => database_transaction_rollback
+          if transaction_open
+            transaction_open = false
+            rollback_db_transaction
+          end
+          raise
+        end
+      ensure
+        commit_db_transaction if transaction_open
+      end
+      
+      def initialize(pool)
+        @pool=pool
+      end
+      
+      def indexes(table_name, name=nil)
+        borrow_connection {|c| c.indexes(table_name, name) }
+      end
+      
+      def tables(name=nil)
+        borrow_connection {|c| c.tables(name) }
+      end
+      
+      # Provides a real connection to a block and returns the value of the block.
+      # Primary purpose is the enforce proper use of pooled connections
+      def borrow_connection(&block)
+        result = nil 
+        if block_given?
+          if Thread.current['activerecord_connection'].nil?
+            begin              
+              Thread.current['activerecord_connection'] = @pool.acquire_connection
+              result = yield Thread.current['activerecord_connection']
+            ensure
+              if !Thread.current['activerecord_connection'].nil?
+                @pool.release_connection(Thread.current['activerecord_connection'])
+                Thread.current['activerecord_connection'] = nil
+              end
+            end            
+          else
+            result = yield Thread.current['activerecord_connection']
+          end
+        end
+        result
+      end
+      alias_method :borrow, :borrow_connection
+
+      def begin_db_transaction
+        if Thread.current['activerecord_connection'].nil?
+          Thread.current['activerecord_connection'] = @pool.acquire_connection
+          Thread.current['activerecord_connection'].begin_db_transaction
+        end
+      end
+
+      def commit_db_transaction
+        if !Thread.current['activerecord_connection'].nil?
+          Thread.current['activerecord_connection'].commit_db_transaction
+          @pool.release_connection(Thread.current['activerecord_connection'])
+          Thread.current['activerecord_connection'] = nil
+        end
+      end
+
+      def rollback_db_transaction
+        if !Thread.current['activerecord_connection'].nil?
+          Thread.current['activerecord_connection'].rollback_db_transaction        
+          @pool.release_connection(Thread.current['activerecord_connection'])
+          Thread.current['activerecord_connection'] = nil
+        end
+      end
+      
+      def method_missing(method, *args, &proc)
+        borrow_connection do |conn|
+          conn.send(method, *args, &proc)
+        end
+      end
+      
+      # For testability more than anything
+      def pool_size
+        return @pool.current_size
+      end
+      
+      def shutdown_pool
+        @pool.shutdown
+      end
+    end
+    
+  end  
+end

activerecord/lib/active_record.rb

@@ -73 +73 @@
-  require "active_record/connection_adapters/#{adapter}_adapter"
-end
-
+require 'active_record/connection_pool'
-require 'active_record/query_cache'