[Data] Cap op concurrency with exponential ramp-up #40275

raulchen · 2023-10-11T22:21:05Z

Why are these changes needed?

Today when a new dataset is launched, the StreamingExecutor will allocate all resources to the first operator.
Consider a simple case read -> map where the read and map are not fused, the scheduler will submit as many read tasks as possible up front. Assuming the read tasks will output many blocks, the blocks will pile up because we don’t have resources to submit map tasks to consume the data.

This PR tries to mitigate this issue by cap the concurrency of each op with an initial value and ramp up the cap exponentially as the execution goes on (see ConcurrencyCapBackpressurePolicy docstring for details).

This PR also introduces a BackpressurePolicy interface, making the backpressure policies configurable and pluggable. Later we should migrate existing backpressure mechanisms to this new interface.

Known limitations:

If the config is not properly set, perf may regress for some workloads. Thus we disable this feature by default in 2.8, and will enable it in 2.9.
This feature only caps the initial concurrency. Once the cap has ramped up, data can still pile up. This issue will be resolved by a different backpressure policy that will profile the runtime metrics (e.g, object store increase, RAM usage) of each op.
It doesn't backpressure tasks that output many blocks. This issue will be solved by streaming generator backpressure instead.

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: Hao Chen <chenh1024@gmail.com>

c21

LG overall.

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

c21 · 2023-10-16T19:12:20Z

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

+
+    # Environment variable to configure this policy.
+    # The format is: "<init_cap>,<cap_multiply_threshold>,<cap_multiplier>"
+    CONFIG_ENV_VAR = "RAY_DATA_CONCURRENCY_CAP_CONFIG"


this is not blocking comment, but better to have separate environment variables.

This is to facilitate internal tests in 2.9. When we officially release this feature, we will probably simplify the configs. So I'll keep it for now.

c21 · 2023-10-16T19:15:33Z

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

+    # The intial concurrency cap for each operator.
+    INIT_CAP = 4
+    # When the number of finished tasks reaches this threshold, the concurrency cap
+    # will be multiplied by the multiplier.
+    CAP_MULTIPLY_THRESHOLD = 0.5
+    # The multiplier to multiply the concurrency cap by.
+    CAP_MULTIPLIER = 2.0


another way is to define these constant on the file level, so we don't need to parse environment variables. It's also maybe easier for advanced users to test out.

The issue with constants is that if the executor doesn't run on the driver (e.g., on the SplitCoordinator actor), it's hard to change the configs. I've seen this issue for other configs that depend on constants. Do you know a good solution to bypass this issue?

I think you'll need to put them into the DataContext.

@stephanie-wang @c21 If we put them into DataContext, I'd like to to save them in a dict and add a key-value interface in DataContext.
The reason is because this is a plugin and DataContext shouldn't need to know about the plugin configs.
What do you think?

The API will be something like:

data_context.set("concucrrency_cap_backpressure_policy.init_cap", 4)

That sounds good to me for now while it's still experimental. Actually, could you prepend the name with "experimental" or something like that? Makes deprecation a bit smoother.

c21 · 2023-10-16T19:18:25Z

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

+
+
+# TODO(hchen): Enable ConcurrencyCapBackpressurePolicy by default.
+DEFAULT_BACKPRESSURE_PLOCIES = []


So do we foresee in the near future, we will have multiple policies enabled at same time?

I think it is too early to abstract backpressure policies, considering we only have one right now. Can we just put the new concurrency caps inside the current backpressure code under a feature flag?

I do plan to migrate existing backpressure code to this interface.
Another reason why I wanted to introduce this interface is that it will allow us experimenting new back-pressure policies without touching the code base. E.g., one idea is to take the real runtime metrics into consideration for backpressure.

Sounds good, I don't feel strongly about it, just think that it may be too early to abstract (we don't know yet if this is the right interface).

yes, the interface is likely to change as we implement other policies. it's an internal interface, so that should be fine.

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

stephanie-wang

I'm fine merging if we turn it off by default, but I don't think this policy is going to be usable. The performance regression is scary, and the configuration parameters here are going to be pretty confusing for the average user, I think. Also, off the top of my head, there are likely many cases that won't work or will regress (e.g., concurrency=4 cap may or may not work depending on num_cpus/task and num_cpus total).

Offline, let's try to come up with a policy that minimizes regressions and has as few configuration parameters as possible, ideally none. I believe we can come up with something reasonable by considering the total number of operators when assigning the initial caps. For example, if we have N operators to run, we give each 1/N cores initially then adjust based on which operators are ready.

