​作者​：昇腾实战派
​知识地图​：https://blog.csdn.net/Lumos_Lovegood/article/details/161601003

背景概述

在基于950PR环境的多卡并行训练场景中，使用PyTorch框架调用NPU通信算子进行模型参数同步是常见需求。为验证通信算子在高并发下的稳定性与性能表现，需在8张NPU卡上并行执行指定时长的通信任务。然而，在实际测试过程中，部分卡在规定时间结束后未能及时退出，导致整体任务因同步超时而失败，影响了测试用例的可靠性与可重复性。本文针对该问题展开分析，并提出有效的解决方案。
问题脚本如下：

# npu_5d_parallel_4gb_24h_stress_test.py
import os
import time
import torch
import torch.distributed as dist
import torch_npu

def main():
    local_rank = int(os.environ.get("LOCAL_RANK", "0"))
    torch.npu.set_device(local_rank)
    device = f"npu:{local_rank}"

    dist.init_process_group(backend="hccl")
    rank = dist.get_rank()
    world_size = dist.get_world_size()
    assert world_size == 8, f"Expected 8 NPUs, got {world_size}"

    full_group = dist.new_group(ranks=list(range(8)), backend="hccl")

    # === 4GB per communication ===
    # num_elements_4gb = (4 * 1024 ** 3) // 4
    num_elements_4gb = (128 * 1024**2) // 4
    chunk_elements_4gb = num_elements_4gb // 8

    # --- 初始化张量 ---
    tp_tensor = torch.full((num_elements_4gb,), float(rank + 1), dtype=torch.float32, device=device)
    dp_tensor = torch.full((num_elements_4gb,), float(rank + 10), dtype=torch.float32, device=device)

    ep_input_list = [
        torch.full((chunk_elements_4gb,), float(rank + 1), dtype=torch.float32, device=device)
        for _ in range(world_size)
    ]
    ep_output_list = [torch.empty_like(ep_input_list[0]) for _ in range(world_size)]

    sp_local = torch.full((chunk_elements_4gb,), float(rank + 20), dtype=torch.float32, device=device)
    sp_output = [torch.empty_like(sp_local) for _ in range(world_size)]

    pp_local = torch.full((chunk_elements_4gb,), float(rank + 30), dtype=torch.float32, device=device)
    pp_output = [torch.empty_like(pp_local) for _ in range(world_size)]

    # --- Streams ---
    streams = {
        name: torch.npu.Stream(device=device)
        for name in ['tp', 'dp', 'ep', 'sp', 'pp']
    }

    print(f"[Rank {rank}] Starting 24-hour 5D parallel stress test (4GB each)...")

    # === Profiler: only first iteration ===
    experimental_config = torch_npu.profiler._ExperimentalConfig(
        aic_metrics=torch_npu.profiler.AiCMetrics.PipeUtilization,
        profiler_level=torch_npu.profiler.ProfilerLevel.Level1,
        l2_cache=False,
        data_simplification=False
    )

    prof_ctx = torch_npu.profiler.profile(
        activities=[torch_npu.profiler.ProfilerActivity.CPU, torch_npu.profiler.ProfilerActivity.NPU],
        schedule=torch_npu.profiler.schedule(wait=0, warmup=0, active=1, repeat=1),
        on_trace_ready=torch_npu.profiler.tensorboard_trace_handler("./5d_72h_first_profile"),
        record_shapes=False,
        with_stack=False,
        experimental_config=experimental_config
    )

    # === 运行 24 小时 ===
    DURATION_HOURS = 0.02
    DURATION_SECONDS = int(DURATION_HOURS * 3600)
    start_time = time.time()
    step = 0
    last_log_time = start_time
    LOG_INTERVAL_SECONDS = 300  # 每 5 分钟打印一次
    # step = 0

    with prof_ctx as prof:
        while (time.time() - start_time) < DURATION_SECONDS:
        # while step<2:
            # Launch all communications
            with torch.npu.stream(streams['tp']):
                dist.all_reduce(tp_tensor, op=dist.ReduceOp.SUM, group=full_group)
            with torch.npu.stream(streams['dp']):
                dist.all_reduce(dp_tensor, op=dist.ReduceOp.SUM, group=full_group)
            with torch.npu.stream(streams['ep']):
                dist.all_to_all(ep_output_list, ep_input_list, group=full_group)
            with torch.npu.stream(streams['sp']):
                dist.all_gather(sp_output, sp_local, group=full_group)
            with torch.npu.stream(streams['pp']):
                dist.all_gather(pp_output, pp_local, group=full_group)

            torch.npu.synchronize()

            # Only profile the very first step
            # if step == 0:
            #     prof.step()
            #     print(f"[Rank {rank}] First iteration profiled.")

            step += 1

            # Periodic logging (only rank 0)
            current_time = time.time()
            if rank == 0 and (current_time - last_log_time) >= LOG_INTERVAL_SECONDS:
                elapsed = current_time - start_time
                hours = elapsed / 3600
                iter_per_sec = step / elapsed if elapsed > 0 else 0
                total_gb = step * 5 * 4  # 5 ops × 4GB
                print(f"[LOG] Elapsed: {hours:.2f}h | Iter: {step} | "
                      f"Speed: {iter_per_sec:.2f} iter/s | Total traffic: {total_gb:.1f} GB")
                last_log_time = current_time



    if rank == 0:
        total_gb = step * 5 * 4
        print(f"✅ 24-hour stress test completed! Total iterations: {step}, Total traffic: {total_gb:.1f} GB")
        print("Profiling of first iteration saved to ./5d_24h_first_profile/")

    dist.destroy_process_group()


