Lazy concurrency (async)

README.md

@@ -386,6 +386,86 @@ class ChainingResolver < GraphQL::Breadth::FieldResolver
 end
 ```
 
+### Loader concurrency
+
+Lazy loaders may engage Ruby's fiber-based concurrency to asynchronously queue scheduler-compatible I/O, such as `async-http` requests (CPU-bound work and thread-blocking clients cannot be parallelized). Running the scheduler is not free, so loaders operate synchronously by default and must manually opt into async workflows when built to leverage them.
+
+#### Install
+
+Lazy concurrency requires the `async` gem as an opt-in dependency. Add `async` to your Gemfile and then enable it with an initializer:
+
+```ruby
+# Gemfile
+gem "async", "~> 2.0"
+
+# config/initializers/graphql_breadth.rb
+GraphQL::Breadth.enable_async!
+```
+
+#### Async loaders
+
+A LazyLoader class may opt into asynchronous execution by calling `async` with concurrency settings. All async loader classes will parallelize during common lazy execution cycles:
+
+```ruby
+class RemoteInventoryLoader < GraphQL::Breadth::LazyLoader
+  async resource: :inventory_api, limit: 4, timeout: 2
+
+  def perform(keys, context)
+    client = context[:inventory_client]
+
+    keys.each do |key|
+      fulfill_key(key, client.fetch_inventory(key))
+    end
+  end
+end
+```
+
+All concurrency settings are optional:
+
+* `resource: Symbol`, coordinates limits across related loaders hitting the same upstream resource. Defaults to a unique resource identity per loader class.
+* `limit: Integer`, number of concurrent operations allowed while hitting the resource. Defaults to 8.
+* `timeout: Integer`, number of seconds to wait on load operations before timing out. Defaults to no limit.
+
+#### Async fan-out
+
+Async LazyLoader classes may also fan-out their internal implementations using the `async` _instance_ method.
+
+```ruby
+class RemoteInventoryLoader < GraphQL::Breadth::LazyLoader
+  # class is async across lazy loaders...
+  async resource: :inventory_api, limit: 8, timeout: 2
+
+  def perform(keys, context)
+    client = context[:inventory_client]
+    futures_by_key = keys.each_with_object({}) do |key, memo|
+      # internals add async fan-out...
+      memo[key] = async { client.fetch_inventory(key) }
+    end
+
+    futures_by_key.each_pair do |key, future|
+      fulfill_key(key, future.wait)
+    end
+  end
+end
+```
+
+There's also an `async_map` variation available:
+
+```ruby
+class RemoteInventoryLoader < GraphQL::Breadth::LazyLoader
+  async resource: :inventory_api, limit: 8, timeout: 2
+
+  def map? = true
+
+  def perform_map(keys, context)
+    client = context[:inventory_client]
+    async_map(keys) { |key| client.fetch_inventory(key) }
+  end
+end
+```
+
+These fan-out instance methods share resource budgeting with their loader class by default. They can also specify their own `resource:`, `limit:`, and/or `timeout:` arguments to manage separate budgets from their loader class.
+
 ## Query planning
 
 The breadth executor operates on an [execution tree](#execution-taxonomy) in three phases:

graphql-breadth.gemspec

@@ -11,7 +11,7 @@ Gem::Specification.new do |spec|
   spec.homepage      = 'https://github.com/gmac/graphql-breadth'
   spec.license       = 'MIT'
 
-  spec.required_ruby_version = '>= 2.7.0'
+  spec.required_ruby_version = '>= 3.2.0'
 
   spec.metadata    = {
     'homepage_uri' => 'https://github.com/gmac/graphql-breadth',
@@ -31,6 +31,7 @@ Gem::Specification.new do |spec|
   spec.add_development_dependency 'bundler', '~> 2.0'
   spec.add_development_dependency 'rake', '~> 12.0'
   spec.add_development_dependency 'minitest', '~> 5.12'
+  spec.add_development_dependency 'async', '~> 2.0'
   spec.add_development_dependency 'benchmark-ips', '~> 2.0'
   spec.add_development_dependency 'memory_profiler'
   spec.add_development_dependency 'debug'

lib/graphql/breadth.rb

@@ -14,6 +14,26 @@ class << self
       def report_error(error)
         on_report_error&.call(error)
       end
+
+      #: -> bool
+      def async_enabled?
+        @async_enabled == true
+      end
+
+      #: -> void
+      def enable_async!
+        return if async_enabled?
+
+        require "async"
+        require "async/barrier"
+        require "async/queue"
+        require "async/semaphore"
+
+        @async_enabled = true
+      rescue LoadError => e
+        @async_enabled = false
+        raise ImplementationError, "Async lazy loaders require the `async` gem. Add `gem \"async\"` to your bundle. #{e.message}"
+      end
     end
 
     EMPTY_OBJECT = {}.freeze