Also, I think this PR needs some doc changes to instruct on how to turn on the feature and configure it.

stephanie-wang · 2023-10-16T19:50:30Z

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

+
+
+# TODO(hchen): Enable ConcurrencyCapBackpressurePolicy by default.
+DEFAULT_BACKPRESSURE_PLOCIES = []


I think it is too early to abstract backpressure policies, considering we only have one right now. Can we just put the new concurrency caps inside the current backpressure code under a feature flag?

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

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

python/ray/data/tests/test_backpressure_policies.py

raulchen · 2023-10-16T23:34:12Z

@stephanie-wang You are right. The current default config isn't optimal. I also thought of making the default config smarter by taking into consideration the number of ops and their resource requirements. The plan is to implement the basic framework and turn this off in 2.8. In 2.9, we'll need to do more experiments to figure out the best configuration and officially release this feature.

Signed-off-by: Hao Chen <chenh1024@gmail.com>

raulchen · 2023-10-17T01:12:42Z

@c21 @stephanie-wang thanks for your comments. They are either addressed (resolved threads) or replied (non-resolved threads). Please take a look again.

stephanie-wang

Let's merge after making it configurable through DataContext instead of the env variables.

Signed-off-by: Hao Chen <chenh1024@gmail.com>

raulchen · 2023-10-17T19:10:36Z

Moved the config to DataContext and added the set_plugin_config/get_plugin_config API. Reason was commented here.

Signed-off-by: Hao Chen <chenh1024@gmail.com>

c21

LG w/ two minor comments.

c21 · 2023-10-17T21:52:13Z

python/ray/data/context.py

