[Data] Streamlining DefaultActorPoolAutoscaler#61385
Conversation
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request refactors the DefaultActorPoolAutoscaler by inlining the ActorPoolResizingPolicy, which simplifies the structure. It also improves the autoscaling logic by calculating utilization based on all actors (including pending ones) and allowing scale-up operations even when there are pending actors. The changes are well-reasoned and the accompanying test updates are thorough. I have a few suggestions to improve code clarity and fix a potential issue in one of the tests.
| assert actor_pool.get_pool_util() == 1.5 | ||
| assert_autoscaling_action( | ||
| delta=1, | ||
| delta=5, |
There was a problem hiding this comment.
The expected delta of 5 seems incorrect. Given the test setup, actor_pool_max_upscaling_delta defaults to 1. The calculated scale-up delta is ceil(10 * (1.5 - 1)) = 5, but this should be capped by max_scale_up, which evaluates to min(inf, 1, 5) = 1. Therefore, the final delta should be 1. Please verify the logic or update the test expectation.
| delta=5, | |
| delta=1, |
| delta = min(delta, max_scale_up) | ||
| delta = max(1, delta) # At least scale up by 1 | ||
|
|
||
| delta = self._compute_upscale_delta(actor_pool, op_state) |
There was a problem hiding this comment.
The op_state parameter is unused in the _compute_upscale_delta method (defined at line 230). Consider removing it from the method signature and this call site for cleaner code.
| delta = self._compute_upscale_delta(actor_pool, op_state) | |
| delta = self._compute_upscale_delta(actor_pool) |
| def _estimate_expected_tasks( | ||
| op_state: OpState, | ||
| ) -> float: | ||
| # Each task consumes `average_num_inputs_per_task` input blocks on average, | ||
| # so the total expected number of tasks: | ||
| # | ||
| # ceil(num enqueued blocks / avg_inputs_per_task) | ||
| # | ||
| avg_input_blocks_per_task = op_state.op.metrics.average_num_inputs_per_task or 1 | ||
| return math.ceil(op_state.total_enqueued_input_blocks() / avg_input_blocks_per_task) |
There was a problem hiding this comment.
The function _estimate_expected_tasks estimates a number of tasks but returns a float. Since tasks are discrete units, it would be more idiomatic and type-safe to return an int.
| def _estimate_expected_tasks( | |
| op_state: OpState, | |
| ) -> float: | |
| # Each task consumes `average_num_inputs_per_task` input blocks on average, | |
| # so the total expected number of tasks: | |
| # | |
| # ceil(num enqueued blocks / avg_inputs_per_task) | |
| # | |
| avg_input_blocks_per_task = op_state.op.metrics.average_num_inputs_per_task or 1 | |
| return math.ceil(op_state.total_enqueued_input_blocks() / avg_input_blocks_per_task) | |
| def _estimate_expected_tasks( | |
| op_state: OpState, | |
| ) -> int: | |
| # Each task consumes `average_num_inputs_per_task` input blocks on average, | |
| # so the total expected number of tasks: | |
| # | |
| # ceil(num enqueued blocks / avg_inputs_per_task) | |
| # | |
| avg_input_blocks_per_task = op_state.op.metrics.average_num_inputs_per_task or 1 | |
| return int(math.ceil(op_state.total_enqueued_input_blocks() / avg_input_blocks_per_task)) |
![cursor[bot]](https://avatars.githubusercontent.com/in/1210556?s=60&v=4){kind=small}
| ) | ||
|
|
||
| def _compute_upscale_delta( | ||
| self, actor_pool: AutoscalingActorPool, op_state: OpState |
There was a problem hiding this comment.
op_state is not being used
Signed-off-by: Alexey Kudinkin <ak@anyscale.com>
Signed-off-by: Alexey Kudinkin <ak@anyscale.com>
… running; Removed dead methods; Signed-off-by: Alexey Kudinkin <ak@anyscale.com>
Signed-off-by: Alexey Kudinkin <ak@anyscale.com>
Signed-off-by: Alexey Kudinkin <ak@anyscale.com>
alexeykudinkin
force-pushed
the
ak/act-ascl-clup
branch
from
365c5c1 to
3c2232f
Compare
There was a problem hiding this comment.
Cursor Bugbot has reviewed your changes and found 1 potential issue.
Bugbot Autofix is OFF. To automatically fix reported issues with cloud agents, have a team admin enable autofix in the Cursor dashboard.
| return math.ceil( | ||
| actor_pool.current_size() | ||
| * (actor_pool.get_pool_util() / self._actor_pool_scaling_up_threshold - 1) | ||
| ) |
There was a problem hiding this comment.
Unused op_state parameter in _compute_upscale_delta
Low Severity
The op_state parameter in _compute_upscale_delta is accepted but never used in the function body. The function only uses actor_pool and self._actor_pool_scaling_up_threshold. This was confirmed in the PR discussion, where the author acknowledged the issue and said they would fix it.
2 participants
Description
ActorPoolResizingPolicy_ActorPoolto compute utilization based on all actors, not just runningRelated issues
Additional information