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 "aicpu_one_side_service.h"
12 : #include "log.h"
13 : #include "common/aicpu_sqe_context.h"
14 : #include "aicpu_hccl_common.h"
15 : #include "transport_pub.h"
16 : #include "adapter_hal_pub.h"
17 : #include "dispatcher.h"
18 : #include "executor_tracer.h"
19 : #include "aicpu_hccl_process.h"
20 :
21 : namespace hccl {
22 : constexpr u64 MAX_RDMA_WQE_SIZE = 2ULL * 1024 * 1024 * 1024; // RDMA最大WQE限制是2GB
23 :
24 : std::shared_mutex HcclOneSideServiceAicpu::serviceMapMutex_;
25 : std::unordered_map<std::string, std::shared_ptr<HcclOneSideServiceAicpu>> HcclOneSideServiceAicpu::services_;
26 :
27 6 : HcclOneSideServiceAicpu::HcclOneSideServiceAicpu()
28 : {
29 6 : }
30 :
31 6 : HcclOneSideServiceAicpu::~HcclOneSideServiceAicpu()
32 : {
33 6 : CHK_PRT_CONT(ReportHcclTaskInfo() != HCCL_SUCCESS, HCCL_WARNING("[~] ReportHcclTaskInfo failed"));
34 6 : CHK_PRT_CONT(ClearStreamLocalBuff() != HCCL_SUCCESS, HCCL_WARNING("[~] ClearStreamLocalBuff failed"));
35 6 : rdmaLinks_.clear();
36 6 : if (dispatcher_ != nullptr) {
37 0 : HcclDispatcherDestroy(dispatcher_);
38 0 : dispatcher_ = nullptr;
39 : }
40 6 : }
41 :
42 2 : HcclResult HcclOneSideServiceAicpu::Process(const OpTilingData *tilingData)
43 : {
44 2 : std::string tag = tilingData->tag;
45 2 : const HcclCMDType cmdType = static_cast<HcclCMDType>(tilingData->opType);
46 2 : HCCL_DEBUG("[Process] Entry, tag[%s] cmdType[%u]", tag.c_str(), cmdType);
47 :
48 2 : const u8 *dynamicDataPtr = reinterpret_cast<const u8 *>(tilingData) + sizeof(OpTilingData);
49 2 : CHK_PRT_RET(tilingData->length < sizeof(OpTilingOneSideCommDataDes),
50 : HCCL_ERROR("[Process] dynamicDataSize[%llu] should be greater than or equal to "
51 : "OpTilingOneSideCommDataDes[%llu]", tilingData->length, sizeof(OpTilingOneSideCommDataDes)), HCCL_E_PARA);
52 1 : const auto *vDataPtr = reinterpret_cast<const OpTilingOneSideCommDataDes *>(dynamicDataPtr);
53 1 : if (vDataPtr->finalize) {
54 1 : std::unique_lock<std::shared_mutex> rwlock(serviceMapMutex_);
55 1 : services_.erase(tag);
56 1 : HCCL_INFO("[Finalize] tag[%s], services[%u]", tag.c_str(), services_.size());
57 1 : return HCCL_SUCCESS;
58 1 : }
59 :
60 0 : auto service = GetService(tag, tilingData);
61 0 : CHK_PRT_RET(service == nullptr, HCCL_ERROR("[Process] Service not found, tag[%s].", tag.c_str()), HCCL_E_INTERNAL);
62 0 : return service->DoProcess(tag, tilingData);
63 2 : }
64 :
65 2 : std::shared_ptr<HcclOneSideServiceAicpu> HcclOneSideServiceAicpu::GetService(const std::string &tag,
66 : const OpTilingData *tilingData)
67 : {
68 : {
69 2 : std::shared_lock<std::shared_mutex> rwlock(serviceMapMutex_);
70 2 : auto serviceIter = services_.find(tag);
71 2 : if (serviceIter != services_.cend()) {
72 1 : return serviceIter->second;
73 : }
74 2 : }
75 1 : std::shared_ptr<HcclOneSideServiceAicpu> service;
76 1 : EXCEPTION_CATCH(service = std::make_shared<HcclOneSideServiceAicpu>(), return nullptr);
77 1 : CHK_PRT_RET(service == nullptr, HCCL_ERROR("[GetService] Alloc failed, tag[%s].", tag.c_str()), nullptr);
78 1 : HcclResult ret = service->Init(tag, tilingData);
79 1 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[GetService] Init failed, tag[%s].", tag.c_str()), nullptr);
80 : {
81 0 : std::unique_lock<std::shared_mutex> rwlock(serviceMapMutex_);
82 0 : services_[tag] = service;
83 0 : }
84 0 : AicpuComContext *ctx = AicpuGetComContext();
85 0 : AicpuHcclProcess::CallMC2MaintenanceThread(ctx);
86 0 : return service;
87 1 : }
88 :
89 1 : HcclResult HcclOneSideServiceAicpu::Init(const std::string &tag, const OpTilingData *tilingData)
90 : {
91 1 : if (isInited_) {
92 0 : HCCL_WARNING("[Init] Already inited, tag[%s]", tag.c_str());
93 0 : return HCCL_SUCCESS;
94 : }
95 :
96 1 : const u8 *dynamicDataPtr = reinterpret_cast<const u8 *>(tilingData) + sizeof(OpTilingData);
97 1 : CHK_PRT_RET(tilingData->length < sizeof(OpTilingOneSideCommDataDes),
98 : HCCL_ERROR("[Init] dynamicDataSize[%llu] should be greater than or equal to OpTilingOneSideCommDataDes[%llu]",
99 : tilingData->length, sizeof(OpTilingOneSideCommDataDes)), HCCL_E_PARA);
100 1 : const auto *vDataPtr = reinterpret_cast<const OpTilingOneSideCommDataDes *>(dynamicDataPtr);
101 1 : CHK_PRT_RET(vDataPtr->commResParaSize != sizeof(HcclOneSideCommResParam),
102 : HCCL_ERROR("[Init] commResParaSize[%llu] should be equal to HcclOneSideCommResParam[%llu]",
103 : vDataPtr->commResParaSize, sizeof(HcclOneSideCommResParam)), HCCL_E_PARA);
104 0 : commResParaPtr_ = reinterpret_cast<const HcclOneSideCommResParam *>(vDataPtr->commResParaAddr);
105 0 : CHK_PTR_NULL(commResParaPtr_);
106 :
107 0 : identifier_ = tag;
108 0 : rankId_ = tilingData->srcRank;
109 0 : rankSize_ = vDataPtr->rankSize;
110 :
111 0 : const u32 hostDevId = commResParaPtr_->aicpuOpNotify[0].devId;
112 0 : CHK_RET(hrtDrvGetLocalDevIDByHostDevID(hostDevId, &devId_));
113 0 : CHK_RET(hrtHalGetDeviceType(devId_, devType_));
114 0 : CHK_PRT_RET(devType_ != DevType::DEV_TYPE_910_93 && devType_ != DevType::DEV_TYPE_910B,
115 : HCCL_ERROR("[Init] Expect devType[%u] is A2 or A3", devType_), HCCL_E_NOT_SUPPORT);
116 0 : CHK_RET(hrtHalGetDeviceInfo(devId_, MODULE_TYPE_SYSTEM, INFO_TYPE_PHY_CHIP_ID, &chipId_));
117 :
118 0 : s32 devLogicId = INVALID_INT;
119 0 : CHK_RET(hrtGetDevice(&devLogicId));
120 0 : if (devLogicId == INVALID_INT) { // standalone mode, run without ccl op
121 0 : CHK_RET(hrtSetlocalDevice(hostDevId));
122 0 : CHK_RET(hrtSetlocalDeviceType(devType_));
123 0 : CHK_RET(hrtSetLocalDeviceSatMode(static_cast<aclrtFloatOverflowMode>(tilingData->floatOverflowMode)));
124 : }
125 0 : logicDevId_ = hostDevId;
126 :
127 0 : CHK_RET(HcclDispatcherAicpuInit(&dispatcher_, devId_, SDMA_QOS_DEFAULT, DispatcherType::DISPATCHER_AICPU));
128 :
129 0 : CHK_RET(InitOpNotifyObj());
130 0 : CHK_RET(InitStream(execStream_, execComStreamInfo_, commResParaPtr_->execStreamParam, tag));
131 :
132 0 : CHK_RET(InitProfiling());
133 :
134 0 : isInited_ = true;
135 :
136 0 : HCCL_RUN_INFO("[Init] End. rankId[%u] hostDevId[%u] chipId[%u] devId[%u] streamId[%u] commResPara[%p]", rankId_,
137 : hostDevId, chipId_, devId_, execStream_.id(), commResParaPtr_);
138 :
139 0 : return HCCL_SUCCESS;
140 : }
141 :
142 0 : HcclResult HcclOneSideServiceAicpu::InitOpNotifyObj()
143 : {
144 0 : for (u32 i = 0; i < AICPU_OP_NOTIFY_MAX_NUM; i++) {
145 0 : const HcclSignalInfo &signalInfo = commResParaPtr_->aicpuOpNotify[i];
146 0 : CHK_PRT_RET(signalInfo.resId == INVALID_U64,
147 : HCCL_ERROR("[InitOpNotifyObj] resId[%llu] is invalid", signalInfo.resId),
148 : HCCL_E_PARA);
149 :
150 0 : std::shared_ptr<LocalNotify> notify;
151 0 : EXCEPTION_CATCH((notify = std::make_shared<LocalNotify>()), return HCCL_E_PTR);
152 0 : CHK_SMART_PTR_NULL(notify);
153 0 : CHK_RET(notify->Init(signalInfo, NotifyLoadType::DEVICE_NOTIFY));
154 0 : opNotifies_.push_back(notify);
155 0 : HCCL_INFO("[InitOpNotifyObj] tag[%s] resId[%llu] tsId[%u] devId[%u]",
156 : identifier_.c_str(), signalInfo.resId, signalInfo.tsId, signalInfo.devId);
157 0 : }
158 0 : return HCCL_SUCCESS;
159 : }
160 :
161 0 : HcclResult HcclOneSideServiceAicpu::InitStream(Stream &stream, HcclComStreamInfo &comStreamInfo,
162 : const HcclStreamParam &streamParam, const std::string &tag)
163 : {
164 0 : const HcclStreamInfo &streamInfo = streamParam.streamInfo;
165 0 : comStreamInfo.sqId = streamInfo.sqIds;
166 0 : comStreamInfo.actualStreamId = streamInfo.streamIds;
167 0 : comStreamInfo.logicCqId = streamInfo.logicCqids;
168 :
169 0 : u64 sqAddr = 0;
170 0 : CHK_RET(QuerySqBaseAddr(devId_, streamInfo.sqIds, sqAddr));
171 0 : comStreamInfo.sqBaseAddr = reinterpret_cast<void *>(sqAddr);
172 0 : CHK_PRT_RET(comStreamInfo.sqBaseAddr == nullptr, HCCL_ERROR("[Init] sqe base addr is nullptr."), HCCL_E_PTR);
173 0 : CHK_RET(QuerySqStatusByType(devId_, streamInfo.sqIds, DRV_SQCQ_PROP_SQ_DEPTH, comStreamInfo.sqDepth));
174 0 : u32 sqHead = 0;
175 0 : CHK_RET(QuerySqStatusByType(devId_, streamInfo.sqIds, DRV_SQCQ_PROP_SQ_HEAD, sqHead));
176 0 : u32 sqTail = 0;
177 0 : CHK_RET(QuerySqStatusByType(devId_, streamInfo.sqIds, DRV_SQCQ_PROP_SQ_TAIL, sqTail));
178 0 : HCCL_DEBUG("[Init] get stream data success, tag[%s], streamId[%d], sqId[%d], logicCqId[%u], sqDepth[%u], "
179 : "sqHead[%u], sqTail[%u]", tag.c_str(), comStreamInfo.actualStreamId, comStreamInfo.sqId,
180 : comStreamInfo.logicCqId, comStreamInfo.sqDepth, sqHead, sqTail);
181 0 : stream = Stream(comStreamInfo);
182 0 : u64 sqCqeContextSize = streamParam.sqCqContextSize;
183 0 : CHK_PRT_RET(sqCqeContextSize != sizeof(SqCqeContext),
184 : HCCL_ERROR("[%s] sqCqeContextSize[%llu] is not equal to sizeof(SqCqeContext)[%llu], tag[%s]", __func__,
185 : sqCqeContextSize, sizeof(SqCqeContext), tag.c_str()), HCCL_E_PARA);
186 0 : SqCqeContext *sqCqeContext = reinterpret_cast<SqCqeContext *>(streamParam.sqCqContextAddr);
187 0 : CHK_PRT_RET(sqCqeContext == nullptr,
188 : HCCL_ERROR("[%s] sqCqeContext[%llu] is nullptr, tag[%s]", __func__, streamParam.sqCqContextAddr, tag.c_str()),
189 : HCCL_E_PARA);
190 0 : CHK_RET(stream.InitSqAndCqeContext(sqHead, sqTail, sqCqeContext));
191 0 : HCCL_INFO("[%s] Create stream success, tag[%s], streamId[%u], devId[%u]", __func__, tag.c_str(), stream.id(),
192 : devId_);
193 :
194 0 : return HCCL_SUCCESS;
195 : }
196 :
197 0 : HcclResult HcclOneSideServiceAicpu::FillMemDetails(MemDetails &localMems, MemDetails &remoteMems,
198 : const HcclOneSideOpDescParam *descPtr, u32 index)
199 : {
200 0 : CHK_PTR_NULL(descPtr);
201 0 : const HcclDataType dataType = static_cast<HcclDataType>(descPtr[index].dataType);
202 0 : if (dataType_ == HcclDataType::HCCL_DATA_TYPE_RESERVED) {
203 0 : dataType_ = dataType;
204 : }
205 0 : const u32 perDataSize = DataUnitSize(dataType);
206 0 : CHK_PRT_RET(perDataSize == 0, HCCL_ERROR("[FillMemDetails] dataType[%u] DataUnitSize is 0", dataType), HCCL_E_PARA);
207 0 : const u64 count = descPtr[index].count;
208 0 : const u64 buffSize = count * perDataSize;
209 0 : totalCount_ += count;
210 0 : localMems.addr = descPtr[index].localAddr;
211 0 : localMems.size = buffSize;
212 0 : localMems.key = descPtr[index].lkey;
213 0 : remoteMems.addr = descPtr[index].remoteAddr;
214 0 : remoteMems.size = buffSize;
215 0 : remoteMems.key = descPtr[index].rkey;
216 0 : HCCL_DEBUG("[FillMemDetails] local addr[%#llx], remote addr[%#llx], size[%llu]", localMems.addr, remoteMems.addr,
217 : localMems.size);
218 0 : return HCCL_SUCCESS;
219 : }
220 :
221 0 : HcclResult HcclOneSideServiceAicpu::PrepareRdmaLink(u32 remoteRankId, const struct HcclQpInfoV2 &qpInfo)
222 : {
223 0 : if (rdmaLinks_.find(remoteRankId) == rdmaLinks_.end()) {
224 0 : const int UNIT_CONVERSION = 1000;
225 0 : linkTimeout_ = 4096ULL * (1 << qpInfo.retryTime) * (qpInfo.retryCnt + 1) / UNIT_CONVERSION; // RDMA超时基数是4.096us
226 0 : TransportMem::AttrInfo attrInfo{};
227 0 : attrInfo.localRankId = rankId_;
228 0 : attrInfo.remoteRankId = remoteRankId;
229 0 : attrInfo.timeout = linkTimeout_;
230 0 : std::shared_ptr<TransportMem> link;
231 0 : EXCEPTION_CATCH(link = TransportMem::Create(TransportMem::TpType::ROCE_DEVICE, qpInfo, dispatcher_, attrInfo),
232 : return HCCL_E_MEMORY);
233 0 : CHK_SMART_PTR_NULL(link);
234 0 : rdmaLinks_[remoteRankId] = link;
235 0 : HCCL_INFO("[Init] PrepareRdmaLink. rankId[%u] chipId[%u] devId[%u] remoteRankId[%u] linkTimeout[%u us]"
236 : "retryTime[%u] retryCnt[%u]", rankId_, chipId_, devId_, remoteRankId, linkTimeout_, qpInfo.retryTime,
237 : qpInfo.retryCnt);
238 0 : }
239 0 : return HCCL_SUCCESS;
240 : }
241 :
242 0 : HcclResult HcclOneSideServiceAicpu::DoProcess(const std::string &tag, const OpTilingData *tilingData)
243 : {
244 0 : const HcclCMDType cmdType = static_cast<HcclCMDType>(tilingData->opType);
245 0 : const u32 remoteRankId = tilingData->dstRank;
246 0 : HCCL_DEBUG("[DoProcess] Entry. tag[%s] cmdType[%u] remoteRankId[%u]", tag.c_str(), cmdType, remoteRankId);
247 :
248 0 : const u8 *dynamicDataPtr = reinterpret_cast<const u8 *>(tilingData) + sizeof(OpTilingData);
249 0 : CHK_PRT_RET(tilingData->length < sizeof(OpTilingOneSideCommDataDes),
250 : HCCL_ERROR("[DoProcess] dynamicDataSize[%llu] should be greater than or equal to "
251 : "OpTilingOneSideCommDataDes[%llu]", tilingData->length, sizeof(OpTilingOneSideCommDataDes)), HCCL_E_PARA);
252 0 : const auto *vDataPtr = reinterpret_cast<const OpTilingOneSideCommDataDes *>(dynamicDataPtr);
253 0 : CHK_PRT_RET(vDataPtr->commResParaSize != sizeof(HcclOneSideCommResParam),
254 : HCCL_ERROR("[DoProcess] commResParaSize[%llu] should be equal to HcclOneSideCommResParam[%llu]",
255 : vDataPtr->commResParaSize, sizeof(HcclOneSideCommResParam)), HCCL_E_PARA);
256 0 : const auto *commResParaPtr = reinterpret_cast<const HcclOneSideCommResParam *>(vDataPtr->commResParaAddr);
257 0 : CHK_PRT_RET(commResParaPtr != commResParaPtr_,
258 : HCCL_ERROR("[DoProcess] not support commResParaPtr[%p/%p] address update", commResParaPtr, commResParaPtr_),
259 : HCCL_E_PARA);
260 0 : LinkType linkType = static_cast<LinkType>(vDataPtr->linkType);
261 0 : CHK_PRT_RET((linkType != LinkType::LINK_ROCE) && (linkType != LinkType::LINK_HCCS),
262 : HCCL_ERROR("[DoProcess] not support linkType[%u]", vDataPtr->linkType), HCCL_E_PARA);
263 0 : const u32 descNum = vDataPtr->descNum; // Batch descNum + 1(signal)
264 0 : CHK_PRT_RET(vDataPtr->descDataLen != descNum * sizeof(HcclOneSideOpDescParam),
265 : HCCL_ERROR("[DoProcess] descDataLen[%llu] should be equal to "
266 : "descNum[%u] * sizeof(HcclOneSideOpDescParam)[%llu]", vDataPtr->descDataLen, descNum,
267 : sizeof(HcclOneSideOpDescParam)),
268 : HCCL_E_PARA);
269 0 : const auto *desc = reinterpret_cast<const HcclOneSideOpDescParam *>(
270 : dynamicDataPtr + sizeof(OpTilingOneSideCommDataDes));
271 :
272 0 : CHK_RET(WorkStart(cmdType, remoteRankId));
273 :
274 0 : if (linkType == LinkType::LINK_ROCE) {
275 0 : CHK_RET(DoRdmaProcess(cmdType, remoteRankId, vDataPtr, desc, descNum));
276 : } else {
277 0 : CHK_RET(DoSdmaProcess(cmdType, remoteRankId, vDataPtr, desc, descNum));
278 : }
279 :
280 0 : CHK_RET(WorkEnd(cmdType, remoteRankId));
281 0 : return HCCL_SUCCESS;
282 : }
283 :
284 0 : HcclResult HcclOneSideServiceAicpu::DoRdmaProcess(HcclCMDType cmdType, u32 remoteRankId,
285 : const OpTilingOneSideCommDataDes *vDataPtr, const HcclOneSideOpDescParam *desc, u32 descNum)
286 : {
287 0 : CHK_PRT_RET(vDataPtr->transportDataSize != sizeof(TransportDeviceNormalData),
288 : HCCL_ERROR("[DoProcess] transportDataSize[%llu] should be equal to TransportDeviceNormalData[%llu]",
289 : vDataPtr->transportDataSize, sizeof(TransportDeviceNormalData)), HCCL_E_PARA);
290 0 : const auto *transportDataPtr = reinterpret_cast<const TransportDeviceNormalData *>(vDataPtr->transportDataAddr);
291 0 : const TransportDeviceNormalData &ibvData = *transportDataPtr;
292 :
293 0 : CHK_RET(PrepareRdmaLink(remoteRankId, ibvData.qpInfo));
294 0 : auto link = rdmaLinks_[remoteRankId];
295 0 : CHK_SMART_PTR_NULL(link);
296 :
297 0 : const u32 userDescNum = descNum - 1;
298 0 : std::vector<MemDetails> localMems(userDescNum);
299 0 : std::vector<MemDetails> remoteMems(userDescNum);
300 0 : for (u32 index = 0; index < userDescNum; ++index) {
301 0 : CHK_RET(FillMemDetails(localMems[index], remoteMems[index], desc, index));
302 : }
303 0 : const bool isRead = (cmdType == HcclCMDType::HCCL_CMD_BATCH_GET);
304 0 : if (isRead) {
305 0 : CHK_RET(link->BatchRead(localMems, remoteMems, execStream_));
306 : } else {
307 0 : CHK_RET(link->BatchWrite(remoteMems, localMems, execStream_));
308 : }
309 :
310 : // fence signal at last
311 0 : MemDetails localFenceMem{};
312 0 : MemDetails remoteFenceMem{};
313 0 : CHK_RET(FillMemDetails(localFenceMem, remoteFenceMem, desc, descNum - 1));
314 0 : CHK_RET(link->AddOpFence(localFenceMem, remoteFenceMem, execStream_));
315 :
316 0 : HCCL_DEBUG("[DoProcess] End. tag[%s] cmdType[%u] remoteRankId[%u] transportData[%p] rdma desc[%p] "
317 : "descNum[%u]", identifier_.c_str(), cmdType, remoteRankId, transportDataPtr, desc, descNum);
318 0 : return HCCL_SUCCESS;
319 0 : }
320 :
321 0 : HcclResult HcclOneSideServiceAicpu::DoSdmaProcess(HcclCMDType cmdType, u32 remoteRankId,
322 : const OpTilingOneSideCommDataDes *vDataPtr, const HcclOneSideOpDescParam *desc, u32 descNum)
323 : {
324 0 : CHK_PTR_NULL(desc);
325 0 : u32 userDescNum = descNum - 1; // fence signal at last, but sdma needn't fence, so keep reserve
326 0 : for (u32 index = 0; index < userDescNum; ++index) {
327 0 : HcclDataType dataType = static_cast<HcclDataType>(desc[index].dataType);
328 0 : u64 dataSize = desc[index].count * DataUnitSize(dataType);
329 0 : DeviceMem localMem = DeviceMem::create(reinterpret_cast<void *>(desc[index].localAddr), dataSize);
330 0 : DeviceMem remoteMem = DeviceMem::create(reinterpret_cast<void *>(desc[index].remoteAddr), dataSize);
331 0 : if (cmdType == HcclCMDType::HCCL_CMD_BATCH_GET) {
332 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, localMem, remoteMem, execStream_, remoteRankId, LinkType::LINK_HCCS));
333 : } else {
334 0 : CHK_RET(HcclD2DMemcpyAsync(dispatcher_, remoteMem, localMem, execStream_, remoteRankId, LinkType::LINK_HCCS));
335 : }
336 0 : }
337 :
338 0 : CHK_RET(LocalNotify::Post(execStream_, dispatcher_, opNotifies_[1]));
339 0 : CHK_RET(LaunchTask(dispatcher_, const_cast<Stream &>(execStream_)));
340 :
341 0 : HCCL_DEBUG("[DoProcess] End. tag[%s] cmdType[%u] remoteRankId[%u] sdma desc[%p] descNum[%u]",
342 : identifier_.c_str(), cmdType, remoteRankId, desc, descNum);
343 0 : return HCCL_SUCCESS;
344 : }
345 :
346 0 : HcclResult HcclOneSideServiceAicpu::InitProfiling()
347 : {
348 0 : CHK_RET(RegisterLoadTaskCallBack(dispatcher_, nullptr, dfx::TaskProfilingCallBack));
349 0 : groupHashId_ = dfx::ProfilingManager::GetProfHashId(identifier_.c_str(), identifier_.length());
350 0 : HCCL_INFO("[InitProfiling]group[%s], groupHashId[%llu].", identifier_.c_str(), groupHashId_);
351 0 : dfx::ProfCommInfo profInfo{ groupHashId_, rankSize_, rankId_ };
352 0 : CHK_RET(dfx::ProfilingManager::AddProfInfoByStreamId(execStream_.id(), identifier_, profInfo));
353 0 : dfx::ProfilingExtendInfoHelper::InitProfItemId();
354 0 : return HCCL_SUCCESS;
355 : }
356 :
357 0 : HcclResult HcclOneSideServiceAicpu::WorkStart(HcclCMDType cmdType, u32 remoteRankId)
358 : {
359 0 : CHK_RET(ReportMainStreamTask(HEAD_TASK));
360 0 : CHK_RET(UpdateProfReportStartSqeIdx());
361 : // 刷新profiling开关, 支持profiling从中间迭代采集
362 0 : const bool profL0Open = dfx::ProfilingManager::IsProfL0On();
363 0 : const bool profL1Open = dfx::ProfilingManager::IsProfL1On();
364 0 : HCCL_INFO("[WorkStart] streamId[%u] tag[%s] cmdType[%u] remoteRankId[%u] profL0Open[%u/%u] profL1Open[%u/%u]",
365 : execStream_.id(), identifier_.c_str(), cmdType, remoteRankId, profL0Open,
366 : dfx::ProfilingManager::GetProfL0State(), profL1Open, dfx::ProfilingManager::GetProfL1State());
367 0 : return HCCL_SUCCESS;
368 : }
369 :
370 0 : HcclResult HcclOneSideServiceAicpu::WorkEnd(HcclCMDType cmdType, u32 remoteRankId)
371 : {
372 0 : CHK_RET(CombineReportOpInfo(cmdType, dataType_, totalCount_));
373 0 : dataType_ = HcclDataType::HCCL_DATA_TYPE_RESERVED;
374 0 : totalCount_ = 0;
375 0 : HCCL_DEBUG("[WorkEnd] streamId[%u] tag[%s] cmdType[%u] remoteRankId[%u]", execStream_.id(), identifier_.c_str(),
376 : cmdType, remoteRankId);
377 0 : CHK_RET(ReportMainStreamTask(TAIL_TASK));
378 0 : return HCCL_SUCCESS;
379 : }
380 :
381 0 : HcclResult HcclOneSideServiceAicpu::ReportMainStreamTask(u16 type)
382 : {
383 0 : HcclSqeContext *sqeContext = execStream_.GetSqeContextPtr();
384 0 : const SqeRingBuffer &sqeBuffer = sqeContext->buffer;
385 0 : u16 taskId = (type == TAIL_TASK) ? (sqeBuffer.tailSqeTaskId - 1) : sqeBuffer.tailSqeTaskId;
386 0 : return dfx::ProfilingManager::ReportMainStreamTask(execStream_, taskId, type);
387 : }
388 :
389 0 : HcclResult HcclOneSideServiceAicpu::UpdateProfReportStartSqeIdx()
390 : {
391 0 : if (dfx::ProfilingManager::IsL1fromOffToOn()) {
392 0 : HcclSqeContext *sqeContext = execStream_.GetSqeContextPtr();
393 0 : const SqeRingBuffer &sqeBuffer = sqeContext->buffer;
394 0 : CHK_RET(dfx::ProfilingManager::UpdateStartReportSqeIdx(execStream_.id(), sqeBuffer.tailSqeIdx));
395 : }
396 0 : return HCCL_SUCCESS;
397 : }
398 :
399 0 : HcclResult HcclOneSideServiceAicpu::CombineReportOpInfo(HcclCMDType cmdType, u8 dataType, u64 count)
400 : {
401 0 : MsprofAicpuHCCLOPInfo hcclOpInfo{};
402 0 : hcclOpInfo.dataType = dataType;
403 0 : hcclOpInfo.count = count;
404 0 : hcclOpInfo.groupName = groupHashId_;
405 0 : hcclOpInfo.ranksize = rankSize_;
406 0 : std::string typeStr = (cmdType == HcclCMDType::HCCL_CMD_BATCH_GET) ? "BatchGet" : "BatchPut";
407 0 : CHK_RET(dfx::ProfilingManager::ReportHcclOpInfo(hcclOpInfo, typeStr));
408 0 : return HCCL_SUCCESS;
409 0 : }
410 :
411 6 : HcclResult HcclOneSideServiceAicpu::ReportHcclTaskInfo()
412 : {
413 6 : return dfx::ProfilingManager::ReportTaskInfo(execStream_.id(), execStream_.GetSqeContextPtr());
414 : }
415 :
416 6 : HcclResult HcclOneSideServiceAicpu::ClearStreamLocalBuff()
417 : {
418 6 : CHK_RET(execStream_.ClearLocalBuff());
419 1 : return dfx::ProfilingManager::UpdateStartReportSqeIdx(execStream_.id(), 0);
420 : }
421 :
422 2 : HcclResult HcclOneSideServiceAicpu::CleanStreamFunc()
423 : {
424 2 : if (execStreamEnable_) {
425 0 : return HCCL_SUCCESS;
426 : }
427 2 : HCCL_RUN_INFO("Entry HcclOneSideServiceAicpu::CleanStreamFunc tag[%s]", identifier_.c_str());
428 2 : const HcclComStreamInfo &streamInfo = execStream_.GetHcclStreamInfo();
429 2 : CHK_RET(ConfigSqStatusByType(devId_, streamInfo.sqId, DRV_SQCQ_PROP_SQ_DISABLE_TO_ENABLE, 1));
430 1 : CHK_RET(CleanStream(execStream_));
431 1 : execStreamEnable_ = true;
432 1 : HCCL_RUN_INFO("Entry HcclOneSideServiceAicpu::CleanStreamFunc reset stream sq buffer success, "
433 : "SetStreamEnable streamid[%d]", streamInfo.actualStreamId);
434 1 : return HCCL_SUCCESS;
435 : }
436 :
437 1 : HcclResult HcclOneSideServiceAicpu::CleanAllStreamFunc()
438 : {
439 1 : HCCL_INFO("Entry HcclOneSideServiceAicpu::CleanAllStreamFunc");
440 1 : std::shared_lock<std::shared_mutex> rwlock(serviceMapMutex_);
441 1 : for (auto &serviceIter : services_) {
442 1 : HcclResult ret = serviceIter.second->CleanStreamFunc();
443 1 : if (ret != HCCL_SUCCESS) {
444 1 : return ret;
445 : }
446 : }
447 0 : return HCCL_SUCCESS;
448 1 : }
449 :
450 0 : HcclResult HcclOneSideServiceAicpu::DisableStreamFunc()
451 : {
452 0 : HCCL_INFO("Entry HcclOneSideServiceAicpu::DisableStreamFunc tag[%s]", identifier_.c_str());
453 0 : execStreamEnable_ = false;
454 0 : return HCCL_SUCCESS;
455 : }
456 :
457 1 : HcclResult HcclOneSideServiceAicpu::DisableAllStreamFunc()
458 : {
459 1 : HCCL_INFO("Entry HcclOneSideServiceAicpu::DisableAllStreamFunc");
460 1 : std::shared_lock<std::shared_mutex> rwlock(serviceMapMutex_);
461 1 : for (auto &serviceIter : services_) {
462 0 : HcclResult ret = serviceIter.second->DisableStreamFunc();
463 0 : if (ret != HCCL_SUCCESS) {
464 0 : return ret;
465 : }
466 : }
467 1 : return HCCL_SUCCESS;
468 1 : }
469 :
470 1 : HcclResult HcclOneSideServiceAicpu::CleanStream(Stream &stream)
471 : {
472 1 : CHK_RET(stream.ClearLocalBuff());
473 1 : CHK_RET(UpdateSqStatus(stream));
474 1 : HCCL_INFO("Entry HcclOneSideServiceAicpu::CleanStream %u success tag[%s]", stream.sqId(), identifier_.c_str());
475 1 : return HCCL_SUCCESS;
476 : }
477 :
478 1 : HcclResult HcclOneSideServiceAicpu::UpdateSqStatus(Stream &stream)
479 : {
480 1 : HcclSqeContext *sqeContext = stream.GetSqeContextPtr();
481 1 : CHK_PTR_NULL(sqeContext);
482 1 : SqeRingBuffer *sqeContextBuffer = &(sqeContext->buffer);
483 1 : auto &head = sqeContextBuffer->sqHead;
484 1 : auto &tail = sqeContextBuffer->sqTail;
485 :
486 1 : CHK_RET(QuerySqStatusByType(devId_, stream.sqId(), DRV_SQCQ_PROP_SQ_TAIL, head));
487 1 : CHK_RET(QuerySqStatusByType(devId_, stream.sqId(), DRV_SQCQ_PROP_SQ_HEAD, tail));
488 1 : HCCL_INFO("Entry HcclOneSideServiceAicpu::UpdateSqStatus, sqid:%u head:%u tail:%u tag[%s]",
489 : stream.sqId(), head, tail, identifier_.c_str());
490 1 : return HCCL_SUCCESS;
491 : }
492 :
493 8100 : HcclResult HcclOneSideServiceAicpu::HandleErrCqe()
494 : {
495 8100 : std::shared_lock<std::shared_mutex> rwlock(serviceMapMutex_);
496 8100 : for (auto &serviceIter : services_) {
497 0 : serviceIter.second->HandleCqeMessage(true);
498 : }
499 8100 : return HCCL_SUCCESS;
500 8100 : }
501 :
502 0 : void HcclOneSideServiceAicpu::HandleCqeMessage(bool isReadClear)
503 : {
504 : rtLogicCqReport_t cqeException;
505 0 : CqeStatus cqeStatus = CqeStatus::kDefault;
506 0 : PollCqeException(execStream_, isReadClear, cqeException, cqeStatus);
507 0 : }
508 :
509 0 : void HcclOneSideServiceAicpu::PollCqeException(Stream &stream, bool isReadClear, rtLogicCqReport_t &cqeException, CqeStatus &cqeStatus)
510 : {
511 0 : const HcclComStreamInfo &streamInfo = stream.GetHcclStreamInfo();
512 0 : bool isPollCqe = isReadClear;
513 0 : while (isPollCqe) {
514 : CqeQueryInput cqeQueryInput;
515 0 : dfx_tracer::ExecutorTracer::SetCqeQueryInput(devId_, streamInfo, cqeQueryInput);
516 0 : constexpr u32 reportSize = 256;
517 : rtLogicCqReport_t streamReport[reportSize];
518 0 : cqeQueryInput.cqeAddr = reinterpret_cast<uint8_t *>(streamReport);
519 0 : cqeStatus = CqReportRecv(cqeQueryInput, cqeException);
520 0 : isPollCqe = (cqeStatus == dfx::CqeStatus::kCqeException);
521 : }
522 0 : }
523 :
524 8101 : bool HcclOneSideServiceAicpu::isAllDestroy()
525 : {
526 8101 : std::shared_lock<std::shared_mutex> rwlock(serviceMapMutex_);
527 8101 : bool isEmpty = services_.empty();
528 8101 : return isEmpty;
529 8101 : }
530 : }
|
