[core][compiled graph] Support all-to-one collective ops (e.g. gather)

[jeffreyjeffreywang]

Description

As part of the effort (meta-issue: #47983) to support collective communication ops, we need to support all-to-one (gather) patterns. As discussed in the RFC, we will pass the reader worker handle to the collective call.

workers = [Worker.options(num_gpus=1).remote() for _ in range(3)]
nccl_group_handle = ray.collective.NcclGroup(workers)
with InputNode() as inp:
  results = [worker.fwd.bind(inp) for worker in workers]
  # Pass the worker handle to the collective call.
  dag = ray.collective.gather.bind(
    results, workers[0],
    transport=nccl_group_handle)
  dag = workers[0].sync.bind(dag)

# Errors if `gather` reader is not part of the group.
dag = dag.experimental_compile()

Use case

No response

[jeffreyjeffreywang]

I'm getting a head start on this issue.

[Moonquakes]

Maybe we can apply RDMA directly to all Ray Object transmissions? #30094