diff --git a/README.md b/README.md index 23bb156..93cfc43 100644 --- a/README.md +++ b/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: diff --git a/graphql-breadth.gemspec b/graphql-breadth.gemspec index 4ab55e9..3427479 100644 --- a/graphql-breadth.gemspec +++ b/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' diff --git a/lib/graphql/breadth.rb b/lib/graphql/breadth.rb index 8fae326..f16b0d2 100644 --- a/lib/graphql/breadth.rb +++ b/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 diff --git a/lib/graphql/breadth/executor.rb b/lib/graphql/breadth/executor.rb index 801529f..679c900 100644 --- a/lib/graphql/breadth/executor.rb +++ b/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,57 +457,82 @@ 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) @@ -512,7 +540,7 @@ def execute_lazy(lazy_elements) 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 diff --git a/lib/graphql/breadth/executor/execution_scope.rb b/lib/graphql/breadth/executor/execution_scope.rb index aaa88f3..5726cbc 100644 --- a/lib/graphql/breadth/executor/execution_scope.rb +++ b/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 diff --git a/lib/graphql/breadth/executor/lazy_async.rb b/lib/graphql/breadth/executor/lazy_async.rb new file mode 100644 index 0000000..7a9eefd --- /dev/null +++ b/lib/graphql/breadth/executor/lazy_async.rb @@ -0,0 +1,310 @@ +# typed: true +# frozen_string_literal: true + +module GraphQL + module Breadth + class Executor + # @requires_ancestor: Executor + module LazyAsync + class ExecutorContext + #: Executor + attr_reader :executor + + #: ::Async::Barrier + attr_reader :barrier + + #: ::Async::Queue + attr_reader :completed_batches + + #: Set[LazyLoader[untyped]] + attr_reader :active_loaders + + #: Set[LazyElement] + attr_reader :waiting_elements + + #: Hash[Symbol, ::Async::Semaphore] + attr_reader :semaphores_by_resource + + #: (Executor, ::Async::Task) -> void + def initialize(executor, task) + @executor = executor + @barrier = ::Async::Barrier.new(parent: task) + @completed_batches = ::Async::Queue.new + @active_loaders = Set.new.compare_by_identity + @waiting_elements = Set.new.compare_by_identity + @semaphores_by_resource = {} + end + + #: (Symbol, Integer) -> ::Async::Semaphore + def semaphore_for(resource, limit) + semaphore = @semaphores_by_resource[resource] + if semaphore + if semaphore.limit != limit + Kernel.raise ArgumentError, "Conflicting lazy async limits for resource #{resource.inspect}: #{semaphore.limit} and #{limit}" + end + + return semaphore + end + + @semaphores_by_resource[resource] = ::Async::Semaphore.new(limit, parent: @barrier) + end + end + + class LoaderContext + #: (LazyLoader[untyped], ExecutorContext, ::Async::Task) -> void + def initialize(loader, async_context, parent_task) + @loader = loader + @async_context = async_context + @barrier = ::Async::Barrier.new(parent: parent_task) + @futures = [] + @parent_semaphore = nil + end + + #: (?resource: Symbol?, ?limit: Integer?, ?timeout: Numeric?) ?{ -> untyped } -> Future + def async(resource: nil, limit: nil, timeout: nil, &block) + Kernel.raise ArgumentError, "LazyLoader#async requires a block" unless block + + settings = @loader.async_settings + resource = settings.resource if resource.nil? + limit = settings.limit if limit.nil? + timeout = settings.timeout if timeout.nil? + + Kernel.raise ArgumentError, "Lazy async limit must be positive" unless limit.nil? || limit > 0 + Kernel.raise ArgumentError, "Lazy async timeout must be positive" unless timeout.nil? || timeout > 0 + + # child fan-out shares the parent resource budget, so release the parent slot before waiting on child slots. + release_parent_permit if @parent_semaphore && @loader.async_settings.resource == resource + + parent = if limit + semaphore_resource = resource #: as !nil + @async_context.semaphore_for(semaphore_resource, limit) + else + @barrier + end + + task = parent.async(parent: @barrier) do |async_task| + begin + if timeout + async_task.with_timeout(timeout, ::Async::TimeoutError, &block) + else + block.call + end + rescue StandardError => e + @async_context.executor.handle_or_reraise(e) + end + end + + @futures << Future.new(task) + @futures.last + end + + #: [T, U] (Enumerable[T], ?resource: Symbol?, ?limit: Integer?, ?timeout: Numeric?) ?{ (T) -> U } -> Array[U] + def async_map(collection, resource: nil, limit: nil, timeout: nil, &block) + Kernel.raise ArgumentError, "LazyLoader#async_map requires a block" unless block + + completed = false + futures = collection.map do |item| + async(resource: resource, limit: limit, timeout: timeout) { block.call(item) } + end + + results = futures.map(&:wait) + completed = true + results + ensure + stop unless completed + end + + #: () { -> untyped } -> untyped + def run(&block) + acquire_parent_permit + value = yield + wait + value + ensure + stop + release_parent_permit + end + + private + + #: -> void + def wait + @futures.each do |future| + future.wait unless future.observed? + end + + @barrier.wait + end + + #: -> void + def stop + @barrier.stop unless @barrier.empty? + end + + #: -> void + def acquire_parent_permit + settings = @loader.async_settings + return unless settings.limit + + @parent_semaphore = @async_context.semaphore_for(settings.resource, settings.limit) + @parent_semaphore.acquire + end + + #: -> void + def release_parent_permit + semaphore = @parent_semaphore + return unless semaphore + + @parent_semaphore = nil + semaphore.release + end + end + + class Future + #: (::Async::Task) -> void + def initialize(task) + @task = task + @observed = false + @resolved = false + @value = nil + end + + #: -> bool + def observed? + @observed + end + + #: -> untyped + def wait + @observed = true + return @value if @resolved + + @value = @task.wait + @resolved = true + @value + end + end + + private + + #: (Array[LazyLoader::Batch], Array[LazyLoader::Batch]) -> void + def execute_async_lazy_batches(sync_batches, async_batches) + Kernel.Sync do |task| + executor = self #: as Executor + async_context = ExecutorContext.new(executor, task) + + begin + async_batches.each { schedule_async_lazy_batch(async_context, _1) } + sync_batches.each { execute_sync_lazy_batch(async_context, _1) } + + until async_context.active_loaders.empty? + batch = async_context.completed_batches.dequeue + async_context.active_loaders.delete(batch.loader) + + resume_lazy_elements_and_drain_requeued(async_context, batch.elements) + retry_waiting_lazy_elements(async_context) + end + + # terminal sweep... + retry_waiting_lazy_elements(async_context) + ensure + async_context.barrier.stop unless async_context.active_loaders.empty? + end + end + end + + #: (ExecutorContext, LazyLoader::Batch) -> void + def schedule_async_lazy_batch(async_context, batch) + async_context.active_loaders.add(batch.loader) + + async_context.barrier.async do |async_task| + settings = batch.loader.async_settings + loader_context = LoaderContext.new(batch.loader, async_context, async_task) + + begin + if settings.timeout + async_task.with_timeout(settings.timeout) do + loader_context.run { execute_lazy_batch(batch, async_context: loader_context) } + end + else + loader_context.run { execute_lazy_batch(batch, async_context: loader_context) } + end + rescue StandardError => e + apply_lazy_error(batch, handle_or_reraise(e)) + end + + async_context.completed_batches << batch + end + end + + #: (ExecutorContext, LazyLoader::Batch) -> void + def execute_sync_lazy_batch(async_context, batch) + execute_lazy_batch(batch) + resume_lazy_elements_and_drain_requeued(async_context, batch.elements) + end + + #: (ExecutorContext, Array[LazyElement]) -> void + def resume_lazy_elements_and_drain_requeued(async_context, elements) + queue_start = @lazy_queue.length + resume_lazy_elements(elements) + + return unless @lazy_queue.length > queue_start + + requeued_elements = @lazy_queue.slice!(queue_start, @lazy_queue.length - queue_start) #: as !nil + scheduled_elements = schedule_pending_lazy_batches(async_context) + + if scheduled_elements.empty? && async_context.active_loaders.empty? + raise ImplementationError, "Lazy #{requeued_elements.first} produced a promise without a loader" + end + + requeued_elements.each do |element| + next if scheduled_elements.include?(element) + + async_context.waiting_elements.add(element) + end + end + + #: (ExecutorContext) -> Set[LazyElement] + def schedule_pending_lazy_batches(async_context) + sync_pending_batches = nil #: Array[LazyLoader::Batch]? + async_pending_batches = nil #: Array[LazyLoader::Batch]? + scheduled_elements = nil #: Set[LazyElement]? + + (@loader_cache || EMPTY_OBJECT).each_value do |loader| + next if loader.promised.empty? || async_context.active_loaders.include?(loader) + + batch = loader.to_batch + if batch.aborted? + loader.reset! + else + (scheduled_elements ||= Set.new.compare_by_identity).merge(batch.elements) + + if batch.loader.async_settings.enabled? + (async_pending_batches ||= []) << batch + else + (sync_pending_batches ||= []) << batch + end + end + end + + return EMPTY_SET if scheduled_elements.nil? + + async_pending_batches&.each { schedule_async_lazy_batch(async_context, _1) } + sync_pending_batches&.each { execute_sync_lazy_batch(async_context, _1) } + + scheduled_elements + end + + #: (ExecutorContext) -> void + def retry_waiting_lazy_elements(async_context) + return if async_context.waiting_elements.empty? + + elements = async_context.waiting_elements.to_a + async_context.waiting_elements.clear + + resume_lazy_elements_and_drain_requeued(async_context, elements) + end + end + end + end +end diff --git a/lib/graphql/breadth/lazy_loader.rb b/lib/graphql/breadth/lazy_loader.rb index 6ca4378..b00a66f 100644 --- a/lib/graphql/breadth/lazy_loader.rb +++ b/lib/graphql/breadth/lazy_loader.rb @@ -5,6 +5,26 @@ module GraphQL module Breadth #: [ContextType < GraphQL::Query::Context] class LazyLoader + AsyncSettings = Data.define(:enabled, :limit, :resource, :timeout) do + alias_method :enabled?, :enabled + end + + # Snapshots a loader's promised elements before execution resets loader state. + Batch = Data.define(:loader, :elements) do + def aborted? + elements.all? do |element| + case element + when Executor::ExecutionField + element.scope.aborted_subtree? + when Executor::ExecutionScope + element.aborted_subtree? + else + raise ImplementationError, "Invalid lazy element: #{element.inspect}" + end + end + end + end + class LazyFulfillment #: Executor::LazyElement attr_reader :element @@ -45,6 +65,37 @@ def resolve(results) KEY_OMISSION = Object.new.freeze + DEFAULT_ASYNC_SETTINGS = AsyncSettings.new( + enabled: false, + limit: nil, + resource: nil, + timeout: nil, + ).freeze + + class << self + #: (?limit: Integer?, ?resource: Symbol?, ?timeout: Numeric?) -> void + def async(limit: 8, resource: nil, timeout: nil) + unless GraphQL::Breadth.async_enabled? + Kernel.raise ImplementationError, "Async lazy loaders require `GraphQL::Breadth.enable_async!` during initialization." + end + + raise ArgumentError, "Lazy async limit must be positive" unless limit.nil? || limit > 0 + raise ArgumentError, "Lazy async timeout must be positive" unless timeout.nil? || timeout > 0 + + @async_settings = AsyncSettings.new( + enabled: true, + limit: limit, + resource: resource || name&.to_sym || :"lazy_loader_#{object_id}", + timeout: timeout, + ).freeze + end + + #: -> AsyncSettings + def async_settings + @async_settings || DEFAULT_ASYNC_SETTINGS + end + end + #: Hash[untyped, untyped] attr_reader :pending_keys_by_identity @@ -59,6 +110,12 @@ def initialize @pending_keys_by_identity = {} @results_by_identity = {} @promised = [] + @async_context = nil + end + + #: -> Batch + def to_batch + Batch.new(loader: self, elements: promised.map(&:element)) end #: -> bool @@ -71,6 +128,29 @@ def resolve_one? false end + #: -> AsyncSettings + def async_settings + self.class.async_settings + end + + #: (?resource: Symbol?, ?limit: Integer?, ?timeout: Numeric?) ?{ -> untyped } -> Executor::LazyAsync::Future + def async(resource: nil, limit: nil, timeout: nil, &block) + unless @async_context + raise ImplementationError, "LazyLoader#async requires the loader class to opt into async features via `async(...)`." + end + + @async_context.async(resource: resource, limit: limit, timeout: timeout, &block) + end + + #: [T, U] (Enumerable[T], ?resource: Symbol?, ?limit: Integer?, ?timeout: Numeric?) ?{ (T) -> U } -> Array[U] + def async_map(collection, resource: nil, limit: nil, timeout: nil, &block) + unless @async_context + raise ImplementationError, "LazyLoader#async_map requires the loader class to opt into async features via `async(...)`." + end + + @async_context.async_map(collection, resource: resource, limit: limit, timeout: timeout, &block) + end + #: (Array[untyped], ContextType) -> void def perform(_keys, _context) raise NotImplementedError, "LazyLoader#perform must be implemented." @@ -147,8 +227,10 @@ def collect_results(fulfillment) end end - #: (ContextType) -> void - def execute!(context) + #: (ContextType, ?async_context: Executor::LazyAsync::LoaderContext?) -> void + def execute!(context, async_context: nil) + previous_async_context = @async_context + @async_context = async_context fulfillments = @promised unless @pending_keys_by_identity.empty? pending_loader_keys = @pending_keys_by_identity.values @@ -176,6 +258,8 @@ def execute!(context) end fulfillments.each { |fulfillment| fulfillment.resolve(collect_results(fulfillment)) } + ensure + @async_context = previous_async_context end #: -> void diff --git a/sorbet/config b/sorbet/config index e2628a7..c15176e 100644 --- a/sorbet/config +++ b/sorbet/config @@ -2,6 +2,7 @@ lib --parser=prism --enable-experimental-rbs-comments +--enable-experimental-requires-ancestor --dir sorbet/rbi --ignore=/test/ diff --git a/sorbet/rbi/shims/async.rbi b/sorbet/rbi/shims/async.rbi new file mode 100644 index 0000000..9dd0ac3 --- /dev/null +++ b/sorbet/rbi/shims/async.rbi @@ -0,0 +1,36 @@ +# typed: true + +module Async + class TimeoutError < StandardError; end + + class Barrier + def initialize(parent: T.unsafe(nil)); end + def async(*arguments, parent: T.unsafe(nil), **options, &block); end + def empty?; end + def wait; end + def stop; end + end + + class Queue + def initialize(parent: T.unsafe(nil)); end + def <<(item); end + def dequeue; end + end + + class Semaphore + def initialize(limit = T.unsafe(nil), parent: T.unsafe(nil)); end + def async(*arguments, parent: T.unsafe(nil), **options, &block); end + def acquire; end + def limit; end + def release; end + end + + class Task + def wait; end + def with_timeout(duration, exception = T.unsafe(nil), message = T.unsafe(nil), &block); end + end +end + +module Kernel + def Sync(annotation: T.unsafe(nil), &block); end +end diff --git a/test/graphql/breadth/executor/loaders_test.rb b/test/graphql/breadth/executor/loaders_test.rb index 71c3219..5e43d37 100644 --- a/test/graphql/breadth/executor/loaders_test.rb +++ b/test/graphql/breadth/executor/loaders_test.rb @@ -26,6 +26,206 @@ def perform_map(keys, _ctx) end end + class ConcurrentLoader < GraphQL::Breadth::LazyLoader + class << self + attr_accessor :active, :max_active, :events + + def reset_tracking! + self.active = 0 + self.max_active = 0 + self.events = [] + end + end + + async limit: 2 + reset_tracking! + + def initialize(group:) + super() + @group = group + end + + def map? + true + end + + def perform_map(keys, _ctx) + self.class.events << [:start, @group, keys.dup] + self.class.active += 1 + self.class.max_active = [self.class.max_active, self.class.active].max + sleep 0.01 + keys.map { |key| "#{key}-#{@group}" } + ensure + self.class.events << [:finish, @group] + self.class.active -= 1 + end + end + + class LimitedConcurrentLoader < ConcurrentLoader + async limit: 1, resource: :limited_concurrent_loader + reset_tracking! + end + + class BoomConcurrentLoader < GraphQL::Breadth::LazyLoader + async limit: 2 + + def map? + true + end + + def perform_map(_keys, _ctx) + sleep 0 + raise GraphQL::ExecutionError, "async loader boom" + end + end + + class FatalConcurrentLoader < GraphQL::Breadth::LazyLoader + FatalError = Class.new(Exception) + + async limit: 2 + + def map? + true + end + + def perform_map(_keys, _ctx) + raise FatalError, "fatal async loader boom" + end + end + + class ChainedConcurrentLoader < GraphQL::Breadth::LazyLoader + class << self + attr_accessor :events + + def reset_tracking! + self.events = [] + end + end + + async limit: 3 + reset_tracking! + + def initialize(group:, delay:) + super() + @group = group + @delay = delay + end + + def map? + true + end + + def perform_map(keys, _ctx) + self.class.events << [:start, @group] + sleep @delay + keys.map { |key| "#{key}-#{@group}" } + ensure + self.class.events << [:finish, @group] + end + end + + class FanoutMapLoader < GraphQL::Breadth::LazyLoader + class << self + attr_accessor :active, :max_active + + def reset_tracking! + self.active = 0 + self.max_active = 0 + end + end + + async limit: 2, resource: :fanout_map_loader + reset_tracking! + + def map? + true + end + + def perform_map(keys, _ctx) + async_map(keys) do |key| + self.class.active += 1 + self.class.max_active = [self.class.max_active, self.class.active].max + sleep 0.01 + "#{key}-fanout" + ensure + self.class.active -= 1 + end + end + end + + class BoomFanoutMapLoader < GraphQL::Breadth::LazyLoader + async limit: 2, resource: :boom_fanout_map_loader + + def map? + true + end + + def perform_map(keys, _ctx) + async_map(keys) do |key| + sleep 0 + raise GraphQL::ExecutionError, "async map boom" if key == "Apple" + + "#{key}-fanout" + end + end + end + + class SharedResourceTracker + class << self + attr_accessor :active, :max_active + + def reset_tracking! + self.active = 0 + self.max_active = 0 + end + + def enter + self.active += 1 + self.max_active = [max_active, active].max + end + + def leave + self.active -= 1 + end + end + + reset_tracking! + end + + class SharedFanoutLoader < GraphQL::Breadth::LazyLoader + async limit: 1, resource: :shared_plane_b_resource + + def map? + true + end + + def perform_map(keys, _ctx) + async_map(keys) do |key| + SharedResourceTracker.enter + sleep 0.01 + "#{key}-fanout" + ensure + SharedResourceTracker.leave + end + end + end + + class SharedPlainLoader < GraphQL::Breadth::LazyLoader + async limit: 1, resource: :shared_plane_b_resource + + def map? + true + end + + def perform_map(keys, _ctx) + SharedResourceTracker.enter + sleep 0.01 + keys.map { |key| "#{key}-plain" } + ensure + SharedResourceTracker.leave + end + end + class FirstResolver < GraphQL::Breadth::FieldResolver def resolve(exec_field, _ctx) exec_field.lazy(loader_class: FancyLoader, args: { group: "a" }, keys: exec_field.objects.map { _1["first"] }) @@ -94,6 +294,71 @@ def resolve(exec_field, _ctx) end end + class LoaderResolver < GraphQL::Breadth::FieldResolver + def initialize(loader_class:, key:, group: nil) + super() + @loader_class = loader_class + @key = key + @group = group + end + + def resolve(exec_field, _ctx) + args = @group ? { group: @group } : nil + exec_field.lazy( + loader_class: @loader_class, + args: args, + keys: exec_field.objects.map { _1[@key] }, + ) + end + end + + class ChainedLoaderResolver < GraphQL::Breadth::FieldResolver + def initialize(key:) + super() + @key = key + end + + def resolve(exec_field, _ctx) + exec_field + .lazy( + loader_class: ChainedConcurrentLoader, + args: { group: "fast", delay: 0.001 }, + keys: exec_field.objects.map { _1[@key] }, + ) + .then do |values| + exec_field.lazy( + loader_class: ChainedConcurrentLoader, + args: { group: "chain", delay: 0.001 }, + keys: values, + ) + end + end + end + + class SlowChainedLoaderResolver < GraphQL::Breadth::FieldResolver + def resolve(exec_field, _ctx) + exec_field.lazy( + loader_class: ChainedConcurrentLoader, + args: { group: "slow", delay: 0.05 }, + keys: exec_field.objects.map { _1["second"] }, + ) + end + end + + class NoLoaderPromiseResolver < GraphQL::Breadth::FieldResolver + def resolve(exec_field, _ctx) + exec_field + .lazy( + loader_class: ConcurrentLoader, + args: { group: "a" }, + keys: exec_field.objects.map { _1["first"] }, + ) + .then do + GraphQL::Breadth::Executor::ExecutionPromise.new + end + end + end + LOADER_SCHEMA = GraphQL::Schema.from_definition(%| type Widget { first: String @@ -123,6 +388,11 @@ def resolve(exec_field, _ctx) def setup FancyLoader.perform_keys = [] + ConcurrentLoader.reset_tracking! + LimitedConcurrentLoader.reset_tracking! + ChainedConcurrentLoader.reset_tracking! + FanoutMapLoader.reset_tracking! + SharedResourceTracker.reset_tracking! end def test_splits_loaders_by_group_across_fields @@ -157,6 +427,348 @@ def test_splits_loaders_by_group_across_fields assert_equal [["Apple", "Banana"], ["Coconut"]], FancyLoader.perform_keys end + def test_sync_lazy_loaders_do_not_enter_async_scheduler + document = GraphQL.parse(%|{ + widget { + first + second + } + }|) + + source = { + "widget" => { + "first" => "Apple", + "second" => "Banana", + }, + } + + expected = { + "data" => { + "widget" => { + "first" => "Apple-a", + "second" => "Banana-a", + }, + }, + } + + executor_class = Class.new(GraphQL::Breadth::Executor) do + private + + def execute_async_lazy_batches(*) + raise "Async scheduler should not run for sync lazy loaders" + end + end + + executor = executor_class.new(LOADER_SCHEMA, document, resolvers: LOADER_RESOLVERS, root_object: source) + assert_equal expected, executor.result + end + + def test_concurrency_loaders_overlap_when_opted_in + document = GraphQL.parse(%|{ + widget { + first + second + } + }|) + + source = { + "widget" => { + "first" => "Apple", + "second" => "Banana", + }, + } + + resolvers = LOADER_RESOLVERS.merge( + "Widget" => LOADER_RESOLVERS["Widget"].merge( + "first" => LoaderResolver.new(loader_class: ConcurrentLoader, key: "first", group: "a"), + "second" => LoaderResolver.new(loader_class: ConcurrentLoader, key: "second", group: "b"), + ), + ) + + expected = { + "data" => { + "widget" => { + "first" => "Apple-a", + "second" => "Banana-b", + }, + }, + } + + executor = GraphQL::Breadth::Executor.new(LOADER_SCHEMA, document, resolvers: resolvers, root_object: source) + assert_equal expected, executor.result + assert_equal 2, ConcurrentLoader.max_active + end + + def test_concurrency_loaders_respect_resource_limits + document = GraphQL.parse(%|{ + widget { + first + second + } + }|) + + source = { + "widget" => { + "first" => "Apple", + "second" => "Banana", + }, + } + + resolvers = LOADER_RESOLVERS.merge( + "Widget" => LOADER_RESOLVERS["Widget"].merge( + "first" => LoaderResolver.new(loader_class: LimitedConcurrentLoader, key: "first", group: "a"), + "second" => LoaderResolver.new(loader_class: LimitedConcurrentLoader, key: "second", group: "b"), + ), + ) + + expected = { + "data" => { + "widget" => { + "first" => "Apple-a", + "second" => "Banana-b", + }, + }, + } + + executor = GraphQL::Breadth::Executor.new(LOADER_SCHEMA, document, resolvers: resolvers, root_object: source) + assert_equal expected, executor.result + assert_equal 1, LimitedConcurrentLoader.max_active + end + + def test_concurrency_loader_errors_are_applied_to_field_results + document = GraphQL.parse(%|{ + widget { + first + } + }|) + + source = { + "widget" => { + "first" => "Apple", + }, + } + + resolvers = LOADER_RESOLVERS.merge( + "Widget" => LOADER_RESOLVERS["Widget"].merge( + "first" => LoaderResolver.new(loader_class: BoomConcurrentLoader, key: "first"), + ), + ) + + executor = GraphQL::Breadth::Executor.new(LOADER_SCHEMA, document, resolvers: resolvers, root_object: source) + result = executor.result + + assert_equal({ "widget" => { "first" => nil } }, result["data"]) + assert_equal "async loader boom", result.dig("errors", 0, "message") + assert_equal ["widget", "first"], result.dig("errors", 0, "path") + end + + def test_concurrency_loader_fatal_exceptions_raise + document = GraphQL.parse(%|{ + widget { + first + } + }|) + + source = { + "widget" => { + "first" => "Apple", + }, + } + + resolvers = LOADER_RESOLVERS.merge( + "Widget" => LOADER_RESOLVERS["Widget"].merge( + "first" => LoaderResolver.new(loader_class: FatalConcurrentLoader, key: "first"), + ), + ) + + executor = GraphQL::Breadth::Executor.new(LOADER_SCHEMA, document, resolvers: resolvers, root_object: source) + + error = assert_raises(FatalConcurrentLoader::FatalError) { executor.result } + assert_equal "fatal async loader boom", error.message + end + + def test_concurrency_chains_are_scheduled_while_slow_sibling_loader_is_running + document = GraphQL.parse(%|{ + widget { + first + second + } + }|) + + source = { + "widget" => { + "first" => "Apple", + "second" => "Banana", + }, + } + + resolvers = LOADER_RESOLVERS.merge( + "Widget" => LOADER_RESOLVERS["Widget"].merge( + "first" => ChainedLoaderResolver.new(key: "first"), + "second" => SlowChainedLoaderResolver.new, + ), + ) + + expected = { + "data" => { + "widget" => { + "first" => "Apple-fast-chain", + "second" => "Banana-slow", + }, + }, + } + + executor = GraphQL::Breadth::Executor.new(LOADER_SCHEMA, document, resolvers: resolvers, root_object: source) + assert_equal expected, executor.result + + events = ChainedConcurrentLoader.events + assert_operator events.index([:start, "chain"]), :<, events.index([:finish, "slow"]) + end + + def test_async_requeued_promise_without_loader_raises + document = GraphQL.parse(%|{ + widget { + first + } + }|) + + source = { + "widget" => { + "first" => "Apple", + }, + } + + resolvers = LOADER_RESOLVERS.merge( + "Widget" => LOADER_RESOLVERS["Widget"].merge( + "first" => NoLoaderPromiseResolver.new, + ), + ) + + executor = GraphQL::Breadth::Executor.new(LOADER_SCHEMA, document, resolvers: resolvers, root_object: source) + + error = assert_raises(GraphQL::Breadth::ImplementationError) { executor.result } + assert_match(/produced a promise without a loader/, error.message) + end + + def test_async_map_fans_out_inside_loader_with_ordered_results + document = GraphQL.parse(%|{ + widget { + first + second + } + }|) + + source = { + "widget" => { + "first" => "Apple", + "second" => "Banana", + }, + } + + resolvers = LOADER_RESOLVERS.merge( + "Widget" => LOADER_RESOLVERS["Widget"].merge( + "first" => LoaderResolver.new(loader_class: FanoutMapLoader, key: "first"), + "second" => LoaderResolver.new(loader_class: FanoutMapLoader, key: "second"), + ), + ) + + expected = { + "data" => { + "widget" => { + "first" => "Apple-fanout", + "second" => "Banana-fanout", + }, + }, + } + + executor = GraphQL::Breadth::Executor.new(LOADER_SCHEMA, document, resolvers: resolvers, root_object: source) + assert_equal expected, executor.result + assert_equal 2, FanoutMapLoader.max_active + end + + def test_async_map_shares_resource_limits_with_plain_concurrent_loaders + document = GraphQL.parse(%|{ + widget { + first + second + } + }|) + + source = { + "widget" => { + "first" => "Apple", + "second" => "Banana", + }, + } + + resolvers = LOADER_RESOLVERS.merge( + "Widget" => LOADER_RESOLVERS["Widget"].merge( + "first" => LoaderResolver.new(loader_class: SharedFanoutLoader, key: "first"), + "second" => LoaderResolver.new(loader_class: SharedPlainLoader, key: "second"), + ), + ) + + expected = { + "data" => { + "widget" => { + "first" => "Apple-fanout", + "second" => "Banana-plain", + }, + }, + } + + executor = GraphQL::Breadth::Executor.new(LOADER_SCHEMA, document, resolvers: resolvers, root_object: source) + assert_equal expected, executor.result + assert_equal 1, SharedResourceTracker.max_active + end + + def test_async_map_errors_are_applied_to_field_results + document = GraphQL.parse(%|{ + widget { + first + second + } + }|) + + source = { + "widget" => { + "first" => "Apple", + "second" => "Banana", + }, + } + + resolvers = LOADER_RESOLVERS.merge( + "Widget" => LOADER_RESOLVERS["Widget"].merge( + "first" => LoaderResolver.new(loader_class: BoomFanoutMapLoader, key: "first"), + "second" => LoaderResolver.new(loader_class: BoomFanoutMapLoader, key: "second"), + ), + ) + + executor = GraphQL::Breadth::Executor.new(LOADER_SCHEMA, document, resolvers: resolvers, root_object: source) + result = executor.result + + assert_equal({ "widget" => { "first" => nil, "second" => "Banana-fanout" } }, result["data"]) + assert_equal "async map boom", result.dig("errors", 0, "message") + assert_equal ["widget", "first"], result.dig("errors", 0, "path") + end + + def test_async_requires_async_initialization + GraphQL::Breadth.instance_variable_set(:@async_enabled, false) + + error = assert_raises(GraphQL::Breadth::ImplementationError) do + Class.new(GraphQL::Breadth::LazyLoader) do + async + end + end + assert_match(/enable_async!/, error.message) + ensure + GraphQL::Breadth.enable_async! + end + + def test_async_defaults_to_per_loader_symbol_resource + assert_instance_of Symbol, ConcurrentLoader.async_settings.resource + end + def test_maintains_ordered_selections_around_object_fields document = GraphQL.parse(%|{ widget { diff --git a/test/test_helper.rb b/test/test_helper.rb index bbafeb4..983c09e 100644 --- a/test/test_helper.rb +++ b/test/test_helper.rb @@ -5,6 +5,7 @@ # ignore warnings from auto-generated GraphQL lib code. Warning.ignore(/.*mismatched indentations.*/) Warning.ignore(/.*lib\/graphql\/language\/nodes.rb:.*/) + Warning.ignore(/.*io-event.*IO::Buffer is experimental.*/) end require 'bundler/setup' @@ -15,6 +16,7 @@ require 'minitest/stub_const' require 'graphql/breadth' +GraphQL::Breadth.enable_async! require 'graphql/batch' require_relative './fixtures' require_relative './star_wars_fixtures'