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