[core][cgraph] Rework DagRef Destruction

[dayshah]

Why are these changes needed?

There are two different approaches we can take when deserializing dagrefs.

Approach #1 is taken here #49781 and is much simpler.
The downside however is that when a dagref is destructed it requires the execution and caching of all previous executions. We have no guarantee Python desructs in order, so if there are multiple dagrefs being destructed it becomes possible that we're still deserializing and caching them even if we don't need to. We are also forced to execute up to the dagref that is being destructed even if the user hasn't called get on previous dagrefs yet.

Approach #2 is taken in this pr, and here we hold a destructed_ref_idxs dict which is the execution_idxs -> set of channel_idxs of destructed CompiledDagRefs. Then, we release the buffer whenever the max_finished_execution_index is 1 less than any destructed execution index with a complete set of channel_idxs. There's three places we check for this, during destruction, during execute, and during get (in the loop of execute_until). The upside is that the destruction of the dagrefs will never require the executing any previous dagrefs. Execution for the destructed dagrefs will only happen if the max_finished_execution_index has reached the index before the destructed dagref. The downside here is the complexity of having the logic of checking if buffers can be released at 3 separate places.

Related issue number

Closes #49782

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

[ruisearch42]

"(except for refs that have been destructed)" is hard for the user to understand, elaborate a bit more?

[dayshah]

I updated a bit lmk if it's clearer now

[ruisearch42]

will have another pass to see if this is comprehensive enough

[ruisearch42]

explain why we are doing this rather than what the code does, sth like "check if native buffers corresponding to destructed CompiledDAGRefs are ready to be released, and release as many as possible"

[dayshah]

The main doc of the function describes what this does so removed this comment

[ruisearch42]

Can you please also write down in the description the pros and cons for the 3 approaches and why this is preferred so that we don't lose context?

[ruisearch42]

If a prior execute() returns two refs, ref1 and ref2, and ref1 has gone out of scope, and ray.get(ref2) is called, this will cache the value for both ref1 and ref2, and then pop only value-for-ref2 but leaves value-for-ref1, which is leaked?

For this to work, I think you will need to make _destructed_execution_idxs a map from int to set of int. i.e., the value is the set of channel indexes.
Can you add a test case?

[dayshah]

added test case and logic to handle inside cache, so we never cache if the ref for that channel idx has been destructed

[ruisearch42]

Otherwise LGTM

[kevin85421]

reviewing

[kevin85421]

maybe in this function we should not only call release_channel_buffers but also check the cached results and clean it up if needed.

[dayshah]

I think they don't have to always be called at the same time.

We should never cache if already destructed because we know ahead of time. So the only times we should ever remove from the cache is during __del__ or get

[kevin85421]

It’s just a safeguard. It’s fine if you prefer not to add it.

[ruisearch42]

This is not consistent with test name? or maybe I misunderstood the test name

[dayshah]

i had trouble coming up with the names for these tests lol, but the idea is that we're getting the ref that was executed first, and destructing the second ref. so the ref we're getting was made before the ref we destructed. Agree the test_get_ref_before_destructed_ref name isn't easy to understand, but can't think of a good one for these

[kevin85421]

Others LGTM

[kevin85421]

For the tests, could you use different inputs for different execute calls so that we can ensure get retrieves the expected execution index?

[dayshah]

I think it shouldn't matter because each call is incrementing the counter so will still be different result for each execution idx.
For first example, ref is 1 and ref2 is 2

[kevin85421]

test_basic_destruction uses a.echo.bind(i). Maybe it's the only one test that doesn't use inc.