Lazy Concurrency Per Evaluation Layer

This change adds support for the concurrent resolution of lazy
objects. While it is currently possible to return something like
a `::Concurrent::Promise.execute` from a GraphQL method, there is
currently no way to make this work in tandem with lazy objects. Consider
the case in which we would like to use a Gem like `graphql-batch` (or
a thread-safe alternative) but execute queries in parallel whenever
possible:

```
{
  post(id: 10) {
    author { name }
    comments { count }
  }
}
```

In this query you could imagine that the each of `post`, `author`, and
`comments` are separate DB calls. While it may not be possible for
`post` and `author` to be executed in parallel, certainly `author` and
`comments` can. But we would still like to perform this operation lazily
because if this query is expanded:

```
{
  a: post(id: 10) {
    author { name }
    comments { count }
  }
  b: post(id: 11) {
    author { name }
    comments { count }
  }
}
```

We would like to be able to load the authors for post 10 and 11 in a
single query.

I have implemented this solution by allowing the `lazy_resolve`
directive to accept an additional method name (the concurrent execution
method). This method will be called on all layers (breadth first) before
the `value` method is called. This ensures that concurrent execution can
be delayed until the last possible moment (to enable batching) but it
also ensures that multiple batches can be run in parallel if they are
resolved in the same graph layer.

Although intended for concurrent execution, it is not necessary for
this new method to actually perform an operation concurrently (i.e it
does not need to return a Thread or anything like that). This allows
`graphql-ruby` to not enforce any specific parallel execution primitive
(threads or `concurrent-ruby` could be used interchangeably).

I know this is a large PR, so I am happy to split it up into multiple
PRs if the overall approach is agreeable.

Author: panthomakos

lib/graphql/execution/execute.rb

@@ -115,20 +115,29 @@ def resolve_field(object, field_ctx)
           end
 
           if field_ctx.schema.lazy?(raw_value)