if __name__ == "__main__":
    main()

问题描述

在执行多卡通信算子测试用例时，脚本采用时间作为循环终止条件，期望所有卡在达到预设时长后同步退出。然而，实际运行中出现以下异常现象：

• 部分卡在执行 torch.npu.synchronize() 时触发超时异常；
• 仅部分卡报错，其余卡正常完成任务并退出；
• 由于先完成的卡提前退出，导致其余卡在等待已退出卡的同步信号，最终因等待超时而失败。

日志如下所示：

EE9999[PID: 19223] 2026-06-01-17:56:50.608.148 (EE9999):  rtDeviceSynchronizeWithTimeout execution failed, reason=tsfw unknown error[FUNC:FuncErrorReason][FILE:error_message_manage.cc][LINE:65]
TraceBack (most recent call last):
       wait for compute device to finish failed, runtime result = 507001.[FUNC:ReportCallError][FILE:log_inner.cpp][LINE:148]
, error code is 507001
[ERROR] 2026-06-01-17:56:50 (PID:19223, Device:4, RankID:4) ERR00100 PTA call acl api failed
[rank4]: Traceback (most recent call last):
[rank4]:   File "/home/5d_parallel_8p_tp8_48H.py", line 121, in <module>
[rank4]:     main()
[rank4]:   File "/home/5d_parallel_8p_tp8_48H.py", line 90, in main
[rank4]:     torch.npu.synchronize()
[rank4]:   File "/home/miniconda3/envs/zgg/lib/python3.10/site-packages/torch_npu/npu/utils.py", line 72, in synchronize
[rank4]:     return torch_npu._C._npu_synchronize()
[rank4]: RuntimeError: npuSynchronizeDevice:../torch_npu/csrc/core/npu/NPUStream.cpp:565 NPU function error: device error type 0xFFFF, [PID: 19223] 2026-06-01-17:56:50.606.418 Communication_Error_Timeout(EI0002): An timeout occurs when the Notify register waits for execution. Waiting peer end: 4294967295; task information: streamID(sqId):[39], taskID(sqeId):[2151055025], taskType:[TaskParamType::TASK_NOTIFY_WAIT]; communication operator information: notify id:[0x0000000000000042], value:[1], remote rank:[local]; communicator: .
[rank4]:         Possible Cause: 1. An exception occurs during the execution on some NPUs in the cluster. As a result, collective communication operation failed.2. The execution speed on some NPU in the cluster is too slow to complete a communication operation within the timeout interval. (The default timeout interval is 1800s, You can set the interval by using HCCL_EXEC_TIMEOUT.)3. The number of training samples of each NPU is inconsistent.4. Packet loss or other connectivity problems occur on the communication link.
[rank4]:         Solution: 1. If this error is reported on part of these ranks, check other ranks to see whether other errors have been reported earlier.2. If this error is reported for all ranks, check whether the error reporting time is consistent (the maximum difference must not exceed 1800s). If not, locate the cause or set the HCCL_EXEC_TIMEOUT environment variable to a larger value. 3. Ensure that the number of training samples of each NPU is consistent. 4. Check whether the completion queue element (CQE) of the error exists in the plog(grep -rn 'error cqe'). If so, check the network connection status. For details about the troubleshooting method, search for the keyword "EI0002" on https://www.hiascend.com/en/document/.
[rank4]: TraceBack (most recent call last):
[rank4]:         The error from device(chipId:0, dieId:0), serial number is 1, wait timeout occurred during task execution, stream_id:54, sq_id:9, task_pos:1531, id=2, timeout=627s, cntFlag = 0, subType = 1(single notify wait), cntValue = 0, clrFlag = 1, waitMode = 0, bitmap = 0.[FUNC:ProcessDavidStarsWaitTimeoutErrorInfo][FILE:device_error_proc_c.cc][LINE:922]
[rank4]:         An timeout occurs when the Notify register waits for execution. Waiting peer end: 4294967295; task information: streamID(sqId):[39], taskID(sqeId):[2151055025], taskType:[TaskParamType::TASK_NOTIFY_WAIT]; communication operator information: notify id:[0x0000000000000042], value:[1], remote rank:[local]; communicator: .
[rank4]:         rtDeviceSynchronizeWithTimeout execution failed, reason=tsfw unknown error[FUNC:FuncErrorReason][FILE:error_message_manage.cc][LINE:65]
[rank4]:         wait for compute device to finish failed, runtime result = 507001.[FUNC:ReportCallError][FILE:log_inner.cpp][LINE:148]
[rank4]: , error code is 507001
[rank4]: [ERROR] 2026-06-01-17:56:50 (PID:19223, Device:4, RankID:4) ERR00100 PTA call acl api failed
[2026-06-01 17:56:50] [WARNING] [19213] profiler.py: Stop profiler while current state is RECORD_AND_SAVE, perhaps the scheduling cycle has not yet completed.
[2026-06-01 17:56:50] [ERROR] [19213] profiler.py: Call stop failed. Exception: npuSynchronizeDevice:../torch_npu/csrc/core/npu/NPUStream.cpp:565 NPU function error: device error type 0xFFFF, EE9999: Inner Error!
EE9999[PID: 19213] 2026-06-01-17:56:50.685.686 (EE9999):  rtDeviceSynchronizeWithTimeout execution failed, reason=tsfw unknown error[FUNC:FuncErrorReason][FILE:error_message_manage.cc][LINE:65]
TraceBack (most recent call last):
       wait for compute device to finish failed, runtime result = 507001.[FUNC:ReportCallError][FILE:log_inner.cpp][LINE:148]