lib/graphql/breadth/executor.rb

@@ -3,6 +3,7 @@
 
 require_relative "./executor/has_attributes"
 require_relative "./executor/lazy_element"
+require_relative "./executor/lazy_async"
 require_relative "./executor/execution_scope"
 require_relative "./executor/execution_field"
 require_relative "./executor/execution_directive"
@@ -23,6 +24,8 @@ class Executor
       UNDEFINED = Util::NilLike.new("undefined")
       EMPTY_TIMER = 0.0
 
+      include LazyAsync
+
       #: GraphQL::Query
       attr_reader :query
 
@@ -454,65 +457,90 @@ def authorize_scope_objects(exec_scope, invalidated_indices)
 
       #: (Array[LazyElement]) -> void
       def execute_lazy(lazy_elements)
-        aborted_status_cache = Hash.new { |h, exec_scope| h[exec_scope] = exec_scope.aborted_subtree? }.compare_by_identity
-        pending_loaders = (@loader_cache || EMPTY_OBJECT).each_value.reject { _1.promised.empty? }
+        pending_loaders = 0
+        sync_batches = nil #: Array[LazyLoader::Batch]?
+        async_batches = nil #: Array[LazyLoader::Batch]?
+
+        (@loader_cache || EMPTY_OBJECT).each_value do |loader|
+          next if loader.promised.empty?
+
+          pending_loaders += 1
+          batch = loader.to_batch
+
+          if batch.aborted?
+            loader.reset!
+          elsif batch.loader.async_settings.enabled?
+            (async_batches ||= []) << batch
+          else
+            (sync_batches ||= []) << batch
+          end
+        end
 
-        if pending_loaders.empty?
+        if pending_loaders.zero?
           raise ImplementationError, "Lazy #{lazy_elements.first} produced a promise without a loader"
         end
 
-        pending_loaders.each do |loader|
-          loader_elements = loader.promised.map(&:element)
-          all_aborted = loader_elements.all? do |element|
-            aborted_status_cache[element.is_a?(ExecutionField) ? element.scope : element]
+        sync_batches ||= EMPTY_ARRAY
+
+        if async_batches
+          execute_async_lazy_batches(sync_batches, async_batches)
+        else
+          sync_batches.each { execute_lazy_batch(_1) }
+          resume_lazy_elements(lazy_elements)
+        end
+      end
+
+      #: (LazyLoader::Batch, ?async_context: LazyAsync::LoaderContext?) -> void
+      def execute_lazy_batch(batch, async_context: nil)
+        loader = batch.loader
+        lazy_start_time = EMPTY_TIMER
+
+        begin
+          unless @tracers.empty?
+            lazy_start_time = timer
+            @tracers.each { _1.before_lazy_set(loader, batch.elements, @context) }
           end
 
-          if all_aborted
-            loader.reset!
-            next
+          loader.execute!(@context, async_context: async_context)
+        rescue StandardError => e
+          apply_lazy_error(batch, handle_or_reraise(e))
+        ensure
+          unless @tracers.empty?
+            duration = timer(lazy_start_time)
+            @tracers.each { _1.after_lazy_set(loader, batch.elements, @context, duration:) }
           end
+        end
+      end
 
