Merge pull request rails#41495 from eileencodes/make-executor-configurable

Allow async executor to be configurable

Author: eileencodes

activerecord/CHANGELOG.md

@@ -1,3 +1,17 @@
+*   Allow applications to configure the thread pool for async queries
+
+    Some applications may want one thread pool per database whereas others want to use
+    a single global thread pool for all queries. By default Rails will set `async_query_executor`
+    to `:immediate` and create a `Concurrent::ImmediateExecutor` object which is essentially a no-op.
+    To create one thread pool for all database connections to use applications can set
+    `config.active_record.async_query_executor` to `:global_thread_pool` and optionally define
+    `config.active_record.global_executor_concurrency`. This defaults to 4. For applications that want
+    to have a thread pool for each database connection, `config.active_record.async_query_executor` can
+    be set to `:multi_thread_pool`. The configuration for each thread pool is set in the database
+    configuration.
+
+    *Eileen M. Uchitelle*
+
 *   Allow new syntax for `enum` to avoid leading `_` from reserved options.
 
     Before:

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

@@ -144,12 +144,7 @@ def initialize(pool_config)
 
         @lock_thread = false
 
-        @async_executor = Concurrent::ThreadPoolExecutor.new(
-          min_threads: 0,
-          max_threads: @size,
-          max_queue: @size * 4,
-          fallback_policy: :caller_runs
-        )
+        @async_executor = build_async_executor
 
         @reaper = Reaper.new(self, db_config.reaping_frequency)
         @reaper.run
@@ -463,6 +458,22 @@ def schedule_query(future_result) # :nodoc:
       end
 
       private
+        def build_async_executor
+          case Base.async_query_executor
+          when :multi_thread_pool
+            Concurrent::ThreadPoolExecutor.new(
+              min_threads: @db_config.min_threads,
+              max_threads: @db_config.max_threads,
+              max_queue: @db_config.max_queue,
+              fallback_policy: :caller_runs
+            )
+          when :global_thread_pool
+            Base.global_thread_pool_async_query_executor
+          else
+            Base.immediate_query_executor
+          end
+        end
+
         #--
         # this is unfortunately not concurrent
         def bulk_make_new_connections(num_new_conns_needed)

activerecord/lib/active_record/core.rb

@@ -157,6 +157,44 @@ def self.configurations
 
       mattr_accessor :application_record_class, instance_accessor: false, default: nil
 
+      # Sets the async_query_executor for an application. By default the thread pool executor
+      # set to `:immediate. Options are:
+      #
+      #   * :immediate - Initializes a single +Concurrent::ImmediateExecutor+
+      #   * :global_thread_pool - Initializes a single +Concurrent::ThreadPoolExecutor+
+      #   that uses the +async_query_concurrency+ for the +max_threads+ value.
+      #   * :multi_thread_pool - Initializes a +Concurrent::ThreadPoolExecutor+ for each
+      #   database connection. The initializer values are defined in the configuration hash.
+      mattr_accessor :async_query_executor, instance_accessor: false, default: :immediate
+
+      def self.immediate_query_executor # :nodoc:
+        @@immediate_query_executor ||= Concurrent::ImmediateExecutor.new
+      end
+
+      def self.global_thread_pool_async_query_executor # :nodoc:
+        concurrency = global_executor_concurrency || 4
+        @@global_thread_pool_async_query_executor ||= Concurrent::ThreadPoolExecutor.new(
+          min_threads: 0,
+          max_threads: concurrency,
+          max_queue: concurrency * 4,
+          fallback_policy: :caller_runs
+        )
+      end
+
+      # Set the +global_executor_concurrency+. This configuration value can only be used
+      # with the global thread pool async query executor.
+      def self.global_executor_concurrency=(global_executor_concurrency)
+        if async_query_executor == :immediate || async_query_executor == :multi_thread_pool
+          raise ArgumentError, "`global_executor_concurrency` cannot be set when using either immediate or multiple thread pools. For multiple thread pools, please set the concurrency in your database configuration. Immediate thread pools are essentially a no-op."
+        end
+
+        @@global_executor_concurrency = global_executor_concurrency
+      end
+
+      def self.global_executor_concurrency # :nodoc:
+        @@global_executor_concurrency ||= nil
+      end
+
       def self.application_record_class? # :nodoc:
         if Base.application_record_class
           self == Base.application_record_class

activerecord/lib/active_record/database_configurations/database_config.rb

@@ -48,6 +48,18 @@ def pool
         raise NotImplementedError
       end
 
+      def min_threads
+        raise NotImplementedError
+      end
+
+      def max_threads
+        raise NotImplementedError
+      end
+
+      def max_queue
+        raise NotImplementedError
+      end
+
       def checkout_timeout
         raise NotImplementedError
       end

