[core][adag] Support node having readers as driver and the actor it was generated from
What happened + What you expected to happen
This is separate from https://github.com/ray-project/ray/issues/47156, because it seems that it needs more to work vs. just the work for supporting reading from a node and a driver. Experimented with changes to support reading from driver and another node from https://github.com/ray-project/ray/pull/48028. It seems that if the actor that produced a node, has an operation that depends on that node, and that node is read by the driver, we fail. The failure occurs because num_read_acquires_remaining in PlasmaObjectHeader hits 0 for x while there's more reads actually awaiting. Suspect that something is messing with the number of readers that the channels pass to the cpp code specifically in this situation.
Versions / Dependencies
3.0dev
Reproduction script
with ray.dag.InputNode() as inp:
x = sqA.square.bind(inp)
y = sqA.square.bind(x)
dag = ray.dag.MultiOutputNode([x, y])
dag = dag.experimental_compile()
result = dag.execute(2)
print(ray.get(result))
Issue Severity
None