已合并
skip sub-comm creation when global process group was reinit'd and fix p2p longname err #38997
limuan创建于 6月23日
skip sub-comm creation when global process group was reinit'd and fix p2p longname err #38997
已合并
共 4 个文件变更+99-51
| @@ -102,6 +102,7 @@ class HcclSendRecvDistTest(TestCase): | |||
| 102 | for p in ps: | 102 | for p in ps: |
| 103 | p.join(2) | 103 | p.join(2) |
| 104 | 104 | ||
| 105 | + | ||
| 105 | 106 | ||
| 106 | def test_avoid_isend_irecv_hccl(self): | 107 | def test_avoid_isend_irecv_hccl(self): |
| 107 | self._test_multiprocess( | 108 | self._test_multiprocess( |
| @@ -109,6 +110,7 @@ class HcclSendRecvDistTest(TestCase): | |||
| 109 | torch.randn(400, 1024, 1024), | 110 | torch.randn(400, 1024, 1024), |
| 110 | HcclSendRecvDistTest._init_dist_hccl) | 111 | HcclSendRecvDistTest._init_dist_hccl) |
| 111 | 112 | ||
| 113 | + | ||
 | |||
| 112 | 114 | ||
| 113 | def test_avoid_batch_isend_irecv_hccl(self): | 115 | def test_avoid_batch_isend_irecv_hccl(self): |
| 114 | self._test_multiprocess( | 116 | self._test_multiprocess( |
| @@ -130,6 +130,9 @@ class SubCommDerivationTest(TestCase): | |||
| 130 | ranks = list(range(world_size)) | 130 | ranks = list(range(world_size)) |
| 131 | sub_pg = dist.new_group(backend='hccl', ranks=ranks) | 131 | sub_pg = dist.new_group(backend='hccl', ranks=ranks) |
| 132 | 132 | ||
| 133 | + global_tensor = xs[0].clone() | ||
| 134 | + dist.all_reduce(global_tensor, op=dist.ReduceOp.SUM) | ||
| 135 | + | ||
| 133 | with StderrCapture(label=f"allreduce_rank{rank}") as cap: | 136 | with StderrCapture(label=f"allreduce_rank{rank}") as cap: |
| 134 | dist.all_reduce(xs[0], group=sub_pg, op=dist.ReduceOp.SUM) | 137 | dist.all_reduce(xs[0], group=sub_pg, op=dist.ReduceOp.SUM) |
| 135 | 138 | ||
| @@ -162,6 +165,9 @@ class SubCommDerivationTest(TestCase): | |||
| 162 | tensor_pp = xs[0].clone() | 165 | tensor_pp = xs[0].clone() |
| 163 | tensor_dp = xs[0].clone() | 166 | tensor_dp = xs[0].clone() |
| 164 | 167 | ||
| 168 | + global_tensor = xs[0].clone() | ||
| 169 | + dist.all_reduce(global_tensor, op=dist.ReduceOp.SUM) | ||
| 170 | + | ||
| 165 | with StderrCapture(label=f"multi_comms_rank{rank}") as cap: | 171 | with StderrCapture(label=f"multi_comms_rank{rank}") as cap: |
| 166 | dist.all_reduce(tensor_tp, group=tp_pg, op=dist.ReduceOp.SUM) | 172 | dist.all_reduce(tensor_tp, group=tp_pg, op=dist.ReduceOp.SUM) |
| 167 | 173 | ||
| @@ -195,6 +201,9 @@ class SubCommDerivationTest(TestCase): | |||
| 195 | ranks = list(range(world_size)) | 201 | ranks = list(range(world_size)) |
| 196 | sub_pg = dist.new_group(backend='hccl', ranks=ranks) | 202 | sub_pg = dist.new_group(backend='hccl', ranks=ranks) |
| 197 | 203 | ||
| 204 | + global_tensor = xs[0].clone() | ||
| 205 | + dist.all_reduce(global_tensor, op=dist.ReduceOp.SUM) | ||
| 206 | + | ||
| 198 | ys = [torch.zeros(xs[0].shape, dtype=xs[0].dtype, device=xs[0].device) for _ in range(world_size)] | 207 | ys = [torch.zeros(xs[0].shape, dtype=xs[0].dtype, device=xs[0].device) for _ in range(world_size)] |
| 199 | 208 | ||
| 200 | with StderrCapture(label=f"allgather_rank{rank}") as cap: | 209 | with StderrCapture(label=f"allgather_rank{rank}") as cap: |
| @@ -226,6 +235,9 @@ class SubCommDerivationTest(TestCase): | |||
| 226 | ranks = list(range(world_size)) | 235 | ranks = list(range(world_size)) |
| 227 | sub_pg = dist.new_group(backend='hccl', ranks=ranks) | 236 | sub_pg = dist.new_group(backend='hccl', ranks=ranks) |
| 228 | 237 | ||
| 238 | + global_tensor = xs[0].clone() | ||
| 239 | + dist.all_reduce(global_tensor, op=dist.ReduceOp.SUM) | ||
| 240 | + | ||
| 229 | with StderrCapture(label=f"broadcast_rank{rank}") as cap: | 241 | with StderrCapture(label=f"broadcast_rank{rank}") as cap: |
| 230 | dist.broadcast(xs[0], src=0, group=sub_pg) | 242 | dist.broadcast(xs[0], src=0, group=sub_pg) |
| 231 | 243 | ||
| @@ -290,6 +302,9 @@ class SubCommDerivationTest(TestCase): | |||
| 290 | } | 302 | } |
| 291 | sub_pg = dist.new_group(backend='hccl', ranks=ranks, pg_options=options) | 303 | sub_pg = dist.new_group(backend='hccl', ranks=ranks, pg_options=options) |
| 292 | 304 | ||
| 305 | + global_tensor = xs[0].clone() | ||
| 306 | + dist.all_reduce(global_tensor, op=dist.ReduceOp.SUM) | ||
| 307 | + | ||
| 293 | with StderrCapture(label=f"custom_config_rank{rank}") as cap: | 308 | with StderrCapture(label=f"custom_config_rank{rank}") as cap: |
| 294 | dist.all_reduce(xs[0], group=sub_pg, op=dist.ReduceOp.SUM) | 309 | dist.all_reduce(xs[0], group=sub_pg, op=dist.ReduceOp.SUM) |
| 295 | 310 | ||
| @@ -316,6 +331,9 @@ class SubCommDerivationTest(TestCase): | |||
| 316 | ranks = list(range(world_size)) | 331 | ranks = list(range(world_size)) |
| 317 | sub_pg = dist.new_group(backend='hccl', ranks=ranks) | 332 | sub_pg = dist.new_group(backend='hccl', ranks=ranks) |
| 318 | 333 | ||
| 334 | + global_tensor = xs[0].clone() | ||
| 335 | + dist.all_reduce(global_tensor, op=dist.ReduceOp.SUM) | ||
| 336 | + | ||
| 319 | with StderrCapture(label=f"async_rank{rank}") as cap: | 337 | with StderrCapture(label=f"async_rank{rank}") as cap: |
| 320 | work = dist.all_reduce(xs[0], group=sub_pg, op=dist.ReduceOp.SUM, async_op=True) | 338 | work = dist.all_reduce(xs[0], group=sub_pg, op=dist.ReduceOp.SUM, async_op=True) |
| 321 | work.wait() | 339 | work.wait() |
| @@ -342,6 +360,9 @@ class SubCommDerivationTest(TestCase): | |||
| 342 | 360 | ||
| 343 | ranks = list(range(world_size)) | 361 | ranks = list(range(world_size)) |
| 344 | 362 | ||
| 363 | + global_tensor = xs[0].clone() | ||
| 364 | + dist.all_reduce(global_tensor, op=dist.ReduceOp.SUM) | ||
| 365 | + | ||
| 345 | sub_pg1 = dist.new_group(backend='hccl', ranks=ranks) | 366 | sub_pg1 = dist.new_group(backend='hccl', ranks=ranks) |
| 346 | with StderrCapture(label=f"recreate1_rank{rank}") as cap1: | 367 | with StderrCapture(label=f"recreate1_rank{rank}") as cap1: |
| 347 | dist.all_reduce(xs[0], group=sub_pg1, op=dist.ReduceOp.SUM) | 368 | dist.all_reduce(xs[0], group=sub_pg1, op=dist.ReduceOp.SUM) |
| @@ -376,6 +397,9 @@ class SubCommDerivationTest(TestCase): | |||
| 376 | ranks = list(range(world_size)) | 397 | ranks = list(range(world_size)) |
| 377 | sub_pg = dist.new_group(backend='hccl', ranks=ranks) | 398 | sub_pg = dist.new_group(backend='hccl', ranks=ranks) |
| 378 | 399 | ||
| 400 | + global_tensor = xs[0].clone() | ||
| 401 | + dist.all_reduce(global_tensor, op=dist.ReduceOp.SUM) | ||
| 402 | + | ||
| 379 | with StderrCapture(label=f"reduce_rank{rank}") as cap: | 403 | with StderrCapture(label=f"reduce_rank{rank}") as cap: |
| 380 | dist.reduce(xs[0], dst=0, op=dist.ReduceOp.SUM, group=sub_pg) | 404 | dist.reduce(xs[0], dst=0, op=dist.ReduceOp.SUM, group=sub_pg) |
| 381 | 405 | ||
| @@ -406,6 +430,9 @@ class SubCommDerivationTest(TestCase): | |||
| 406 | 430 | ||
| 407 | sub_pg = dist.new_group(backend='hccl', ranks=partial_ranks) | 431 | sub_pg = dist.new_group(backend='hccl', ranks=partial_ranks) |
| 408 | 432 | ||
| 433 | + global_tensor = xs[0].clone() | ||
| 434 | + dist.all_reduce(global_tensor, op=dist.ReduceOp.SUM) | ||
| 435 | + | ||
| 409 | if rank in partial_ranks: | 436 | if rank in partial_ranks: |
| 410 | with StderrCapture(label=f"partial_rank{rank}") as cap: | 437 | with StderrCapture(label=f"partial_rank{rank}") as cap: |
| 411 | dist.all_reduce(xs[0], group=sub_pg, op=dist.ReduceOp.SUM) | 438 | dist.all_reduce(xs[0], group=sub_pg, op=dist.ReduceOp.SUM) |
| @@ -2679,6 +2679,14 @@ void ProcessGroupHCCL::createHCCLCommOrigin( | |||
| 2679 | std::vector<c10_npu::NPUStream> &streamVal, | 2679 | std::vector<c10_npu::NPUStream> &streamVal, |
| 2680 | int p2pRank) | 2680 | int p2pRank) |
| 2681 | { | 2681 | { |
| 2682 | + bool isSub = !options_->global_ranks_in_group.empty(); | ||
| 2683 | + if (isSub) { | ||
| 2684 | + if (createHCCLCommSub(devicesKey, devices, commType, commConfig, hcclComms, streamVal, p2pRank, true)) { | ||
| 2685 | + return; | ||
| 2686 | + } | ||
| 2687 | + TORCH_NPU_HCCL_LOGI("Sub comm derivation failed in createHCCLCommOrigin, fallback to rootinfo."); | ||
| 2688 | + } | ||
| 2689 | + | ||
| 2682 | HcclRootInfo hcclID; | 2690 | HcclRootInfo hcclID; |
| 2683 | bool isSingleP2POp = commType == HcclCommType::P2P ? true : false; | 2691 | bool isSingleP2POp = commType == HcclCommType::P2P ? true : false; |
| 2684 | if (rank_ == 0 || (isSingleP2POp && p2pRank == 0)) { | 2692 | if (rank_ == 0 || (isSingleP2POp && p2pRank == 0)) { |
| @@ -2756,34 +2764,35 @@ bool ProcessGroupHCCL::createHCCLCommEx( | |||
| 2756 | return false; | 2764 | return false; |
| 2757 | } | 2765 | } |
| 2758 | c10_npu::OptionalNPUGuard npuGuard; | 2766 | c10_npu::OptionalNPUGuard npuGuard; |
| 2759 | - // global process group only; sub comm derivation is handled by createHCCLCommSub | 2767 | + // global process group |
| 2760 | - if (!(options_->global_ranks_in_group.empty() && commType == HcclCommType::DEFAULT)) { | 2768 | + if (options_->global_ranks_in_group.empty() && commType == HcclCommType::DEFAULT) { |
| 2761 | - TORCH_NPU_HCCL_LOGI("createHCCLCommEx only handles global comm with ranktable, skip for sub/P2P comm."); | 2769 | + auto startTime = std::chrono::steady_clock::now(); |
| 2762 | - return false; | 2770 | + for (size_t i = 0; i < devices.size(); ++i) { |
| 2763 | - } | 2771 | + int rank = getRank() * static_cast<int>(devices.size()) + static_cast<int>(i); |
| 2764 | - auto startTime = std::chrono::steady_clock::now(); | ||
| 2765 | - for (size_t i = 0; i < devices.size(); ++i) { | ||
| 2766 | - int rank = getRank() * static_cast<int>(devices.size()) + static_cast<int>(i); | ||
| 2767 | 2772 | ||
| 2768 | - npuGuard.set_index(devices[i].index()); | 2773 | + npuGuard.set_index(devices[i].index()); |
| 2769 | - HcclCommConfig config; | 2774 | + HcclCommConfig config; |
| 2770 | - if (commConfig == nullptr) { | 2775 | + if (commConfig == nullptr) { |
| 2771 | - config = createHcclCommConfigWithOptions(); | 2776 | + config = createHcclCommConfigWithOptions(); |
| 2772 | - commConfig = &config; | 2777 | + commConfig = &config; |
| 2778 | + } | ||
| 2779 | + auto comm = HCCLComm::createGlobalHcclComm(rankTableFile.c_str(), rank, commConfig); | ||
| 2780 | + if (comm == nullptr) { | ||
| 2781 | + TORCH_NPU_HCCL_LOGI("Create global hccl comm with ranktable failed, switch to original interface."); | ||
| 2782 | + return false; | ||
| 2783 | + } | ||
| 2784 | + hcclComms[i] = comm; | ||
| 2785 | + // Creates the HCCL streams | ||
| 2786 | + streamVal.push_back(getHcclNPUStream(devices[i])); | ||
| 2773 | } | 2787 | } |
| 2774 | - auto comm = HCCLComm::createGlobalHcclComm(rankTableFile.c_str(), rank, commConfig); | 2788 | + auto endTime = std::chrono::steady_clock::now(); |
| 2775 | - if (comm == nullptr) { | 2789 | + auto timeElapsed = std::chrono::duration_cast<std::chrono::milliseconds>(endTime - startTime); |
| 2776 | - TORCH_NPU_HCCL_LOGI("Create global hccl comm with ranktable failed, switch to original interface."); | 2790 | + TORCH_NPU_HCCL_LOGI("Create global hccl comm with ranktable success, take %d milliseconds", static_cast<int>(timeElapsed.count())); |
| 2777 | - return false; | 2791 | + return true; |
| 2778 | - } | ||
| 2779 | - hcclComms[i] = comm; | ||
| 2780 | - // Creates the HCCL streams | ||
| 2781 | - streamVal.push_back(getHcclNPUStream(devices[i])); | ||
| 2782 | } | 2792 | } |
| 2783 | - auto endTime = std::chrono::steady_clock::now(); | 2793 | + |
| 2784 | - auto timeElapsed = std::chrono::duration_cast<std::chrono::milliseconds>(endTime - startTime); | 2794 | + // sub process group |
| 2785 | - TORCH_NPU_HCCL_LOGI("Create global hccl comm with ranktable success, take %d milliseconds", static_cast<int>(timeElapsed.count())); | 2795 | + return createHCCLCommSub(devicesKey, devices, commType, commConfig, hcclComms, streamVal, p2pRank); |
| 2786 | - return true; | ||
| 2787 | } | 2796 | } |
| 2788 | 2797 | ||
| 2789 | bool ProcessGroupHCCL::createHCCLCommSub( | 2798 | bool ProcessGroupHCCL::createHCCLCommSub( |
| @@ -2793,7 +2802,8 @@ bool ProcessGroupHCCL::createHCCLCommSub( | |||
| 2793 | HcclCommConfig* commConfig, | 2802 | HcclCommConfig* commConfig, |
| 2794 | std::vector<std::shared_ptr<HCCLComm>> &hcclComms, | 2803 | std::vector<std::shared_ptr<HCCLComm>> &hcclComms, |
| 2795 | std::vector<c10_npu::NPUStream> &streamVal, | 2804 | std::vector<c10_npu::NPUStream> &streamVal, |
| 2796 | - int p2pRank) | 2805 | + int p2pRank, |
| 2806 | + bool isOrigin) | ||
| 2797 | { | 2807 | { |
| 2798 | if (!hcclCreateSubCommConfigExist()) { | 2808 | if (!hcclCreateSubCommConfigExist()) { |
| 2799 | TORCH_NPU_HCCL_LOGI("The hcclCreateSubCommConfig is not exist, switch to original interface."); | 2809 | TORCH_NPU_HCCL_LOGI("The hcclCreateSubCommConfig is not exist, switch to original interface."); |
| @@ -2804,17 +2814,24 @@ bool ProcessGroupHCCL::createHCCLCommSub( | |||
| 2804 | return false; | 2814 | return false; |
| 2805 | } | 2815 | } |
| 2806 | std::shared_ptr<HCCLComm> globalHcclComm = nullptr; | 2816 | std::shared_ptr<HCCLComm> globalHcclComm = nullptr; |
| 2807 | - try { | 2817 | + if (!isOrigin) { |
| 2808 | - globalHcclComm = global_->getHcclCommByDevices(devices); | 2818 | + try { |
| 2809 | - } catch (const std::exception& e) { | 2819 | + globalHcclComm = global_->getHcclCommByDevices(devices); |
| 2810 | - TORCH_NPU_HCCL_LOGI("Get global HCCL communicator failed: %s, switch to original interface.", e.what()); | 2820 | + } catch (const std::exception& e) { |
| 2811 | - return false; | 2821 | + TORCH_NPU_HCCL_LOGI("Get global HCCL communicator failed: %s, switch to original interface.", e.what()); |
| 2822 | + return false; | ||
| 2823 | + } | ||
| 2824 | + if (!globalHcclComm) { | ||
| 2825 | + TORCH_NPU_HCCL_LOGI("Create sub hccl comm failed, globalHcclComm is nullptr, switch to original interface."); | ||
| 2826 | + return false; | ||
| 2827 | + } | ||
| 2828 | + } else { | ||
| 2829 | + globalHcclComm = global_->getHcclCommByDevicesFromCache(devices); | ||
| 2830 | + if (!globalHcclComm) { | ||
| 2831 | + TORCH_NPU_HCCL_LOGI("Get global HCCL communicator failed, switch to original interface."); | ||
| 2832 | + return false; | ||
| 2833 | + } | ||
| 2812 | } | 2834 | } |
| 2813 | - if (!globalHcclComm) { | ||
| 2814 | - TORCH_NPU_HCCL_LOGI("Create sub hccl comm failed, globalHcclComm is nullptr, switch to original interface."); | ||
| 2815 | - return false; | ||
| 2816 | - } | ||
| 2817 | - | ||
| 2818 | c10_npu::OptionalNPUGuard npuGuard; | 2835 | c10_npu::OptionalNPUGuard npuGuard; |
| 2819 | uint64_t hcclid = (std::hash<string>{}(options_->group_id)); | 2836 | uint64_t hcclid = (std::hash<string>{}(options_->group_id)); |
| 2820 | auto subStartTime = std::chrono::steady_clock::now(); | 2837 | auto subStartTime = std::chrono::steady_clock::now(); |
| @@ -2855,7 +2872,7 @@ bool ProcessGroupHCCL::createHCCLCommSub( | |||
| 2855 | commConfig->hcclCommName[COMM_NAME_MAX_LENGTH - 1] = '\0'; | 2872 | commConfig->hcclCommName[COMM_NAME_MAX_LENGTH - 1] = '\0'; |
| 2856 | 2873 | ||
| 2857 | } | 2874 | } |
| 2858 | - if (strlen(commConfig->hcclUdi) > 0) { | 2875 | + if ((!isOrigin) && strlen(commConfig->hcclUdi) > 0) { |
| 2859 | 2876 | ||
| 2860 | torch_npu::toolkit::profiler::Utils::safe_strcpy_s(commConfig->hcclUdi, p2pName.c_str(), UDI_MAX_LENGTH); | 2877 | torch_npu::toolkit::profiler::Utils::safe_strcpy_s(commConfig->hcclUdi, p2pName.c_str(), UDI_MAX_LENGTH); |
| 2861 | 2878 | ||
| @@ -2965,19 +2982,7 @@ std::vector<std::shared_ptr<HCCLComm>>& ProcessGroupHCCL::createHCCLComm( | |||
| 2965 | }; | 2982 | }; |
| 2966 | at_npu::native::OpCommand::RunOpApiV3("hcclGroupEnd", hccl_call); | 2983 | at_npu::native::OpCommand::RunOpApiV3("hcclGroupEnd", hccl_call); |
| 2967 | } | 2984 | } |
| 2968 | - bool isSubComm = !(options_->global_ranks_in_group.empty() && commType == HcclCommType::DEFAULT); | 2985 | + if (!createHCCLCommEx(devicesKey, devices, commType, commConfig, hcclComms, streamVal, p2pRank)) { |
| 2969 | - bool created = false; | ||
| 2970 | - if (isSubComm) { | ||
| 2971 | - // Sub comm: derive from global_ via hcclCreateSubCommConfig (works for both ranktable and rootinfo) | ||
| 2972 | - created = createHCCLCommSub(devicesKey, devices, commType, commConfig, hcclComms, streamVal, p2pRank); | ||
| 2973 | - if (!created) { | ||
| 2974 | - TORCH_NPU_HCCL_LOGI("Sub comm derivation failed, fallback to original interface."); | ||
| 2975 | - } | ||
| 2976 | - } else { | ||
| 2977 | - // Global comm: try ranktable path first | ||
| 2978 | - created = createHCCLCommEx(devicesKey, devices, commType, commConfig, hcclComms, streamVal, p2pRank); | ||
| 2979 | - } | ||
| 2980 | - if (!created) { | ||
| 2981 | createHCCLCommOrigin(devicesKey, devices, commType, commConfig, hcclComms, streamVal, p2pRank); | 2986 | createHCCLCommOrigin(devicesKey, devices, commType, commConfig, hcclComms, streamVal, p2pRank); |
| 2982 | } | 2987 | } |
| 2983 | npuGuard.set_index(getDeviceForRank(getRank()).index()); | 2988 | npuGuard.set_index(getDeviceForRank(getRank()).index()); |
| @@ -3436,6 +3441,17 @@ ProcessGroupHCCL::Options::Options(bool is_high_priority_stream) | |||
| 3436 | { | 3441 | { |
| 3437 | } | 3442 | } |
| 3438 | 3443 | ||
| 3444 | +std::shared_ptr<HCCLComm> ProcessGroupHCCL::getHcclCommByDevicesFromCache(const std::vector<at::Device>& devices) | ||
| 3445 | +{ | ||
| 3446 | + const auto key = getKeyFromDevices(devices); | ||
| 3447 | + std::lock_guard<std::mutex> lock(mutex_); | ||
| 3448 | + auto it = devHCCLCommMap_.find(key); | ||
| 3449 | + if (it != devHCCLCommMap_.end() && !it->second.empty()) { | ||
| 3450 | + return it->second[0]; | ||
| 3451 | + } | ||
| 3452 | + return nullptr; | ||
| 3453 | +} | ||
| 3454 | + | ||
| 3439 | std::shared_ptr<HCCLComm> ProcessGroupHCCL::getHcclCommByDevices(const std::vector<at::Device>& devices) | 3455 | std::shared_ptr<HCCLComm> ProcessGroupHCCL::getHcclCommByDevices(const std::vector<at::Device>& devices) |
| 3440 | { | 3456 | { |
| 3441 | const auto key = getKeyFromDevices(devices); | 3457 | const auto key = getKeyFromDevices(devices); |
| @@ -759,6 +759,8 @@ public: | |||
| 759 | // may indicate that there is some sort of collective desynchronization. | 759 | // may indicate that there is some sort of collective desynchronization. |
| 760 | uint64_t getSequenceNumberForGroup() override; | 760 | uint64_t getSequenceNumberForGroup() override; |
| 761 | 761 | ||
| 762 | + std::shared_ptr<HCCLComm> getHcclCommByDevicesFromCache(const std::vector<at::Device>& devices); | ||
| 763 | + | ||
| 762 | std::shared_ptr<HCCLComm> getHcclCommByDevices(const std::vector<at::Device>& devices); | 764 | std::shared_ptr<HCCLComm> getHcclCommByDevices(const std::vector<at::Device>& devices); |
| 763 | 765 | ||
| 764 | int64_t getHcclComm(int rankid); | 766 | int64_t getHcclComm(int rankid); |
| @@ -1291,7 +1293,8 @@ private: | |||
| 1291 | HcclCommConfig* commConfig, | 1293 | HcclCommConfig* commConfig, |
| 1292 | std::vector<std::shared_ptr<HCCLComm>> &hcclComms, | 1294 | std::vector<std::shared_ptr<HCCLComm>> &hcclComms, |
| 1293 | std::vector<c10_npu::NPUStream> &streamVal, | 1295 | std::vector<c10_npu::NPUStream> &streamVal, |
| 1294 | - int p2pRank); | 1296 | + int p2pRank, |
| 1297 | + bool isOrigin = false); | ||
| 1295 | 1298 | ||
| 1296 | void createHCCLCommForZeroCopy( | 1299 | void createHCCLCommForZeroCopy( |
| 1297 | std::vector<std::shared_ptr<HCCLComm>> &hcclComms, | 1300 | std::vector<std::shared_ptr<HCCLComm>> &hcclComms, |
🟡 Medium Priority
变更行：test_avoid_record_stream.py 第 105、113 行添加了
@unittest.skip("Temporarily disable")，禁用了test_avoid_isend_irecv_hccl和test_avoid_batch_isend_irecv_hccl两个测试。这些测试验证 P2P 的 isend/irecv 操作，其底层通信器创建路径经过
createHCCLCommSub（P2P 类型的通信器），该函数在本 PR 中被重构（新增isOrigin参数并改变了调用链逻辑）。在修改了相关代码路径的情况下禁用这些测试，将导致回归无法被检测到。PR 标题为 "skip sub-comm creation when global process group was reinit'd"，但跳过的测试并未附带 issue 编号或说明原因，"Temporarily disable" 可能演变为永久禁用。
建议：如果这些测试因代码变更而失败，应在 PR 中修复相关代码路径（例如 P2P 通信器创建逻辑），而非禁用测试。如果确实需要临时禁用，应在 skip 消息中添加 issue 编号或明确的重新启用条件，例如
@unittest.skip("Temporarily disable for PR#38997; re-enable after <condition>")。