MONGOID-5445 Add load_async to criteria

docs/reference/queries.txt

@@ -2388,5 +2388,67 @@ will query the database and separately cache their results.
 To use the cached results, call ``all.to_a.first`` on the model class.
 
 
-.. _legacy-query-cache-limitations:
+.. _load-async:
 
+Asynchronous Queries
+====================
+
+Mongoid allows running database queries asynchronously in the background.
+This can be beneficial when there is a need to get documents from different
+collections.
+
+In order to schedule an asynchronous query call ``load_async`` method on a
+``Criteria``:
+
+.. code-block:: ruby
+
+  class PagesController < ApplicationController
+    def index
+      @active_bands = Band.where(active: true).load_async
+      @best_events = Event.best.load_async
+      @public_articles = Article.where(public: true).load_async
+    end
+  end
+
+In the above example three queries will be scheduled for asynchronous execution.
+Results of the queries can be later accessed as usual:
+
+.. code-block:: html
+
+  <ul>
+    <%- @active_bands.each do -%>
+      <li><%= band.name %></li>
+    <%- end -%>
+  </ul>
+
+Even if a query is scheduled for asynchronous execution, it might be executed
+synchronously on the caller's thread. There are three possible scenarios depending
+on when the query results are being accessed:
+
+#. If the scheduled asynchronous task has been already executed, the results are returned.
+#. If the task has been started, but not finished yet, the caller's thread blocks until the task is finished.
+#. If the task has not been started yet, it is removed from the execution queue, and the query is executed synchronously on the caller's thread.
+
+.. note::
+
+  Even though ``load_async`` method returns ``Criteria`` object, you should not
+  do any operations on this object except accessing query results. The query is
+  scheduled for execution immediately after calling ``load_async``, therefore
+  later changes to the `Criteria`` object may not be applied.
+
+
+Configuring asynchronous query execution
+----------------------------------------
+
+Asynchronous queries are disabled by default. In order to enable them, the
+following config options should be set:
+
+.. code-block:: yaml
+
+  development:
+    ...
+    options:
+      # Execute asynchronous queries using a global thread pool.
+      async_query_executor: :global_thread_pool
+      # Number of threads in the pool. This option is defaulted to 4.
+      global_executor_concurrency: 5

docs/release-notes/mongoid-8.1.txt

@@ -17,6 +17,11 @@ The complete list of releases is available `on GitHub
 please consult GitHub releases for detailed release notes and JIRA for
 the complete list of issues fixed in each release, including bug fixes.
 
+Added ``load_async`` method on ``Criteria`` to asynchronously load documents
+----------------------------------------------------------------------------
+
+The new ``load_async`` method on ``Criteria`` allows :ref:`running database queries asynchronously <load-async>`.
+
 
 Added ``attribute_before_last_save``, ``saved_change_to_attribute``, ``saved_change_to_attribute?``, and ``will_save_change_to_attribute?``  methods
 ----------------------------------------------------------------------------------------------------------------------------------------------------
@@ -245,8 +250,8 @@ for more details.
 Added ``readonly!`` method and ``legacy_readonly`` feature flag
 ---------------------------------------------------------------
 
-Mongoid 8.1 changes the meaning of read-only documents. In Mongoid 8.1 with 
-this feature flag turned off, a document becomes read-only when calling the 
+Mongoid 8.1 changes the meaning of read-only documents. In Mongoid 8.1 with
+this feature flag turned off, a document becomes read-only when calling the
 ``readonly!`` method:
 
 .. code:: ruby
@@ -261,8 +266,8 @@ this feature flag turned off, a document becomes read-only when calling the
 With this feature flag turned off, a ``ReadonlyDocument`` error will be
 raised when destroying or deleting, as well as when saving or updating.
 
