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 <atomic>
12 : #include <algorithm>
13 : #include <arpa/inet.h>
14 : #include <fstream>
15 : #include <unistd.h>
16 : #include <hccl/hccl_types.h>
17 : #include "hccl_communicator.h"
18 : #include "hccl_comm_pub.h"
19 : #include "task_abort_handler_pub.h"
20 : #include "i_hccl_one_sided_service.h"
21 : #include "comm_configer.h"
22 : #include "launch_aicpu.h"
23 : #include "launch_device.h"
24 : #include "sal_pub.h"
25 : #include "coll_comm_config.h"
26 : #include "coll_comm_mgr.h"
27 : #include "env_config/env_config_v2.h"
28 :
29 : namespace hccl {
30 28 : HcclResult hcclComm::AllReduce(
31 : const std::string& tag, void* inputPtr, void* outputPtr, u64 count, HcclDataType dataType, HcclReduceOp op,
32 : HcclRtStream stream, SyncMode syncMode)
33 : {
34 : /* 增加输出日志关键字 */
35 28 : HCCL_INFO(
36 : "HCCL_KEY_INFO: tag[%s], input_ptr[%p], output_ptr[%p], count[%llu], data_type[%s], op[%s]", tag.c_str(),
37 : inputPtr, outputPtr, count, GetDataTypeEnumStr(dataType).c_str(), GetReduceOpEnumStr(op).c_str());
38 :
39 : /* * 入参检查 */
40 51 : CHK_PTR_NULL(stream);
41 51 : CHK_PTR_NULL(inputPtr);
42 51 : CHK_PTR_NULL(outputPtr);
43 :
44 51 : CHK_PRT_RET(
45 : tag.empty(),
46 : HCCL_ERROR("[HcclComm][AllReduce]errNo[0x%016llx] AllReduce tag length is 0", HCCL_ERROR_CODE(HCCL_E_PARA)),
47 : HCCL_E_PARA);
48 :
49 49 : CHK_RET(communicator_->CheckCount(count));
50 38 : CHK_RET(communicator_->CheckDataType(dataType, true));
51 36 : CHK_RET(communicator_->CheckReduceDataType(dataType, op));
52 31 : CHK_RET(communicator_->CheckReductionOp(op));
53 30 : HcclResult ret = communicator_->AllReduce(tag, inputPtr, outputPtr, count, dataType, op, stream, syncMode);
54 44 : if (ret != HCCL_SUCCESS) {
55 4 : PrintSubmittedOpCnt(tag, ret);
56 4 : return ret;
57 : }
58 :
59 40 : return HCCL_SUCCESS;
60 : }
61 :
62 30 : HcclResult hcclComm::AllReduceOutPlace(
63 : const std::string& tag, void* inputPtr, void* outputPtr, u64 count, HcclDataType dataType, HcclReduceOp op,
64 : HcclRtStream stream, SyncMode syncMode)
65 : {
66 : /* 增加输出日志关键字 */
67 30 : HCCL_INFO(
68 : "HCCL_KEY_INFO: tag[%s], input_ptr[%p], output_ptr[%p], count[%llu], data_type[%s], op[%s]", tag.c_str(),
69 : inputPtr, outputPtr, count, GetDataTypeEnumStr(dataType).c_str(), GetReduceOpEnumStr(op).c_str());
70 :
71 : /* * 入参检查 */
72 39 : CHK_RET(communicator_->CheckDataType(dataType, true));
73 34 : CHK_RET(communicator_->CheckReduceDataType(dataType, op));
74 34 : HcclResult ret = communicator_->AllReduceOutPlace(tag, inputPtr, outputPtr, count, dataType, op, stream, syncMode);
75 35 : if (ret != HCCL_SUCCESS) {
76 0 : PrintSubmittedOpCnt(tag, ret);
77 0 : return ret;
78 : }
79 :
80 35 : return HCCL_SUCCESS;
81 : }
82 :
83 0 : HcclResult hcclComm::GetOneSidedService(IHcclOneSidedService** service)
84 : {
85 0 : CHK_RET(communicator_->GetOneSidedService(service));
86 :
87 0 : return HCCL_SUCCESS;
88 : }
89 :
90 0 : HcclResult hcclComm::InitOneSidedServiceNetDevCtx(u32 remoteRankId)
91 : {
92 0 : CHK_RET(communicator_->InitOneSidedServiceNetDevCtx(remoteRankId));
93 0 : return HCCL_SUCCESS;
94 : }
95 :
96 0 : HcclResult hcclComm::OneSidedServiceStartListen(NicType nicType, HcclNetDevCtx netDevCtx)
97 : {
98 0 : CHK_SMART_PTR_NULL(communicator_);
99 0 : CHK_RET(communicator_->OneSidedServiceStartListen(nicType, netDevCtx));
100 0 : return HCCL_SUCCESS;
101 : }
102 :
103 0 : HcclResult hcclComm::GetOneSidedServiceDevIpAndPort(NicType nicType, HcclIpAddress& ipAddress, u32& port)
104 : {
105 0 : CHK_SMART_PTR_NULL(communicator_);
106 0 : CHK_RET(communicator_->GetOneSidedServiceDevIpAndPort(nicType, ipAddress, port));
107 0 : return HCCL_SUCCESS;
108 : }
109 :
110 0 : HcclResult hcclComm::DeinitOneSidedService()
111 : {
112 0 : CHK_SMART_PTR_NULL(communicator_);
113 0 : CHK_RET(communicator_->DeinitOneSidedService());
114 0 : return HCCL_SUCCESS;
115 : }
116 :
117 412 : HcclResult hcclComm::RegistTaskAbortHandler() const
118 : {
119 412 : HCCL_RUN_INFO("RegistTaskAbortHandler begin, group[%s]", identifier_.c_str());
120 412 : CHK_RET(TaskAbortHandler::Init(communicator_.get()));
121 412 : return HCCL_SUCCESS;
122 : }
123 :
124 653 : HcclResult hcclComm::UnRegistTaskAbortHandler() const
125 : {
126 653 : HCCL_RUN_INFO("UnRegistTaskAbortHandler begin, group[%s]", identifier_.c_str());
127 653 : CHK_RET(TaskAbortHandler::DeInit(communicator_.get()));
128 653 : return HCCL_SUCCESS;
129 : }
130 :
131 0 : HcclResult hcclComm::RegisterCommUserMem(void* addr, u64 size, void** handle)
132 : {
133 0 : CHK_SMART_PTR_NULL(communicator_);
134 0 : CHK_RET(communicator_->RegisterCommUserMem(addr, size, handle));
135 0 : return HCCL_SUCCESS;
136 : }
137 :
138 0 : HcclResult hcclComm::DeregisterCommUserMem(void* handle)
139 : {
140 0 : CHK_SMART_PTR_NULL(communicator_);
141 0 : CHK_RET(communicator_->DeregisterCommUserMem(handle));
142 0 : return HCCL_SUCCESS;
143 : }
144 :
145 0 : HcclResult hcclComm::ExchangeCommUserMem(void* handle, std::vector<u32>& peerRanks)
146 : {
147 0 : CHK_SMART_PTR_NULL(communicator_);
148 0 : return communicator_->ExchangeCommUserMem(handle, peerRanks);
149 : }
150 :
151 235 : HcclResult hcclComm::SetIndependentOpConfig(const CommConfig& commConfig, const RankTable_t& rankTable)
152 : {
153 235 : CHK_SMART_PTR_NULL(communicator_);
154 235 : HcclTopoAttr topoAttr = communicator_->GetTopoAttr();
155 235 : aclrtBinHandle binHandle = communicator_->GetBinHandle();
156 235 : HDCommunicateParams kfcControlTransferH2DParams;
157 235 : HDCommunicateParams kfcStatusTransferD2HParams;
158 471 : std::function<bool()> getAicpuCommState = [this]() {
159 1 : return this->GetIndependentOp().GetAicpuCommState();
160 235 : };
161 235 : CHK_RET(communicator_->GetHDCommunicate(kfcControlTransferH2DParams, kfcStatusTransferD2HParams));
162 235 : CHK_RET(communicator_->SetGetAicpuCommState(getAicpuCommState));
163 235 : CHK_RET(GetIndependentOp().SetIndependentOpConfig(
164 : commConfig, rankTable, topoAttr, binHandle, kfcControlTransferH2DParams, kfcStatusTransferD2HParams,
165 : communicator_->GetCCLbufferManager()));
166 235 : return HCCL_SUCCESS;
167 235 : }
168 :
169 403 : HcclResult hcclComm::ReleaseChannel() { return independentOp_.GetChannelManager().ReleaseChannel(); }
170 :
171 404 : HcclResult hcclComm::InitIndependentOp()
172 : {
173 404 : if (communicator_ != nullptr) {
174 404 : communicator_->SetReleaseChannel([this]() -> HcclResult {
175 404 : return this->ReleaseChannel();
176 : });
177 : }
178 404 : ChannelManagerCallbacks channelCallbacks;
179 : channelCallbacks.indOpTransportAlloc
180 808 : = [this](
181 : const std::string& tag, OpCommTransport& opCommTransport, bool isAicpuModeEn,
182 : const HcclMemHandle* memHandles, uint32_t memHandleNum) -> HcclResult {
183 0 : return this->IndOpTransportAlloc(tag, opCommTransport, isAicpuModeEn, memHandles, memHandleNum);
184 404 : };
185 1212 : channelCallbacks.getRankLists = [this]() -> std::vector<RankInfo> {
186 404 : return this->GetRankLists();
187 404 : };
188 808 : return independentOp_.SetChannelCallbacks(channelCallbacks);
189 404 : }
190 :
191 318 : IndependentOp& hcclComm::GetIndependentOp() { return independentOp_; }
192 0 : HcclResult hcclComm::PrepareChannelMem(
193 : const std::string& tag, TransportIOMem& transMem, const HcclMemHandle* memHandles, uint32_t memHandleNum)
194 : {
195 : // 获取本地cclbuffer
196 : CommBuffer commBuffer;
197 0 : CHK_RET(GetIndependentOp().GetCommMemMgr().GetHcclBuffer(&commBuffer));
198 0 : DeviceMem cclbuffer = DeviceMem::create(commBuffer.addr, commBuffer.size);
199 0 : CHK_PTR_NULL(cclbuffer.ptr());
200 :
201 : // 获取通信域内存
202 0 : IndOpMem indOpMem{};
203 0 : std::vector<HcclMem> localMemVec{};
204 0 : CHK_RET(GetIndependentOp().GetCommMemMgr().CommGetLocalRegMemByHandles(memHandles, memHandleNum, localMemVec));
205 0 : for (const HcclMem& mem : localMemVec) {
206 0 : if (mem.type == HCCL_MEM_TYPE_HOST) {
207 0 : indOpMem.userHostMem.push_back(HostMem::create(mem.addr, mem.size));
208 0 : CHK_PTR_NULL(indOpMem.userHostMem.back().ptr());
209 0 : } else if (mem.type == HCCL_MEM_TYPE_DEVICE) {
210 0 : indOpMem.userDeviceMem.push_back(DeviceMem::create(mem.addr, mem.size));
211 0 : CHK_PTR_NULL(indOpMem.userDeviceMem.back().ptr());
212 : }
213 : }
214 0 : transMem.indOpMem = indOpMem;
215 0 : transMem.cclInputMem = cclbuffer;
216 0 : transMem.cclOutputMem = cclbuffer;
217 0 : return HCCL_SUCCESS;
218 0 : }
219 0 : HcclResult hcclComm::IndOpTransportAlloc(
220 : const std::string& tag, OpCommTransport& opCommTransport, bool isAicpuModeEn, const HcclMemHandle* memHandles,
221 : uint32_t memHandleNum)
222 : {
223 0 : CHK_SMART_PTR_NULL(communicator_);
224 0 : TransportIOMem transMem;
225 0 : CHK_RET(PrepareChannelMem(tag, transMem, memHandles, memHandleNum));
226 0 : std::string commId = GetIdentifier();
227 0 : return communicator_->IndOpTransportAlloc(tag, opCommTransport, transMem, isAicpuModeEn);
228 0 : }
229 0 : HcclResult hcclComm::CommGetNetLayers(uint32_t** netLayers, uint32_t* netLayerNum)
230 : {
231 0 : return communicator_->CommGetNetLayers(netLayers, netLayerNum);
232 : }
233 :
234 0 : HcclResult hcclComm::CommGetInstSizeByNetLayer(uint32_t netLayer, uint32_t* rankNum)
235 : {
236 0 : return communicator_->CommGetInstSizeByNetLayer(netLayer, rankNum);
237 : }
238 :
239 0 : HcclResult hcclComm::CommGetInstTopoTypeByNetLayer(uint32_t netLayer, u32* topoType)
240 : {
241 0 : return communicator_->CommGetInstTopoTypeByNetLayer(netLayer, topoType);
242 : }
243 0 : HcclResult hcclComm::GetNetLayers(uint32_t** netLayers, uint32_t* netLayerNum)
244 : {
245 0 : return communicator_->GetNetLayers(netLayers, netLayerNum);
246 : }
247 :
248 0 : HcclResult hcclComm::GetInstSizeByNetLayer(uint32_t netLayer, uint32_t* rankNum)
249 : {
250 0 : return communicator_->GetInstSizeByNetLayer(netLayer, rankNum);
251 : }
252 :
253 0 : HcclResult hcclComm::GetInstTopoTypeByNetLayer(uint32_t netLayer, CommTopo* topoType)
254 : {
255 0 : return communicator_->GetInstTopoTypeByNetLayer(netLayer, topoType);
256 : }
257 :
258 0 : HcclResult hcclComm::GetInstRanksByNetLayer(uint32_t netLayer, uint32_t** rankList, uint32_t* rankNum)
259 : {
260 0 : return communicator_->GetInstRanksByNetLayer(netLayer, rankList, rankNum);
261 : }
262 :
263 0 : HcclResult hcclComm::GetInstSizeListByNetLayer(uint32_t netLayer, uint32_t** instSizeList, uint32_t* listSize)
264 : {
265 0 : return communicator_->GetInstSizeListByNetLayer(netLayer, instSizeList, listSize);
266 : }
267 :
268 0 : HcclResult hcclComm::GetTopoInstsByLayer(uint32_t netLayer, uint32_t** topoInsts, uint32_t* topoInstNum)
269 : {
270 0 : return communicator_->GetTopoInstsByLayer(netLayer, topoInsts, topoInstNum);
271 : }
272 :
273 0 : HcclResult hcclComm::GetTopoType(uint32_t netLayer, uint32_t topoInstId, CommTopo* topoType)
274 : {
275 0 : return communicator_->GetTopoType(netLayer, topoInstId, topoType);
276 : }
277 :
278 0 : HcclResult hcclComm::GetRanksByTopoInst(uint32_t netLayer, uint32_t topoInstId, uint32_t** ranks, uint32_t* rankNum)
279 : {
280 0 : return communicator_->GetRanksByTopoInst(netLayer, topoInstId, ranks, rankNum);
281 : }
282 :
283 0 : HcclResult hcclComm::GetEndpointNum(uint32_t netLayer, uint32_t topoInstId, uint32_t* num)
284 : {
285 0 : return communicator_->GetEndpointNum(netLayer, topoInstId, num);
286 : }
287 :
288 : HcclResult
289 0 : hcclComm::GetEndpointDesc(uint32_t netLayer, uint32_t topoInstId, uint32_t* descNum, EndpointDesc* endpointDesc)
290 : {
291 0 : return communicator_->GetEndpointDesc(netLayer, topoInstId, descNum, endpointDesc);
292 : }
293 :
294 0 : HcclResult hcclComm::GetEndpointInfo(
295 : uint32_t rankId, const EndpointDesc* endPointDesc, EndpointAttr endpointAttr, uint32_t infoLen, void* info)
296 : {
297 0 : return communicator_->GetEndpointInfo(rankId, endPointDesc, endpointAttr, infoLen, info);
298 : }
299 :
300 0 : HcclResult hcclComm::GetRankGraph(GraphType type, void** graph, uint32_t* len)
301 : {
302 0 : return communicator_->GetRankGraph(type, graph, len);
303 : }
304 :
305 235 : uint32_t hcclComm::GetConnectMode() { return communicator_->GetConnectMode(); }
306 0 : HcclResult hcclComm::GetDevMemWorkSpace(const std::string& memTag, uint64_t* size, void** addr, bool* newCreated)
307 : {
308 0 : return communicator_->GetDevMemWorkSpace(memTag, size, addr, newCreated);
309 : }
310 : HcclResult
311 0 : hcclComm::GetLinks(uint32_t netLayer, uint32_t srcRank, uint32_t dstRank, CommLink** linkList, uint32_t* listSize)
312 : {
313 0 : return communicator_->GetLinks(netLayer, srcRank, dstRank, linkList, listSize);
314 : }
315 :
316 0 : HcclResult hcclComm::GetHeterogMode(HcclHeterogMode* mode) { return communicator_->GetHeterogMode(mode); }
317 :
318 154 : inline uint32_t GetCollCommOpExpansionMode(CollComm* collComm)
319 : {
320 154 : auto* myRank = collComm->GetMyRank();
321 154 : CHK_PTR_NULL(myRank);
322 146 : return myRank->GetOpExpansionMode();
323 : }
324 :
325 156 : HcclResult hcclComm::InitCollComm(
326 : void* commV2, void* rankGraph, uint32_t userRank, HcclMem cclBuffer, const std::string& commName,
327 : const HcclCommConfig* config, CollCommInitMode initMode)
328 : {
329 : // 不校验config，为空时配置默认加速模式
330 :
331 : // aicpu侧初始化状态的回调函数
332 156 : ManagerCallbacks callbacks;
333 474 : callbacks.getAicpuCommState = [this]() {
334 162 : return this->GetAicpuCommState();
335 156 : };
336 312 : callbacks.setAicpuCommState = [this](bool state) {
337 0 : this->SetAicpuCommState(state);
338 156 : };
339 312 : callbacks.kernelLaunchAicpuCommInit = [this]() {
340 0 : return this->KernelLaunchAicpuCommInit();
341 156 : };
342 312 : callbacks.reportProfilingKernel = [this](uint64_t beginTime, std::string kernelName) {
343 0 : return this->ReportProfilingKernel(beginTime, kernelName);
344 156 : };
345 :
346 : // Aicpu通信域初始化参数
347 156 : auto ret = snprintf_s(commAicpuParam_.hcomId, HCOMID_MAX_SIZE, HCOMID_MAX_SIZE - 1, "%s", commName.c_str());
348 156 : if (ret < 0) {
349 0 : HCCL_ERROR(
350 : "[InitCollComm]comm id snprintf_s fail, commId: %s, commId maxSize: %u", commName.c_str(), HCOMID_MAX_SIZE);
351 0 : return HCCL_E_PARA;
352 : }
353 :
354 156 : CHK_RET(hrtGetDevice(&(commAicpuParam_.deviceLogicId)));
355 156 : CHK_RET(hrtGetDevicePhyIdByIndex(static_cast<u32>(commAicpuParam_.deviceLogicId), commAicpuParam_.devicePhyId));
356 156 : CHK_RET(hrtGetDeviceType(devType_));
357 156 : commAicpuParam_.deviceType = static_cast<u32>(devType_);
358 156 : CHK_RET(InitBinHandle());
359 :
360 156 : EXCEPTION_CATCH(
361 : collComm_ = std::make_unique<CollComm>(commV2, userRank, commName, callbacks, initMode), return HCCL_E_PTR);
362 :
363 156 : uint32_t configOpExpansionMode = 0;
364 156 : CHK_RET(ApplyHcclCommConfig(config, collComm_->GetCommConfig(), configOpExpansionMode));
365 154 : CHK_RET(collComm_->Init(rankGraph, binHandle_, cclBuffer, configOpExpansionMode));
366 154 : if (initMode == CollCommInitMode::simpleMode) { /* hccl::CommunicatorV1支持CollComm简易流程 */
367 0 : return HCCL_SUCCESS;
368 : }
369 :
370 : // 注册通信域到 CollCommMgr，由 owner(hcclComm) 负责注册/注销，避免 CollComm 反向依赖 CollCommMgr
371 154 : CollCommMgr::GetInstance().RegisteCollComm(collComm_.get());
372 :
373 154 : CHK_RET(collComm_->GetHDCommunicate(
374 : commAicpuParam_.kfcControlTransferH2DParams, commAicpuParam_.kfcStatusTransferD2HParams));
375 154 : commAicpuParam_.userRank = collComm_->GetMyRankId();
376 154 : commAicpuParam_.userRankSize = collComm_->GetRankSize();
377 : commAicpuParam_.commConfig.taskExceptionEnable
378 154 : = Hccl::EnvConfig::GetInstance().GetLogConfig().GetDfsConfig().taskExceptionEnable;
379 154 : commAicpuParam_.commConfig.notifyWaitTimeout = Hccl::EnvConfig::GetInstance().GetRtsConfig().GetExecTimeOut();
380 154 : commAicpuParam_.commConfig.plfDebugConfig = Hccl::EnvConfig::GetInstance().GetPlfDebugConfig().GetConfigValue();
381 154 : const auto opExpansionMode = GetCollCommOpExpansionMode(collComm_.get());
382 154 : HCCL_RUN_INFO(
383 : "[%s]success, commId[%s], deviceLogicId[%u], devicePhyId[%u], devType[%u], "
384 : "userRank[%u], userRankSize[%u], opExpansionMode[%u], taskExceptionEnable[%d], notifyWaitTimeout[%u], "
385 : "plfDebugConfig[0x%llx].",
386 : __func__, collComm_->GetCommId().c_str(), commAicpuParam_.deviceLogicId, commAicpuParam_.devicePhyId,
387 : commAicpuParam_.deviceType, commAicpuParam_.userRank, commAicpuParam_.userRankSize, opExpansionMode,
388 : commAicpuParam_.commConfig.taskExceptionEnable, commAicpuParam_.commConfig.notifyWaitTimeout,
389 : commAicpuParam_.commConfig.plfDebugConfig);
390 :
391 : // 当前需要支持coll comm与legacy comm混跑，coll comm确定加速模式后，需要设置comm加速模式
392 154 : auto* commImplV2 = static_cast<Hccl::HcclCommunicator*>(commV2);
393 154 : constexpr bool isCcuMsAvailable = false; // 禁止legacy通信域使用ms模式，避免抢占过多coll comm ccu可用资源
394 154 : CHK_RET(commImplV2->SetAccelerator(static_cast<int32_t>(opExpansionMode), isCcuMsAvailable));
395 :
396 154 : return HCCL_SUCCESS;
397 156 : }
398 :
399 235 : HcclResult hcclComm::InitCollCommInner(uint32_t userRank)
400 : {
401 235 : if (GetConnectMode() == 0) {
402 234 : return HCCL_SUCCESS;
403 : }
404 :
405 1 : CHK_PRT_RET(
406 : userRank == INVALID_VALUE_RANKID, HCCL_ERROR("[%s] invalid userRank[%u]", __func__, userRank), HCCL_E_PARA);
407 :
408 1 : std::string commName = GetIdentifier();
409 1 : HCCL_INFO("[%s]Init CollComm start, comm[%s], userRank[%u]", __func__, commName.c_str(), userRank);
410 1 : HcclCommunicator* hcclComm = GetHcclCommunicator();
411 1 : if (hcclComm == nullptr) {
412 1 : HCCL_WARNING("[%s] HcclCommunicator NULL, skip CollComm init", __func__);
413 1 : return HCCL_SUCCESS;
414 : }
415 :
416 0 : void* rankGraphV1 = hcclComm->GetRankGraphV1();
417 0 : if (rankGraphV1 == nullptr) {
418 0 : HCCL_WARNING("[%s] rankGraphV1 is nullptr, skip CollComm init, comm[%s]", __func__, commName.c_str());
419 0 : return HCCL_SUCCESS;
420 : }
421 :
422 0 : void* cclBufferAddr = nullptr;
423 0 : u64 cclBufferSize = 0;
424 0 : CHK_RET(CreateCommCCLbuffer());
425 0 : HcclResult ret = hcclComm->GetInCCLbuffer(cclBufferAddr, cclBufferSize);
426 0 : if (ret != HCCL_SUCCESS) {
427 0 : HCCL_ERROR("[%s] GetInCCLbuffer failed, comm[%s], ret=%d", __func__, commName.c_str(), ret);
428 0 : return ret;
429 : }
430 :
431 0 : HcclMem cclBuffer{};
432 0 : cclBuffer.size = static_cast<uint64_t>(cclBufferSize);
433 0 : cclBuffer.addr = cclBufferAddr;
434 0 : cclBuffer.type = HcclMemType::HCCL_MEM_TYPE_DEVICE;
435 0 : constexpr const HcclCommConfig* config = nullptr;
436 :
437 0 : ret = InitCollComm(nullptr, rankGraphV1, userRank, cclBuffer, commName, config, CollCommInitMode::simpleMode);
438 0 : if (ret != HCCL_SUCCESS) {
439 0 : HCCL_ERROR("[%s] InitCollComm failed, comm[%s], ret=%d", __func__, commName.c_str(), ret);
440 0 : return ret;
441 : }
442 :
443 0 : HCCL_INFO("[%s] CollComm init success for V1, comm[%s]", __func__, commName.c_str());
444 0 : return HCCL_SUCCESS;
445 1 : }
446 :
447 156 : HcclResult hcclComm::InitBinHandle()
448 : {
449 156 : std::string jsonPath;
450 156 : CHK_RET(GetKernelFilePath(jsonPath));
451 156 : jsonPath += "ccl_kernel.json";
452 :
453 156 : HcclResult retCode = LoadBinaryFromFile(jsonPath.c_str(), ACL_RT_BINARY_LOAD_OPT_CPU_KERNEL_MODE, 0, binHandle_);
454 156 : CHK_PRT_RET(
455 : retCode != HCCL_SUCCESS,
456 : HCCL_ERROR(
457 : "[InitCollComm]errNo[0x%016llx]load aicpu file fail, path[%s] optionType[%u]"
458 : "cpuKernelMode[%u].",
459 : retCode, jsonPath.c_str(), ACL_RT_BINARY_LOAD_OPT_CPU_KERNEL_MODE, 0),
460 : retCode);
461 156 : return HCCL_SUCCESS;
462 156 : }
463 :
464 656 : void hcclComm::BinaryUnLoad()
465 : {
466 656 : if (binHandle_ != nullptr) {
467 2 : HCCL_INFO("[BinaryUnLoad]aclrtBinaryUnLoad binHandle");
468 2 : aclError ret = aclrtBinaryUnLoad(binHandle_);
469 2 : if (ret != 0) {
470 1 : HCCL_RUN_WARNING("[BinaryUnLoad]aclrtBinaryUnLoad binHandle failed");
471 : }
472 2 : binHandle_ = nullptr;
473 : }
474 656 : }
475 :
476 151 : bool hcclComm::GetAicpuCommState() { return isAicpuCommInit_; }
477 :
478 0 : void hcclComm::SetAicpuCommState(bool aicpuCommState)
479 : {
480 0 : isAicpuCommInit_ = aicpuCommState;
481 0 : return;
482 : }
483 :
484 1 : HcclResult hcclComm::KernelLaunchAicpuCommInit()
485 : {
486 : // 创建局部流
487 1 : u64 beginTime = Hccl::DlProfFunction::GetInstance().dlMsprofSysCycleTime();
488 1 : Stream localStream(StreamType::STREAM_TYPE_ONLINE);
489 1 : constexpr u32 aicpuStreamMode = 1;
490 1 : CHK_RET(hrtStreamSetMode(localStream.ptr(), aicpuStreamMode));
491 :
492 : // 下kernel进行自定义算子aicpu侧通信域的公共初始化
493 1 : std::string kernelName = "RunAicpuCommInit";
494 1 : HCCL_INFO("AicpuAclKernelLaunch start");
495 1 : s32 timeout = 1836;
496 1 : if (IsCommunicatorV2()) {
497 1 : timeout = Hccl::EnvConfig::GetInstance().GetRtsConfig().GetExecTimeOut() + 25; // 多25s，避免超时
498 : } else {
499 0 : timeout = CommConfiger::GetInstance().GetCommConfigExecTimeOut("") + 25; // 多25s，避免超时
500 : }
501 1 : CHK_RET(AicpuAclKernelLaunch(
502 : localStream.ptr(), reinterpret_cast<void*>(&commAicpuParam_), sizeof(commAicpuParam_), binHandle_, kernelName,
503 : true, timeout));
504 1 : HCCL_INFO("AicpuAclKernelLaunch end, hcclStreamSynchronize start");
505 1 : CHK_RET(hcclStreamSynchronize(localStream.ptr(), timeout));
506 1 : HCCL_INFO("[KernelLaunchAicpuCommInit] ReportAicpuCommKernel begin");
507 1 : CHK_PTR_NULL(collComm_);
508 0 : HcclCommDfx* hcclComDfx = collComm_->GetHcclCommDfx();
509 0 : CHK_PTR_NULL(hcclComDfx);
510 : // 通信域初始化在op注册之前，这个地方一定是false，因为还不知道是不是图模式
511 0 : CHK_RET(hcclComDfx->ReportKernel(beginTime, identifier_, kernelName, SalGetTid(), false));
512 0 : HCCL_INFO("[KernelLaunchAicpuCommInit] ReportAicpuCommKernel end");
513 : // 打印增加初始化对应的参数
514 0 : HCCL_RUN_INFO("[%s] KernelLaunchAicpuCommInit Success", __func__);
515 0 : return HCCL_SUCCESS;
516 1 : }
517 :
518 0 : HcclResult hcclComm::ReportProfilingKernel(uint64_t beginTime, std::string kernelName)
519 : {
520 0 : CHK_PTR_NULL(collComm_);
521 0 : HcclCommDfx* hcclComDfx = collComm_->GetHcclCommDfx();
522 0 : CHK_PTR_NULL(hcclComDfx);
523 : // 通信域初始化在op注册之前，这个地方一定是false，因为还不知道是不是图模式
524 0 : CHK_RET(hcclComDfx->ReportKernel(beginTime, identifier_, kernelName, SalGetTid(), false));
525 0 : return HCCL_SUCCESS;
526 : }
527 :
528 0 : HcclComm hcclComm::GetCommunicatorV2()
529 : {
530 0 : if (collComm_ == nullptr) {
531 0 : return nullptr;
532 : }
533 0 : return collComm_->GetCommunicatorV2();
534 : }
535 :
536 2 : HcclCommunicator* hcclComm::GetHcclCommunicator() { return communicator_.get(); }
537 :
538 300 : CollComm* hcclComm::GetCollComm() { return collComm_ != nullptr ? collComm_.get() : nullptr; }
539 :
540 4 : HcclResult hcclComm::Resume()
541 : {
542 4 : if (IsCommunicatorV2()) {
543 2 : CHK_RET(collComm_->Resume());
544 : } else {
545 2 : CHK_RET(communicator_->Resume());
546 : }
547 :
548 3 : return HCCL_SUCCESS;
549 : }
550 3 : HcclResult hcclComm::GetCommStatus(HcclCommStatus& status)
551 : {
552 3 : if (IsCommunicatorV2()) {
553 1 : status = collComm_->GetCommStatus();
554 2 : } else if (devType_ == DevType::DEV_TYPE_910B && collComm_ != nullptr) {
555 0 : status = collComm_->GetCommStatus();
556 : } else {
557 2 : HCCL_ERROR("[%s] deviceType is not support", __func__);
558 2 : return HCCL_E_NOT_SUPPORT;
559 : }
560 1 : return HCCL_SUCCESS;
561 : }
562 :
563 : } // namespace hccl
|
