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 <cstdint>
12 : #include <cstdlib>
13 : #include <cstring>
14 :
15 : #include "aicpu_task_cache_entry.h"
16 : #include "aicpu_task_utils.h"
17 :
18 : #include "aicpu_indop_env.h"
19 : #include "log.h"
20 : #include "rma_buffer_lite.h"
21 : #include "buffer.h"
22 : #include "rmt_rma_buf_slice_lite.h"
23 : #include "sqe_v82.h"
24 :
25 : namespace hcomm {
26 :
27 96 : AddrRefreshInfo::AddrRefreshInfo() : needRefresh(false), memIdx(0), offset(0) {}
28 :
29 4 : AddrRefreshInfo::AddrRefreshInfo(const uint32_t curMemIdx) : needRefresh(true), memIdx(curMemIdx), offset(0) {}
30 :
31 1 : AddrRefreshInfo::AddrRefreshInfo(const AddrRefreshInfo& other)
32 1 : : needRefresh(other.needRefresh),
33 1 : memIdx(other.memIdx),
34 1 : offset(other.offset)
35 1 : {}
36 :
37 101 : AddrRefreshInfo::~AddrRefreshInfo() {}
38 :
39 2 : const AddrRefreshInfo& AddrRefreshInfo::operator=(const AddrRefreshInfo& other)
40 : {
41 2 : if (this != &other) {
42 1 : needRefresh = other.needRefresh;
43 1 : offset = other.offset;
44 1 : memIdx = other.memIdx;
45 : }
46 2 : return *this;
47 : }
48 :
49 95 : AicpuTaskCacheEntry::AicpuTaskCacheEntry()
50 : {
51 95 : if ((UNLIKELY(GetPlfDebugConfigValue() & PLF_TASK)) || UNLIKELY(HcclCheckLogLevel(HCCL_LOG_DEBUG))) {
52 95 : isTaskConfigDebug_ = true;
53 : }
54 95 : }
55 :
56 95 : AicpuTaskCacheEntry::~AicpuTaskCacheEntry()
57 : {
58 95 : size_t sqeArrayCount = sqeArrayInfos_.size();
59 95 : size_t totalSqeCount = 0;
60 128 : for (size_t arrayIdx = 0; arrayIdx < sqeArrayCount; ++arrayIdx) {
61 33 : totalSqeCount += sqeArrayInfos_[arrayIdx].srcAddrRefreshInfoArray.size();
62 :
63 33 : uint8_t* curSqeArray = sqeArrayInfos_[arrayIdx].sqeArray;
64 33 : if (UNLIKELY(curSqeArray == nullptr)) {
65 0 : HCCL_ERROR("[AicpuTaskCacheEntry][~AicpuTaskCacheEntry] curSqeArray is nullptr");
66 : } else {
67 33 : free(curSqeArray);
68 33 : sqeArrayInfos_[arrayIdx].sqeArray = nullptr;
69 : }
70 : }
71 :
72 95 : size_t wqeArrayCount = wqeTaskArrayInfos_.size();
73 95 : size_t totalWqeCount = 0;
74 107 : for (size_t arrayIdx = 0; arrayIdx < wqeArrayCount; ++arrayIdx) {
75 12 : totalWqeCount += wqeTaskArrayInfos_[arrayIdx].wqeTaskArray.size();
76 : }
77 :
78 95 : HCCL_INFO(
79 : "[AicpuTaskCacheEntry][~AicpuTaskCacheEntry] release %u SQE arrays "
80 : "(%u SQEs in total) and %u WQE arrays (%u WQEs in total) from a cache entry",
81 : sqeArrayCount, totalSqeCount, wqeArrayCount, totalWqeCount);
82 95 : }
83 :
84 : HcclResult
85 66 : AicpuTaskCacheEntry::InitCacheEntry(const uint64_t* baseAddrs, const uint64_t* memSizes, const uint32_t count)
86 : {
87 : // 校验count (当前rank的userIn和userOut)
88 66 : constexpr uint64_t ADDRS_COUNT = 2;
89 66 : CHK_PRT_RET(
90 : count != ADDRS_COUNT,
91 : HCCL_ERROR("[AicpuTaskCacheEntry][%s] count[%llu] != ADDRS_COUNT[%llu]", __func__, count, ADDRS_COUNT),
92 : HCCL_E_PARA);
93 :
94 : // 空指针及溢出检查
95 189 : for (uint32_t i = 0; i < count; i++) {
96 126 : CHK_PRT_RET(
97 : baseAddrs[i] == 0,
98 : HCCL_ERROR("[AicpuTaskCacheEntry][InitCacheEntry] baseAddrs[%u] is 0, memSize[%llu]", i, memSizes[i]),
99 : HCCL_E_PARA);
100 126 : CHK_PRT_RET(
101 : baseAddrs[i] + memSizes[i] < baseAddrs[i],
102 : HCCL_ERROR("[AicpuTaskCacheEntry][InitCacheEntry] baseAddrs[%u] + memSizes[%u] overflows", i, i),
103 : HCCL_E_PARA);
104 : }
105 :
106 : // 缓存baseAddrs
107 63 : CHK_PRT_RET(
108 : cachedBaseAddrs_.size() != 0,
109 : HCCL_ERROR("[AicpuTaskCacheEntry][%s] cachedBaseAddrs_.size[%u] != 0", __func__, cachedBaseAddrs_.size()),
110 : HCCL_E_INTERNAL);
111 62 : cachedBaseAddrs_.resize(count);
112 62 : entryBytes_ += count * sizeof(uint64_t);
113 62 : CHK_SAFETY_FUNC_RET(
114 : memcpy_s(cachedBaseAddrs_.data(), count * sizeof(uint64_t), baseAddrs, count * sizeof(uint64_t)));
115 :
116 : // 缓存sizes
117 62 : CHK_PRT_RET(
118 : cachedMemSizes_.size() != 0,
119 : HCCL_ERROR("[AicpuTaskCacheEntry][%s] cachedMemSizes_.size[%u] != 0", __func__, cachedMemSizes_.size()),
120 : HCCL_E_INTERNAL);
121 62 : cachedMemSizes_.resize(count);
122 62 : entryBytes_ += count * sizeof(uint64_t);
123 62 : CHK_SAFETY_FUNC_RET(memcpy_s(cachedMemSizes_.data(), count * sizeof(uint64_t), memSizes, count * sizeof(uint64_t)));
124 :
125 62 : return HCCL_SUCCESS;
126 : }
127 :
128 37 : HcclResult AicpuTaskCacheEntry::AddSqeArray(
129 : RtsqA5* rtsqPtr, AicpuTsThread* aicpuTsThreadPtr, const uint64_t sqeCount, const uint8_t* sqeArray,
130 : const uint32_t streamId)
131 : {
132 37 : CHK_PTR_NULL(rtsqPtr);
133 36 : CHK_PTR_NULL(aicpuTsThreadPtr);
134 35 : CHK_PTR_NULL(sqeArray);
135 34 : CHK_PRT_RET(sqeCount == 0, HCCL_ERROR("[AicpuTaskCacheEntry][AddSqeArray] sqeCount is 0"), HCCL_E_INTERNAL);
136 :
137 33 : const size_t sqeBytes = sqeCount * AC_SQE_SIZE;
138 33 : uint8_t* newSqeArray = reinterpret_cast<uint8_t*>(malloc(sqeBytes));
139 33 : CHK_PTR_NULL(newSqeArray);
140 33 : HcclResult ret = AddSqeArray_(newSqeArray, sqeBytes, sqeArray, streamId);
141 33 : if (ret != HCCL_SUCCESS) {
142 0 : free(newSqeArray);
143 0 : return ret;
144 : }
145 :
146 : // 更新SQE数组
147 33 : sqeArrayInfos_.emplace_back();
148 33 : SqeArrayInfo& sqeArrayInfo = sqeArrayInfos_.back();
149 33 : sqeArrayInfo.sqeArray = newSqeArray;
150 33 : sqeArrayInfo.rtsqPtr = rtsqPtr;
151 33 : sqeArrayInfo.aicpuTsThreadPtr = aicpuTsThreadPtr;
152 33 : sqeArrayInfo.sqeCount = sqeCount;
153 33 : sqeArrayInfo.srcAddrRefreshInfoArray.resize(sqeCount);
154 33 : sqeArrayInfo.dstAddrRefreshInfoArray.resize(sqeCount);
155 33 : entryBytes_ += sqeArrayInfo.GetSize();
156 :
157 : // 更新下发顺序
158 33 : launchOrder_.emplace_back(TaskArrayType::kTaskArrayTypeSqe);
159 33 : entryBytes_ += sizeof(TaskArrayType);
160 :
161 33 : HCCL_INFO(
162 : "[AicpuTaskCacheEntry][AddSqeArray] add %uth sqe array with sqeCount[%llu] streamId[%u]",
163 : sqeArrayInfos_.size() - 1, sqeCount, streamId);
164 :
165 33 : return HCCL_SUCCESS;
166 : }
167 :
168 17 : HcclResult AicpuTaskCacheEntry::AddWqeArray(
169 : UbConnLite* ubConnLitePtr, UbTransportLiteImpl* ubTransportLiteImplPtr, const vector<WqeTask>& wqeTasks,
170 : const uint32_t streamId, const uint32_t dbSqeIdx, const bool isReportTask, const DbSqeProfInfo& dbSqeProfInfo)
171 : {
172 17 : CHK_PRT_RET(
173 : isReportTask != dbSqeProfInfo.isValid,
174 : HCCL_ERROR(
175 : "[AicpuTaskCacheEntry][AddWqeArray] isReportTask[%d] != dbSqeProfInfo.isValid[%d]", isReportTask,
176 : dbSqeProfInfo.isValid),
177 : HCCL_E_INTERNAL);
178 15 : CHK_PTR_NULL(ubConnLitePtr);
179 14 : CHK_PTR_NULL(ubTransportLiteImplPtr);
180 :
181 13 : const uint32_t wqeCount = wqeTasks.size();
182 13 : CHK_PRT_RET(wqeCount == 0, HCCL_ERROR("[AicpuTaskCacheEntry][AddWqeArray] wqeCount is 0"), HCCL_E_INTERNAL);
183 :
184 : // 更新streamIdToDbSqeTmpInfoMap_ (仅cache miss时临时维护, 展开完成后一定为空)
185 12 : DbSqeTmpInfo dbSqeTmpInfo;
186 12 : dbSqeTmpInfo.wqeArrayIdx = wqeTaskArrayInfos_.size();
187 12 : dbSqeTmpInfo.dbSqeIdx = dbSqeIdx;
188 12 : dbSqeTmpInfo.isReportTask = isReportTask;
189 12 : if (isReportTask) {
190 2 : dbSqeTmpInfo.dbSqeProfInfo = dbSqeProfInfo;
191 : }
192 12 : std::unordered_map<uint32_t, vector<DbSqeTmpInfo>>::iterator mapIter = streamIdToDbSqeTmpInfoMap_.find(streamId);
193 12 : if (mapIter == streamIdToDbSqeTmpInfoMap_.end()) {
194 12 : mapIter = streamIdToDbSqeTmpInfoMap_.emplace(streamId, vector<DbSqeTmpInfo>()).first;
195 : }
196 12 : mapIter->second.push_back(dbSqeTmpInfo);
197 :
198 : // 更新WQE数组
199 12 : wqeTaskArrayInfos_.emplace_back();
200 12 : WqeTaskArrayInfo& wqeTaskArrayInfo = wqeTaskArrayInfos_.back();
201 12 : wqeTaskArrayInfo.wqeTaskArray = wqeTasks;
202 12 : wqeTaskArrayInfo.ubConnLitePtr = ubConnLitePtr;
203 12 : wqeTaskArrayInfo.ubTransportLiteImplPtr = ubTransportLiteImplPtr;
204 12 : wqeTaskArrayInfo.dbSqeLocation.dbSqeIdx = dbSqeIdx;
205 12 : wqeTaskArrayInfo.locAddrRefreshInfoArray.resize(wqeCount);
206 12 : wqeTaskArrayInfo.rmtAddrRefreshInfoArray.resize(wqeCount);
207 12 : entryBytes_ += wqeTaskArrayInfo.GetSize();
208 :
209 : // 更新下发顺序
210 12 : launchOrder_.emplace_back(TaskArrayType::kTaskArrayTypeWqe);
211 12 : entryBytes_ += sizeof(TaskArrayType);
212 :
213 : // 初始化token info
214 12 : if (tokenInfosMap_.find(ubTransportLiteImplPtr) == tokenInfosMap_.end()) {
215 12 : const uint64_t addrCnt = cachedBaseAddrs_.size();
216 24 : tokenInfosMap_.emplace(ubTransportLiteImplPtr, vector<TokenInfo>(addrCnt));
217 12 : entryBytes_ += (sizeof(UbTransportLiteImplHandle) + addrCnt * sizeof(TokenInfo));
218 : }
219 :
220 12 : HCCL_INFO(
221 : "[AicpuTaskCacheEntry][AddWqeArray] add %uth wqe array with wqeCount[%u] dbSqeTmpInfo[%u, %d, %u] streamId[%u]",
222 : wqeTaskArrayInfos_.size() - 1, wqeCount, dbSqeTmpInfo.wqeArrayIdx, dbSqeTmpInfo.isReportTask,
223 : dbSqeTmpInfo.dbSqeIdx, streamId);
224 :
225 12 : return HCCL_SUCCESS;
226 : }
227 :
228 31 : HcclResult AicpuTaskCacheEntry::SubmitCacheEntry()
229 : {
230 : // 校验streamIdToDbSqeTmpInfoMap_
231 : // 注意: SubmitCacheEntry时, 算子展开一定完成且task全部launch, 因此所有DbSqe的临时信息一定均已消耗
232 31 : if (UNLIKELY(streamIdToDbSqeTmpInfoMap_.size() != 0)) {
233 1 : HCCL_ERROR(
234 : "[AicpuTaskCacheEntry][SubmitCacheEntry] streamIdToDbSqeTmpInfoMap_.size[%u] != 0",
235 : streamIdToDbSqeTmpInfoMap_.size());
236 2 : for (const auto& mapIter : streamIdToDbSqeTmpInfoMap_) {
237 1 : HCCL_ERROR(
238 : "[AicpuTaskCacheEntry][SubmitCacheEntry] streamId[%u] wqeArrayIdx[%u] dbSqeIdx[%u] isReportTask[%d]",
239 : mapIter.first, mapIter.second[0].wqeArrayIdx, mapIter.second[0].dbSqeIdx,
240 : mapIter.second[0].isReportTask);
241 : }
242 1 : return HCCL_E_INTERNAL;
243 : }
244 :
245 : // 地址信息已经通过InitCacheEntry保存
246 30 : const uint64_t count = cachedBaseAddrs_.size();
247 30 : CHK_PRT_RET(
248 : count == 0, HCCL_ERROR("[AicpuTaskCacheEntry][SubmitCacheEntry] cachedBaseAddrs_.size is 0"), HCCL_E_INTERNAL);
249 :
250 : // 打印dynamic memory ranges
251 28 : if (UNLIKELY(HcclCheckLogLevel(HCCL_LOG_INFO))) {
252 84 : for (size_t memIdx = 0; memIdx < count; memIdx++) {
253 56 : HCCL_INFO(
254 : "[AicpuTaskCacheEntry][SubmitCacheEntry] memRanges[%u]: [0x%016llx, 0x%016llx)", memIdx,
255 : cachedBaseAddrs_[memIdx], cachedBaseAddrs_[memIdx] + cachedMemSizes_[memIdx]);
256 : }
257 : }
258 :
259 28 : CHK_RET(SubmitSqeAddrRefreshInfo_());
260 27 : CHK_RET(SubmitWqeAddrRefreshInfoAndTokenInfo_());
261 26 : CHK_RET(SubmitDbSqeProfRefreshInfo_());
262 26 : CHK_RET(ValidateLaunchOrder_());
263 :
264 26 : return HCCL_SUCCESS;
265 : }
266 :
267 28 : inline HcclResult AicpuTaskCacheEntry::SubmitSqeAddrRefreshInfo_()
268 : {
269 : // 更新每段SQE数组的AddrRefreshInfo
270 55 : for (size_t arrayIdx = 0; arrayIdx < sqeArrayInfos_.size(); arrayIdx++) {
271 28 : SqeArrayInfo& sqeArrayInfo = sqeArrayInfos_[arrayIdx];
272 28 : const uint64_t sqeCount = sqeArrayInfo.sqeCount;
273 28 : const uint8_t* sqePtr = sqeArrayInfo.sqeArray;
274 28 : CHK_PTR_NULL(sqePtr);
275 57 : for (size_t sqeIdx = 0; sqeIdx < sqeCount; sqeIdx++) {
276 30 : CHK_RET(UpdateSqeAddrRefreshInfo_(
277 : sqePtr, sqeArrayInfo.srcAddrRefreshInfoArray[sqeIdx], sqeArrayInfo.dstAddrRefreshInfoArray[sqeIdx]));
278 :
279 : // 打印SQE AddrRefreshInfo
280 29 : HCCL_INFO(
281 : "[AicpuTaskCacheEntry][SubmitCacheEntry] sqeArrayInfos_[%u][%u]: "
282 : "srcAddrRefreshInfo[needRefresh-%d memIdx-%u] dstAddrRefreshInfo[needRefresh-%d memIdx-%u]",
283 : arrayIdx, sqeIdx, sqeArrayInfo.srcAddrRefreshInfoArray[sqeIdx].needRefresh,
284 : sqeArrayInfo.srcAddrRefreshInfoArray[sqeIdx].memIdx,
285 : sqeArrayInfo.dstAddrRefreshInfoArray[sqeIdx].needRefresh,
286 : sqeArrayInfo.dstAddrRefreshInfoArray[sqeIdx].memIdx);
287 29 : sqePtr += AC_SQE_SIZE;
288 : }
289 : }
290 27 : return HCCL_SUCCESS;
291 : }
292 :
293 27 : inline HcclResult AicpuTaskCacheEntry::SubmitWqeAddrRefreshInfoAndTokenInfo_()
294 : {
295 : // 更新每段WQE数组的AddrRefreshInfo, 并更新token info
296 34 : for (size_t arrayIdx = 0; arrayIdx < wqeTaskArrayInfos_.size(); arrayIdx++) {
297 : // 获取token info
298 8 : WqeTaskArrayInfo& wqeTaskArrayInfo = wqeTaskArrayInfos_[arrayIdx];
299 8 : UbTransportLiteImpl* ubTransportLiteImplPtr
300 : = wqeTaskArrayInfo.ubTransportLiteImplPtr; // 注意: AddWqeArray时已校验非空
301 : std::unordered_map<UbTransportLiteImplHandle, vector<TokenInfo>>::iterator iter
302 8 : = tokenInfosMap_.find(ubTransportLiteImplPtr);
303 :
304 9 : CHK_PRT_RET(
305 : iter == tokenInfosMap_.end(),
306 : HCCL_ERROR(
307 : "[AicpuTaskCacheEntry][SubmitCacheEntry] ubTransportLiteImplPtr[%p] not found in tokenInfosMap_",
308 : ubTransportLiteImplPtr),
309 : HCCL_E_INTERNAL);
310 8 : vector<TokenInfo>& tokenInfos = iter->second;
311 :
312 : // 更新AddrRefreshInfo和token info
313 8 : const uint32_t wqeCount = wqeTaskArrayInfo.wqeTaskArray.size();
314 8 : vector<WqeTask>& wqeTasks = wqeTaskArrayInfo.wqeTaskArray;
315 15 : for (size_t wqeIdx = 0; wqeIdx < wqeCount; wqeIdx++) {
316 8 : CHK_RET(UpdateWqeAddrRefreshInfoAndTokenInfo_(
317 : wqeTasks[wqeIdx], wqeTaskArrayInfo.locAddrRefreshInfoArray[wqeIdx],
318 : wqeTaskArrayInfo.rmtAddrRefreshInfoArray[wqeIdx], tokenInfos));
319 :
320 : // 打印WQE AddrRefreshInfo
321 7 : HCCL_INFO(
322 : "[AicpuTaskCacheEntry][SubmitCacheEntry] wqeTaskArrayInfos_[%u][%u]: "
323 : "srcAddrRefreshInfo[needRefresh-%d memIdx-%u] needLocTokenIdFlag[%d] "
324 : "dstAddrRefreshInfo[needRefresh-%d memIdx-%u] needRmtTokenIdAndValueFlag[%d]",
325 : arrayIdx, wqeIdx, wqeTaskArrayInfo.locAddrRefreshInfoArray[wqeIdx].needRefresh,
326 : wqeTaskArrayInfo.locAddrRefreshInfoArray[wqeIdx].memIdx,
327 : tokenInfos[wqeTaskArrayInfo.locAddrRefreshInfoArray[wqeIdx].memIdx].needLocTokenIdFlag,
328 : wqeTaskArrayInfo.rmtAddrRefreshInfoArray[wqeIdx].needRefresh,
329 : wqeTaskArrayInfo.rmtAddrRefreshInfoArray[wqeIdx].memIdx,
330 : tokenInfos[wqeTaskArrayInfo.rmtAddrRefreshInfoArray[wqeIdx].memIdx].needRmtTokenIdAndValueFlag);
331 : }
332 : }
333 26 : return HCCL_SUCCESS;
334 : }
335 :
336 26 : inline HcclResult AicpuTaskCacheEntry::SubmitDbSqeProfRefreshInfo_()
337 : {
338 : // 更新每个DbSqeProfAndRefreshInfo的AddrRefreshInfo
339 26 : for (std::unordered_map<DbSqeLocation, DbSqeProfAndRefreshInfo>::iterator iter = dbSqeLocInfoMap_.begin();
340 26 : iter != dbSqeLocInfoMap_.end(); iter++) {
341 0 : DbSqeProfAndRefreshInfo& dbSqeProfAndRefreshInfo = iter->second;
342 0 : const DbSqeProfInfo& dbSqeProfInfo = dbSqeProfAndRefreshInfo.dbSqeProfInfo;
343 0 : switch (static_cast<u8>(dbSqeProfInfo.taskParamType)) {
344 0 : case TaskParamTypeVal::TASK_UB:
345 : case TaskParamTypeVal::TASK_UB_REDUCE_INLINE:
346 : case TaskParamTypeVal::TASK_WRITE_WITH_NOTIFY:
347 : case TaskParamTypeVal::TASK_WRITE_REDUCE_WITH_NOTIFY:
348 0 : CHK_RET(UpdateAddrRefreshInfo_(dbSqeProfInfo.srcAddr, dbSqeProfAndRefreshInfo.srcAddrRefreshInfo));
349 0 : CHK_RET(UpdateAddrRefreshInfo_(dbSqeProfInfo.dstAddr, dbSqeProfAndRefreshInfo.dstAddrRefreshInfo));
350 0 : break;
351 0 : case TaskParamTypeVal::TASK_UB_INLINE_WRITE:
352 0 : CHK_RET(UpdateAddrRefreshInfo_(dbSqeProfInfo.dstAddr, dbSqeProfAndRefreshInfo.dstAddrRefreshInfo));
353 0 : break;
354 0 : default:
355 0 : HCCL_ERROR(
356 : "[AicpuTaskCacheEntry][%s] invalid taskParamType[%u]", __func__, dbSqeProfInfo.taskParamType);
357 0 : return HCCL_E_INTERNAL;
358 : }
359 : }
360 26 : return HCCL_SUCCESS;
361 : }
362 :
363 26 : inline HcclResult AicpuTaskCacheEntry::ValidateLaunchOrder_()
364 : {
365 : // 按照缓存时的下发顺序, 校验launchOrder_对应的sqe/wqeArrayIdx, 避免后续命中时重复校验
366 26 : size_t sqeArrayIdx = 0;
367 26 : size_t wqeArrayIdx = 0;
368 59 : for (size_t launchIdx = 0; launchIdx < launchOrder_.size(); launchIdx++) {
369 33 : const TaskArrayType taskType = launchOrder_[launchIdx];
370 33 : if (taskType == TaskArrayType::kTaskArrayTypeSqe) {
371 : // 校验arrayIdx
372 26 : CHK_PRT_RET(
373 : sqeArrayIdx >= sqeArrayInfos_.size(),
374 : HCCL_ERROR(
375 : "[AicpuTaskCacheEntry][SubmitCacheEntry] sqeArrayIdx[%u] >= sqeArrayInfos_.size[%u]", sqeArrayIdx,
376 : sqeArrayInfos_.size()),
377 : HCCL_E_PARA);
378 26 : ++sqeArrayIdx;
379 7 : } else if (taskType == TaskArrayType::kTaskArrayTypeWqe) {
380 : // 校验arrayIdx
381 7 : CHK_PRT_RET(
382 : wqeArrayIdx >= wqeTaskArrayInfos_.size(),
383 : HCCL_ERROR(
384 : "[AicpuTaskCacheEntry][SubmitCacheEntry] wqeArrayIdx[%u] >= wqeTaskArrayInfos_.size[%u]",
385 : wqeArrayIdx, wqeTaskArrayInfos_.size()),
386 : HCCL_E_PARA);
387 7 : ++wqeArrayIdx;
388 : } else {
389 0 : HCCL_ERROR("[AicpuTaskCacheEntry][SubmitCacheEntry] invalid task array type");
390 0 : return HCCL_E_INTERNAL;
391 : }
392 : }
393 26 : return HCCL_SUCCESS;
394 : }
395 :
396 : HcclResult
397 4 : AicpuTaskCacheEntry::RefreshAndLaunch(const uint64_t* baseAddrs, const uint64_t* memSizes, const uint32_t count)
398 : {
399 : // 校验count
400 4 : CHK_PRT_RET(
401 : cachedBaseAddrs_.size() != count || cachedMemSizes_.size() != count,
402 : HCCL_ERROR(
403 : "[AicpuTaskCacheEntry][RefreshAndLaunch] cachedBaseAddrs_.size[%u] / cachedMemSizes_.size[%u] != count[%u]",
404 : cachedBaseAddrs_.size(), cachedMemSizes_.size(), count),
405 : HCCL_E_INTERNAL);
406 :
407 : // 校验memSizes
408 3 : for (uint32_t memIdx = 0; memIdx < count; memIdx++) {
409 2 : CHK_PRT_RET(
410 : cachedMemSizes_[memIdx] != memSizes[memIdx],
411 : HCCL_ERROR("[AicpuTaskCacheEntry][RefreshAndLaunch] cachedMemSizes_[%u] != memSizes[%u]", memIdx, memIdx),
412 : HCCL_E_INTERNAL);
413 : }
414 :
415 1 : CHK_RET(RefreshTokenInfos_(baseAddrs, memSizes, count));
416 :
417 1 : const bool enableTaskException = hcomm::GetTaskExceptionEnable();
418 1 : const bool l1State = Hccl::DfxProfilingHandlerLite::GetInstance().GetProfL1State();
419 1 : const bool needTaskParam = (l1State || enableTaskException);
420 1 : HCCL_INFO(
421 : "[AicpuTaskCacheEntry][RefreshAndLaunch] l1State[%d] enableTaskException[%d] -> needTaskParam[%d]", l1State,
422 : enableTaskException, needTaskParam);
423 :
424 1 : CHK_RET(LaunchTasksByOrder_(baseAddrs, memSizes, count, needTaskParam));
425 :
426 1 : if (UNLIKELY(HcclCheckLogLevel(HCCL_LOG_INFO))) {
427 1 : CHK_RET(PrintRefreshResult_(baseAddrs, memSizes, count));
428 : }
429 :
430 1 : return HCCL_SUCCESS;
431 : }
432 :
433 : inline HcclResult
434 1 : AicpuTaskCacheEntry::RefreshTokenInfos_(const uint64_t* baseAddrs, const uint64_t* memSizes, const uint32_t count)
435 : {
436 : // 按需统一获取新传入内存地址的token信息用于WQE刷新, 避免重复构造Buffer
437 1 : for (std::unordered_map<UbTransportLiteImplHandle, vector<TokenInfo>>::iterator iter = tokenInfosMap_.begin();
438 2 : iter != tokenInfosMap_.end(); iter++) {
439 1 : vector<TokenInfo>& tokenInfos = iter->second;
440 1 : CHK_PRT_RET(
441 : tokenInfos.size() != count,
442 : HCCL_ERROR(
443 : "[AicpuTaskCacheEntry][RefreshAndLaunch] tokenInfos.size[%u] != count[%u]", tokenInfos.size(), count),
444 : HCCL_E_INTERNAL);
445 :
446 : // 注意: AddWqeArray时已校验非空, 无需重复校验
447 1 : UbTransportLiteImpl* ubTransportLitePtr = reinterpret_cast<UbTransportLiteImpl*>(iter->first);
448 3 : for (uint32_t memIdx = 0; memIdx < count; memIdx++) {
449 2 : const uint64_t baseAddr = baseAddrs[memIdx];
450 2 : const uint64_t memSize = memSizes[memIdx];
451 2 : TokenInfo& tokenInfo = tokenInfos[memIdx];
452 :
453 2 : if (tokenInfo.needLocTokenIdFlag) {
454 : // 参考hccl_api_data_aicpu_ts.cc (例如HcommWriteOnThread), 获取新loc token id
455 : // 注意: 无需通过ubTransportLitePtr->GetRmaBufSlicelite(locRmaBuf)构造Hccl::RmaBufSliceLite
456 : // locRmaBufSlicelite,
457 : // 再调用locRmaBufSlicelite.GetTokenId()获取token id (一定与Hccl::RmaBufferLite locRmaBuf的token
458 : // id相同)
459 1 : Hccl::RmaBufferLite locRmaBuf;
460 1 : CHK_RET(ubTransportLitePtr->BuildLocRmaBufferLite(
461 : reinterpret_cast<uintptr_t>(baseAddr), memSize, locRmaBuf));
462 1 : tokenInfo.locTokenId = locRmaBuf.GetTokenId();
463 : }
464 :
465 2 : if (tokenInfo.needRmtTokenIdAndValueFlag) {
466 : // 参考hccl_api_data_aicpu_ts.cc (例如HcommRead/Write/WriteReduce/WriteWithNotifyOnThread), 获取新rmt
467 : // token id/value 注意: Hccl::Buffer本身不含token id/value,
468 : // 必须通过调用ubTransportLitePtr->GetRmtRmaBufSliceLite,
469 : // 构造Hccl::RmtRmaBufSliceLite, 再调用GetTokenId/Value获取token id/value
470 1 : const Hccl::Buffer rmtBuf{reinterpret_cast<uintptr_t>(baseAddr), memSize};
471 1 : Hccl::RmtRmaBufSliceLite rmtRmaBufSlicelite = ubTransportLitePtr->GetRmtRmaBufSliceLite(rmtBuf);
472 1 : tokenInfo.rmtTokenId = rmtRmaBufSlicelite.GetTokenId();
473 1 : tokenInfo.rmtTokenValue = rmtRmaBufSlicelite.GetTokenValue();
474 1 : }
475 : }
476 : }
477 1 : return HCCL_SUCCESS;
478 : }
479 :
480 1 : inline HcclResult AicpuTaskCacheEntry::LaunchTasksByOrder_(
481 : const uint64_t* baseAddrs, const uint64_t* memSizes, const uint32_t count, bool needTaskParam)
482 : {
483 : // 按照缓存时的下发顺序, 依次刷新并下发task
484 1 : size_t sqeArrayIdx = 0;
485 1 : size_t wqeArrayIdx = 0;
486 1 : for (size_t launchIdx = 0; launchIdx < launchOrder_.size(); launchIdx++) {
487 0 : const TaskArrayType taskType = launchOrder_[launchIdx];
488 0 : if (taskType == TaskArrayType::kTaskArrayTypeSqe) {
489 : // 注意: sqeArrayIdx已在SubmitCacheEntry校验, 无需重复校验
490 0 : const SqeArrayInfo& sqeArrayInfo = sqeArrayInfos_[sqeArrayIdx];
491 :
492 : // 刷新SQE数组
493 0 : CHK_RET(RefreshSqeTasks_(sqeArrayInfo, baseAddrs));
494 :
495 : // 下发SQE
496 0 : CHK_RET(LaunchSqeTasks_(sqeArrayInfo));
497 :
498 : // 对刷新后的SQE填充DfxTaskInfo并经NextTaskSlot上报
499 0 : if (needTaskParam) {
500 0 : CHK_RET(ReportSqeArrayProfiling_(sqeArrayIdx, baseAddrs, memSizes, count));
501 : }
502 0 : ++sqeArrayIdx;
503 0 : } else if (taskType == TaskArrayType::kTaskArrayTypeWqe) {
504 : // 注意: wqeArrayIdx已在SubmitCacheEntry校验, 无需重复校验
505 0 : WqeTaskArrayInfo& wqeTaskArrayInfo = wqeTaskArrayInfos_[wqeArrayIdx];
506 :
507 : // 刷新WQE数组
508 0 : CHK_RET(RefreshWqeTasks_(wqeTaskArrayInfo, baseAddrs, memSizes, count));
509 :
510 : // 下发WQE数组
511 0 : CHK_RET(LaunchWqeTasks_(wqeTaskArrayInfo));
512 :
513 : // 刷新对应DbSqe
514 0 : CHK_RET(RefreshDbSqe_(wqeTaskArrayInfo));
515 0 : ++wqeArrayIdx;
516 : } else {
517 0 : HCCL_ERROR("[AicpuTaskCacheEntry][RefreshAndLaunch] invalid task array type[%u]", taskType);
518 0 : return HCCL_E_INTERNAL;
519 : }
520 : }
521 1 : return HCCL_SUCCESS;
522 : }
523 :
524 1 : inline HcclResult AicpuTaskCacheEntry::PrintRefreshResult_(
525 : [[maybe_unused]] const uint64_t* baseAddrs, [[maybe_unused]] const uint64_t* memSizes, const uint32_t count)
526 : {
527 : // 打印更新后的token信息
528 1 : for (std::unordered_map<UbTransportLiteImplHandle, vector<TokenInfo>>::iterator iter = tokenInfosMap_.begin();
529 2 : iter != tokenInfosMap_.end(); iter++) {
530 1 : UbTransportLiteImpl* ubTransportLitePtr = reinterpret_cast<UbTransportLiteImpl*>(iter->first);
531 1 : vector<TokenInfo>& tokenInfos = iter->second;
532 3 : for (uint32_t memIdx = 0; memIdx < count; memIdx++) {
533 2 : TokenInfo& tokenInfo = tokenInfos[memIdx];
534 2 : HCCL_INFO(
535 : "[AicpuTaskCacheEntry][RefreshAndLaunch] tokenInfosMap_[0x%016llx][%u]: "
536 : "needLocTokenIdFlag[%d] locTokenId[%u]; rmtTokenIdAndValueFlag[%d] rmtTokenId[%u] rmtTokenValue[%u]",
537 : ubTransportLitePtr, memIdx, tokenInfo.needLocTokenIdFlag, tokenInfo.locTokenId,
538 : tokenInfo.needRmtTokenIdAndValueFlag, tokenInfo.rmtTokenId, tokenInfo.rmtTokenValue);
539 : }
540 : }
541 : // 打印刷新的顺序
542 1 : size_t sqeArrayIdx = 0;
543 1 : size_t wqeArrayIdx = 0;
544 1 : for (size_t launchIdx = 0; launchIdx < launchOrder_.size(); launchIdx++) {
545 0 : const TaskArrayType taskType = launchOrder_[launchIdx];
546 0 : if (taskType == TaskArrayType::kTaskArrayTypeSqe) {
547 0 : HCCL_INFO(
548 : "[AicpuTaskCacheEntry][RefreshAndLaunch] launchedArray[%u] is sqeArray[%u]", launchIdx, sqeArrayIdx);
549 0 : ++sqeArrayIdx;
550 0 : } else if (taskType == TaskArrayType::kTaskArrayTypeWqe) {
551 0 : HCCL_INFO(
552 : "[AicpuTaskCacheEntry][RefreshAndLaunch] launchedArray[%u] is wqeArray[%u]", launchIdx, wqeArrayIdx);
553 0 : ++wqeArrayIdx;
554 : } else {
555 0 : HCCL_ERROR("[AicpuTaskCacheEntry][RefreshAndLaunch] invalid task array type[%u]", taskType);
556 0 : return HCCL_E_INTERNAL;
557 : }
558 : }
559 1 : return HCCL_SUCCESS;
560 : }
561 :
562 33 : inline HcclResult AicpuTaskCacheEntry::AddSqeArray_(
563 : uint8_t* newSqeArray, const size_t sqeBytes, const uint8_t* sqeArray, const uint32_t streamId)
564 : {
565 33 : CHK_SAFETY_FUNC_RET(memcpy_s(newSqeArray, sqeBytes, sqeArray, sqeBytes));
566 :
567 : // 按需更新DbSqe相关信息
568 33 : std::unordered_map<uint32_t, vector<DbSqeTmpInfo>>::iterator mapIter = streamIdToDbSqeTmpInfoMap_.find(streamId);
569 33 : if (mapIter != streamIdToDbSqeTmpInfoMap_.end()) {
570 : // 当前SQE数组可能存在多个DbSqe, 需要逐一更新
571 16 : for (size_t i = 0; i < mapIter->second.size(); ++i) {
572 : // 更新DbSqeLocation (AddWqeArray时已分配并设置相关字段, 此处只需设置sqeArrayIdx)
573 8 : const DbSqeTmpInfo& dbSqeTmpInfo = mapIter->second[i];
574 8 : CHK_PRT_RET(
575 : dbSqeTmpInfo.wqeArrayIdx >= wqeTaskArrayInfos_.size(),
576 : HCCL_ERROR(
577 : "[AicpuTaskCacheEntry][AddSqeArray_] streamIdToDbSqeTmpInfoMap_[%u][%u].wqeArrayIdx[%u] >= "
578 : "wqeTaskArrayInfos_.size[%u]",
579 : streamId, i, dbSqeTmpInfo.wqeArrayIdx, wqeTaskArrayInfos_.size()),
580 : HCCL_E_INTERNAL);
581 8 : DbSqeLocation& dbSqeLocation = wqeTaskArrayInfos_[dbSqeTmpInfo.wqeArrayIdx].dbSqeLocation;
582 8 : dbSqeLocation.sqeArrayIdx = static_cast<uint32_t>(sqeArrayInfos_.size());
583 :
584 : // 按需更新dbSqeLocInfoMap_
585 8 : if (dbSqeTmpInfo.isReportTask) {
586 : // dbSqeLocInfoMap_中一定不存在对应的DbSqeProfInfo
587 0 : CHK_PRT_RET(
588 : dbSqeLocInfoMap_.find(dbSqeLocation) != dbSqeLocInfoMap_.end(),
589 : HCCL_ERROR(
590 : "[AicpuTaskCacheEntry][AddSqeArray_] DbSqeLocation[%u, %u] already exists in "
591 : "dbSqeLocInfoMap_ for streamId[%u]",
592 : dbSqeLocation.sqeArrayIdx, dbSqeLocation.dbSqeIdx, streamId),
593 : HCCL_E_INTERNAL);
594 :
595 : // 保存到dbSqeLocInfoMap_
596 : // 注意: dbSqeProfAndRefreshInfo中的src/dstAddrRefreshInfo在SubmitCacheEntry时更新
597 0 : DbSqeProfAndRefreshInfo dbSqeProfAndRefreshInfo;
598 0 : dbSqeProfAndRefreshInfo.dbSqeProfInfo = dbSqeTmpInfo.dbSqeProfInfo;
599 : dbSqeProfAndRefreshInfo.wqeArrayIdx
600 0 : = dbSqeTmpInfo.wqeArrayIdx; // 注意: 本函数已校验dbSqeTmpInfo.wqeArrayIdx
601 0 : dbSqeLocInfoMap_.emplace(dbSqeLocation, dbSqeProfAndRefreshInfo);
602 0 : }
603 :
604 8 : HCCL_INFO(
605 : "[AicpuTaskCacheEntry][AddSqeArray_] update dbSqeLocation[%u, %u] isReportTask[%d]",
606 : dbSqeLocation.sqeArrayIdx, dbSqeLocation.dbSqeIdx, dbSqeTmpInfo.isReportTask);
607 : }
608 :
609 : // 更新后清理当前SQE数组对应的DbSqe临时信息
610 8 : streamIdToDbSqeTmpInfoMap_.erase(mapIter);
611 : }
612 33 : return HCCL_SUCCESS;
613 : }
614 :
615 30 : HcclResult AicpuTaskCacheEntry::UpdateSqeAddrRefreshInfo_(
616 : const uint8_t* sqePtr, AddrRefreshInfo& srcAddrRefreshInfo, AddrRefreshInfo& dstAddrRefreshInfo) const
617 : {
618 : // 参考sqe_build_a5.h, 提取给定SQE的AddrRefreshInfo
619 :
620 : // 提取sqeType，频繁调用私有函数，上层保证指针不为空
621 30 : Rt91095StarsSqeHeader* sqeHeaderPtr = (Rt91095StarsSqeHeader*)sqePtr;
622 30 : const Rt91095StarsSqeType sqeType = static_cast<Rt91095StarsSqeType>(sqeHeaderPtr->type);
623 :
624 : // 根据sqeType提取AddrRefreshInfo
625 30 : switch (sqeType) {
626 20 : case Rt91095StarsSqeType::RT_91095_SQE_TYPE_NOTIFY_WAIT:
627 : case Rt91095StarsSqeType::RT_91095_SQE_TYPE_NOTIFY_RECORD:
628 : case Rt91095StarsSqeType::RT_91095_SQE_TYPE_UBDMA:
629 : // 无地址字段, 直接跳过
630 20 : break;
631 6 : case Rt91095StarsSqeType::RT_91095_SQE_TYPE_SDMA: {
632 6 : Rt91095StarsMemcpySqe* memcpySqePtr = (Rt91095StarsMemcpySqe*)sqePtr;
633 6 : CHK_RET(UpdateAddrRefreshInfo_(
634 : memcpySqePtr->u.strideMode0.srcAddrLow, memcpySqePtr->u.strideMode0.srcAddrHigh, srcAddrRefreshInfo));
635 6 : CHK_RET(UpdateAddrRefreshInfo_(
636 : memcpySqePtr->u.strideMode0.dstAddrLow, memcpySqePtr->u.strideMode0.dstAddrHigh, dstAddrRefreshInfo));
637 6 : break;
638 : }
639 3 : case Rt91095StarsSqeType::RT_91095_SQE_TYPE_WRITE_VALUE: {
640 3 : Rt91095StarsWriteValueSqe* writeValueSqePtr = (Rt91095StarsWriteValueSqe*)sqePtr;
641 3 : CHK_RET(UpdateAddrRefreshInfo_(
642 : writeValueSqePtr->writeAddrLow, writeValueSqePtr->writeAddrHigh, dstAddrRefreshInfo));
643 3 : break;
644 : }
645 1 : default:
646 : // sqe_build_a5.h中未使用的SQE类型, 告警后报错
647 1 : HCCL_ERROR("[AicpuTaskCacheEntry][UpdateSqeAddrRefreshInfo_] unexpected sqeType[%u]", sqeType);
648 1 : return HCCL_E_INTERNAL;
649 : }
650 :
651 29 : return HCCL_SUCCESS;
652 : }
653 :
654 8 : HcclResult AicpuTaskCacheEntry::UpdateWqeAddrRefreshInfoAndTokenInfo_(
655 : const WqeTask& wqeTask, AddrRefreshInfo& locAddrRefreshInfo, AddrRefreshInfo& rmtAddrRefreshInfo,
656 : vector<TokenInfo>& tokenInfos)
657 : {
658 : // 参考ub_conn_lite.cc, 提取给定WQE的AddrRefreshInfo
659 :
660 : // 提取wqeCode
661 : // 注意: BatchOneSidedRead/Write只是对multi-slice封装的接口, 最终还是规约到normal read/write
662 8 : UdmaSqeCommon* wqeCommonPtr = (UdmaSqeCommon*)(&wqeTask);
663 8 : const uint8_t wqeCode = static_cast<uint8_t>(wqeCommonPtr->opcode); // opcode来自于UdmaSqOpcode, 一定在uint8范围内
664 8 : switch (wqeCode) {
665 4 : case UdmaSqOpcode::UDMA_OPC_READ: // UdmaSqeWrite
666 : // normal read
667 4 : CHK_RET(UpdateAddrRefreshInfo_(
668 : wqeTask.wqeWrite.u.sge.dataAddrLow, wqeTask.wqeWrite.u.sge.dataAddrHigh, locAddrRefreshInfo));
669 4 : CHK_RET(UpdateAddrRefreshInfo_(
670 : wqeTask.wqeWrite.comm.rmtAddrLow, wqeTask.wqeWrite.comm.rmtAddrHigh, rmtAddrRefreshInfo));
671 4 : break;
672 2 : case UdmaSqOpcode::UDMA_OPC_WRITE: { // UdmaSqeWrite
673 2 : const uint32_t inlineEn = wqeTask.wqeWrite.comm.inlineEn;
674 2 : if (inlineEn) { // inline write
675 1 : CHK_RET(UpdateAddrRefreshInfo_(
676 : wqeTask.wqeWrite.comm.rmtAddrLow, wqeTask.wqeWrite.comm.rmtAddrHigh, rmtAddrRefreshInfo));
677 : } else { // normal write or write reduce
678 1 : CHK_RET(UpdateAddrRefreshInfo_(
679 : wqeTask.wqeWrite.u.sge.dataAddrLow, wqeTask.wqeWrite.u.sge.dataAddrHigh, locAddrRefreshInfo));
680 1 : CHK_RET(UpdateAddrRefreshInfo_(
681 : wqeTask.wqeWrite.comm.rmtAddrLow, wqeTask.wqeWrite.comm.rmtAddrHigh, rmtAddrRefreshInfo));
682 : }
683 2 : break;
684 : }
685 1 : case WRITE_WITH_NOTIFY_OPCODE: // UdmaSqeWriteWithNotify
686 : // write with notify (对于给定slice的最后一个UB chunk)
687 1 : CHK_RET(UpdateAddrRefreshInfo_(
688 : wqeTask.wqeWriteWithNotify.localU.sge.dataAddrLow, wqeTask.wqeWriteWithNotify.localU.sge.dataAddrHigh,
689 : locAddrRefreshInfo));
690 1 : CHK_RET(UpdateAddrRefreshInfo_(
691 : wqeTask.wqeWriteWithNotify.comm.rmtAddrLow, wqeTask.wqeWriteWithNotify.comm.rmtAddrHigh,
692 : rmtAddrRefreshInfo));
693 1 : break;
694 1 : default:
695 : // ub_conn_lite.cc中未使用的WQE类型, 告警后报错
696 1 : HCCL_ERROR("[AicpuTaskCacheEntry][UpdateWqeAddrRefreshInfo_] unexpected wqeCode[%u]", wqeCode);
697 1 : return HCCL_E_INTERNAL;
698 : }
699 :
700 7 : CHK_RET(UpdateTokenFlagsByAddrRefreshInfo_(locAddrRefreshInfo, tokenInfos, true));
701 7 : CHK_RET(UpdateTokenFlagsByAddrRefreshInfo_(rmtAddrRefreshInfo, tokenInfos, false));
702 :
703 7 : return HCCL_SUCCESS;
704 : }
705 :
706 14 : inline HcclResult AicpuTaskCacheEntry::UpdateTokenFlagsByAddrRefreshInfo_(
707 : const AddrRefreshInfo& addrRefreshInfo, vector<TokenInfo>& tokenInfos, bool isLoc)
708 : {
709 14 : if (!addrRefreshInfo.needRefresh) {
710 1 : return HCCL_SUCCESS;
711 : }
712 13 : CHK_PRT_RET(
713 : addrRefreshInfo.memIdx >= tokenInfos.size(),
714 : HCCL_ERROR(
715 : "[AicpuTaskCacheEntry][UpdateWqeAddrRefreshInfo_] %s memIdx[%u] >= tokenInfos.size[%u]",
716 : isLoc ? "loc" : "rmt", addrRefreshInfo.memIdx, tokenInfos.size()),
717 : HCCL_E_INTERNAL);
718 13 : if (isLoc) {
719 6 : tokenInfos[addrRefreshInfo.memIdx].needLocTokenIdFlag = true;
720 : } else {
721 7 : tokenInfos[addrRefreshInfo.memIdx].needRmtTokenIdAndValueFlag = true;
722 : }
723 13 : return HCCL_SUCCESS;
724 : }
725 :
726 28 : HcclResult AicpuTaskCacheEntry::UpdateAddrRefreshInfo_(const uint64_t addr, AddrRefreshInfo& addrRefreshInfo) const
727 : {
728 : // 默认不是dynamic memory (e.g., user input/ouput), 认为无需刷新
729 28 : addrRefreshInfo.needRefresh = false;
730 :
731 : // 检查是否为任意一段dynamic memory
732 47 : for (uint32_t memIdx = 0; memIdx < cachedBaseAddrs_.size(); memIdx++) {
733 : // Memory range: [baseAddr, baseAddr + memSize)
734 43 : const uint64_t baseAddr = cachedBaseAddrs_[memIdx];
735 43 : const uint64_t memSize = cachedMemSizes_[memIdx];
736 :
737 : // 在InitCacheEntry已经做了baseAddr的空指针，及baseAddr + memsize的溢出检查。
738 :
739 : // 检查是否在当前dynamic memory range
740 43 : if (AicpuTaskCacheEntry::InRange(baseAddr, memSize, addr)) {
741 24 : addrRefreshInfo.needRefresh = true;
742 24 : addrRefreshInfo.memIdx = memIdx;
743 24 : addrRefreshInfo.offset = addr - baseAddr;
744 24 : break;
745 : }
746 : }
747 :
748 28 : return HCCL_SUCCESS;
749 : }
750 :
751 43 : inline bool AicpuTaskCacheEntry::InRange(const uint64_t baseAddr, const uint64_t memSize, const uint64_t addr)
752 : {
753 43 : return (addr >= baseAddr && addr < baseAddr + memSize);
754 : }
755 :
756 10 : inline HcclResult AicpuTaskCacheEntry::RefreshSqeTasks_(const SqeArrayInfo& sqeArrayInfo, const uint64_t* baseAddrs)
757 : {
758 : // sqe数组
759 10 : uint8_t* sqeArrayPtr = sqeArrayInfo.sqeArray; // 注意: sqeArrayPtr已在AddSqeArray校验, 无需再校验
760 10 : uint64_t sqeCount = sqeArrayInfo.sqeCount;
761 10 : RtsqA5* rtsqA5Ptr = sqeArrayInfo.rtsqPtr; // 注意: rtsqPtr已在AddSqeArray校验, 无需再校验
762 10 : const vector<AddrRefreshInfo>& sqeSrcAddrRefreshInfoArray = sqeArrayInfo.srcAddrRefreshInfoArray;
763 10 : const vector<AddrRefreshInfo>& sqeDstAddrRefreshInfoArray = sqeArrayInfo.dstAddrRefreshInfoArray;
764 10 : if (UNLIKELY(isTaskConfigDebug_)) {
765 10 : PLF_CONFIG_DEBUG(
766 : PLF_TASK, "[AicpuTaskCacheEntry][RefreshSqeTasks_] dump %llu cached SQEs in stream[%u]", sqeCount,
767 : rtsqA5Ptr->GetStreamId());
768 : }
769 21 : for (size_t sqeIdx = 0; sqeIdx < sqeCount; ++sqeIdx) {
770 : // 获取当前SQE的信息
771 12 : Rt91095StarsSqeHeader* sqeHeaderPtr = (Rt91095StarsSqeHeader*)sqeArrayPtr;
772 12 : CHK_RET(RefreshOneSqe_(
773 : sqeArrayPtr, sqeSrcAddrRefreshInfoArray[sqeIdx], sqeDstAddrRefreshInfoArray[sqeIdx], baseAddrs));
774 :
775 : // 刷新streamId, taskId
776 11 : rtsqA5Ptr->RefreshSqeHeaderTaskField(sqeHeaderPtr);
777 :
778 : // 切换至下一个SQE
779 11 : sqeArrayPtr += AC_SQE_SIZE;
780 : }
781 :
782 : // 按需打印刷新后的SQE内容
783 9 : if (UNLIKELY(isTaskConfigDebug_)) {
784 : // 循环打印刷新后的SQE内容
785 9 : sqeArrayPtr = sqeArrayInfo.sqeArray;
786 20 : for (size_t sqeIdx = 0; sqeIdx < sqeCount; ++sqeIdx) {
787 : // 打印当前SQE
788 11 : PLF_CONFIG_DEBUG(
789 : PLF_TASK, "[AicpuTaskCacheEntry][RefreshSqeTasks_] %uth cached SQE in stream[%u]", sqeIdx,
790 : rtsqA5Ptr->GetStreamId());
791 11 : CHK_RET(AicpuTaskUtils::DumpSqeContent(sqeArrayPtr));
792 :
793 : // 切换至下一个SQE
794 11 : sqeArrayPtr += AC_SQE_SIZE;
795 : }
796 : }
797 9 : return HCCL_SUCCESS;
798 : }
799 :
800 12 : inline HcclResult AicpuTaskCacheEntry::RefreshOneSqe_(
801 : uint8_t* sqeArrayPtr, const AddrRefreshInfo& srcAddrRefreshInfo, const AddrRefreshInfo& dstAddrRefreshInfo,
802 : const uint64_t* baseAddrs)
803 : {
804 12 : Rt91095StarsSqeHeader* sqeHeaderPtr = (Rt91095StarsSqeHeader*)sqeArrayPtr;
805 12 : const Rt91095StarsSqeType sqeType = static_cast<Rt91095StarsSqeType>(sqeHeaderPtr->type);
806 : // 根据SQE type进行对应刷新 (task id始终要刷新; addr相关字段有条件刷新)
807 12 : switch (sqeType) {
808 5 : case Rt91095StarsSqeType::RT_91095_SQE_TYPE_NOTIFY_WAIT:
809 : case Rt91095StarsSqeType::RT_91095_SQE_TYPE_NOTIFY_RECORD:
810 : case Rt91095StarsSqeType::RT_91095_SQE_TYPE_UBDMA:
811 : // 无地址字段, 直接跳过
812 5 : break;
813 4 : case Rt91095StarsSqeType::RT_91095_SQE_TYPE_SDMA: {
814 4 : Rt91095StarsMemcpySqe* memcpySqePtr = (Rt91095StarsMemcpySqe*)sqeArrayPtr;
815 : // 刷新地址
816 4 : if (srcAddrRefreshInfo.needRefresh) {
817 3 : RefreshTaskAddr_(
818 3 : memcpySqePtr->u.strideMode0.srcAddrLow, memcpySqePtr->u.strideMode0.srcAddrHigh, srcAddrRefreshInfo,
819 : baseAddrs);
820 : }
821 4 : if (dstAddrRefreshInfo.needRefresh) {
822 3 : RefreshTaskAddr_(
823 3 : memcpySqePtr->u.strideMode0.dstAddrLow, memcpySqePtr->u.strideMode0.dstAddrHigh, dstAddrRefreshInfo,
824 : baseAddrs);
825 : }
826 4 : break;
827 : }
828 2 : case Rt91095StarsSqeType::RT_91095_SQE_TYPE_WRITE_VALUE: {
829 2 : Rt91095StarsWriteValueSqe* writeValueSqePtr = (Rt91095StarsWriteValueSqe*)sqeArrayPtr;
830 : // 刷新地址
831 2 : if (dstAddrRefreshInfo.needRefresh) {
832 : // 注意: writeAddrHigh是位域, 无法直接作为u32&传入
833 2 : uint32_t tmpAddrHigh = writeValueSqePtr->writeAddrHigh;
834 2 : RefreshTaskAddr_(writeValueSqePtr->writeAddrLow, tmpAddrHigh, dstAddrRefreshInfo, baseAddrs);
835 2 : writeValueSqePtr->writeAddrHigh = tmpAddrHigh;
836 : }
837 2 : break;
838 : }
839 1 : default:
840 : // sqe_build_a5.h中未使用的SQE类型, 告警后报错
841 1 : HCCL_ERROR("[AicpuTaskCacheEntry][%s] unexpected sqeType[%u]", __func__, sqeType);
842 1 : return HCCL_E_INTERNAL;
843 : }
844 11 : return HCCL_SUCCESS;
845 : }
846 :
847 0 : inline HcclResult AicpuTaskCacheEntry::LaunchSqeTasks_(const SqeArrayInfo& sqeArrayInfo)
848 : {
849 : // 注意: rtsqPtr和sqeArray在AddSqeArray时已校验, 这里无需再校验
850 0 : RtsqA5* rtsqA5Ptr = sqeArrayInfo.rtsqPtr;
851 0 : rtsqA5Ptr->LaunchNewTask(sqeArrayInfo.sqeArray, (u32)sqeArrayInfo.sqeCount);
852 0 : return HCCL_SUCCESS;
853 : }
854 :
855 : // 频繁调用函数，参数外层已经校验，不会失败，无需返回HcclResult让调用方判断。
856 10 : inline void AicpuTaskCacheEntry::RefreshTaskAddr_(
857 : uint32_t& addrLow, uint32_t& addrHigh, const AddrRefreshInfo& addrRefreshInfo, const uint64_t* baseAddrs) const
858 : {
859 : // 注意: 调用RefreshTaskAddr_时已判断addrRefreshInfo.needRefresh, 无需再次校验
860 :
861 : // 注意: memIdx已经在SubmitCacheEntry的UpdateAddrRefreshInfo_中校验, 确保在cachedMemSizes_范围内
862 :
863 : // 计算newAddr
864 10 : const uint64_t newAddr = baseAddrs[addrRefreshInfo.memIdx] + addrRefreshInfo.offset;
865 :
866 : // 返回新地址
867 10 : AicpuTaskCacheEntry::SplitUint64ToUint32(newAddr, addrHigh, addrLow);
868 10 : }
869 :
870 : // 频繁调用函数，参数外层已经校验，不会失败，无需返回HcclResult让上层判断。
871 0 : inline void AicpuTaskCacheEntry::RefreshTaskAddr_(
872 : uint64_t& addr, const AddrRefreshInfo& addrRefreshInfo, const uint64_t* baseAddrs) const
873 : {
874 : // 注意: 调用RefreshTaskAddr_时已判断addrRefreshInfo.needRefresh, 无需再次校验
875 :
876 : // 注意: memIdx已经在SubmitCacheEntry的UpdateAddrRefreshInfo_中校验, 确保在cachedMemSizes_范围内
877 :
878 : // 计算newAddr, 返回新地址
879 0 : addr = baseAddrs[addrRefreshInfo.memIdx] + addrRefreshInfo.offset;
880 0 : }
881 :
882 1 : inline void AicpuTaskCacheEntry::DumpWqeTasksHeader_(uint64_t wqeCount, const UbConnLite* ubConnLitePtr) const
883 : {
884 1 : PLF_CONFIG_DEBUG(
885 : PLF_TASK, "[AicpuTaskCacheEntry][RefreshWqeTasks_] dump %llu cached WQEs in jetty[%u, %u, %u]", wqeCount,
886 : ubConnLitePtr->GetUbJettyLiteId().GetDieId(), ubConnLitePtr->GetUbJettyLiteId().GetFuncId(),
887 : ubConnLitePtr->GetUbJettyLiteId().GetJettyId());
888 1 : }
889 :
890 : inline HcclResult
891 1 : AicpuTaskCacheEntry::DumpWqeTasksPerWqe_(size_t wqeIdx, const WqeTask& wqeTask, const UbConnLite* ubConnLitePtr) const
892 : {
893 1 : PLF_CONFIG_DEBUG(
894 : PLF_TASK, "[AicpuTaskCacheEntry][%s] %uth cached WQE in jetty[%u, %u, %u]", __func__, wqeIdx,
895 : ubConnLitePtr->GetUbJettyLiteId().GetDieId(), ubConnLitePtr->GetUbJettyLiteId().GetFuncId(),
896 : ubConnLitePtr->GetUbJettyLiteId().GetJettyId());
897 1 : CHK_RET(AicpuTaskUtils::DumpWqeContent(reinterpret_cast<const uint8_t*>(&wqeTask)));
898 1 : return HCCL_SUCCESS;
899 : }
900 :
901 1 : inline HcclResult AicpuTaskCacheEntry::RefreshWqeTasks_(
902 : WqeTaskArrayInfo& wqeTaskArrayInfo, const uint64_t* baseAddrs, [[maybe_unused]] const uint64_t* memSizes,
903 : [[maybe_unused]] const uint32_t count)
904 : {
905 : // 逐个刷新地址
906 1 : vector<WqeTask>& wqeTasks = wqeTaskArrayInfo.wqeTaskArray;
907 1 : const vector<AddrRefreshInfo>& wqeLocAddrRefreshInfoArray = wqeTaskArrayInfo.locAddrRefreshInfoArray;
908 1 : const vector<AddrRefreshInfo>& wqeRmtAddrRefreshInfoArray = wqeTaskArrayInfo.rmtAddrRefreshInfoArray;
909 1 : UbTransportLiteImpl* ubTransportLitePtr
910 : = wqeTaskArrayInfo.ubTransportLiteImplPtr; // 注意: 已在AddWqeArray时校验非空
911 1 : const uint64_t wqeCount = wqeTasks.size();
912 1 : if (UNLIKELY(isTaskConfigDebug_)) {
913 1 : DumpWqeTasksHeader_(wqeCount, wqeTaskArrayInfo.ubConnLitePtr);
914 : }
915 : // 当前UbTransportLiteImpl在对应内存上, 一定已经提前获取了rmtTokenId/Value
916 : std::unordered_map<UbTransportLiteImplHandle, vector<TokenInfo>>::const_iterator constIter
917 1 : = tokenInfosMap_.find(ubTransportLitePtr);
918 1 : CHK_PRT_RET(
919 : constIter == tokenInfosMap_.end(),
920 : HCCL_ERROR(
921 : "[AicpuTaskCacheEntry][RefreshWqeTasks_] ubTransportLitePtr[%p] not found in tokenInfosMap_",
922 : ubTransportLitePtr),
923 : HCCL_E_INTERNAL);
924 1 : const vector<TokenInfo>& tokenInfos = constIter->second;
925 2 : for (size_t wqeIdx = 0; wqeIdx < wqeCount; wqeIdx++) {
926 1 : WqeTask& wqeTask = wqeTasks[wqeIdx];
927 1 : const AddrRefreshInfo& locAddrRefreshInfo = wqeLocAddrRefreshInfoArray[wqeIdx];
928 1 : const AddrRefreshInfo& rmtAddrRefreshInfo = wqeRmtAddrRefreshInfoArray[wqeIdx];
929 :
930 : // 根据WQE类型刷新对应地址字段及token id/value
931 1 : UdmaSqeCommon* wqeCommonPtr = (UdmaSqeCommon*)(&wqeTask);
932 1 : const uint8_t wqeCode
933 : = static_cast<uint8_t>(wqeCommonPtr->opcode); // opcode来自于UdmaSqOpcode, 一定在uint8范围内
934 1 : switch (wqeCode) {
935 1 : case UdmaSqOpcode::UDMA_OPC_READ: // UdmaSqeWrite
936 : // normal read
937 1 : CHK_RET(RefreshWqeRead_(wqeTask, locAddrRefreshInfo, rmtAddrRefreshInfo, baseAddrs, tokenInfos));
938 1 : break;
939 0 : case UdmaSqOpcode::UDMA_OPC_WRITE: { // UdmaSqeWrite
940 0 : CHK_RET(RefreshWqeWrite_(wqeTask, locAddrRefreshInfo, rmtAddrRefreshInfo, baseAddrs, tokenInfos));
941 0 : break;
942 : }
943 0 : case WRITE_WITH_NOTIFY_OPCODE: // UdmaSqeWriteWithNotify
944 : // write with notify (对于给定slice的最后一个UB chunk)
945 0 : CHK_RET(
946 : RefreshWqeWriteWithNotify_(wqeTask, locAddrRefreshInfo, rmtAddrRefreshInfo, baseAddrs, tokenInfos));
947 0 : break;
948 0 : default:
949 : // ub_conn_lite.cc中未使用的WQE类型, 告警后报错
950 0 : HCCL_ERROR("[AicpuTaskCacheEntry][%s] unexpected wqeCode[%u]", __func__, wqeCode);
951 0 : return HCCL_E_INTERNAL;
952 : }
953 1 : if (UNLIKELY(isTaskConfigDebug_)) {
954 1 : CHK_RET(DumpWqeTasksPerWqe_(wqeIdx, wqeTask, wqeTaskArrayInfo.ubConnLitePtr));
955 : }
956 : }
957 1 : return HCCL_SUCCESS;
958 : }
959 :
960 1 : inline HcclResult AicpuTaskCacheEntry::RefreshWqeRead_(
961 : WqeTask& wqeTask, const AddrRefreshInfo& locAddrRefreshInfo, const AddrRefreshInfo& rmtAddrRefreshInfo,
962 : const uint64_t* baseAddrs, const vector<TokenInfo>& tokenInfos)
963 : {
964 : // 如果需要刷新loc地址信息
965 1 : if (locAddrRefreshInfo.needRefresh) {
966 : // 刷新loc addr
967 1 : RefreshTaskAddr_(
968 1 : wqeTask.wqeWrite.u.sge.dataAddrLow, wqeTask.wqeWrite.u.sge.dataAddrHigh, locAddrRefreshInfo, baseAddrs);
969 : // 根据刷新后的loc addr, 刷新loc token id (注意: loc不需要刷新token value)
970 1 : CHK_RET(RefreshWqeLocTokenId_(wqeTask.wqeWrite.u.sge.tokenId, locAddrRefreshInfo, tokenInfos));
971 : }
972 : // 如果需要刷新rmt地址信息
973 1 : if (rmtAddrRefreshInfo.needRefresh) {
974 : // 刷新rmt addr
975 1 : RefreshTaskAddr_(
976 1 : wqeTask.wqeWrite.comm.rmtAddrLow, wqeTask.wqeWrite.comm.rmtAddrHigh, rmtAddrRefreshInfo, baseAddrs);
977 :
978 : // 注意: rmtObjId是位域, 不能直接传引用
979 1 : uint32_t rmtObjId = wqeTask.wqeWrite.comm.rmtObjId;
980 1 : CHK_RET(RefreshWqeRmtTokenIdAndValue_(
981 : rmtObjId, wqeTask.wqeWrite.comm.rmtTokenValue, rmtAddrRefreshInfo, tokenInfos));
982 1 : wqeTask.wqeWrite.comm.rmtObjId = rmtObjId;
983 : }
984 1 : return HCCL_SUCCESS;
985 : }
986 :
987 0 : inline HcclResult AicpuTaskCacheEntry::RefreshWqeWrite_(
988 : WqeTask& wqeTask, const AddrRefreshInfo& locAddrRefreshInfo, const AddrRefreshInfo& rmtAddrRefreshInfo,
989 : const uint64_t* baseAddrs, const vector<TokenInfo>& tokenInfos)
990 : {
991 0 : const uint32_t inlineEn = wqeTask.wqeWrite.comm.inlineEn;
992 0 : if (inlineEn) { // inline write
993 : // 注意: inline write使用WriteWqe实现notify功能 (类似A3使用WriteValue实现notify功能)
994 : // 因为notify token id/value以及notify addr不会改变, 因此inline write无需刷新WQE
995 0 : if (UNLIKELY(rmtAddrRefreshInfo.needRefresh)) {
996 : // 打印告警信息
997 0 : HCCL_ERROR(
998 : "[AicpuTaskCacheEntry][RefreshWqeTasks_] inline write should not refresh rmt addr: "
999 : "rmtAddrRefreshInfo.needRefresh[%d] rmtAddrRefreshInfo.memIdx[%u]",
1000 : rmtAddrRefreshInfo.needRefresh, rmtAddrRefreshInfo.memIdx);
1001 :
1002 : // 打印地址信息
1003 : // 注意: memIdx在SubmitCacheEntry中已校验, 这里直接使用
1004 0 : uint64_t rmtAddr = 0;
1005 0 : AicpuTaskCacheEntry::CombineUint32ToUint64(
1006 : rmtAddr, wqeTask.wqeWrite.comm.rmtAddrHigh, wqeTask.wqeWrite.comm.rmtAddrLow);
1007 0 : const uint64_t cachedBaseAddr = cachedBaseAddrs_[rmtAddrRefreshInfo.memIdx];
1008 0 : const uint64_t cachedMemSize = cachedMemSizes_[rmtAddrRefreshInfo.memIdx];
1009 0 : HCCL_ERROR(
1010 : "[AicpuTaskCacheEntry][RefreshWqeTasks_] rmtAddr[0x%016llx] "
1011 : "cachedBaseAddr[0x%016llx] endAddr[0x%016llx] memSize[%llu]",
1012 : rmtAddr, cachedBaseAddr, cachedBaseAddr + cachedMemSize, cachedMemSize);
1013 :
1014 0 : return HCCL_E_INTERNAL;
1015 : }
1016 0 : return HCCL_SUCCESS;
1017 : }
1018 : // normal write or write reduce
1019 0 : return RefreshWqeRead_(wqeTask, locAddrRefreshInfo, rmtAddrRefreshInfo, baseAddrs, tokenInfos);
1020 : }
1021 :
1022 0 : inline HcclResult AicpuTaskCacheEntry::RefreshWqeWriteWithNotify_(
1023 : WqeTask& wqeTask, const AddrRefreshInfo& locAddrRefreshInfo, const AddrRefreshInfo& rmtAddrRefreshInfo,
1024 : const uint64_t* baseAddrs, const vector<TokenInfo>& tokenInfos)
1025 : {
1026 : // 如果需要刷新loc地址信息
1027 0 : if (locAddrRefreshInfo.needRefresh) {
1028 : // 刷新loc addr
1029 0 : RefreshTaskAddr_(
1030 0 : wqeTask.wqeWriteWithNotify.localU.sge.dataAddrLow, wqeTask.wqeWriteWithNotify.localU.sge.dataAddrHigh,
1031 : locAddrRefreshInfo, baseAddrs);
1032 : // 根据刷新后的loc addr, 刷新loc token id (注意: loc不需要刷新token value)
1033 0 : CHK_RET(RefreshWqeLocTokenId_(wqeTask.wqeWriteWithNotify.localU.sge.tokenId, locAddrRefreshInfo, tokenInfos));
1034 : }
1035 : // 如果需要刷新rmt地址信息
1036 0 : if (rmtAddrRefreshInfo.needRefresh) {
1037 : // 刷新rmt addr
1038 0 : RefreshTaskAddr_(
1039 0 : wqeTask.wqeWriteWithNotify.comm.rmtAddrLow, wqeTask.wqeWriteWithNotify.comm.rmtAddrHigh, rmtAddrRefreshInfo,
1040 : baseAddrs);
1041 :
1042 : // 注意: rmtObjId是位域, 不能直接传引用
1043 0 : uint32_t rmtObjId = wqeTask.wqeWriteWithNotify.comm.rmtObjId;
1044 0 : CHK_RET(RefreshWqeRmtTokenIdAndValue_(
1045 : rmtObjId, wqeTask.wqeWriteWithNotify.comm.rmtTokenValue, rmtAddrRefreshInfo, tokenInfos));
1046 0 : wqeTask.wqeWriteWithNotify.comm.rmtObjId = rmtObjId;
1047 : }
1048 0 : return HCCL_SUCCESS;
1049 : }
1050 :
1051 1 : inline HcclResult AicpuTaskCacheEntry::RefreshWqeLocTokenId_(
1052 : uint32_t& tokenId, const AddrRefreshInfo& addrRefreshInfo, const vector<TokenInfo>& tokenInfos) const
1053 : {
1054 : // 调用方保证addrRefreshInfo.needRefresh为true
1055 :
1056 : // 当前UbTransportLiteImpl在对应内存上, 一定已经提前获取了locTokenId
1057 : // 注意: ubTransportLitePtr已在AddWqeArray时校验非空, 这里无需再校验
1058 1 : const TokenInfo& tokenInfo = tokenInfos[addrRefreshInfo.memIdx];
1059 :
1060 1 : CHK_PRT_RET(
1061 : !tokenInfo.needLocTokenIdFlag,
1062 : HCCL_ERROR("[AicpuTaskCacheEntry][RefreshWqeLocTokenId_] needLocTokenIdFlag is false"), HCCL_E_INTERNAL);
1063 :
1064 : // 刷新loc token id
1065 1 : tokenId = tokenInfo.locTokenId;
1066 :
1067 1 : return HCCL_SUCCESS;
1068 : }
1069 :
1070 1 : inline HcclResult AicpuTaskCacheEntry::RefreshWqeRmtTokenIdAndValue_(
1071 : uint32_t& tokenId, uint32_t& tokenValue, const AddrRefreshInfo& addrRefreshInfo,
1072 : const vector<TokenInfo>& tokenInfos) const
1073 : {
1074 : // 调用方保证addrRefreshInfo.needRefresh为true
1075 :
1076 : // 当前UbTransportLiteImpl在对应内存上, 一定已经提前获取了rmtTokenId/Value
1077 1 : const TokenInfo& tokenInfo = tokenInfos[addrRefreshInfo.memIdx];
1078 :
1079 1 : CHK_PRT_RET(
1080 : !tokenInfo.needRmtTokenIdAndValueFlag,
1081 : HCCL_ERROR("[AicpuTaskCacheEntry][RefreshWqeRmtTokenIdAndValue_] needRmtTokenIdAndValueFlag is false"),
1082 : HCCL_E_INTERNAL);
1083 :
1084 : // 刷新remote token id/value
1085 1 : tokenId = tokenInfo.rmtTokenId;
1086 1 : tokenValue = tokenInfo.rmtTokenValue;
1087 :
1088 1 : return HCCL_SUCCESS;
1089 : }
1090 :
1091 0 : inline HcclResult AicpuTaskCacheEntry::LaunchWqeTasks_(WqeTaskArrayInfo& wqeTaskArrayInfo)
1092 : {
1093 : // 逐个下发WQE
1094 0 : vector<WqeTask>& wqeTasks = wqeTaskArrayInfo.wqeTaskArray;
1095 0 : UbConnLite* ubConnLitePtr = wqeTaskArrayInfo.ubConnLitePtr; // 注意: AddWqeArray时已校验非空
1096 0 : const size_t wqeCount = wqeTasks.size();
1097 0 : for (size_t wqeIdx = 0; wqeIdx < wqeCount; wqeIdx++) {
1098 0 : WqeTask& wqeTask = wqeTasks[wqeIdx];
1099 :
1100 : // 根据WQE类型下发 (下发过程中会更新ubConnLitePtr中的pi)
1101 0 : UdmaSqeCommon* wqeCommonPtr = (UdmaSqeCommon*)(&wqeTask);
1102 0 : const uint8_t wqeCode
1103 : = static_cast<uint8_t>(wqeCommonPtr->opcode); // opcode来自于UdmaSqOpcode, 一定在uint8范围内
1104 0 : switch (wqeCode) {
1105 0 : case UdmaSqOpcode::UDMA_OPC_READ: // UdmaSqeWrite
1106 : // normal read
1107 0 : ubConnLitePtr->LaunchOneWqe(&wqeTask.wqeWrite, UdmaSqOpcode::UDMA_OPC_READ);
1108 0 : break;
1109 0 : case UdmaSqOpcode::UDMA_OPC_WRITE: // UdmaSqeWrite
1110 : // inline write, normal write, or write reduce
1111 0 : ubConnLitePtr->LaunchOneWqe(&wqeTask.wqeWrite, UdmaSqOpcode::UDMA_OPC_WRITE);
1112 0 : break;
1113 0 : case WRITE_WITH_NOTIFY_OPCODE: // UdmaSqeWriteWithNotify
1114 : // write with notify (对于给定slice的最后一个UB chunk)
1115 0 : ubConnLitePtr->LaunchOneWqeWithNotify(&wqeTask.wqeWriteWithNotify, WRITE_WITH_NOTIFY_OPCODE);
1116 0 : break;
1117 0 : default:
1118 : // ub_conn_lite.cc中未使用的WQE类型, 告警后报错
1119 0 : HCCL_ERROR("[AicpuTaskCacheEntry][LaunchWqeTasks_] unexpected wqeCode[%u]", wqeCode);
1120 0 : return HCCL_E_INTERNAL;
1121 : }
1122 : }
1123 :
1124 0 : return HCCL_SUCCESS;
1125 : }
1126 :
1127 0 : inline HcclResult AicpuTaskCacheEntry::RefreshDbSqe_(WqeTaskArrayInfo& wqeTaskArrayInfo)
1128 : {
1129 : // 获取pi
1130 0 : UbConnLite* ubConnLitePtr = wqeTaskArrayInfo.ubConnLitePtr; // 注意: AddWqeArray时已校验非空
1131 0 : const uint16_t pi = ubConnLitePtr->GetPi();
1132 :
1133 : // 校验dbSqeLocation, AddSqeArray_中已经校验
1134 0 : const DbSqeLocation& dbSqeLocation = wqeTaskArrayInfo.dbSqeLocation;
1135 :
1136 : // 校验SQE type
1137 0 : uint8_t* sqePtr = sqeArrayInfos_[dbSqeLocation.sqeArrayIdx].sqeArray + dbSqeLocation.dbSqeIdx * AC_SQE_SIZE;
1138 :
1139 : // 更新SQE pi value
1140 0 : Rt91095StarsUbdmaDBmodeSqe* dbSqePtr = (Rt91095StarsUbdmaDBmodeSqe*)sqePtr;
1141 0 : dbSqePtr->piValue1 = pi;
1142 :
1143 : // 注意: UbTransportLiteImpl只针对WQE按需填充DfxTaskInfo上报profiling, DB SQE无需上报profiling
1144 :
1145 0 : HCCL_INFO(
1146 : "[AicpuTaskCacheEntry][RefreshDbSqe_] update pi[%u] at dbSqeLocation[%u, %u]", pi, dbSqeLocation.sqeArrayIdx,
1147 : dbSqeLocation.dbSqeIdx);
1148 :
1149 0 : return HCCL_SUCCESS;
1150 : }
1151 :
1152 0 : HcclResult AicpuTaskCacheEntry::ReportDbSqeProfiling_(
1153 : uint8_t* dbSqePtr, size_t arrayIdx, uint32_t dbSqeIdx, const uint64_t* baseAddrs, const uint64_t* memSizes,
1154 : const uint32_t count, StreamLite* streamLite, const u32 sqId, const u32 taskId)
1155 : {
1156 : // 注意: 参考ub_transport_lite_impl.cc填充DfxTaskInfo并经NextTaskSlot上报
1157 :
1158 0 : DbSqeLocation dbSqeLocation;
1159 0 : dbSqeLocation.sqeArrayIdx = arrayIdx;
1160 0 : dbSqeLocation.dbSqeIdx = dbSqeIdx;
1161 0 : auto it = dbSqeLocInfoMap_.find(dbSqeLocation);
1162 :
1163 : // 注意: 少数场景下DbSqe不需要上报, 例如Drain, BatchOneSidedRead, BatchOneSidedWrite
1164 0 : if (UNLIKELY(it == dbSqeLocInfoMap_.end())) {
1165 0 : HCCL_INFO(
1166 : "[AicpuTaskCacheEntry][ReportDbSqeProfiling_] dbSqeLocInfoMap_[%u, %u] is not found",
1167 : dbSqeLocation.sqeArrayIdx, dbSqeLocation.dbSqeIdx);
1168 0 : return HCCL_SUCCESS;
1169 : }
1170 :
1171 0 : CHK_PRT_RET(
1172 : it->second.dbSqeProfInfo.isValid == false,
1173 : HCCL_ERROR(
1174 : "[AicpuTaskCacheEntry][ReportDbSqeProfiling_] dbSqeLocInfoMap_[%u, %u] is invalid",
1175 : dbSqeLocation.sqeArrayIdx, dbSqeLocation.dbSqeIdx),
1176 : HCCL_E_INTERNAL);
1177 :
1178 : // 校验SQE type
1179 0 : Rt91095StarsSqeHeader* sqeHeaderPtr = (Rt91095StarsSqeHeader*)dbSqePtr; // 已在AddSqeArray校验, 无需再校验
1180 0 : CHK_PRT_RET(
1181 : static_cast<Rt91095StarsSqeType>(sqeHeaderPtr->type) != Rt91095StarsSqeType::RT_91095_SQE_TYPE_UBDMA,
1182 : HCCL_ERROR(
1183 : "[AicpuTaskCacheEntry][ReportDbSqeProfiling_] sqeHeaderPtr->type[%u] is not RT_91095_SQE_TYPE_UBDMA",
1184 : sqeHeaderPtr->type),
1185 : HCCL_E_INTERNAL);
1186 :
1187 0 : HCCL_INFO(
1188 : "[AicpuTaskCacheEntry][ReportDbSqeProfiling_] report %lluth DbSqe profiling in %lluth SQE array: "
1189 : "dbSqeType[%u], taskId[%u], sqId[%u]",
1190 : dbSqeIdx, arrayIdx, sqeHeaderPtr->type, taskId, sqId);
1191 :
1192 0 : CHK_RET(RefreshDbSqeProfAddrs_(it->second, baseAddrs, memSizes, count));
1193 :
1194 : // 从DbSqeProfAndRefreshInfo中获取DbSqe对应的UbTransportLiteImpl
1195 : // 注意: wqeArrayIdx已在AddSqeArray_校验, 无需再校验
1196 0 : const uint32_t wqeArrayIdx = it->second.wqeArrayIdx;
1197 : // 注意: ubTransportLiteImplPtr已在AddWqeArray校验, 无需再校验
1198 0 : UbTransportLiteImpl* ubTransportLitePtr = wqeTaskArrayInfos_[wqeArrayIdx].ubTransportLiteImplPtr;
1199 :
1200 0 : Hccl::DfxTaskInfo* slot = streamLite->NextTaskSlot();
1201 0 : switch (static_cast<u8>(it->second.dbSqeProfInfo.taskParamType)) {
1202 0 : case TaskParamTypeVal::TASK_UB_INLINE_WRITE:
1203 : case TaskParamTypeVal::TASK_UB:
1204 : case TaskParamTypeVal::TASK_WRITE_WITH_NOTIFY:
1205 0 : CHK_RET(FillSlotUbDma_(slot, dbSqePtr, it->second, ubTransportLitePtr, streamLite, taskId));
1206 0 : break;
1207 0 : case TaskParamTypeVal::TASK_UB_REDUCE_INLINE:
1208 : case TaskParamTypeVal::TASK_WRITE_REDUCE_WITH_NOTIFY:
1209 0 : CHK_RET(FillSlotReduce_(slot, dbSqePtr, it->second, ubTransportLitePtr, streamLite, taskId));
1210 0 : break;
1211 0 : default:
1212 0 : HCCL_ERROR(
1213 : "[AicpuTaskCacheEntry][ReportDbSqeProfiling_] taskType[%u] is unsupported",
1214 : it->second.dbSqeProfInfo.taskParamType);
1215 0 : return HCCL_E_INTERNAL;
1216 : }
1217 :
1218 0 : return HCCL_SUCCESS;
1219 : }
1220 :
1221 0 : inline HcclResult AicpuTaskCacheEntry::RefreshDbSqeProfAddrs_(
1222 : DbSqeProfAndRefreshInfo& profAndRefreshInfo, const uint64_t* baseAddrs, [[maybe_unused]] const uint64_t* memSizes,
1223 : [[maybe_unused]] const uint32_t count)
1224 : {
1225 : // 注意: dbSqeProfInfo中的src/dstAddr, 需要根据DbSqeProfAndRefreshInfo中的src/dstAddrRefreshInfo进行刷新,
1226 : // 才能填充DfxTaskInfo
1227 0 : if (profAndRefreshInfo.srcAddrRefreshInfo.needRefresh) {
1228 0 : RefreshTaskAddr_(profAndRefreshInfo.dbSqeProfInfo.srcAddr, profAndRefreshInfo.srcAddrRefreshInfo, baseAddrs);
1229 : }
1230 0 : if (profAndRefreshInfo.dstAddrRefreshInfo.needRefresh) {
1231 0 : RefreshTaskAddr_(profAndRefreshInfo.dbSqeProfInfo.dstAddr, profAndRefreshInfo.dstAddrRefreshInfo, baseAddrs);
1232 : }
1233 0 : return HCCL_SUCCESS;
1234 : }
1235 :
1236 0 : inline void AicpuTaskCacheEntry::FillSlotCommonFields_(
1237 : Hccl::DfxTaskInfo* slot, StreamLite* streamLite, u32 taskId, u8 linkType, u8 transportType, u64 channelHandle) const
1238 : {
1239 0 : slot->sqId = streamLite->GetSqId();
1240 0 : slot->taskId = taskId;
1241 0 : const void* opInfo = streamLite->GetLatestDfxOpInfo();
1242 0 : slot->dfxOpInfo = (opInfo != nullptr) ? reinterpret_cast<u64>(opInfo) : DFX_INVALID_U64;
1243 0 : slot->linkType = linkType;
1244 0 : slot->transportType = transportType;
1245 0 : slot->channelHandle = channelHandle;
1246 0 : }
1247 :
1248 0 : inline u8 AicpuTaskCacheEntry::ConvertSdmaOpCodeToReduceOp_(uint8_t opcode) const
1249 : {
1250 : // opcode 低4位为 RtStarsMemcpyAsyncOperationKind: ADD=0x01, MAX=0x02, MIN=0x03
1251 : // 映射到 HcclReduceOp: SUM=0, MAX=2, MIN=3
1252 0 : const uint8_t opKind = opcode & 0x0F;
1253 0 : switch (opKind) {
1254 0 : case 0x01:
1255 0 : return static_cast<u8>(HCCL_REDUCE_SUM);
1256 0 : case 0x02:
1257 0 : return static_cast<u8>(HCCL_REDUCE_MAX);
1258 0 : case 0x03:
1259 0 : return static_cast<u8>(HCCL_REDUCE_MIN);
1260 0 : default:
1261 0 : return static_cast<u8>(HCCL_REDUCE_RESERVED);
1262 : }
1263 : }
1264 :
1265 0 : inline HcclResult AicpuTaskCacheEntry::FillSlotUbDma_(
1266 : Hccl::DfxTaskInfo* slot, const uint8_t* sqePtr, const DbSqeProfAndRefreshInfo& profAndRefreshInfo,
1267 : UbTransportLiteImpl* ubTransportLiteImplPtr, StreamLite* streamLite, u32 taskId) const
1268 : {
1269 0 : const DbSqeProfInfo& profInfo = profAndRefreshInfo.dbSqeProfInfo;
1270 0 : slot->taskType = static_cast<u8>(profInfo.taskParamType);
1271 0 : FillSlotCommonFields_(
1272 : slot, streamLite, taskId, Hccl::DfxLinkTypeVal::LINK_UB,
1273 : static_cast<u8>(Hccl::DfxTransportType::DFX_TRANSPORT_TYPE_UB), reinterpret_cast<u64>(ubTransportLiteImplPtr));
1274 0 : slot->taskPara.ubDma.sqeAddr = reinterpret_cast<u64>(sqePtr);
1275 0 : slot->taskPara.ubDma.srcAddr = profInfo.srcAddr;
1276 0 : slot->taskPara.ubDma.dstAddr = profInfo.dstAddr;
1277 0 : slot->taskPara.ubDma.size = profInfo.size;
1278 0 : if (static_cast<u8>(profInfo.taskParamType) == TaskParamTypeVal::TASK_UB_INLINE_WRITE) {
1279 0 : slot->taskPara.ubDma.notifyId = static_cast<u32>(profInfo.dstAddr);
1280 0 : } else if (static_cast<u8>(profInfo.taskParamType) == TaskParamTypeVal::TASK_UB) {
1281 0 : slot->taskPara.ubDma.notifyId = INVALID_U32;
1282 : } else {
1283 0 : slot->taskPara.ubDma.notifyId = static_cast<u32>(profInfo.notifyId);
1284 : }
1285 0 : return HCCL_SUCCESS;
1286 : }
1287 :
1288 0 : inline HcclResult AicpuTaskCacheEntry::FillSlotReduce_(
1289 : Hccl::DfxTaskInfo* slot, const uint8_t* sqePtr, const DbSqeProfAndRefreshInfo& profAndRefreshInfo,
1290 : UbTransportLiteImpl* ubTransportLiteImplPtr, StreamLite* streamLite, u32 taskId) const
1291 : {
1292 0 : const DbSqeProfInfo& profInfo = profAndRefreshInfo.dbSqeProfInfo;
1293 0 : slot->taskType = static_cast<u8>(profInfo.taskParamType);
1294 0 : FillSlotCommonFields_(
1295 : slot, streamLite, taskId, Hccl::DfxLinkTypeVal::LINK_UB,
1296 : static_cast<u8>(Hccl::DfxTransportType::DFX_TRANSPORT_TYPE_UB), reinterpret_cast<u64>(ubTransportLiteImplPtr));
1297 0 : slot->taskPara.Reduce.sqeAddr = reinterpret_cast<u64>(sqePtr);
1298 0 : slot->taskPara.Reduce.srcAddr = profInfo.srcAddr;
1299 0 : slot->taskPara.Reduce.dstAddr = profInfo.dstAddr;
1300 0 : slot->taskPara.Reduce.size = profInfo.size;
1301 0 : slot->taskPara.Reduce.reduceOp = static_cast<u8>(profInfo.reduceOp);
1302 0 : if (static_cast<u8>(profInfo.taskParamType) == TaskParamTypeVal::TASK_UB_REDUCE_INLINE) {
1303 0 : slot->taskPara.Reduce.notifyId = INVALID_U32;
1304 : } else {
1305 0 : slot->taskPara.Reduce.notifyId = static_cast<u32>(profInfo.notifyId);
1306 : }
1307 0 : return HCCL_SUCCESS;
1308 : }
1309 :
1310 0 : HcclResult AicpuTaskCacheEntry::ReportSqeProfiling_(
1311 : uint8_t* sqePtr, size_t arrayIdx, uint32_t sqeIdx, const uint64_t* baseAddrs, const uint64_t* memSizes,
1312 : const uint32_t count, StreamLite* streamLite, const u32 sqId)
1313 : {
1314 : // 注意: 参考aicpu_ts_thread.cc填充DfxTaskInfo并经NextTaskSlot上报
1315 :
1316 : // 获取SQE对应的sqeType和taskId
1317 0 : Rt91095StarsSqeHeader* sqeHeaderPtr = (Rt91095StarsSqeHeader*)sqePtr; // 已在AddSqeArray校验, 无需再校验
1318 0 : const Rt91095StarsSqeType sqeType = static_cast<Rt91095StarsSqeType>(sqeHeaderPtr->type);
1319 0 : const u32 taskId = (sqeHeaderPtr->taskId << 16) | (sqeHeaderPtr->rtStreamId);
1320 :
1321 0 : switch (sqeType) {
1322 0 : case Rt91095StarsSqeType::RT_91095_SQE_TYPE_UBDMA:
1323 : // DbSqe由ReportDbSqeProfiling_内部按需获取NextTaskSlot并填充
1324 0 : CHK_RET(
1325 : ReportDbSqeProfiling_(sqePtr, arrayIdx, sqeIdx, baseAddrs, memSizes, count, streamLite, sqId, taskId));
1326 0 : break;
1327 0 : case Rt91095StarsSqeType::RT_91095_SQE_TYPE_NOTIFY_RECORD:
1328 : case Rt91095StarsSqeType::RT_91095_SQE_TYPE_NOTIFY_WAIT: {
1329 0 : Hccl::DfxTaskInfo* slot = streamLite->NextTaskSlot();
1330 0 : CHK_RET(FillSlotNotify_(slot, sqePtr, streamLite, taskId));
1331 0 : break;
1332 : }
1333 0 : case Rt91095StarsSqeType::RT_91095_SQE_TYPE_SDMA: {
1334 0 : Hccl::DfxTaskInfo* slot = streamLite->NextTaskSlot();
1335 0 : CHK_RET(FillSlotSdma_(slot, sqePtr, streamLite, taskId));
1336 0 : break;
1337 : }
1338 0 : default:
1339 0 : HCCL_ERROR("[AicpuTaskCacheEntry][ReportSqeProfiling_] sqeType[%u] is unsupported", sqeType);
1340 0 : return HCCL_E_INTERNAL;
1341 : }
1342 0 : return HCCL_SUCCESS;
1343 : }
1344 :
1345 0 : inline HcclResult AicpuTaskCacheEntry::FillSlotNotify_(
1346 : Hccl::DfxTaskInfo* slot, const uint8_t* sqePtr, StreamLite* streamLite, u32 taskId) const
1347 : {
1348 0 : Rt91095StarsSqeHeader* sqeHeaderPtr = (Rt91095StarsSqeHeader*)sqePtr;
1349 0 : const Rt91095StarsSqeType sqeType = static_cast<Rt91095StarsSqeType>(sqeHeaderPtr->type);
1350 0 : slot->taskType = static_cast<u8>(
1351 : (sqeType == Rt91095StarsSqeType::RT_91095_SQE_TYPE_NOTIFY_RECORD) ? Hccl::TaskParamTypeVal::TASK_NOTIFY_RECORD :
1352 : Hccl::TaskParamTypeVal::TASK_NOTIFY_WAIT);
1353 0 : FillSlotCommonFields_(
1354 : slot, streamLite, taskId, Hccl::DfxLinkTypeVal::LINK_ONCHIP,
1355 : static_cast<u8>(Hccl::DfxTransportType::DFX_TRANSPORT_TYPE_LOCAL), DFX_INVALID_U64);
1356 0 : slot->taskPara.Notify.sqeAddr = reinterpret_cast<u64>(sqePtr);
1357 0 : return HCCL_SUCCESS;
1358 : }
1359 :
1360 0 : inline HcclResult AicpuTaskCacheEntry::FillSlotSdma_(
1361 : Hccl::DfxTaskInfo* slot, const uint8_t* sqePtr, StreamLite* streamLite, u32 taskId) const
1362 : {
1363 0 : Hccl::Rt91095StarsMemcpySqe* sdmaSqe = (Hccl::Rt91095StarsMemcpySqe*)sqePtr;
1364 0 : FillSlotCommonFields_(
1365 : slot, streamLite, taskId, Hccl::DfxLinkTypeVal::LINK_ONCHIP,
1366 : static_cast<u8>(Hccl::DfxTransportType::DFX_TRANSPORT_TYPE_LOCAL), DFX_INVALID_U64);
1367 0 : if (sdmaSqe->opcode == 0) {
1368 0 : slot->taskType = static_cast<u8>(Hccl::TaskParamTypeVal::TASK_SDMA);
1369 0 : slot->taskPara.Dma.sqeAddr = reinterpret_cast<u64>(sqePtr);
1370 : } else {
1371 0 : slot->taskType = static_cast<u8>(Hccl::TaskParamTypeVal::TASK_REDUCE_INLINE);
1372 0 : slot->taskPara.Reduce.sqeAddr = reinterpret_cast<u64>(sqePtr);
1373 : slot->taskPara.Reduce.srcAddr
1374 0 : = (static_cast<uint64_t>(sdmaSqe->u.strideMode0.srcAddrHigh) << 32) | sdmaSqe->u.strideMode0.srcAddrLow;
1375 : slot->taskPara.Reduce.dstAddr
1376 0 : = (static_cast<uint64_t>(sdmaSqe->u.strideMode0.dstAddrHigh) << 32) | sdmaSqe->u.strideMode0.dstAddrLow;
1377 0 : slot->taskPara.Reduce.size = sdmaSqe->u.strideMode0.lengthMove;
1378 0 : slot->taskPara.Reduce.notifyId = INVALID_U32;
1379 0 : slot->taskPara.Reduce.reduceOp = ConvertSdmaOpCodeToReduceOp_(sdmaSqe->opcode);
1380 : }
1381 0 : return HCCL_SUCCESS;
1382 : }
1383 :
1384 0 : HcclResult AicpuTaskCacheEntry::ReportSqeArrayProfiling_(
1385 : size_t arrayIdx, const uint64_t* baseAddrs, const uint64_t* memSizes, const uint32_t count)
1386 : {
1387 : // 注意: arrayIdx已在SubmitCacheEntry_校验, 这里无需重复校验
1388 0 : const SqeArrayInfo& sqeArrayInfo = sqeArrayInfos_[arrayIdx];
1389 :
1390 : // sqe数组
1391 0 : uint8_t* sqePtr = sqeArrayInfo.sqeArray; // 注意: sqePtr已在AddSqeArray校验, 无需再校验
1392 0 : uint64_t sqeCount = sqeArrayInfo.sqeCount;
1393 :
1394 : // 获取SQE对应的sqId
1395 : // 注意: aicpuTsThreadPtr已在AddSqeArray校验, 无需再校验
1396 0 : StreamLite* streamLite = reinterpret_cast<StreamLite*>(sqeArrayInfo.aicpuTsThreadPtr->GetStreamLitePtr());
1397 0 : CHK_PTR_NULL(streamLite);
1398 0 : const u32 sqId = streamLite->GetSqId();
1399 0 : for (size_t sqeIdx = 0; sqeIdx < sqeCount; ++sqeIdx) {
1400 0 : CHK_PRT(ReportSqeProfiling_(sqePtr, arrayIdx, sqeIdx, baseAddrs, memSizes, count, streamLite, sqId));
1401 :
1402 : // 切换到下一个SQE
1403 0 : sqePtr += AC_SQE_SIZE;
1404 : }
1405 :
1406 0 : return HCCL_SUCCESS;
1407 : }
1408 :
1409 : } // namespace hcomm
|