-Prior to Mongoid 8.1 and in 8.1 with the ``legacy_readonly`` feature flag 
-turned on, documents become read-only when they are projected (i.e. using 
+Prior to Mongoid 8.1 and in 8.1 with the ``legacy_readonly`` feature flag
+turned on, documents become read-only when they are projected (i.e. using
 ``#only`` or ``#without``).
 
 .. code:: ruby
@@ -278,7 +283,7 @@ turned on, documents become read-only when they are projected (i.e. using
   band.destroy # => raises ReadonlyDocument error
 
 Note that with this feature flag on, a ``ReadonlyDocument`` error will only be
-raised when destroying or deleting, and not on saving or updating. See the 
+raised when destroying or deleting, and not on saving or updating. See the
 section on :ref:`Read-only Documents <readonly-documents>` for more details.
 
 

lib/mongoid.rb

@@ -132,4 +132,14 @@ def discriminator_key=(value)
   class << self
     prepend GlobalDiscriminatorKeyAssignment
   end
+
+  def global_thread_pool_async_query_executor
+    concurrency = Mongoid.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
 end

lib/mongoid/config.rb

@@ -127,7 +127,20 @@ module Config
     # always return a Hash.
     option :legacy_attributes, default: false
 
-    # When this flag is false, a document will become read-only only once the 
+    # Sets the async_query_executor for an application. By default the thread pool executor
+    #   is 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.
+    option :async_query_executor, default: :immediate
+
+    # Defines how many asynchronous queries can be executed concurrently.
+    # This option should be set only if `async_query_executor` is set
+    # to `:global_thread_pool`, it will be ignored otherwise.
+    option :global_executor_concurrency, default: nil
+
+    # When this flag is false, a document will become read-only only once the
     # #readonly! method is called, and an error will be raised on attempting
     # to save or update such documents, instead of just on delete. When this
     # flag is true, a document is only read-only if it has been projected

lib/mongoid/contextual.rb

@@ -35,6 +35,18 @@ def context
       @context ||= create_context
     end
 
+    # Instructs the context to schedule an asynchronous loading of documents
+    # specified by the criteria.
+    #
+    # Note that depending on the context and on the Mongoid configuration,
+    # documents can be loaded synchronously on the caller's thread.
+    #
+    # @return [ Criteria ] Returns self.
+    def load_async
+      context.load_async if context.respond_to?(:load_async)
+      self
+    end
+
     private
 
     # Create the context for the queries to execute. Will be memory for

lib/mongoid/contextual/mongo.rb

@@ -1,5 +1,6 @@
 # frozen_string_literal: true
 
+require "mongoid/contextual/mongo/documents_loader"
 require "mongoid/contextual/atomic"
 require "mongoid/contextual/aggregable/mongo"
 require "mongoid/contextual/command"
@@ -37,6 +38,8 @@ class Mongo
       # @attribute [r] view The Mongo collection view.
       attr_reader :view
 
+      attr_reader :documents_loader
+
       # Get the number of documents matching the query.
       #
       # @example Get the number of matching documents.
@@ -777,6 +780,17 @@ def third_to_last!
         third_to_last || raise_document_not_found_error
       end
 
+      # Schedule a task to load documents for the context.
+      #
+      # Depending on the Mongoid configuration, the scheduled task can be executed
+      # immediately on the caller's thread, or can be scheduled for an
+      # asynchronous execution.
+      #
+      # @api private
+      def load_async
+        @documents_loader ||= DocumentsLoader.new(view, klass, criteria)
+      end
+
       private
 
       # Update the documents for the provided method.
@@ -844,24 +858,29 @@ def inverse_sorting
         Hash[sort.map{|k, v| [k, -1*v]}]
       end
 
-      # Get the documents the context should iterate. This follows 3 rules:
-      #
-      # 1. If the query is cached, and we already have documents loaded, use
-      #   them.
-      # 2. If we are eager loading, then eager load the documents and use
-      #   those.
-      # 3. Use the query.
-      #
-      # @api private
+      # Get the documents the context should iterate.
       #
-      # @example Get the documents for iteration.
-      #   context.documents_for_iteration
+      # If the documents have been already preloaded by `Document::Loader`
+      # instance, they will be used.
       #
       # @return [ Array<Document> | Mongo::Collection::View ] The docs to iterate.
+      #
+      # @api private
       def documents_for_iteration
-        return view unless eager_loadable?
-        docs = view.map{ |doc| Factory.from_db(klass, doc, criteria) }
-        eager_load(docs)
+        if @documents_loader
+          if @documents_loader.started?
+            @documents_loader.value!
+          else
+            @documents_loader.unschedule
+            @documents_loader.execute
+          end
+        else
+          return view unless eager_loadable?
+          docs = view.map do |doc|
+            Factory.from_db(klass, doc, criteria)
+          end
+          eager_load(docs)
+        end
       end
 
       # Yield to the document.

lib/mongoid/contextual/mongo/documents_loader.rb

@@ -0,0 +1,135 @@
+require "mongoid/association/eager_loadable"
+
+module Mongoid
+  module Contextual
+    class Mongo
+      # Loads documents for the provided criteria.
+      #
+      # @api private
+      class DocumentsLoader
+        extend Forwardable
+        include Association::EagerLoadable
+
+        def_delegators :@future, :value!, :value, :wait!, :wait
+
+        # Synchronous executor to be used when async_query_executor config option
+        # is set to :immediate. This excutor runs all operations on the current
+        # thread, blocking as necessary.
+        IMMEDIATE_EXECUTOR = Concurrent::ImmediateExecutor.new
+
+        # Returns suitable executor according to Mongoid config options.
+        #
+        # @return [ Concurrent::ImmediateExecutor | Concurrent::ThreadPoolExecutor ] The executor
+        #   to be used to execute document loading tasks.
+        def self.executor
+          case Mongoid.async_query_executor
+          when :immediate
+            IMMEDIATE_EXECUTOR
+          when :global_thread_pool
+            Mongoid.global_thread_pool_async_query_executor
+          end
+        end
+
+        # @return [ Mongoid::Criteria ] Criteria that specifies which documents should
+        #   be loaded. Exposed here because `eager_loadable?` method from
+        #   `Association::EagerLoadable` expects is to be available.
+        attr_accessor :criteria
+
+        # Instantiates the document loader instance and immediately schedules
+        # its execution using the provided executor.
+        #
+        # @param [ Mongo::Collection::View ] view The collection view to get
+        #   records from the database.
+        # @param [ Class ] klass Mongoid model class to instantiate documents.
+        #   All records obtained from the database will be converted to an
+        #   instance of this class, if possible.
+        # @param [ Mongoid::Criteria ] criteria. Criteria that specifies which
+        #   documents should be loaded.
+        # @param [ Concurrent::AbstractExecutorService ] executor. Executor that
+        #   is capable of running `Concurrent::Promises::Future` instances.
+        def initialize(view, klass, criteria, executor: self.class.executor)
+          @view = view
+          @klass = klass
+          @criteria = criteria
+          @mutex = Mutex.new
+          @state = :pending
+          @future = Concurrent::Promises.future_on(executor, self) do |task|
+            if task.pending?
+              task.send(:start)
+              task.execute
+            end
+          end
+        end
+
+        # Returns false or true whether the loader is in pending state.
+        #
+        # Pending state means that the loader execution has been scheduled,
+        # but has not been started yet.
+        #
+        # @return [ true | false ] true if the loader is in pending state,
+        #   otherwise false.
+        def pending?
+          @mutex.synchronize do
+            @state == :pending
+          end
+        end
+
+        # Returns false or true whether the loader is in started state.
+        #
+        # Started state means that the loader execution has been started.
+        # Note that the loader stays in this state even after the execution
+        # completed (successfully or failed).
+        #
+        # @return [ true | false ] true if the loader is in started state,
+        #   otherwise false.
+        def started?
+          @mutex.synchronize do
+            @state == :started
+          end
+        end
+
+        # Mark the loader as unscheduled.
+        #
+        # If the loader is marked unscheduled, it will not be executed. The only
+        # option to load the documents is to call `execute` method directly.
+        #
+        # Please note that if execution of a task has been already started,
+        # unscheduling does not have any effect.
+        def unschedule
+          @mutex.synchronize do
+            @state = :cancelled unless @state == :started
+          end
+        end
+
+        # Loads records specified by `@criteria` from the database, and convert
+        # them to Mongoid documents of `@klass` type.
+        #
+        # This method is called by the task (possibly asynchronous) scheduled
+        # when creating an instance of the loader. However, this method can be
+        # called directly, if it is desired to execute loading on the caller
+        # thread immediately.
+        #
+        # Calling this method does not change the state of the loader.
+        #
+        # @return [ Array<Mongoid::Document> ] Array of document loaded from
+        #   the database.
+        def execute
+          documents = @view.map do |doc|
+            Factory.from_db(@klass, doc, @criteria)
+          end
+          eager_load(documents) if eager_loadable?
+          documents
+        end
+
+        private
+
+        # Mark the loader as started.
+        def start
+          @mutex.synchronize do
+            @state = :started
+          end
+        end
+      end
+    end
+  end
+end

mongoid.gemspec

@@ -38,6 +38,7 @@ Gem::Specification.new do |s|
   # See: https://github.com/rails/rails/pull/43951
   s.add_dependency("activemodel", ['>=5.1', '<7.1', '!= 7.0.0'])
   s.add_dependency("mongo", ['>=2.18.0', '<3.0.0'])
+  s.add_dependency("concurrent-ruby", ['>= 1.1.10', '<1.2'])
 
   # The ruby2_keywords gem is recommended for handling argument delegation issues,
   # especially if support for 2.6 or prior is required.