, error code is 507001
[ERROR] 2026-06-01-17:56:50 (PID:19213, Device:0, RankID:0) ERR00100 PTA call acl api failed
[rank0]: Traceback (most recent call last):
[rank0]:   File "/home/5d_parallel_8p_tp8_48H.py", line 121, in <module>
[rank0]:     main()
[rank0]:   File "/home/5d_parallel_8p_tp8_48H.py", line 90, in main
[rank0]:     torch.npu.synchronize()
[rank0]:   File "/home/miniconda3/envs/zgg/lib/python3.10/site-packages/torch_npu/npu/utils.py", line 72, in synchronize
[rank0]:     return torch_npu._C._npu_synchronize()
[rank0]: RuntimeError: npuSynchronizeDevice:../torch_npu/csrc/core/npu/NPUStream.cpp:565 NPU function error: device error type 0xFFFF, [PID: 19213] 2026-06-01-17:56:50.683.460 Communication_Error_Timeout(EI0002): An timeout occurs when the Notify register waits for execution. Waiting peer end: 4294967295; task information: streamID(sqId):[39], taskID(sqeId):[2151055025], taskType:[TaskParamType::TASK_NOTIFY_WAIT]; communication operator information: notify id:[0x0000000000000042], value:[1], remote rank:[local]; communicator: .
[rank0]:         Possible Cause: 1. An exception occurs during the execution on some NPUs in the cluster. As a result, collective communication operation failed.2. The execution speed on some NPU in the cluster is too slow to complete a communication operation within the timeout interval. (The default timeout interval is 1800s, You can set the interval by using HCCL_EXEC_TIMEOUT.)3. The number of training samples of each NPU is inconsistent.4. Packet loss or other connectivity problems occur on the communication link.
[rank0]:         Solution: 1. If this error is reported on part of these ranks, check other ranks to see whether other errors have been reported earlier.2. If this error is reported for all ranks, check whether the error reporting time is consistent (the maximum difference must not exceed 1800s). If not, locate the cause or set the HCCL_EXEC_TIMEOUT environment variable to a larger value. 3. Ensure that the number of training samples of each NPU is consistent. 4. Check whether the completion queue element (CQE) of the error exists in the plog(grep -rn 'error cqe'). If so, check the network connection status. For details about the troubleshooting method, search for the keyword "EI0002" on https://www.hiascend.com/en/document/.
[rank0]: TraceBack (most recent call last):
[rank0]:         The error from device(chipId:0, dieId:0), serial number is 3, wait timeout occurred during task execution, stream_id:54, sq_id:9, task_pos:1538, id=2, timeout=627s, cntFlag = 0, subType = 1(single notify wait), cntValue = 0, clrFlag = 1, waitMode = 0, bitmap = 0.[FUNC:ProcessDavidStarsWaitTimeoutErrorInfo][FILE:device_error_proc_c.cc][LINE:922]
[rank0]:         An timeout occurs when the Notify register waits for execution. Waiting peer end: 4294967295; task information: streamID(sqId):[39], taskID(sqeId):[2151055025], taskType:[TaskParamType::TASK_NOTIFY_WAIT]; communication operator information: notify id:[0x0000000000000042], value:[1], remote rank:[local]; communicator: .
[rank0]:         rtDeviceSynchronizeWithTimeout execution failed, reason=tsfw unknown error[FUNC:FuncErrorReason][FILE:error_message_manage.cc][LINE:65]
[rank0]:         wait for compute device to finish failed, runtime result = 507001.[FUNC:ReportCallError][FILE:log_inner.cpp][LINE:148]
[rank0]: , error code is 507001
[rank0]: [ERROR] 2026-06-01-17:56:50 (PID:19213, Device:0, RankID:0) ERR00100 PTA call acl api failed
[2026-06-01 17:56:52] [ERROR] [19216] profiler.py: Call stop failed. Exception: npuSynchronizeDevice:../torch_npu/csrc/core/npu/NPUStream.cpp:565 NPU function error: device error type 0xFFFF, EE9999: Inner Error!
EE9999[PID: 19216] 2026-06-01-17:56:52.960.986 (EE9999):  rtDeviceSynchronizeWithTimeout execution failed, reason=tsfw unknown error[FUNC:FuncErrorReason][FILE:error_message_manage.cc][LINE:65]
TraceBack (most recent call last):
       wait for compute device to finish failed, runtime result = 507001.[FUNC:ReportCallError][FILE:log_inner.cpp][LINE:148]
