[Datasets] Provide more efficient + intuitive block clearing semantics for different execution modes #24127

clarkzinzow · 2022-04-22T23:40:49Z

TL;DR: Don't clear for eager, clear all but non-lazy input blocks if lazy, clear everything if pipelining.

This PR provides more efficient and intuitive block clearing semantics for eager mode, lazy mode, and pipelining, while still supporting multiple operations applied to the same base dataset, i.e. fan-out. For example, two different map operations are applied to the same base ds in this example:

ds = ray.data.range(10).map(lambda x: x+1)
ds1 = ds.map(lambda x: 2*x)
ds2 = ds.map(lambda x: 3*x)

If naively clear the blocks when executing the map to produce ds1, the map producing ds2 will fail.

Desired Semantics

Eager mode - don’t clear input blocks, thereby supporting fan-out from cached data at any point in the stage chain without triggering unexpected recomputation.
Lazy mode - if lazy datasource, clear the input blocks for every stage, relying on recomputing via stage lineage if fan-out occurs; if non-lazy datasource, do not clear source blocks for execution plan when executing first stage, but do clear input blocks for every subsequent stage.
Pipelines - Same as lazy mode, although the only fan-out that can occur is from the pipeline source blocks when repeating a dataset/pipeline, so unintended intermediate recomputation will never happen.

Related issue number

Closes #18791

