Skip to content
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

No longer double count transfer cost in stealing #7036

Merged
merged 10 commits into from
Sep 16, 2022
Merged

No longer double count transfer cost in stealing #7036

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
10 commits
Select commit Hold shift + click to select a range
a69b602
No longer double count transfer cost in stealing
fjetter Sep 8, 2022
c11b38c
Remove stealable_all
fjetter Sep 12, 2022
a3dd984
Fix combined_nprocessing
hendrikmakait Sep 15, 2022
cfae9ee
Fix test_steal_cheap_data_slow_computation
hendrikmakait Sep 15, 2022
cb26af8
Fix in-flight calculations
hendrikmakait Sep 15, 2022
80a274a
Revert "Fix test_steal_cheap_data_slow_computation"
hendrikmakait Sep 15, 2022
cf3747b
Refactoring
hendrikmakait Sep 15, 2022
b4db26a
Extract logic for unoccupied worker determination
hendrikmakait Sep 15, 2022
4f33c16
Fix thief selection
hendrikmakait Sep 16, 2022
9c9438e
Cleanup
hendrikmakait Sep 16, 2022
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
9 changes: 8 additions & 1 deletion distributed/scheduler.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2892,7 +2892,7 @@ def check_idle_saturated(self, ws: WorkerState, occ: float = -1.0):
idle = self.idle
saturated = self.saturated
if (
(p < nc or occ < nc * avg / 2)
self.is_unoccupied(ws, occ, p)
if math.isinf(self.WORKER_SATURATION)
else not _worker_full(ws, self.WORKER_SATURATION)
):
Expand All @@ -2910,6 +2910,13 @@ def check_idle_saturated(self, ws: WorkerState, occ: float = -1.0):

saturated.discard(ws)

def is_unoccupied(
self, ws: WorkerState, occupancy: float, nprocessing: int
) -> bool:
nthreads = ws.nthreads
avg_occ_per_thread = self.total_occupancy / self.total_nthreads
return nprocessing < nthreads or occupancy < nthreads * avg_occ_per_thread / 2

def get_comm_cost(self, ts: TaskState, ws: WorkerState) -> float:
"""
Get the estimated communication cost (in s.) to compute the task
Expand Down
Loading