class GraphQL::Dataloader::AsyncDataloader::Run

# File lib/graphql/dataloader/async_dataloader.rb, line 52
def initialize(dataloader, total_fiber_limit, jobs_fiber_limit)
  @dataloader = dataloader
  @root_task = nil
  @trace = nil
  @jobs = []

  @total_fiber_limit = total_fiber_limit
  @jobs_fiber_limit = jobs_fiber_limit
  @lazies_at_depth = Hash.new { |h, k| h[k] = [] }

  @running_tasks = nil
  @tasks_channel = nil
  @tasks_channel_task = nil
  @activity = nil
  @task_error = nil
  @expected_resumes = 0
  @mode = nil

  @snoozed_jobs_condition = Async::Condition.new
  @snoozed_sources_condition = Async::Condition.new
end

# File lib/graphql/dataloader/async_dataloader.rb, line 127
def check_error!
  if (err = @task_error)
    @task_error = nil
    raise err
  end
end

# File lib/graphql/dataloader/async_dataloader.rb, line 86
def close_queues
  @tasks_channel.close
  @tasks_channel_task.cancel
end

# File lib/graphql/dataloader/async_dataloader.rb, line 168
def current_sources_fiber_limit
  within_limit = @total_fiber_limit - running_count
  if within_limit < 1
    1
  else
    within_limit
  end
end

# File lib/graphql/dataloader/async_dataloader.rb, line 123
def expect_resumes(count)
  @expected_resumes = count
end

# File lib/graphql/dataloader/async_dataloader.rb, line 117
def has_bandwidth?
  @mode == :jobs ? jobs_bandwidth? : sources_bandwidth?
end

# File lib/graphql/dataloader/async_dataloader.rb, line 113
def has_pending_work?
  @mode == :jobs ? @jobs.any? : @dataloader.pending_sources.any?(&:pending?) # rubocop:disable Development/NoneWithoutBlockCop
end

# File lib/graphql/dataloader/async_dataloader.rb, line 78
def jobs_bandwidth?
  running_count < @jobs_fiber_limit
end

# File lib/graphql/dataloader/async_dataloader.rb, line 134
def new_queues(mode)
  @mode = mode
  @tasks_channel = Async::Queue.new(parent: @root_task)
  @activity = Async::Condition.new
  @task_error = nil
  @expected_resumes = 0
  @running_tasks = []
  @tasks_channel_task = @root_task.async do |_t|
    while ((msg, data) = @tasks_channel.wait)
      case msg
      when :started_task
        @running_tasks.push(data)
        data.run
      when :resumed_task
        if @expected_resumes > 0
          @expected_resumes -= 1
        end
        @running_tasks.push(data)
      when :finished_task, :paused_task
        @running_tasks.delete(data)
      when :task_error
        @task_error ||= data
      else
        raise ArgumentError, "Unknown tasks_channel action: #{msg.inspect}"
      end
      @activity.signal
    end
  end
end

# File lib/graphql/dataloader/async_dataloader.rb, line 95
def push_task_message(msg, data)
  @tasks_channel.push([msg, data])
  true
rescue Async::Queue::ClosedError
  if msg == :task_error
    @task_error ||= data
  end
  false
end

# File lib/graphql/dataloader/async_dataloader.rb, line 109
def quiesced?
  @running_tasks.empty? && @tasks_channel.empty? && @expected_resumes == 0
end

# File lib/graphql/dataloader/async_dataloader.rb, line 164
def running?
  @snoozed_jobs_condition.waiting? || @snoozed_sources_condition.waiting?
end

# File lib/graphql/dataloader/async_dataloader.rb, line 82
def sources_bandwidth?
  running_count < current_sources_fiber_limit
end

# File lib/graphql/dataloader/async_dataloader.rb, line 105
def wait_for_activity
  @activity.wait
end