, error code is 507001
[ERROR] 2026-06-01-17:56:52 (PID:19216, Device:1, RankID:1) ERR00100 PTA call acl api failed
[rank1]: Traceback (most recent call last):
[rank1]:   File "/home/5d_parallel_8p_tp8_48H.py", line 121, in <module>
[rank1]:     main()
[rank1]:   File "/home/5d_parallel_8p_tp8_48H.py", line 90, in main
[rank1]:     torch.npu.synchronize()
[rank1]:   File "/home/miniconda3/envs/zgg/lib/python3.10/site-packages/torch_npu/npu/utils.py", line 72, in synchronize
[rank1]:     return torch_npu._C._npu_synchronize()
[rank1]: RuntimeError: npuSynchronizeDevice:../torch_npu/csrc/core/npu/NPUStream.cpp:565 NPU function error: device error type 0xFFFF, [PID: 19216] 2026-06-01-17:56:52.959.622 Communication_Error_Timeout(EI0002): An timeout occurs when the Notify register waits for execution. Waiting peer end: 4294967295; task information: streamID(sqId):[39], taskID(sqeId):[2151055025], taskType:[TaskParamType::TASK_NOTIFY_WAIT]; communication operator information: notify id:[0x0000000000000042], value:[1], remote rank:[local]; communicator: .
[rank1]:         Possible Cause: 1. An exception occurs during the execution on some NPUs in the cluster. As a result, collective communication operation failed.2. The execution speed on some NPU in the cluster is too slow to complete a communication operation within the timeout interval. (The default timeout interval is 1800s, You can set the interval by using HCCL_EXEC_TIMEOUT.)3. The number of training samples of each NPU is inconsistent.4. Packet loss or other connectivity problems occur on the communication link.
[rank1]:         Solution: 1. If this error is reported on part of these ranks, check other ranks to see whether other errors have been reported earlier.2. If this error is reported for all ranks, check whether the error reporting time is consistent (the maximum difference must not exceed 1800s). If not, locate the cause or set the HCCL_EXEC_TIMEOUT environment variable to a larger value. 3. Ensure that the number of training samples of each NPU is consistent. 4. Check whether the completion queue element (CQE) of the error exists in the plog(grep -rn 'error cqe'). If so, check the network connection status. For details about the troubleshooting method, search for the keyword "EI0002" on https://www.hiascend.com/en/document/.
[rank1]: TraceBack (most recent call last):
[rank1]:         The error from device(chipId:0, dieId:0), serial number is 2, wait timeout occurred during task execution, stream_id:54, sq_id:9, task_pos:1536, id=2, timeout=627s, cntFlag = 0, subType = 1(single notify wait), cntValue = 0, clrFlag = 1, waitMode = 0, bitmap = 0.[FUNC:ProcessDavidStarsWaitTimeoutErrorInfo][FILE:device_error_proc_c.cc][LINE:922]
[rank1]:         An timeout occurs when the Notify register waits for execution. Waiting peer end: 4294967295; task information: streamID(sqId):[39], taskID(sqeId):[2151055025], taskType:[TaskParamType::TASK_NOTIFY_WAIT]; communication operator information: notify id:[0x0000000000000042], value:[1], remote rank:[local]; communicator: .
[rank1]:         rtDeviceSynchronizeWithTimeout execution failed, reason=tsfw unknown error[FUNC:FuncErrorReason][FILE:error_message_manage.cc][LINE:65]
[rank1]:         wait for compute device to finish failed, runtime result = 507001.[FUNC:ReportCallError][FILE:log_inner.cpp][LINE:148]
[rank1]: , error code is 507001
[rank1]: [ERROR] 2026-06-01-17:56:52 (PID:19216, Device:1, RankID:1) ERR00100 PTA call acl api failed

