…cator=comm) (ray-project#50023)

Currently, `with_tensor_transport(transport=arg)` supports 3 types of
arg: "nccl", "auto", or a `Communicator`. If "nccl" is used, or "auto"
is used and resolves to using NCCL, Ray creates a communicator
internally and uses it.

In order to better manage the communicator lifecycle, we make the
following changes:

- Add API to support
`experimental_compile(_default_communicator=default_comm)`, default_comm
is a default communicator user provided
- `_default_communicator` in the above API can be "create", which
explicitly tells Ray to create a default communicator and use it for all
`with_tensor_transport()` sites where a specific communicator was not
provided
- If `_default_communicator` is not specified, Ray will not create a
default communicator or reuse communicators passed in from other
`with_tensor_transport()` sites, but throw an error when a communicator
is needed. Note that this is backward incompatible.
- For collectives, if a custom communicator is specified at a specific
site, we use it; otherwise, if default is provided, we use the default;
otherwise if default is "create", we create a communicator per
collective op, but reuse the communicator op when actor set is the same.
For all passed-in communicator, we check its actor set is the same as
that of the communicator.
- For "create", a single p2p communicator is created for all involved
actors without a passed-in communicator. We reuse a previously created
collective communicator if its actor set includes all p2p actors.
- `_init_communicator()` is called for all communicators (passed in or
created) on its involved actors, and `_destroy_communicator()` is only
called for the ones created by Compiled Graph.

By default, `experimental_compile(_default_communicator="create")` is
used, therefore:
- Any custom communicator passed in (via `with_tensor_transport()` for
p2p, and `collective.allreduce.bind()` for collective) will take
precedence and be used in the corresponding operations. These
communicators won't be reused at other sites. (User should explicitly
specify if they want.)
- For each collective operation without a custom communicator, a new
communicator is created, and this communicator is reused for all other
collective-operation-without-custom-communicator that has the same set
of actors.
- For all p2p communications without custom communicator, a single
communicator is created for all involved actors. Creation is skipped if
there is a communicator created for collective operation, whose actors
include all p2p actors: the collective communicator is reused in p2p
communications in this case.

This PR also refactors `CompiledDAG._preprocess()` so that code is more
organized.

Closes ray-project#47540


---------

Signed-off-by: Rui Qiao <ruisearch42@gmail.com>

…cator=comm) (ray-project#50023)

Currently, `with_tensor_transport(transport=arg)` supports 3 types of
arg: "nccl", "auto", or a `Communicator`. If "nccl" is used, or "auto"
is used and resolves to using NCCL, Ray creates a communicator
internally and uses it.

In order to better manage the communicator lifecycle, we make the
following changes:

- Add API to support
`experimental_compile(_default_communicator=default_comm)`, default_comm
is a default communicator user provided
- `_default_communicator` in the above API can be "create", which
explicitly tells Ray to create a default communicator and use it for all
`with_tensor_transport()` sites where a specific communicator was not
provided
- If `_default_communicator` is not specified, Ray will not create a
default communicator or reuse communicators passed in from other
`with_tensor_transport()` sites, but throw an error when a communicator
is needed. Note that this is backward incompatible.
- For collectives, if a custom communicator is specified at a specific
site, we use it; otherwise, if default is provided, we use the default;
otherwise if default is "create", we create a communicator per
collective op, but reuse the communicator op when actor set is the same.
For all passed-in communicator, we check its actor set is the same as
that of the communicator.
- For "create", a single p2p communicator is created for all involved
actors without a passed-in communicator. We reuse a previously created
collective communicator if its actor set includes all p2p actors.
- `_init_communicator()` is called for all communicators (passed in or
created) on its involved actors, and `_destroy_communicator()` is only
called for the ones created by Compiled Graph.

By default, `experimental_compile(_default_communicator="create")` is
used, therefore:
- Any custom communicator passed in (via `with_tensor_transport()` for
p2p, and `collective.allreduce.bind()` for collective) will take
precedence and be used in the corresponding operations. These
communicators won't be reused at other sites. (User should explicitly
specify if they want.)
- For each collective operation without a custom communicator, a new
communicator is created, and this communicator is reused for all other
collective-operation-without-custom-communicator that has the same set
of actors.
- For all p2p communications without custom communicator, a single
communicator is created for all involved actors. Creation is skipped if
there is a communicator created for collective operation, whose actors
include all p2p actors: the collective communicator is reused in p2p
communications in this case.

This PR also refactors `CompiledDAG._preprocess()` so that code is more
organized.

Closes ray-project#47540


---------

Signed-off-by: Rui Qiao <ruisearch42@gmail.com>

…cator=comm) (ray-project#50023)

Currently, `with_tensor_transport(transport=arg)` supports 3 types of
arg: "nccl", "auto", or a `Communicator`. If "nccl" is used, or "auto"
is used and resolves to using NCCL, Ray creates a communicator
internally and uses it.

In order to better manage the communicator lifecycle, we make the
following changes:

- Add API to support
`experimental_compile(_default_communicator=default_comm)`, default_comm
is a default communicator user provided
- `_default_communicator` in the above API can be "create", which
explicitly tells Ray to create a default communicator and use it for all
`with_tensor_transport()` sites where a specific communicator was not
provided
- If `_default_communicator` is not specified, Ray will not create a
default communicator or reuse communicators passed in from other
`with_tensor_transport()` sites, but throw an error when a communicator
is needed. Note that this is backward incompatible.
- For collectives, if a custom communicator is specified at a specific
site, we use it; otherwise, if default is provided, we use the default;
otherwise if default is "create", we create a communicator per
collective op, but reuse the communicator op when actor set is the same.
For all passed-in communicator, we check its actor set is the same as
that of the communicator.
- For "create", a single p2p communicator is created for all involved
actors without a passed-in communicator. We reuse a previously created
collective communicator if its actor set includes all p2p actors.
- `_init_communicator()` is called for all communicators (passed in or
created) on its involved actors, and `_destroy_communicator()` is only
called for the ones created by Compiled Graph.

By default, `experimental_compile(_default_communicator="create")` is
used, therefore:
- Any custom communicator passed in (via `with_tensor_transport()` for
p2p, and `collective.allreduce.bind()` for collective) will take
precedence and be used in the corresponding operations. These
communicators won't be reused at other sites. (User should explicitly
specify if they want.)
- For each collective operation without a custom communicator, a new
communicator is created, and this communicator is reused for all other
collective-operation-without-custom-communicator that has the same set
of actors.
- For all p2p communications without custom communicator, a single
communicator is created for all involved actors. Creation is skipped if
there is a communicator created for collective operation, whose actors
include all p2p actors: the collective communicator is reused in p2p
communications in this case.

This PR also refactors `CompiledDAG._preprocess()` so that code is more
organized.

Closes ray-project#47540


---------

Signed-off-by: Rui Qiao <ruisearch42@gmail.com>