Repository: pytorch
Issue: [funcol] functional collectives are 67% slower than torch.distributed collectives
Labels: module: performance,oncall: distributed,triaged

### 🐛 Describe the bug

Hi torch distributed team!

As we discussed in PTC, we found functional collectives are 34%~67% slower than c10d collectives due to the heavy CPU overhead.

To be specific, we benchmarked functional collectives (the big three: all_gather, reduce_scatter, all_reduce) in torch 2.4 and compared them with the c10d collectives. Here is the summary:

| collective    | c10d time | funcol time  |
|---------------|-----------|--------------|
| AllGather     | 131us     | 205us (156%) |
| ReduceScatter | 122us     | 164us (134%) |
| AllReduce     | 89us      | 149us (167%) |

We believe the overhead of funcol comes from extra copy, extra wrapping, and extra aten ops due to the tracing requirement for torch.compile.

Here are the profiles:

- AllGather

    #### torch.distributed.all_gather (totally 131us with creating empty tensor)
    ![image](https://github.com/user-attachments/assets/87df7aef-1dc7-4345-82e6-e9d89bb8d409)

    #### funcol.all_gather (205us)
    ![image](https://github.com/user-attachments/assets/7c4fcfba-685f-4f31-8958-81d845e8717a)

- ReduceScatter
    
    #### torch.distributed.reduce_scatter
    ![image](https://github.com/user-attachments/assets/c56e3d6b-3a6e-4a2a-b0af-1263b1aff48b)
    #### funcol.reduce_scatter
    ![image](https://github.com/user-attachments/assets/a5916e58-682c-4f0c-b1b5-39c73f321850)

- AllReduce 

    #### torch.distributed.all_reduce
    ![image](https://github.com/user-attachments/assets/3e01f159-97e8-4885-be88-61057641108d)
    #### funcol.all_reduce
    ![image](https://github.com/user-attachments/assets/af3e3f8f-0e6a-4da6-b64d-2ccfd3551a89)

We can reproduce the trace with the following code. Thanks : )

```
import os
import torch
import torch.distributed as dist
from torch.distributed.device_mesh import init_device_mesh
import torch.distributed._functional_collectives as funcol

world_size = int(os.environ["WORLD_SIZE"])


class TorchCollectiveProfile:
    @property
    def world_size(self):
        return world_size

    @property
    def device_type(self):
        return "cuda"

    def profile_funcol_all_gather(self):
        device_mesh = init_device_mesh(self.device_type, (self.world_size,))
        tensor = torch.randn(128, 128).cuda()
        torch.cuda.synchronize()
        dist.barrier()
        if device_mesh.get_rank() == 0:
            with torch.profiler.profile(
                schedule=torch.profiler.schedule(wait=5, warmup=10, active=5),
                on_trace_ready=lambda p: p.export_chrome_trace(
                    "test_profile_funcol_all_gather_" + str(p.step_num) + ".json"
                ),
                with_stack=True,
            ) as p:
                for _ in range(20):
                    _ = funcol.all_gather_tensor(tensor, 1, device_mesh)
                    p.step()
        else:
            for _ in range(20):
                _ = funcol.all_gather_tensor(tensor, 1, device_mesh)

        dist.barrier()
        torch.cuda.synchronize()

    def profile_c10d_all_gather(self):
        device_mesh = init_device_mesh(self.device_type, (self.world_size,))
        tensor = torch.randn(128, 128).cuda()

        def c10d_call():
            output_tensor = torch.empty(128, 128 * self.world_size, dtype=tensor.dtype, device=tensor.device)
            _ = dist.all_gather_into_tensor(output_tensor, tensor, device_mesh.get_group())

        torch.cuda.synchronize()
        dist.barrier()
        if device_mesh.get_rank() == 0:
            with torch.profiler.profile(
                schedule=torch.profiler.schedule(wait=5, warmup=10, active=5),
                on_trace_ready=lambda p: p.export_chrome_trace(
                    "test_profile_c10d_all_gather_" + str(p.step_num) + ".json"
                ),
                with_stack=True,
            ) as p:
                for _ in range(20):
                    c10d_call()
                    p.step()
        else:
            for _ in range(20):
                c10d_call()

        dist.barrier()
        torch.cuda.synchronize()
```


### Versions

PyTorch version: 2.4.1+cu121
Is debug build: False
CUDA used to build PyTorch: 12.1
ROCM used to build PyTorch: N/A

OS: Debian GNU/Linux 11 (bullseye) (x86_64)
GCC version: (Debian 10.2.1-6) 10.2.1 20210110
Clang version: Could not collect
CMake version: Could not collect
Libc version: glibc-2.31

Python version: 3.9.2 (default, Feb 28 2021, 17:03:44) [GCC 10.2.1 20210110] (64-bit runtime)
Python platform: Linux-5.15.120.bsk.2-amd64-x86_64-with-glibc2.31
Is CUDA available: True
CUDA runtime version: 12.1.105
CUDA_MODULE_LOADING set to: LAZY
GPU models and configuration:
GPU 0: NVIDIA A800-SXM4-40GB
GPU 1: NVIDIA A800-SXM4-40GB
GPU 2: NVIDIA A800-SXM4-40GB
GPU 3: NVIDIA A800-SXM4-40GB
GPU 4: NVIDIA A800-SXM4-40GB
GPU 5: NVIDIA A800-SXM4-40GB
GPU 6: NVIDIA A800-SXM4-40GB
GPU 7: NVIDIA A800-SXM4-40GB

Nvidia driver version: 535.161.08
cuDNN version: Probably one of the following:
/usr/lib/x86_64-linux-gnu/libcudnn.so.8.9.0
/usr/lib/x86_64-linux-gnu/libcudnn_adv_infer.so.8.9.0
/usr/lib/x86_64-linux-gnu/libcudnn_adv_train.so.8.9.0
/usr/lib/x86_64-linux-gnu/libcudnn_cnn_infer.so.8.9.0
/usr/lib/x86_64-linux-gnu/libcudnn_cnn_train.so.8.9.0
/usr/lib/x86_64-linux-gnu/libcudnn_ops_infer.so.8.9.0
/usr/lib/x86_64-linux-gnu/libcudnn_ops_train.so.8.9.0
HIP runtime version: N/A
MIOpen runtime version: N/A
Is XNNPACK available: True

CPU:
Architecture: x86_64
CPU op-mode(s): 32-bit, 64-bit
Byte Order: Little Endian
Address sizes: 52 bits physical, 57 bits virtual
CPU(s): 120
On-line CPU(s) list: 0-119
Thread(s) per core: 2
Core(s) per socket: 30
Socket(s): 2
NUMA node(s): 2
Vendor ID: GenuineIntel
CPU family: 6
Model: 106
Model name: Intel(R) Xeon(R) Platinum 8336C CPU @ 2.30GHz
Stepping: 6
CPU MHz: 2294.635
BogoMIPS: 4589.27
Hypervisor vendor: KVM
Virtualization type: full
L1d cache: 2.8 MiB
L1i cache: 1.9 MiB
L2 cache: 75 MiB
L3 cache: 108 MiB
NUMA node0 CPU(s): 0-59
NUMA node1 CPU(s): 60-119
Vulnerability Itlb multihit: Not affected
Vulnerability L1tf: Not affected
Vulnerability Mds: Not affected
Vulnerability Meltdown: Not affected
Vulnerability Mmio stale data: Vulnerable: Clear CPU buffers attempted, no microcode; SMT Host state unknown
Vulnerability Retbleed: Not affected
Vulnerability Spec store bypass: Mitigation; Speculative Store Bypass disabled via prctl and seccomp
Vulnerability Spectre v1: Mitigation; usercopy/swapgs barriers and __user pointer sanitization
Vulnerability Spectre v2: Mitigation; Enhanced IBRS, IBPB conditional, RSB filling, PBRSB-eIBRS SW sequence
Vulnerability Srbds: Not affected
Vulnerability Tsx async abort: Mitigation; TSX disabled
Flags: fpu vme de pse tsc msr pae mce cx8 apic sep mtrr pge mca cmov pat pse36 clflush mmx fxsr sse sse2 ss ht syscall nx pdpe1gb rdtscp lm constant_tsc rep_good nopl xtopology nonstop_tsc cpuid pni pclmulqdq monitor ssse3 fma cx16 pcid sse4_1 sse4_2 x2apic movbe popcnt tsc_deadline_timer aes xsave avx f16c rdrand hypervisor lahf_lm abm 3dnowprefetch cpuid_fault invpcid_single ssbd ibrs ibpb stibp ibrs_enhanced fsgsbase tsc_adjust bmi1 avx2 smep bmi2 erms invpcid avx512f avx512dq rdseed adx smap avx512ifma clflushopt clwb avx512cd sha_ni avx512bw avx512vl xsaveopt xsavec xgetbv1 xsaves wbnoinvd arat avx512vbmi umip pku ospke avx512_vbmi2 gfni vaes vpclmulqdq avx512_vnni avx512_bitalg avx512_vpopcntdq rdpid fsrm md_clear arch_capabilities

Versions of relevant libraries:
[pip3] numpy==2.0.2
[pip3] torch==2.4.1
[pip3] torchaudio==2.4.1
[pip3] torchvision==0.19.1
[pip3] triton==3.0.0
[conda] Could not collect

cc @msaroufim @H-Huang @awgu @kwen2501 @wanchaol @fegin @fduwjj @wz337 @wconstab @d4l3k @c-p-i-o