从日志分析可知，问题根源在于：各卡执行的通信次数不一致。由于时间控制逻辑未保证所有卡同步退出，导致部分卡提前退出，破坏了多卡间的同步一致性。

根本原因分析

• 使用时间作为循环结束条件时，各卡因启动时间、执行速度差异，导致部分卡率先达到时间阈值并退出；
• 退出后未主动通知其他卡，其余卡仍在等待其完成同步操作；
• torch.npu.synchronize() 在等待未完成的进程时，因无响应而触发超时，最终导致整个任务失败。

解决方案

修改测试用例代码，保证所有卡执行相同次数的通信，都能正常退出。

3.1 使用step作为循环结束条件

# npu_5d_parallel_4gb_24h_stress_test.py
import os
import time
import torch
import torch.distributed as dist
import torch_npu

def main():
    local_rank = int(os.environ.get("LOCAL_RANK", "0"))
    torch.npu.set_device(local_rank)
    device = f"npu:{local_rank}"

    dist.init_process_group(backend="hccl")
    rank = dist.get_rank()
    world_size = dist.get_world_size()
    assert world_size == 8, f"Expected 8 NPUs, got {world_size}"

    full_group = dist.new_group(ranks=list(range(8)), backend="hccl")

    # === 4GB per communication ===
    # num_elements_4gb = (4 * 1024 ** 3) // 4
    num_elements_4gb = (128 * 1024**2) // 4
    chunk_elements_4gb = num_elements_4gb // 8

    # --- 初始化张量 ---
    tp_tensor = torch.full((num_elements_4gb,), float(rank + 1), dtype=torch.float32, device=device)
    dp_tensor = torch.full((num_elements_4gb,), float(rank + 10), dtype=torch.float32, device=device)

    ep_input_list = [
        torch.full((chunk_elements_4gb,), float(rank + 1), dtype=torch.float32, device=device)
        for _ in range(world_size)
    ]
    ep_output_list = [torch.empty_like(ep_input_list[0]) for _ in range(world_size)]

    sp_local = torch.full((chunk_elements_4gb,), float(rank + 20), dtype=torch.float32, device=device)
    sp_output = [torch.empty_like(sp_local) for _ in range(world_size)]

    pp_local = torch.full((chunk_elements_4gb,), float(rank + 30), dtype=torch.float32, device=device)
    pp_output = [torch.empty_like(pp_local) for _ in range(world_size)]

    # --- Streams ---
    streams = {
        name: torch.npu.Stream(device=device)
        for name in ['tp', 'dp', 'ep', 'sp', 'pp']
    }

    print(f"[Rank {rank}] Starting 24-hour 5D parallel stress test (4GB each)...")

    # === Profiler: only first iteration ===
    experimental_config = torch_npu.profiler._ExperimentalConfig(
        aic_metrics=torch_npu.profiler.AiCMetrics.PipeUtilization,
        profiler_level=torch_npu.profiler.ProfilerLevel.Level1,
        l2_cache=False,
        data_simplification=False
    )

    prof_ctx = torch_npu.profiler.profile(
        activities=[torch_npu.profiler.ProfilerActivity.CPU, torch_npu.profiler.ProfilerActivity.NPU],
        schedule=torch_npu.profiler.schedule(wait=0, warmup=0, active=1, repeat=1),
        on_trace_ready=torch_npu.profiler.tensorboard_trace_handler("./5d_72h_first_profile"),
        record_shapes=False,
        with_stack=False,
        experimental_config=experimental_config
    )

    # === 运行 24 小时 ===
    DURATION_HOURS = 0.02
    DURATION_SECONDS = int(DURATION_HOURS * 3600)
    start_time = time.time()
    step = 0
    last_log_time = start_time
    LOG_INTERVAL_SECONDS = 300  # 每 5 分钟打印一次
    # step = 0

    with prof_ctx as prof:
        # while (time.time() - start_time) < DURATION_SECONDS:
        while step<20:
            # Launch all communications
            with torch.npu.stream(streams['tp']):
                dist.all_reduce(tp_tensor, op=dist.ReduceOp.SUM, group=full_group)
            with torch.npu.stream(streams['dp']):
                dist.all_reduce(dp_tensor, op=dist.ReduceOp.SUM, group=full_group)
            with torch.npu.stream(streams['ep']):
                dist.all_to_all(ep_output_list, ep_input_list, group=full_group)
            with torch.npu.stream(streams['sp']):
                dist.all_gather(sp_output, sp_local, group=full_group)
            with torch.npu.stream(streams['pp']):
                dist.all_gather(pp_output, pp_local, group=full_group)

            torch.npu.synchronize()

            # Only profile the very first step
            # if step == 0:
            #     prof.step()
            #     print(f"[Rank {rank}] First iteration profiled.")

            step += 1

            # Periodic logging (only rank 0)
            current_time = time.time()
            if rank == 0 and (current_time - last_log_time) >= LOG_INTERVAL_SECONDS:
                elapsed = current_time - start_time
                hours = elapsed / 3600
                iter_per_sec = step / elapsed if elapsed > 0 else 0
                total_gb = step * 5 * 4  # 5 ops × 4GB
                print(f"[LOG] Elapsed: {hours:.2f}h | Iter: {step} | "
                      f"Speed: {iter_per_sec:.2f} iter/s | Total traffic: {total_gb:.1f} GB")
                last_log_time = current_time



    if rank == 0:
        total_gb = step * 5 * 4
        print(f"✅ 24-hour stress test completed! Total iterations: {step}, Total traffic: {total_gb:.1f} GB")
        print("Profiling of first iteration saved to ./5d_24h_first_profile/")

    dist.destroy_process_group()