-          lazy_start_time = EMPTY_TIMER
-          begin
-            unless @tracers.empty?
-              lazy_start_time = timer
-              @tracers.each { _1.before_lazy_set(loader, loader_elements, @context) }
-            end
-            loader.execute!(@context)
-          rescue StandardError => e
-            handled_error = handle_or_reraise(e)
-            loader_elements.each do |element|
-              case element
-              when ExecutionField
-                field_error = ExecutionError.from(handled_error, exec_field: element)
-                element.result = element.resolve_all(field_error)
-              when ExecutionScope
-                scope_error = ExecutionError.from(handled_error, exec_field: element.parent_field)
-                element.results.each { add_error(scope_error, _1, exec_field: element.parent_field) }
-                element.abort!
-              end
-            end
-          ensure
-            unless @tracers.empty?
-              duration = timer(lazy_start_time)
-              @tracers.each { _1.after_lazy_set(loader, loader_elements, @context, duration:) }
-            end
+      #: (LazyLoader::Batch, ExecutionError) -> void
+      def apply_lazy_error(batch, error)
+        batch.elements.each do |element|
+          case element
+          when ExecutionField
+            field_error = ExecutionError.from(error, exec_field: element)
+            element.result = element.resolve_all(field_error)
+          when ExecutionScope
+            scope_error = ExecutionError.from(error, exec_field: element.parent_field)
+            element.results.each { add_error(scope_error, _1, exec_field: element.parent_field) }
+            element.abort!
           end
         end
+      end
 
-        aborted_status_cache.clear
-        lazy_elements.each do |element|
+      #: (Array[LazyElement]) -> void
+      def resume_lazy_elements(elements)
+        elements.each do |element|
           case element
           when ExecutionField
-            next if aborted_status_cache[element.scope]
+            next if element.scope.aborted_subtree?
 
             if element.has_result?
               resume_lazy_field_result(element)
             else
               resume_lazy_field_execute(element)
             end
           when ExecutionScope
-            next if aborted_status_cache[element]
+            next if element.aborted_subtree?
 
             resume_lazy_scope_execute(element)
           end
@@ -607,8 +635,6 @@ def execute_field(exec_field)
           return
         end
 
-        parent_objects = exec_field.scope.objects
-
         resolve_start_time = EMPTY_TIMER
         begin
           unless @tracers.empty?
@@ -633,7 +659,7 @@ def execute_field(exec_field)
         rescue StandardError => e
           error = handle_or_reraise(e, exec_field: exec_field)
           @errors << error if error.base_error?
-          exec_field.result = Array.new(parent_objects.length, error)
+          exec_field.result = Array.new(exec_field.objects.length, error)
         ensure
           unless @tracers.empty?
             duration = timer(resolve_start_time)
@@ -772,8 +798,7 @@ def build_leaf_result(exec_field, current_type, val)
           val.map { build_leaf_result(exec_field, current_type.of_type, _1) }
         else
           begin
-            unwrapped_type = current_type.unwrap
-            coerced_val = unwrapped_type.coerce_result(val, @context)
+            coerced_val = current_type.unwrap.coerce_result(val, @context)
             return coerced_val unless coerced_val.nil?
 
             build_missing_value(exec_field, current_type, nil)
@@ -832,7 +857,7 @@ def propagate_null!(exec_field)
         current_exec_field = exec_field #: ExecutionField[untyped]?
         propagating = !!exec_field.propagates_null?
         highest_nulled_depth = exec_field.scope.depth
-        highest_list_depth = Integer(-1)
+        highest_list_depth = -1
 
         while current_exec_field
           if current_exec_field.propagates_null? && propagating
@@ -842,7 +867,7 @@ def propagate_null!(exec_field)
           end
 
           if current_exec_field.type.list?
-            highest_list_depth = Integer(current_exec_field.scope.depth)
+            highest_list_depth = current_exec_field.scope.depth
           end
 
           current_exec_field = current_exec_field.scope.parent_field

lib/graphql/breadth/executor/execution_scope.rb

@@ -158,8 +158,7 @@ def aborted_subtree?
           exec_field = parent_field #: ExecutionField[untyped]?
           while exec_field
             if exec_field.scope.aborted?
-              abort!
-              return true
+              return abort!
             end
             exec_field = exec_field.scope.parent_field
           end