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