if __name__ == "__main__":
    main()

3.2 使用时间作为循环结束条件，但是有一张卡到时间退出，其余卡必需同步退出

# npu_5d_parallel_4gb_24h_stress_test.py
import os
import time
import torch
import torch.distributed as dist
import torch_npu

def main():
    local_rank = int(os.environ.get("LOCAL_RANK", "0"))
    torch.npu.set_device(local_rank)
    device = f"npu:{local_rank}"

    dist.init_process_group(backend="hccl")
    rank = dist.get_rank()
    world_size = dist.get_world_size()
    assert world_size == 8, f"Expected 8 NPUs, got {world_size}"

    full_group = dist.new_group(ranks=list(range(8)), backend="hccl")

    # === 4GB per communication ===
    # num_elements_4gb = (4 * 1024 ** 3) // 4
    num_elements_4gb = (128 * 1024**2) // 4
    chunk_elements_4gb = num_elements_4gb // 8

    # --- 初始化张量 ---
    tp_tensor = torch.full((num_elements_4gb,), float(rank + 1), dtype=torch.float32, device=device)
    dp_tensor = torch.full((num_elements_4gb,), float(rank + 10), dtype=torch.float32, device=device)

    ep_input_list = [
        torch.full((chunk_elements_4gb,), float(rank + 1), dtype=torch.float32, device=device)
        for _ in range(world_size)
    ]
    ep_output_list = [torch.empty_like(ep_input_list[0]) for _ in range(world_size)]

    sp_local = torch.full((chunk_elements_4gb,), float(rank + 20), dtype=torch.float32, device=device)
    sp_output = [torch.empty_like(sp_local) for _ in range(world_size)]

    pp_local = torch.full((chunk_elements_4gb,), float(rank + 30), dtype=torch.float32, device=device)
    pp_output = [torch.empty_like(pp_local) for _ in range(world_size)]

    # --- Streams ---
    streams = {
        name: torch.npu.Stream(device=device)
        for name in ['tp', 'dp', 'ep', 'sp', 'pp']
    }

    print(f"[Rank {rank}] Starting 24-hour 5D parallel stress test (4GB each)...")

    # === Profiler: only first iteration ===
    experimental_config = torch_npu.profiler._ExperimentalConfig(
        aic_metrics=torch_npu.profiler.AiCMetrics.PipeUtilization,
        profiler_level=torch_npu.profiler.ProfilerLevel.Level1,
        l2_cache=False,
        data_simplification=False
    )

    prof_ctx = torch_npu.profiler.profile(
        activities=[torch_npu.profiler.ProfilerActivity.CPU, torch_npu.profiler.ProfilerActivity.NPU],
        schedule=torch_npu.profiler.schedule(wait=0, warmup=0, active=1, repeat=1),
        on_trace_ready=torch_npu.profiler.tensorboard_trace_handler("./5d_72h_first_profile"),
        record_shapes=False,
        with_stack=False,
        experimental_config=experimental_config
    )

    # === 运行 24 小时 ===
    DURATION_HOURS = 0.02
    DURATION_SECONDS = int(DURATION_HOURS * 3600)
    start_time = time.time()
    step = 0
    last_log_time = start_time
    LOG_INTERVAL_SECONDS = 300  # 每 5 分钟打印一次
    # step = 0

    with prof_ctx as prof:
        # while (time.time() - start_time) < DURATION_SECONDS:
        # while step<20:
        while True:
            # 这种修改需要所有卡都在相同的通信进程组内
            # --------------------------
            # 【第一步】所有卡判断是否超时（本地判断）
            # --------------------------
            need_exit = 1 if (time.time() - start_time > DURATION_SECONDS) else 0

            # --------------------------
            # 【第二步】全局同步：是否所有卡都可以退出？
            # --------------------------
            exit_flag = torch.tensor([need_exit], dtype=torch.int32, device=device)
            dist.all_reduce(exit_flag, op=dist.ReduceOp.MAX, group=full_group)

            # 如果全局标志为1 → 所有卡一起退出
            if exit_flag.item() == 1:
                break
            # Launch all communications
            with torch.npu.stream(streams['tp']):
                dist.all_reduce(tp_tensor, op=dist.ReduceOp.SUM, group=full_group)
            with torch.npu.stream(streams['dp']):
                dist.all_reduce(dp_tensor, op=dist.ReduceOp.SUM, group=full_group)
            with torch.npu.stream(streams['ep']):
                dist.all_to_all(ep_output_list, ep_input_list, group=full_group)
            with torch.npu.stream(streams['sp']):
                dist.all_gather(sp_output, sp_local, group=full_group)
            with torch.npu.stream(streams['pp']):
                dist.all_gather(pp_output, pp_local, group=full_group)

            torch.npu.synchronize()

            # Only profile the very first step
            # if step == 0:
            #     prof.step()
            #     print(f"[Rank {rank}] First iteration profiled.")

            step += 1

            # Periodic logging (only rank 0)
            current_time = time.time()
            if rank == 0 and (current_time - last_log_time) >= LOG_INTERVAL_SECONDS:
                elapsed = current_time - start_time
                hours = elapsed / 3600
                iter_per_sec = step / elapsed if elapsed > 0 else 0
                total_gb = step * 5 * 4  # 5 ops × 4GB
                print(f"[LOG] Elapsed: {hours:.2f}h | Iter: {step} | "
                      f"Speed: {iter_per_sec:.2f} iter/s | Total traffic: {total_gb:.1f} GB")
                last_log_time = current_time



    if rank == 0:
        total_gb = step * 5 * 4
        print(f"✅ 24-hour stress test completed! Total iterations: {step}, Total traffic: {total_gb:.1f} GB")
        print("Profiling of first iteration saved to ./5d_24h_first_profile/")

    dist.destroy_process_group()


if __name__ == "__main__":
    main()

总结

在多卡并行通信测试中，时间控制虽灵活，但易因执行差异导致同步不一致。建议优先采用 step 控制 以保证通信次数一致；若必须使用时间控制，则需引入显式同步退出机制，确保所有卡在退出前完成同步操作。通过上述优化，可显著提升测试用例的稳定性与可靠性，为通信算子的性能验证提供坚实保障。