[Data] Add `PhysicalOperator.min_max_resource_usage_bounds` #52502

bveeramani · 2025-04-21T17:18:43Z

Why are these changes needed?

This change is necessary to ensure we don't over-reserve resources for operators. For example:

If an actor-pool only uses GPU resources, we don't need to reserve CPU resources for it.
If a task-pool has finished receiving inputs and launching tasks, we don't ned to reserve more resources than required for the currently active tasks.

Related issue number

Checks

I've signed off every commit(by using the -s flag, i.e., git commit -s) in this PR.
I've run scripts/format.sh to lint the changes in this PR.
I've included any doc changes needed for https://docs.ray.io/en/master/.
- I've added any new APIs to the API Reference. For example, if I added a
  method in Tune, I've added it in doc/source/tune/api/ under the
  corresponding .rst file.
I've made sure the tests are passing. Note that there might be a few flaky tests, see the recent failures at https://flakey-tests.ray.io/
Testing Strategy
- Unit tests
- Release tests
- This PR is not tested :(

python/ray/data/_internal/execution/interfaces/physical_operator.py

alexeykudinkin · 2025-04-22T04:52:51Z

python/ray/data/_internal/execution/interfaces/physical_operator.py

+    def min_max_resource_usage_bounds(
+        self,
+    ) -> Tuple[ExecutionResources, ExecutionResources]:
+        """Returns the min and max resources to start the operator and make progress.


Let's expand that these are derived from operator's concurrency configuration multiplied by single task/actor resource requirements.

alexeykudinkin · 2025-04-22T04:53:37Z

python/ray/data/_internal/execution/interfaces/physical_operator.py


-    def base_resource_usage(self) -> ExecutionResources:
-        """Returns the minimum amount of resources required for execution.
+    def min_max_resource_usage_bounds(


Sorry for the back and forth, but on a second thought i've realized that min_max_resource_requirements might be a better fit here (will defer to you to decide)

min_max_resource_requirements sounds slightly better to me as well

alexeykudinkin · 2025-04-22T04:54:30Z

python/ray/data/_internal/execution/interfaces/physical_operator.py

-        """Returns the minimum amount of resources required for execution.
+    def min_max_resource_usage_bounds(
+        self,
+    ) -> Tuple[ExecutionResources, ExecutionResources]:


Should be optional

How would we handle None in the resource manager? Would it be equivalent to the default implementation?

Yeah, it'd default to not knowing resource reqs

Ok, i see that you're defaulting to [0, inf) that's fine too

python/ray/data/_internal/execution/operators/actor_pool_map_operator.py

alexeykudinkin · 2025-04-22T05:13:04Z

python/ray/data/_internal/execution/operators/map_operator.py

+    def min_max_resource_usage_bounds(
+        self,
+    ) -> Tuple[ExecutionResources, ExecutionResources]:
        raise NotImplementedError


Let's make sure there's a default implementation (so that it doesn't break any custom operators)

this is the map op. there will only be 2 sub classes: task/actor pool map ops.

See my other comment:

#52502 (comment)

Responded in the thread for #52502 (comment)

alexeykudinkin · 2025-04-22T05:14:24Z

python/ray/data/_internal/execution/operators/task_pool_map_operator.py

+    def _min_resource_usage(self) -> ExecutionResources:
+        # Make sure the reserved resources are at least to allow one task.
+        return self.incremental_resource_usage()
+
+    def _max_resource_usage(self) -> ExecutionResources:
+        if self._inputs_complete:
+            # If the operator has already received all input data, we know it won't
+            # launch more tasks. So, we only need to reserve resources for the tasks
+            # that are currently running.
+            num_cpus_per_task = self._ray_remote_args.get("num_cpus", 0)
+            num_gpus_per_task = self._ray_remote_args.get("num_gpus", 0)
+            object_store_memory_per_task = (
+                self._metrics.obj_store_mem_max_pending_output_per_task or 0
+            )
+            resources = ExecutionResources.for_limits(
+                cpu=num_cpus_per_task * self.num_active_tasks(),
+                gpu=num_gpus_per_task * self.num_active_tasks(),
+                object_store_memory=object_store_memory_per_task
+                * self.num_active_tasks(),
+            )
+        else:
+            resources = ExecutionResources.for_limits()
+
+        return resources


Let's generalize this to be base method in PhysicalOperator

I don't think this is a sensible default for PhysicalOperator. The implementation makes sense for TaskPoolOperator, but I don't think we can assume it makes sense for all PhysicalOperator subclasses (e.g., the current all-to-all operator implementation)

alexeykudinkin · 2025-04-22T05:15:31Z

python/ray/data/_internal/execution/resource_manager.py

+            min_resource_usage, max_resource_usage = op.min_max_resource_usage_bounds()
+            reserved_for_tasks = default_reserved.subtract(reserved_for_outputs)
+            reserved_for_tasks = reserved_for_tasks.max(min_resource_usage)
+            reserved_for_tasks = reserved_for_tasks.min(max_resource_usage)


This is much, much cleaner!

raulchen · 2025-04-22T18:00:18Z

python/ray/data/_internal/execution/interfaces/physical_operator.py


-    def base_resource_usage(self) -> ExecutionResources:
-        """Returns the minimum amount of resources required for execution.
+    def min_max_resource_usage_bounds(


min_max_resource_requirements sounds slightly better to me as well

python/ray/data/_internal/execution/operators/actor_pool_map_operator.py

raulchen · 2025-04-22T18:11:42Z

python/ray/data/_internal/execution/operators/map_operator.py

+    def min_max_resource_usage_bounds(
+        self,
+    ) -> Tuple[ExecutionResources, ExecutionResources]:
        raise NotImplementedError


this is the map op. there will only be 2 sub classes: task/actor pool map ops.

python/ray/data/_internal/execution/operators/task_pool_map_operator.py

python/ray/data/_internal/execution/operators/actor_pool_map_operator.py

python/ray/data/_internal/execution/operators/task_pool_map_operator.py

python/ray/data/_internal/execution/interfaces/physical_operator.py

python/ray/data/_internal/execution/operators/actor_pool_map_operator.py

Signed-off-by: Balaji Veeramani <[email protected]> Add limit Signed-off-by: Balaji Veeramani <[email protected]> Update files Signed-off-by: Balaji Veeramani <[email protected]> Appease lint Signed-off-by: Balaji Veeramani <[email protected]> Fix test Signed-off-by: Balaji Veeramani <[email protected]> Address review comments Signed-off-by: Balaji Veeramani <[email protected]> Address review comments Signed-off-by: Balaji Veeramani <[email protected]> Update stuff Signed-off-by: Balaji Veeramani <[email protected]>

Signed-off-by: Balaji Veeramani <[email protected]>

raulchen · 2025-04-30T17:47:24Z

python/ray/data/_internal/execution/operators/actor_pool_map_operator.py

+            * min_actors,
        )

+        return min_resource_usage, ExecutionResources.for_limits()


Suggested change

return min_resource_usage, ExecutionResources.for_limits()

return min_resource_usage, ExecutionResources.inf()

alexeykudinkin · 2025-04-30T18:40:18Z

python/ray/data/_internal/execution/interfaces/physical_operator.py

-        """Returns the minimum amount of resources required for execution.
+    def min_max_resource_usage_bounds(
+        self,
+    ) -> Tuple[ExecutionResources, ExecutionResources]:


Yeah, it'd default to not knowing resource reqs

alexeykudinkin · 2025-04-30T18:40:53Z

python/ray/data/_internal/execution/interfaces/physical_operator.py

-        """Returns the minimum amount of resources required for execution.
+    def min_max_resource_usage_bounds(
+        self,
+    ) -> Tuple[ExecutionResources, ExecutionResources]:


Ok, i see that you're defaulting to [0, inf) that's fine too

python/ray/data/_internal/execution/operators/actor_pool_map_operator.py

python/ray/data/_internal/execution/interfaces/physical_operator.py

python/ray/data/_internal/execution/operators/map_operator.py

Signed-off-by: Balaji Veeramani <[email protected]>

## Why are these changes needed?  This change is necessary to ensure we don't over-reserve resources for operators. For example: * If an actor-pool only uses GPU resources, we don't need to reserve CPU resources for it. * If a task-pool has finished receiving inputs and launching tasks, we don't ned to reserve more resources than required for the currently active tasks. ## Related issue number  ## Checks - [ ] I've signed off every commit(by using the -s flag, i.e., `git commit -s`) in this PR. - [ ] I've run `scripts/format.sh` to lint the changes in this PR. - [ ] I've included any doc changes needed for https://docs.ray.io/en/master/. - [ ] I've added any new APIs to the API Reference. For example, if I added a method in Tune, I've added it in `doc/source/tune/api/` under the corresponding `.rst` file. - [ ] I've made sure the tests are passing. Note that there might be a few flaky tests, see the recent failures at https://flakey-tests.ray.io/ - Testing Strategy - [ ] Unit tests - [ ] Release tests - [ ] This PR is not tested :( --------- Signed-off-by: Balaji Veeramani <[email protected]> Signed-off-by: jhsu <[email protected]>

#52502 removed the cache in `update_reservation`. now the low-resource warning will spam the console when cluster resources are low. Add a `log_once` check to fix this. --------- Signed-off-by: Hao Chen <[email protected]>

bveeramani requested a review from a team as a code owner April 21, 2025 17:18

bveeramani marked this pull request as draft April 21, 2025 17:18

alexeykudinkin reviewed Apr 21, 2025

View reviewed changes

python/ray/data/_internal/execution/interfaces/physical_operator.py Outdated Show resolved Hide resolved

bveeramani changed the title ~~[Data] Add PhysicalOperator.max_resource_usage~~ [Data] Add PhysicalOperator.min_max_resource_usage_bounds Apr 22, 2025

bveeramani assigned raulchen Apr 22, 2025

bveeramani marked this pull request as ready for review April 22, 2025 03:02

bveeramani added the go add ONLY when ready to merge, run all tests label Apr 22, 2025

alexeykudinkin reviewed Apr 22, 2025

View reviewed changes

raulchen reviewed Apr 22, 2025

View reviewed changes

raulchen reviewed Apr 23, 2025

View reviewed changes

python/ray/data/_internal/execution/interfaces/physical_operator.py Outdated Show resolved Hide resolved

python/ray/data/_internal/execution/operators/actor_pool_map_operator.py Outdated Show resolved Hide resolved

bveeramani force-pushed the max-resource-usage2 branch from 0d4b23c to 8edacd4 Compare April 29, 2025 03:30

bveeramani force-pushed the max-resource-usage2 branch from 6e9754f to 0babe2d Compare April 29, 2025 04:07

bveeramani added 3 commits April 30, 2025 06:46

Update tests

c7819ce

Signed-off-by: Balaji Veeramani <[email protected]>

Merge branch 'master' into max-resource-usage2

88bffd7

Signed-off-by: Balaji Veeramani <[email protected]>

Merge branch 'master' into max-resource-usage2

6270c19

Signed-off-by: Balaji Veeramani <[email protected]>

raulchen approved these changes Apr 30, 2025

View reviewed changes

alexeykudinkin reviewed Apr 30, 2025

View reviewed changes

bveeramani added 4 commits April 30, 2025 13:38

Fix tests

528110b

Signed-off-by: Balaji Veeramani <[email protected]>

Fix tests

d6df3e1

Signed-off-by: Balaji Veeramani <[email protected]>

Address review comments

2297450

Signed-off-by: Balaji Veeramani <[email protected]>

Address review comments

9118e4f

Signed-off-by: Balaji Veeramani <[email protected]>

bveeramani enabled auto-merge (squash) May 1, 2025 14:35

Fix edge case

5b06704

Signed-off-by: Balaji Veeramani <[email protected]>

github-actions bot disabled auto-merge May 1, 2025 15:50

bveeramani enabled auto-merge (squash) May 1, 2025 16:32

bveeramani merged commit c664886 into master May 1, 2025
6 checks passed

bveeramani deleted the max-resource-usage2 branch May 1, 2025 17:03

raulchen mentioned this pull request May 13, 2025

[data] limit low-resource warning log #52973

Merged

hainesmichaelc added the community-backlog label May 22, 2025

	return min_resource_usage, ExecutionResources.for_limits()
	return min_resource_usage, ExecutionResources.inf()

[Data] Add PhysicalOperator.min_max_resource_usage_bounds #52502

[Data] Add PhysicalOperator.min_max_resource_usage_bounds #52502

Uh oh!

Conversation

bveeramani commented Apr 21, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Why are these changes needed?

Related issue number

Checks

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

5 participants

[Data] Add `PhysicalOperator.min_max_resource_usage_bounds` #52502

[Data] Add `PhysicalOperator.min_max_resource_usage_bounds` #52502

bveeramani commented Apr 21, 2025 •

edited

Loading