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.



52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
 
# 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

Instance Attribute Details

#dataloader ⇒ Object (readonly)

Returns the value of attribute dataloader.



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

def dataloader
  @dataloader
end

#jobs ⇒ Object (readonly)

Returns the value of attribute jobs.



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

def jobs
  @jobs
end

#jobs_fiber_limit ⇒ Object (readonly)

Returns the value of attribute jobs_fiber_limit.



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

def jobs_fiber_limit
  @jobs_fiber_limit
end

#lazies_at_depth ⇒ Object (readonly)

Returns the value of attribute lazies_at_depth.



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

def lazies_at_depth
  @lazies_at_depth
end

#root_task ⇒ Object

Returns the value of attribute root_task.



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

def root_task
  @root_task
end

#snoozed_jobs_condition ⇒ Object (readonly)

Returns the value of attribute snoozed_jobs_condition.



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

def snoozed_jobs_condition
  @snoozed_jobs_condition
end

#snoozed_sources_condition ⇒ Object (readonly)

Returns the value of attribute snoozed_sources_condition.



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

def snoozed_sources_condition
  @snoozed_sources_condition
end

#trace ⇒ Object

Returns the value of attribute trace.



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

def trace
  @trace
end

Instance Method Details

#check_error! ⇒ Object 



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

def check_error!
  if (err = @task_error)
    @task_error = nil
    raise err
  end
end

#close_queues ⇒ Object 



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

def close_queues
  @tasks_channel.close
  @tasks_channel_task.cancel
end

#current_sources_fiber_limit ⇒ Object 



168
169
170
171
172
173
174
175
 
# 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

#expect_resumes(count) ⇒ Object

Signalled tasks don't appear in any accounting until their first slice pushes :resumed_task, so they have to be counted at signal time:



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

def expect_resumes(count)
  @expected_resumes = count
end

#has_bandwidth? ⇒ Boolean

Returns:

  • (Boolean) 


117
118
119
 
# File 'lib/graphql/dataloader/async_dataloader.rb', line 117

def has_bandwidth?
  @mode == :jobs ? jobs_bandwidth? : sources_bandwidth?
end

#has_pending_work? ⇒ Boolean

Returns:

  • (Boolean) 


113
114
115
 
# 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

#jobs_bandwidth? ⇒ Boolean

Returns:

  • (Boolean) 


78
79
80
 
# File 'lib/graphql/dataloader/async_dataloader.rb', line 78

def jobs_bandwidth?
  running_count < @jobs_fiber_limit
end

#new_queues(mode) ⇒ Object 



134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
 
# 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

#push_task_message(msg, data) ⇒ Object

Push to the tasks_channel, tolerating a closed channel: on the error path, run_queue closes the channel while sibling tasks can still run one more slice before root_task.cancel reaches them. Record :task_error payloads so they aren't lost, and return false so the caller can stop the task instead of raising ClosedError into user code.



95
96
97
98
99
100
101
102
103
 
# 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

#quiesced? ⇒ Boolean

Returns:

  • (Boolean) 


109
110
111
 
# File 'lib/graphql/dataloader/async_dataloader.rb', line 109

def quiesced?
  @running_tasks.empty? && @tasks_channel.empty? && @expected_resumes == 0
end

#running? ⇒ Boolean

Returns:

  • (Boolean) 


164
165
166
 
# File 'lib/graphql/dataloader/async_dataloader.rb', line 164

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

#sources_bandwidth? ⇒ Boolean

Returns:

  • (Boolean) 


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

def sources_bandwidth?
  running_count < current_sources_fiber_limit
end

#wait_for_activity ⇒ Object 



105
106
107
 
# File 'lib/graphql/dataloader/async_dataloader.rb', line 105

def wait_for_activity
  @activity.wait
end