Allow async executor to be configurable

This is a followup/alternative to rails#41406. This change wouldn't work for
GitHub because we intend to implement an executor for each database and
use the database configuration to set the `min_threads` and
`max_threads` for each one.

The changes here borrow from rails#41406 by implementing an
`Concurrent::ImmediateExecutor` by default. Otherwise applications have
the option of having one global thread pool that is used by all connections
or a thread pool for each connection. A global thread pool can set with
`config.active_record.async_query_executor = :global_thread_pool`. This
will create a single `Concurrent::ThreadPoolExecutor` for applications
to utilize. By default the concurrency is 4, but it can be changed for the
`global_thread_pool` by setting `global_executor_concurrency` to another
number. If applications want to use a thread pool per database
connection they can set `config.active_record.async_query_executor =
:multi_thread_pool`. This will create a `Concurrent::ThreadPoolExecutor`
for each database connection and set the `min_threads` and `max_threads`
by their configuration values or the defaults.

I've also moved the async tests out of the adapter test and into their
own tests and added tests for all the new functionality. This change
would allow us at GitHub to control threads per database and per
writer/reader or other apps to use one global executor. The immediate
executor allows apps to no-op by default.

Took the immediate executor idea from Jean's PR.
Co-authored-by: Jean Boussier <[email protected]>

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