[core][experimental] Correct num_input_consumers for CachedChannel

[kevin85421]

Why are these changes needed?

Problem statement

Without this PR, the num_input_consumers would be 1 because both inp[0] and inp[1] are only referred to in one task on the actor, so CachedChannel will not be created. The read will eventually time out because the mutable object is being read by the same actor twice.

def test_actor_method_bind_diff_input_attr_1(ray_start_regular):
    actor = Actor.remote(0)
    with InputNode() as inp:
        # Two class methods are bound to two different input
        # attribute nodes.
        output1 = actor.inc.bind(inp[0])
        output2 = actor.inc.bind(inp[1])
        dag = MultiOutputNode([output1, output2])
    compiled_dag = dag.experimental_compile()
    ref = compiled_dag.execute(0, 1)
    assert ray.get(ref) == [0, 1]

    ref = compiled_dag.execute(1, 2)
    assert ray.get(ref) == [2, 4]

    compiled_dag.teardown()

Solution

If a task is bound to either an InputNode or an InputAttributeNode, increment num_input_consumers by 1. Therefore, num_input_consumers will be 2, and a CachedChannel will be created.

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

[ruisearch42]

Thanks for the fix!

Need to change title?

[kevin85421]

Need to change title?

How about "Correct num_input_consumers for CachedChannel"?

[rkooo567]

ready to be merged @kevin85421 ?

[ruisearch42]

LGTM when tests pass!

[kevin85421]

ready to be merged @kevin85421 ?

It will be ready to be merged after tests pass.

[rkooo567]

premerge failures

[kevin85421]

Retry the unrelated RLlib failures.