Class: GraphQL::Dataloader::AsyncDataloader::Run

Inherits:
Object
  • Object
show all
Defined in:
lib/graphql/dataloader/async_dataloader.rb

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(dataloader, total_fiber_limit, jobs_fiber_limit) ⇒ Run

Returns a new instance of Run.



44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
 
# File 'lib/graphql/dataloader/async_dataloader.rb', line 44

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
  @finished_all_tasks = nil

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

Instance Attribute Details

#jobs ⇒ Object (readonly)

Returns the value of attribute jobs.



65
66
67
 
# File 'lib/graphql/dataloader/async_dataloader.rb', line 65

def jobs
  @jobs
end

#jobs_fiber_limit ⇒ Object (readonly)

Returns the value of attribute jobs_fiber_limit.



65
66
67
 
# File 'lib/graphql/dataloader/async_dataloader.rb', line 65

def jobs_fiber_limit
  @jobs_fiber_limit
end

#lazies_at_depth ⇒ Object (readonly)

Returns the value of attribute lazies_at_depth.



65
66
67
 
# File 'lib/graphql/dataloader/async_dataloader.rb', line 65

def lazies_at_depth
  @lazies_at_depth
end

#root_task ⇒ Object

Returns the value of attribute root_task.



63
64
65
 
# File 'lib/graphql/dataloader/async_dataloader.rb', line 63

def root_task
  @root_task
end

#snoozed_jobs_condition ⇒ Object (readonly)

Returns the value of attribute snoozed_jobs_condition.



65
66
67
 
# File 'lib/graphql/dataloader/async_dataloader.rb', line 65

def snoozed_jobs_condition
  @snoozed_jobs_condition
end

#snoozed_sources_condition ⇒ Object (readonly)

Returns the value of attribute snoozed_sources_condition.



65
66
67
 
# File 'lib/graphql/dataloader/async_dataloader.rb', line 65

def snoozed_sources_condition
  @snoozed_sources_condition
end

#tasks_channel ⇒ Object (readonly)

Returns the value of attribute tasks_channel.



65
66
67
 
# File 'lib/graphql/dataloader/async_dataloader.rb', line 65

def tasks_channel
  @tasks_channel
end

#trace ⇒ Object

Returns the value of attribute trace.



63
64
65
 
# File 'lib/graphql/dataloader/async_dataloader.rb', line 63

def trace
  @trace
end

Instance Method Details

#close_queues ⇒ Object 



75
76
77
78
 
# File 'lib/graphql/dataloader/async_dataloader.rb', line 75

def close_queues
  @tasks_channel.close
  @tasks_channel_task.cancel
end

#current_sources_fiber_limit ⇒ Object 



127
128
129
130
131
132
133
134
 
# File 'lib/graphql/dataloader/async_dataloader.rb', line 127

def current_sources_fiber_limit
  within_limit = @total_fiber_limit - running_count
  if within_limit < 1
    1
  else
    within_limit
  end
end

#jobs_bandwidth? ⇒ Boolean

Returns:

  • (Boolean) 


67
68
69
 
# File 'lib/graphql/dataloader/async_dataloader.rb', line 67

def jobs_bandwidth?
  running_count < @jobs_fiber_limit
end

#new_queues(mode) ⇒ Object 



90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
 
# File 'lib/graphql/dataloader/async_dataloader.rb', line 90

def new_queues(mode)
  @tasks_channel = Async::Queue.new(parent: @root_task)
  @no_running_tasks = Async::Promise.new
  @finished_all_tasks = Async::Promise.new
  @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
        @running_tasks.push(data)
      when :finished_task, :paused_task
        @running_tasks.delete(data)
        has_pending_work = mode == :jobs ? @jobs.any? : @dataloader.pending_sources.any?(&:pending?) # rubocop:disable Development/NoneWithoutBlockCop
        if @running_tasks.empty?
          @no_running_tasks.resolve(true)
          has_bandwidth = mode == :jobs ? jobs_bandwidth? : sources_bandwidth?
          if (!has_pending_work) || (!has_bandwidth)
            @finished_all_tasks.resolve(true)
          end
        end
      when :task_error
        @no_running_tasks.resolve(true)
        @finished_all_tasks.reject(data)
      else
        raise ArgumentError, "Unknown tasks_channel action: #{msg.inspect}"
      end
    end
  end
end

#running? ⇒ Boolean

Returns:

  • (Boolean) 


123
124
125
 
# File 'lib/graphql/dataloader/async_dataloader.rb', line 123

def running?
  @snoozed_jobs_condition.waiting? || @snoozed_sources_condition.waiting?
end

#sources_bandwidth? ⇒ Boolean

Returns:

  • (Boolean) 


71
72
73
 
# File 'lib/graphql/dataloader/async_dataloader.rb', line 71

def sources_bandwidth?
  running_count < current_sources_fiber_limit
end

#wait_for_no_running_tasks ⇒ Object 



85
86
87
88
 
# File 'lib/graphql/dataloader/async_dataloader.rb', line 85

def wait_for_no_running_tasks
  @no_running_tasks.wait
  @no_running_tasks = Async::Promise.new
end

#wait_for_queues ⇒ Object 



80
81
82
83
 
# File 'lib/graphql/dataloader/async_dataloader.rb', line 80

def wait_for_queues
  @finished_all_tasks.wait
  @finished_all_tasks = Async::Promise.new
end