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