[Autoscaler][Placement Group] Skip placed bundle when requesting resource #48924
Conversation
| shapes = [dict(bundle.unit_resources) for bundle in placement_group.bundles] | ||
| # Skip **placed** bundle (which has node id associated with it). | ||
| for bundle in placement_group.bundles: | ||
| if bundle.node_id: |
There was a problem hiding this comment.
Is it an empty string or None? If it is None, use is instead.
| if bundle.node_id: | |
| if bundle.node_id is not None: |
There was a problem hiding this comment.
fix in 7a44207, it should be an empty byte string.
Signed-off-by: Mimi Liao <mimiliao2000@gmail.com>
Signed-off-by: Mimi Liao <mimiliao2000@gmail.com>
mimiliaogo
force-pushed
the
pg-overprovision
branch
from
5818b44 to
6353847
Compare
| break | ||
|
|
||
| # TODO(mimi): kill_raylet won't trigger reschedule in autoscaler v1 | ||
| # kill_raylet(node["NodeManagerAddress"], node["NodeManagerPort"]) |
There was a problem hiding this comment.
I found that when using kill_raylet, reschedule won't be triggered in autoscaler v1, even when the cluster status shows the node is killed. In this case, v1 will fail and v2 pass.
Both v1 and v2 pass when using kill_node.
Signed-off-by: Mimi Liao <mimiliao2000@gmail.com>
|
|
||
| from ray.autoscaler.v2.sdk import get_cluster_status | ||
|
|
||
| def verify_nodes(active=3, idle=1): |
There was a problem hiding this comment.
| def verify_nodes(active=3, idle=1): | |
| def verify_nodes(active, idle): |
|
|
||
| def kill_node(node_id): | ||
| # kill -9 | ||
| import subprocess |
There was a problem hiding this comment.
Move import to top-level. Typically, Ray uses deferred import only to avoid circular dependencies.
| wait_for_condition(lambda: verify_nodes(3, 1)) | ||
|
|
||
| # Kill a node | ||
| def kill_raylet(ip, port, graceful=True): |
There was a problem hiding this comment.
Remove this function because it is not used for now.
| # Wait for the node to be removed | ||
| wait_for_condition(lambda: verify_nodes(2, 1), 20) | ||
|
|
||
| # Check that the placement group is rescheduled |
There was a problem hiding this comment.
where is the logic to check the placement group is rescheduled?
There was a problem hiding this comment.
wait_for_condition(lambda: verify_nodes(3, 1)) is to check the autoscaler rescheduling. However, this comment is redundant, I've already removed it.
|
|
||
| ray.get(pg.ready()) | ||
|
|
||
| from ray.autoscaler.v2.sdk import get_cluster_status |
There was a problem hiding this comment.
Do we need to import this? It seems to have already been imported at the top level.
There was a problem hiding this comment.
Above suggestions fixed in 415dcf8
Signed-off-by: Mimi Liao <mimiliao2000@gmail.com>
|
CI fails. Can you fix the CI errors? |
Why are these changes needed?
Before the PR, when a node in a placement group (PG) goes down, the autoscaler attempts to reschedule the entire PG (all bundles). However, this will lead to overprovisioning. Details: #40212
This PR solved this by skipping already placed bundles (i.e., bundles with an associated node_id) when demanding resources in autoscaler.
Before: Every bundles get rescheduled
After: Only one node will be scaled up
Related issue number
Closes #40212
Checks
git commit -s) in this PR.scripts/format.shto lint the changes in this PR.method in Tune, I've added it in
doc/source/tune/api/under thecorresponding
.rstfile.