Enable symmetric memory all reduce by default only enabling for TP

[ilmarkov]

Purpose

Enable torch symm mem by default and fix DataParallel crash. Enable torch symm mem only for TP communicator.

Conflict with DataParallel

In DataParallel we have multiple workers that are assigned to the same set of devices in their environment (with different CUDA_VISIBLE_DEVICES).

In case of DP communicator, we have wrong local_rank:rank mapping, e.g.
unique_name: dp:0, device: cuda:0, rank: 1, world_size: 2.

In case of EP communicator in DP setup we have two workers with the same device a different ranks within group:

unique_name: ep:0, device: cuda:0, rank: 0, world_size: 4                                                                                                                                                 
unique_name: ep:0, device: cuda:0, rank: 2, world_size: 4

So in following line, we basically use the same self.device in the same group. That makes torch here think that we overlap devices, although the allocated tensors are in different environment. Torch symm mem fails to do rendevouz in this setup so we basically can not use symm mem
with DataParallel. This PR only avoids the crash.

         self.buffer = torch_symm_mem.empty(
            self.max_size // self.dtype.itemsize,
            device=self.device,
            dtype=self.dtype,
        )

Fixes #24694.

Test Plan

Added test to test_symm_mem_allreduce.py

Test Result

Passed test

[gemini-code-assist]

Code Review

This pull request enables symmetric memory all-reduce by default and fixes a crash when used with DataParallel. The core fix correctly sets TORCH_SYMM_MEM_ALLOW_OVERLAPPING_DEVICES when a DataParallel setup is detected, which is a sound approach. The new test case validates this fix. The changes are generally good, but I have identified a critical bug in the test cleanup logic that could lead to resource leaks, and a high-severity issue regarding environment variable modification that should be addressed to prevent potential side effects.

[gemini-code-assist]

There is a critical bug in the cleanup logic. pytest.skip() raises a special exception to stop the test execution. Because it's called before cleanup_dist_env_and_memory() within the finally block, the cleanup function will not be executed if a worker process sends a skip message. This will lead to resource leaks (e.g., dangling distributed processes), which can cause subsequent tests to fail or hang. The cleanup must be guaranteed to run before the test is skipped.

Suggested change

	if val is not None:
	pytest.skip(val)
	cleanup_dist_env_and_memory()
	cleanup_dist_env_and_memory()
	if val is not None:
	pytest.skip(val)

[mgoin]

LGTM!