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.



48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
 
# File 'lib/graphql/dataloader/async_dataloader.rb', line 48

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

#jobs ⇒ Object (readonly)

Returns the value of attribute jobs.



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

def jobs
  @jobs
end

#jobs_fiber_limit ⇒ Object (readonly)

Returns the value of attribute jobs_fiber_limit.



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

def jobs_fiber_limit
  @jobs_fiber_limit
end

#lazies_at_depth ⇒ Object (readonly)

Returns the value of attribute lazies_at_depth.



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

def lazies_at_depth
  @lazies_at_depth
end

#root_task ⇒ Object

Returns the value of attribute root_task.



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

def root_task
  @root_task
end

#snoozed_jobs_condition ⇒ Object (readonly)

Returns the value of attribute snoozed_jobs_condition.



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

def snoozed_jobs_condition
  @snoozed_jobs_condition
end

#snoozed_sources_condition ⇒ Object (readonly)

Returns the value of attribute snoozed_sources_condition.



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

def snoozed_sources_condition
  @snoozed_sources_condition
end

#tasks_channel ⇒ Object (readonly)

Returns the value of attribute tasks_channel.



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

def tasks_channel
  @tasks_channel
end

#trace ⇒ Object

Returns the value of attribute trace.



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

def trace
  @trace
end

Instance Method Details

#check_error! ⇒ Object 



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

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

#close_queues ⇒ Object 



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

def close_queues
  @tasks_channel.close
  @tasks_channel_task.cancel
end

#current_sources_fiber_limit ⇒ Object 



150
151
152
153
154
155
156
157
 
# File 'lib/graphql/dataloader/async_dataloader.rb', line 150

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:



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

def expect_resumes(count)
  @expected_resumes = count
end

#has_bandwidth? ⇒ Boolean

Returns:

  • (Boolean) 


99
100
101
 
# File 'lib/graphql/dataloader/async_dataloader.rb', line 99

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

#has_pending_work? ⇒ Boolean

Returns:

  • (Boolean) 


95
96
97
 
# File 'lib/graphql/dataloader/async_dataloader.rb', line 95

def has_pending_work?
  @mode == :jobs ? @jobs.any? : @dataloader.pending_sources.any?(&:pending?) # rubocop:disable Development/NoneWithoutBlockCop
end

#jobs_bandwidth? ⇒ Boolean

Returns:

  • (Boolean) 


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

def jobs_bandwidth?
  running_count < @jobs_fiber_limit
end

#new_queues(mode) ⇒ Object 



116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
 
# File 'lib/graphql/dataloader/async_dataloader.rb', line 116

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

#quiesced? ⇒ Boolean

Returns:

  • (Boolean) 


91
92
93
 
# File 'lib/graphql/dataloader/async_dataloader.rb', line 91

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

#running? ⇒ Boolean

Returns:

  • (Boolean) 


146
147
148
 
# File 'lib/graphql/dataloader/async_dataloader.rb', line 146

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

#sources_bandwidth? ⇒ Boolean

Returns:

  • (Boolean) 


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

def sources_bandwidth?
  running_count < current_sources_fiber_limit
end

#wait_for_activity ⇒ Object 



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

def wait_for_activity
  @activity.wait
end