-            field_ctx.value = Execution::Lazy.new {
-              inner_value = field_ctx.trace("execute_field_lazy", {context: field_ctx}) {
-                begin
+            field_ctx.value = Execution::Lazy.new(
+              value: -> {
+                inner_value = field_ctx.trace("execute_field_lazy", {context: field_ctx}) {
                   begin
-                    field_ctx.field.lazy_resolve(raw_value, arguments, field_ctx)
-                  rescue GraphQL::UnauthorizedError => err
-                    field_ctx.schema.unauthorized_object(err)
+                    begin
+                      field_ctx.field.lazy_resolve(raw_value, arguments, field_ctx)
+                    rescue GraphQL::UnauthorizedError => err
+                      field_ctx.schema.unauthorized_object(err)
+                    end
+                  rescue GraphQL::ExecutionError => err
+                    err
                   end
-                rescue GraphQL::ExecutionError => err
-                  err
+                }
+                continue_or_wait(inner_value, field_ctx.type, field_ctx)
+              },
+              exec: -> {
+                if field_ctx.schema.concurrent?(raw_value)
+                  field_ctx.trace("execute_field_concurrent", {context: field_ctx}) {
+                    field_ctx.field.concurrent_exec(raw_value, arguments, field_ctx)
+                  }
                 end
               }
-              continue_or_wait(inner_value, field_ctx.type, field_ctx)
-            }
+            )
           else
             continue_or_wait(raw_value, field_ctx.type, field_ctx)
           end
@@ -143,8 +152,10 @@ def resolve_field(object, field_ctx)
         # and resolve child fields
         def continue_or_wait(raw_value, field_type, field_ctx)
           if field_ctx.schema.lazy?(raw_value)
-            field_ctx.value = Execution::Lazy.new {
-              inner_value = begin
+            field_ctx.value = Execution::Lazy.new(
+              value: -> {
+                inner_value =
+                begin
                   begin
                     field_ctx.schema.sync_lazy(raw_value)
                   rescue GraphQL::UnauthorizedError => err
@@ -154,8 +165,14 @@ def continue_or_wait(raw_value, field_type, field_ctx)
                   err
                 end
 
-              field_ctx.value = continue_or_wait(inner_value, field_type, field_ctx)
-            }
+                field_ctx.value = continue_or_wait(inner_value, field_type, field_ctx)
+              },
+              exec: -> {
+                if field_ctx.schema.concurrent?(raw_value)
+                  field_ctx.schema.exec_concurrent(raw_value)
+                end
+              }
+            )
           else
             field_ctx.value = continue_resolve_field(raw_value, field_type, field_ctx)
           end

lib/graphql/execution/lazy.rb

@@ -20,18 +20,39 @@ def self.resolve(val)
       end
 
       # Create a {Lazy} which will get its inner value by calling the block
-      # @param get_value_func [Proc] a block to get the inner value (later)
-      def initialize(&get_value_func)
-        @get_value_func = get_value_func
+      # @param value_proc [Proc] a block to get the inner value (later)
+      def initialize(original = nil, value:, exec:)
+        @original = original
+        @value_proc = value
+        @exec_proc = exec
         @resolved = false
       end
 
+      def execute
+        return if @resolved
+
+        exec =
+          begin
+            e = @exec_proc.call
+            if e.is_a?(Lazy)
+              e = e.execute
+            end
+            e
+          rescue GraphQL::ExecutionError => err
+            err
+          end
+
+        if exec.is_a?(StandardError)
+          raise exec
+        end
+      end
+
       # @return [Object] The wrapped value, calling the lazy block if necessary
       def value
         if !@resolved
           @resolved = true
           @value = begin
-            v = @get_value_func.call
+            v = @value_proc.call
             if v.is_a?(Lazy)
               v = v.value
             end
@@ -50,22 +71,24 @@ def value
 
       # @return [Lazy] A {Lazy} whose value depends on another {Lazy}, plus any transformations in `block`
       def then
-        self.class.new {
-          yield(value)
-        }
+        self.class.new(
+          value: -> { yield(value) },
+          exec: -> { execute }
+        )
       end
 
       # @param lazies [Array<Object>] Maybe-lazy objects
       # @return [Lazy] A lazy which will sync all of `lazies`
       def self.all(lazies)
-        self.new {
-          lazies.map { |l| l.is_a?(Lazy) ? l.value : l }
-        }
+        self.new(
+          value: -> { lazies.map { |l| l.is_a?(Lazy) ? l.value : l } },
+          exec: -> { lazies.each { |l| l.is_a?(Lazy) ? l.execute : l } }
+        )
       end
 
       # This can be used for fields which _had no_ lazy results
       # @api private
-      NullResult = Lazy.new(){}
+      NullResult = Lazy.new(value: -> {}, exec: -> {})
       NullResult.value
     end
   end

lib/graphql/execution/lazy/lazy_method_map.rb

@@ -24,10 +24,13 @@ def initialize_copy(other)
           @storage = other.storage.dup
         end
 
+        LazySpec = Struct.new(:value_method, :exec_method)
+        private_constant :LazySpec
+
         # @param lazy_class [Class] A class which represents a lazy value (subclasses may also be used)
         # @param lazy_value_method [Symbol] The method to call on this class to get its value
-        def set(lazy_class, lazy_value_method)
-          @storage[lazy_class] = lazy_value_method
+        def set(lazy_class, lazy_value_method, concurrent_exec_method)
+          @storage[lazy_class] = LazySpec.new(lazy_value_method, concurrent_exec_method)
         end
 
         # @param value [Object] an object which may have a `lazy_value_method` registered for its class or superclasses

lib/graphql/execution/lazy/resolve.rb

@@ -35,12 +35,17 @@ def self.resolve_in_place(value)
           if acc.empty?
             Lazy::NullResult
           else
-            Lazy.new {
-              acc.each_with_index { |ctx, idx|
-                acc[idx] = ctx.value.value
-              }
-              resolve_in_place(acc)
-            }
+            Lazy.new(
+              value: -> {
+                acc.each { |ctx| ctx.value.execute }
+
+                acc.each_with_index { |ctx, idx|
+                  acc[idx] = ctx.value.value
+                }
+                resolve_in_place(acc)
+              },
+              exec: -> {}
+            )
           end
         end
 

lib/graphql/field.rb

@@ -123,6 +123,7 @@ class Field
     include GraphQL::Define::InstanceDefinable
     accepts_definitions :name, :description, :deprecation_reason,
       :resolve, :lazy_resolve,
+      :concurrent_exec,
       :type, :arguments,
       :property, :hash_key, :complexity,
       :mutation, :function,
@@ -138,6 +139,7 @@ class Field
       :name, :deprecation_reason, :description, :description=, :property, :hash_key,
       :mutation, :arguments, :complexity, :function,
       :resolve, :resolve=, :lazy_resolve, :lazy_resolve=, :lazy_resolve_proc, :resolve_proc,
+      :concurrent_exec, :concurrent_exec=, :concurrent_exec_proc,
       :type, :type=, :name=, :property=, :hash_key=,
       :relay_node_field, :relay_nodes_field, :edges?, :edge_class, :subscription_scope,
       :introspection?
@@ -155,6 +157,9 @@ class Field
     # @return [<#call(obj, args, ctx)>] A proc-like object which can be called trigger a lazy resolution
     attr_reader :lazy_resolve_proc
 
+    # @return [<#call(obj, args, ctx)>] A proc-like object which can be called trigger a concurrent execution
+    attr_reader :concurrent_exec_proc
+
     # @return [String] The name of this field on its {GraphQL::ObjectType} (or {GraphQL::InterfaceType})
     attr_reader :name
     alias :graphql_name :name
@@ -218,6 +223,7 @@ def initialize
       @arguments = {}
       @resolve_proc = build_default_resolver
       @lazy_resolve_proc = DefaultLazyResolve
+      @concurrent_exec_proc = DefaultConcurrentExec
       @relay_node_field = false
       @connection = false
       @connection_max_page_size = nil
@@ -307,12 +313,21 @@ def lazy_resolve=(new_lazy_resolve_proc)
       @lazy_resolve_proc = new_lazy_resolve_proc
     end
 
+    def concurrent_exec(obj, args, ctx)
+      @concurrent_exec_proc.call(obj, args, ctx)
+    end
+
+    def concurrent_exec=(new_concurrent_exec_proc)
+      @concurrent_exec_proc = new_concurrent_exec_proc
+    end
+
     # Prepare a lazy value for this field. It may be `then`-ed and resolved later.
     # @return [GraphQL::Execution::Lazy] A lazy wrapper around `obj` and its registered method name
     def prepare_lazy(obj, args, ctx)
-      GraphQL::Execution::Lazy.new {
-        lazy_resolve(obj, args, ctx)
-      }
+      GraphQL::Execution::Lazy.new(
+        value: -> { lazy_resolve(obj, args, ctx) },
+        exec: -> { concurrent_exec(obj, args, ctx) }
+      )
     end
 
     private
@@ -326,5 +341,11 @@ def self.call(obj, args, ctx)
         ctx.schema.sync_lazy(obj)
       end
     end
+
+    module DefaultConcurrentExec
+      def self.call(obj, args, ctx)
+        ctx.schema.exec_concurrent(obj)
+      end
+    end
   end
 end

lib/graphql/schema.rb

@@ -92,7 +92,9 @@ class Schema
       query_analyzer: ->(schema, analyzer) { schema.query_analyzers << analyzer },
       multiplex_analyzer: ->(schema, analyzer) { schema.multiplex_analyzers << analyzer },
       middleware: ->(schema, middleware) { schema.middleware << middleware },
-      lazy_resolve: ->(schema, lazy_class, lazy_value_method) { schema.lazy_methods.set(lazy_class, lazy_value_method) },
+      lazy_resolve: ->(schema, lazy_class, lazy_value_method, concurrent_exec_method = nil) {
+        schema.lazy_methods.set(lazy_class, lazy_value_method, concurrent_exec_method)
+      },
       rescue_from: ->(schema, err_class, &block) { schema.rescue_from(err_class, &block)},
       tracer: ->(schema, tracer) { schema.tracers.push(tracer) }
 
@@ -157,7 +159,7 @@ def initialize
       @parse_error_proc = DefaultParseError
       @instrumenters = Hash.new { |h, k| h[k] = [] }
       @lazy_methods = GraphQL::Execution::Lazy::LazyMethodMap.new
-      @lazy_methods.set(GraphQL::Execution::Lazy, :value)
+      @lazy_methods.set(GraphQL::Execution::Lazy, :value, :execute)
       @cursor_encoder = Base64Encoder
       # Default to the built-in execution strategy:
       @query_execution_strategy = self.class.default_execution_strategy
@@ -604,14 +606,26 @@ class InvalidDocumentError < Error; end;
 
     # @return [Symbol, nil] The method name to lazily resolve `obj`, or nil if `obj`'s class wasn't registered wtih {#lazy_resolve}.
     def lazy_method_name(obj)
-      @lazy_methods.get(obj)
+      spec = @lazy_methods.get(obj)
+      spec && spec.value_method
+    end
+
+    # @return [Symbol, nil] The method name to concurrently resolve `obj`, or nil if `obj`'s class wasn't registered wtih {#lazy_resolve} with a concurrent method.
+    def concurrent_method_name(obj)
+      spec = @lazy_methods.get(obj)
+      spec && spec.exec_method
     end
 
     # @return [Boolean] True if this object should be lazily resolved
     def lazy?(obj)
       !!lazy_method_name(obj)
     end
 
+    # @return [Boolean] True if this object should be concurrently executed
+    def concurrent?(obj)
+      !!concurrent_method_name(obj)
+    end
+
     # Return the GraphQL IDL for the schema
     # @param context [Hash]
     # @param only [<#call(member, ctx)>]
@@ -656,7 +670,8 @@ class << self
         :execute, :multiplex,
         :static_validator, :introspection_system,
         :query_analyzers, :tracers, :instrumenters,
-        :validate, :multiplex_analyzers, :lazy?, :lazy_method_name, :after_lazy, :sync_lazy,
+        :validate, :multiplex_analyzers,
+        :lazy?, :lazy_method_name, :concurrent_method_name, :after_lazy, :sync_lazy,
         # Configuration
         :max_complexity=, :max_depth=,
         :metadata,
@@ -717,8 +732,8 @@ def to_graphql
           end
         end
         schema_defn.instrumenters[:query] << GraphQL::Schema::Member::Instrumentation
-        lazy_classes.each do |lazy_class, value_method|
-          schema_defn.lazy_methods.set(lazy_class, value_method)
+        lazy_classes.each do |lazy_class, (value_method, exec_method)|
+          schema_defn.lazy_methods.set(lazy_class, value_method, exec_method)
         end
         if @rescues
           @rescues.each do |err_class, handler|
@@ -915,8 +930,8 @@ def type_error(type_err, ctx)
         DefaultTypeError.call(type_err, ctx)
       end
 
-      def lazy_resolve(lazy_class, value_method)
-        lazy_classes[lazy_class] = value_method
+      def lazy_resolve(lazy_class, value_method, exec_method = nil)
+        lazy_classes[lazy_class] = [value_method, exec_method]
       end
 
       def instrument(instrument_step, instrumenter, options = {})
@@ -1028,13 +1043,16 @@ def resolve_type(type, obj, ctx = :__undefined__)
     # @api private
     def after_lazy(value)
       if lazy?(value)
-        GraphQL::Execution::Lazy.new do
-          result = sync_lazy(value)
-          # The returned result might also be lazy, so check it, too
-          after_lazy(result) do |final_result|
-            yield(final_result) if block_given?
-          end
-        end
+        GraphQL::Execution::Lazy.new(
+          value: -> {
+            result = sync_lazy(value)
+            # The returned result might also be lazy, so check it, too
+            after_lazy(result) do |final_result|
+              yield(final_result) if block_given?
+            end
+          },
+          exec: -> { exec_concurrent(value) }
+        )
       else
         yield(value) if block_given?
       end
@@ -1062,6 +1080,24 @@ def sync_lazy(value)
       }
     end
 
+    # Override this method to handle lazy concurrent objects in a custom way.
+    # @param value [Object] an instance of a class registered with {.lazy_resolve}
+    # @param ctx [GraphQL::Query::Context] the context for this query
+    # @return [Object] A GraphQL-ready (non-lazy) object
+    def self.exec_concurrent(value)
+      yield(value)
+    end
+
+    # @see Schema.exec_concurrent for a hook to override
+    # @api private
+    def exec_concurrent(value)
+      self.class.exec_concurrent(value) do |v|
+        if concurrent_method = concurrent_method_name(v)
+          value.public_send(concurrent_method)
+        end
+      end
+    end
+
     protected
 
     def rescues?