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(total_fiber_limit, jobs_fiber_limit) ⇒ Run

Returns a new instance of Run.



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

def initialize(total_fiber_limit, jobs_fiber_limit)
  @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] = [] }

  @finished_tasks = nil
  @started_tasks = nil
  @started_count_task = nil
  @finished_count_task = nil
  @finished_all_tasks = nil
  @finished_first_pass = nil

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

Instance Attribute Details

#finished_tasks ⇒ Object (readonly)

Returns the value of attribute finished_tasks.



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

def finished_tasks
  @finished_tasks
end

#jobs ⇒ Object (readonly)

Returns the value of attribute jobs.



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

def jobs
  @jobs
end

#jobs_fiber_limit ⇒ Object (readonly)

Returns the value of attribute jobs_fiber_limit.



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

def jobs_fiber_limit
  @jobs_fiber_limit
end

#lazies_at_depth ⇒ Object (readonly)

Returns the value of attribute lazies_at_depth.



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

def lazies_at_depth
  @lazies_at_depth
end

#root_task ⇒ Object

Returns the value of attribute root_task.



61
62
63
 
# File 'lib/graphql/dataloader/async_dataloader.rb', line 61

def root_task
  @root_task
end

#snoozed_jobs_condition ⇒ Object (readonly)

Returns the value of attribute snoozed_jobs_condition.



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

def snoozed_jobs_condition
  @snoozed_jobs_condition
end

#snoozed_sources_condition ⇒ Object (readonly)

Returns the value of attribute snoozed_sources_condition.



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

def snoozed_sources_condition
  @snoozed_sources_condition
end

#started_tasks ⇒ Object (readonly)

Returns the value of attribute started_tasks.



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

def started_tasks
  @started_tasks
end

#total_fiber_limit ⇒ Object (readonly)

Returns the value of attribute total_fiber_limit.



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

def total_fiber_limit
  @total_fiber_limit
end

#trace ⇒ Object

Returns the value of attribute trace.



61
62
63
 
# File 'lib/graphql/dataloader/async_dataloader.rb', line 61

def trace
  @trace
end

Instance Method Details

#allowed_sources_tasks ⇒ Object 



69
70
71
72
73
74
75
76
 
# File 'lib/graphql/dataloader/async_dataloader.rb', line 69

def allowed_sources_tasks
  within_limit = total_fiber_limit - running_count
  if within_limit < 1
    1
  else
    within_limit
  end
end

#close_queues ⇒ Object 



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

def close_queues
  @finished_tasks.close
  @finished_count_task.cancel

  @started_tasks.close
  @started_count_task.cancel
end

#jobs_bandwidth? ⇒ Boolean

Returns:

  • (Boolean) 


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

def jobs_bandwidth?
  running_count < jobs_fiber_limit
end

#new_queues ⇒ Object 



103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
 
# File 'lib/graphql/dataloader/async_dataloader.rb', line 103

def new_queues
  @finished_tasks = Async::Queue.new
  @finished_count = 0
  @started_tasks = Async::Queue.new
  @started_count = 0
  @finished_first_pass = Async::Promise.new
  @finished_all_tasks = Async::Promise.new

  @started_count_task = @root_task.async do |task|
    @finished_first_pass.wait
    while task = @started_tasks.wait
      @started_count += 1
      if task.status == :initialized # could also be resumed after waiting
        task.run
      end
    end
  end

  @finished_count_task = @root_task.async do |task|
    while t_or_err = @finished_tasks.wait
      if t_or_err.is_a?(StandardError)
        @finished_all_tasks.reject(t_or_err)
      else
        @finished_count += 1
        if @finished_count == @started_count
          @finished_all_tasks.resolve(true)
        end
      end
    end
  end
end

#running? ⇒ Boolean

Returns:

  • (Boolean) 


135
136
137
 
# File 'lib/graphql/dataloader/async_dataloader.rb', line 135

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

#running_count ⇒ Object 



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

def running_count
  @snoozed_jobs_condition.instance_variable_get(:@ready).num_waiting +
    @snoozed_sources_condition.instance_variable_get(:@ready).num_waiting +
    @started_count +
    @started_tasks.size -
    @finished_count
end

#wait_for_queues ⇒ Object 



94
95
96
97
98
99
100
101
 
# File 'lib/graphql/dataloader/async_dataloader.rb', line 94

def wait_for_queues
  if !@finished_first_pass.resolved?
    @finished_first_pass.resolve(true)
  end

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