Skip to content

Share lazy job scheduling across dataloaders #5669

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
ydah wants to merge 1 commit into
base: master
Choose a base branch
from
+19 −43
Open

Share lazy job scheduling across dataloaders #5669

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
29 changes: 12 additions & 17 deletions lib/graphql/dataloader.rb
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -212,7 +212,9 @@ def run(trace_query_lazy: nil)

if !@lazies_at_depth.empty?
with_trace_query_lazy(trace_query_lazy) do
run_next_pending_lazies(job_fibers, trace)
run_next_pending_lazies(@lazies_at_depth) do
job_fibers.unshift(spawn_job_fiber(trace))
end
run_pending_steps(trace, job_fibers, next_job_fibers, jobs_fiber_limit, source_fibers, next_source_fibers, total_fiber_limit)
end
end
Expand Down Expand Up @@ -274,24 +276,17 @@ def merge_records(records, index_by: :id)

private

def run_next_pending_lazies(job_fibers, trace)
smallest_depth = nil
@lazies_at_depth.each_key do |depth_key|
smallest_depth ||= depth_key
if depth_key < smallest_depth
smallest_depth = depth_key
end
end
def run_next_pending_lazies(lazies_at_depth)
smallest_depth = lazies_at_depth.each_key.min
return if smallest_depth.nil?

if smallest_depth
lazies = @lazies_at_depth.delete(smallest_depth)
if !lazies.empty?
lazies.each_with_index do |l, idx|
append_job { l.value }
end
job_fibers.unshift(spawn_job_fiber(trace))
end
lazies = lazies_at_depth.delete(smallest_depth)
return if lazies.empty?

lazies.each do |lazy|
append_job { lazy.value }
end
yield
end

def run_pending_steps(trace, job_fibers, next_job_fibers, jobs_fiber_limit, source_fibers, next_source_fibers, total_fiber_limit)
Expand Down
33 changes: 7 additions & 26 deletions lib/graphql/dataloader/async_dataloader.rb
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -208,7 +208,7 @@ def run(trace_query_lazy: nil)

if !run.lazies_at_depth.empty?
with_trace_query_lazy(trace_query_lazy) do
run_next_pending_lazies(run)
run_next_pending_lazies(run.lazies_at_depth) { run_lazy_jobs(run) }
run_pending_steps(run)
end
end
Expand Down Expand Up @@ -291,31 +291,12 @@ def run_sources(run)
run.close_queues
end

#### TODO DRY Had to duplicate to remove spawn_job_fiber
def run_next_pending_lazies(run)
smallest_depth = nil
run.lazies_at_depth.each_key do |depth_key|
smallest_depth ||= depth_key
if depth_key < smallest_depth
smallest_depth = depth_key
end
end

if smallest_depth
lazies = run.lazies_at_depth.delete(smallest_depth)
if !lazies.empty?
begin
run.new_queues
lazies.each_with_index do |l, idx|
append_job { l.value }
end
spawn_job_task(run) # Todo what was the last `true` condition?
run.wait_for_queues
ensure
run.close_queues
end
end
end
def run_lazy_jobs(run)
run.new_queues
spawn_job_task(run)
run.wait_for_queues
ensure
run.close_queues
end

def spawn_source_task(run, num_tasks)
Expand Down
Loading