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