[core][aDAG] asyncio run hangs upon shutdown

[rkooo567]

What happened + What you expected to happen

When I ran the script below, it hangs. I don't know the root cause yet. Probably related to shutdown

Versions / Dependencies

master

Reproduction script

import asyncio
import ray
from ray.dag import InputNode, MultiOutputNode

async def main():
    @ray.remote
    class A:
        def f(self, i):
            return i

    a = A.remote()
    b = A.remote()

    with InputNode() as inp:
        x = a.f.bind(inp)
        y = b.f.bind(inp)
        dag =  MultiOutputNode([x, y])

    adag = dag.experimental_compile(enable_asyncio=True)
    refs = await adag.execute_async(1)
    outputs = []
    # works
    for ref in refs:
        outputs.append(await ref)
    # doesn't work.
    # outputs = await asyncio.gather(*refs)
    print(outputs)

asyncio.run(main())

Issue Severity

None

[jeffreyjeffreywang]

It seems like the program doesn't hang anymore if adag.teardown() is called after print(outputs).

[jeffreyjeffreywang]

Perhaps we can tear down the aDAG implicitly when it goes out of scope which will be more intuitive to the client.

[rkooo567]

actually the teardown is called on the destructor of CompiledDag class. Idk why it is not properly triggered..

[rkooo567]

Potential fix here; #47685 (comment) waiting for CI to see if it passes all tests