activerecord/lib/active_record/database_configurations/hash_config.rb

@@ -66,6 +66,18 @@ def pool
         (configuration_hash[:pool] || 5).to_i
       end
 
+      def min_threads
+        (configuration_hash[:min_threads] || 0).to_i
+      end
+
+      def max_threads
+        (configuration_hash[:max_threads] || pool).to_i
+      end
+
+      def max_queue
+        max_threads * 4
+      end
+
       def checkout_timeout
         (configuration_hash[:checkout_timeout] || 5).to_f
       end

activerecord/test/cases/adapter_test.rb

@@ -343,122 +343,6 @@ def test_in_clause_length_is_deprecated
     end
   end
 
-  module AsynchronousQueriesSharedTests
-    def test_async_select_failure
-      ActiveRecord::Base.asynchronous_queries_tracker.start_session
-
-      future_result = @connection.select_all "SELECT * FROM does_not_exists", async: true
-      assert_kind_of ActiveRecord::FutureResult, future_result
-      assert_raises ActiveRecord::StatementInvalid do
-        future_result.result
-      end
-    ensure
-      ActiveRecord::Base.asynchronous_queries_tracker.finalize_session
-    end
-
-    def test_async_query_from_transaction
-      ActiveRecord::Base.asynchronous_queries_tracker.start_session
-
-      assert_nothing_raised do
-        @connection.select_all "SELECT * FROM posts", async: true
-      end
-
-      @connection.transaction do
-        assert_raises AsynchronousQueryInsideTransactionError do
-          @connection.select_all "SELECT * FROM posts", async: true
-        end
-      end
-    ensure
-      ActiveRecord::Base.asynchronous_queries_tracker.finalize_session
-    end
-
-    def test_async_query_cache
-      ActiveRecord::Base.asynchronous_queries_tracker.start_session
-
-      @connection.enable_query_cache!
-
-      @connection.select_all "SELECT * FROM posts"
-      result = @connection.select_all "SELECT * FROM posts", async: true
-      assert_equal Result, result.class
-    ensure
-      ActiveRecord::Base.asynchronous_queries_tracker.finalize_session
-      @connection.disable_query_cache!
-    end
-
-    def test_async_query_foreground_fallback
-      status = {}
-
-      subscriber = ActiveSupport::Notifications.subscribe("sql.active_record") do |event|
-        if event.payload[:sql] == "SELECT * FROM does_not_exists"
-          status[:executed] = true
-          status[:async] = event.payload[:async]
-        end
-      end
-
-      @connection.pool.stub(:schedule_query, proc { }) do
-        future_result = @connection.select_all "SELECT * FROM does_not_exists", async: true
-        assert_kind_of ActiveRecord::FutureResult, future_result
-        assert_raises ActiveRecord::StatementInvalid do
-          future_result.result
-        end
-      end
-
-      assert_equal true, status[:executed]
-      assert_equal false, status[:async]
-    ensure
-      ActiveSupport::Notifications.unsubscribe(subscriber) if subscriber
-    end
-  end
-
-  class AsynchronousQueriesTest < ActiveRecord::TestCase
-    self.use_transactional_tests = false
-
-    include AsynchronousQueriesSharedTests
-
-    def setup
-      @connection = ActiveRecord::Base.connection
-    end
-
-    def test_async_select_all
-      ActiveRecord::Base.asynchronous_queries_tracker.start_session
-      status = {}
-
-      monitor = Monitor.new
-      condition = monitor.new_cond
-
-      subscriber = ActiveSupport::Notifications.subscribe("sql.active_record") do |event|
-        if event.payload[:sql] == "SELECT * FROM posts"
-          status[:executed] = true
-          status[:async] = event.payload[:async]
-          monitor.synchronize { condition.signal }
-        end
-      end
-
-      future_result = @connection.select_all "SELECT * FROM posts", async: true
-      assert_kind_of ActiveRecord::FutureResult, future_result
-
-      monitor.synchronize do
-        condition.wait_until { status[:executed] }
-      end
-      assert_kind_of ActiveRecord::Result, future_result.result
-      assert_equal @connection.supports_concurrent_connections?, status[:async]
-    ensure
-      ActiveRecord::Base.asynchronous_queries_tracker.finalize_session
-      ActiveSupport::Notifications.unsubscribe(subscriber) if subscriber
-    end
-  end
-
-  class AsynchronousQueriesWithTransactionalTest < ActiveRecord::TestCase
-    self.use_transactional_tests = true
-
-    include AsynchronousQueriesSharedTests
-
-    def setup
-      @connection = ActiveRecord::Base.connection
-      @connection.materialize_transactions
-    end
-  end
-
   class AdapterForeignKeyTest < ActiveRecord::TestCase
     self.use_transactional_tests = false