Line data Source code
1 : /**
2 : * Copyright (c) 2025 Huawei Technologies Co., Ltd.
3 : * This program is free software, you can redistribute it and/or modify it under the terms and conditions of
4 : * CANN Open Software License Agreement Version 2.0 (the "License").
5 : * Please refer to the License for details. You may not use this file except in compliance with the License.
6 : * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND, EITHER EXPRESS OR IMPLIED,
7 : * INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT, MERCHANTABILITY, OR FITNESS FOR A PARTICULAR PURPOSE.
8 : * See LICENSE in the root of the software repository for the full text of the License.
9 : */
10 :
11 : #include "communicator_impl.h"
12 : #include <memory>
13 : #include <op_type.h>
14 : #include <adapter_error_manager_pub.h>
15 : #include "orion_adapter_rts.h"
16 : #include "orion_adapter_hal.h"
17 : #include "hccl_exception.h"
18 : #include "null_ptr_exception.h"
19 : #include "runtime_api_exception.h"
20 : #include "exception_util.h"
21 : #include "hccp_hdc_manager.h"
22 : #include "hccp_peer_manager.h"
23 : #include "rdma_handle_manager.h"
24 : #include "env_config.h"
25 : #include "coll_service_ai_cpu_impl.h"
26 : #include "checkcrc.h"
27 : #include "task_exception_handler.h"
28 : #include "coll_service_device_mode.h"
29 : #include "dlprof_function.h"
30 : #include "kfc.h"
31 : #include "op_params_checker.h"
32 : #include "diff_rank_updater.h"
33 : #include "coll_alg_component_builder.h"
34 : #include "coll_alg_component.h"
35 : #include "hccl_common_v2.h"
36 : #include "tp_manager.h"
37 : #include "hccl_aiv_utils.h"
38 : #include "rts_1ton_cnt_notify.h"
39 : #include "rts_cnt_notify.h"
40 : #include "stream_utils.h"
41 : #include "port.h"
42 : #include "net_instance.h"
43 : #include "ascend_hal_base.h"
44 : #include "acl/acl_rt.h"
45 : #include "types.h"
46 : #include "json_parser.h"
47 : #include "rank_graph_builder_bridge.h"
48 : #include "ccu_jetty_mgr.h"
49 : #include "comm_topo_desc.h"
50 : #include "hostdpu/flush_manager.h"
51 : #include "hostdpu/dpu_kernel_entrance.h"
52 : #include "p2p_enable_manager.h"
53 : #include "adapter_error_manager_pub.h"
54 : #include "ccu_context_all_to_all_v_mesh1d.h"
55 : #include "topo_addr_info.h"
56 :
57 : namespace Hccl {
58 : constexpr u64 HCCL_CCL_COMM_FIXED_CALC_BUFFER_SIZE = (1 * 1024 * 1024); // 指定bufferSize的单位为MB
59 : constexpr u64 HCCL_AIV_OFFLOAD_TAG_BUFFER_SIZE = (4 * 1024 * 1024); // 指定bufferSize的单位为MB
60 : constexpr u64 HCCL_MC2_ON_AICPU_FIXED_CALC_BUFFER_SIZE = 1 * HCCL_CCL_COMM_FIXED_CALC_BUFFER_SIZE; // MC2适配AICPU，额外需要1M
61 : std::atomic<u32> Hccl::CommunicatorImpl::globalIndex(0);
62 : constexpr u64 HCCL_CCL_AIV_TAG_BUFFER_SIZE = 2; // 指定存放aiv tag的大小为2M
63 : constexpr u32 HCCL_CCL_AIV_CLEAR_STEP_MAX = 1000; // aiv tag算子下发时++，大于1000置位
64 : constexpr u32 BASE_BIT = 1; // 用于左移设置二进制数的特定位
65 : constexpr u64 SHARE_HBM_MEMORY_SIZE = (100 * 1024 * 1024);
66 : constexpr u64 DPU_TASKEXCEPTION_MEMORY_SIZE = 10; // DPU TASKEXCEPTION共享内存大小 |stopflag[1]|hcclret[2]|hcclret[2]|
67 : constexpr u64 ALIGN_4K = 4096U;
68 : constexpr const char* DPUTAG = "DPUTAG";
69 : constexpr const char* DPUTASKEXCEPTION = "DPUTASKEXCEPTION";
70 : constexpr u64 INDEPENDENT_OP_BUFFER_SIZE_TIMES = 2; //自定义算子buffer倍数
71 : constexpr uint8_t DEVICE_SIGNAL_SECOND = 2;
72 : constexpr uint8_t DEVICE_SIGNAL_THIRD = 3;
73 : constexpr uint32_t TEMP_DEV_TYPE_DPU = 0; // 临时适配，后续rts接口上库之后使用rts的定义
74 : static std::atomic<u32> g_commNum(0); // 一个进程内创建的通信域数量
75 :
76 :
77 : // 支持零拷贝算子的白名单
78 : std::set<OpType> opWhiteSet = {
79 : OpType::BROADCAST,
80 : OpType::ALLTOALL,
81 : OpType::ALLTOALLV,
82 : OpType::SEND,
83 : OpType::RECV,
84 : OpType::ALLGATHER
85 : };
86 :
87 13 : static void PrintBackTrace(HcclException &e)
88 : {
89 13 : auto backTraces = e.GetBackTraceStrings();
90 13 : std::for_each(backTraces.begin(), backTraces.end(), [](string item) {
91 585 : HCCL_INFO(item.c_str());
92 195 : });
93 13 : }
94 :
95 2 : static std::shared_ptr<RankGraph> TakeRankGraphOwnership(std::unique_ptr<RankGraph> &rankGraph)
96 : {
97 2 : CHK_PRT_THROW(rankGraph == nullptr, HCCL_ERROR("[%s] input rankGraph is nullptr.", __func__),
98 : NullPtrException, "input rankGraph is nullptr.");
99 :
100 2 : const RankGraphBuilderBridge *bridge = GetRankGraphBuilderBridge();
101 2 : CHK_PRT_THROW(bridge == nullptr || bridge->adoptRankGraph == nullptr,
102 : HCCL_ERROR("[%s] RankGraphBuilder ownership bridge is not registered.", __func__),
103 : InternalException, "RankGraphBuilder ownership bridge is not registered.");
104 :
105 2 : std::shared_ptr<RankGraph> sharedRankGraph;
106 2 : HcclResult ret = bridge->adoptRankGraph(std::move(rankGraph), sharedRankGraph);
107 2 : CHK_PRT_THROW(ret != HCCL_SUCCESS,
108 : HCCL_ERROR("[%s] adopt RankGraph failed, errNo[0x%016llx].", __func__, HCCL_ERROR_CODE(ret)),
109 : InternalException, "adopt RankGraph failed.");
110 2 : CHK_PRT_THROW(sharedRankGraph == nullptr,
111 : HCCL_ERROR("[%s] adopted RankGraph is nullptr.", __func__),
112 : InternalException, "adopted RankGraph is nullptr.");
113 2 : return sharedRankGraph;
114 0 : }
115 :
116 5 : static void PrepareRankGraphMetadata(const RankTableInfo &inputRankTableInfo, const TopoInfo &inputTopoInfo,
117 : std::unique_ptr<RankTableInfo> &outputRankTableInfo, std::shared_ptr<TopoInfo> &outputTopoInfo)
118 : {
119 5 : auto preparedRankTableInfo = std::make_unique<RankTableInfo>(inputRankTableInfo);
120 5 : auto preparedTopoInfo = std::make_shared<TopoInfo>(inputTopoInfo);
121 :
122 5 : outputRankTableInfo = std::move(preparedRankTableInfo);
123 5 : outputTopoInfo = std::move(preparedTopoInfo);
124 5 : }
125 :
126 46 : HcclResult CommunicatorImpl::Init(const CommParams &commParams, const std::string &ranktableM,
127 : const HcclCommConfig &config)
128 : {
129 46 : if (!initFlag) {
130 45 : initFlag = true;
131 : try {
132 45 : InitCommonData(commParams, config);
133 45 : InitHccpHdc(); // tsdOpen + rainit
134 45 : InitRankGraph(ranktableM);
135 38 : CHK_RET(InitCommResource(commParams));
136 7 : } catch (HcclException &e) {
137 21 : HCCL_ERROR(e.what());
138 7 : PrintBackTrace(e);
139 7 : return e.GetErrorCode();
140 7 : } catch (exception &e) {
141 0 : HCCL_ERROR(e.what());
142 0 : return HcclResult::HCCL_E_INTERNAL;
143 0 : } catch (...) {
144 0 : HCCL_ERROR("Unknown error occurs!");
145 0 : return HcclResult::HCCL_E_INTERNAL;
146 0 : }
147 38 : return HcclResult::HCCL_SUCCESS;
148 : }
149 3 : HCCL_ERROR("Repeated calling init method!");
150 1 : return HcclResult::HCCL_E_INTERNAL;
151 : }
152 :
153 38 : HcclResult CommunicatorImpl::InitCommResource(const CommParams &commParams)
154 : {
155 38 : HrtSetDevice(devLogicId);
156 38 : if (IsNeedDpu()) {
157 5 : InitHccpPeer();
158 : }
159 38 : AppendLocalDieIdForLinks();
160 38 : InitCcuSuperFastLoad();
161 38 : InitNotifyManager();
162 38 : InitStreamManager();
163 38 : InitPreResource();
164 38 : InitSocketManager();
165 38 : InitRmaConnManager();
166 38 : InitDataBufferManager();
167 38 : InitMemTransportManager();
168 38 : InitHostDeviceSyncNotifyManager();
169 38 : InitUbMemoryTransportMgr();
170 38 : CollAlgComponentInit(); // 初始化算法组件
171 38 : RegisterAicpuKernel();
172 38 : InitCollService();
173 38 : InitTraceManager();
174 38 : DlProfFunction::GetInstance().DlProfFunctionInit();
175 38 : InitMirrorTaskManager();
176 38 : CHK_RET(InitProfilingReporter());
177 38 : InitTaskExceptionHandler();
178 38 : InitHDCommunicate();
179 38 : notifyTimeoutCfg.Init();
180 38 : SetCommStatus(CommStatus::COMM_READY);
181 38 : SnapShotParser::GetInstance().SerializeCommonInfo(commParams, config, std::move(ranktableInfo), topoInfo, staticBinaryInfo);
182 38 : InitOneSidedService();
183 38 : RegisterKernel();
184 38 : InitDpuKernel();
185 38 : return HCCL_SUCCESS;
186 : }
187 :
188 39 : void CommunicatorImpl::InitDpuKernel() {
189 39 : std::unordered_set<IpAddress> hostIps = GetHostIpFromRankGraph();
190 39 : if (hostIps.empty()) {
191 39 : return;
192 : }
193 0 : for (auto ip: hostIps) {
194 0 : FlushManager::GetInstance().initFlushHandle(ip, devPhyId);
195 : }
196 0 : HCCL_INFO("[InitDpuKernel]all FlushHandle init success.");
197 : /* kernel Launch */
198 0 : CHK_RET_THROW(RuntimeApiException, "InitAndLaunchDpuKernel Failed", InitAndLaunchDpuKernel());
199 0 : CHK_RET_THROW(RuntimeApiException, "InitAndLaunchAicpuKernel Failed", InitAndLaunchAicpuKernel()); // 下aicpukernel,将taskexception共享内存保存到aicpu全局map中
200 39 : }
201 :
202 39 : std::unordered_set<IpAddress> CommunicatorImpl::GetHostIpFromRankGraph()
203 : {
204 117 : HCCL_INFO("[GetHostIpFromRankGraph]Start get host ip.");
205 39 : std::unordered_set<IpAddress> ips;
206 39 : if (rankGraph->GetPeer(myRank) == nullptr) {
207 0 : HCCL_ERROR("[GetHostIpFromRankGraph] rankGraph peer is null!");
208 0 : return ips;
209 : }
210 39 : std::vector<std::shared_ptr<NetInstance::ConnInterface>> interfaces = rankGraph->GetPeer(myRank)->GetIfaces();
211 64 : for (auto interface : interfaces) {
212 : // 找到所有在host上和LinkProtocol有rdma的ip进行注册
213 25 : if (interface->GetPos() == AddrPosition::HOST && interface->GetLinkProtocols().count(LinkProtocol::ROCE) != 0) {
214 0 : IpAddress ip = interface->GetAddr();
215 0 : ips.insert(ip);
216 : }
217 25 : }
218 117 : HCCL_INFO("[GetHostIpFromRankGraph] Successfully completed: GetHostIp finished.");
219 39 : return ips;
220 39 : }
221 :
222 0 : HcclResult CommunicatorImpl::Init(const CommParams &commParams, const RankTableInfo &ranktable,
223 : const HcclCommConfig &config)
224 : {
225 0 : if (!initFlag) {
226 0 : initFlag = true;
227 : try {
228 0 : InitCommonData(commParams, config);
229 0 : InitHccpHdc(); // tsdOpen + rainit
230 0 : InitRankGraph(ranktable);
231 0 : CHK_RET(InitCommResource(commParams));
232 0 : } catch (HcclException &e) {
233 0 : HCCL_ERROR(e.what());
234 0 : PrintBackTrace(e);
235 0 : return e.GetErrorCode();
236 0 : } catch (exception &e) {
237 0 : HCCL_ERROR(e.what());
238 0 : return HcclResult::HCCL_E_INTERNAL;
239 0 : } catch (...) {
240 0 : HCCL_ERROR("Unknown error occurs!");
241 0 : return HcclResult::HCCL_E_INTERNAL;
242 0 : }
243 0 : return HcclResult::HCCL_SUCCESS;
244 : }
245 0 : HCCL_ERROR("Repeated calling init method!");
246 0 : return HcclResult::HCCL_E_INTERNAL;
247 : }
248 :
249 1 : HcclResult CommunicatorImpl::Init(const CommParams &commParams, std::unique_ptr<RankGraph> &inputRankGraph, DevId inputDevLogicId)
250 : {
251 1 : if (!initFlag) {
252 1 : initFlag = true;
253 : try {
254 1 : HrtSetDevice(inputDevLogicId);
255 1 : InitCommonData(commParams);
256 1 : InitRankGraph(inputRankGraph);
257 1 : HrtSetDevice(devLogicId);
258 1 : if (IsNeedDpu()) {
259 0 : InitHccpPeer();
260 : }
261 1 : InitHccpHdc();
262 1 : AppendLocalDieIdForLinks();
263 1 : InitCcuSuperFastLoad();
264 1 : InitNotifyManager();
265 1 : InitStreamManager();
266 1 : InitSocketManager();
267 1 : InitRmaConnManager();
268 1 : InitDataBufferManager();
269 1 : InitMemTransportManager();
270 1 : InitHostDeviceSyncNotifyManager();
271 1 : InitUbMemoryTransportMgr();
272 1 : CollAlgComponentInit();
273 1 : RegisterAicpuKernel();
274 1 : InitCollService();
275 1 : InitTraceManager();
276 1 : InitHDCommunicate();
277 1 : InitMirrorTaskManager();
278 1 : CHK_RET(InitProfilingReporter());
279 1 : InitTaskExceptionHandler();
280 1 : RegisterKernel();
281 1 : InitDpuKernel();
282 1 : SetCommStatus(CommStatus::COMM_READY);
283 0 : } catch (HcclException &e) {
284 0 : HCCL_ERROR(e.what());
285 0 : PrintBackTrace(e);
286 0 : return e.GetErrorCode();
287 0 : } catch (exception &e) {
288 0 : HCCL_ERROR(e.what());
289 0 : return HcclResult::HCCL_E_INTERNAL;
290 0 : } catch (...) {
291 0 : HCCL_ERROR("Unknown error occurs!");
292 0 : return HcclResult::HCCL_E_INTERNAL;
293 0 : }
294 1 : return HcclResult::HCCL_SUCCESS;
295 : }
296 0 : HCCL_ERROR("Repeated calling init method!");
297 0 : return HcclResult::HCCL_E_INTERNAL;
298 : }
299 :
300 0 : HcclResult CommunicatorImpl::Init(const CommParams &commParams, std::unique_ptr<RankGraph> &inputRankGraph,
301 : HcclCommConfig &subConfig, DevId inputDevLogicId)
302 : {
303 0 : if (!initFlag) {
304 0 : initFlag = true;
305 0 : TRY_CATCH_RETURN(
306 : HrtSetDevice(inputDevLogicId);
307 : InitCommonData(commParams, subConfig);
308 : InitHccpHdc();
309 : InitCcuSuperFastLoad();
310 : InitNotifyManager();
311 : InitStreamManager();
312 : InitSocketManager();
313 : InitRmaConnManager();
314 : InitDataBufferManager();
315 : InitMemTransportManager();
316 : InitHostDeviceSyncNotifyManager();
317 : InitTraceManager();
318 : InitHDCommunicate();
319 : notifyTimeoutCfg.Init();
320 : InitRankGraph(inputRankGraph);
321 : if (IsNeedDpu()) {
322 : InitHccpPeer();
323 : }
324 : AppendLocalDieIdForLinks();
325 : InitUbMemoryTransportMgr();
326 : CollAlgComponentInit();
327 : RegisterAicpuKernel();
328 : InitCollService();
329 : DlProfFunction::GetInstance().DlProfFunctionInit();
330 : InitMirrorTaskManager();
331 : CHK_RET(InitProfilingReporter());
332 : InitTaskExceptionHandler();
333 : RegisterKernel();
334 : InitDpuKernel();
335 : SetCommStatus(CommStatus::COMM_READY);
336 : SnapShotParser::GetInstance().SerializeSubCommInfo(commParams, subConfig, rankIdsVec, staticBinaryInfo);
337 : );
338 0 : return HcclResult::HCCL_SUCCESS;
339 : } else {
340 0 : HCCL_ERROR("Repeated calling init method!");
341 0 : return HcclResult::HCCL_E_INTERNAL;
342 : }
343 : }
344 :
345 1 : HcclResult CommunicatorImpl::CreateSubComm(const CommParams &subCommParams, const std::vector<u32> &rankIds,
346 : CommunicatorImpl *subCommImpl)
347 : {
348 65 : TRY_CATCH_RETURN(
349 : if (initFlag) {
350 : // 创建子虚拟拓扑
351 : std::unique_ptr<RankGraph> subRankGraph = rankGraph->CreateSubRankGraph(rankIds);
352 : // 初始化子通信域
353 : CHK_RET(subCommImpl->Init(subCommParams, subRankGraph, devLogicId));
354 : auto rankIpPortMap = GetSocketManager().GetSubCommDeviceServerListenPortMap(rankIds);
355 : RankIpPortMapPtr rankIpPortMapPtr = std::make_shared<decltype(rankIpPortMap)>(std::move(rankIpPortMap));
356 : CHK_RET(subCommImpl->SetRankIpPortMap(rankIpPortMapPtr));
357 : subCommImpl->GetSocketManager().SetDeviceServerListenPortMap(*rankIpPortMapPtr);
358 : return HcclResult::HCCL_SUCCESS;
359 : } else {
360 : std::string msg = StringFormat("CreateSubComm fail, communicator has not been initialized, please check.");
361 : THROW<InternalException>(msg);
362 : }
363 : );
364 : HCCL_ERROR("CreateSubComm fail !");
365 : return HcclResult::HCCL_E_INTERNAL;
366 : }
367 :
368 1 : HcclResult CommunicatorImpl::CreateSubComm(const CommParams &subCommParams, const std::vector<u32> &rankIds,
369 : CommunicatorImpl *subCommImpl, HcclCommConfig &subConfig)
370 : {
371 65 : TRY_CATCH_RETURN(
372 : if (initFlag) {
373 : // 创建子虚拟拓扑
374 : std::unique_ptr<RankGraph> subRankGraph = rankGraph->CreateSubRankGraph(rankIds);
375 : subCommImpl->rankIdsVec = rankIds;
376 : HCCL_INFO("[%s]rankIds size[%u], rankIdsVec size[%u]", __func__, rankIds.size(), subCommImpl->rankIdsVec.size());
377 : // 初始化子通信域
378 : CHK_RET(subCommImpl->Init(subCommParams, subRankGraph, subConfig, devLogicId));
379 : auto rankIpPortMap = GetSocketManager().GetSubCommDeviceServerListenPortMap(rankIds);
380 : RankIpPortMapPtr rankIpPortMapPtr = std::make_shared<decltype(rankIpPortMap)>(std::move(rankIpPortMap));
381 : CHK_RET(subCommImpl->SetRankIpPortMap(rankIpPortMapPtr));
382 : subCommImpl->GetSocketManager().SetDeviceServerListenPortMap(*rankIpPortMapPtr);
383 : return HcclResult::HCCL_SUCCESS;
384 : } else {
385 : std::string msg = StringFormat("CreateSubComm fail, communicator has not been initialized, please check.");
386 : THROW<InternalException>(msg);
387 : }
388 : );
389 : HCCL_ERROR("CreateSubComm fail !");
390 : return HcclResult::HCCL_E_INTERNAL;
391 : }
392 :
393 1 : void CommunicatorImpl::RefreshSubmittedOpcnt()
394 : {
395 1 : if (currentCollOperator->opType == OpType::SEND || currentCollOperator->opType == OpType::RECV) {
396 0 : sendRecvIndex++;
397 0 : submittedOpCnt = sendRecvIndex;
398 : } else {
399 1 : collOpIndex++;
400 1 : submittedOpCnt = collOpIndex;
401 : }
402 3 : HCCL_INFO("[%s] end, opType[%s], submittedOpCnt[%u], sendRecvIndex[%u], collOpIndex[%u].", __func__,
403 : currentCollOperator->opType.Describe().c_str(), submittedOpCnt, sendRecvIndex, collOpIndex);
404 1 : }
405 :
406 12 : void CommunicatorImpl::SingleRankProc(const CollOpParams &opParams, void *stream) const
407 : {
408 23 : if (opParams.opType == Hccl::OpType::BATCHSENDRECV || opParams.opType == Hccl::OpType::SEND
409 23 : || opParams.opType == Hccl::OpType::RECV) {
410 6 : HCCL_WARNING("[CommunicatorImpl][%s] ranksize == 1 is not support BATCHSENDRECV SEND RECV", __func__);
411 2 : return;
412 : }
413 10 : if (opParams.sendBuf == opParams.recvBuf) {
414 21 : HCCL_WARNING("[CommunicatorImpl][%s] sendBuf == recvBuf, return success", __func__);
415 7 : return;
416 : }
417 3 : u64 len{0};
418 3 : if (opParams.opType == Hccl::OpType::ALLTOALL) {
419 1 : len = DataTypeSizeGet(opParams.all2AllDataDes.sendType) * opParams.all2AllDataDes.sendCount;
420 2 : } else if (opParams.opType == Hccl::OpType::ALLTOALLV) {
421 1 : len = DataTypeSizeGet(opParams.all2AllVDataDes.sendType) * *(static_cast<const u64 *>(opParams.all2AllVDataDes.sendCounts));
422 1 : } else if (opParams.opType == Hccl::OpType::ALLTOALLVC) {
423 0 : len = DataTypeSizeGet(opParams.all2AllVCDataDes.sendType) * *(static_cast<const u64 *>(opParams.all2AllVCDataDes.sendCountMatrix));
424 : } else {
425 1 : len = DataTypeSizeGet(opParams.dataType) * opParams.count;
426 : }
427 :
428 9 : HCCL_INFO("[CommunicatorImpl][%s] sendBuf[%p], recvBuf[%p], len[%llu].", __func__, opParams.sendBuf, opParams.recvBuf, len);
429 3 : if (len > 0) {
430 3 : HrtMemAsyncCopy(opParams.recvBuf, len, opParams.sendBuf, len, ACL_MEMCPY_DEVICE_TO_DEVICE, stream);
431 : }
432 : }
433 :
434 12 : bool CommunicatorImpl::TryFastCcuLaunch(const CollOpParams &opParams, aclrtStream const stream)
435 : {
436 12 : InitCcuSuperFastLoad(); // 存在profiling开关在多次下发算子时动态变化的场景，每次下发流程中都需要更新开关
437 16 : superFasterLoad = (opParams.opType == OpType::ALLREDUCE || opParams.opType == OpType::ALLGATHER ||
438 6 : opParams.opType == OpType::REDUCESCATTER || opParams.opType == OpType::BROADCAST ||
439 6 : opParams.opType == OpType::ALLTOALL || opParams.opType == OpType::REDUCE ||
440 16 : opParams.opType == OpType::SCATTER || opParams.opType == OpType::ALLTOALLV
441 : );
442 15 : bool canUpdate = superFasterLoad && (commExecuteConfig.accState == AcceleratorState::CCU_MS ||
443 3 : commExecuteConfig.accState == AcceleratorState::CCU_SCHED);
444 12 : if (OpType::ALLTOALL == opParams.opType) {
445 0 : ccuParamsMappingKey = {static_cast<u32>(opParams.reduceOp), static_cast<u32>(opParams.all2AllDataDes.sendType), static_cast<u32>(opParams.all2AllDataDes.sendCount)};
446 12 : } else if (OpType::ALLTOALLV == opParams.opType) {
447 0 : ccuParamsMappingKey = {static_cast<u32>(opParams.reduceOp), static_cast<u32>(opParams.all2AllVDataDes.sendType), 0};
448 12 : } else if (OpType::BROADCAST == opParams.opType || OpType::SCATTER == opParams.opType) {
449 0 : ccuParamsMappingKey = {static_cast<u32>(opParams.root), static_cast<u32>(opParams.dataType), static_cast<u32>(opParams.count)};
450 : } else {
451 12 : ccuParamsMappingKey = {static_cast<u32>(opParams.reduceOp), static_cast<u32>(opParams.dataType), static_cast<u32>(opParams.count)};
452 : }
453 12 : auto &ccuParamsMapping = colCcuParamMapping[opParams.opType];
454 12 : auto ccuParamsMappingKeyIter = ccuParamsMapping.find(ccuParamsMappingKey);
455 12 : bool isCCUChangeModel = canUpdate && ccuParamsMappingKeyIter != ccuParamsMapping.end();
456 12 : if (!isCCUChangeModel) {
457 7 : return false;
458 : }
459 5 : CachedCCUParams ¶ms = ccuParamsMappingKeyIter->second;
460 :
461 5 : if (opParams.opType == OpType::ALLTOALLV && params.insType != CcuInstType::CCU_ALLTOALLV_MESH_1D_DIRECT) {
462 0 : return false;
463 : }
464 5 : if (enableProfilingEnv) {
465 1 : uint64_t beginTime = DlProfFunction::GetInstance().dlMsprofSysCycleTime();
466 1 : UpdateProfStat();
467 1 : auto dfxOpInfo = std::make_shared<DfxOpInfo>();
468 1 : CovertToCurrentCollOperator(id, opParams, OpMode::OPBASE);
469 1 : dfxOpInfo->op_ = *GetCurrentCollOperator();
470 1 : dfxOpInfo->tag_ = dfxOpInfo->op_.opTag;
471 1 : dfxOpInfo->algType_ = GetCurAlgName().c_str();
472 1 : dfxOpInfo->commIndex_ = GetIdIndex();
473 1 : dfxOpInfo->comm_ = this;
474 1 : dfxOpInfo->beginTime_ = DlProfFunction::GetInstance().dlMsprofSysCycleTime();
475 1 : dfxOpInfo->commId_ = id;
476 1 : dfxOpInfo->opIndex_ = opIndex;
477 1 : GetMirrorTaskManager().SetCurrDfxOpInfo(dfxOpInfo);
478 1 : ExecuteFastCcuLaunch(opParams, stream, params);
479 1 : ReportProfInfo(beginTime, opParams.staticShape, true);
480 1 : } else {
481 4 : ExecuteFastCcuLaunch(opParams, stream, params);
482 : }
483 5 : return true;
484 : }
485 :
486 9 : static void FastCcuLaunchSaveDfxTaskInfo(const CommunicatorImpl &comm, const TaskParam &taskParam, bool isMaster,
487 : const u32 remoteRankId = INVALID_VALUE_RANKID)
488 : {
489 : u32 taskId;
490 : u32 streamId;
491 9 : HrtGetTaskIdAndStreamID(taskId, streamId);
492 :
493 : std::unique_ptr<TaskInfo> taskInfo = std::make_unique<TaskInfo>(streamId, taskId, remoteRankId, taskParam,
494 9 : comm.GetMirrorTaskManager().GetCurrDfxOpInfo(), isMaster);
495 :
496 27 : HCCL_INFO("Begin to AddTaskInfo: streamId[%lu], taskId[%lu], remoteRankId[%u].", streamId, taskId, remoteRankId);
497 9 : comm.GetMirrorTaskManager().AddTaskInfo(std::move(taskInfo));
498 9 : }
499 :
500 0 : void CommunicatorImpl::FillAllToAllVArgs(const CollOpParams &opParams, rtCcuTaskInfo_t *&ccuParams) const
501 : {
502 0 : std::vector<uint64_t> args;
503 0 : CcuContextAllToAllVMesh1D::RefreshArgs(opParams, rankSize, args, myRank);
504 0 : rtCcuTaskInfo_t *currCcuParam = ccuParams;
505 0 : for (u32 i = 0; i < args.size(); i++) {
506 : // skip token info
507 0 : if (i == 2) {
508 0 : continue;
509 : }
510 0 : currCcuParam->args[i % RT_CCU_SQE_ARGS_LEN] = args[i];
511 0 : if ((i + 1) % RT_CCU_SQE_ARGS_LEN == 0) {
512 0 : currCcuParam += 1;
513 : }
514 : }
515 0 : }
516 :
517 3 : void CommunicatorImpl::ExecuteFastCcuLaunch(const CollOpParams &opParams, aclrtStream const stream, CachedCCUParams ¶ms)
518 : {
519 : static thread_local int slaveIndex = 0;
520 : static thread_local u32 mStreamId = 0;
521 : static thread_local u32 value = 0;
522 : static thread_local Rts1ToNCntNotify *cntNotify1ToN = nullptr;
523 3 : static thread_local u32 timeout = notifyTimeoutCfg.GetNotifyTimeout();
524 :
525 3 : rtCcuTaskInfo_t *&ccuParams = params.ccuParams;
526 :
527 3 : if (params.insType == CcuInstType::CCU_ALLTOALLV_MESH_1D_DIRECT) {
528 0 : FillAllToAllVArgs(opParams, ccuParams);
529 6 : } else if (params.insType == CcuInstType::CCU_ALLTOALL_MESH_1D_2DIE ||
530 6 : params.insType == CcuInstType::CCU_ALLGATHER_MESH_1D_2DIE ||
531 3 : params.insType == CcuInstType::CCU_REDUCE_SCATTER_MESH_1D_2DIE) {
532 0 : for (std::size_t i = 0; i < params.totalCounts; ++i) {
533 0 : (void)memcpy_s(&ccuParams[i].args[0], sizeof(ccuParams[i].args[0]), &opParams.sendBuf,
534 : sizeof(ccuParams[i].args[0]));
535 0 : (void)memcpy_s(&ccuParams[i].args[1], sizeof(ccuParams[i].args[1]), &opParams.recvBuf,
536 : sizeof(ccuParams[i].args[1]));
537 : }
538 : } else {
539 3 : (void)memcpy_s(&ccuParams[0].args[0], sizeof(ccuParams[0].args[0]), &opParams.sendBuf,
540 : sizeof(ccuParams[0].args[0]));
541 3 : (void)memcpy_s(&ccuParams[0].args[1], sizeof(ccuParams[0].args[1]), &opParams.recvBuf,
542 : sizeof(ccuParams[0].args[1]));
543 : }
544 :
545 3 : auto vector_zero_count = params.count[0];
546 3 : auto &opbaseStream = GetStreamManager().opbase;
547 3 : auto mStream = params.isSlave ? opbaseStream->GetSlave(slaveIndex)->GetPtr() : stream;
548 3 : u32 streamNum = params.count.size();
549 3 : if (streamNum > 1) {
550 3 : timeout = notifyTimeoutCfg.GetNotifyTimeout();
551 3 : mStreamId = params.isSlave ? opbaseStream->GetSlave(slaveIndex++)->GetId() : HrtGetStreamId(mStream);
552 3 : cntNotify1ToN = GetCcuStreamSyncNotifyManager().GetRts1ToNCntNotify(mStreamId);
553 : // launch LocalPostTo on stream
554 3 : value = 0;
555 9 : for (u32 i = 0; i < streamNum - 1; ++i) {
556 6 : value |= BASE_BIT << i;
557 : }
558 3 : cntNotify1ToN->PostValue(value, mStream);
559 : }
560 3 : if (taskExceptionEnv || enableProfilingEnv) {
561 3 : params.taskParams[0].beginTime = DlProfFunction::GetInstance().dlMsprofSysCycleTime();
562 3 : SuperFastLoad(ccuParams, mStream, vector_zero_count);
563 3 : params.taskParams[0].endTime = DlProfFunction::GetInstance().dlMsprofSysCycleTime();
564 3 : FastCcuLaunchSaveDfxTaskInfo(*this, params.taskParams[0], (!params.isSlave));
565 : } else {
566 0 : SuperFastLoad(ccuParams, mStream, vector_zero_count);
567 : }
568 :
569 3 : if (streamNum > 1) {
570 3 : RtsCntNotify *cntNotifyNTo1 = GetCcuStreamSyncNotifyManager().GetRtsNTo1CntNotify(mStreamId);
571 3 : opbaseStream->RegisterMaster(std::make_unique<Stream>(stream));
572 : // launch LocalWaitFrom on stream
573 3 : cntNotifyNTo1->WaitValue(value, timeout, mStream);
574 9 : for (std::size_t i = 0, len = streamNum - 1; i < len; ++i) {
575 6 : u32 bitValue = BASE_BIT << i;
576 6 : auto slave = opbaseStream->GetSlave(slaveIndex++);
577 6 : auto master = opbaseStream->GetMaster();
578 6 : GetStreamManager().CaptureSlaveStream(master, slave);
579 6 : cntNotify1ToN->WaitBits(bitValue, timeout, *slave);
580 6 : if (taskExceptionEnv || enableProfilingEnv) {
581 6 : params.taskParams[i + 1].beginTime = DlProfFunction::GetInstance().dlMsprofSysCycleTime();
582 6 : SuperFastLoad(ccuParams + params.count[i], slave->GetPtr(), params.count[i + 1]);
583 6 : params.taskParams[i + 1].endTime = DlProfFunction::GetInstance().dlMsprofSysCycleTime();
584 6 : FastCcuLaunchSaveDfxTaskInfo(*this, params.taskParams[i + 1], slave->IsMaster());
585 : }
586 : else{
587 0 : SuperFastLoad(ccuParams + params.count[i], slave->GetPtr(), params.count[i + 1]);
588 : }
589 : // launch localPostTo on extra streams
590 6 : cntNotifyNTo1->PostBits(bitValue, *slave);
591 : }
592 : }
593 3 : if(params.insType == CcuInstType::CCU_REDUCE_SCATTER_MESH_1D_2DIE) {
594 : //硬编码
595 0 : if (taskExceptionEnv || enableProfilingEnv) {
596 0 : TaskParam taskParam{};
597 0 : taskParam.beginTime = DlProfFunction::GetInstance().dlMsprofSysCycleTime();
598 0 : aclrtReduceKind rtReduceOp = static_cast<aclrtReduceKind>(static_cast<int>(RtReduceOpGet(opParams.reduceOp)));
599 0 : aclDataType rtDataType = static_cast<aclDataType>(static_cast<int>(RtDataTypeGet(opParams.dataType)));
600 0 : constexpr std::size_t myScratchPlace = 4;
601 0 : const u32 scratchSize = ccuParams[0].args[myScratchPlace];
602 0 : auto src = reinterpret_cast<void *>(ccuParams[0].args[3]);
603 0 : auto dst = reinterpret_cast<void *>(ccuParams[0].args[1]);
604 0 : HrtReduceAsync(dst, scratchSize, src, scratchSize, rtReduceOp, rtDataType, stream);
605 0 : taskParam.taskType = TaskParamType::TASK_REDUCE_INLINE;
606 0 : taskParam.endTime = DlProfFunction::GetInstance().dlMsprofSysCycleTime();
607 0 : taskParam.taskPara.Reduce.src = src;
608 0 : taskParam.taskPara.Reduce.dst = dst;
609 0 : taskParam.taskPara.Reduce.size = scratchSize;
610 0 : taskParam.taskPara.Reduce.notifyID = INVALID_VALUE_NOTIFYID;
611 0 : taskParam.taskPara.Reduce.linkType = DfxLinkType::ONCHIP;
612 0 : taskParam.taskPara.Reduce.dataType = DataTypeToHcclDataType(opParams.dataType);
613 0 : taskParam.taskPara.Reduce.reduceOp = ReduceOpToHcclReduceOp(opParams.reduceOp);
614 0 : FastCcuLaunchSaveDfxTaskInfo(*this, taskParam, true, GetMyRank()); // stream为主流
615 0 : } else {
616 0 : aclrtReduceKind rtReduceOp = static_cast<aclrtReduceKind>(static_cast<int>(RtReduceOpGet(opParams.reduceOp)));
617 0 : aclDataType rtDataType = static_cast<aclDataType>(static_cast<int>(RtDataTypeGet(opParams.dataType)));
618 0 : constexpr std::size_t myScratchPlace = 4;
619 0 : const u32 scratchSize = ccuParams[0].args[myScratchPlace];
620 0 : auto src = reinterpret_cast<void *>(ccuParams[0].args[3]);
621 0 : auto dst = reinterpret_cast<void *>(ccuParams[0].args[1]);
622 0 : HrtReduceAsync(dst, scratchSize, src, scratchSize, rtReduceOp, rtDataType, stream);
623 : }
624 : }
625 :
626 3 : slaveIndex = 0;
627 3 : collOpIndex++;
628 3 : submittedOpCnt = collOpIndex;
629 3 : opBaseOpIndex++;
630 3 : opIndex++;
631 3 : SetCommStatus(CommStatus::COMM_READY);
632 3 : }
633 :
634 0 : HcclResult CommunicatorImpl::SetAivControledCoreNum(bool isAiv)
635 : {
636 0 : if (isAiv) {
637 0 : u32 numBlocksLimit = MAX_NUM_BLOCKS;
638 0 : aclError acl_ret = aclrtGetResInCurrentThread(ACL_RT_DEV_RES_VECTOR_CORE, &numBlocksLimit);
639 0 : CHK_PRT_RET(acl_ret != ACL_SUCCESS,
640 : HCCL_ERROR("[CommunicatorImpl::SetAivControledCoreNum] aclrtGetResInCurrentThread failed, ret=[%d]", acl_ret),
641 : HCCL_E_PARA);
642 0 : CHK_PRT_RET(numBlocksLimit < 1,
643 : HCCL_ERROR("[CommunicatorImpl::SetAivControledCoreNum] block num less than 1, block num[%u]", numBlocksLimit),
644 : HCCL_E_PARA);
645 0 : currentCollOperator->numBlocksLimit = numBlocksLimit;
646 0 : HCCL_INFO("[CommunicatorImpl::SetAivControledCoreNum] Aiv core limit is [%u].", numBlocksLimit);
647 : }
648 0 : return HCCL_SUCCESS;
649 : }
650 :
651 12 : static HcclResult MatchAclgraph(const rtStream_t stream, bool &isCapture)
652 : {
653 12 : rtModel_t rtModel = nullptr;
654 12 : CHK_RET(GetStreamCaptureInfo(stream, rtModel, isCapture));
655 12 : return HCCL_SUCCESS;
656 : }
657 :
658 0 : HcclResult CommunicatorImpl::OffloadResourcePre(std::string &opTag, const CollOpParams &opParams)
659 : {
660 0 : CollOffloadOpResReq resReq;
661 0 : auto dataSize = opParams.count * DataTypeSizeGet(opParams.dataType);
662 0 : auto dataType = DataTypeToHcclDataType(opParams.dataType);
663 0 : CHK_RET(CalcCollOffloadOpRes(opParams.opType, dataSize, dataType, resReq));
664 :
665 : // 设定workspace内存资源
666 0 : std::vector<rtStream_t> slaveStreams;
667 0 : slaveStreams.resize(resReq.requiredSubQueNum);
668 0 : for (u64 i = 0; i < resReq.requiredSubQueNum; ++i) {
669 0 : slaveStreams[i] = static_cast<rtStream_t>(std::make_unique<Stream>(true, false).get());
670 : }
671 0 : CHK_RET(SetCollOffloadSlaveStreams(opTag, slaveStreams));
672 0 : CHK_RET(SetCollOffloadScratchBuf(opTag, reinterpret_cast<void *>(GetCclBuffer()->GetAddr()),
673 : GetCclBuffer()->GetSize()));
674 0 : return HCCL_SUCCESS;
675 0 : }
676 :
677 21 : HcclResult CommunicatorImpl::LoadOpbasedCollOp(const CollOpParams &opParams, void *stream)
678 : {
679 : try {
680 21 : isLoadOp = true;
681 45 : CHK_RET(CheckCommStatus());
682 : // 等待通信域状态为Ready，执行算子下发
683 18 : WaitReady();
684 18 : SnapShotParser::GetInstance().SetIsNeedLoadOp(false);
685 18 : if (rankSize == 1) {
686 30 : HCCL_WARNING("[CommunicatorImpl][%s] ranksize == 1, enter SingleRankProc", __func__);
687 10 : SingleRankProc(opParams, stream);
688 10 : return HcclResult::HCCL_SUCCESS;
689 : }
690 : // 判断是否为aclgraph
691 8 : bool isCapture = false; // isCapture为true表示aclgraph
692 8 : CHK_RET(MatchAclgraph(stream, isCapture));
693 8 : if (!isCapture && TryFastCcuLaunch(opParams, stream)) { // 若是aclgraph则不走快速下发
694 3 : return HcclResult::HCCL_SUCCESS;
695 : }
696 5 : curOpParams = opParams;
697 5 : CovertToCurrentCollOperator(id, opParams, OpMode::OPBASE);
698 3 : opExecuteConfig = commExecuteConfig;
699 3 : ExecAlgSelect(opParams, OpMode::OPBASE); // 根据配置选择对应的collService
700 2 : if (dynamic_cast<CollServiceDefaultImpl *>(collService) != nullptr) {
701 3 : HCCL_ERROR("Opbase mode is not supported in expanding on the host in 950");
702 1 : return HcclResult::HCCL_E_NOT_SUPPORT;
703 : }
704 1 : bool isAiv = (opExecuteConfig.accState == AcceleratorState::AIV || opExecuteConfig.accState == AcceleratorState::AIV_ONLY);
705 1 : SetCommStatus(CommStatus::COMM_READY);
706 4 : CHK_RET(OpParamsChecker::CheckOpDataTypeOpbase(opParams, GetOpCcuFeatureFlag(), GetOpAiCpuTSFeatureFlag(), isAiv));
707 :
708 : // AICPU aclgraph场景传入的stream被capture且算子时支持零拷贝算法的,会切换到图模式
709 0 : if (opExecuteConfig.accState == AcceleratorState::AICPU_TS && isCapture && (opWhiteSet.find(opParams.opType) != opWhiteSet.end())) {
710 0 : std::string tag = opParams.opTag + "_" + std::to_string(tagResourceIndex_++);
711 0 : OffloadResourcePre(tag, opParams);
712 0 : HCCL_INFO("[CommunicatorImpl][%s]current op support zero copy in aicpu aclgraph, change to offload", __func__);
713 0 : return LoadOffloadCollOp(tag, opParams, stream);
714 0 : }
715 0 : CHK_RET(SetAivControledCoreNum(isAiv));
716 :
717 : // 避免transport建链前，通讯域被摧毁
718 0 : SetCommStatus(CommStatus::COMM_INUSE);
719 0 : if (opParams.sendBuf != nullptr) {
720 0 : PrintMemoryAttr(opParams.sendBuf);
721 : }
722 0 : if (opParams.recvBuf != nullptr) {
723 0 : PrintMemoryAttr(opParams.recvBuf);
724 : }
725 0 : uint64_t beginTime = DlProfFunction::GetInstance().dlMsprofSysCycleTime();
726 :
727 : // 更新开关状态
728 0 : UpdateProfStat();
729 0 : collService->LoadWithOpBasedMode(*currentCollOperator, std::make_unique<Stream>(stream));
730 0 : if (++aivTag > HCCL_CCL_AIV_CLEAR_STEP_MAX) {
731 0 : aivTag = 1;
732 : }
733 : // ReportProfInfok:opinfo, allTaskInfo
734 0 : bool cachedReq = opParams.staticShape || isCapture;
735 0 : ReportProfInfo(beginTime, cachedReq, true);
736 0 : RefreshSubmittedOpcnt();
737 0 : opBaseOpIndex++;
738 0 : opIndex++;
739 0 : SetCommStatus(CommStatus::COMM_READY);
740 3 : } catch (HcclException &e) {
741 1 : SetCommStatus(CommStatus::COMM_READY);
742 3 : HCCL_ERROR(e.what());
743 1 : PrintBackTrace(e);
744 1 : u32 idxHcclException = GetSubmittedOpCnt();
745 3 : HCCL_ERROR("SubmittedOpCnt: %u, OperatorParams: %s", idxHcclException, opParams.Describe().c_str());
746 1 : return e.GetErrorCode();
747 2 : } catch (exception &e) {
748 1 : SetCommStatus(CommStatus::COMM_READY);
749 3 : HCCL_ERROR(e.what());
750 1 : u32 idxException = GetSubmittedOpCnt();
751 3 : HCCL_ERROR("SubmittedOpCnt: %u, OperatorParams: %s", idxException, opParams.Describe().c_str());
752 1 : return HcclResult::HCCL_E_INTERNAL;
753 2 : } catch (...) {
754 1 : SetCommStatus(CommStatus::COMM_READY);
755 1 : u32 idxOthers = GetSubmittedOpCnt();
756 3 : HCCL_ERROR("SubmittedOpCnt: %u, OperatorParams: %s", idxOthers, opParams.Describe().c_str());
757 3 : HCCL_ERROR("Unknown error occurs!");
758 1 : return HcclResult::HCCL_E_INTERNAL;
759 1 : }
760 0 : return HcclResult::HCCL_SUCCESS;
761 : }
762 :
763 26 : HcclResult CommunicatorImpl::CheckCommStatus() const
764 : {
765 26 : if (GetCommStatus() == CommStatus::COMM_ERROR) {
766 12 : HCCL_ERROR("Comm has been error, can not load opbased operator now!");
767 4 : return HcclResult::HCCL_E_INTERNAL;
768 : }
769 :
770 22 : if (isSuspended) {
771 3 : HCCL_ERROR("Comm has been suspended, can not load opbased operator now!");
772 1 : return HcclResult::HCCL_E_SUSPENDING;
773 : }
774 21 : return HcclResult::HCCL_SUCCESS;
775 : }
776 :
777 6 : HcclResult CommunicatorImpl::AllocCollOpResource(const CollOpParams &opParams, void **addr)
778 : {
779 : try {
780 6 : if (opParams.commEngine != HcclAccelerator::AICPU_TS) {
781 3 : HCCL_ERROR("[CommunicatorImpl][%s] Only AICPU_TS is supported for aicpu unfold on mc2. input is %s", __func__, opParams.commEngine.Describe().c_str());
782 5 : return HCCL_E_NOT_SUPPORT;
783 : }
784 11 : CHK_RET(CheckCommStatus());
785 :
786 3 : WaitReady();
787 3 : curOpParams = opParams;
788 3 : CovertToCurrentCollOperator(id, opParams, OpMode::OPBASE, false);
789 3 : opExecuteConfig = commExecuteConfig;
790 3 : ExecAlgSelect(opParams, OpMode::OPBASE);
791 3 : CHK_PTR_NULL(collService);
792 3 : if (dynamic_cast<CollServiceDefaultImpl *>(collService) != nullptr) {
793 3 : HCCL_ERROR("The op base is not supported in expanding on the host in 950 with MC2.");
794 1 : return HcclResult::HCCL_E_NOT_SUPPORT;
795 : }
796 :
797 2 : SetCommStatus(CommStatus::COMM_READY);
798 5 : CHK_RET(OpParamsChecker::CheckOpDataTypeOpbase(opParams, GetOpCcuFeatureFlag(), GetOpAiCpuTSFeatureFlag(), false));
799 1 : SetCommStatus(CommStatus::COMM_INUSE);
800 1 : std::string opAlgTag = opParams.opTag + "_" + curAlgName;
801 1 : CHK_RET(collService->AllocCollOpResource(*currentCollOperator, opAlgTag, addr));
802 1 : SetCommStatus(CommStatus::COMM_READY);
803 1 : } catch (HcclException &e) {
804 0 : SetCommStatus(CommStatus::COMM_READY);
805 0 : HCCL_ERROR(e.what());
806 0 : PrintBackTrace(e);
807 0 : HCCL_ERROR("AllocCollOpResource OperatorParams: %s", opParams.Describe().c_str());
808 0 : return e.GetErrorCode();
809 0 : } catch (exception &e) {
810 0 : SetCommStatus(CommStatus::COMM_READY);
811 0 : HCCL_ERROR(e.what());
812 0 : HCCL_ERROR("AllocCollOpResource OperatorParams: %s", opParams.Describe().c_str());
813 0 : return HcclResult::HCCL_E_INTERNAL;
814 0 : } catch (...) {
815 0 : SetCommStatus(CommStatus::COMM_READY);
816 0 : HCCL_ERROR("AllocCollOpResource OperatorParams: %s", opParams.Describe().c_str());
817 0 : HCCL_ERROR("Unkown error occurs!");
818 0 : return HcclResult::HCCL_E_INTERNAL;
819 0 : }
820 1 : return HcclResult::HCCL_SUCCESS;
821 : }
822 :
823 1 : HcclResult CommunicatorImpl::CalcCollOffloadOpRes(const OpType opType, u64 dataSize, HcclDataType dataType, CollOffloadOpResReq &resReq)
824 : {
825 3 : HCCL_INFO("[CommunicatorImpl][%s] start, opType[%s], dataSize[%llu].", __func__, opType.Describe().c_str(),
826 : dataSize);
827 : try {
828 : // 资源计算
829 : HcclResult errCode
830 1 : = collAlgComponent->CalcResOffload(opType, dataSize, dataType, GetCommExecuteConfig(), resReq); // 通信域粒度
831 1 : if (errCode != HcclResult::HCCL_SUCCESS) {
832 : std::string msg
833 : = StringFormat("[CommunicatorImpl][%s] Error occurs when call collAlgComponent.CalcResOffload, "
834 : "error code: %d",
835 0 : __func__, errCode);
836 0 : HCCL_ERROR(msg.c_str());
837 0 : return errCode;
838 0 : }
839 0 : } catch (HcclException &e) {
840 0 : HCCL_ERROR(e.what());
841 0 : return e.GetErrorCode();
842 0 : } catch (exception &e) {
843 0 : HCCL_ERROR(e.what());
844 0 : return HcclResult::HCCL_E_INTERNAL;
845 0 : } catch (...) {
846 0 : HCCL_ERROR("Unknown error occurs!");
847 0 : return HcclResult::HCCL_E_INTERNAL;
848 0 : }
849 3 : HCCL_INFO("[CommunicatorImpl][%s] end.", __func__);
850 1 : return HcclResult::HCCL_SUCCESS;
851 : }
852 :
853 0 : HcclResult CommunicatorImpl::SetCollOffloadSlaveStreams(const std::string &opTag,
854 : std::vector<void *> slaveStreams)
855 : {
856 : try {
857 0 : HCCL_INFO("[CommunicatorImpl][%s] start, opTag[%s].", __func__, opTag.c_str());
858 : // 将slaveStreams注册到streamManager中
859 0 : RegisterOffloadSlaveStreams(opTag, slaveStreams);
860 0 : HCCL_INFO("[CommunicatorImpl][%s] end.", __func__);
861 0 : } catch (HcclException &e) {
862 0 : HCCL_ERROR(e.what());
863 0 : return e.GetErrorCode();
864 0 : } catch (exception &e) {
865 0 : HCCL_ERROR(e.what());
866 0 : return HcclResult::HCCL_E_INTERNAL;
867 0 : } catch (...) {
868 0 : HCCL_ERROR("Unknown error occurs!");
869 0 : return HcclResult::HCCL_E_INTERNAL;
870 0 : }
871 0 : return HcclResult::HCCL_SUCCESS;
872 : }
873 :
874 6 : HcclResult CommunicatorImpl::SetCollOffloadScratchBuf(const std::string &opTag,
875 : void *scratchMemPtr,
876 : u64 requiredScratchMemSize)
877 : {
878 : try {
879 18 : HCCL_INFO("[CommunicatorImpl][%s] start, opTag[%s] requiredScratchMemSize[%llu].", __func__, opTag.c_str(), requiredScratchMemSize);
880 : // 将scratchBuf注册到dataBufManager中
881 6 : RegisterOffloadScratchBuffer(opTag, scratchMemPtr, requiredScratchMemSize);
882 18 : HCCL_INFO("[CommunicatorImpl][%s] end.", __func__);
883 0 : } catch (HcclException &e) {
884 0 : HCCL_ERROR(e.what());
885 0 : return e.GetErrorCode();
886 0 : } catch (exception &e) {
887 0 : HCCL_ERROR(e.what());
888 0 : return HcclResult::HCCL_E_INTERNAL;
889 0 : } catch (...) {
890 0 : HCCL_ERROR("Unknown error occurs!");
891 0 : return HcclResult::HCCL_E_INTERNAL;
892 0 : }
893 6 : return HcclResult::HCCL_SUCCESS;
894 : }
895 :
896 0 : void CommunicatorImpl::RegisterOffloadSlaveStreams(const std::string &opTag, std::vector<void *> slaveStreams) const
897 : {
898 0 : StreamManager &sm = GetStreamManager();
899 0 : sm.offload->RegisterSlaves(opTag, slaveStreams);
900 0 : }
901 :
902 6 : void CommunicatorImpl::RegisterOffloadScratchBuffer(const std::string &opTag, void *scratchMemPtr,
903 : u64 requiredScratchMemSize)
904 : {
905 6 : auto scratchBuffer = DevBuffer::Create(reinterpret_cast<uintptr_t>(scratchMemPtr), requiredScratchMemSize);
906 6 : if(scratchBuffer){
907 6 : offloadScrachBufferMap[opTag] = scratchBuffer;
908 18 : HCCL_RUN_INFO("[CommunicatorImpl] offloadScratchBuffer register, opTag[%s], offloadScrachBufferAddr[%llu], "
909 : "offloadScrachBufferBufSize[%llu]M",
910 : opTag.c_str(), scratchBuffer->GetAddr(),
911 : scratchBuffer->GetSize() / HCCL_CCL_COMM_FIXED_CALC_BUFFER_SIZE);
912 : }
913 6 : }
914 :
915 8 : HcclResult CommunicatorImpl::LoadOffloadCollOp(std::string &opTag, const CollOpParams &opParams, void *stream)
916 : {
917 : try {
918 24 : HCCL_INFO("CommunicatorImpl::LoadOffloadCollOp dataType[%s]", opParams.dataType.Describe().c_str());
919 8 : isLoadOp = true;
920 8 : curOpParams = opParams;
921 8 : if (GetCommStatus() == CommStatus::COMM_ERROR) {
922 9 : HCCL_ERROR("Comm has been error, can not offload operator now!");
923 7 : return HcclResult::HCCL_E_INTERNAL;
924 : }
925 :
926 5 : if (isSuspended) {
927 0 : HCCL_ERROR("Comm has been suspended, can not offload operator now!");
928 0 : return HcclResult::HCCL_E_SUSPENDING;
929 : }
930 :
931 : // 等待通信域状态为Ready，执行算子下发
932 5 : WaitReady();
933 5 : SnapShotParser::GetInstance().SetIsNeedLoadOp(false);
934 5 : if (rankSize == 1) {
935 3 : HCCL_WARNING("[CommunicatorImpl][%s] ranksize == 1, enter SingleRankProc", __func__);
936 1 : SingleRankProc(opParams, stream);
937 1 : return HcclResult::HCCL_SUCCESS;
938 : }
939 4 : uint64_t beginTime = DlProfFunction::GetInstance().dlMsprofSysCycleTime();
940 :
941 : // 更新开关状态
942 4 : UpdateProfStat();
943 : // 判断是否为aclgraph(aicpu场景零拷贝会切图模式)
944 4 : bool isCapture = false; // isCapture为true表示aclgraph，profiling需要
945 4 : CHK_RET(MatchAclgraph(stream, isCapture));
946 12 : HCCL_INFO("CommunicatorImpl::LoadOffloadCollOp opParams dataType[%s]", opParams.dataType.Describe().c_str());
947 4 : CovertToCurrentCollOperator(opTag, opParams, OpMode::OFFLOAD);
948 12 : HCCL_INFO("CommunicatorImpl::LoadOffloadCollOp currentCollOperator dataType[%s]", currentCollOperator->dataType.Describe().c_str());
949 : // 图模式算子加载选择CollService
950 4 : opExecuteConfig = commExecuteConfig;
951 4 : ExecAlgSelect(opParams, OpMode::OFFLOAD);
952 :
953 4 : if (opExecuteConfig.accState == AcceleratorState::HOSTCPU_TS) { // 950不支持HOST_TS模式
954 3 : HCCL_ERROR("[CommunicatorImpl::LoadOffloadCollOp] HOSTCPU_TS is not support.");
955 1 : return HcclResult::HCCL_E_NOT_SUPPORT;
956 : }
957 :
958 3 : bool isAiv = (opExecuteConfig.accState == AcceleratorState::AIV || opExecuteConfig.accState == AcceleratorState::AIV_ONLY);
959 9 : CHK_RET(OpParamsChecker::CheckOpDataTypeOffload(opParams, GetOpCcuFeatureFlag(), GetOpAiCpuTSFeatureFlag(), isAiv)); // 算子粒度
960 :
961 1 : if (isAiv) {
962 0 : currentCollOperator->numBlocksLimit = aivCoreLimit;
963 0 : HCCL_INFO("[CommunicatorImpl::LoadOffloadCollOp] Aiv core limit is [%u].", aivCoreLimit);
964 : }
965 1 : if (isAiv && aivClearEnable) {
966 0 : aivOffloadTag = 1;
967 1 : } else if (isAiv) {
968 0 : aivOffloadTag++;
969 : }
970 :
971 : // 避免transport建链前，通讯域被摧毁
972 1 : SetCommStatus(CommStatus::COMM_INUSE);
973 1 : collService->LoadWithOffloadMode(*currentCollOperator, std::make_unique<Stream>(stream));
974 1 : SetCommStatus(CommStatus::COMM_READY);
975 1 : bool cachedReq = opParams.staticShape || isCapture;
976 1 : ReportProfInfo(beginTime, cachedReq, isCapture); // profiling对于aclgraph场景的处理与单算子一致
977 1 : opIndex++;
978 0 : } catch (HcclException &e) {
979 0 : SetCommStatus(CommStatus::COMM_READY);
980 0 : HCCL_ERROR(e.what());
981 0 : return e.GetErrorCode();
982 0 : } catch (exception &e) {
983 0 : SetCommStatus(CommStatus::COMM_READY);
984 0 : HCCL_ERROR(e.what());
985 0 : return HcclResult::HCCL_E_INTERNAL;
986 0 : } catch (...) {
987 0 : SetCommStatus(CommStatus::COMM_READY);
988 0 : HCCL_ERROR("Unknown error occurs!");
989 0 : return HcclResult::HCCL_E_INTERNAL;
990 0 : }
991 1 : return HcclResult::HCCL_SUCCESS;
992 : }
993 :
994 18 : void CommunicatorImpl::CalcA2ASendRecvMem(const CollOpParams &opParams, u64 &sendSize, u64 &recvSize, bool isHcomSelectAlg) const
995 : {
996 18 : u64 sendCount = 0;
997 18 : u64 recvCount = 0;
998 18 : u32 sendTypeSize = 0;
999 18 : u32 recvTypeSize = 0;
1000 18 : if (opParams.opType == OpType::ALLTOALLV && !isHcomSelectAlg) { // alltoallv场景hcomSeletAlg不填写alltoallv的参数
1001 2 : for (u32 i = 0; i < rankSize; i++) {
1002 1 : CHECK_NULLPTR((static_cast<const u64 *>(opParams.all2AllVDataDes.sendCounts) + i),
1003 2 : StringFormat("%s failed, opParams.all2AllVDataDes.sendCounts[%u] is nullptr", __func__, i));
1004 1 : CHECK_NULLPTR((static_cast<const u64 *>(opParams.all2AllVDataDes.sdispls) + i),
1005 2 : StringFormat("%s failed, opParams.all2AllVDataDes.sdispls[%u] is nullptr", __func__, i));
1006 1 : CHECK_NULLPTR((static_cast<const u64 *>(opParams.all2AllVDataDes.recvCounts) + i),
1007 2 : StringFormat("%s failed, opParams.all2AllVDataDes.recvCounts[%u] is nullptr", __func__, i));
1008 1 : CHECK_NULLPTR((static_cast<const u64 *>(opParams.all2AllVDataDes.rdispls) + i),
1009 2 : StringFormat("%s failed, opParams.all2AllVDataDes.rdispls[%u] is nullptr", __func__, i));
1010 1 : u64 curSendCount = *(static_cast<const u64 *>(opParams.all2AllVDataDes.sendCounts) + i) +
1011 1 : *(static_cast<const u64 *>(opParams.all2AllVDataDes.sdispls) + i);
1012 1 : sendCount = std::max(sendCount, curSendCount);
1013 1 : u64 curRecvCount = *(static_cast<const u64 *>(opParams.all2AllVDataDes.recvCounts) + i) +
1014 1 : *(static_cast<const u64 *>(opParams.all2AllVDataDes.rdispls) + i);
1015 1 : recvCount = std::max(recvCount, curRecvCount);
1016 : }
1017 1 : sendTypeSize = DataTypeSizeGet(opParams.all2AllVDataDes.sendType);
1018 1 : recvTypeSize = DataTypeSizeGet(opParams.all2AllVDataDes.recvType);
1019 17 : } else if (opParams.opType == OpType::ALLTOALLVC && !isHcomSelectAlg) { // alltoallvc场景hcomSeletAlg不填写alltoallvc的参数
1020 2 : for (u32 i = 0; i < rankSize; i++) {
1021 1 : CHECK_NULLPTR((static_cast<const u64 *>(opParams.all2AllVCDataDes.sendCountMatrix) + myRank * rankSize + i),
1022 2 : StringFormat("%s failed, opParams.all2AllVCDataDes.sendCountMatrix[%u] is nullptr", __func__, (myRank * rankSize + i)));
1023 1 : sendCount += *(static_cast<const u64 *>(opParams.all2AllVCDataDes.sendCountMatrix) +
1024 1 : myRank * rankSize + i);
1025 1 : recvCount += *(static_cast<const u64 *>(opParams.all2AllVCDataDes.sendCountMatrix) +
1026 1 : myRank + rankSize * i);
1027 : }
1028 1 : sendTypeSize = DataTypeSizeGet(opParams.all2AllVCDataDes.sendType);
1029 1 : recvTypeSize = DataTypeSizeGet(opParams.all2AllVCDataDes.recvType);
1030 : } else {
1031 16 : sendCount = opParams.all2AllDataDes.sendCount * rankSize;
1032 16 : recvCount = opParams.all2AllDataDes.recvCount * rankSize;
1033 16 : sendTypeSize = DataTypeSizeGet(opParams.all2AllDataDes.sendType);
1034 16 : recvTypeSize = DataTypeSizeGet(opParams.all2AllDataDes.recvType);
1035 : }
1036 18 : sendSize = sendCount * sendTypeSize;
1037 18 : recvSize = recvCount * recvTypeSize;
1038 18 : }
1039 :
1040 18 : void CommunicatorImpl::ConvertCollOperatorA2A(const CollOpParams &opParams, bool isLaunch, bool isHcomSelectAlg)
1041 : {
1042 18 : if (currentCollOperator == nullptr) {
1043 0 : std::string msg = StringFormat("currentCollOperator is nullptr");
1044 0 : THROW<NullPtrException>(msg);
1045 0 : }
1046 :
1047 18 : if (isLaunch) {
1048 18 : LaunchConvertCollOperatorA2A(opParams, isHcomSelectAlg);
1049 : } else {
1050 0 : DefaultConvertCollOperatorA2A(opParams);
1051 : }
1052 18 : }
1053 :
1054 0 : void CommunicatorImpl::DefaultConvertCollOperatorA2A(const CollOpParams &opParams)
1055 : {
1056 : // MC2场景、图模式算法选择场景准备资源场景下只需默认值
1057 0 : HCCL_INFO("DefaultConvertCollOperatorA2A start.");
1058 0 : if (opParams.opType == OpType::ALLTOALL) {
1059 0 : currentCollOperator->all2AllDataDes.sendCount = 0;
1060 0 : currentCollOperator->all2AllDataDes.recvCount = 0;
1061 0 : currentCollOperator->all2AllDataDes.sendType = DataType::FP16;
1062 0 : currentCollOperator->all2AllDataDes.recvType = DataType::FP16;
1063 0 : currentCollOperator->dataType = DataType::FP16;
1064 0 : } else if (opParams.opType == OpType::ALLTOALLV) {
1065 0 : currentCollOperator->all2AllVDataDes.sendType = DataType::FP16;
1066 0 : currentCollOperator->all2AllVDataDes.recvType = DataType::FP16;
1067 0 : currentCollOperator->dataType = DataType::FP16;
1068 0 : } else if (opParams.opType == OpType::ALLTOALLVC) {
1069 0 : currentCollOperator->all2AllVCDataDes.sendType = DataType::FP16;
1070 0 : currentCollOperator->all2AllVCDataDes.recvType = DataType::FP16;
1071 0 : currentCollOperator->dataType = DataType::FP16;
1072 : }
1073 0 : }
1074 :
1075 18 : void CommunicatorImpl::LaunchConvertCollOperatorA2A(const CollOpParams &opParams, bool isHcomSelectAlg)
1076 : {
1077 : // 下发算子场景下需要继承值并准备Mem
1078 54 : HCCL_INFO("LaunchConvertCollOperatorA2A start.");
1079 18 : if (opParams.opType == OpType::ALLTOALL) {
1080 8 : currentCollOperator->all2AllDataDes.sendCount = opParams.all2AllDataDes.sendCount;
1081 8 : currentCollOperator->all2AllDataDes.recvCount = opParams.all2AllDataDes.recvCount;
1082 8 : currentCollOperator->all2AllDataDes.sendType = opParams.all2AllDataDes.sendType;
1083 8 : currentCollOperator->all2AllDataDes.recvType = opParams.all2AllDataDes.recvType;
1084 8 : currentCollOperator->dataType = opParams.all2AllDataDes.sendType;
1085 24 : HCCL_INFO("sendCount[%llu], recvCount[%llu]", opParams.all2AllDataDes.sendCount, opParams.all2AllDataDes.recvCount);
1086 10 : } else if (opParams.opType == OpType::ALLTOALLV) {
1087 7 : currentCollOperator->all2AllVDataDes.sendCounts = opParams.all2AllVDataDes.sendCounts;
1088 7 : currentCollOperator->all2AllVDataDes.recvCounts = opParams.all2AllVDataDes.recvCounts;
1089 7 : currentCollOperator->all2AllVDataDes.sdispls = opParams.all2AllVDataDes.sdispls;
1090 7 : currentCollOperator->all2AllVDataDes.rdispls = opParams.all2AllVDataDes.rdispls;
1091 7 : currentCollOperator->all2AllVDataDes.sendType = opParams.all2AllVDataDes.sendType;
1092 7 : currentCollOperator->all2AllVDataDes.recvType = opParams.all2AllVDataDes.recvType;
1093 7 : currentCollOperator->dataType = opParams.all2AllVDataDes.sendType;
1094 3 : } else if (opParams.opType == OpType::ALLTOALLVC) {
1095 3 : currentCollOperator->all2AllVCDataDes.sendType = opParams.all2AllVCDataDes.sendType;
1096 3 : currentCollOperator->all2AllVCDataDes.recvType = opParams.all2AllVCDataDes.recvType;
1097 3 : currentCollOperator->all2AllVCDataDes.sendCountMatrix = opParams.all2AllVCDataDes.sendCountMatrix;
1098 3 : currentCollOperator->dataType = opParams.all2AllVCDataDes.sendType;
1099 : }
1100 :
1101 18 : u64 sendSize = 0;
1102 18 : u64 recvSize = 0;
1103 18 : CalcA2ASendRecvMem(opParams, sendSize, recvSize, isHcomSelectAlg);
1104 54 : HCCL_INFO("sendSize[%llu], recvSize[%llu]", sendSize, recvSize);
1105 18 : currentCollOperator->inputMem = DevBuffer::Create(reinterpret_cast<uintptr_t >(opParams.sendBuf), sendSize);
1106 18 : currentCollOperator->outputMem = DevBuffer::Create(reinterpret_cast<uintptr_t >(opParams.recvBuf), recvSize);
1107 18 : }
1108 :
1109 50 : void CommunicatorImpl::ConvertCollOperatorMem(const CollOpParams &opParams, u64 size)
1110 : {
1111 150 : HCCL_INFO("[CommunicatorImpl][%s] start, opType[%s], size[%llu]", __func__, opParams.opType.Describe().c_str(), size);
1112 :
1113 50 : if (opParams.opType == OpType::REDUCESCATTER || opParams.opType == OpType::SCATTER) {
1114 9 : currentCollOperator->inputMem = DevBuffer::Create(reinterpret_cast<uintptr_t>(opParams.sendBuf), size * rankSize);
1115 : } else {
1116 41 : currentCollOperator->inputMem = DevBuffer::Create(reinterpret_cast<uintptr_t>(opParams.sendBuf), size);
1117 : }
1118 :
1119 50 : if (opParams.opType == OpType::ALLGATHER || opParams.opType == OpType::GATHER) {
1120 7 : currentCollOperator->outputMem = DevBuffer::Create(reinterpret_cast<uintptr_t>(opParams.recvBuf), size * rankSize);
1121 : } else {
1122 43 : currentCollOperator->outputMem = DevBuffer::Create(reinterpret_cast<uintptr_t>(opParams.recvBuf), size);
1123 : }
1124 :
1125 150 : HCCL_INFO("[CommunicatorImpl][%s] end.", __func__);
1126 50 : }
1127 :
1128 6 : void CommunicatorImpl::ConvertCollOperatorMemV(const CollOpParams &opParams, bool isHcomSelectAlg)
1129 : {
1130 6 : if (isHcomSelectAlg) {
1131 4 : return; // isHcomSeletAlg表示是否为图插件接口进来，若是跳过该步。未来aiv支持reducescatterv/allgatherv算子时，改处需做对应适配。
1132 : }
1133 6 : HCCL_INFO("[CommunicatorImpl::%s] start, opType[%s]", __func__, opParams.opType.Describe().c_str());
1134 2 : u64 size = DataTypeSizeGet(opParams.dataType) * opParams.count;
1135 :
1136 2 : u64 *counts = static_cast<u64 *>(opParams.vDataDes.counts);
1137 2 : u64 totalCount = 0;
1138 6 : for (size_t index = 0; index < rankSize; index++) {
1139 4 : totalCount += counts[index];
1140 : }
1141 2 : u64 totalSize = DataTypeSizeGet(opParams.dataType) * totalCount;
1142 :
1143 2 : if (opParams.opType == OpType::REDUCESCATTERV) {
1144 1 : currentCollOperator->inputMem = DevBuffer::Create(reinterpret_cast<uintptr_t>(opParams.sendBuf), totalSize);
1145 : } else {
1146 1 : currentCollOperator->inputMem = DevBuffer::Create(reinterpret_cast<uintptr_t>(opParams.sendBuf), size);
1147 : }
1148 :
1149 2 : if (opParams.opType == OpType::ALLGATHERV) {
1150 1 : currentCollOperator->outputMem = DevBuffer::Create(reinterpret_cast<uintptr_t>(opParams.recvBuf), totalSize);
1151 : } else {
1152 1 : currentCollOperator->outputMem = DevBuffer::Create(reinterpret_cast<uintptr_t>(opParams.recvBuf), size);
1153 : }
1154 :
1155 6 : HCCL_INFO("[CommunicatorImpl::%s] end.", __func__);
1156 : }
1157 :
1158 81 : void CommunicatorImpl::CovertToCurrentCollOperator(std::string &opTag, const CollOpParams &opParams, OpMode opMode, bool isLaunch, bool isHcomSelectAlg)
1159 : {
1160 81 : std::string errorMsg = "CovertToCurrentCollOperator make_unique<CollOperator> failed";
1161 81 : TRY_CATCH_THROW(InternalException, errorMsg, currentCollOperator = make_unique<CollOperator>(););
1162 81 : CHECK_NULLPTR(currentCollOperator, StringFormat("[CommunicatorImpl][%s] currentCollOperator is nullptr", __func__));
1163 81 : currentCollOperator->opMode = opMode;
1164 81 : currentCollOperator->opTag = opTag; // 单算子 标签 为通信域id, 图模式 标签 为传入的opTag
1165 81 : currentCollOperator->staticAddr = opParams.staticAddr;
1166 81 : currentCollOperator->staticShape = opParams.staticShape;
1167 81 : currentCollOperator->myRank = GetMyRank();
1168 81 : if (opMode == OpMode::OPBASE) { // 单算子Scratch buffer为CCL Buffer
1169 14 : currentCollOperator->scratchMem = DevBuffer::Create(GetCclBuffer()->GetAddr(), GetCclBuffer()->GetSize());
1170 67 : } else if (opMode == OpMode::OFFLOAD) {
1171 67 : if (offloadScrachBufferMap.find(opTag) != offloadScrachBufferMap.end()) {
1172 0 : auto scratchMem = offloadScrachBufferMap[opTag];
1173 0 : HCCL_INFO("[CommunicatorImpl::CovertToCurrentCollOperator] offloadScrachBufferMap[%s] is [%s]",
1174 : opTag.c_str(), scratchMem->Describe().c_str());
1175 0 : currentCollOperator->scratchMem = scratchMem;
1176 0 : }
1177 : }
1178 :
1179 81 : currentCollOperator->opType = opParams.opType;
1180 81 : currentCollOperator->reduceOp = opParams.reduceOp;
1181 81 : currentCollOperator->root = opParams.root;
1182 81 : currentCollOperator->outputDataType = opParams.outputDataType;
1183 81 : currentCollOperator->sendRecvRemoteRank = opParams.dstRank;
1184 81 : if (opParams.opType == OpType::ALLTOALL || opParams.opType == OpType::ALLTOALLV || opParams.opType == OpType::ALLTOALLVC) {
1185 16 : ConvertCollOperatorA2A(opParams, isLaunch, isHcomSelectAlg);
1186 65 : } else if (opParams.opType == OpType::BATCHSENDRECV) {
1187 0 : currentCollOperator->batchSendRecvDataDes.sendRecvItemsPtr = opParams.batchSendRecvDataDes.sendRecvItemsPtr;
1188 0 : currentCollOperator->batchSendRecvDataDes.itemNum = opParams.batchSendRecvDataDes.itemNum;
1189 0 : currentCollOperator->dataType = HcclDataTypeToDataType(static_cast<HcclSendRecvItem*>(opParams.batchSendRecvDataDes.sendRecvItemsPtr)->dataType);
1190 : } else {
1191 65 : currentCollOperator->dataType = opParams.dataType;
1192 65 : currentCollOperator->dataCount = opParams.count;
1193 65 : if(opParams.opType == OpType::REDUCESCATTERV || opParams.opType == OpType::ALLGATHERV){
1194 6 : currentCollOperator->vDataDes.counts = opParams.vDataDes.counts;
1195 6 : currentCollOperator->vDataDes.displs = opParams.vDataDes.displs;
1196 6 : currentCollOperator->vDataDes.dataType = opParams.vDataDes.dataType;
1197 6 : ConvertCollOperatorMemV(opParams, isHcomSelectAlg);
1198 : } else {
1199 59 : u64 size = DataTypeSizeGet(opParams.dataType) * opParams.count;
1200 59 : if (size != 0) {
1201 50 : ConvertCollOperatorMem(opParams, size);
1202 : } else {
1203 27 : HCCL_WARNING("[CommunicatorImpl::%s] size is 0", __func__);
1204 : }
1205 : }
1206 : }
1207 243 : HCCL_INFO("CommunicatorImpl::%s op dataType[%s], dataCount[%llu]", __func__, currentCollOperator->dataType.Describe().c_str(), currentCollOperator->dataCount);
1208 81 : }
1209 :
1210 47 : void CommunicatorImpl::InitCommonData(const CommParams &commParams, const HcclCommConfig &commConfig)
1211 : {
1212 47 : InitCommonDataNotInitDevType(commParams, commConfig);
1213 : // 设定devType，初始化能力，算法及其他模块通过Get获取能力
1214 47 : DevCapability::GetInstance().Init(devType);
1215 47 : }
1216 :
1217 49 : void CommunicatorImpl::InitCommonDataNotInitDevType(const CommParams &commParams, const HcclCommConfig &commConfig)
1218 : {
1219 49 : InitCommonData(commParams);
1220 49 : config = commConfig;
1221 49 : cclBufferSize = config.hcclBufferSize;
1222 49 : }
1223 :
1224 50 : void CommunicatorImpl::InitCommonData(const CommParams &commParams)
1225 : {
1226 50 : id = commParams.commId;
1227 50 : idIndex = globalIndex.fetch_add(1);
1228 50 : establishLinkSocketTag = id + "_establish_link" + "_" + "exchanger";
1229 50 : myRank = commParams.myRank;
1230 50 : rankSize = commParams.rankSize;
1231 50 : rankInParentComm = commParams.rankInParentComm;
1232 50 : devType = commParams.devType;
1233 50 : isWorldGroup = commParams.isWorldGroup;
1234 50 : devLogicId = HrtGetDevice();
1235 50 : devPhyId = HrtGetDevicePhyIdByIndex(devLogicId);
1236 50 : }
1237 :
1238 7 : void CommunicatorImpl::CheckRankGraph() const
1239 : {
1240 : // 校验虚拟拓扑中的rankSize和通信域的rankSize一致
1241 7 : u32 virtRankSize = rankGraph->GetRankSize();
1242 7 : if (virtRankSize != rankSize) {
1243 : std::string msg
1244 : = StringFormat("Check rankGraph failed, communicator rankSize[%u] does not equal rankTable rankSize[%u]",
1245 1 : rankSize, virtRankSize);
1246 1 : THROW<InvalidParamsException>(msg);
1247 1 : }
1248 :
1249 : // 校验0值
1250 6 : u32 num = rankGraph->GetInnerRankSize();
1251 6 : if (num == 0) {
1252 : std::string msg
1253 : = StringFormat("Check rankGraph failed, inner rankSize should not be %u",
1254 1 : num);
1255 1 : THROW<InvalidParamsException>(msg);
1256 1 : }
1257 5 : CheckRankGraphAddrs();
1258 5 : }
1259 :
1260 5 : void CommunicatorImpl::CheckRankGraphAddrs() const
1261 : {
1262 5 : if (rankGraph == nullptr || ranktableInfo == nullptr) {
1263 0 : HCCL_WARNING("[CommunicatorImpl][%s] rankGraph or ranktableInfo is nullptr, skip.", __func__);
1264 5 : return;
1265 : }
1266 :
1267 5 : if (rankGraph->GetRankSize() == 1) {
1268 0 : HCCL_WARNING("[CommunicatorImpl][%s] single rank no need do this check!", __func__);
1269 0 : return;
1270 : }
1271 :
1272 : // 仅能获取到当前进程所在卡的ip，每个卡独立check自己的部分
1273 5 : std::unordered_set<Eid> localEidSet;
1274 5 : NewRankInfo localRankInfo;
1275 10 : for (auto &rank : ranktableInfo->ranks) {
1276 10 : if (rank.deviceId == devPhyId) { // 获取本卡的ip地址
1277 5 : HRaInfo info(HrtNetworkMode::HDC, rank.deviceId);
1278 5 : std::vector<HrtDevEidInfo> localEidInfos = HrtRaGetDevEidInfoList(info);
1279 5 : for (auto &eidInfo : localEidInfos) {
1280 0 : localEidSet.insert(eidInfo.ipAddress.GetEid());
1281 : }
1282 5 : localRankInfo = rank;
1283 5 : break;
1284 5 : }
1285 : }
1286 :
1287 5 : if (localEidSet.empty()) {
1288 5 : return;
1289 : }
1290 :
1291 0 : const std::shared_ptr<NetInstance::Peer> &peer = rankGraph->GetPeer(myRank);
1292 0 : const std::vector<std::shared_ptr<NetInstance::ConnInterface>> &interfaces = peer->GetIfaces();
1293 0 : for(auto &interface : interfaces) {
1294 0 : const std::set<LinkProtocol> &protocols = interface->GetLinkProtocols(); // PCIE和UBOE没有EID
1295 0 : if (interface->GetPos() == AddrPosition::DEVICE && protocols.count(LinkProtocol::PCIE) == 0
1296 0 : && protocols.count(LinkProtocol::UBOE) == 0 && localEidSet.count(interface->GetAddr().GetEid()) == 0) {
1297 0 : RPT_INPUT_ERR(true, "EI0014", std::vector<std::string>({"value", "variable", "expect"}),
1298 : std::vector<std::string>({interface->GetAddr().GetIpStr(), "addr", "A right ip address"}));
1299 0 : THROW<InvalidParamsException>(StringFormat("[CommunicatorImpl][%s]"
1300 : "the ip address %s of ranktable in rank %u is error!",
1301 0 : __func__, interface->GetAddr().Describe().c_str(), devPhyId));
1302 : }
1303 0 : }
1304 10 : }
1305 :
1306 :
1307 40 : u32 GetLocalDieId(PortData&& port, LinkProtocol linkProtocol)
1308 : {
1309 40 : auto devLogicId = HrtGetDevice();
1310 40 : uint32_t devPhyId = HrtGetDevicePhyIdByIndex(devLogicId);
1311 :
1312 40 : auto &rdmaHandleMgr = RdmaHandleManager::GetInstance();
1313 40 : auto rdmaHandle = rdmaHandleMgr.Get(devPhyId, port, linkProtocol);
1314 40 : auto dieId = rdmaHandleMgr.GetDieAndFuncId(rdmaHandle).first;
1315 40 : return dieId;
1316 : }
1317 :
1318 : constexpr u32 localPortId = 0;
1319 :
1320 12 : void CommunicatorImpl::InitRankGraph(const string &ranktableM)
1321 : {
1322 12 : string topoPath = GetTopoFilePath();
1323 : // legacy 通信域保留调用入口，实际构建通过 bridge 转交给 hcomm 中的 RankGraphBuilder。
1324 5 : const RankGraphBuilderBridge *bridge = GetRankGraphBuilderBridge();
1325 5 : CHK_PRT_THROW(bridge == nullptr || bridge->buildFromString == nullptr,
1326 : HCCL_ERROR("[%s] RankGraphBuilder bridge is not registered.", __func__),
1327 : InternalException, "RankGraphBuilder bridge is not registered.");
1328 :
1329 5 : RankGraphBuildResult buildResult{};
1330 5 : HcclResult ret = bridge->buildFromString(ranktableM, topoPath, myRank, buildResult);
1331 5 : CHK_PRT_THROW(ret != HCCL_SUCCESS,
1332 : HCCL_ERROR("[%s] build rank graph failed, errNo[0x%016llx].", __func__, HCCL_ERROR_CODE(ret)),
1333 : InternalException, "Build rank graph failed.");
1334 : // 三项结果来自同一个 builder，必须一起更新，避免通信域持有不一致的拓扑快照。
1335 5 : InitRankGraph(std::move(buildResult.rankGraph), buildResult.rankTableInfo, buildResult.topoInfo);
1336 5 : }
1337 :
1338 7 : std::string CommunicatorImpl::GetTopoFilePath()
1339 : {
1340 21 : HCCL_INFO("[CommunicatorImpl::%s] start.", __func__);
1341 :
1342 7 : std::string filePath = "/etc/hccl_rootinfo.json";
1343 : JsonParser jsonParser{};
1344 7 : nlohmann::json parseJson{};
1345 7 : std::string topoFilePath{};
1346 7 : std::ifstream file(filePath);
1347 7 : if (file.good()) {
1348 0 : jsonParser.ParseFileToJson(filePath, parseJson);
1349 0 : std::string msgRankTopoFile = "error occurs when parser object of propName \"topo_file_path\"";
1350 0 : TRY_CATCH_THROW(InvalidParamsException, msgRankTopoFile, topoFilePath = GetJsonProperty(parseJson, "topo_file_path"););
1351 0 : } else {
1352 7 : const size_t bufSize = 1024;
1353 7 : auto devLogicId = HrtGetDevice();
1354 7 : auto devPhyId = HrtGetDevicePhyIdByIndex(devLogicId);
1355 7 : std::vector<char> buffer(bufSize, '\0');
1356 7 : int result = TopoAddrInfoGetTopoFilePath(devPhyId, buffer.data(), buffer.size());
1357 42 : CHK_PRT_THROW(result != 0,
1358 : HCCL_ERROR("[%s] Get topo file path failed.", __func__),
1359 : InvalidParamsException, "Get topo file path failed.");
1360 0 : topoFilePath = std::string(buffer.data());
1361 7 : }
1362 :
1363 : // check topo_file_path
1364 0 : char resolvedPath[PATH_MAX] = {0};
1365 0 : CHK_PRT_THROW(realpath(topoFilePath.c_str(), resolvedPath) == nullptr,
1366 : HCCL_ERROR("[%s] topo_file_path[%s] is not a valid real path", __func__, topoFilePath.c_str()),
1367 : InvalidParamsException, "topo_file_path error");
1368 0 : return topoFilePath;
1369 28 : }
1370 :
1371 0 : void CommunicatorImpl::InitRankGraph(const RankTableInfo &ranktable)
1372 : {
1373 0 : string topoPath = GetTopoFilePath();
1374 : // RootInfoDetect 已产出结构化 RankTableInfo 时，避免再次解析字符串，直接调用对应 provider 回调。
1375 0 : const RankGraphBuilderBridge *bridge = GetRankGraphBuilderBridge();
1376 0 : CHK_PRT_THROW(bridge == nullptr || bridge->buildFromRankTable == nullptr,
1377 : HCCL_ERROR("[%s] RankGraphBuilder bridge is not registered.", __func__),
1378 : InternalException, "RankGraphBuilder bridge is not registered.");
1379 :
1380 0 : RankGraphBuildResult buildResult{};
1381 0 : HcclResult ret = bridge->buildFromRankTable(ranktable, topoPath, myRank, buildResult);
1382 0 : CHK_PRT_THROW(ret != HCCL_SUCCESS,
1383 : HCCL_ERROR("[%s] build rank graph failed, errNo[0x%016llx].", __func__, HCCL_ERROR_CODE(ret)),
1384 : InternalException, "Build rank graph failed.");
1385 : // 同步安装本次构建得到的 RankGraph、RankTableInfo 和 TopoInfo。
1386 0 : InitRankGraph(std::move(buildResult.rankGraph), buildResult.rankTableInfo, buildResult.topoInfo);
1387 0 : }
1388 :
1389 2 : void CommunicatorImpl::InitRankGraph(std::unique_ptr<RankGraph> &inputRankGraph)
1390 : {
1391 2 : if (inputRankGraph != nullptr) {
1392 : // 子通信域等旧路径仍传入 unique_ptr，需要补上 hcomm 侧 deleter 后再由通信域共享持有。
1393 2 : rankGraph = TakeRankGraphOwnership(inputRankGraph);
1394 : } else {
1395 0 : std::string msg = StringFormat("Init RankGraph failed, inputRankGraph is nullptr");
1396 0 : THROW<NullPtrException>(msg);
1397 0 : }
1398 2 : CheckRankGraph();
1399 0 : SaveTopoDesc(id);
1400 0 : }
1401 :
1402 5 : void CommunicatorImpl::InitRankGraph(std::shared_ptr<RankGraph> inputRankGraph,
1403 : const RankTableInfo &inputRankTableInfo, const TopoInfo &inputTopoInfo)
1404 : {
1405 5 : if (inputRankGraph == nullptr) {
1406 0 : THROW<NullPtrException>("Init RankGraph failed, inputRankGraph is nullptr");
1407 : }
1408 :
1409 5 : std::unique_ptr<RankTableInfo> preparedRankTableInfo;
1410 5 : std::shared_ptr<TopoInfo> preparedTopoInfo;
1411 5 : PrepareRankGraphMetadata(inputRankTableInfo, inputTopoInfo, preparedRankTableInfo, preparedTopoInfo);
1412 :
1413 5 : const u32 preparedRankSize = inputRankGraph->GetRankSize();
1414 : // 以下移动赋值均不分配内存，确保三个拓扑对象只在准备完整后一起发布。
1415 5 : rankGraph = std::move(inputRankGraph);
1416 5 : ranktableInfo = std::move(preparedRankTableInfo);
1417 5 : topoInfo = std::move(preparedTopoInfo);
1418 5 : rankSize = preparedRankSize;
1419 15 : HCCL_RUN_INFO("[CommunicatorImpl::%s] rankTableInfo: %s", __func__, ranktableInfo->Describe().c_str());
1420 15 : HCCL_RUN_INFO("[CommunicatorImpl][%s] topoInfo[%s]", __func__, topoInfo->Describe().c_str());
1421 5 : CheckRankGraph();
1422 5 : SaveTopoDesc(id);
1423 5 : std::vector<LinkData> fullLinks = GetFullMeshLinks();
1424 20 : for (auto link : fullLinks) {
1425 45 : HCCL_RUN_INFO("[CommunicatorImpl][%s] link[%s]", __func__, link.Describe().c_str());
1426 : }
1427 5 : }
1428 :
1429 40 : void CommunicatorImpl::InitDataBufferManager()
1430 : {
1431 : // 申请scratchMem
1432 40 : u64 scratchBufSize = static_cast<u64>(GetBufferSize());
1433 40 : if (scratchBufSize == 0) {
1434 17 : scratchBufSize = EnvConfig::GetInstance().GetAlgoConfig().GetBuffSize();
1435 : } else {
1436 23 : scratchBufSize = scratchBufSize * HCCL_CCL_COMM_FIXED_CALC_BUFFER_SIZE;
1437 : }
1438 : // 如果是自定义算子流程，cclBufferSize的大小为2倍
1439 40 : scratchBufSize = scratchBufSize * INDEPENDENT_OP_BUFFER_SIZE_TIMES;
1440 40 : cclBufferSize = scratchBufSize;
1441 :
1442 : // aiv mc2预埋1M，并不暴露在内部算子执行逻辑里
1443 40 : scratchBufSize += HCCL_MC2_ON_AICPU_FIXED_CALC_BUFFER_SIZE;
1444 :
1445 40 : if (rankSize > 1) {
1446 39 : aivOffloadTagBuffer = std::move(DevBuffer::CreateHugePageBuf(HCCL_AIV_OFFLOAD_TAG_BUFFER_SIZE));
1447 39 : HrtMemset(reinterpret_cast<void*>(aivOffloadTagBuffer->GetAddr()), aivOffloadTagBuffer->GetSize(), aivOffloadTagBuffer->GetSize());
1448 39 : cclBuffer = std::move(DevBuffer::CreateHugePageBuf(scratchBufSize));
1449 117 : HCCL_RUN_INFO(
1450 : "[CommunicatorImpl][InitDataBufferManager] cclBuffer create, commId[%s], addr[%llu], size[%llu]M",
1451 : GetId().c_str(), cclBuffer->GetAddr(), cclBufferSize / HCCL_CCL_COMM_FIXED_CALC_BUFFER_SIZE);
1452 :
1453 39 : u64 aivTagBufSize = HCCL_CCL_AIV_TAG_BUFFER_SIZE * HCCL_CCL_COMM_FIXED_CALC_BUFFER_SIZE;
1454 117 : HCCL_INFO("[CommunicatorImpl][InitDataBufferManager] aivTagBufSize[%llu]M", aivTagBufSize / HCCL_CCL_COMM_FIXED_CALC_BUFFER_SIZE);
1455 39 : aivTagBuffer = std::move(DevBuffer::CreateHugePageBuf(aivTagBufSize));
1456 39 : HrtMemset(reinterpret_cast<void*>(aivTagBuffer->GetAddr()), aivTagBuffer->GetSize(), aivTagBuffer->GetSize());
1457 : }
1458 40 : dataBufferManager = std::make_unique<DataBufManager>();
1459 :
1460 40 : localRmaBufManager = std::make_unique<LocalRmaBufManager>(*this);
1461 :
1462 40 : remoteRmaBufManager = std::make_unique<RemoteRmaBufManager>(*this);
1463 40 : }
1464 :
1465 242 : void CommunicatorImpl::InitNotifyManager()
1466 : {
1467 242 : aicpuQueueNotifyManager_ = std::make_unique<QueueNotifyManager>(*this);
1468 :
1469 242 : ccuQueueNotifyManager_ = std::make_unique<QueueNotifyManager>(*this);
1470 :
1471 242 : queueWaitGroupCntNotifyManager = std::make_unique<QueueWaitGroupCntNotifyManager>();
1472 :
1473 242 : queueBcastPostCntNotifyManager = std::make_unique<QueueBcastPostCntNotifyManager>();
1474 :
1475 242 : connLocalNotifyManager = std::make_unique<ConnLocalNotifyManager>(this);
1476 :
1477 242 : connLocalCntNotifyManager = std::make_unique<ConnLocalCntNotifyManager>(this);
1478 :
1479 242 : ccuStreamSyncNotifyManager = std::make_unique<CcuStreamSyncNotifyManager>();
1480 242 : }
1481 :
1482 248 : void CommunicatorImpl::InitStreamManager()
1483 : {
1484 248 : streamManager = std::make_unique<StreamManager>(this);
1485 248 : aicpuStreamManager = std::make_unique<AicpuStreamManager>();
1486 248 : }
1487 :
1488 193 : void CommunicatorImpl::InitCollService()
1489 : {
1490 579 : HCCL_INFO("CommunicatorImpl::InitCollServices start");
1491 :
1492 193 : auto ccuCollService = std::make_shared<CollServiceDeviceMode>(this); // host 展开，ccu使用
1493 193 : auto aiCpuCollService = std::make_shared<CollServiceAiCpuImpl>(this); // aicpu 展开
1494 193 : auto hostCollService = std::make_shared<CollServiceDefaultImpl>(this); // host 展开，图模式使用
1495 193 : ccuCollService->Init();
1496 193 : aiCpuCollService->Init();
1497 193 : hostCollService->Init();
1498 :
1499 193 : collServices[AcceleratorState::AIV] = ccuCollService; // host 展开，aiv使用
1500 193 : collServices[AcceleratorState::AIV_ONLY] = ccuCollService; // host 展开，aiv使用
1501 193 : collServices[AcceleratorState::CCU_MS] = ccuCollService; // host 展开，ccu使用
1502 193 : collServices[AcceleratorState::CCU_SCHED] = ccuCollService; // host 展开，ccu使用
1503 193 : collServices[AcceleratorState::AICPU_TS] = aiCpuCollService; // aicpu 展开
1504 193 : collServices[AcceleratorState::HOSTCPU_TS] = hostCollService; // host 展开，图模式使用
1505 :
1506 579 : HCCL_INFO("CommunicatorImpl::InitCollServices end");
1507 386 : return;
1508 193 : }
1509 :
1510 41 : HcclResult CommunicatorImpl::InitTraceManager()
1511 : {
1512 : /* 申请trace资源信息 */
1513 41 : std::string logInfo = "HCCL_";
1514 41 : logInfo.append(std::to_string(SalGetTid()));
1515 41 : logInfo.append("_");
1516 41 : logInfo.append(std::to_string(GetDeviceLogicId()));
1517 41 : logInfo.append("_");
1518 41 : logInfo.append(std::to_string(idIndex));
1519 41 : trace = std::make_unique<Trace>();
1520 41 : CHK_PTR_NULL(trace);
1521 41 : CHK_RET(trace->Init(logInfo));
1522 41 : return HCCL_SUCCESS;
1523 41 : }
1524 :
1525 47 : void CommunicatorImpl::InitHDCommunicate()
1526 : {
1527 : // 不管是aicpu还是ccu都初始化
1528 141 : HCCL_INFO("Enter [CommunicatorImpl::InitHDCommunicate]");
1529 47 : kfcControlTransferH2D = std::make_unique<HDCommunicate>(devLogicId, HCCLV2_HDC_TYPE_H2D, sizeof(KfcCommand));
1530 47 : kfcControlTransferH2D->Init();
1531 47 : kfcStatusTransferD2H = std::make_unique<HDCommunicate>(devLogicId, HCCLV2_HDC_TYPE_D2H, sizeof(KfcExecStatus));
1532 47 : kfcStatusTransferD2H->Init();
1533 47 : }
1534 :
1535 50 : void CommunicatorImpl::InitHccpHdc() const
1536 : {
1537 50 : HccpHdcManager::GetInstance().Init(devLogicId);
1538 50 : }
1539 :
1540 1 : void CommunicatorImpl::TryInitCcuFeature() const
1541 : {
1542 1 : TpManager::GetInstance(devLogicId).Init();
1543 3 : HCCL_INFO("[CommunicatorImpl][%s] passed, "
1544 : "will use open source ccu feature.", __func__);
1545 1 : return;
1546 : }
1547 :
1548 47 : void CommunicatorImpl::InitCcuSuperFastLoad()
1549 : {
1550 : //ccu 模式 快速下发模式需要的变量初始化
1551 47 : taskExceptionEnv = EnvConfig::GetInstance().GetLogConfig().GetDfsConfig().taskExceptionEnable;
1552 :
1553 47 : bool hostApiState = ProfilingHandler::GetInstance().GetHostApiState();
1554 47 : bool nodeState = ProfilingHandler::GetInstance().GetHcclNodeState();
1555 47 : bool l0State = ProfilingHandler::GetInstance().GetHcclL0State();
1556 47 : bool l1State = ProfilingHandler::GetInstance().GetHcclL1State();
1557 :
1558 47 : enableProfilingEnv = hostApiState || nodeState || l0State || l1State;
1559 :
1560 141 : HCCL_INFO("taskExceptionEnv[%d], enableProfilingEnv: hostApiState[%d] nodeState[%d] l0State[%d] l1State[%d]",
1561 : taskExceptionEnv, hostApiState, nodeState, l0State, l1State);
1562 47 : }
1563 :
1564 38 : void CommunicatorImpl::InitPreResource()
1565 : {
1566 : // PCIE链路的两端实现enableP2P
1567 38 : auto links = GetFullMeshLinks();
1568 53 : for (auto link : links) {
1569 15 : if (link.GetLinkProtocol() == LinkProtocol::PCIE) {
1570 0 : DeviceId remotePhyId = link.GetRemoteDeviceId();
1571 0 : enableP2PDevices_.push_back(remotePhyId);
1572 : }
1573 : }
1574 38 : CHK_RET_THROW(RuntimeApiException, "EnableP2P Failed", P2PEnableManager::GetInstance().EnableP2P(enableP2PDevices_));
1575 38 : }
1576 :
1577 839 : void CommunicatorImpl::DeInitPreResource()
1578 : {
1579 839 : (void)P2PEnableManager::GetInstance().DisableP2P(devLogicId, enableP2PDevices_);
1580 839 : }
1581 :
1582 242 : void CommunicatorImpl::InitSocketManager()
1583 : {
1584 242 : socketManager = std::make_unique<SocketManager>(*this, myRank, devPhyId, devLogicId);
1585 242 : if (ranktableInfo != nullptr) {
1586 5 : auto rankIpPortMap = ranktableInfo->GetRankDeviceListenPortMap();
1587 5 : rankIpPortMap_ = std::make_shared<decltype(rankIpPortMap)>(std::move(rankIpPortMap));
1588 5 : socketManager->SetDeviceServerListenPortMap(*rankIpPortMap_);
1589 5 : }
1590 242 : }
1591 :
1592 245 : void CommunicatorImpl::InitRmaConnManager()
1593 : {
1594 245 : rmaConnectionManager = std::make_unique<RmaConnManager>(*this);
1595 245 : }
1596 :
1597 252 : void CommunicatorImpl::InitMemTransportManager()
1598 : {
1599 252 : memTransportManager = std::make_unique<MemTransportManager>(*this);
1600 252 : }
1601 :
1602 37 : void CommunicatorImpl::InitHostDeviceSyncNotifyManager()
1603 : {
1604 37 : hostDeviceSyncNotifyManager = std::make_unique<HostDeviceSyncNotifyManager>();
1605 37 : }
1606 :
1607 550 : const string &CommunicatorImpl::GetId() const
1608 : {
1609 550 : return id;
1610 : }
1611 :
1612 20 : u32 CommunicatorImpl::GetIdIndex() const
1613 : {
1614 20 : return idIndex;
1615 : }
1616 :
1617 298 : const string &CommunicatorImpl::GetEstablishLinkSocketTag() const
1618 : {
1619 298 : return establishLinkSocketTag;
1620 : }
1621 :
1622 846 : RankId CommunicatorImpl::GetMyRank() const
1623 : {
1624 846 : return myRank;
1625 : }
1626 :
1627 617 : u32 CommunicatorImpl::GetRankSize() const
1628 : {
1629 617 : return rankSize;
1630 : }
1631 :
1632 591 : u32 CommunicatorImpl::GetDeviceLogicId() const
1633 : {
1634 591 : return devLogicId;
1635 : }
1636 :
1637 34 : u32 CommunicatorImpl::GetDevicePhyId() const
1638 : {
1639 34 : return devPhyId;
1640 : }
1641 :
1642 67 : u64 CommunicatorImpl::GetBufferSize() const
1643 : {
1644 67 : return cclBufferSize;
1645 : }
1646 :
1647 3 : u32 CommunicatorImpl::GetSubmittedOpCnt() const
1648 : {
1649 3 : return submittedOpCnt;
1650 : }
1651 :
1652 0 : u32 CommunicatorImpl::GetOpBaseOpIndex() const
1653 : {
1654 0 : return opBaseOpIndex;
1655 : }
1656 :
1657 12 : u32 CommunicatorImpl::GetOpIndex() const
1658 : {
1659 12 : return opIndex;
1660 : }
1661 :
1662 147 : bool CommunicatorImpl::GetOpAiCpuTSFeatureFlag() const
1663 : {
1664 147 : return opExecuteConfig.accState == AcceleratorState::AICPU_TS;
1665 : }
1666 :
1667 0 : bool CommunicatorImpl::GetCommAiCpuTSFeatureFlag() const
1668 : {
1669 0 : return commExecuteConfig.accState == AcceleratorState::AICPU_TS;
1670 : }
1671 :
1672 235 : const DevType &CommunicatorImpl::GetDevType() const
1673 : {
1674 705 : HCCL_INFO("CommunicatorImpl::DevType is %s", devType.Describe().c_str());
1675 235 : return devType;
1676 : }
1677 :
1678 2469 : shared_ptr<RankGraph> CommunicatorImpl::GetRankGraph() const
1679 : {
1680 7407 : HCCL_INFO("CommunicatorImpl::GetRankGraph ");
1681 2469 : return rankGraph;
1682 : }
1683 :
1684 35 : DataBufManager &CommunicatorImpl::GetDataBufferManager() const
1685 : {
1686 35 : CHECK_NULLPTR(dataBufferManager, "dataBufferManager is nullptr!");
1687 35 : return *dataBufferManager;
1688 : }
1689 :
1690 185 : LocalRmaBufManager &CommunicatorImpl::GetLocalRmaBufManager() const
1691 : {
1692 185 : CHECK_NULLPTR(localRmaBufManager, "localRmaBufManager is nullptr!");
1693 185 : return *localRmaBufManager;
1694 : }
1695 :
1696 0 : RemoteRmaBufManager &CommunicatorImpl::GetRemoteRmaBufManager() const
1697 : {
1698 0 : CHECK_NULLPTR(remoteRmaBufManager, "remoteRmaBufManager is nullptr!");
1699 0 : return *remoteRmaBufManager;
1700 : }
1701 :
1702 5 : QueueNotifyManager &CommunicatorImpl::GetAicpuQueueNotifyManager() const
1703 : {
1704 5 : CHECK_NULLPTR(aicpuQueueNotifyManager_, "aicpuQueueNotifyManager is nullptr!");
1705 5 : return *aicpuQueueNotifyManager_;
1706 : }
1707 :
1708 4 : QueueNotifyManager &CommunicatorImpl::GetCcuQueueNotifyManager() const
1709 : {
1710 4 : CHECK_NULLPTR(ccuQueueNotifyManager_, "ccuQueueNotifyManager is nullptr!");
1711 4 : return *ccuQueueNotifyManager_;
1712 : }
1713 :
1714 1 : ConnLocalNotifyManager &CommunicatorImpl::GetConnLocalNotifyManager() const
1715 : {
1716 1 : CHECK_NULLPTR(connLocalNotifyManager, "connLocalNotifyManager is nullptr!");
1717 1 : return *connLocalNotifyManager;
1718 : }
1719 :
1720 1 : ConnLocalCntNotifyManager &CommunicatorImpl::GetConnLocalCntNotifyManager() const
1721 : {
1722 1 : CHECK_NULLPTR(connLocalCntNotifyManager, "connLocalCntNotifyManager is nullptr!");
1723 1 : return *connLocalCntNotifyManager;
1724 : }
1725 :
1726 13 : QueueWaitGroupCntNotifyManager &CommunicatorImpl::GetQueueWaitGroupCntNotifyManager() const
1727 : {
1728 13 : CHECK_NULLPTR(queueWaitGroupCntNotifyManager, "queueWaitGroupCntNotifyManager is nullptr!");
1729 13 : return *queueWaitGroupCntNotifyManager;
1730 : }
1731 :
1732 13 : QueueBcastPostCntNotifyManager &CommunicatorImpl::GetBcastPostCntNotifyManager() const
1733 : {
1734 13 : CHECK_NULLPTR(queueBcastPostCntNotifyManager, "queueBcastPostCntNotifyManager is nullptr!");
1735 13 : return *queueBcastPostCntNotifyManager;
1736 : }
1737 :
1738 11 : CcuStreamSyncNotifyManager &CommunicatorImpl::GetCcuStreamSyncNotifyManager() const
1739 : {
1740 11 : CHECK_NULLPTR(ccuStreamSyncNotifyManager, "ccuStreamSyncNotifyManager is nullptr!");
1741 11 : return *ccuStreamSyncNotifyManager;
1742 : }
1743 :
1744 113 : StreamManager &CommunicatorImpl::GetStreamManager() const
1745 : {
1746 114 : CHECK_NULLPTR(streamManager, "streamManager is nullptr!");
1747 112 : return *streamManager;
1748 : }
1749 :
1750 28 : AicpuStreamManager &CommunicatorImpl::GetAicpuStreamManager() const
1751 : {
1752 28 : CHECK_NULLPTR(aicpuStreamManager, "aicpuStreamManager is nullptr!");
1753 28 : return *aicpuStreamManager;
1754 : }
1755 :
1756 50 : CollServiceBase *CommunicatorImpl::GetCollService() const
1757 : {
1758 50 : return collService;
1759 : }
1760 :
1761 7 : CollServiceBase *CommunicatorImpl::GetCcuCollService() const
1762 : {
1763 : // 仅在Task Exception下使用，异常捕获由TaskExceptionHandler::Process管理
1764 7 : if (collServices.find(AcceleratorState::CCU_SCHED) != collServices.end()) {
1765 6 : return collServices.at(AcceleratorState::CCU_SCHED).get();
1766 : }
1767 : else {
1768 1 : std::string msg{"[CommunicatorImpl] Communicator uninitialized, this should not be arrived"};
1769 4 : MACRO_THROW(NullPtrException, msg);
1770 1 : }
1771 : }
1772 :
1773 65 : SocketManager &CommunicatorImpl::GetSocketManager() const
1774 : {
1775 65 : CHECK_NULLPTR(socketManager, "socketManager is nullptr!");
1776 65 : return *socketManager;
1777 : }
1778 :
1779 219 : RmaConnManager &CommunicatorImpl::GetRmaConnManager() const
1780 : {
1781 219 : CHECK_NULLPTR(rmaConnectionManager, "rmaConnectionManager is nullptr!");
1782 219 : return *rmaConnectionManager;
1783 : }
1784 :
1785 124 : CollOperator *CommunicatorImpl::GetCurrentCollOperator() const
1786 : {
1787 124 : CHECK_NULLPTR(currentCollOperator, "currentCollOperator is nullptr!");
1788 124 : return currentCollOperator.get();
1789 : }
1790 :
1791 2023579 : MemTransportManager *CommunicatorImpl::GetMemTransportManager() const
1792 : {
1793 2023579 : return memTransportManager.get();
1794 : }
1795 :
1796 79 : bool CommunicatorImpl::GetOpCcuFeatureFlag() const
1797 : {
1798 79 : return IsOpUsingCcuMs() || IsOpUsingCcuSched(); // 算子粒度
1799 : }
1800 :
1801 1 : bool CommunicatorImpl::GetCommCcuFeatureFlag() const
1802 : {
1803 1 : return IsCommUsingCcuMs() || IsCommUsingCcuSched(); // 通信域粒度
1804 : }
1805 :
1806 0 : bool CommunicatorImpl::IsOpUsingAiv() const
1807 : {
1808 0 : return GetOpExecuteConfig().accState == AcceleratorState::AIV;
1809 : }
1810 :
1811 0 : bool CommunicatorImpl::IsOpUsingAivOnly() const
1812 : {
1813 0 : return GetOpExecuteConfig().accState == AcceleratorState::AIV_ONLY;
1814 : }
1815 :
1816 0 : bool CommunicatorImpl::GetOpAivFeatureFlag() const
1817 : {
1818 0 : return IsOpUsingAiv() || IsOpUsingAivOnly();
1819 : }
1820 :
1821 7 : HcclResult CommunicatorImpl::AllocCommResource(void *mc2Tiling, void **commContext)
1822 : {
1823 : try {
1824 7 : AcceleratorState acceleratorState;
1825 22 : CHK_RET(GetTilingAccelerator(mc2Tiling, acceleratorState));
1826 2 : OpExecuteConfig mc2AcceConfig;
1827 2 : mc2AcceConfig.accState = acceleratorState;
1828 2 : SetOpExecuteConfig(mc2AcceConfig);
1829 2 : SelectCollService();
1830 2 : isLoadOp = true;
1831 2 : WaitReady();
1832 0 : collService->AllocCommResource(mc2Tiling, commContext, acceleratorState);
1833 2 : } catch (HcclException &e) {
1834 6 : HCCL_ERROR(e.what());
1835 2 : PrintBackTrace(e);
1836 2 : return e.GetErrorCode();
1837 2 : } catch (exception &e) {
1838 0 : HCCL_ERROR(e.what());
1839 0 : return HcclResult::HCCL_E_INTERNAL;
1840 0 : } catch (...) {
1841 0 : HCCL_ERROR("Unknown error occurs!");
1842 0 : return HcclResult::HCCL_E_INTERNAL;
1843 0 : }
1844 0 : return HcclResult::HCCL_SUCCESS;
1845 : }
1846 :
1847 1 : HcclResult CommunicatorImpl::GetCcuTaskInfo(void *tilingData, void *ccuTaskGroup) const
1848 : {
1849 1 : if (!GetCommCcuFeatureFlag()) { // 通信域粒度
1850 3 : HCCL_ERROR("CommunicatorImpl::GetCcuTaskInfo: ccu is not used, can't GetCcuTaskInfo.");
1851 1 : return HCCL_E_NOT_SUPPORT;
1852 : }
1853 : try {
1854 0 : WaitReady();
1855 0 : collService->GetCcuTaskInfo(tilingData, ccuTaskGroup);
1856 0 : } catch (HcclException &e) {
1857 0 : HCCL_ERROR(e.what());
1858 0 : PrintBackTrace(e);
1859 0 : return e.GetErrorCode();
1860 0 : } catch (exception &e) {
1861 0 : HCCL_ERROR(e.what());
1862 0 : return HcclResult::HCCL_E_INTERNAL;
1863 0 : } catch (...) {
1864 0 : HCCL_ERROR("Unknown error occurs!");
1865 0 : return HcclResult::HCCL_E_INTERNAL;
1866 0 : }
1867 0 : return HcclResult::HCCL_SUCCESS;
1868 : }
1869 :
1870 4 : u32 CommunicatorImpl::GetCcuMc2ServerNum()
1871 : {
1872 7 : if (collServices.find(AcceleratorState::CCU_MS) == collServices.end() ||
1873 7 : collServices.find(AcceleratorState::CCU_SCHED) == collServices.end()) {
1874 1 : THROW<InternalException>("[CommunicatorImpl][%s] not create collServices type "
1875 : "CCU_MS and CCU_SCHED", __func__);
1876 : }
1877 :
1878 3 : auto ccuMc2ServerNum = collServices[AcceleratorState::CCU_MS]->GetCcuMc2ServerNum();
1879 :
1880 3 : return ccuMc2ServerNum;
1881 : }
1882 :
1883 : /* topoDescs 当前只支持l0和l1 */
1884 2 : HcclResult CommunicatorImpl::GetTopoDesc(HcclTopoDescs *topoDescs, uint32_t topoSize) const
1885 : {
1886 2 : if (topoSize < static_cast<uint32_t>(HcclTopoLevel::HCCL_TOPO_MAX)) {
1887 3 : HCCL_ERROR("topoDescs size is not enough, please check topoSize[%u]", topoSize);
1888 1 : return HCCL_E_PARA;
1889 : }
1890 :
1891 1 : topoDescs[static_cast<uint32_t>(HcclTopoLevel::HCCL_TOPO_L0)].algSets = HCCL_ALG_MESH;
1892 1 : topoDescs[static_cast<uint32_t>(HcclTopoLevel::HCCL_TOPO_L1)].algSets = 0;
1893 :
1894 1 : topoDescs[static_cast<uint32_t>(HcclTopoLevel::HCCL_TOPO_L0)].rankSize = rankSize;
1895 1 : topoDescs[static_cast<uint32_t>(HcclTopoLevel::HCCL_TOPO_L1)].rankSize = 0;
1896 :
1897 1 : return HcclResult::HCCL_SUCCESS;
1898 : }
1899 :
1900 16 : HostDeviceSyncNotifyManager &CommunicatorImpl::GetHostDeviceSyncNotifyManager() const
1901 : {
1902 16 : return *hostDeviceSyncNotifyManager;
1903 : }
1904 :
1905 28 : Trace &CommunicatorImpl::GetTrace() const
1906 : {
1907 28 : return *trace;
1908 : }
1909 :
1910 7 : HDCommunicate &CommunicatorImpl::GetKfcControlTransferH2D() const
1911 : {
1912 7 : return *kfcControlTransferH2D;
1913 : }
1914 :
1915 7 : HDCommunicate &CommunicatorImpl::GetKfcStatusTransferD2H() const
1916 : {
1917 7 : return *kfcStatusTransferD2H;
1918 : }
1919 :
1920 : constexpr u32 WAIT_CMD_TIMEOUT = 10 * 1000; // 最大等待10秒
1921 :
1922 3 : HcclResult CommunicatorImpl::Suspend()
1923 : {
1924 15 : TRY_CATCH_RETURN(
1925 : if (isSuspended) {
1926 : HCCL_WARNING("[NsRecovery][Suspend] The current communication has been suspended, no need to suspend again.");
1927 : return HcclResult::HCCL_SUCCESS;
1928 : }
1929 : isSuspended = true;
1930 : if (!isAicpuKernelLaunched) {
1931 : HCCL_INFO("[NsRecovery][Suspend] Aicpu kernel is not launched yet. Suspend host only.");
1932 : return HcclResult::HCCL_SUCCESS;
1933 : }
1934 : KfcCommand opCmd = KfcCommand::NS_STOP_LAUNCH;
1935 : CHK_RET(kfcControlTransferH2D->Put(0, sizeof(KfcCommand), reinterpret_cast<uint8_t *>(&opCmd)));
1936 : HCCL_INFO("[NsRecovery][Suspend] send KfcCommand[%d] success, which is NS_STOP_LAUNCH.", opCmd);
1937 : KfcExecStatus opInfo;
1938 : auto timeout = std::chrono::milliseconds(WAIT_CMD_TIMEOUT);
1939 : auto startTime = std::chrono::steady_clock::now();
1940 : while (true) {
1941 : CHK_RET(kfcStatusTransferD2H->Get(0, sizeof(KfcExecStatus), reinterpret_cast<uint8_t *>(&opInfo)));
1942 : if (opInfo.kfcStatus == KfcStatus::STOP_LAUNCH_DONE) {
1943 : HCCL_INFO("[NsRecovery][Suspend] received KfcStatus[%d], which is STOP_LAUNCH_DONE", opInfo.kfcStatus);
1944 : return HcclResult::HCCL_E_SUSPENDING;
1945 : } else if (opInfo.kfcStatus == KfcStatus::ERROR){
1946 : HCCL_ERROR("[NsRecovery][Suspend] received KfcStatus[%d], which is ERROR", opInfo.kfcStatus);
1947 : return HcclResult::HCCL_E_INTERNAL;
1948 : } else {
1949 : if((std::chrono::steady_clock::now() - startTime) >= timeout){
1950 : HCCL_ERROR("[NsRecovery][Suspend] Wait suspend response status timeout[%u ms] and get the opExecStatus is [%u].", WAIT_CMD_TIMEOUT,
1951 : opInfo.kfcStatus);
1952 : return HcclResult::HCCL_E_TIMEOUT;
1953 : }
1954 : continue;
1955 : }
1956 : }
1957 : );
1958 : return HcclResult::HCCL_SUCCESS;
1959 : }
1960 :
1961 4 : HcclResult CommunicatorImpl::Clean()
1962 : {
1963 28 : TRY_CATCH_RETURN(
1964 : if (!isSuspended) {
1965 : HCCL_ERROR("[NsRecovery][Clean] The current communication is not suspended, cannot clean.");
1966 : return HcclResult::HCCL_E_NOT_SUPPORT;
1967 : }
1968 : isSuspended = true;
1969 : if (isCleaned) {
1970 : HCCL_WARNING("[NsRecovery][Clean] The current communication has been cleaned, no need to clean again.");
1971 : return HcclResult::HCCL_SUCCESS;
1972 : }
1973 : isCleaned = true;
1974 : // 清理host侧资源
1975 : if (GetOpCcuFeatureFlag()) { // 算子粒度加速模式
1976 : if (collService == nullptr) { // 当前通信域没下发过算子
1977 : HCCL_WARNING("[NsRecovery][Clean] The current communication has not loaded op, no need to clean.");
1978 : return HcclResult::HCCL_SUCCESS;
1979 : }
1980 : HCCL_INFO("[NsRecovery][Clean] start to clean host. ccu flag is true");
1981 : auto collServiceCcu = dynamic_cast<CollServiceDeviceMode *>(collService);
1982 : CHECK_NULLPTR(collServiceCcu, "collServiceBase cast to CollServiceDeviceMode failed.");
1983 :
1984 : CcuInsPreprocessor *ccuInsPreprocessor = collServiceCcu->GetCcuInsPreprocessor();
1985 : CHECK_NULLPTR(ccuInsPreprocessor, "ccuInsPreprocessor is nullptr!");
1986 :
1987 : CcuCommunicator *ccuComm = ccuInsPreprocessor->GetCcuComm();
1988 : CHECK_NULLPTR(ccuComm, "ccuComm is nullptr!");
1989 :
1990 : CcuTransportMgr *ccuTransportMgr = ccuComm->GetCcuTransportMgr();
1991 : CHECK_NULLPTR(ccuTransportMgr, "ccuTransportMgr is nullptr!");
1992 : ccuTransportMgr->Clean();
1993 : return HcclResult::HCCL_SUCCESS;
1994 : } else {
1995 : HCCL_INFO("[NsRecovery][Clean] start to clean host. ccu flag is false");
1996 : rmaConnectionManager->Clear();
1997 : memTransportManager->Clear();
1998 : }
1999 : if (!isAicpuKernelLaunched) {
2000 : HCCL_INFO("[NsRecovery][Clean] Aicpu kernel is not launched yet. Clean host only.");
2001 : return HcclResult::HCCL_SUCCESS;
2002 : }
2003 : HCCL_INFO("[NsRecovery][Clean] start to clean device, waiting for device STOP_LAUNCH_DONE");
2004 : KfcExecStatus opInfo;
2005 : CHK_RET(kfcStatusTransferD2H->Get(0, sizeof(KfcExecStatus), reinterpret_cast<uint8_t *>(&opInfo)));
2006 : if (opInfo.kfcStatus == KfcStatus::STOP_LAUNCH_DONE) {
2007 : HCCL_INFO("[NsRecovery][Clean] received KfcStatus[%d], which is STOP_LAUNCH_DONE", opInfo.kfcStatus);
2008 : // 通知背景线程清理device侧资源
2009 : KfcCommand opCmd = KfcCommand::NS_CLEAN;
2010 : CHK_RET(kfcControlTransferH2D->Put(0, sizeof(KfcCommand), reinterpret_cast<uint8_t *>(&opCmd)));
2011 : HCCL_INFO("[NsRecovery][Clean] send KfcCommand [%d] success, which is NS_CLEAN", opCmd);
2012 : // 监听背景线程状态
2013 : auto timeout = std::chrono::milliseconds(WAIT_CMD_TIMEOUT);
2014 : auto startTime = std::chrono::steady_clock::now();
2015 : while (true) {
2016 : CHK_RET(kfcStatusTransferD2H->Get(0, sizeof(KfcExecStatus), reinterpret_cast<uint8_t *>(&opInfo)));
2017 : if (opInfo.kfcStatus == KfcStatus::CLEAN_DONE) {
2018 : HCCL_INFO("[NsRecovery][Clean] received KfcStatus[%d], which is CLEAN_DONE", opInfo.kfcStatus);
2019 : return HcclResult::HCCL_E_SUSPENDING;
2020 : } else if (opInfo.kfcStatus == KfcStatus::ERROR){
2021 : HCCL_ERROR("[NsRecovery][Clean] received KfcStatus[%d], which is ERROR", opInfo.kfcStatus);
2022 : return HcclResult::HCCL_E_INTERNAL;
2023 : } else {
2024 : if ((std::chrono::steady_clock::now() - startTime) >= timeout) {
2025 : HCCL_ERROR("[NsRecovery][Clean] Wait clean response status timeout[%u ms] and get the opExecStatus is [%u].", WAIT_CMD_TIMEOUT,
2026 : opInfo.kfcStatus);
2027 : return HcclResult::HCCL_E_TIMEOUT;
2028 : }
2029 : continue;
2030 : }
2031 : }
2032 : } else {
2033 : std::string msg = StringFormat("[NsRecovery][Clean] Aicpu kernel is not stopped yet. Cannot clean.");
2034 : THROW<InternalException>(msg);
2035 : return HcclResult::HCCL_E_INTERNAL;
2036 : }
2037 : );
2038 : return HcclResult::HCCL_SUCCESS;
2039 : }
2040 :
2041 3 : HcclResult CommunicatorImpl::Resume()
2042 : {
2043 17 : TRY_CATCH_RETURN(
2044 : if (GetCommStatus() == CommStatus::COMM_ERROR) {
2045 : HCCL_ERROR("[NsRecovery][Resume] Comm has been error, can not resume now!");
2046 : return HcclResult::HCCL_E_INTERNAL;
2047 : }
2048 : if (!isSuspended) {
2049 : HCCL_WARNING("[NsRecovery][Resume] The current communication is normal, no need to resume.");
2050 : return HcclResult::HCCL_SUCCESS;
2051 : }
2052 : if (GetOpCcuFeatureFlag() || GetOpAiCpuTSFeatureFlag() || GetOpAivFeatureFlag()) { // CCU和AICPU AIV // 算子粒度加速模式
2053 : HCCL_INFO("[NsRecovery][Resume] start to Resume.");
2054 : if (collService != nullptr) {
2055 : collService->Resume();
2056 : } else { // 当前通信域没下发过算子
2057 : HCCL_WARNING("[NsRecovery][Resume] The current communication has not loaded op, no need to resume.");
2058 : }
2059 : isSuspended = false;
2060 : isCleaned = false;
2061 : HCCL_INFO("[NsRecovery][Resume] Resume success.");
2062 : } else { // HOST场景不支持
2063 : HCCL_ERROR("[NsRecovery][Resume] HOST is not supported to resume.");
2064 : return HcclResult::HCCL_E_NOT_SUPPORT;
2065 : }
2066 : );
2067 1 : return HcclResult::HCCL_SUCCESS;
2068 : }
2069 :
2070 31 : const NotifyTimeoutCfg &CommunicatorImpl::GetNotifyTimeoutCfg() const
2071 : {
2072 31 : return notifyTimeoutCfg;
2073 : }
2074 :
2075 : /* 当前接口中申请的buffer都是ge图模式下使用 */
2076 9 : HcclResult CommunicatorImpl::CreateCommCclBuf()
2077 : {
2078 27 : HCCL_INFO("[%s] start.", __func__);
2079 9 : if (inCclBuffer == nullptr) {
2080 2 : inCclBuffer = std::make_shared<DevBuffer>(cclBufferSize);
2081 6 : HCCL_INFO("CommunicatorImpl::CreateCommCclBuf, inCclBuffer is %p", inCclBuffer.get());
2082 : }
2083 9 : if (outCclBuffer == nullptr) {
2084 2 : outCclBuffer = std::make_shared<DevBuffer>(cclBufferSize);
2085 6 : HCCL_INFO("CommunicatorImpl::CreateCommCclBuf, outCclBuffer is %p", outCclBuffer.get());
2086 : }
2087 9 : if (indirectInCclBuffer == nullptr) {
2088 2 : indirectInCclBuffer = std::make_shared<DevBuffer>(sizeof(uintptr_t));
2089 6 : HCCL_INFO("Create Indirect In CclBuf success, indirectInCclBuffer = %p", indirectInCclBuffer.get());
2090 : }
2091 9 : if (indirectOutCclBuffer == nullptr) {
2092 2 : indirectOutCclBuffer = std::make_shared<DevBuffer>(sizeof(uintptr_t));
2093 6 : HCCL_INFO("Create Indirect out CclBuf success, indirectOutCclBuffer = %p", indirectOutCclBuffer.get());
2094 : }
2095 9 : return HcclResult::HCCL_SUCCESS;
2096 : }
2097 :
2098 1 : HcclResult CommunicatorImpl::GetInCclBuf(void *&commInputPtr, u64 &commInputSize)
2099 : {
2100 1 : CHK_PTR_NULL(inCclBuffer);
2101 1 : commInputSize = inCclBuffer->GetSize();
2102 1 : commInputPtr = reinterpret_cast<void*>(inCclBuffer->GetAddr());
2103 1 : return HcclResult::HCCL_SUCCESS;
2104 : }
2105 :
2106 1 : HcclResult CommunicatorImpl::GetOutCclBuf(void *&commOutputPtr, u64 &commOutputSize)
2107 : {
2108 1 : CHK_PTR_NULL(outCclBuffer);
2109 1 : commOutputSize = outCclBuffer->GetSize();
2110 1 : commOutputPtr = reinterpret_cast<void*>(outCclBuffer->GetAddr());
2111 1 : return HcclResult::HCCL_SUCCESS;
2112 : }
2113 :
2114 1 : HcclResult CommunicatorImpl::GetIndirectInCclBuf(void *&commIndirectInputPtr, u64 &commIndirectInputSize)
2115 : {
2116 3 : HCCL_INFO("[%s] start.", __func__);
2117 1 : CreateCommCclBuf();
2118 1 : commIndirectInputPtr = reinterpret_cast<void*>(indirectInCclBuffer->GetAddr());
2119 1 : commIndirectInputSize = indirectInCclBuffer->GetSize();
2120 3 : HCCL_INFO("GetIndirectInCclBuf: commIndirectInputPtr[%p], commIndirectInputSize[%lu]", commIndirectInputPtr, commIndirectInputSize);
2121 1 : return HcclResult::HCCL_SUCCESS;
2122 : }
2123 :
2124 1 : HcclResult CommunicatorImpl::GetIndirectOutCclBuf(void *&commIndirectOutputPtr, u64 &commIndirectOutputSize)
2125 : {
2126 3 : HCCL_INFO("[%s] start.", __func__);
2127 1 : CreateCommCclBuf();
2128 1 : commIndirectOutputPtr = reinterpret_cast<void*>(indirectOutCclBuffer->GetAddr());
2129 1 : commIndirectOutputSize = indirectOutCclBuffer->GetSize();
2130 3 : HCCL_INFO("GetIndirectOutCclBuf: commIndirectOutputPtr[%p], commIndirectOutputSize[%lu]", commIndirectOutputPtr, commIndirectOutputSize);
2131 1 : return HcclResult::HCCL_SUCCESS;
2132 : }
2133 :
2134 10 : bool CommunicatorImpl::IsWorldGroup() const
2135 : {
2136 10 : return isWorldGroup;
2137 : }
2138 :
2139 0 : bool CommunicatorImpl::IsCommReady()
2140 : {
2141 0 : CHECK_NULLPTR(collService, "[CommunicatorImpl::IsCommReady] collService is nullptr!");
2142 0 : if (collService->IsAllTransportRecoveredReady(GetId())) {
2143 : // 遗留问题：对Comm状态置为ready
2144 0 : SetCommStatus(CommStatus::COMM_READY);
2145 0 : return true;
2146 : } else {
2147 0 : return false;
2148 : }
2149 : }
2150 :
2151 6 : HcclResult CommunicatorImpl::GetSnapShotDynamicBuf(BinaryStream &buf) const
2152 : {
2153 18 : HCCL_INFO("[CommunicatorImpl][%s] opExecuteConfig.accState is [%u], commExecuteConfig.accState "
2154 : "is [%u], isLoadOp is [%d]",
2155 : __func__, static_cast<u32>(opExecuteConfig.accState), static_cast<u32>(commExecuteConfig.accState),
2156 : isLoadOp);
2157 6 : buf << static_cast<u32>(opExecuteConfig.accState); // 算子粒度 和 通信域粒度 都保存
2158 6 : buf << static_cast<u32>(commExecuteConfig.accState);
2159 6 : buf << isLoadOp;
2160 :
2161 6 : buf << submittedOpCnt;
2162 18 : HCCL_INFO("[CommunicatorImpl][%s], rank[%d], submittedOpCnt[%u]", __func__, myRank, submittedOpCnt);
2163 6 : if (submittedOpCnt == 0) {
2164 5 : return HcclResult::HCCL_SUCCESS;
2165 : }
2166 :
2167 1 : if (currentCollOperator) {
2168 3 : HCCL_INFO("[CommunicatorImpl][%s] opMode is %u", __func__, static_cast<u32>(currentCollOperator->opMode));
2169 1 : buf << static_cast<u32>(currentCollOperator->opMode);
2170 :
2171 3 : HCCL_INFO("[CommunicatorImpl][%s] rank[%d], currentCollOperator", __func__, myRank);
2172 1 : TRY_CATCH_RETURN(collService->GetSnapShotDynamicBuf(*currentCollOperator, buf));
2173 : }
2174 1 : return HcclResult::HCCL_SUCCESS;
2175 : }
2176 :
2177 2 : u32 CommunicatorImpl::GetRanktableCrc(bool isContainLoaId) const
2178 : {
2179 6 : HCCL_INFO("[CommunicatorImpl][%s], rank[%d], id[%s], idIdex[%u]", __func__, myRank, id.c_str(), idIndex);
2180 8 : CHK_PTR_NULL(ranktableInfo);
2181 0 : vector<char> ranktableBuf = ranktableInfo->GetUniqueId(isContainLoaId);
2182 0 : CheckCrc crc;
2183 0 : u32 crcValue = 0;
2184 0 : auto ret = crc.Calc32Crc(reinterpret_cast<const char*>(ranktableBuf.data()), ranktableBuf.size(), &crcValue);
2185 0 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[CommunicatorImpl][GetRanktableCrc] calculate crc failed, ret[%d]", ret),
2186 : ret);
2187 0 : return crcValue;
2188 0 : }
2189 :
2190 : // 恢复全局通信域
2191 3 : HcclResult CommunicatorImpl::RecoverComm(SnapShotComm &snapShotComm, u32 stepParam, const char *changeInfo)
2192 : {
2193 3 : if (!initFlag) {
2194 2 : initFlag = true;
2195 : try {
2196 6 : HCCL_INFO("[CommunicatorImpl][%s], rank[%d]", __func__, myRank);
2197 : // 将状态设置为resuming
2198 2 : if (GetCommStatus() == CommStatus::COMM_IDLE) {
2199 2 : SetCommStatus(CommStatus::COMM_RESUMING);
2200 : } else {
2201 0 : HCCL_ERROR("Communicator status is not idle, can not resume!");
2202 0 : return HcclResult::HCCL_E_INTERNAL;
2203 : }
2204 2 : RecoverOpMode(snapShotComm.opMode);
2205 2 : InitCommonData(snapShotComm.commParams, snapShotComm.config);
2206 2 : HrtSetDevice(devLogicId);
2207 2 : InitHccpHdc(); // 选择ccu加速模式依赖hdc通道打开ccu驱动
2208 2 : RecoverExeCfgData(snapShotComm.opExecuteConfig, snapShotComm.commExecuteConfig, snapShotComm.isLoadOp); // 算子粒度 和 通信域粒度都恢复
2209 2 : HcclResult recoverRet = RecoverRankGraphData(snapShotComm, changeInfo);
2210 1 : if (recoverRet != HCCL_SUCCESS) {
2211 0 : SetCommStatus(CommStatus::COMM_IDLE);
2212 0 : return recoverRet;
2213 : }
2214 1 : InitNotifyManager();
2215 1 : InitStreamManager();
2216 1 : InitSocketManager();
2217 1 : InitRmaConnManager();
2218 1 : InitDataBufferManager();
2219 1 : InitMemTransportManager();
2220 1 : InitHostDeviceSyncNotifyManager();
2221 1 : InitUbMemoryTransportMgr();
2222 1 : CollAlgComponentInit();
2223 1 : RegisterAicpuKernel();
2224 1 : InitCollService();
2225 1 : SelectCollService();
2226 1 : InitTraceManager();
2227 1 : DlProfFunction::GetInstance().DlProfFunctionInit();
2228 1 : InitMirrorTaskManager();
2229 1 : CHK_RET(InitProfilingReporter());
2230 1 : InitTaskExceptionHandler();
2231 1 : InitHDCommunicate();
2232 1 : notifyTimeoutCfg.Init();
2233 1 : RecoverTransportData(snapShotComm.submittedOpCnt, snapShotComm.levelRankPairs, stepParam, snapShotComm.linkGroupPair);
2234 1 : } catch (HcclException &e) {
2235 : // 异常时状态返回IDLE
2236 1 : SetCommStatus(CommStatus::COMM_IDLE);
2237 3 : HCCL_ERROR(e.what());
2238 1 : PrintBackTrace(e);
2239 1 : return e.GetErrorCode();
2240 1 : } catch (exception &e) {
2241 : // 异常时状态返回IDLE
2242 0 : SetCommStatus(CommStatus::COMM_IDLE);
2243 0 : HCCL_ERROR(e.what());
2244 0 : return HcclResult::HCCL_E_INTERNAL;
2245 0 : } catch (...) {
2246 : // 异常时状态返回IDLE
2247 0 : SetCommStatus(CommStatus::COMM_IDLE);
2248 0 : HCCL_ERROR("Unknown error occurs!");
2249 0 : return HcclResult::HCCL_E_INTERNAL;
2250 0 : }
2251 1 : return HcclResult::HCCL_SUCCESS;
2252 : }
2253 3 : HCCL_ERROR("[CommunicatorImpl][%s] Repeated calling init method!", __func__);
2254 1 : return HcclResult::HCCL_E_INTERNAL;
2255 : }
2256 :
2257 : // 恢复子通信域
2258 3 : HcclResult CommunicatorImpl::RecoverComm(const SnapShotSubComm &snapShotSubComm, std::unique_ptr<RankGraph> &inputRankGraph, u32 inputStep)
2259 : {
2260 3 : if (!initFlag) {
2261 2 : initFlag = true;
2262 : try {
2263 6 : HCCL_INFO("[CommunicatorImpl][%s], rank[%d]", __func__, myRank);
2264 : // 将状态设置为resuming
2265 2 : if (GetCommStatus() == CommStatus::COMM_IDLE) {
2266 2 : SetCommStatus(CommStatus::COMM_RESUMING);
2267 : } else {
2268 0 : HCCL_ERROR("Communicator status is not idle, can not resume!");
2269 0 : return HcclResult::HCCL_E_INTERNAL;
2270 : }
2271 2 : RecoverOpMode(snapShotSubComm.opMode);
2272 2 : InitCommonDataNotInitDevType(snapShotSubComm.commParams, snapShotSubComm.config);
2273 2 : HrtSetDevice(devLogicId);
2274 2 : InitHccpHdc(); // 选择ccu加速模式依赖hdc通道打开ccu驱动
2275 2 : RecoverExeCfgData(snapShotSubComm.opExecuteConfig, snapShotSubComm.commExecuteConfig, snapShotSubComm.isLoadOp); // 算子粒度 和 通信域粒度都恢复
2276 2 : InitRankGraph(inputRankGraph);
2277 1 : InitNotifyManager();
2278 1 : InitStreamManager();
2279 1 : InitSocketManager();
2280 1 : InitRmaConnManager();
2281 1 : InitDataBufferManager();
2282 1 : InitMemTransportManager();
2283 1 : InitHostDeviceSyncNotifyManager();
2284 1 : InitUbMemoryTransportMgr();
2285 1 : CollAlgComponentInit();
2286 1 : RegisterAicpuKernel();
2287 1 : InitCollService();
2288 1 : SelectCollService();
2289 1 : InitTraceManager();
2290 1 : DlProfFunction::GetInstance().DlProfFunctionInit();
2291 1 : InitMirrorTaskManager();
2292 1 : CHK_RET(InitProfilingReporter());
2293 1 : InitTaskExceptionHandler();
2294 1 : InitHDCommunicate();
2295 1 : RecoverTransportData(snapShotSubComm.submittedOpCnt, snapShotSubComm.levelRankPairs, inputStep, snapShotSubComm.linkGroupPair);
2296 1 : } catch (HcclException &e) {
2297 : // 异常时状态返回IDLE
2298 1 : SetCommStatus(CommStatus::COMM_IDLE);
2299 3 : HCCL_ERROR(e.what());
2300 1 : PrintBackTrace(e);
2301 1 : return e.GetErrorCode();
2302 1 : } catch (exception &e) {
2303 : // 异常时状态返回IDLE
2304 0 : SetCommStatus(CommStatus::COMM_IDLE);
2305 0 : HCCL_ERROR(e.what());
2306 0 : return HcclResult::HCCL_E_INTERNAL;
2307 0 : } catch (...) {
2308 : // 异常时状态返回IDLE
2309 0 : SetCommStatus(CommStatus::COMM_IDLE);
2310 0 : HCCL_ERROR("Unknown error occurs!");
2311 0 : return HcclResult::HCCL_E_INTERNAL;
2312 0 : }
2313 1 : return HcclResult::HCCL_SUCCESS;
2314 : }
2315 3 : HCCL_ERROR("Repeated calling init method!");
2316 1 : return HcclResult::HCCL_E_INTERNAL;
2317 : }
2318 5 : HcclResult CommunicatorImpl::RecoverOpMode(u32 opMode)
2319 : {
2320 5 : if (currentCollOperator == nullptr) {
2321 3 : currentCollOperator = make_unique<CollOperator>();
2322 : }
2323 5 : currentCollOperator->opMode = static_cast<OpMode::Value>(opMode);
2324 5 : return HcclResult::HCCL_SUCCESS;
2325 : }
2326 : // 创建子虚拟拓扑并恢复子通信域
2327 1 : HcclResult CommunicatorImpl::RecoverSubComm(const SnapShotSubComm &snapShotSubComm, CommunicatorImpl *subCommImpl, u32 step)
2328 : {
2329 3 : HCCL_INFO("[CommunicatorImpl][%s] start, myRank is [%d]", __func__, myRank);
2330 1 : vector<u32> rankIds;
2331 4 : for(u32 i = 0; i < snapShotSubComm.rankIds.size(); ++i) {
2332 3 : rankIds.push_back(static_cast<u32>(snapShotSubComm.rankIds[i]));
2333 : }
2334 : try {
2335 1 : if (initFlag) {
2336 : // 创建子虚拟拓扑
2337 0 : std::unique_ptr<RankGraph> subRankGraph = rankGraph->CreateSubRankGraph(rankIds);
2338 : // 初始化子通信域
2339 0 : return subCommImpl->RecoverComm(snapShotSubComm, subRankGraph, step);
2340 0 : } else {
2341 : // 异常时状态返回IDLE
2342 1 : SetCommStatus(CommStatus::COMM_IDLE);
2343 1 : std::string msg = StringFormat("CreateSubComm fail, communicator has not been initialized, please check.");
2344 1 : THROW<InternalException>(msg);
2345 1 : }
2346 1 : } catch (HcclException &e) {
2347 : // 异常时状态返回IDLE
2348 1 : SetCommStatus(CommStatus::COMM_IDLE);
2349 3 : HCCL_ERROR(e.what());
2350 1 : PrintBackTrace(e);
2351 1 : return e.GetErrorCode();
2352 1 : } catch (exception &e) {
2353 : // 异常时状态返回IDLE
2354 0 : SetCommStatus(CommStatus::COMM_IDLE);
2355 0 : HCCL_ERROR(e.what());
2356 0 : return HcclResult::HCCL_E_INTERNAL;
2357 0 : } catch (...) {
2358 : // 异常时状态返回IDLE
2359 0 : SetCommStatus(CommStatus::COMM_IDLE);
2360 0 : HCCL_ERROR("Unknown error occurs!");
2361 0 : return HcclResult::HCCL_E_INTERNAL;
2362 0 : }
2363 : HCCL_ERROR("CreateSubComm fail !");
2364 : return HcclResult::HCCL_E_INTERNAL;
2365 1 : }
2366 : // 恢复全局通信域拓扑信息
2367 2 : HcclResult CommunicatorImpl::RecoverRankGraphData(SnapShotComm &snapShotComm, const char *changeInfo)
2368 : {
2369 6 : HCCL_INFO("[CommunicatorImpl][%s] start, rank[%d]", __func__, myRank);
2370 :
2371 : // 根据changedInfo更新快照信息
2372 2 : auto ret = DiffRankUpdater(changeInfo, snapShotComm.rankTableInfo);
2373 2 : if (ret != HcclResult::HCCL_SUCCESS) {
2374 2 : THROW<InternalException>("DiffRankUpdater failed");
2375 : }
2376 :
2377 0 : const RankGraphBuilderBridge *bridge = GetRankGraphBuilderBridge();
2378 0 : CHK_PRT_RET(bridge == nullptr || bridge->recoverBuild == nullptr,
2379 : HCCL_ERROR("[%s] RankGraphBuilder bridge is not registered.", __func__), HCCL_E_INTERNAL);
2380 :
2381 0 : RankGraphBuildResult buildResult{};
2382 : // 恢复构建也由 hcomm 完成，并返回同一版本的 graph、rank table 和 topo 快照。
2383 0 : ret = bridge->recoverBuild(snapShotComm.rankTableInfo, snapShotComm.topoInfo, myRank, buildResult);
2384 0 : CHK_PRT_RET(ret != HCCL_SUCCESS,
2385 : HCCL_ERROR("[%s] recover rank graph failed, errNo[0x%016llx].", __func__, HCCL_ERROR_CODE(ret)), ret);
2386 0 : CHK_PRT_RET(buildResult.rankGraph == nullptr,
2387 : HCCL_ERROR("[%s] recover rank graph result is invalid.", __func__), HCCL_E_INTERNAL);
2388 :
2389 0 : std::unique_ptr<RankTableInfo> preparedRankTableInfo;
2390 0 : std::shared_ptr<TopoInfo> preparedTopoInfo;
2391 0 : PrepareRankGraphMetadata(
2392 : buildResult.rankTableInfo, buildResult.topoInfo, preparedRankTableInfo, preparedTopoInfo);
2393 :
2394 0 : const u32 preparedRankSize = buildResult.rankGraph->GetRankSize();
2395 : // 仅在恢复结果和两份元数据准备完整后，用不分配内存的移动赋值整体替换拓扑状态。
2396 0 : rankGraph = std::move(buildResult.rankGraph);
2397 0 : ranktableInfo = std::move(preparedRankTableInfo);
2398 0 : topoInfo = std::move(preparedTopoInfo);
2399 0 : rankSize = preparedRankSize;
2400 0 : HCCL_INFO(
2401 : "[CommunicatorImpl][%s] Recover topo data from snapshot, rank[%d], id[%s], idIndex[%u], RankTableInfo[%s]", __func__,
2402 : myRank, id.c_str(), idIndex, ranktableInfo->Describe().c_str());
2403 :
2404 0 : CheckRankGraph();
2405 0 : HCCL_INFO("Recover topo data from snapshot success.");
2406 0 : return HcclResult::HCCL_SUCCESS;
2407 0 : }
2408 : // 恢复通信域transport信息
2409 0 : HcclResult CommunicatorImpl::RecoverTransportData(u32 savedSubmittedOpCnt, const vector<std::pair<u32, RankId>> &levelRankPairs, u32 savedStep, vector<std::pair<LinkGroup, u32>> linkGroupPair)
2410 : {
2411 0 : HCCL_INFO("[CommunicatorImpl][%s] Recover transport data from snapshot.levelRankPairs size is %u", __func__, levelRankPairs.size());
2412 0 : vector<LinkData> links;
2413 :
2414 0 : for (uint32_t i = 0; i < levelRankPairs.size(); ++i) {
2415 0 : CHK_PTR_NULL(rankGraph);
2416 0 : std::vector<NetInstance::Path> paths = rankGraph->GetPaths(levelRankPairs[i].first, myRank, levelRankPairs[i].second);
2417 0 : for (NetInstance::Path &path : paths) {
2418 0 : links.emplace_back(LinkData(path));
2419 : }
2420 0 : }
2421 : // 指令的下标是指令的个数 - 1
2422 0 : collOpIndex = savedSubmittedOpCnt - 1;
2423 0 : step = savedStep;
2424 : // 建transport
2425 0 : collService->RecoverTransport(links, linkGroupPair);
2426 0 : HCCL_INFO("Recover transport data from snapshot success.");
2427 0 : return HcclResult::HCCL_SUCCESS;
2428 0 : }
2429 :
2430 84 : void CommunicatorImpl::WaitReady() const
2431 : {
2432 84 : constexpr u32 loadWaitTimeOut = 10 * 1000; // 待修改，定义最大等待10秒
2433 84 : auto timeout = std::chrono::milliseconds(loadWaitTimeOut);
2434 :
2435 252 : HCCL_INFO("[CommunicatorImpl][%s] start", __func__);
2436 84 : HcclUs startTime = std::chrono::steady_clock::now();
2437 : while (true) {
2438 400298776 : if (GetCommStatus() == CommStatus::COMM_READY) {
2439 82 : break;
2440 : }
2441 400298694 : if ((std::chrono::steady_clock::now() - startTime) >= timeout) {
2442 2 : THROW<InternalException>("Wait COMM_READY timeout, commId[%s]", id.c_str());
2443 : }
2444 : }
2445 246 : HCCL_INFO("[CommunicatorImpl][%s] end", __func__);
2446 82 : }
2447 :
2448 5 : u32 CommunicatorImpl::GetCollOpIndex() const
2449 : {
2450 5 : return collOpIndex;
2451 : }
2452 :
2453 5 : u32 CommunicatorImpl::GetStep() const
2454 : {
2455 5 : return step;
2456 : }
2457 :
2458 4 : std::set<RankId> CommunicatorImpl::GetNeighboorRanks() const
2459 : {
2460 4 : return rankGraph->GetNetInstanceByRankId(0,myRank)->GetRankIds();
2461 : }
2462 :
2463 284 : void CommunicatorImpl::InitMirrorTaskManager()
2464 : {
2465 284 : mirrorTaskManager = std::make_unique<MirrorTaskManager>(devLogicId,
2466 284 : &GlobalMirrorTasks::Instance(), false); // host侧写死
2467 284 : }
2468 :
2469 130 : MirrorTaskManager &CommunicatorImpl::GetMirrorTaskManager() const
2470 : {
2471 131 : CHECK_NULLPTR(mirrorTaskManager, "mirrorTaskManager is nullptr!");
2472 129 : return *mirrorTaskManager;
2473 : }
2474 :
2475 841 : CommunicatorImpl::~CommunicatorImpl()
2476 : {
2477 2523 : HCCL_INFO("[~CommunicatorImpl] start CommunicatorImpl destroy, commId[%s]", id.c_str());
2478 841 : (void)DestroyDpuKernelResource();
2479 841 : (void)DestroyDpuTaskexpShmemInDevice();
2480 841 : (void)DestroyKFCWorkSpaceVA();
2481 : // 释放
2482 841 : if (hostShareBuf != nullptr) {
2483 2 : free(hostShareBuf);
2484 2 : hostShareBuf = nullptr;
2485 : }
2486 841 : auto outerIt = g_taskServiceMap.find(id);
2487 841 : if (outerIt != g_taskServiceMap.end()) {
2488 0 : outerIt->second.erase(devLogicId);
2489 0 : if (outerIt->second.empty()) {
2490 0 : g_taskServiceMap.erase(id);
2491 : }
2492 : }
2493 841 : g_taskExpMemMap.erase(id);
2494 841 : (void)NotifyAicpuDestroyComm();
2495 841 : ccuDrvHandle = nullptr;
2496 :
2497 841 : DeInitPreResource();
2498 841 : }
2499 :
2500 841 : HcclResult CommunicatorImpl::DestroyDpuKernelResource()
2501 : {
2502 : // 终止Dpu Kernel的TaskRun
2503 841 : if (!isDpuKernelLaunched) {
2504 830 : return HCCL_SUCCESS;
2505 : }
2506 :
2507 17 : CHK_RET(WaitDpuKernelThreadTerminate());
2508 :
2509 : // 切换回 dpu ctx
2510 9 : aclError aclRet = aclrtSetCurrentContext(dpuContext);
2511 9 : if (ACL_SUCCESS != aclRet) {
2512 0 : HCCL_ERROR("set dpu Ctx Failed, aclReturn[%d]", aclRet);
2513 0 : return HCCL_E_RUNTIME;
2514 : }
2515 : // 销毁局部流
2516 9 : aclRet = aclrtDestroyStreamForce(dpuStream);
2517 9 : if (ACL_SUCCESS != aclRet) {
2518 6 : HCCL_ERROR("Destroy Stream Failed, aclReturn[%d]", aclRet);
2519 2 : aclRet = aclrtSetCurrentContext(npuContext);
2520 8 : CHK_PRT_RET(aclRet == ACL_SUCCESS, HCCL_ERROR("set npu Ctx Failed, aclReturn[%d]", aclRet), HCCL_E_RUNTIME);
2521 0 : return HCCL_E_RUNTIME;
2522 : }
2523 7 : if (g_commNum > 1) {
2524 1 : g_commNum--;
2525 : } else {
2526 : // reset DPU kernel 线程
2527 6 : HcclResult ret = HrtResetXpuDevice(TEMP_DEV_TYPE_DPU, 0);
2528 6 : if (HCCL_SUCCESS != ret) {
2529 18 : HCCL_ERROR("ResetXpuDevice Failed, return[%d]", ret);
2530 6 : aclRet = aclrtSetCurrentContext(npuContext);
2531 24 : CHK_PRT_RET(aclRet == ACL_SUCCESS, HCCL_ERROR("set npu Ctx Failed, aclReturn[%d]", aclRet), HCCL_E_RUNTIME);
2532 0 : return HCCL_E_RUNTIME;
2533 : }
2534 : }
2535 : // 切回 npu ctx
2536 1 : aclRet = aclrtSetCurrentContext(npuContext);
2537 1 : if (ACL_SUCCESS != aclRet) {
2538 0 : HCCL_ERROR("set npu Ctx Failed, aclReturn[%d]", aclRet);
2539 0 : return HCCL_E_RUNTIME;
2540 : }
2541 :
2542 1 : return HCCL_SUCCESS;
2543 : }
2544 :
2545 10 : HcclResult CommunicatorImpl::WaitDpuKernelThreadTerminate()
2546 : {
2547 10 : if (!isDpuKernelLaunched) {
2548 1 : return HCCL_SUCCESS;
2549 : }
2550 :
2551 39 : if (tagDpuShmemArgsMap_[DPUTAG].va_ == nullptr || tagDpuShmemArgsMap_[DPUTAG].accessVA_ == nullptr) {
2552 21 : HCCL_ERROR("[CommunicatorImpl::%s] va_[%p] or accessVA_[%p] is nullptr", __func__, tagDpuShmemArgsMap_[DPUTAG].va_, tagDpuShmemArgsMap_[DPUTAG].accessVA_);
2553 3 : return HCCL_E_MEMORY;
2554 : }
2555 24 : void* hostPtr = tagDpuShmemArgsMap_[DPUTAG].connectType_ == HOST_DEVICE_CONNECT_TYPE_UB ? tagDpuShmemArgsMap_[DPUTAG].va_ : tagDpuShmemArgsMap_[DPUTAG].accessVA_;
2556 : // 先读取是否有dpu线程回写的退出信号
2557 6 : uint8_t flag = 0;
2558 6 : errno_t ret = memcpy_s(&flag, sizeof(flag), hostPtr, sizeof(flag));
2559 6 : if (ret != EOK) {
2560 0 : HCCL_ERROR("Read Terminate TaskRun Signal Fail, return[%d]", ret);
2561 0 : return HCCL_E_INTERNAL;
2562 : }
2563 6 : if (flag == DEVICE_SIGNAL_THIRD) {
2564 5 : return HCCL_SUCCESS; // dpu线程已退出
2565 : }
2566 1 : flag = DEVICE_SIGNAL_SECOND;
2567 1 : ret = memcpy_s(hostPtr, sizeof(flag), &flag, sizeof(flag));
2568 1 : if (ret != EOK) {
2569 0 : HCCL_ERROR("Terminate TaskRun Fail, return[%d]", ret);
2570 0 : return HCCL_E_INTERNAL;
2571 : }
2572 : do {
2573 3522411 : ret = memcpy_s(&flag, sizeof(flag), hostPtr, sizeof(flag));
2574 3522411 : if (ret != EOK) {
2575 0 : HCCL_ERROR("Read Terminate TaskRun Signal Fail, return[%d]", ret);
2576 0 : return HCCL_E_INTERNAL;
2577 : }
2578 3522411 : } while (flag != DEVICE_SIGNAL_THIRD);
2579 1 : return HCCL_SUCCESS;
2580 : }
2581 :
2582 842 : HcclResult CommunicatorImpl::DestroyDpuTaskexpShmemInDevice()
2583 : {
2584 842 : if (!isDpuKernelLaunched) { // 非dpu场景,无共享内存申请
2585 832 : return HCCL_SUCCESS;
2586 : }
2587 :
2588 30 : if (tagDpuShmemArgsMap_[DPUTASKEXCEPTION].va_ == nullptr || tagDpuShmemArgsMap_[DPUTASKEXCEPTION].accessVA_ == nullptr) {
2589 70 : HCCL_ERROR("[CommunicatorImpl::%s] va_[%p] or accessVA_[%p] is nullptr", __func__, tagDpuShmemArgsMap_[DPUTASKEXCEPTION].va_, tagDpuShmemArgsMap_[DPUTASKEXCEPTION].accessVA_);
2590 10 : return HCCL_E_MEMORY;
2591 : }
2592 0 : void* hostPtr = tagDpuShmemArgsMap_[DPUTASKEXCEPTION].connectType_ == HOST_DEVICE_CONNECT_TYPE_UB ? tagDpuShmemArgsMap_[DPUTASKEXCEPTION].va_ : tagDpuShmemArgsMap_[DPUTASKEXCEPTION].accessVA_;
2593 0 : uint8_t flag = 1;
2594 0 : errno_t ret = memcpy_s(hostPtr, sizeof(flag), &flag, sizeof(flag)); // taskexception首字节作为是否停止标志位；
2595 0 : if (ret != EOK) {
2596 0 : HCCL_ERROR("DestroyDpuTaskexpShmemInDevice Fail, return[%d]", ret);
2597 0 : return HCCL_E_INTERNAL;
2598 : }
2599 0 : auto timeout = std::chrono::milliseconds(WAIT_CMD_TIMEOUT);
2600 0 : auto startTime = std::chrono::steady_clock::now();
2601 : do { // 等待aicpu背景线程response
2602 0 : ret = memcpy_s(&flag, sizeof(flag), hostPtr, sizeof(flag));
2603 0 : if (ret != EOK) {
2604 0 : HCCL_ERROR("Read Device Response Fail, return[%d]", ret);
2605 0 : return HCCL_E_INTERNAL;
2606 : }
2607 0 : if ((std::chrono::steady_clock::now() - startTime) >= timeout) {
2608 0 : HCCL_WARNING("[CommunicatorImpl][DestroyDpuTaskexpShmemInDevice] Wait device destroy response timeout[%u ms].", WAIT_CMD_TIMEOUT);
2609 0 : return HcclResult::HCCL_E_TIMEOUT;
2610 : }
2611 0 : } while (flag != 0);
2612 0 : return HCCL_SUCCESS;
2613 : }
2614 :
2615 841 : HcclResult CommunicatorImpl::NotifyAicpuDestroyComm()
2616 : {
2617 841 : if (!isAicpuKernelLaunched) {
2618 2499 : HCCL_WARNING("[%s] isAicpuKernelLaunched is false", __func__);
2619 833 : return HcclResult::HCCL_SUCCESS;
2620 : }
2621 :
2622 8 : if (kfcControlTransferH2D == nullptr) {
2623 18 : HCCL_WARNING("[%s] kfcControlTransferH2D is null", __func__);
2624 6 : return HcclResult::HCCL_SUCCESS;
2625 : }
2626 :
2627 2 : KfcCommand opCmd = KfcCommand::DESTROY_AICPU_COMM;
2628 6 : HCCL_INFO("[%s] send KfcCommand[%d] begin, which is DESTROY_AICPU_COMM.", __func__, opCmd);
2629 2 : CHK_RET(kfcControlTransferH2D->Put(0, sizeof(KfcCommand), reinterpret_cast<uint8_t *>(&opCmd)));
2630 6 : HCCL_INFO("[%s] send KfcCommand[%d] success, which is DESTROY_AICPU_COMM.", __func__, opCmd);
2631 2 : KfcExecStatus opInfo;
2632 2 : auto timeout = std::chrono::milliseconds(WAIT_CMD_TIMEOUT);
2633 2 : auto startTime = std::chrono::steady_clock::now();
2634 : while (true) {
2635 4753126 : CHK_RET(kfcStatusTransferD2H->Get(0, sizeof(KfcExecStatus), reinterpret_cast<uint8_t *>(&opInfo)));
2636 4753126 : if (opInfo.kfcStatus == KfcStatus::DESTROY_AICPU_COMM_DONE) {
2637 0 : HCCL_INFO("[%s] get KfcStatus[%d], which is DESTROY_AICPU_COMM_DONE", __func__, opInfo.kfcStatus);
2638 0 : return HcclResult::HCCL_SUCCESS;
2639 : }
2640 4753126 : if ((std::chrono::steady_clock::now() - startTime) >= timeout) {
2641 6 : HCCL_WARNING("[%s] Wait suspend response status timeout[%u ms] and get the "
2642 : "opExecStatus is [%u].", __func__,
2643 : WAIT_CMD_TIMEOUT, opInfo.kfcStatus);
2644 2 : return HcclResult::HCCL_E_TIMEOUT;
2645 : }
2646 4753124 : }
2647 :
2648 : return HcclResult::HCCL_SUCCESS;
2649 : }
2650 :
2651 257 : HcclResult CommunicatorImpl::InitProfilingReporter()
2652 : {
2653 514 : profilingReporter = std::make_unique<ProfilingReporter>(mirrorTaskManager.get(),
2654 514 : &ProfilingHandler::GetInstance());
2655 257 : CHK_RET(ProfilingHandler::GetInstance().Init());
2656 257 : CHK_RET(profilingReporter->Init());
2657 257 : return HCCL_SUCCESS;
2658 : }
2659 :
2660 0 : ProfilingReporter &CommunicatorImpl::GetProfilingReporter() const
2661 : {
2662 0 : CHECK_NULLPTR(profilingReporter, "profilingReporter is nullptr!");
2663 0 : return *profilingReporter;
2664 : }
2665 :
2666 8 : HcclResult CommunicatorImpl::GetOneSidedService(HcclOneSidedService** service) const
2667 : {
2668 8 : CHECK_NULLPTR(oneSidedService, "oneSidedService is nullptr!");
2669 8 : *service = oneSidedService.get();
2670 8 : return HCCL_SUCCESS;
2671 : }
2672 :
2673 3 : void CommunicatorImpl::UpdateProfStat()
2674 : {
2675 3 : profilingReporter->UpdateProfStat();
2676 3 : }
2677 :
2678 1 : void CommunicatorImpl::ReportProfInfo(uint64_t beginTime, bool cachedReq, bool opbased)
2679 : {
2680 : // 上报task信息
2681 1 : profilingReporter->ReportAllTasks(cachedReq);
2682 :
2683 : // 上报opInfo信息
2684 1 : profilingReporter->ReportOp(beginTime, cachedReq, opbased);
2685 1 : }
2686 :
2687 41 : void CommunicatorImpl::InitTaskExceptionHandler() const
2688 : {
2689 41 : TaskExceptionHandler* handler = TaskExceptionHandlerManager::GetHandler(static_cast<size_t>(devLogicId));
2690 41 : CHECK_NULLPTR(handler, "handler is nullptr!");
2691 41 : }
2692 :
2693 39 : void CommunicatorImpl::InitOneSidedService()
2694 : {
2695 117 : HCCL_INFO("[CommunicatorImpl][InitOneSidedService] start!");
2696 39 : oneSidedService = std::make_unique<HcclOneSidedService>(*this);
2697 117 : HCCL_INFO("[CommunicatorImpl][InitOneSidedService] end!");
2698 39 : }
2699 :
2700 2 : u32 CommunicatorImpl::GetUsedChannelCount(u32 dieId)
2701 : {
2702 2 : CHECK_NULLPTR(collService, "collService is nullptr!");
2703 2 : if (!GetOpCcuFeatureFlag()) { return 0; } // 防止非ccu模式进入
2704 2 : CHECK_NULLPTR(dynamic_cast<CollServiceDeviceMode *>(collService), "CollServiceDeviceMode is nullptr!");
2705 2 : CcuJettyMgr *ccuJettyMgr = dynamic_cast<CollServiceDeviceMode *>(collService)
2706 : ->GetCcuInsPreprocessor()
2707 : ->GetCcuComm()
2708 4 : ->GetCcuJettyMgr();
2709 :
2710 2 : if (ccuJettyMgr == nullptr) {
2711 3 : HCCL_WARNING("[CommunicatorImpl][%s] failed, ccuJettyMgr is nullptr, commId[%s].",
2712 : __func__, id.c_str());
2713 1 : return 0;
2714 : }
2715 1 : return ccuJettyMgr->GetUsedChannelCount(dieId);
2716 : }
2717 :
2718 13 : void CommunicatorImpl::RegisterPrintChannelInfoCallback(std::function<void()> callback)
2719 : {
2720 13 : printChannelInfoCallback = callback;
2721 13 : }
2722 :
2723 3 : void CommunicatorImpl::PrintChannelInfoCallback() const
2724 : {
2725 3 : if (printChannelInfoCallback == nullptr) {
2726 6 : HCCL_WARNING("[CommunicatorImpl][PrintChannelInfoCallback] commId[%s], callback function not registered.", id.c_str());
2727 2 : return;
2728 : }
2729 : // ccu建链时channel资源不足，调用回调函数做维测打印
2730 1 : printChannelInfoCallback();
2731 : }
2732 :
2733 276 : void CommunicatorImpl::SetCommStatus(CommStatus commStatus)
2734 : {
2735 276 : status_.store(commStatus);
2736 276 : }
2737 :
2738 400298898 : CommStatus CommunicatorImpl::GetCommStatus() const
2739 : {
2740 400298898 : return status_.load();
2741 : }
2742 :
2743 : std::map<HcclAccelerator, AcceleratorState> accStateMap = {
2744 : {HcclAccelerator::AICPU, AcceleratorState::AICPU_TS},
2745 : {HcclAccelerator::AICPU_TS, AcceleratorState::AICPU_TS},
2746 : {HcclAccelerator::CCU_SCHED, AcceleratorState::CCU_SCHED},
2747 : {HcclAccelerator::DEFAULT, AcceleratorState::CCU_SCHED},
2748 : {HcclAccelerator::CCU_MS, AcceleratorState::CCU_MS}
2749 : };
2750 :
2751 : // 初始化 算子粒度 = 通信域粒度 选择用 算子粒度 ok
2752 21 : void CommunicatorImpl::ExecAlgSelect(const CollOpParams &opParams, const OpMode &opMode)
2753 : {
2754 63 : HCCL_INFO("[CommunicatorImpl][%s] opType[%s], opMode[%s], primary accelerator[%s]", __func__, opParams.opType.Describe().c_str(),
2755 : opMode.Describe().c_str(), opExecuteConfig.accState.Describe().c_str());
2756 : // 调用算法选择接口，获取algName、展开方式、执行方式
2757 21 : CollAlgParams params;
2758 21 : params.opMode = opMode;
2759 21 : params.maxTmpMemSize = GetBufferSize();
2760 21 : params.isMc2 = opParams.isMc2;
2761 21 : if (opParams.isMc2) {
2762 0 : if(accStateMap.find(opParams.commEngine) == accStateMap.end()) {
2763 0 : THROW<NotSupportException>("[CommunicatorImpl][ExecAlgSelect] not support commEngine type[%s]!", opParams.commEngine.Describe().c_str());
2764 : }
2765 0 : opExecuteConfig.accState = accStateMap.find(opParams.commEngine)->second;
2766 : }
2767 21 : OpExecuteConfig inOpExecuteConfig = opExecuteConfig;
2768 21 : params.opExecuteConfig = inOpExecuteConfig;
2769 21 : params.algConfig = opParams.algConfig;
2770 :
2771 63 : HCCL_DEBUG("CommunicatorImpl::ExecAlgSelect currentCollOperator dataType[%s]", currentCollOperator->dataType.Describe().c_str());
2772 21 : auto ret = collAlgComponent->ExecAlgSelect(*currentCollOperator, params, curAlgName, inOpExecuteConfig);
2773 21 : if (ret != HcclResult::HCCL_SUCCESS) {
2774 : std::vector<HcclAlgoType> algos
2775 1 : = std::vector<HcclAlgoType>(HCCL_ALGO_LEVEL_NUM, HcclAlgoType::HCCL_ALGO_TYPE_DEFAULT);
2776 1 : auto configAlgMap = EnvConfig::GetInstance().GetAlgoConfig().GetAlgoConfig();
2777 1 : auto it = configAlgMap.find(opParams.opType);
2778 1 : if (it != configAlgMap.end()) {
2779 0 : algos = it->second;
2780 : }
2781 1 : auto dataSize = opParams.count * DataTypeSizeGet(opParams.dataType);
2782 6 : THROW<NotSupportException>(
2783 : "[CommunicatorImpl][ExecAlgSelect] failed. Error code :%u, opType[%s], opMode[%s], accState[%s], "
2784 : "dataType[%s], reduceOp[%s]. Current algName[%s],algos[0]:[%u],algos[1]:[%u],algos[2]:[%u],algos[3]:[%u], dataSize[%u Bytes] .",
2785 5 : ret, opParams.opType.Describe().c_str(), opMode.Describe().c_str(),
2786 5 : opExecuteConfig.accState.Describe().c_str(), opParams.dataType.Describe().c_str(),
2787 4 : opParams.reduceOp.Describe().c_str(), curAlgName.c_str(), algos[0], algos[1], algos[2], algos[3], dataSize);
2788 2 : }
2789 20 : if(params.isMc2 && (opExecuteConfig.accState == AcceleratorState::CCU_SCHED || opExecuteConfig.accState == AcceleratorState::CCU_MS)) {
2790 0 : algorithmType_ = collAlgComponent->GetAlgorithmTypeForMC2CCU(curAlgName);
2791 : }
2792 20 : auto opAcceStateCacheIt = opAcceStateCache.find({opParams.opType, curAlgName});
2793 20 : if (opAcceStateCacheIt != opAcceStateCache.end()) {
2794 0 : HCCL_INFO("[CommunicatorImpl][%s] opAcceStateCache find, reset accelerator[%s], algName[%s]", __func__, opAcceStateCacheIt->second.first.Describe().c_str(), opAcceStateCacheIt->second.second.c_str());
2795 0 : opExecuteConfig.accState = opAcceStateCacheIt->second.first;
2796 0 : curAlgName = opAcceStateCacheIt->second.second;
2797 0 : ExecAlgSelect(opParams, opMode); // 重新走算法选择(数据量、数据类型、reduce类型不一样，算法可能不一样)
2798 0 : return;
2799 : }
2800 20 : SetOpExecuteConfig(inOpExecuteConfig); // 算子粒度 ok
2801 60 : HCCL_INFO("[CommunicatorImpl][%s] current accelerator[%s], algName[%s], algorithmType[%u]", __func__,
2802 : opExecuteConfig.accState.Describe().c_str(), curAlgName.c_str(), algorithmType_);
2803 20 : SelectCollService();
2804 21 : }
2805 :
2806 52 : void CommunicatorImpl::SelectCollService()
2807 : {
2808 : // 根据执行方式和展开方式，选择对应的CollService
2809 52 : auto mapIt = collServices.find(GetOpExecuteConfig().accState); // 算子粒度
2810 52 : if (mapIt == collServices.end()) {
2811 : auto msg = StringFormat("[CommunicatorImpl][%s] not support, accelerator is %s", __func__,
2812 3 : GetOpExecuteConfig().accState.Describe().c_str());
2813 3 : THROW<NotSupportException>(msg);
2814 3 : }
2815 49 : collService = mapIt->second.get();
2816 49 : }
2817 :
2818 226 : void CommunicatorImpl::CollAlgComponentInit()
2819 : {
2820 226 : HcclMainboardId hcclMainboardId;
2821 226 : HrtGetMainboardId(devLogicId, hcclMainboardId);
2822 226 : CollAlgComponentBuilder collAlgComponentBuilder;
2823 226 : collAlgComponent = collAlgComponentBuilder.SetRankGraph(GetRankGraph().get())
2824 226 : .SetDevType(GetDevType())
2825 226 : .SetMyRank(GetMyRank())
2826 226 : .SetRankSize(GetRankSize())
2827 226 : .SetDmaMode(DmaMode::PUT)
2828 226 : .SetMainboardId(static_cast<uint8_t>(hcclMainboardId))
2829 678 : .EnableDetour(EnvConfig::GetInstance().GetDetourConfig().GetDetourType()
2830 226 : == HcclDetourType::HCCL_DETOUR_ENABLE_2P) // 当前仅支持2P绕路
2831 226 : .Build();
2832 226 : if (collAlgComponent == nullptr) {
2833 0 : HCCL_ERROR("collAlgComponent is a null pointer!");
2834 0 : throw NullPtrException("collAlgComponent is a null pointer!");
2835 : }
2836 678 : HCCL_INFO("[CommunicatorImpl][%s] finished initializing collAlgComponent.", __func__);
2837 226 : }
2838 :
2839 16 : HcclResult CommunicatorImpl::SetAccelerator(HcclAccelerator hcclAccelerator, bool isCcuMsAvailable)
2840 : {
2841 16 : if (isLoadOp) {
2842 : // 已下发过算子，不允许再设置accelerator
2843 3 : HCCL_ERROR("[CommunicatorImpl]SetAccelerator is not allowed after load op.");
2844 1 : return HCCL_E_NOT_SUPPORT;
2845 : }
2846 15 : AcceleratorState commAccelerator;
2847 15 : if (hcclAccelerator == HcclAccelerator::DEFAULT) { // 用户没有配，读环境变量
2848 14 : hcclAccelerator = EnvConfig::GetInstance().GetAlgoConfig().GetHcclAccelerator();
2849 42 : HCCL_RUN_INFO("[CommunicatorImpl][%s] env OpExpansionMode is [%s]", __func__, hcclAccelerator.Describe().c_str());
2850 : }
2851 15 : HcclMainboardId hcclMainboardId;
2852 15 : CHK_RET(HrtGetMainboardId(devLogicId, hcclMainboardId));
2853 :
2854 : // 开启新流程时，仅mc2场景走回legacy通信域，此时不允许使用ms模式
2855 45 : HCCL_WARNING("[CommunicatorImpl][%s] legacy communicator not support ccu ms mode for mc2.",
2856 : __func__);
2857 15 : isCcuMsAvailable = false;
2858 :
2859 15 : switch (hcclAccelerator) {
2860 1 : case HcclAccelerator::CCU_MS:
2861 1 : if (hcclMainboardId == HcclMainboardId::MAINBOARD_PCIE_STD) { // 标卡环境下配置CCU_MS加速模式拦截报错
2862 3 : HCCL_ERROR("[SetAccelerator] hcclAccelerator[%s] not support in %s", hcclAccelerator.Describe().c_str(), hcclMainboardId.Describe().c_str());
2863 1 : return HCCL_E_NOT_SUPPORT;
2864 : }
2865 0 : commAccelerator = isCcuMsAvailable ? AcceleratorState::CCU_MS : AcceleratorState::CCU_SCHED;
2866 0 : break;
2867 0 : case HcclAccelerator::CCU_SCHED:
2868 0 : commAccelerator = AcceleratorState::CCU_SCHED;
2869 0 : if (IsCommWithPCIEProtocol()) {
2870 : // 若当前通信域存在PCIE链路,不支持ccu展开,默认切换为aicpu展开,在大于8卡不支持aicpu场景由后续算法选择部分切换至aiv展开
2871 0 : commAccelerator = AcceleratorState::AICPU_TS;
2872 : }
2873 0 : break;
2874 0 : case HcclAccelerator::AIV:
2875 0 : commAccelerator = AcceleratorState::AIV;
2876 0 : break;
2877 0 : case HcclAccelerator::AIV_ONLY:
2878 0 : commAccelerator = AcceleratorState::AIV_ONLY;
2879 0 : break;
2880 14 : case HcclAccelerator::AICPU_TS:
2881 14 : commAccelerator = AcceleratorState::AICPU_TS;
2882 14 : break;
2883 0 : case HcclAccelerator::HOSTCPU_TS: // 950不支持HOST展开，进行拦截
2884 0 : HCCL_ERROR("[SetAccelerator] hcclAccelerator[%s] not support in 950", hcclAccelerator.Describe().c_str());
2885 0 : return HCCL_E_NOT_SUPPORT;
2886 0 : case HcclAccelerator::AICPU:
2887 0 : HCCL_ERROR("[SetAccelerator] hcclAccelerator[%s] not support", hcclAccelerator.Describe().c_str());
2888 0 : return HCCL_E_NOT_SUPPORT;
2889 0 : default:
2890 0 : HCCL_ERROR("[SetAccelerator] hcclAccelerator[%s] internal error", hcclAccelerator.Describe().c_str());
2891 0 : return HCCL_E_INTERNAL;
2892 : }
2893 14 : OpExecuteConfig inCommExecuteConfig;
2894 14 : inCommExecuteConfig.accState = commAccelerator;
2895 42 : HCCL_DEBUG("[CommunicatorImpl][%s] inCommExecuteConfig[%s]", __func__, inCommExecuteConfig.accState.Describe().c_str());
2896 14 : TRY_CATCH_RETURN(SetCommExecuteConfig(inCommExecuteConfig)); // 设置通信域粒度加速模式，ccu模式需打开ccu驱动
2897 14 : SetOpExecuteConfig(inCommExecuteConfig); // 算子粒度加速模式 同步为 通信域粒度加速模式
2898 42 : HCCL_DEBUG("[CommunicatorImpl][%s] comm accelerator [%s], isCcuMsAvailable is [%d]", __func__, GetCommExecuteConfig().accState.Describe().c_str(), isCcuMsAvailable);
2899 14 : return HCCL_SUCCESS;
2900 : }
2901 :
2902 0 : bool CommunicatorImpl::IsCommWithPCIEProtocol() const
2903 : {
2904 0 : auto links = GetFullMeshLinks();
2905 0 : for (auto link : links) {
2906 0 : if (link.GetLinkProtocol() == LinkProtocol::PCIE) {
2907 0 : HCCL_INFO("[CommunicatorImpl][%s]the current communicator has PCIE link", __func__);
2908 0 : return true;
2909 : }
2910 : }
2911 0 : HCCL_INFO("[CommunicatorImpl][%s]the current communicator does not have a PCIE link", __func__);
2912 0 : return false;
2913 0 : }
2914 :
2915 0 : HcclResult CommunicatorImpl::GetAccelerator(int32_t *accelerator) const
2916 : {
2917 0 : HcclAccelerator hcclAccelerator{HcclAccelerator::DEFAULT};
2918 0 : auto commAccelerator = GetCommExecuteConfig().accState;
2919 0 : std::string acceleraToStr = AcceleratorStateToString.at(commAccelerator);
2920 0 : HCCL_INFO("[CommunicatorImpl][%s] commId[%s], commAccelerator[%s]", __func__, GetId().c_str(),
2921 : acceleraToStr.c_str());
2922 :
2923 0 : switch (commAccelerator) {
2924 0 : case AcceleratorState::CCU_MS:
2925 0 : hcclAccelerator = HcclAccelerator::CCU_MS;
2926 0 : break;
2927 0 : case AcceleratorState::CCU_SCHED:
2928 0 : hcclAccelerator = HcclAccelerator::CCU_SCHED;
2929 0 : break;
2930 0 : case AcceleratorState::AIV:
2931 0 : hcclAccelerator = HcclAccelerator::AIV;
2932 0 : break;
2933 0 : case AcceleratorState::AIV_ONLY:
2934 0 : hcclAccelerator = HcclAccelerator::AIV_ONLY;
2935 0 : break;
2936 0 : case AcceleratorState::AICPU_TS:
2937 0 : hcclAccelerator = HcclAccelerator::AICPU_TS;
2938 0 : break;
2939 0 : case AcceleratorState::HOSTCPU_TS:
2940 0 : hcclAccelerator = HcclAccelerator::HOSTCPU_TS;
2941 0 : break;
2942 0 : case AcceleratorState::AICPU:
2943 0 : hcclAccelerator = HcclAccelerator::AICPU;
2944 0 : break;
2945 0 : default:
2946 0 : HCCL_ERROR("[GetAccelerator] commAccelerator[%s] internal error", acceleraToStr.c_str());
2947 0 : return HCCL_E_INTERNAL;
2948 : }
2949 0 : *accelerator = static_cast<int32_t>(hcclAccelerator);
2950 0 : return HCCL_SUCCESS;
2951 0 : }
2952 :
2953 176 : bool CommunicatorImpl::IsOpUsingCcuMs() const
2954 : {
2955 176 : return GetOpExecuteConfig().accState == AcceleratorState::CCU_MS;
2956 : }
2957 :
2958 163 : bool CommunicatorImpl::IsOpUsingCcuSched() const
2959 : {
2960 163 : return GetOpExecuteConfig().accState == AcceleratorState::CCU_SCHED;
2961 : }
2962 :
2963 4 : bool CommunicatorImpl::IsCommUsingCcuMs() const
2964 : {
2965 4 : return GetCommExecuteConfig().accState == AcceleratorState::CCU_MS;
2966 : }
2967 :
2968 4 : bool CommunicatorImpl::IsCommUsingCcuSched() const
2969 : {
2970 4 : return GetCommExecuteConfig().accState == AcceleratorState::CCU_SCHED;
2971 : }
2972 :
2973 4 : HcclResult CommunicatorImpl::RecoverExeCfgData(const OpExecuteConfig &inOpExeCfg, const OpExecuteConfig &inCommExeCfg, bool inIsLoadOp)
2974 : {
2975 : // mc2目前没有快照恢复，如果增加需要调用该接口
2976 12 : HCCL_INFO("CommunicatorImpl[%s] Recover ExecuteConfig, opAcceState is %s, commAcceState is %s, isLoadOp is %d", __func__,
2977 : inOpExeCfg.accState.Describe().c_str(), inCommExeCfg.accState.Describe().c_str(), inIsLoadOp);
2978 :
2979 : // 恢复加速器类型
2980 4 : SetOpExecuteConfig(inOpExeCfg); // 算子粒度 和 通信域粒度 都考虑
2981 4 : SetCommExecuteConfig(inCommExeCfg);
2982 4 : isLoadOp = inIsLoadOp;
2983 :
2984 12 : HCCL_INFO("Recover OpExecuteConfig data from snapshot success.");
2985 4 : return HcclResult::HCCL_SUCCESS;
2986 : }
2987 :
2988 188 : void CommunicatorImpl::RegisterAcceStateCallBack(std::function<HcclResult(const std::string &commId, bool isUsingCcuMs, bool isUsingCcuSched)> inCallback)
2989 : {
2990 188 : callback = inCallback;
2991 188 : }
2992 :
2993 97 : void CommunicatorImpl::SetOpExecuteConfig(const OpExecuteConfig &inConfig)
2994 : {
2995 97 : opExecuteConfig = inConfig;
2996 291 : HCCL_DEBUG(
2997 : "[CommunicatorImpl][%s] comm id [%s], IsOpUsingCcuMs [%d], IsOpUsingCcuSched [%d]",
2998 : __func__, GetId().c_str(), IsOpUsingCcuMs(), IsOpUsingCcuSched()); // 算子粒度
2999 97 : }
3000 :
3001 1 : void CommunicatorImpl::SetCommExecuteConfig(const OpExecuteConfig& inConfig)
3002 : {
3003 1 : commExecuteConfig = inConfig;
3004 3 : HCCL_DEBUG(
3005 : "[CommunicatorImpl][%s] update comm manager ccu status, comm id [%s], IsCommUsingCcuMs [%d], IsCommUsingCcuSched [%d]",
3006 : __func__, GetId().c_str(), IsCommUsingCcuMs(), IsCommUsingCcuSched()); // 通信域粒度
3007 :
3008 1 : TryInitCcuFeature(); // 单例结构整改前临时方案
3009 :
3010 1 : callback(GetId(), IsCommUsingCcuMs(), IsCommUsingCcuSched());
3011 1 : }
3012 :
3013 2 : HcclResult CommunicatorImpl::CalcTaskNum(OpType opType, DataType dataType, u64 count, u32 &taskNum) const
3014 : {
3015 6 : HCCL_INFO("[CommunicatorImpl][CalcTaskNum] start!");
3016 2 : return collAlgComponent->CalcTaskNum(opType, dataType, count, taskNum);
3017 : }
3018 :
3019 194 : void CommunicatorImpl::InitUbMemoryTransportMgr()
3020 : {
3021 194 : ubMemoryTransportMgr = std::make_unique<UbMemoryTransportMgr>(*this);
3022 194 : }
3023 :
3024 11 : UbMemoryTransportMgr *CommunicatorImpl::GetUbMemoryTransportMgr() const
3025 : {
3026 11 : return ubMemoryTransportMgr.get();
3027 : }
3028 :
3029 58 : HcclResult CommunicatorImpl::HcomSelectAlg(const CollOpParams& opParams, int32_t aivCoreLimit, bool &ifAiv, std::string &algName)
3030 : {
3031 174 : HCCL_INFO("CommunicatorImpl::HcomSelectAlg opType[%s], count[%llu], dataType[%s], HcclReduceOp[%s], aivCoreLimit[%d]",
3032 : opParams.opType.Describe().c_str(), opParams.count, opParams.dataType.Describe().c_str(), opParams.reduceOp.Describe().c_str(), aivCoreLimit);
3033 :
3034 58 : if (GetCommStatus() == CommStatus::COMM_ERROR) {
3035 3 : HCCL_ERROR("Comm has been error, can not select alg now!");
3036 1 : return HcclResult::HCCL_E_INTERNAL;
3037 : }
3038 :
3039 57 : if (isSuspended) {
3040 3 : HCCL_ERROR("Comm has been suspended, can not select alg now!");
3041 1 : return HcclResult::HCCL_E_SUSPENDING;
3042 : }
3043 : // 等待通信域状态为Ready，执行算子下发
3044 56 : WaitReady();
3045 :
3046 56 : std::string tag = ""; // 算法选择不需要传入tag，获取kernel arg的时候会用到
3047 56 : CovertToCurrentCollOperator(tag, opParams, OpMode::OFFLOAD, true, true);
3048 : // 图模式算子加载选择CollService
3049 56 : opExecuteConfig = commExecuteConfig;
3050 56 : ExecAlgSelect(opParams, OpMode::OFFLOAD);
3051 56 : ifAiv = (opExecuteConfig.accState == AcceleratorState::AIV || opExecuteConfig.accState == AcceleratorState::AIV_ONLY);
3052 56 : HcclResult dataTypeChkRes = OpParamsChecker::CheckOpDataTypeOffload(opParams, GetOpCcuFeatureFlag(),
3053 56 : GetOpAiCpuTSFeatureFlag(), ifAiv);
3054 56 : if (dataTypeChkRes != HcclResult::HCCL_SUCCESS) {
3055 0 : HCCL_ERROR("[CommunicatorImpl::HcomSelectAlg] DataType check fail.");
3056 0 : SetCommStatus(CommStatus::COMM_READY);
3057 0 : return dataTypeChkRes;
3058 : }
3059 56 : algName = curAlgName;
3060 :
3061 56 : return HcclResult::HCCL_SUCCESS;
3062 56 : }
3063 :
3064 2 : void CommunicatorImpl::ReportHcclMC2Info(const Stream &kfcStream, Stream &stream, const std::vector<Stream*> &aicpuStreams)
3065 : {
3066 2 : HcclResult hcclRet = InitProfilingReporter();
3067 2 : if (hcclRet != HCCL_SUCCESS) {
3068 0 : HCCL_ERROR("[ReportHcclMC2Info]InitProfilingReporter failed, ret[%d]", static_cast<s32>(hcclRet));
3069 0 : return;
3070 : }
3071 2 : profilingReporter->CallReportMc2CommInfo(kfcStream, stream, aicpuStreams, id, myRank, rankSize, rankInParentComm);
3072 : }
3073 :
3074 4 : void CommunicatorImpl::OpAcceleratorStateFallback()
3075 : {
3076 4 : OpExecuteConfig inOpExecuteConfig;
3077 : // 只要ccu出问题，直接回退到CCU_FALLBACK，走AICPU
3078 4 : switch (opExecuteConfig.accState) {
3079 2 : case AcceleratorState::CCU_MS:
3080 2 : inOpExecuteConfig.accState = AcceleratorState::CCU_FALLBACK;
3081 2 : break;
3082 1 : case AcceleratorState::CCU_SCHED:
3083 1 : inOpExecuteConfig.accState = AcceleratorState::CCU_FALLBACK;
3084 1 : break;
3085 1 : default:
3086 1 : THROW<NotSupportException>(
3087 3 : StringFormat("[CommunicatorImpl::%s] Only supports CCU accelerator rollback", __func__));
3088 : break;
3089 : }
3090 3 : SetOpExecuteConfig(inOpExecuteConfig);
3091 3 : }
3092 :
3093 2 : HcclResult CommunicatorImpl::AcceleratorFallback()
3094 : {
3095 6 : HCCL_RUN_INFO("[CommunicatorImpl][%s] opMode[%s]", __func__, currentCollOperator->opMode.Describe().c_str());
3096 2 : string needFallBackAlgName = curAlgName;
3097 2 : OpAcceleratorStateFallback();
3098 :
3099 2 : HcclResult ret = HCCL_SUCCESS;
3100 2 : switch (currentCollOperator->opMode) {
3101 1 : case OpMode::OPBASE:
3102 1 : ret = ReLoadOpbasedOp();
3103 0 : break;
3104 1 : case OpMode::OFFLOAD:
3105 1 : ret = ReLoadOffloadOp();
3106 0 : break;
3107 0 : default:
3108 0 : THROW<InternalException>(
3109 0 : StringFormat("[CommunicatorImpl::%s] OpMode error, accelerator rollback failed", __func__));
3110 : break;
3111 : }
3112 :
3113 : // 缓存当前算子的加速模式；
3114 : // 下一个算子下发时，做完算法选择后，查找上述加速模式缓存，
3115 : // 若能命中，按照上述已缓存的加速模式下发算子(大概率也是资源不足，走回退)；
3116 : // 否则，按照算法选择的加速模式下发算子。
3117 0 : opAcceStateCache.insert({{curOpParams.opType, needFallBackAlgName}, {opExecuteConfig.accState, curAlgName}});
3118 0 : HCCL_INFO("[CommunicatorImpl][%s] opAcceStateCache opType[%s], needFallBackAlgName[%s], accelerator[%s], curAlgName[%s]", __func__,
3119 : curOpParams.opType.Describe().c_str(), needFallBackAlgName.c_str(), opExecuteConfig.accState.Describe().c_str(), curAlgName.c_str());
3120 :
3121 0 : HCCL_INFO("[CommunicatorImpl][%s] end", __func__);
3122 0 : return ret;
3123 2 : }
3124 :
3125 0 : HcclResult CommunicatorImpl::GetCacheMap(AivOpCacheArgs& opCacheParam , std::shared_ptr<InsQueue>& tempInsQue)
3126 : {
3127 0 : if (hcclCacheMap_.size() > CACHEMAP_MAXSIZE) {
3128 0 : size_t clearCount = static_cast<size_t>(CACHEMAP_MAXSIZE * CACHEMAP_CLEARPERCENT);
3129 0 : for (auto it = hcclCacheMap_.begin(); clearCount > 0 && it != hcclCacheMap_.end(); clearCount--) {
3130 0 : it = hcclCacheMap_.erase(it);
3131 : }
3132 : }
3133 0 : hcclCacheMap_.emplace(std::make_pair(opCacheParam, std::move(tempInsQue)));
3134 0 : HCCL_INFO("[CommunicatorImpl][GetCacheMap]");
3135 0 : return HCCL_SUCCESS;
3136 : }
3137 :
3138 6 : HcclResult CommunicatorImpl::ReLoadOpbasedOp()
3139 : {
3140 18 : HCCL_DEBUG("[CommunicatorImpl][%s] status is [%s], isSuspended is [%d]", __func__, GetCommStatus().Describe().c_str(),
3141 : isSuspended);
3142 6 : ExecAlgSelect(curOpParams, OpMode::OPBASE); // 根据配置选择对应的collService
3143 5 : if (dynamic_cast<CollServiceDefaultImpl *>(collService) != nullptr) {
3144 3 : HCCL_ERROR("ReLoadOpbasedOp is not supported in CollServiceDefaultImpl.");
3145 1 : return HcclResult::HCCL_E_NOT_SUPPORT;
3146 : }
3147 4 : bool isAiv = (opExecuteConfig.accState == AcceleratorState::AIV || opExecuteConfig.accState == AcceleratorState::AIV_ONLY);
3148 4 : HcclResult dataTypeChkRes = OpParamsChecker::CheckOpDataTypeOpbase(curOpParams, GetOpCcuFeatureFlag(),
3149 4 : GetOpAiCpuTSFeatureFlag(), isAiv); // 算子粒度
3150 4 : if (dataTypeChkRes != HcclResult::HCCL_SUCCESS) {
3151 3 : HCCL_ERROR("[CommunicatorImpl::ReLoadOpbasedOp] DataType check fail.");
3152 1 : SetCommStatus(CommStatus::COMM_READY);
3153 1 : return dataTypeChkRes;
3154 : }
3155 :
3156 3 : if (currentCollOperator == nullptr) {
3157 3 : HCCL_ERROR("CurrentCollOperator not initialized.");
3158 1 : return HcclResult::HCCL_E_PTR;
3159 : }
3160 2 : collService->ReLoadWithOpBasedMode(*currentCollOperator);
3161 1 : return HcclResult::HCCL_SUCCESS;
3162 : }
3163 :
3164 6 : HcclResult CommunicatorImpl::ReLoadOffloadOp()
3165 : {
3166 18 : HCCL_DEBUG("[CommunicatorImpl][%s] status is [%s], isSuspended is [%d]", __func__, GetCommStatus().Describe().c_str(),
3167 : isSuspended);
3168 :
3169 6 : ExecAlgSelect(curOpParams, OpMode::OFFLOAD); // 根据配置选择对应的collService
3170 :
3171 5 : if (opExecuteConfig.accState == AcceleratorState::HOSTCPU_TS) { // 950不支持HOST_TS模式
3172 3 : HCCL_ERROR("[CommunicatorImpl::ReLoadOffloadOp] HOSTCPU_TS is not support.");
3173 1 : return HcclResult::HCCL_E_NOT_SUPPORT;
3174 : }
3175 4 : bool isAiv = (opExecuteConfig.accState == AcceleratorState::AIV || opExecuteConfig.accState == AcceleratorState::AIV_ONLY);
3176 4 : HcclResult dataTypeChkRes = OpParamsChecker::CheckOpDataTypeOffload(curOpParams, GetOpCcuFeatureFlag(),
3177 4 : GetOpAiCpuTSFeatureFlag(), isAiv); // 算子粒度
3178 4 : if (dataTypeChkRes != HcclResult::HCCL_SUCCESS) {
3179 3 : HCCL_ERROR("[CommunicatorImpl::ReLoadOffloadCollOp] DataType check fail.");
3180 1 : SetCommStatus(CommStatus::COMM_READY);
3181 1 : return dataTypeChkRes;
3182 : }
3183 :
3184 3 : if (currentCollOperator == nullptr) {
3185 3 : HCCL_ERROR("CurrentCollOperator not initialized.");
3186 1 : return HcclResult::HCCL_E_PTR;
3187 : }
3188 2 : collService->ReLoadWithOffloadMode(*currentCollOperator);
3189 1 : return HcclResult::HCCL_SUCCESS;
3190 : }
3191 :
3192 : template<typename BufferType>
3193 0 : std::shared_ptr<BufferType> CommunicatorImpl::BarrierAllocBuffer(std::size_t size)
3194 : {
3195 0 : return std::make_shared<BufferType>(size);
3196 : }
3197 :
3198 0 : HcclResult CommunicatorImpl::CreateBarrierMemory(void *&sendBuf, void *&recvBuf, uint64_t count)
3199 : {
3200 0 : HCCL_INFO("CreateBarrierMemory start.");
3201 0 : if (isFirstBarrier) {
3202 0 : barrierInMemory = BarrierAllocBuffer<DevBuffer>(count * sizeof(float));
3203 0 : barrierOutMemory = BarrierAllocBuffer<DevBuffer>(count * sizeof(float));
3204 : // 申请host侧内存，并将初始值设置为0
3205 0 : std::shared_ptr<HostBuffer> barrierHostMem = BarrierAllocBuffer<HostBuffer>(count * sizeof(float));
3206 0 : s32 sRet = memset_s(reinterpret_cast<void *>(barrierHostMem->GetAddr()), barrierHostMem->GetSize(), 0,
3207 : count * sizeof(float));
3208 0 : if (sRet != EOK) {
3209 0 : barrierInMemory.reset();
3210 0 : barrierOutMemory.reset();
3211 0 : barrierHostMem.reset();
3212 0 : HCCL_ERROR("[CreateBarrierMemory] mem set failed.errorno[%d]", sRet);
3213 0 : return HCCL_E_MEMORY;
3214 : }
3215 : // H2D拷贝
3216 0 : HrtMemcpy(reinterpret_cast<void *>(barrierInMemory->GetAddr()), barrierInMemory->GetSize(), reinterpret_cast<void *>(barrierHostMem->GetAddr()),
3217 0 : barrierHostMem->GetSize(), RT_MEMCPY_HOST_TO_DEVICE);
3218 0 : HrtMemcpy(reinterpret_cast<void *>(barrierOutMemory->GetAddr()), barrierOutMemory->GetSize(), reinterpret_cast<void *>(barrierHostMem->GetAddr()),
3219 0 : barrierHostMem->GetSize(), RT_MEMCPY_HOST_TO_DEVICE);
3220 0 : isFirstBarrier = false;
3221 0 : }
3222 : // 将内存指针赋值给传入参数
3223 0 : sendBuf = reinterpret_cast<void *>(barrierInMemory->GetAddr());
3224 0 : if (sendBuf == nullptr) {
3225 0 : HCCL_ERROR("[CreateBarrierMemory] Failed to get barrierInMemory.");
3226 0 : return HCCL_E_PTR;
3227 : }
3228 0 : recvBuf = reinterpret_cast<void *>(barrierOutMemory->GetAddr());
3229 0 : if (recvBuf == nullptr) {
3230 0 : HCCL_ERROR("[CreateBarrierMemory] Failed to get barrierOutMemory.");
3231 0 : return HCCL_E_PTR;
3232 : }
3233 0 : HCCL_INFO("CreateBarrierMemory success.");
3234 0 : return HCCL_SUCCESS;
3235 : }
3236 :
3237 0 : HcclResult CommunicatorImpl::CreateWorkspaceBuf(const char *memTag, uint64_t *size, bool *newCreated)
3238 : {
3239 0 : std::string tag = memTag != nullptr ? std::string(memTag) : "";
3240 : // empty tag is global workspace
3241 0 : if (tagWorkspaceMap_.find(tag) == tagWorkspaceMap_.end()) {
3242 0 : shared_ptr<DevBuffer> workspace = std::make_shared<DevBuffer>(*size);
3243 0 : tagWorkspaceMap_.insert(make_pair(tag, workspace));
3244 0 : HCCL_INFO("Create tagMem[%s] WorkspaceBuf success, WorkspaceBuf = %p", tag.c_str(), workspace.get());
3245 0 : if (newCreated != nullptr) {
3246 0 : *newCreated = true;
3247 : }
3248 0 : }
3249 0 : return HcclResult::HCCL_SUCCESS;
3250 0 : }
3251 :
3252 : // dpu相关
3253 39 : bool CommunicatorImpl::IsNeedDpu()
3254 : {
3255 39 : if (rankGraph == nullptr) {
3256 0 : return false;
3257 : }
3258 39 : if (rankGraph->GetPeer(myRank) == nullptr) {
3259 0 : HCCL_ERROR("[GetHostIpFromRankGraph] rankGraph peer is null!");
3260 0 : return false;
3261 : }
3262 : // 根据rankgraph直接找peer对应的ConnInterface列表
3263 39 : std::vector<std::shared_ptr<NetInstance::ConnInterface>> interfaces = rankGraph->GetPeer(myRank)->GetIfaces();
3264 54 : for (auto interface : interfaces) {
3265 20 : if (interface->GetPos() == AddrPosition::HOST) {
3266 15 : HCCL_INFO("[CommunicatorImpl][IsNeedDpu] need host dpu");
3267 5 : return true;
3268 : }
3269 20 : }
3270 34 : return false;
3271 39 : }
3272 :
3273 5 : void CommunicatorImpl::InitHccpPeer() const
3274 : {
3275 5 : RaSocketSetWhiteListStatus(1); // PEER模式需要手动开启白名单模式
3276 5 : HccpPeerManager::GetInstance().Init(devLogicId);
3277 5 : }
3278 :
3279 0 : HcclResult CommunicatorImpl::PrepareDpuKernelResource(aclrtFuncHandle &funcHandle)
3280 : {
3281 : // 获取二进制文件路径
3282 0 : std::string jsonPath;
3283 0 : std::string getPath = getenv("ASCEND_HOME_PATH");
3284 0 : if (!getPath.empty()) {
3285 0 : jsonPath = getPath;
3286 : } else {
3287 0 : jsonPath = "/usr/local/Ascend/cann/";
3288 0 : HCCL_WARNING("[CommunicatorImpl::%s] ENV:ASCEND_HOME_PATH is not set", __func__);
3289 : }
3290 :
3291 0 : jsonPath += "/opp/built-in/op_impl/dpu/";
3292 0 : HCCL_DEBUG("[CommunicatorImpl::%s] kernel folder path[%s]", __func__, jsonPath.c_str());
3293 :
3294 : // cpuKernelMode为1时，json命名需与so命名保持一致， 即libccl_dpu.json与libccl_dpu.so
3295 0 : jsonPath += "libccl_dpu.json";
3296 0 : char realPath[PATH_MAX] = {0};
3297 0 : CHK_PRT_RET(realpath(jsonPath.c_str(), realPath) == nullptr,
3298 : HCCL_ERROR("[CommunicatorImpl::%s]: %s is not a valid real path, err[%d]", __func__, jsonPath.c_str(), errno),
3299 : HCCL_E_INTERNAL);
3300 0 : HCCL_INFO("[CommunicatorImpl::%s] realPath: %s", __func__, realPath);
3301 :
3302 : aclrtBinHandle binHandle;
3303 : aclrtBinaryLoadOptions options;
3304 : aclrtBinaryLoadOption option;
3305 0 : option.type = ACL_RT_BINARY_LOAD_OPT_CPU_KERNEL_MODE; // AI CPU算子注册模式 ????
3306 0 : option.value.cpuKernelMode = 1; // 0 ：仅需要加载json，1 ：加载cpu so & json，2: LoadFromData
3307 0 : options.numOpt = 1;
3308 0 : options.options = &option;
3309 0 : if (aclrtBinaryLoadFromFile(realPath, &options, &binHandle) != ACL_SUCCESS) {
3310 0 : HCCL_ERROR("[CommunicatorImpl::%s] load binary from file error.", __func__);
3311 0 : return HCCL_E_OPEN_FILE_FAILURE;
3312 : }
3313 :
3314 : // 创建dpustream
3315 0 : if (aclrtCreateStreamWithConfig(&dpuStream, 0, ACL_STREAM_FAST_LAUNCH) != ACL_SUCCESS) {
3316 0 : HCCL_ERROR("[CommunicatorImpl::%s] Create Local Stream Failed", __func__);
3317 0 : return HCCL_E_INTERNAL;
3318 : }
3319 :
3320 : // 查找核函数
3321 0 : if (aclrtBinaryGetFunction(binHandle, "RunDpuRpcSrvLaunch", &funcHandle) != ACL_SUCCESS) {
3322 0 : HCCL_ERROR("[CommunicatorImpl::%s] Get Function Failed", __func__);
3323 0 : return HCCL_E_INTERNAL;
3324 : }
3325 :
3326 0 : return HCCL_SUCCESS;
3327 0 : }
3328 :
3329 2 : HcclResult CommunicatorImpl::LaunchDpuKernel(aclrtFuncHandle &funcHandle)
3330 : {
3331 : // 下发
3332 6 : HCCL_INFO("[CommunicatorImpl::%s] Launch Dpu Kernel", __func__);
3333 : aclrtLaunchKernelCfg cfg;
3334 : aclrtLaunchKernelAttr kernelAttr;
3335 2 : kernelAttr.id = ACL_RT_LAUNCH_KERNEL_ATTR_TIMEOUT;
3336 2 : kernelAttr.value.timeout = NOTIFY_DEFAULT_WAIT_TIME > std::numeric_limits<uint16_t>::max() ?
3337 0 : std::numeric_limits<uint16_t>::max() : NOTIFY_DEFAULT_WAIT_TIME;
3338 2 : cfg.numAttrs = 1;
3339 2 : cfg.attrs = &kernelAttr;
3340 2 : constexpr u32 numBlocks = 1;
3341 2 : hostArgs.commId = id;
3342 2 : hostArgs.memorySize = SHARE_HBM_MEMORY_SIZE;
3343 2 : hostArgs.hostMem = hostShareBuf;
3344 8 : hostArgs.shareHBM = tagDpuShmemArgsMap_[DPUTAG].connectType_ == HOST_DEVICE_CONNECT_TYPE_UB ? tagDpuShmemArgsMap_[DPUTAG].va_ : tagDpuShmemArgsMap_[DPUTAG].accessVA_;
3345 2 : hostArgs.deviceId = devLogicId;
3346 8 : hostArgs.taskexceptionVa = tagDpuShmemArgsMap_[DPUTASKEXCEPTION].connectType_ == HOST_DEVICE_CONNECT_TYPE_UB ? tagDpuShmemArgsMap_[DPUTASKEXCEPTION].va_ : tagDpuShmemArgsMap_[DPUTASKEXCEPTION].accessVA_;
3347 6 : HCCL_INFO("[CommunicatorImpl::%s] DpuKernelLaunchParam{commId:%s; memorySize:%u; shareHBM:%p; taskexceptionVa:%p}",
3348 : __func__, hostArgs.commId.c_str(), hostArgs.memorySize, hostArgs.shareHBM, hostArgs.taskexceptionVa);
3349 2 : CHK_RET(SaveDpuStreamId());
3350 2 : size_t argsSize = sizeof(hostArgs);
3351 : aclrtPlaceHolderInfo placeHolderArrays;
3352 2 : size_t placeHolderNum = 0;
3353 2 : if (aclrtLaunchKernelWithHostArgs(funcHandle, numBlocks, dpuStream, &cfg, &hostArgs, argsSize,
3354 : &placeHolderArrays, placeHolderNum)
3355 2 : != ACL_SUCCESS) {
3356 0 : HCCL_ERROR("[CommunicatorImpl::%s] Launch Dpu Kernel Failed", __func__);
3357 0 : return HCCL_E_INTERNAL;
3358 : }
3359 2 : return HCCL_SUCCESS;
3360 : }
3361 :
3362 3 : HcclResult CommunicatorImpl::InitAndLaunchDpuKernel()
3363 : {
3364 9 : HCCL_INFO("[CommunicatorImpl::%s] Start to Launch Dpu Kernel", __func__);
3365 : // 申请共享内存(需要在npu ctx 下进行)
3366 9 : for (auto &tmpShmem : tagDpuShmemArgsMap_) {
3367 6 : bool newCreate = false;
3368 6 : uint64_t memSize = tmpShmem.first == DPUTAG ? static_cast<uint64_t>(SHARE_HBM_MEMORY_SIZE) : static_cast<uint64_t>(DPU_TASKEXCEPTION_MEMORY_SIZE);
3369 6 : HcclResult memRet = GetKFCWorkSpaceVA(tmpShmem.first, &memSize, &tmpShmem.second.accessVA_, &newCreate);
3370 6 : if (memRet != HCCL_SUCCESS) {
3371 0 : HCCL_ERROR("[CommunicatorImpl::InitCommResource] Alloc Share HBM Failed");
3372 0 : return HCCL_E_RUNTIME;
3373 : }
3374 : }
3375 : // 设置XPU
3376 9 : HCCL_INFO("[CommunicatorImpl::%s] Switch to Dpu Ctx", __func__);
3377 3 : if (aclrtGetCurrentContext(&npuContext) != ACL_SUCCESS) {
3378 0 : HCCL_ERROR("[CommunicatorImpl::%s] Get Npu Ctx Failed", __func__);
3379 0 : return HCCL_E_INTERNAL;
3380 : }
3381 3 : if (HrtSetXpuDevice(TEMP_DEV_TYPE_DPU, 0) != HCCL_SUCCESS) {
3382 0 : HCCL_ERROR("[CommunicatorImpl::%s] Switch to Dpu Ctx Failed", __func__);
3383 0 : return HCCL_E_INTERNAL;
3384 : }
3385 3 : if (aclrtGetCurrentContext(&dpuContext) != ACL_SUCCESS) {
3386 0 : HCCL_ERROR("[CommunicatorImpl::%s] Get Dpu Ctx Failed", __func__);
3387 0 : return HCCL_E_INTERNAL;
3388 : }
3389 :
3390 : // 准备资源
3391 : aclrtFuncHandle funcHandle;
3392 3 : CHK_RET(PrepareDpuKernelResource(funcHandle));
3393 :
3394 3 : hostShareBuf = malloc(SHARE_HBM_MEMORY_SIZE);
3395 3 : CHK_PTR_NULL(hostShareBuf);
3396 :
3397 : // 下发
3398 3 : HcclResult ret = LaunchDpuKernel(funcHandle);
3399 3 : if (ret != HCCL_SUCCESS) {
3400 3 : HCCL_ERROR("[CommunicatorImpl::%s] Launch Dpu Kernel Failed", __func__);
3401 1 : free(hostShareBuf);
3402 1 : hostShareBuf = nullptr;
3403 1 : return ret;
3404 : }
3405 :
3406 : // 切换回当前Ctx
3407 6 : HCCL_INFO("[CommunicatorImpl::%s] Switch to Npu Ctx", __func__);
3408 2 : if (ACL_SUCCESS != aclrtSetCurrentContext(npuContext)) {
3409 0 : HCCL_ERROR("[CommunicatorImpl::%s] Reset Current Ctx Failed", __func__);
3410 0 : free(hostShareBuf);
3411 0 : hostShareBuf = nullptr;
3412 0 : return HCCL_E_INTERNAL;
3413 : }
3414 :
3415 6 : HCCL_INFO("[CommunicatorImpl::%s] Launch Dpu Kernel End", __func__);
3416 2 : isDpuKernelLaunched = true;
3417 2 : g_commNum++;
3418 2 : return HCCL_SUCCESS;
3419 : }
3420 :
3421 4 : HcclResult CommunicatorImpl::InitAndLaunchAicpuKernel()
3422 : {
3423 12 : HCCL_INFO("[CommunicatorImpl::%s] Start to Launch Aicpu Kernel", __func__);
3424 : // context切换
3425 :
3426 : // 准备资源
3427 4 : std::string kernelName = "HcclDpuTaskexpShmemRestore";
3428 4 : aclrtFuncHandle funcHandle = GetAicpuKernelFuncHandle(kernelName.c_str());
3429 4 : constexpr u32 numBlocks = 1;
3430 : aclrtStream tempAicpuStream; // 创建局部流
3431 4 : if (aclrtCreateStreamWithConfig(&tempAicpuStream, 0, ACL_STREAM_FAST_LAUNCH) != ACL_SUCCESS) { // 后两个入参？
3432 3 : HCCL_ERROR("[CommunicatorImpl::%s] Create Local Stream Failed", __func__);
3433 1 : return HCCL_E_INTERNAL;
3434 : }
3435 : aclrtLaunchKernelCfg cfg;
3436 : aclrtLaunchKernelAttr kernelAttr;
3437 3 : kernelAttr.id = ACL_RT_LAUNCH_KERNEL_ATTR_TIMEOUT;
3438 3 : kernelAttr.value.timeout = NOTIFY_DEFAULT_WAIT_TIME > std::numeric_limits<uint16_t>::max() ?
3439 0 : std::numeric_limits<uint16_t>::max() : NOTIFY_DEFAULT_WAIT_TIME;
3440 3 : cfg.numAttrs = 1;
3441 3 : cfg.attrs = &kernelAttr;
3442 : struct AicpuKernelLaunchParam {
3443 : char commId[COMM_NAME_MAX_LENGTH];
3444 : void *taskexceptionVa;
3445 : u64 memorySize;
3446 : uint32_t deviceId;
3447 : };
3448 : AicpuKernelLaunchParam hostArgsTmp;
3449 3 : auto sRet = strncpy_s(hostArgsTmp.commId, COMM_NAME_MAX_LENGTH, id.c_str(), COMM_NAME_MAX_LENGTH - 1);
3450 3 : if (sRet != 0) {
3451 0 : HCCL_ERROR("[CommunicatorImpl::InitAndLaunchAicpuKernel] commId strncpy_s failed, ret[%d]", sRet);
3452 0 : return HCCL_E_INTERNAL;
3453 : }
3454 6 : hostArgsTmp.taskexceptionVa = tagDpuShmemArgsMap_[DPUTASKEXCEPTION].accessVA_;
3455 3 : hostArgsTmp.memorySize = DPU_TASKEXCEPTION_MEMORY_SIZE;
3456 3 : hostArgsTmp.deviceId = devLogicId;
3457 3 : size_t argsSize = sizeof(hostArgsTmp);
3458 : aclrtPlaceHolderInfo placeHolderArrays;
3459 3 : size_t placeHolderNum = 0;
3460 : // 下发
3461 3 : rtError_t ret = aclrtLaunchKernelWithHostArgs(funcHandle, numBlocks, tempAicpuStream, &cfg, &hostArgsTmp, argsSize,
3462 : &placeHolderArrays, placeHolderNum);
3463 3 : if (ret != RT_ERROR_NONE) {
3464 3 : HCCL_ERROR("[CommunicatorImpl::%s] Launch Aicpu Kernel Failed", __func__);
3465 1 : return HCCL_E_INTERNAL;
3466 : }
3467 2 : HcclStreamSynchronize(tempAicpuStream); // 同步
3468 2 : aclError aclRet = aclrtDestroyStreamForce(tempAicpuStream); // 销毁局部流
3469 2 : if (ACL_SUCCESS != aclRet) {
3470 3 : HCCL_ERROR("Destroy Stream Failed, aclReturn[%d]", aclRet);
3471 1 : return HCCL_E_RUNTIME;
3472 : }
3473 3 : HCCL_INFO("[CommunicatorImpl::%s] Launch Aicpu Kernel End", __func__);
3474 1 : return HCCL_SUCCESS;
3475 4 : }
3476 :
3477 40 : void CommunicatorImpl::AppendLocalDieIdForLinks()
3478 : {
3479 40 : if (rankSize == 1) {
3480 3 : HCCL_INFO("[AppendLocalDieIdForLinks] rankSize = 1, No RankGraph exists");
3481 1 : return;
3482 : }
3483 :
3484 39 : auto srcRankNode = rankGraph->GetPeer(myRank)->GetNodeId();
3485 :
3486 140 : auto processLinks = [&](const std::vector<std::shared_ptr<NetInstance::Link>>& links, bool isSource) {
3487 190 : for (auto link : links) {
3488 50 : auto iface = isSource ? link->GetSourceIface() : link->GetTargetIface();
3489 50 : if (iface->GetPos() == AddrPosition::HOST || *(iface->GetLinkProtocols().begin()) == LinkProtocol::PCIE) {
3490 10 : continue;
3491 : }
3492 40 : u32 dieId = GetLocalDieId({myRank, *iface}, *(link->GetLinkProtocols().begin()));
3493 120 : HCCL_INFO("[CommunicatorImpl][AppendLocalDieIdForLinks] get link dieid[%u]", dieId);
3494 40 : iface->SetLocalDieId(dieId);
3495 60 : }
3496 140 : };
3497 :
3498 54 : for (auto level : rankGraph->GetLevels(myRank)) {
3499 15 : auto netInstance = rankGraph->GetNetInstanceByRankId(level, myRank);
3500 15 : auto& vGraph = netInstance->GetGraph();
3501 :
3502 : // Process fabric links
3503 25 : for (auto fabric : netInstance->GetFabrics()) {
3504 10 : auto dstRankNode = fabric->GetNodeId();
3505 10 : processLinks(vGraph.GetEdges(srcRankNode, dstRankNode), true);
3506 10 : processLinks(vGraph.GetEdges(dstRankNode, srcRankNode), false);
3507 10 : }
3508 :
3509 : // Process direct peer links
3510 75 : for (u32 dstRank = 0; dstRank < rankSize; ++dstRank) {
3511 60 : auto dstRankNode = rankGraph->GetPeer(dstRank)->GetNodeId();
3512 60 : processLinks(vGraph.GetEdges(srcRankNode, dstRankNode), true);
3513 60 : processLinks(vGraph.GetEdges(dstRankNode, srcRankNode), false);
3514 : }
3515 39 : }
3516 : }
3517 :
3518 1 : HcclResult CommunicatorImpl::GetLocalCclBuffer(void **addr, uint64_t *size)
3519 : {
3520 1 : CHK_PTR_NULL(inCclBuffer.get());
3521 1 : *addr = reinterpret_cast<void*>(inCclBuffer.get()->GetAddr());
3522 1 : *size = static_cast<uint64_t>(inCclBuffer.get()->GetSize());
3523 3 : HCCL_INFO("CommunicatorImpl::GetLocalCclBuffer success, addr[%p], size[%llu]", *addr, *size);
3524 1 : return HcclResult::HCCL_SUCCESS;
3525 : }
3526 :
3527 5 : HcclResult CommunicatorImpl::GetDevMemWorkSpace(const std::string &memTag, uint64_t *size, void **addr, bool *newCreated)
3528 : {
3529 5 : if (memTag == DPUTAG) {
3530 0 : return GetKFCWorkSpaceVA(memTag, size, addr, newCreated);
3531 : }
3532 5 : auto iter = tagWorkspaceMap_.find(memTag);
3533 5 : if (iter != tagWorkspaceMap_.end()) {
3534 2 : std::shared_ptr<DevBuffer> oldWorkspace = iter->second;
3535 2 : if (*size != static_cast<uint64_t>(oldWorkspace.get()->GetSize())) {
3536 3 : HCCL_ERROR("HcclCommunicator::GetDevMemWorkSpace, The size of oldWorkspace %p is non-consistent, target size compare now size: %llu->%llu", *addr, *size, oldWorkspace.get()->GetSize());
3537 1 : return HCCL_E_PARA;
3538 : }
3539 1 : *addr = reinterpret_cast<void *>(oldWorkspace.get()->GetAddr());
3540 1 : if (newCreated != nullptr) {
3541 1 : *newCreated = false;
3542 : }
3543 1 : return HcclResult::HCCL_SUCCESS;
3544 2 : }
3545 :
3546 3 : shared_ptr<DevBuffer> newWorkspace = std::make_shared<DevBuffer>(*size);
3547 3 : tagWorkspaceMap_.insert(make_pair(memTag, newWorkspace));
3548 9 : HCCL_INFO("Create tagMem[%s] WorkspaceBuf success, WorkspaceBuf: %p -> %p, size[%llu]", memTag.c_str(), newWorkspace.get(), newWorkspace.get()->GetAddr(), *size);
3549 3 : if (newCreated != nullptr) {
3550 2 : *newCreated = true;
3551 : }
3552 3 : *addr = reinterpret_cast<void *>(newWorkspace.get()->GetAddr());
3553 3 : return HcclResult::HCCL_SUCCESS;
3554 3 : }
3555 :
3556 4 : HcclResult CommunicatorImpl::AllocAndRegKFCWorkSpace(uint64_t size, const std::string &memTag) // 完成共享内存的申请和注册
3557 : {
3558 4 : auto it = tagDpuShmemArgsMap_.find(memTag);
3559 4 : if (it == tagDpuShmemArgsMap_.end()) {
3560 0 : HCCL_ERROR("memTag is invalid, memTag: %s", memTag.c_str());
3561 0 : return HCCL_E_PARA;
3562 : }
3563 :
3564 4 : if (it->second.accessVA_ != nullptr && it->second.va_ != nullptr) {
3565 0 : HCCL_WARNING("[CommunicatorImpl::%s] memTag[%s] accessVA_ is not nullptr, maybe already register, accessVA_: %p", __func__, memTag.c_str(), it->second.accessVA_);
3566 0 : CHK_RET(DestroyKFCWorkSpaceVA());
3567 : }
3568 4 : int32_t deviceLogicId = 0;
3569 4 : aclError aclRet = aclrtGetLogicDevIdByUserDevId(devLogicId, &deviceLogicId); // userDevId 转 logicDevId
3570 4 : if (aclRet != ACL_SUCCESS) {
3571 0 : HCCL_ERROR("[CommunicatorImpl::%s] aclrtGetLogicDevIdByUserDevId failed, devLogicId: %u, ret: %d", __func__, devLogicId, aclRet);
3572 0 : return HCCL_E_RUNTIME;
3573 : }
3574 4 : CHK_RET(HrtHalGetDeviceInfo(deviceLogicId, MODULE_TYPE_SYSTEM, INFO_TYPE_HD_CONNECT_TYPE, it->second.connectType_));
3575 12 : HCCL_INFO("[CommunicatorImpl::%s] connectType_: %ld", __func__, it->second.connectType_);
3576 :
3577 4 : HcclResult ret = HCCL_SUCCESS;
3578 4 : if (it->second.connectType_ == HOST_DEVICE_CONNECT_TYPE_PCIE) {
3579 2 : it->second.va_ = HrtMalloc(size, ACL_MEM_TYPE_HIGH_BAND_WIDTH);
3580 2 : ret = HrtHalHostRegister(it->second.va_, size, DEV_SVM_MAP_HOST, deviceLogicId, &it->second.accessVA_);
3581 2 : } else if (it->second.connectType_ == HOST_DEVICE_CONNECT_TYPE_UB) {
3582 1 : it->second.originVa_ = malloc(size + ALIGN_4K); // 保证足够的对齐余量（最多偏移 ALIGN_4K-1）
3583 1 : CHK_PTR_NULL(it->second.originVa_);
3584 1 : void* ptr = it->second.originVa_;
3585 1 : std::size_t storeSize = 0; // 不预留对象空间，仅做指针向上对齐
3586 1 : std::size_t space = static_cast<std::size_t>(size + ALIGN_4K);
3587 : // 如果对齐失败，va_为nullptr，在HrtHalHostRegister中校验返回HCCL_E_PTR
3588 1 : it->second.va_ = std::align(static_cast<std::size_t>(ALIGN_4K), storeSize, ptr, space);
3589 1 : ret = HrtHalHostRegister(it->second.va_, size, HOST_MEM_MAP_DEV_PCIE_TH, deviceLogicId, &it->second.accessVA_);
3590 : } else {
3591 1 : return HCCL_E_NOT_SUPPORT;
3592 : }
3593 3 : if (ret != HCCL_SUCCESS) {
3594 3 : HCCL_ERROR("[CommunicatorImpl::%s] HrtHalHostRegister failed, ret: %d, connect type: %ld", __func__, ret, it->second.connectType_);
3595 1 : if (it->second.va_ != nullptr) {
3596 1 : if (it->second.connectType_ == HOST_DEVICE_CONNECT_TYPE_PCIE) {
3597 1 : HrtFree(it->second.va_);
3598 0 : } else if (it->second.connectType_ == HOST_DEVICE_CONNECT_TYPE_UB) {
3599 0 : free(it->second.originVa_);
3600 0 : it->second.originVa_ = nullptr;
3601 : }
3602 1 : it->second.va_ = nullptr;
3603 : }
3604 1 : it->second.accessVA_ = nullptr;
3605 1 : return ret;
3606 : }
3607 6 : HCCL_INFO("CommunicatorImpl::AllocAndRegKFCWorkSpace va_[%p], accessVA_[%p]", it->second.va_, it->second.accessVA_);
3608 2 : return HCCL_SUCCESS;
3609 : }
3610 :
3611 7 : HcclResult CommunicatorImpl::GetKFCWorkSpaceVA(const std::string &memTag, const uint64_t *size, void **addr, bool *newCreated)
3612 : {
3613 21 : HCCL_INFO("CommunicatorImpl::GetKFCWorkSpaceVA tag[%s]", memTag.c_str()); // 调试日志
3614 7 : auto it = tagDpuShmemArgsMap_.find(memTag);
3615 7 : if (it == tagDpuShmemArgsMap_.end()) {
3616 0 : HCCL_ERROR("memTag is invalid, memTag: %s", memTag.c_str());
3617 0 : return HCCL_E_PARA;
3618 : }
3619 7 : auto iter = tagWorkspaceVAMap_.find(memTag);
3620 7 : if (iter != tagWorkspaceVAMap_.end()) {
3621 3 : std::shared_ptr<DevBuffer> oldWorkspace = iter->second;
3622 3 : if (*size != static_cast<uint64_t>(oldWorkspace.get()->GetSize())) {
3623 3 : HCCL_ERROR("HcclCommunicator::GetKFCWorkSpaceVA, The size of oldWorkspace %p is non-consistent, target size compare now size: %llu->%llu", *addr, *size, oldWorkspace.get()->GetSize());
3624 1 : return HCCL_E_PARA;
3625 : }
3626 2 : *addr = reinterpret_cast<void *>(oldWorkspace.get()->GetAddr());
3627 2 : if (newCreated != nullptr) {
3628 1 : *newCreated = false;
3629 : }
3630 2 : return HcclResult::HCCL_SUCCESS;
3631 3 : }
3632 :
3633 10 : CHK_RET(AllocAndRegKFCWorkSpace(*size, memTag));
3634 2 : shared_ptr<DevBuffer> newWorkspace = DevBuffer::Create(reinterpret_cast<uintptr_t>(it->second.accessVA_), *size);
3635 2 : tagWorkspaceVAMap_.insert(make_pair(memTag, newWorkspace));
3636 2 : if (newCreated != nullptr) {
3637 2 : *newCreated = true;
3638 : }
3639 2 : *addr = reinterpret_cast<void *>(newWorkspace.get()->GetAddr());
3640 2 : return HcclResult::HCCL_SUCCESS;
3641 2 : }
3642 :
3643 839 : HcclResult CommunicatorImpl::DestroyKFCWorkSpaceVA()
3644 : {
3645 2517 : for (auto &tmpShmem : tagDpuShmemArgsMap_) {
3646 1678 : if (tmpShmem.second.accessVA_ == nullptr && tmpShmem.second.va_ == nullptr) {
3647 1667 : continue;
3648 : }
3649 11 : int32_t deviceLogicId = 0;
3650 11 : aclError aclRet = aclrtGetLogicDevIdByUserDevId(devLogicId, &deviceLogicId); // userDevId 转 logicDevId
3651 11 : if (aclRet != ACL_SUCCESS) {
3652 0 : HCCL_ERROR("[CommunicatorImpl::%s] aclrtGetLogicDevIdByUserDevId failed, devLogicId: %u, ret: %d", __func__, devLogicId, aclRet);
3653 0 : return HCCL_E_RUNTIME;
3654 : }
3655 :
3656 : // 必须先halHostUnregister解除映射，再释放设备内存，否则HrtFree会因内存被pin住而异常
3657 11 : if (tmpShmem.second.va_ != nullptr) {
3658 6 : HcclResult ret = HrtHalHostUnregister(tmpShmem.second.va_, deviceLogicId);
3659 6 : if (ret != HCCL_SUCCESS) {
3660 0 : HCCL_ERROR("[CommunicatorImpl::%s] HrtHalHostUnregister failed, ret[%d]", __func__, ret);
3661 : }
3662 :
3663 6 : if (tmpShmem.second.connectType_ == HOST_DEVICE_CONNECT_TYPE_PCIE) {
3664 4 : DECTOR_TRY_CATCH("CommunicatorImpl", HrtFree(tmpShmem.second.va_));
3665 2 : } else if (tmpShmem.second.connectType_ == HOST_DEVICE_CONNECT_TYPE_UB) {
3666 2 : DECTOR_TRY_CATCH("CommunicatorImpl", free(tmpShmem.second.originVa_));
3667 2 : tmpShmem.second.originVa_ = nullptr;
3668 : }
3669 : }
3670 :
3671 11 : tmpShmem.second.va_ = nullptr;
3672 11 : tmpShmem.second.accessVA_ = nullptr;
3673 11 : tagWorkspaceVAMap_.erase(tmpShmem.first);
3674 : }
3675 839 : return HCCL_SUCCESS;
3676 : }
3677 :
3678 0 : HcclResult CommunicatorImpl::GetAicpuOpStreamNotify(rtStream_t *opStream, u8 aicpuNotifyNum, void** aicpuNotify) const
3679 : {
3680 0 : GetAicpuStreamManager().AllocFreeStream();
3681 0 : Stream *stream = GetAicpuStreamManager().GetFreeStream();
3682 0 : *opStream = stream->GetPtr();
3683 0 : GetHostDeviceSyncNotifyManager().GetMc2AiCpuNotifys(aicpuNotifyNum, aicpuNotify);
3684 0 : return HCCL_SUCCESS;
3685 : }
3686 :
3687 1 : HcclResult CommunicatorImpl::GetNetLayers(uint32_t **netLayers, uint32_t *netLayerNum)
3688 : {
3689 : try {
3690 1 : CHK_PTR_NULL(rankGraph);
3691 1 : u32 rankId = rankGraph->GetMyRank();
3692 1 : std::set<u32> levels = rankGraph->GetLevels(rankId);
3693 1 : u32 num = rankGraph->GetLevelNum();
3694 1 : netLayersVec.clear();
3695 1 : netLayersVec = std::vector<u32>(levels.begin(), levels.end());
3696 1 : *netLayers = netLayersVec.data();
3697 1 : *netLayerNum = num;
3698 1 : return HCCL_SUCCESS;
3699 1 : } catch (const InvalidParamsException& e) {
3700 0 : HCCL_ERROR(e.what());
3701 0 : return HCCL_E_PARA;
3702 0 : } catch (const NullPtrException& e) {
3703 0 : HCCL_ERROR(e.what());
3704 0 : return HCCL_E_PTR;
3705 0 : } catch (const std::exception& e) {
3706 0 : HCCL_ERROR(e.what());
3707 0 : return HCCL_E_INTERNAL;
3708 0 : }
3709 : }
3710 :
3711 2 : HcclResult CommunicatorImpl::GetInstRanksByNetLayer(uint32_t netLayer, uint32_t **ranks, uint32_t *rankNum)
3712 : {
3713 2 : CHK_PTR_NULL(rankGraph);
3714 2 : u32 num = 0;
3715 2 : rankListVec.clear();
3716 65 : TRY_CATCH_RETURN(rankGraph->GetLocalInstRanks(netLayer, rankListVec, num));
3717 1 : *ranks = rankListVec.data();
3718 1 : *rankNum = num;
3719 1 : return HCCL_SUCCESS;
3720 : }
3721 :
3722 7 : HcclResult CommunicatorImpl::GetInstTopoTypeByNetLayer(uint32_t netLayer, uint32_t* topoType)
3723 : {
3724 7 : CHK_PTR_NULL(rankGraph);
3725 70 : TRY_CATCH_RETURN(rankGraph->GetNetType(netLayer));
3726 6 : auto type = rankGraph->GetNetType(netLayer);
3727 : static const std::unordered_map<NetType, uint32_t> netTypeMap = {
3728 : {NetType::CLOS, static_cast<uint32_t>(CommTopo::COMM_TOPO_CLOS)},
3729 : {NetType::MESH_1D, static_cast<uint32_t>(CommTopo::COMM_TOPO_1DMESH)},
3730 : {NetType::A3_SERVER, static_cast<uint32_t>(CommTopo::COMM_TOPO_910_93)},
3731 : {NetType::A2_AX_SERVER, static_cast<uint32_t>(CommTopo::COMM_TOPO_A2AXSERVER)},
3732 8 : {NetType::TOPO_FILE_DESC, static_cast<uint32_t>(CommTopo::COMM_TOPO_CUSTOM)}};
3733 :
3734 6 : auto it = netTypeMap.find(type);
3735 6 : if (it != netTypeMap.end()) {
3736 6 : *topoType = it->second;
3737 6 : return HCCL_SUCCESS;
3738 : }
3739 0 : return HCCL_E_PARA;
3740 : }
3741 :
3742 4 : HcclResult CommunicatorImpl::GetInstSizeListByNetLayer(uint32_t netLayer, uint32_t** instSizeList, uint32_t* listSize)
3743 : {
3744 : try {
3745 10 : CHK_PTR_NULL(rankGraph);
3746 2 : u32 size = 0;
3747 2 : instSizeVec.clear();
3748 2 : auto ret = rankGraph->GetNetInstanceList(netLayer, instSizeVec, size);
3749 2 : if (ret != HCCL_SUCCESS) {
3750 3 : HCCL_ERROR("[CommunicatorImpl::GetInstSizeListByNetLayer] Failed to get instSzie[%p] at netLayer[%u]",
3751 : listSize, netLayer);
3752 1 : return ret;
3753 : }
3754 1 : *instSizeList = instSizeVec.data();
3755 1 : *listSize = size;
3756 1 : return HCCL_SUCCESS;
3757 0 : } catch (const InvalidParamsException& e) {
3758 0 : HCCL_ERROR(e.what());
3759 0 : return HCCL_E_PARA;
3760 0 : } catch (const NullPtrException& e) {
3761 0 : HCCL_ERROR(e.what());
3762 0 : return HCCL_E_PTR;
3763 0 : } catch (const std::exception& e) {
3764 0 : HCCL_ERROR(e.what());
3765 0 : return HCCL_E_INTERNAL;
3766 0 : }
3767 : }
3768 :
3769 :
3770 3 : static HcclResult InsertInnerLink(const NetInstance::Path& path, std::vector<CommLink>& linkListVec)
3771 : {
3772 6 : for (const auto& link : path.links) {
3773 3 : const NetInstance::Link *peer2peer = &link;
3774 6 : for (LinkProtocol protocol : link.GetLinkProtocols()) {
3775 : CommLink commLink;
3776 3 : CommLinkInit(&commLink, 1);
3777 3 : const CommProtocol &commProtocol = LinkProtocolToCommProtocol(protocol);
3778 3 : commLink.linkAttr.linkProtocol = commProtocol;
3779 3 : commLink.linkAttr.hop = peer2peer->GetHop();
3780 3 : commLink.srcEndpointDesc.protocol = commProtocol;
3781 3 : commLink.dstEndpointDesc.protocol = commProtocol;
3782 :
3783 : // 设置源端点
3784 3 : std::shared_ptr<NetInstance::ConnInterface> srcConnInterface = link.GetSourceIface();
3785 3 : CHK_PTR_NULL(srcConnInterface);
3786 3 : HcclResult result = GetCommAddr(commLink.srcEndpointDesc.commAddr, srcConnInterface->GetAddr());
3787 3 : if (result != HCCL_SUCCESS)
3788 0 : return result;
3789 :
3790 : // 设置目标端点
3791 3 : std::shared_ptr<NetInstance::ConnInterface> dstConnInterface = link.GetTargetIface();
3792 3 : CHK_PTR_NULL(dstConnInterface);
3793 3 : result = GetCommAddr(commLink.dstEndpointDesc.commAddr, dstConnInterface->GetAddr());
3794 3 : if (result != HCCL_SUCCESS)
3795 0 : return result;
3796 :
3797 3 : linkListVec.emplace_back(std::move(commLink));
3798 6 : }
3799 : }
3800 :
3801 3 : return HCCL_SUCCESS;
3802 : }
3803 :
3804 12 : static HcclResult InsertClosLinks(const NetInstance::Path &path, std::vector<CommLink> &linkListVec)
3805 : {
3806 12 : const NetInstance::Link *peer2net = nullptr;
3807 12 : const NetInstance::Link *net2peer = nullptr;
3808 36 : for (const auto &link : path.links) {
3809 24 : bool srcNull = (link.GetSourceIface() == nullptr);
3810 24 : bool dstNull = (link.GetTargetIface() == nullptr);
3811 24 : if (!srcNull && dstNull) {
3812 12 : peer2net = &link ;
3813 12 : } else if (srcNull && !dstNull) {
3814 12 : net2peer = &link ;
3815 : }
3816 : }
3817 12 : auto srcInterface = peer2net->GetSourceIface();
3818 12 : auto dstInterface = net2peer->GetTargetIface();
3819 12 : CHK_PTR_NULL(srcInterface);
3820 12 : CHK_PTR_NULL(dstInterface);
3821 24 : for (LinkProtocol protocol : peer2net->GetLinkProtocols()) {
3822 : CommLink commLink;
3823 12 : CommLinkInit(&commLink, 1);
3824 12 : const CommProtocol &commProtocol = LinkProtocolToCommProtocol(protocol);
3825 :
3826 12 : commLink.linkAttr.linkProtocol = commProtocol;
3827 12 : commLink.linkAttr.hop = peer2net->GetHop();
3828 :
3829 12 : commLink.srcEndpointDesc.protocol = commProtocol;
3830 12 : commLink.dstEndpointDesc.protocol = commProtocol;
3831 :
3832 : // 设置源端点
3833 12 : HcclResult result = GetCommAddr(commLink.srcEndpointDesc.commAddr, srcInterface->GetAddr());
3834 12 : if (result != HCCL_SUCCESS)
3835 0 : return result;
3836 : // 设置目标端点
3837 12 : result = GetCommAddr(commLink.dstEndpointDesc.commAddr, dstInterface->GetAddr());
3838 12 : if (result != HCCL_SUCCESS)
3839 0 : return result;
3840 12 : linkListVec.emplace_back(std::move(commLink));
3841 12 : }
3842 12 : return HCCL_SUCCESS;
3843 12 : }
3844 :
3845 5 : HcclResult CommunicatorImpl::GetLinks(uint32_t netLayer, uint32_t srcRank, uint32_t dstRank, CommLink** linkList,
3846 : uint32_t* listSize)
3847 : {
3848 : try {
3849 5 : CHK_PTR_NULL(rankGraph);
3850 5 : std::vector<NetInstance::Path> paths = rankGraph->GetPaths(netLayer, srcRank, dstRank);
3851 5 : linkListVec.clear();
3852 : // 遍历每条path
3853 20 : for (const auto& path : paths) {
3854 : // 检查是否是Clos网络（有nullptr接口）
3855 15 : bool isClos = false;
3856 18 : for (const auto& link : path.links) {
3857 : // fabric没有接口
3858 15 : if (link.GetSourceIface() == nullptr || link.GetTargetIface() == nullptr) {
3859 12 : isClos = true;
3860 12 : break;
3861 : }
3862 : }
3863 15 : if (!isClos) {
3864 : // Peer2Peer网络：直接处理每条link
3865 3 : HcclResult ret = InsertInnerLink(path, linkListVec);
3866 3 : if (ret != HCCL_SUCCESS)
3867 0 : return ret;
3868 : } else {
3869 : // Clos网络：找到peer2net和net2peer，组合成一条链路
3870 12 : HcclResult ret = InsertClosLinks(path, linkListVec);
3871 12 : if (ret != HCCL_SUCCESS)
3872 0 : return ret;
3873 : }
3874 : }
3875 5 : *linkList = linkListVec.data();
3876 5 : *listSize = linkListVec.size();
3877 5 : return HCCL_SUCCESS;
3878 5 : } catch (const InvalidParamsException& e) {
3879 0 : HCCL_ERROR(e.what());
3880 0 : return HCCL_E_PARA;
3881 0 : } catch (const NullPtrException& e) {
3882 0 : HCCL_ERROR(e.what());
3883 0 : return HCCL_E_PTR;
3884 0 : } catch (const std::exception& e) {
3885 0 : HCCL_ERROR(e.what());
3886 0 : return HCCL_E_INTERNAL;
3887 0 : }
3888 : }
3889 :
3890 3 : HcclResult CommunicatorImpl::GetTopoInstsByLayer(uint32_t netLayer, uint32_t **topoInsts, uint32_t *topoInstNum)
3891 : {
3892 : try {
3893 3 : CHK_PTR_NULL(rankGraph);
3894 3 : auto currNetType = rankGraph->GetNetType(netLayer);
3895 2 : if (currNetType != NetType::TOPO_FILE_DESC) {
3896 3 : HCCL_ERROR(
3897 : "[CommunicatorImpl::GetTopoInstsByLayer] Only support TOPO_FILE_DESC netType ,current netType is [%d]",
3898 : currNetType);
3899 1 : return HCCL_E_PARA;
3900 : }
3901 :
3902 1 : u32 num = 0;
3903 1 : rankGraph->GetTopoInstsByLayer(netLayer, topoInstsVec, num);
3904 :
3905 1 : *topoInsts = topoInstsVec.data();
3906 1 : *topoInstNum = topoInstsVec.size();
3907 :
3908 1 : return HCCL_SUCCESS;
3909 1 : } catch (const InvalidParamsException &e) {
3910 0 : HCCL_ERROR(e.what());
3911 0 : return HCCL_E_PARA;
3912 1 : } catch (const NullPtrException &e) {
3913 3 : HCCL_ERROR(e.what());
3914 1 : return HCCL_E_PTR;
3915 1 : } catch (const std::exception &e) {
3916 0 : HCCL_ERROR(e.what());
3917 0 : return HCCL_E_INTERNAL;
3918 0 : }
3919 : }
3920 :
3921 3 : HcclResult CommunicatorImpl::GetTopoType(uint32_t netLayer, uint32_t topoInstId, CommTopo* topoType)
3922 : {
3923 : try {
3924 3 : CHK_PTR_NULL(rankGraph);
3925 3 : auto currNetType = rankGraph->GetNetType(netLayer);
3926 2 : if (currNetType != NetType::TOPO_FILE_DESC) {
3927 3 : HCCL_ERROR(
3928 : "[CommunicatorImpl::GetTopoInstsByLayer] Only support TOPO_FILE_DESC netType, current netType is [%d]",
3929 : currNetType);
3930 1 : return HCCL_E_PARA;
3931 : }
3932 1 : Hccl::TopoType type;
3933 1 : HcclResult ret = rankGraph->GetTopoType(netLayer, topoInstId, type);
3934 1 : if (ret != HCCL_SUCCESS) {
3935 0 : HCCL_ERROR("[CommunicatorImpl::GetTopoType] Failed to get topo type at netLayer [%u] ret=%d", netLayer, ret);
3936 0 : return ret;
3937 : }
3938 : static const std::unordered_map<Hccl::TopoType, CommTopo> topoTypeMap = {
3939 : {Hccl::TopoType::CLOS, COMM_TOPO_CLOS},
3940 : {Hccl::TopoType::MESH_1D, COMM_TOPO_1DMESH},
3941 : {Hccl::TopoType::A3_SERVER, COMM_TOPO_910_93},
3942 3 : {Hccl::TopoType::A2_AX_SERVER, COMM_TOPO_A2AXSERVER}};
3943 1 : auto it = topoTypeMap.find(type);
3944 1 : if (it != topoTypeMap.end()) {
3945 1 : *topoType = it->second;
3946 1 : return HCCL_SUCCESS;
3947 : }
3948 0 : return HCCL_E_PARA;
3949 1 : } catch (const InvalidParamsException& e) {
3950 0 : HCCL_ERROR(e.what());
3951 0 : return HCCL_E_PARA;
3952 1 : } catch (const NullPtrException& e) {
3953 3 : HCCL_ERROR(e.what());
3954 1 : return HCCL_E_PTR;
3955 1 : } catch (const std::exception& e) {
3956 0 : HCCL_ERROR(e.what());
3957 0 : return HCCL_E_INTERNAL;
3958 0 : }
3959 : }
3960 :
3961 3 : HcclResult CommunicatorImpl::GetRanksByTopoInst(uint32_t netLayer, uint32_t topoInstId, uint32_t **ranks,
3962 : uint32_t *rankNum)
3963 : {
3964 : try {
3965 3 : CHK_PTR_NULL(rankGraph);
3966 3 : auto currNetType = rankGraph->GetNetType(netLayer);
3967 2 : if (currNetType != NetType::TOPO_FILE_DESC) {
3968 3 : HCCL_ERROR(
3969 : "[CommunicatorImpl::GetTopoInstsByLayer] Only support TOPO_FILE_DESC netType, current netType is [%d]",
3970 : currNetType);
3971 1 : return HCCL_E_PARA;
3972 : }
3973 1 : u32 num = 0;
3974 1 : auto ret = rankGraph->GetRanksByTopoInst(netLayer, topoInstId, ranksVec, num);
3975 1 : if (ret != HCCL_SUCCESS) {
3976 0 : HCCL_ERROR("[CommunicatorImpl::GetRanksByTopoInst] Failed to get topo type at netLayer [%u] ret=%d", netLayer, ret);
3977 0 : return ret;
3978 : }
3979 1 : *ranks = ranksVec.data();
3980 1 : *rankNum = ranksVec.size();
3981 1 : return HCCL_SUCCESS;
3982 1 : } catch (const InvalidParamsException &e) {
3983 0 : HCCL_ERROR(e.what());
3984 0 : return HCCL_E_PARA;
3985 1 : } catch (const NullPtrException &e) {
3986 3 : HCCL_ERROR(e.what());
3987 1 : return HCCL_E_PTR;
3988 1 : } catch (const std::exception &e) {
3989 0 : HCCL_ERROR(e.what());
3990 0 : return HCCL_E_INTERNAL;
3991 0 : }
3992 : }
3993 :
3994 1 : HcclResult CommunicatorImpl::GetInstSizeByNetLayer(uint32_t netLayer, uint32_t* rankNum)
3995 : {
3996 : try {
3997 1 : CHK_PTR_NULL(rankGraph);
3998 1 : u32 num = rankGraph->GetLocalInstSize(netLayer);
3999 1 : *rankNum = static_cast<uint32_t>(num);
4000 1 : return HCCL_SUCCESS;
4001 0 : } catch (const InvalidParamsException& e) {
4002 0 : HCCL_ERROR(e.what());
4003 0 : return HCCL_E_PARA;
4004 0 : } catch (const NullPtrException& e) {
4005 0 : HCCL_ERROR(e.what());
4006 0 : return HCCL_E_PTR;
4007 0 : } catch (const std::exception& e) {
4008 0 : HCCL_ERROR(e.what());
4009 0 : return HCCL_E_INTERNAL;
4010 0 : }
4011 : }
4012 :
4013 4 : HcclResult CommunicatorImpl::GetEndpointNum(uint32_t layer, uint32_t topoInstId, uint32_t* num)
4014 : {
4015 4 : CHK_PTR_NULL(rankGraph);
4016 4 : HcclResult ret = rankGraph->GetEndpointNum(layer, topoInstId, num);
4017 4 : if (ret != HCCL_SUCCESS) {
4018 0 : HCCL_ERROR("[CommunicatorImpl::GetEndpointNum] Faild to get endpoint num at netLayer [%u] with topoInstId[%u]", layer, topoInstId);
4019 0 : return ret;
4020 : }
4021 4 : return HCCL_SUCCESS;
4022 : }
4023 :
4024 3 : HcclResult CommunicatorImpl::GetEndpointDesc(uint32_t layer, uint32_t topoInstId, uint32_t *descNum, EndpointDesc *endpointDesc)
4025 : {
4026 3 : CHK_PTR_NULL(rankGraph);
4027 3 : HcclResult ret = rankGraph->GetEndpointDesc(layer, topoInstId, descNum, endpointDesc);
4028 3 : if (ret != HCCL_SUCCESS) {
4029 0 : HCCL_ERROR("[CommunicatorImpl::GetEndpointDesc] Failed to get endpoint desc at netLayer [%u] with descNum [%p]", layer, descNum);
4030 0 : return ret;
4031 : }
4032 3 : return HCCL_SUCCESS;
4033 : }
4034 :
4035 3 : HcclResult CommunicatorImpl::GetEndpointInfo(uint32_t rankId, const EndpointDesc* endPointDesc, EndpointAttr endpointAttr,
4036 : uint32_t infoLen, void* info)
4037 : {
4038 3 : CHK_PTR_NULL(rankGraph);
4039 3 : HcclResult ret = rankGraph->GetEndpointInfo(rankId, endPointDesc, endpointAttr, infoLen, info);
4040 3 : if (ret != HCCL_SUCCESS) {
4041 3 : HCCL_ERROR("[CommunicatorImpl::GetEndpointInfo] Faild to get endpoint info with rank [%u]", rankId);
4042 1 : return ret;
4043 : }
4044 2 : return HCCL_SUCCESS;
4045 : }
4046 :
4047 :
4048 5 : HcclResult CommunicatorImpl::SaveTopoDesc(std::string &identifier)
4049 : {
4050 5 : uint32_t topoType = 0;
4051 5 : CHK_RET(GetInstTopoTypeByNetLayer(0, &topoType)); // layer 0
4052 :
4053 5 : CommTopoDesc::GetInstance().SaveRankSize(identifier, rankSize);
4054 5 : CommTopoDesc::GetInstance().SaveL0TopoType(identifier, static_cast<CommTopo>(topoType));
4055 5 : return HCCL_SUCCESS;
4056 : }
4057 :
4058 3 : void CommunicatorImpl::CheckAcceleratorConsistency(AcceleratorState commAccelerator, AcceleratorState tilingAccelerator) const
4059 : {
4060 3 : bool isCommAiv = (commAccelerator == AcceleratorState::AIV || commAccelerator == AcceleratorState::AIV_ONLY);
4061 3 : bool isTilingCcu = (tilingAccelerator == AcceleratorState::CCU_MS || tilingAccelerator == AcceleratorState::CCU_SCHED);
4062 :
4063 3 : bool isCommCcu = (commAccelerator == AcceleratorState::CCU_MS || commAccelerator == AcceleratorState::CCU_SCHED);
4064 3 : bool isTilingAiv = (tilingAccelerator == AcceleratorState::AIV || tilingAccelerator == AcceleratorState::AIV_ONLY);
4065 :
4066 3 : if ((isCommAiv && isTilingCcu) || (isCommCcu && isTilingAiv)) {
4067 3 : HCCL_WARNING("CommunicatorImpl::GetTilingAccelerator comm accelerator is [%s] but tiling accelerator is [%s]",
4068 : commAccelerator.Describe().c_str(), tilingAccelerator.Describe().c_str());
4069 : }
4070 3 : }
4071 :
4072 7 : HcclResult CommunicatorImpl::GetTilingAccelerator(void *mc2Tiling, AcceleratorState& acceleratorState) const
4073 : {
4074 21 : HCCL_INFO("[CommunicatorImpl::%s] start.", __func__);
4075 7 : auto tilingVersion = *static_cast<uint32_t *>(mc2Tiling);
4076 21 : HCCL_INFO("[CommunicatorImpl:%s] Tiling version [%u]", __func__, tilingVersion);
4077 7 : if (tilingVersion != UNKNOWN_TILING_V1 && tilingVersion != UNKNOWN_TILING_V2) {
4078 3 : HCCL_ERROR("[CommunicatorImpl::GetTilingAccelerator] Tiling version not support, version[%u]", tilingVersion);
4079 1 : return HCCL_E_NOT_SUPPORT;
4080 : }
4081 6 : uint8_t accelerator{0};
4082 6 : if (tilingVersion == UNKNOWN_TILING_V1) {
4083 : // 从mc2Tiling中获取需要的算法信息,校验所有commConfig的communicationEngine是否一致
4084 6 : Mc2Tiling *mc2TilingPtr = reinterpret_cast<Mc2Tiling *>(mc2Tiling);
4085 6 : accelerator = static_cast<Mc2Tiling *>(mc2Tiling)->commConfig.communicationEngine;
4086 6 : Mc2CommConfig *commConfigPtr = reinterpret_cast<Mc2CommConfig *>(
4087 : reinterpret_cast<uint8_t *>(mc2TilingPtr) + sizeof(uint32_t) + sizeof(uint32_t) + sizeof(Mc2ServerCfg));
4088 12 : for (uint32_t index = 0; index < mc2TilingPtr->commConfigNum; index++) {
4089 6 : const Mc2CommConfig &commConfig = *(commConfigPtr + index);
4090 6 : if (commConfig.communicationEngine != accelerator) {
4091 0 : HCCL_ERROR("[CommunicatorImpl::GetTilingAccelerator] Input communicationEngine [%u] and [%u] not equal", commConfig.communicationEngine, accelerator);
4092 0 : return HCCL_E_PARA;
4093 : }
4094 : }
4095 : } else {
4096 0 : Mc2InitTilingInner *mc2TilingPtr = reinterpret_cast<Mc2InitTilingInner *>(mc2Tiling);
4097 0 : const auto offset = mc2TilingPtr->offset[0];
4098 0 : const auto &commConfig
4099 0 : = *(reinterpret_cast<const Mc2CcTilingInner *>(reinterpret_cast<const uint8_t *>(mc2TilingPtr) + offset));
4100 0 : accelerator = commConfig.communicationEngine;
4101 :
4102 0 : HCCL_INFO("[CommunicatorImpl::%s] tilingAccelerator[%u].", __func__, accelerator);
4103 : }
4104 :
4105 6 : HcclAccelerator hcclAccelerator = HcclAccelerator::DEFAULT;
4106 6 : if (accelerator <= HcclAccelerator::AICPU) {
4107 5 : hcclAccelerator = static_cast<HcclAccelerator::Value>(accelerator);
4108 : }
4109 18 : HCCL_INFO("[CommunicatorImpl::%s] hcclAccelerator[%s].", __func__, hcclAccelerator.Describe().c_str());
4110 6 : switch (hcclAccelerator) {
4111 1 : case HcclAccelerator::CCU_SCHED:
4112 1 : acceleratorState = AcceleratorState::CCU_SCHED;
4113 1 : break;
4114 1 : case HcclAccelerator::AIV:
4115 1 : acceleratorState = AcceleratorState::AIV;
4116 1 : break;
4117 0 : case HcclAccelerator::AIV_ONLY:
4118 0 : acceleratorState = AcceleratorState::AIV_ONLY;
4119 0 : break;
4120 4 : default:
4121 12 : HCCL_ERROR("[GetTilingAccelerator] Tiling hcclAccelerator not support, hcclAccelerator[%s]", hcclAccelerator.Describe().c_str());
4122 4 : return HCCL_E_NOT_SUPPORT;
4123 : }
4124 :
4125 2 : AcceleratorState commAccelerator = GetCommExecuteConfig().accState;
4126 2 : CheckAcceleratorConsistency(commAccelerator, acceleratorState);
4127 :
4128 2 : return HCCL_SUCCESS;
4129 : }
4130 :
4131 2 : HcclResult CommunicatorImpl::CalcNumBlocks(const CollOpParams &opParams, int32_t aivCoreLimit, std::string &algName,
4132 : u32 &numBlocks) const
4133 : {
4134 6 : HCCL_INFO("[CommunicatorImpl::CalcNumBlocks] count[%llu], dataType[%s], opType[%s], aivCoreLimit[%d], algName[%s].",
4135 : opParams.count, opParams.dataType.Describe().c_str(), opParams.opType.Describe().c_str(), aivCoreLimit,
4136 : algName.c_str());
4137 2 : numBlocks = aivCoreLimit;
4138 2 : return HCCL_SUCCESS;
4139 : }
4140 :
4141 2 : HcclResult CommunicatorImpl::GetAlgExecParam(const CollOpParams &opParams, bool clearEnable, void *&commContext,
4142 : u64 &len, u32 aivCoreLimit)
4143 : {
4144 6 : HCCL_INFO("[CommunicatorImpl::GetAlgExecParam] clearEnable[%d], aivCoreLimit[%u].", clearEnable, aivCoreLimit);
4145 2 : bool ifAiv = true;
4146 2 : std::string algName = "";
4147 2 : CHK_RET(HcomSelectAlg(opParams, aivCoreLimit, ifAiv, algName));
4148 2 : bool isAiv = (opExecuteConfig.accState == AcceleratorState::AIV || opExecuteConfig.accState == AcceleratorState::AIV_ONLY);
4149 2 : if (!isAiv) {
4150 3 : HCCL_WARNING("GetAlgExecParam only supported aiv.");
4151 1 : return HCCL_E_NOT_SUPPORT;
4152 : }
4153 :
4154 1 : u32 numBlocks = 0;
4155 1 : CHK_RET(CalcNumBlocks(opParams, aivCoreLimit, algName, numBlocks));
4156 :
4157 1 : return collService->GetAlgExecParam(clearEnable, numBlocks, commContext, len);
4158 2 : }
4159 :
4160 : HcclResult DeregisterOffloadSlaveStreams(const std::string &opTag);
4161 :
4162 1 : HcclResult CommunicatorImpl::ClearOpResource(const std::string &opTag)
4163 : {
4164 3 : HCCL_INFO("CommunicatorImpl::%s] opTag[%s]", __func__, opTag.c_str());
4165 : // 清空stream资源
4166 1 : CHK_RET(GetStreamManager().offload->ClearOpStream(opTag));
4167 : // 清空workspaceMem资源
4168 1 : offloadScrachBufferMap.erase(opTag);
4169 3 : HCCL_RUN_INFO("[CommunicatorImpl][%s] offloadScrachBuffer free, opTag[%s]", __func__, opTag.c_str());
4170 : // 清空input/output/scratch资源
4171 1 : CHK_RET(GetDataBufferManager().Deregister(opTag));
4172 1 : CHK_RET(GetLocalRmaBufManager().Dereg(opTag));
4173 : // 清空transport资源
4174 1 : CHK_RET(GetMemTransportManager()->ClearOpTransport(opTag));
4175 : // 清空aicpu_ts—host侧打包资源
4176 1 : CollServiceAiCpuImpl *aiCpuCollService = dynamic_cast<CollServiceAiCpuImpl *>(collServices[AcceleratorState::AICPU_TS].get());
4177 1 : CHK_PTR_NULL(aiCpuCollService);
4178 1 : CHK_RET(aiCpuCollService->ClearOpLoadedInfo(opTag));
4179 1 : return HCCL_SUCCESS;
4180 : }
4181 :
4182 47 : std::vector<LinkData> CommunicatorImpl::GetFullMeshLinks() const
4183 : {
4184 141 : HCCL_INFO("[CommunicatorImpl::%s] start.", __func__);
4185 :
4186 : // 遍历所有rank，两两建链
4187 47 : std::vector<LinkData> links;
4188 47 : std::unordered_set<LinkData> linkDataSet;
4189 47 : int rankSize = GetRankSize();
4190 47 : int myRank = GetMyRank();
4191 355 : for (int dRank = 0; dRank < rankSize; dRank++) {
4192 308 : if (myRank == dRank) {
4193 46 : continue;
4194 : }
4195 2086 : for (u32 level = 0; level < MAX_NET_LAYER; level++) {
4196 1858 : vector<LinkData> tempLinks;
4197 1858 : std::vector<NetInstance::Path> paths = GetRankGraph()->GetPaths(level, myRank, dRank);
4198 1892 : for (NetInstance::Path &path : paths) {
4199 34 : tempLinks.emplace_back(LinkData(path));
4200 : }
4201 :
4202 1858 : if (!tempLinks.empty()) {
4203 34 : linkDataSet.insert(tempLinks.at(0));
4204 34 : break;
4205 : }
4206 1892 : }
4207 : }
4208 :
4209 47 : links.assign(linkDataSet.begin(), linkDataSet.end());
4210 :
4211 141 : HCCL_INFO("[CommunicatorImpl::%s] end, links size[%zu]", __func__, links.size());
4212 47 : return links;
4213 47 : }
4214 :
4215 0 : ErrorMessageReport CommunicatorImpl::GetAicpuTaskException()
4216 : {
4217 0 : HcclResult ret = HCCL_SUCCESS;
4218 0 : ErrorMessageReport errorMessage;
4219 0 : if (kfcStatusTransferD2H != nullptr)
4220 : {
4221 0 : ret = kfcStatusTransferD2H->Get(sizeof(KfcStatus) + sizeof(KfcErrType),
4222 : sizeof(errorMessage), reinterpret_cast<uint8_t *>(&errorMessage));
4223 0 : if (ret != HCCL_SUCCESS)
4224 : {
4225 0 : HCCL_ERROR("GetAicpuTaskException get aicpu task exception failed.ret[%u]", ret);
4226 : }
4227 : } else {
4228 0 : HCCL_ERROR("GetAicpuTaskException kfcStatusTransferD2H is nullptr");
4229 : }
4230 0 : HCCL_INFO("[CommunicatorImpl::GetAicpuTaskException] end");
4231 0 : return errorMessage;
4232 : }
4233 :
4234 :
4235 0 : u32 CommunicatorImpl::GetRankInParentComm() const {
4236 0 : return static_cast<u32>(rankInParentComm);
4237 : }
4238 67 : void CommunicatorImpl::RegisterAicpuKernel()
4239 : {
4240 67 : aicpuKernelHolder_.Load();
4241 67 : }
4242 :
4243 6 : aclrtFuncHandle CommunicatorImpl::GetAicpuKernelFuncHandle(const char *kernelName) const
4244 : {
4245 6 : return aicpuKernelHolder_.GetAicpuKernelFuncHandle(kernelName);
4246 : }
4247 :
4248 0 : HcclResult CommunicatorImpl::Mc2AiCpuStreamAllocAndGetV2(rtStream_t *aiCpuStream)
4249 : {
4250 0 : if (aicpuStreamManager == nullptr)
4251 : {
4252 0 : HCCL_ERROR("[CommunicatorImpl::Mc2AiCpuStreamAllocAndGetV2] aicpuStreamManager is nullPtr!");
4253 0 : return HCCL_E_PTR;
4254 : }
4255 0 : aicpuStreamManager->AllocFreeStream();
4256 0 : Stream *stream = aicpuStreamManager->GetFreeStream();
4257 0 : *aiCpuStream = stream->GetPtr();
4258 0 : HCCL_RUN_INFO("[CommunicatorImpl::Mc2AiCpuStreamAllocAndGetV2] success, stream %s", stream->Describe().c_str());
4259 0 : return HCCL_SUCCESS;
4260 : }
4261 :
4262 2 : HcclResult CommunicatorImpl::SaveDpuStreamId()
4263 : {
4264 2 : dpuStreamId = HrtGetStreamId(dpuStream);
4265 6 : HCCL_INFO("[CommunicatorImpl::SaveDpuStreamId] dpuStreamId_[%u]", dpuStreamId);
4266 2 : return HCCL_SUCCESS;
4267 : }
4268 :
4269 0 : HcclResult CommunicatorImpl::GetRankIpPortMap(RankIpPortMapPtr& rankIpPortMap)
4270 : {
4271 0 : CHK_PTR_NULL(rankIpPortMap_);
4272 0 : rankIpPortMap = rankIpPortMap_;
4273 0 : return HCCL_SUCCESS;
4274 : }
4275 :
4276 0 : HcclResult CommunicatorImpl::SetRankIpPortMap(const RankIpPortMapPtr& rankIpPortMap)
4277 : {
4278 0 : CHK_PTR_NULL(rankIpPortMap);
4279 0 : rankIpPortMap_ = rankIpPortMap;
4280 0 : return HCCL_SUCCESS;
4281 : }
4282 :
4283 : } // namespace Hccl
|