Checks

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 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 :(

ericl · 2022-04-23T01:01:37Z

python/ray/data/impl/plan.py

-        self, clear_input_blocks: bool = True, force_read: bool = False
+        self,
+        clear_input_blocks: bool = True,
+        destroy_unrecoverable_input_blocks: bool = False,


I like the name of this argument!

ericl

LGTM at a high level. @jianoaix can you review?

jjyao · 2022-04-24T22:17:28Z

What's fan-out here?

clarkzinzow · 2022-04-24T23:09:13Z

@jjyao By fan-out, we mean more than one dataset operation applied to a single base dataset, e.g. two different map operations applied to the same base ds in this example:

ds = ray.data.range(10).map(lambda x: x+1)
ds1 = ds.map(lambda x: 2*x)
ds2 = ds.map(lambda x: 3*x)

I'll add this to the PR description!

jianoaix · 2022-04-24T23:16:37Z

Actually, with fan-out, I think there is an issue with execution plan, which will be a DAG, rather than a chain.

For example, continue the code snippet, we can have a fan-in: ds3 = ds1.union(ds2). Now the lineage of ds3 is a DAG. And user can continue such pattern to construct complex DAG.

clarkzinzow · 2022-04-25T00:20:19Z

@jianoaix Agreed that an explicit DAG representation would be more ideal (that's the future plan), but what do you see as the concrete current issue with fan-out here? Fan-out is supported in eager mode via reusing the base dataset's snapshot blocks, and is supported in lazy mode via recomputation from source, so fan-out should be covered. I.e., the stage DAG for fan-out of lazy datasets is currently supported by independent chains for each dataset sink, with the chain going back to the shared source dataset.

For example, continue the code snippet, we can have a fan-in: ds3 = ds1.union(ds2). Now the lineage of ds3 is a DAG. And user can continue such pattern to construct complex DAG.

Right now, lazy fan-in isn't supported: all fan-in operations trigger computation for lazy datasets. One thing that's currently untested is lazy fan-out after a union. Upon a quick eye-test, it looks like it might not work if one or more of the unioned datasets are created from non-lazy datasources, since recomputing from the base won't work on the dummy read tasks.

jianoaix · 2022-04-25T03:49:07Z

With the lineage/execution plan being used for both eager and lazy dataset (e.g. the other PR claimed the eager dataset can also serialize the lineage), isn't this already a problem for both cases using linear data structure for DAG?

clarkzinzow · 2022-04-26T00:42:28Z

Synced offline, opened this PR to make it explicit that lineage-based serialization of unions/zips are not supported.

python/ray/data/impl/plan.py

python/ray/data/impl/pipeline_executor.py

python/ray/data/impl/plan.py

python/ray/data/tests/test_optimize.py

…lear everything if pipelining.

…ack.

clarkzinzow · 2022-04-29T23:23:45Z

python/ray/data/tests/test_optimize.py

    meminfo = memory_summary(info["address"], stats_only=True)
-    # TODO(ekl) we can add this back once we clear inputs on eager exec as well.
-    # assert "Spilled" not in meminfo, meminfo
+    assert "Spilled" not in meminfo, meminfo


This test might prove to be flaky due to the memory reclaiming issues within Ray, so we may need to put this in a retry loop similar to test_memory_release_lazy_shuffle.

Ack. Even 2s not enough to escape the reclaiming uncertainty?

1s was a bit flaky, didn't see 2s fail while testing locally, but who knows in CI. For context, we use an "OOM grace period" of 2 seconds in core Ray, where we let reference counting + garbage collection and object spilling free up space, so I'd expect this to be ~enough.

jianoaix

LGTM, just a few minor comments

jianoaix · 2022-04-30T00:00:37Z

python/ray/data/dataset.py

        """Force full evaluation of the blocks of this dataset.

        This can be used to read all blocks into memory. By default, Datasets
        doesn't read blocks from the datasource until the first transform.

+        Args:
+            preserve_original: Whether the original unexecuted dataset should be


What is "original dataset" for a dataset? (I think users may have same question if they read this API)

Ah I thought that when calling d2 = ds.fully_executed(preserve_original=False), original referring to ds would be obvious. 🤔

I see. I was having a chain of operations in mind, in that case it's unclear which one of them should be thought as "original" (maybe the first one? maybe each intermediate ones? or others?).

python/ray/data/impl/plan.py

jianoaix · 2022-04-30T00:30:15Z

python/ray/data/tests/test_optimize.py

+        import time
+
+        # TODO(Clark): Remove this sleep once we have fixed memory pressure handling.
+        time.sleep(2)


File a bug to track this issue?

Yep can do!

jianoaix · 2022-04-30T00:37:01Z

python/ray/data/tests/test_optimize.py

    meminfo = memory_summary(info["address"], stats_only=True)
-    # TODO(ekl) we can add this back once we clear inputs on eager exec as well.
-    # assert "Spilled" not in meminfo, meminfo
+    assert "Spilled" not in meminfo, meminfo


Ack. Even 2s not enough to escape the reclaiming uncertainty?

clarkzinzow requested review from ericl, scv119 and jjyao as code owners April 22, 2022 23:40

ericl reviewed Apr 23, 2022

View reviewed changes

ericl assigned jianoaix Apr 23, 2022

clarkzinzow force-pushed the datasets/feat/block-clearing branch from 3920ff9 to 4935ad3 Compare April 23, 2022 01:40

jianoaix reviewed Apr 26, 2022

View reviewed changes

Don't clear for eager, clear all but non-lazy input blocks if lazy, c…

3a4ce26

…lear everything if pipelining.

clarkzinzow force-pushed the datasets/feat/block-clearing branch 2 times, most recently from 37d9297 to 7473238 Compare April 29, 2022 21:13

Ensure aggressive memory reclamation while pipelining; misc. PR feedb…

e68a5e9

…ack.

clarkzinzow force-pushed the datasets/feat/block-clearing branch from 7473238 to e68a5e9 Compare April 29, 2022 23:21

clarkzinzow commented Apr 29, 2022

View reviewed changes

jianoaix reviewed Apr 30, 2022

View reviewed changes

jianoaix approved these changes Apr 30, 2022

View reviewed changes

ericl approved these changes Apr 30, 2022

View reviewed changes

ericl merged commit f725552 into ray-project:master Apr 30, 2022

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[Datasets] Provide more efficient + intuitive block clearing semantics for different execution modes #24127

[Datasets] Provide more efficient + intuitive block clearing semantics for different execution modes #24127

clarkzinzow commented Apr 22, 2022 •

edited

Loading

ericl Apr 23, 2022

ericl left a comment

jjyao commented Apr 24, 2022

clarkzinzow commented Apr 24, 2022

jianoaix commented Apr 24, 2022

clarkzinzow commented Apr 25, 2022

jianoaix commented Apr 25, 2022

clarkzinzow commented Apr 26, 2022

clarkzinzow Apr 29, 2022

jianoaix Apr 30, 2022

clarkzinzow Apr 30, 2022

jianoaix left a comment

jianoaix Apr 30, 2022

clarkzinzow Apr 30, 2022

jianoaix Apr 30, 2022

jianoaix Apr 30, 2022

clarkzinzow Apr 30, 2022

jianoaix Apr 30, 2022

[Datasets] Provide more efficient + intuitive block clearing semantics for different execution modes #24127

[Datasets] Provide more efficient + intuitive block clearing semantics for different execution modes #24127

Conversation

clarkzinzow commented Apr 22, 2022 • edited Loading

Desired Semantics

Related issue number

Checks

Choose a reason for hiding this comment

ericl left a comment

Choose a reason for hiding this comment

jjyao commented Apr 24, 2022

clarkzinzow commented Apr 24, 2022

jianoaix commented Apr 24, 2022

clarkzinzow commented Apr 25, 2022

jianoaix commented Apr 25, 2022

clarkzinzow commented Apr 26, 2022

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

jianoaix 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

clarkzinzow commented Apr 22, 2022 •

edited

Loading