Repository: pytorch
Issue: Flight recorder cannot mark the state of the batch P2P OP entry as completed.
Labels: oncall: distributed,triaged

### 🐛 Describe the bug

We use the `torchrun --standalone --nproc-per-node=8 test.py` on a single node. After the `batch_isend_irecv` is completed, we dump the entries and find that the state of `nccl:send` and `nccl:recv` are also "scheduled".  The reason may be that the following codes in ProcessGroupNCCL.cpp does not set the ncclEndEvent_ to batch P2P OPs.

https://github.com/pytorch/pytorch/blob/56cc22eb01639ebd1ca3bbe8ba0381cd8fdbcff8/torch/csrc/distributed/c10d/ProcessGroupNCCL.cpp#L3213-L3232

The codes to reproduce the issue.

```Python
import os
import time
import pickle
import torch
import torch.distributed as dist

os.environ["TORCH_NCCL_TRACE_BUFFER_SIZE"] = "200"


def dump_nccl_trace(rank):
    trace = torch._C._distributed_c10d._dump_nccl_trace(
        includeStackTraces=False,
        onlyActive=False,
    )
    trace = pickle.loads(trace)
    print(f"Rank = {rank}, trace = {trace}")


def test_batch_p2p():
    rank = int(os.getenv("RANK", "0"))
    local_rank = int(os.getenv("LOCAL_RANK", "0"))
    world_size = int(os.getenv("WORLD_SIZE", "1"))
    send_tensor = torch.arange(2, dtype=torch.float32) + 2 * rank
    recv_tensor = torch.randn(2, dtype=torch.float32)
    send_tensor = send_tensor.to(local_rank)
    recv_tensor = recv_tensor.to(local_rank)
    send_op = dist.P2POp(dist.isend, send_tensor, (rank + 1) % world_size)
    recv_op = dist.P2POp(dist.irecv, recv_tensor, (rank - 1 + world_size)%world_size)
    reqs = dist.batch_isend_irecv([send_op, recv_op])
    for req in reqs:
        req.wait()
    torch.cuda.synchronize()
    print(f"Rank = {rank}, tensor = {recv_tensor}")
    dump_nccl_trace(rank)


if __name__ == "__main__":
    dist.init_process_group(backend="nccl")
    test_batch_p2p()
```

The trace of flight recorder.

```
{
  "version": "2.1",
  "pg_config": {
    "0": {
      "name": "0",
      "desc": "default_pg",
      "ranks": "[0, 1, 2, 3, 4, 5, 6, 7]"
    }
  },
  "entries": [
    {
      "record_id": 0,
      "pg_id": 0,
      "process_group": [
        "0",
        "default_pg"
      ],
      "collective_seq_id": 1,
      "p2p_seq_id": 0,
      "op_id": 1,
      "profiling_name": "nccl:send 0->1",
      "time_created_ns": 1728976538249745700,
      "input_sizes": [
        [
          2
        ]
      ],
      "input_dtypes": [
        "Float"
      ],
      "output_sizes": [
        [
          2
        ]
      ],
      "output_dtypes": [
        "Float"
      ],
      "state": "scheduled",
      "time_discovered_started_ns": null,
      "time_discovered_completed_ns": null,
      "retired": false,
      "is_p2p": true
    },
    {
      "record_id": 1,
      "pg_id": 0,
      "process_group": [
        "0",
        "default_pg"
      ],
      "collective_seq_id": 1,
      "p2p_seq_id": 0,
      "op_id": 2,
      "profiling_name": "nccl:recv 0<-7",
      "time_created_ns": 1728976538250029300,
      "input_sizes": [
        [
          2
        ]
      ],
      "input_dtypes": [
        "Float"
      ],
      "output_sizes": [
        [
          2
        ]
      ],
      "output_dtypes": [
        "Float"
      ],
      "state": "scheduled",
      "time_discovered_started_ns": null,
      "time_discovered_completed_ns": null,
      "retired": false,
      "is_p2p": true
    },
    {
      "record_id": 2,
      "pg_id": 0,
      "process_group": [
        "0",
        "default_pg"
      ],
      "collective_seq_id": 1,
      "p2p_seq_id": 0,
      "op_id": 2,
      "profiling_name": "nccl:coalesced",
      "time_created_ns": 1728976538250107000,
      "input_sizes": [],
      "input_dtypes": [],
      "output_sizes": [],
      "output_dtypes": [],
      "state": "completed",
      "time_discovered_started_ns": 1728976541309922000,
      "time_discovered_completed_ns": 1728976541309923600,
      "retired": false,
      "is_p2p": false
    }
  ]
}
```



### Versions


Versions of relevant libraries:
[pip3] numpy==1.24.4
[pip3] onnx==1.16.2
[pip3] torch==2.4.0
[pip3] torchaudio==2.4.0
[pip3] torchlibrosa==0.1.0
[pip3] torchvision==0.19.0
[pip3] triton==3.0.0

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