@@ -219,6 +219,7 @@ def __init__(
        self.enable_get_object_locations_for_metrics = (
            enable_get_object_locations_for_metrics
        )
+        self._plugin_configs: Dict[str, Any] = {}


can we rename it as _backpressure_plugin_configs? In the future, we may introduce other plugin components.

c21 · 2023-10-17T21:53:22Z

python/ray/data/context.py

+    def get_plugin_config(self, key: str, default: Any = None) -> Any:
+        return self._plugin_configs.get(key, default)
+
+    def set_plugin_config(self, key: str, value: Any) -> None:
+        self._plugin_configs[key] = value
+
+    def remove_plugin_config(self, key: str) -> None:
+        self._plugin_configs.pop(key, None)


let's declare these methods with _ prefix, so they remain private and easy to change later.

Discussed offline, we'll make this API reusable for other components.

Signed-off-by: Hao Chen <chenh1024@gmail.com>

Today when a new dataset is launched, the StreamingExecutor will allocate all resources to the first operator. Consider a simple case `read -> map` where the read and map are not fused, the scheduler will submit as many read tasks as possible up front. Assuming the read tasks will output many blocks, the blocks will pile up because we don’t have resources to submit map tasks to consume the data. This PR tries to mitigate this issue by cap the concurrency of each op with an initial value and ramp up the cap exponentially as the execution goes on (see `ConcurrencyCapBackpressurePolicy` docstring for details). This PR also introduces a `BackpressurePolicy` interface, making the backpressure policies configurable and pluggable. Later we should migrate existing backpressure mechanisms to this new interface. Known limitations: - If the config is not properly set, perf may regress for some workloads. Thus we disable this feature by default in 2.8, and will enable it in 2.9. - This feature only caps the initial concurrency. Once the cap has ramped up, data can still pile up. This issue will be resolved by a different backpressure policy that will profile the runtime metrics (e.g, object store increase, RAM usage) of each op. - It doesn't backpressure tasks that output many blocks. This issue will be solved by streaming generator backpressure instead. --------- Signed-off-by: Hao Chen <chenh1024@gmail.com>

raulchen requested review from ericl, scv119, c21, amogkam, scottjlee, bveeramani and stephanie-wang as code owners October 11, 2023 22:21

raulchen force-pushed the concurrency-cap-backpressure branch from 514a0af to 4ff0a9c Compare October 13, 2023 19:01

raulchen added 9 commits October 16, 2023 10:57

back pressure policy

2d580a7

Signed-off-by: Hao Chen <chenh1024@gmail.com>

refine

ed288f3

Signed-off-by: Hao Chen <chenh1024@gmail.com>

refine

78d2b10

Signed-off-by: Hao Chen <chenh1024@gmail.com>

fix

b699382

Signed-off-by: Hao Chen <chenh1024@gmail.com>

fix

c08f1b6

Signed-off-by: Hao Chen <chenh1024@gmail.com>

disable

647128a

Signed-off-by: Hao Chen <chenh1024@gmail.com>

refine

a517c30

Signed-off-by: Hao Chen <chenh1024@gmail.com>

add test

37913b0

Signed-off-by: Hao Chen <chenh1024@gmail.com>

rename

fc14b94

Signed-off-by: Hao Chen <chenh1024@gmail.com>

raulchen force-pushed the concurrency-cap-backpressure branch from 4ff0a9c to fc14b94 Compare October 16, 2023 19:05

lint

600f197

Signed-off-by: Hao Chen <chenh1024@gmail.com>

raulchen assigned ericl, stephanie-wang and c21 Oct 16, 2023

c21 reviewed Oct 16, 2023

View reviewed changes

stephanie-wang reviewed Oct 16, 2023

View reviewed changes

python/ray/data/tests/test_backpressure_policies.py Outdated Show resolved Hide resolved

stephanie-wang added the @author-action-required The PR author is responsible for the next step. Remove tag to send back to the reviewer. label Oct 16, 2023

raulchen mentioned this pull request Oct 17, 2023

[data] implement streaming output backpressure #40387

Merged

8 tasks

address comments

061faf8

Signed-off-by: Hao Chen <chenh1024@gmail.com>

stephanie-wang approved these changes Oct 17, 2023

View reviewed changes

raulchen added 2 commits October 17, 2023 12:06

config and e2e test

1a35d2d

Signed-off-by: Hao Chen <chenh1024@gmail.com>

lint

527606d

Signed-off-by: Hao Chen <chenh1024@gmail.com>

raulchen added 3 commits October 17, 2023 13:00

fix

681b589

Signed-off-by: Hao Chen <chenh1024@gmail.com>

fix test

ec46dd1

Signed-off-by: Hao Chen <chenh1024@gmail.com>

Merge branch 'master' into concurrency-cap-backpressure

26e59b4

c21 approved these changes Oct 17, 2023

View reviewed changes

raulchen added 3 commits October 17, 2023 16:39

refine

04e4e5a

Signed-off-by: Hao Chen <chenh1024@gmail.com>

lint

9443aeb

Signed-off-by: Hao Chen <chenh1024@gmail.com>

fix

ee0dd39

Signed-off-by: Hao Chen <chenh1024@gmail.com>

raulchen merged commit 6ba659f into ray-project:master Oct 18, 2023
26 of 37 checks passed

raulchen deleted the concurrency-cap-backpressure branch October 18, 2023 03:02

rickyyx mentioned this pull request Oct 26, 2023

Release test long_running_many_actor_tasks failed #40568

Closed

raulchen mentioned this pull request Oct 27, 2023

[Data] Streaming executor backpressure #40754

Closed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[Data] Cap op concurrency with exponential ramp-up #40275

[Data] Cap op concurrency with exponential ramp-up #40275

raulchen commented Oct 11, 2023 •

edited

Loading

c21 left a comment

c21 Oct 16, 2023

raulchen Oct 17, 2023

c21 Oct 16, 2023

raulchen Oct 17, 2023

stephanie-wang Oct 17, 2023

raulchen Oct 17, 2023

stephanie-wang Oct 17, 2023

c21 Oct 16, 2023

stephanie-wang Oct 16, 2023

raulchen Oct 17, 2023

stephanie-wang Oct 17, 2023

raulchen Oct 17, 2023

stephanie-wang left a comment

stephanie-wang Oct 16, 2023

raulchen commented Oct 16, 2023

raulchen commented Oct 17, 2023

stephanie-wang left a comment

raulchen commented Oct 17, 2023

c21 left a comment

c21 Oct 17, 2023

c21 Oct 17, 2023

raulchen Oct 17, 2023



		# TODO(hchen): Enable ConcurrencyCapBackpressurePolicy by default.
		DEFAULT_BACKPRESSURE_PLOCIES = []

[Data] Cap op concurrency with exponential ramp-up #40275

[Data] Cap op concurrency with exponential ramp-up #40275

Conversation

raulchen commented Oct 11, 2023 • edited Loading

Why are these changes needed?

Related issue number

Checks

c21 left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

stephanie-wang left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

raulchen commented Oct 16, 2023

raulchen commented Oct 17, 2023

stephanie-wang left a comment

Choose a reason for hiding this comment

raulchen commented Oct 17, 2023

c21 left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

raulchen commented Oct 11, 2023 •

edited

Loading