[core][aDAG] Non-dag output cannot be an input of other dag

[rkooo567]

What happened + What you expected to happen

not sure if it is intended. But we cannot pass non-dag-value to be an input of other dag.

Versions / Dependencies

master

Reproduction script

import ray
ray.init()

from ray.dag import InputNode

@ray.remote
class A:
    def g(self, i):
        return 1

    def f(self, *a):
        print(a)

a = A.remote()
actors = [A.remote() for _ in range(4)]

with InputNode() as inp:
    activations = []
    for i, actor in enumerate(actors):
        activations.append((actor.g.bind(inp), i))
    dag = a.f.bind(*activations)
adag = dag.experimental_compile()
print(ray.get(adag.execute(1)))

Issue Severity

None

[stephanie-wang]

@rkooo567 can you elaborate? I'm not sure from the example what you mean. Which line in the example shows the issue?

[rkooo567]

@stephanie-wang it looks like if we pass a tuple as an argument everything other than dag ref is ignored (in this case i, activations.append((actor.g.bind(inp), i)))

[stephanie-wang]

Oh got it, thanks! Hmm should we just disallow passing DAG ref inside other data structures for now? Otherwise I guess the other option is to serialize with placeholders for the DAG refs and then deserialize for each DAG invocation.