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

Jump to bottom

[tune/execution] Update staged resources in a fixed counter for faster lookup #32087

Merged
merged 6 commits into from
Jan 31, 2023
Merged

[tune/execution] Update staged resources in a fixed counter for faster lookup #32087

Show file tree
Hide file tree
Changes from 4 commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
6c4795e
[tune] Cache staged resources for faster lookup
Jan 31, 2023
ee82d48
adjust timeout
Jan 31, 2023
d7cd168
Revert throughput changes
Jan 31, 2023
7e31cf1
comments
Jan 31, 2023
1bd5be1
Merge remote-tracking branch 'upstream/master' into tune/cache-staged…
Jan 31, 2023
836833e
Comment
Jan 31, 2023
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
12 changes: 6 additions & 6 deletions python/ray/tune/execution/ray_trial_executor.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -232,7 +232,8 @@ def __init__(
] = defaultdict(list)

# Trials for which we requested resources
self._staged_trials = set()
self._staged_trials = set() # Staged trials
self._staged_resources = Counter() # Resources of staged trials
self._trial_to_acquired_resources: Dict[Trial, AcquiredResources] = {}

# Result buffer
Expand Down Expand Up @@ -319,6 +320,7 @@ def _stage_and_update_status(self, trials: Iterable[Trial]):
resource_request = trial.placement_group_factory

self._staged_trials.add(trial)
self._staged_resources[trial.placement_group_factory] += 1
self._resource_manager.request_resources(resource_request=resource_request)

self._resource_manager.update_state()
Expand Down Expand Up @@ -533,6 +535,7 @@ def _unstage_trial_with_resources(self, trial: Trial):
# Case 1: The trial we started was staged. Just remove it
if trial in self._staged_trials:
self._staged_trials.remove(trial)
self._staged_resources[trial.placement_group_factory] -= 1
return

# Case 2: We staged a trial "A" with the same resources, but our trial "B"
Expand All @@ -551,6 +554,7 @@ def _unstage_trial_with_resources(self, trial: Trial):

if candidate_trial:
self._staged_trials.remove(candidate_trial)
self._staged_resources[candidate_trial.placement_group_factory] -= 1
return

raise RuntimeError(
Expand Down Expand Up @@ -848,11 +852,7 @@ def on_step_end(self, search_ended: bool = False) -> None:
self._do_force_trial_cleanup()

def _count_staged_resources(self):
counter = Counter()
for trial in self._staged_trials:
resource_request = trial.placement_group_factory
counter[resource_request] += 1
return counter
return self._staged_resources

def _cleanup_cached_actors(
self, search_ended: bool = False, force_all: bool = False
Expand Down
2 changes: 1 addition & 1 deletion release/ray_release/alerts/tune_tests.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -44,7 +44,7 @@ def handle_result(
target_time = 900 if not was_smoke_test else 400
elif test_name == "result_throughput_cluster":
target_terminated = 1000
target_time = 160
target_time = 130
krfricke marked this conversation as resolved.
Show resolved Hide resolved
elif test_name == "result_throughput_single_node":
target_terminated = 96
target_time = 120
Expand Down
7 changes: 1 addition & 6 deletions release/tune_tests/scalability_tests/workloads/test_result_throughput_cluster.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,6 @@

import ray
from ray import tune
from ray.tune.execution.cluster_info import _is_ray_cluster

from ray.tune.utils.release_test_util import timed_tune_run

Expand All @@ -31,11 +30,7 @@ def main():
results_per_second = 0.5
trial_length_s = 100

max_runtime = 160

if _is_ray_cluster():
# Add constant overhead for SSH connection
max_runtime = 160
max_runtime = 130

timed_tune_run(
name="result throughput cluster",
Expand Down