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 :
14 : #include "op_unfold_cache_entry.h"
15 :
16 : #include "aicpu_hccl_sqcq.h"
17 : #include "aicpu_hccl_sqcqv1.h"
18 : #include "aicpu_hccl_sqcqv2.h"
19 : #include "dispatcher_pub.h" // HCCL_SDMA_MAX_COUNT_4GB
20 : #include "log.h"
21 : #include "sal.h"
22 :
23 : namespace hccl {
24 :
25 : // struct OpUnfoldMemRange
26 :
27 0 : OpUnfoldMemRange::OpUnfoldMemRange() : isValid(false), baseAddr(0), memSize(0) {}
28 :
29 0 : OpUnfoldMemRange::OpUnfoldMemRange(const uint64_t curBaseAddr, const uint64_t curMemSize)
30 0 : : isValid(true),
31 0 : baseAddr(curBaseAddr),
32 0 : memSize(curMemSize)
33 : {
34 : // 检查地址有效性
35 0 : CHK_PRT_CONT(curBaseAddr == 0, HCCL_ERROR("[OpUnfoldMemRange][OpUnfoldMemRange] curBaseAddr is 0"));
36 0 : }
37 :
38 0 : OpUnfoldMemRange::OpUnfoldMemRange(const OpUnfoldMemRange& other)
39 0 : : isValid(other.isValid),
40 0 : baseAddr(other.baseAddr),
41 0 : memSize(other.memSize)
42 0 : {}
43 :
44 0 : OpUnfoldMemRange::~OpUnfoldMemRange() {}
45 :
46 0 : const OpUnfoldMemRange& OpUnfoldMemRange::operator=(const OpUnfoldMemRange& other)
47 : {
48 0 : if (this != &other) {
49 0 : isValid = other.isValid;
50 0 : baseAddr = other.baseAddr;
51 0 : memSize = other.memSize;
52 : }
53 0 : return *this;
54 : }
55 :
56 0 : HcclResult OpUnfoldMemRange::GetEndAddr(uint64_t& endAddr) const
57 : {
58 : // 检查地址是否溢出
59 0 : CHK_PRT_RET(
60 : baseAddr + memSize < baseAddr,
61 : HCCL_ERROR("[OpUnfoldMemRange][InRange] baseAddr[0x%016llx] + memSize[%llu] overflows", baseAddr, memSize),
62 : HCCL_E_INTERNAL);
63 :
64 0 : endAddr = baseAddr + memSize;
65 0 : return HCCL_SUCCESS;
66 : }
67 :
68 0 : HcclResult OpUnfoldMemRange::InRange(const uint64_t addr, bool& isInRange) const
69 : {
70 0 : uint64_t endAddr = 0;
71 0 : CHK_RET(GetEndAddr(endAddr));
72 :
73 0 : if (isValid && addr >= baseAddr && addr < endAddr) {
74 0 : isInRange = true;
75 : } else {
76 0 : isInRange = false;
77 : }
78 :
79 0 : return HCCL_SUCCESS;
80 : }
81 :
82 : // struct RefreshAddrInfo
83 :
84 0 : RefreshAddrInfo::RefreshAddrInfo() : rankId(INVALID_VALUE_RANKID), memType(RefreshAddrInfo::INVALID_MEMTYPE) {}
85 :
86 0 : RefreshAddrInfo::RefreshAddrInfo(const uint32_t curRankId, const uint8_t curMemType)
87 0 : : rankId(curRankId),
88 0 : memType(curMemType)
89 : {
90 : // 注意: rankId可以为INVALID_VALUE_RANKID, 表示访问本地rank
91 0 : CHK_PRT_CONT(memType == INVALID_MEMTYPE, HCCL_ERROR("[RefreshAddrInfo][RefreshAddrInfo] invalid memType"));
92 0 : }
93 :
94 0 : RefreshAddrInfo::RefreshAddrInfo(const RefreshAddrInfo& other) : rankId(other.rankId), memType(other.memType) {}
95 :
96 0 : RefreshAddrInfo::~RefreshAddrInfo() {}
97 :
98 0 : const RefreshAddrInfo& RefreshAddrInfo::operator=(const RefreshAddrInfo& other)
99 : {
100 0 : if (this != &other) {
101 0 : rankId = other.rankId;
102 0 : memType = other.memType;
103 : }
104 0 : return *this;
105 : }
106 :
107 : // struct AlltoallvMetadata
108 :
109 33 : AlltoallvMetadata::AlltoallvMetadata()
110 : {
111 33 : sdmaDataBlockSize = 0;
112 33 : hcclInputMemRanges.clear();
113 33 : notifyIdRankRflagMap.clear();
114 33 : signalAddrRankRflagMap.clear();
115 :
116 33 : hcclOffsetDstRanksIdxMap.clear();
117 33 : }
118 :
119 0 : void AlltoallvMetadata::Clear()
120 : {
121 0 : sdmaDataBlockSize = 0;
122 0 : hcclInputMemRanges.clear();
123 0 : notifyIdRankRflagMap.clear();
124 0 : signalAddrRankRflagMap.clear();
125 0 : hcclOffsetDstRanksIdxMap.clear();
126 0 : return;
127 : }
128 :
129 0 : HcclResult AlltoallvMetadata::Check(const bool afterFirstOrch) const
130 : {
131 0 : CHK_PRT_RET(
132 : sdmaDataBlockSize == 0, HCCL_ERROR("[AlltoallvMetadata][Check] sdmaDataBlockSize is zero"), HCCL_E_INTERNAL);
133 :
134 0 : const uint32_t rankSize = hcclInputMemRanges.size();
135 0 : CHK_PRT_RET(rankSize == 0, HCCL_ERROR("[AlltoallvMetadata][Check] empty hcclInputMemRanges"), HCCL_E_INTERNAL);
136 : // 注意: 每个remote rank各有两个NotifyId/SignalAddr分别用于send/recv count对应的Wait/Record同步
137 0 : CHK_PRT_RET(
138 : notifyIdRankRflagMap.size() != 2 * (rankSize - 1),
139 : HCCL_ERROR(
140 : "[AlltoallvMetadata][Check] notifyIdRankRflagMap.size[%u] != rankSize-1[%u]", notifyIdRankRflagMap.size(),
141 : 2 * (rankSize - 1)),
142 : HCCL_E_INTERNAL);
143 0 : CHK_PRT_RET(
144 : signalAddrRankRflagMap.size() != 2 * (rankSize - 1),
145 : HCCL_ERROR(
146 : "[AlltoallvMetadata][Check] signalAddrRankRflagMap.size[%u] != rankSize-1[%u]",
147 : signalAddrRankRflagMap.size(), 2 * (rankSize - 1)),
148 : HCCL_E_INTERNAL);
149 :
150 : // 注意: 只有在第一次cache miss的executor->Orchestrate之后, 相关mapping才会被初始化
151 0 : if (afterFirstOrch) {
152 0 : CHK_PRT_RET(
153 : ((rankSize > 1) && (hcclOffsetDstRanksIdxMap.size() == 0)),
154 : HCCL_ERROR("[AlltoallvMetadata][Check] empty hcclOffsetDstRanksIdxMap for rankSize[%u]", rankSize),
155 : HCCL_E_INTERNAL); // 注意: 只有当rankSize为1时, hcclOffsetDstRanksIdxMap的size才可以是0
156 : }
157 :
158 0 : return HCCL_SUCCESS;
159 : }
160 :
161 : // struct AlltoallvSendRecvInfo
162 :
163 0 : AlltoallvSendRecvInfo::AlltoallvSendRecvInfo()
164 : {
165 0 : sendType = HcclDataType::HCCL_DATA_TYPE_RESERVED;
166 0 : recvType = HcclDataType::HCCL_DATA_TYPE_RESERVED;
167 0 : sendCounts.clear();
168 0 : recvCounts.clear();
169 0 : sendOffsets.clear();
170 0 : recvOffsets.clear();
171 :
172 0 : return;
173 : }
174 :
175 0 : HcclResult AlltoallvSendRecvInfo::Check() const
176 : {
177 0 : CHK_PRT_RET(
178 : sendType == HcclDataType::HCCL_DATA_TYPE_RESERVED,
179 : HCCL_ERROR("[AlltoallvSendRecvInfo][Check] sendType is reserved"), HCCL_E_INTERNAL);
180 0 : CHK_PRT_RET(
181 : recvType == HcclDataType::HCCL_DATA_TYPE_RESERVED,
182 : HCCL_ERROR("[AlltoallvSendRecvInfo][Check] recvType is reserved"), HCCL_E_INTERNAL);
183 :
184 0 : const uint32_t rankSize = sendCounts.size();
185 0 : CHK_PRT_RET(rankSize == 0, HCCL_ERROR("[AlltoallvSendRecvInfo][Check] empty sendCounts"), HCCL_E_INTERNAL);
186 0 : CHK_PRT_RET(
187 : recvCounts.size() != rankSize,
188 : HCCL_ERROR("[AlltoallvSendRecvInfo][Check] recvCounts.size[%u] != rankSize[%u]", recvCounts.size(), rankSize),
189 : HCCL_E_INTERNAL);
190 0 : CHK_PRT_RET(
191 : sendOffsets.size() != rankSize,
192 : HCCL_ERROR("[AlltoallvSendRecvInfo][Check] sendOffsets.size[%u] != rankSize[%u]", sendOffsets.size(), rankSize),
193 : HCCL_E_INTERNAL);
194 0 : CHK_PRT_RET(
195 : recvOffsets.size() != rankSize,
196 : HCCL_ERROR("[AlltoallvSendRecvInfo][Check] recvOffsets.size[%u] != rankSize[%u]", recvOffsets.size(), rankSize),
197 : HCCL_E_INTERNAL);
198 :
199 0 : return HCCL_SUCCESS;
200 : }
201 :
202 : // class OpUnfoldCacheEntry
203 :
204 0 : OpUnfoldCacheEntry::OpUnfoldCacheEntry(
205 0 : const std::vector<OpUnfoldMemRange>& userInputMemRanges, const std::vector<OpUnfoldMemRange>& userOutputMemRanges)
206 0 : : userInputMemRanges_(userInputMemRanges),
207 0 : userOutputMemRanges_(userOutputMemRanges)
208 : {
209 0 : HCCL_INFO(
210 : "[OpUnfoldCacheEntry][OpUnfoldCacheEntry] create a cache entry with %llu userInputMemRanges and %llu "
211 : "userOutputMemRanges",
212 : userInputMemRanges_.size(), userOutputMemRanges_.size());
213 0 : }
214 :
215 0 : OpUnfoldCacheEntry::~OpUnfoldCacheEntry()
216 : {
217 0 : size_t sqeArrayCount = sqeArrays_.size();
218 0 : size_t totalSqeCount = 0;
219 0 : for (size_t arrayIdx = 0; arrayIdx < sqeArrayCount; ++arrayIdx) {
220 0 : totalSqeCount += srcRefreshAddrInfoArrays_[arrayIdx].size();
221 :
222 : // 如果存在当前这段连续的SQE数组，则指向内容必不为空
223 : // 因为SQE数量为0时, DispatcherAicpu::LaunchTask()会直接返回, 不会添加SQE到OpUnfoldCache中
224 0 : uint8_t* curSqeArray = sqeArrays_[arrayIdx];
225 :
226 : // 释放当前SQE数组
227 0 : if (UNLIKELY(curSqeArray == nullptr)) { // 不能使用CHK_PTR_NULL，因为会return HcclResult
228 0 : HCCL_ERROR("[OpUnfoldCacheEntry][~OpUnfoldCacheEntry] curSqeArray is nullptr");
229 : } else {
230 0 : free(curSqeArray);
231 0 : curSqeArray = nullptr;
232 : }
233 :
234 : // 同理释放其他空间
235 :
236 : // 释放当前SQE type数组
237 0 : uint8_t* curSqeTypeArray = sqeTypeArrays_[arrayIdx];
238 0 : if (UNLIKELY(curSqeTypeArray == nullptr)) {
239 0 : HCCL_ERROR("[OpUnfoldCacheEntry][~OpUnfoldCacheEntry] curSqeTypeArray is nullptr");
240 : } else {
241 0 : free(curSqeTypeArray);
242 0 : curSqeTypeArray = nullptr;
243 : }
244 :
245 : // 释放当前SQE DfxInfo数组
246 0 : AicpuDfxInfo* curSqeDfxInfoArray = sqeDfxInfoArrays_[arrayIdx];
247 0 : if (UNLIKELY(curSqeDfxInfoArray == nullptr)) {
248 0 : HCCL_ERROR("[OpUnfoldCacheEntry][~OpUnfoldCacheEntry] curSqeDfxInfoArray is nullptr");
249 : } else {
250 0 : free(curSqeDfxInfoArray);
251 0 : curSqeDfxInfoArray = nullptr;
252 : }
253 : }
254 :
255 0 : HCCL_INFO(
256 : "[OpUnfoldCacheEntry][~OpUnfoldCacheEntry] release %u SQE arrays (%u SQEs in total) from the cache entry",
257 : sqeArrayCount, totalSqeCount);
258 0 : }
259 :
260 0 : HcclResult OpUnfoldCacheEntry::GetSqeArrayCount(size_t& sqeArrayCount) const
261 : {
262 0 : sqeArrayCount = sqeArrays_.size();
263 0 : CHK_PRT_RET(
264 : sqeArrayCount == 0, HCCL_ERROR("[OpUnfoldCacheEntry][OpUnfoldCacheEntry] sqeArrayCount is 0"), HCCL_E_INTERNAL);
265 0 : return HCCL_SUCCESS;
266 : }
267 :
268 0 : HcclResult OpUnfoldCacheEntry::AllocSqeArray(const size_t sqeCount, const int32_t streamId, size_t& arrayIdx)
269 : {
270 : // Allocate a new SQE array
271 0 : const size_t sqeBytes = sqeCount * HCCL_SQE_SIZE;
272 0 : uint8_t* newSqeArray = reinterpret_cast<uint8_t*>(malloc(sqeBytes));
273 0 : CHK_PTR_NULL(newSqeArray);
274 0 : sqeArrays_.emplace_back(newSqeArray);
275 :
276 : // Allocate a new SQE type array
277 0 : const size_t sqeTypeBytes = sqeCount * sizeof(uint8_t);
278 0 : uint8_t* newSqeTypeArray = reinterpret_cast<uint8_t*>(malloc(sqeTypeBytes));
279 0 : CHK_PTR_NULL(newSqeTypeArray);
280 0 : sqeTypeArrays_.emplace_back(newSqeTypeArray);
281 :
282 : // Allocate a new SQE DFX info array
283 0 : const size_t sqeDfxInfoBytes = sqeCount * sizeof(AicpuDfxInfo);
284 0 : AicpuDfxInfo* newSqeDfxInfoArray = reinterpret_cast<AicpuDfxInfo*>(malloc(sqeDfxInfoBytes));
285 0 : CHK_PTR_NULL(newSqeDfxInfoArray);
286 0 : sqeDfxInfoArrays_.emplace_back(newSqeDfxInfoArray);
287 :
288 : // Copy stream pointer
289 0 : CHK_PRT_RET(
290 : streamId < 0, HCCL_ERROR("[OpUnfoldCacheEntry][AllocSqeArray] streamId %d < 0", streamId), HCCL_E_INTERNAL);
291 0 : streamIds_.emplace_back(streamId);
292 :
293 : // 注意: streamSeqIdxes_在cache miss LaunchTask()结束后, HcclCommAicpu通过CalcStreamSeqIdxes更新
294 :
295 : // 初始化src/dst RefreshAddrInfo
296 0 : srcRefreshAddrInfoArrays_.emplace_back(sqeCount);
297 0 : dstRefreshAddrInfoArrays_.emplace_back(sqeCount);
298 :
299 : // Set index of allocated array
300 0 : arrayIdx = sqeArrays_.size() - 1;
301 :
302 0 : HCCL_INFO(
303 : "[OpUnfoldCacheEntry][AllocSqeArray] allocate %uth sqe array with sqeCount of %u and streamId of %d", arrayIdx,
304 : sqeCount, streamId);
305 :
306 0 : return HCCL_SUCCESS;
307 : }
308 :
309 0 : HcclResult OpUnfoldCacheEntry::MemcpySqeArray(
310 : const size_t arrayIdx, const size_t sqeStartIdx, const size_t sqeCount, const uint8_t* sqeArray,
311 : const uint8_t* sqeTypeArray, const AicpuDfxInfo* sqeDfxInfoArray, const bool isAlltoallv,
312 : const AlltoallvMetadata* alltoallvMetadataPtr)
313 : {
314 : // Copy sqeArray[0:sqeCount) -> sqeArrays_[arrayIdx][sqeStartIdx:sqeStartIdx+sqeCount)
315 :
316 : // 检验入参
317 0 : CHK_PRT_RET(
318 : arrayIdx >= sqeArrays_.size(),
319 : HCCL_ERROR(
320 : "[OpUnfoldCacheEntry][MemcpySqeArray] arrayIdx %u is out of range [0, %u)", arrayIdx, sqeArrays_.size()),
321 : HCCL_E_INTERNAL);
322 0 : const size_t totalSqeCount = srcRefreshAddrInfoArrays_[arrayIdx].size();
323 0 : CHK_PRT_RET(
324 : sqeStartIdx + sqeCount - 1 >= totalSqeCount,
325 : HCCL_ERROR(
326 : "[OpUnfoldCacheEntry][MemcpySqeArray] sqeStartIdx %u + sqeCount %u - 1 is out of range [0, %u)",
327 : sqeStartIdx, sqeCount, totalSqeCount),
328 : HCCL_E_INTERNAL);
329 0 : CHK_PTR_NULL(sqeArray);
330 0 : CHK_PTR_NULL(sqeTypeArray);
331 0 : CHK_PTR_NULL(sqeDfxInfoArray);
332 0 : if (isAlltoallv) {
333 0 : CHK_PTR_NULL(alltoallvMetadataPtr);
334 0 : CHK_RET(alltoallvMetadataPtr->Check(false));
335 : }
336 :
337 0 : HCCL_INFO(
338 : "[OpUnfoldCacheEntry][MemcpySqeArray] memcpy %uth sqe array[%u:%u]; isAlltoallv[%u]", arrayIdx, sqeStartIdx,
339 : sqeStartIdx + sqeCount - 1, isAlltoallv);
340 :
341 : // Copy SQE content
342 0 : const size_t sqeBytes = sqeCount * HCCL_SQE_SIZE;
343 0 : uint8_t* dstSqeArray = sqeArrays_[arrayIdx];
344 0 : CHK_PTR_NULL(dstSqeArray);
345 0 : CHK_SAFETY_FUNC_RET(memcpy_s(
346 : dstSqeArray + sqeStartIdx * HCCL_SQE_SIZE, (totalSqeCount - sqeStartIdx) * HCCL_SQE_SIZE, sqeArray, sqeBytes));
347 :
348 : // Copy SQE type
349 0 : const size_t sqeTypeBytes = sqeCount * sizeof(uint8_t);
350 0 : uint8_t* dstSqeTypeArray = sqeTypeArrays_[arrayIdx];
351 0 : CHK_PTR_NULL(dstSqeTypeArray);
352 0 : CHK_SAFETY_FUNC_RET(memcpy_s(
353 : dstSqeTypeArray + sqeStartIdx, (totalSqeCount - sqeStartIdx) * sizeof(uint8_t), sqeTypeArray, sqeTypeBytes));
354 :
355 : // Copy SQE DFX info
356 0 : const size_t sqeDfxInfoBytes = sqeCount * sizeof(AicpuDfxInfo);
357 0 : AicpuDfxInfo* dstSqeDfxInfoArray = sqeDfxInfoArrays_[arrayIdx];
358 0 : CHK_PTR_NULL(dstSqeDfxInfoArray);
359 0 : CHK_SAFETY_FUNC_RET(memcpy_s(
360 : dstSqeDfxInfoArray + sqeStartIdx, (totalSqeCount - sqeStartIdx) * sizeof(AicpuDfxInfo), sqeDfxInfoArray,
361 : sqeDfxInfoBytes));
362 :
363 : // 遍历SQE, 根据type更新src/dst RefreshAddrInfo
364 0 : std::vector<RefreshAddrInfo>& srcRefreshAddrInfoArray = srcRefreshAddrInfoArrays_[arrayIdx];
365 0 : std::vector<RefreshAddrInfo>& dstRefreshAddrInfoArray = dstRefreshAddrInfoArrays_[arrayIdx];
366 0 : uint64_t sqeSrcAddr = 0;
367 0 : uint64_t sqeDstAddr = 0;
368 0 : const uint8_t* sqePtr = sqeArray;
369 0 : for (size_t tmpSqeIdx = 0; tmpSqeIdx < sqeCount; tmpSqeIdx++) {
370 0 : const size_t cacheSqeIdx = sqeStartIdx + tmpSqeIdx;
371 :
372 : // 获得当前SQE的信息
373 : // 注意: 不使用sqeDfxInfoArray[tmpSqeIdx].remoteRank来准备RefreshAddrInfo,
374 : // 因为DfxInfo.remoteRank某些整网用例下存在维护异常
375 0 : const uint8_t sqeType = sqeTypeArray[tmpSqeIdx];
376 :
377 : // 根据SQE type更新RefreshAddrInfo
378 0 : switch (sqeType) {
379 0 : case SqeType::NOTIFY_SQE:
380 : case SqeType::EVENT_SQE: {
381 : // No need to update src/dst RefreshAddrInfo due to no addr fields
382 0 : break;
383 : }
384 0 : case SqeType::WRITE_VALUE_SQE:
385 : case SqeType::RDMA_DB_SEND_SQE: {
386 0 : const rtStarsWriteValueSqe_t* writeValueSqePtr
387 : = reinterpret_cast<const rtStarsWriteValueSqe_t*>(sqePtr);
388 :
389 0 : CombineUint32ToUint64(sqeDstAddr, writeValueSqePtr->write_addr_high, writeValueSqePtr->write_addr_low);
390 0 : CHK_RET(CheckAndPrepareRefreshAddrInfo(
391 : sqeDstAddr, dstRefreshAddrInfoArray[cacheSqeIdx], false,
392 : nullptr)); // 注意: alltoallv算子有WRITE_VALUE_SQE, 但不会存在对于HCCL input buffer的访问,
393 : // 无需确认HCCL input buffer对应的rank id进行地址刷新
394 :
395 0 : break;
396 : }
397 0 : case SqeType::MEMCPY_ASYNC_SQE: {
398 0 : const rtStarsMemcpyAsyncSqe_t* memcpyAsyncSqePtr
399 : = reinterpret_cast<const rtStarsMemcpyAsyncSqe_t*>(sqePtr);
400 :
401 0 : CombineUint32ToUint64(sqeSrcAddr, memcpyAsyncSqePtr->src_addr_high, memcpyAsyncSqePtr->src_addr_low);
402 0 : CHK_RET(CheckAndPrepareRefreshAddrInfo(
403 : sqeSrcAddr, srcRefreshAddrInfoArray[cacheSqeIdx], isAlltoallv, alltoallvMetadataPtr));
404 :
405 0 : CombineUint32ToUint64(sqeDstAddr, memcpyAsyncSqePtr->dst_addr_high, memcpyAsyncSqePtr->dst_addr_low);
406 0 : CHK_RET(CheckAndPrepareRefreshAddrInfo(
407 : sqeDstAddr, dstRefreshAddrInfoArray[cacheSqeIdx], isAlltoallv, alltoallvMetadataPtr));
408 :
409 0 : break;
410 : }
411 0 : case SqeType::CCORE_WAIT_START_SQE: {
412 0 : HCCL_ERROR("[OpUnfoldCacheEntry][MemcpySqeArray] SqeType::CCORE_WAIT_START_SQE is not supported in A3");
413 0 : return HCCL_E_NOT_SUPPORT;
414 : }
415 0 : case SqeType::CCORE_WRITE_VALUE_SQE: {
416 0 : HCCL_ERROR(
417 : "[OpUnfoldCacheEntry][MemcpySqeArray] SqeType::CCORE_WRITE_VALUE_SQE is not supported in A3");
418 0 : return HCCL_E_NOT_SUPPORT;
419 : }
420 0 : case SqeType::NOTIFY_SQE_V2: {
421 0 : HCCL_ERROR("[OpUnfoldCacheEntry][MemcpySqeArray] SqeType::NOTIFY_SQE_V2 is not supported in A3");
422 0 : return HCCL_E_NOT_SUPPORT;
423 : }
424 0 : case SqeType::WRITE_VALUE_SQE_V2: {
425 0 : HCCL_ERROR("[OpUnfoldCacheEntry][MemcpySqeArray] SqeType::WRITE_VALUE_SQE_V2 is not supported in A3");
426 0 : return HCCL_E_NOT_SUPPORT;
427 : }
428 0 : case SqeType::EVENT_SQE_V2: {
429 0 : HCCL_ERROR("[OpUnfoldCacheEntry][MemcpySqeArray] SqeType::EVENT_SQE_V2 is not supported in A3");
430 0 : return HCCL_E_NOT_SUPPORT;
431 : }
432 0 : case SqeType::MEMCPY_ASYNC_SQE_V2: {
433 0 : HCCL_ERROR("[OpUnfoldCacheEntry][MemcpySqeArray] SqeType::MEMCPY_ASYNC_SQE_V2 is not supported in A3");
434 0 : return HCCL_E_NOT_SUPPORT;
435 : }
436 0 : case SqeType::FLIP_PLACEHOLDER_SQE: {
437 0 : HCCL_ERROR(
438 : "[OpUnfoldCacheEntry][MemcpySqeArray] placeholder should not be cached, sqeType[%u] tmpSqeIdx[%u] "
439 : "cacheSqeIdx[%u]",
440 : sqeType, tmpSqeIdx, cacheSqeIdx);
441 0 : return HCCL_E_INTERNAL;
442 : }
443 0 : case SqeType::CACHE_MEMCPY_PLACEHOLDER_SQE: {
444 : // 注意: 只有alltoallv算子在cache时才会有此类SQE
445 0 : CHK_PRT_RET(
446 : !isAlltoallv,
447 : HCCL_ERROR("[OpUnfoldCacheEntry][MemcpySqeArray] non-alltoallv op should not dispatch "
448 : "CACHE_MEMCPY_PLACEHOLDER_SQE"),
449 : HCCL_E_INTERNAL);
450 :
451 0 : const rtStarsPlaceHolderSqe_t* placeholderSqePtr
452 : = reinterpret_cast<const rtStarsPlaceHolderSqe_t*>(sqePtr);
453 0 : const struct rtCacheMemcpyTaskTag_t& unfoldCacheTaskTag = placeholderSqePtr->u.cache_memcpy_task_info;
454 :
455 0 : CombineUint32ToUint64(sqeSrcAddr, unfoldCacheTaskTag.src_addr_high, unfoldCacheTaskTag.src_addr_low);
456 0 : CHK_RET(CheckAndPrepareRefreshAddrInfo(
457 : sqeSrcAddr, srcRefreshAddrInfoArray[cacheSqeIdx], isAlltoallv, alltoallvMetadataPtr));
458 :
459 0 : CombineUint32ToUint64(sqeDstAddr, unfoldCacheTaskTag.dst_addr_high, unfoldCacheTaskTag.dst_addr_low);
460 0 : CHK_RET(CheckAndPrepareRefreshAddrInfo(
461 : sqeDstAddr, dstRefreshAddrInfoArray[cacheSqeIdx], isAlltoallv, alltoallvMetadataPtr));
462 :
463 0 : break;
464 : }
465 0 : case SqeType::CACHE_NOTIFY_PLACEHOLDER_SQE:
466 : case SqeType::CACHE_WRITE_VALUE_PLACEHOLDER_SQE:
467 : case SqeType::CACHE_MEMCPY_RECORD_PLACEHOLDER_SQE: {
468 : // Cache-notify placeholder: 无地址字段, 不需要更新src/dst RefreshAddrInfo
469 : // Cache-write placeholder: 对应WriteValueRecord SQE, 地址为硬件映射的固定内存, 无需更新RefreshAddrInfo
470 : // Cache-memcpy-record placeholder: 对应MemcpyRecord SQE, 地址为硬件映射的固定内存,
471 : // 无需更新RefreshAddrInfo
472 0 : break;
473 : }
474 0 : default: {
475 0 : HCCL_WARNING("[OpUnfoldCacheEntry][MemcpySqeArray] sqeType %u is unsupported", sqeType);
476 0 : return HCCL_E_NOT_SUPPORT;
477 : }
478 : }
479 :
480 0 : sqePtr += HCCL_SQE_SIZE;
481 : }
482 :
483 0 : return HCCL_SUCCESS;
484 : }
485 :
486 0 : HcclResult OpUnfoldCacheEntry::CalcStreamSeqIdxes(Stream& mainStream, std::vector<Stream>& slaveStreams)
487 : {
488 0 : const size_t streamIdCount = streamIds_.size();
489 0 : HCCL_INFO(
490 : "[OpUnfoldCacheEntry][CalcStreamSeqIdxes] calculate stream sequential indexes for %u stream ids",
491 : streamIdCount);
492 :
493 : // 对每个stream id找到对应的sequential stream index
494 0 : streamSeqIdxes_.resize(streamIdCount);
495 0 : for (size_t i = 0; i < streamIdCount; ++i) {
496 0 : const int32_t curStreamId = streamIds_[i];
497 :
498 0 : if (curStreamId == mainStream.GetHcclStreamInfo().actualStreamId) { // 主流
499 0 : streamSeqIdxes_[i] = 0;
500 : } else { // 遍历从流
501 0 : bool isFound = false;
502 0 : for (size_t j = 0; j < slaveStreams.size(); ++j) {
503 0 : if (curStreamId == slaveStreams[j].GetHcclStreamInfo().actualStreamId) { // 匹配某个从流
504 0 : streamSeqIdxes_[i] = j + 1;
505 0 : isFound = true;
506 0 : break;
507 : }
508 : }
509 :
510 : // No stream can match the stream id
511 0 : if (!isFound) {
512 0 : HCCL_ERROR(
513 : "[OpUnfoldCacheEntry][CalcStreamSeqIdxes] cannot find any stream to match streamId %u",
514 : curStreamId);
515 0 : return HCCL_E_INTERNAL;
516 : }
517 : }
518 : }
519 :
520 0 : return HCCL_SUCCESS;
521 : }
522 :
523 : HcclResult
524 0 : OpUnfoldCacheEntry::UpdateRefreshAddrInfoForAlltoallv(const uint32_t curRank, AlltoallvMetadata& alltoallvMetadata)
525 : {
526 : // 校验入参
527 0 : CHK_RET(alltoallvMetadata.Check(true));
528 :
529 : // 获得rankSize
530 0 : const uint32_t rankSize = alltoallvMetadata.hcclInputMemRanges.size();
531 0 : CHK_PRT_RET(
532 : curRank >= rankSize,
533 : HCCL_ERROR(
534 : "[OpUnfoldCacheEntry][UpdateRefreshAddrInfoForAlltoallv] curRank[%u] >= rankSize[%u]", curRank, rankSize),
535 : HCCL_E_INTERNAL);
536 :
537 0 : HCCL_RUN_INFO(
538 : "[OpUnfoldCacheEntry][UpdateRefreshAddrInfoForAlltoallv] curRank[%u] rankSize[%u]", curRank, rankSize);
539 :
540 : // 遍历每一段SQE数组
541 0 : const size_t sqeArrayCnt = dstRefreshAddrInfoArrays_.size();
542 0 : for (size_t arrayIdx = 0; arrayIdx < sqeArrayCnt; ++arrayIdx) {
543 : // 准备当前SQE数组的相关metadata
544 0 : const uint8_t* sqeArray = sqeArrays_[arrayIdx];
545 0 : CHK_PTR_NULL(sqeArray);
546 0 : const uint8_t* sqeTypeArray = sqeTypeArrays_[arrayIdx];
547 0 : CHK_PTR_NULL(sqeTypeArray);
548 0 : std::vector<RefreshAddrInfo>& srcRefreshAddrInfoArray = srcRefreshAddrInfoArrays_[arrayIdx];
549 0 : std::vector<RefreshAddrInfo>& dstRefreshAddrInfoArray = dstRefreshAddrInfoArrays_[arrayIdx];
550 :
551 : // 遍历数组中的每个SQE
552 0 : const size_t sqeCount = dstRefreshAddrInfoArray.size();
553 0 : HCCL_INFO(
554 : "[OpUnfoldCacheEntry][UpdateRefreshAddrInfoForAlltoallv] sqeArrayIdx[%u] sqeCount[%u]", arrayIdx, sqeCount);
555 0 : for (size_t sqeIdx = 0; sqeIdx < sqeCount; ++sqeIdx) {
556 : // 准备当前SQE的相关metadata
557 0 : const uint8_t* sqePtr = sqeArray + sqeIdx * HCCL_SQE_SIZE;
558 0 : const uint8_t sqeType = sqeTypeArray[sqeIdx];
559 0 : RefreshAddrInfo& srcRefreshAddrInfo = srcRefreshAddrInfoArray[sqeIdx];
560 0 : RefreshAddrInfo& dstRefreshAddrInfo = dstRefreshAddrInfoArray[sqeIdx];
561 :
562 : // 只有memcpy / cache-memcpy placeholder SQE可能需要更新RefreshAddrInfo
563 0 : if (sqeType != SqeType::MEMCPY_ASYNC_SQE && sqeType != SqeType::CACHE_MEMCPY_PLACEHOLDER_SQE) {
564 0 : continue;
565 : }
566 :
567 : // 跳过memcpy record SQE (src/dst memType为invalid)
568 0 : const uint8_t srcMemType = srcRefreshAddrInfo.memType;
569 0 : const uint8_t dstMemType = dstRefreshAddrInfo.memType;
570 0 : if (sqeType == SqeType::MEMCPY_ASYNC_SQE && srcMemType == RefreshAddrInfo::INVALID_MEMTYPE
571 0 : && dstMemType == RefreshAddrInfo::INVALID_MEMTYPE) {
572 0 : const rtStarsMemcpyAsyncSqe_t* memcpySqePtr = reinterpret_cast<const rtStarsMemcpyAsyncSqe_t*>(sqePtr);
573 0 : HCCL_DEBUG(
574 : "[OpUnfoldCacheEntry][UpdateRefreshAddrInfoForAlltoallv] skip memcpy-record SQE: "
575 : "streamId[%u] taskId[%u] srcMemType[%u] dstMemType[%u]",
576 : memcpySqePtr->header.rtStreamId, memcpySqePtr->header.taskId, srcMemType, dstMemType);
577 0 : continue;
578 0 : }
579 :
580 : // (i) 更新invalid memType
581 :
582 : // 只有cache-memcpy placeholder才有可能存在user input/output endAddr导致的src/dst invalid memType
583 0 : if (sqeType == SqeType::CACHE_MEMCPY_PLACEHOLDER_SQE) {
584 : // 校验src/dst memType
585 0 : const rtStarsPlaceHolderSqe_t* placeholderSqePtr
586 : = reinterpret_cast<const rtStarsPlaceHolderSqe_t*>(sqePtr);
587 0 : const uint16_t placeholderStreamId = placeholderSqePtr->header.rtStreamId;
588 0 : const uint16_t placeholderTaskId = placeholderSqePtr->header.taskId;
589 0 : if (srcMemType == RefreshAddrInfo::INVALID_MEMTYPE && dstMemType != RefreshAddrInfo::INVALID_MEMTYPE) {
590 : // LocalCopy/PrepareIntraData: dst一定是local user output
591 : // PrepareIntraData: 虽然dst RefreshAddrInfo的rank尚未更新 (see当前函数的第(ii)部分),
592 : // memType一定是(local) hccl input
593 0 : CHK_PRT_RET(
594 : dstMemType != RefreshAddrInfo::USER_OUTPUT_MEMTYPE
595 : && dstMemType != RefreshAddrInfo::HCCL_INPUT_MEMTYPE,
596 : HCCL_ERROR(
597 : "[OpUnfoldCacheEntry][UpdateRefreshAddrInfoForAlltoallv] cache-memcpy placeholder: "
598 : "streamId[%u] taskId[%u] dstMemType[%u] != RefreshAddrInfo::HCCL_INPUT_MEMTYPE[%u]",
599 : placeholderStreamId, placeholderTaskId, dstMemType, RefreshAddrInfo::HCCL_INPUT_MEMTYPE),
600 : HCCL_E_INTERNAL);
601 0 : } else if (
602 0 : srcMemType != RefreshAddrInfo::INVALID_MEMTYPE && dstMemType == RefreshAddrInfo::INVALID_MEMTYPE) {
603 : // LocalCopy/RemoteCopy: src一定是local user input
604 : // RemoteCopy: src memType一定是(remote) hccl input
605 0 : CHK_PRT_RET(
606 : srcMemType != RefreshAddrInfo::USER_INPUT_MEMTYPE
607 : && srcMemType != RefreshAddrInfo::HCCL_INPUT_MEMTYPE,
608 : HCCL_ERROR(
609 : "[OpUnfoldCacheEntry][UpdateRefreshAddrInfoForAlltoallv] cache-memcpy placeholder: "
610 : "streamId[%u] taskId[%u] srcMemType[%u] != RefreshAddrInfo::HCCL_INPUT_MEMTYPE[%u]",
611 : placeholderStreamId, placeholderTaskId, srcMemType, RefreshAddrInfo::HCCL_INPUT_MEMTYPE),
612 : HCCL_E_INTERNAL);
613 : }
614 :
615 : // 更新src/dst memType
616 0 : const struct rtCacheMemcpyTaskTag_t& cacheMemcpyTaskInfo = placeholderSqePtr->u.cache_memcpy_task_info;
617 0 : if (srcMemType == RefreshAddrInfo::INVALID_MEMTYPE) { // LocalCopy/PrepareIntraData
618 : // srcAddr一定是local user input的end addr
619 0 : uint64_t sqeSrcAddr = 0;
620 0 : CombineUint32ToUint64(
621 0 : sqeSrcAddr, cacheMemcpyTaskInfo.src_addr_high, cacheMemcpyTaskInfo.src_addr_low);
622 0 : uint64_t localUserInputEndAddr = 0;
623 0 : CHK_RET(userInputMemRanges_[curRank].GetEndAddr(localUserInputEndAddr));
624 0 : CHK_PRT_RET(
625 : sqeSrcAddr != localUserInputEndAddr,
626 : HCCL_ERROR(
627 : "[OpUnfoldCacheEntry][UpdateRefreshAddrInfoForAlltoallv] sqeSrcAddr[0x%016llx]"
628 : "!= localUserInputEndAddr[0x%016llx]",
629 : sqeSrcAddr, localUserInputEndAddr),
630 : HCCL_E_INTERNAL);
631 :
632 : // 更新src memType和rank, 即local user input
633 0 : srcRefreshAddrInfo.memType = RefreshAddrInfo::USER_INPUT_MEMTYPE;
634 0 : srcRefreshAddrInfo.rankId = curRank;
635 0 : HCCL_INFO(
636 : "[OpUnfoldCacheEntry][UpdateRefreshAddrInfoForAlltoallv] update invalid src "
637 : "RefreshAddrInfo for cache-memcpy placeholder: streamId[%u] taskId[%u] memType[%u] rankId[%u]",
638 : placeholderStreamId, placeholderTaskId, srcRefreshAddrInfo.memType, srcRefreshAddrInfo.rankId);
639 : }
640 0 : if (dstMemType == RefreshAddrInfo::INVALID_MEMTYPE) { // LocalCopy/RemoteCopy
641 : // dstAddr一定是local user output的end addr
642 0 : uint64_t sqeDstAddr = 0;
643 0 : CombineUint32ToUint64(
644 0 : sqeDstAddr, cacheMemcpyTaskInfo.dst_addr_high, cacheMemcpyTaskInfo.dst_addr_low);
645 0 : uint64_t localUserOutputEndAddr = 0;
646 0 : CHK_RET(userOutputMemRanges_[curRank].GetEndAddr(localUserOutputEndAddr));
647 0 : CHK_PRT_RET(
648 : sqeDstAddr != localUserOutputEndAddr,
649 : HCCL_ERROR(
650 : "[OpUnfoldCacheEntry][UpdateRefreshAddrInfoForAlltoallv] sqeDstAddr[0x%016llx]"
651 : "!= localUserOutputEndAddr[0x%016llx]",
652 : sqeDstAddr, localUserOutputEndAddr),
653 : HCCL_E_INTERNAL);
654 :
655 : // 更新dst memType和rank, 即local user output
656 0 : dstRefreshAddrInfo.memType = RefreshAddrInfo::USER_OUTPUT_MEMTYPE;
657 0 : dstRefreshAddrInfo.rankId = curRank;
658 0 : HCCL_INFO(
659 : "[OpUnfoldCacheEntry][UpdateRefreshAddrInfoForAlltoallv] update invalid dst "
660 : "RefreshAddrInfo for cache-memcpy placeholder: streamId[%u] taskId[%u] memType[%u] rankId[%u]",
661 : placeholderStreamId, placeholderTaskId, dstRefreshAddrInfo.memType, dstRefreshAddrInfo.rankId);
662 : }
663 : }
664 :
665 : // (ii) 更新local dst rank (如果dst memType是local hccl input)
666 :
667 : // PrepareIntraData: local user input -> local hccl input
668 : // 注意: invalid memType已经在当前函数的第(i)部分被解决
669 0 : if (srcRefreshAddrInfo.memType == RefreshAddrInfo::USER_INPUT_MEMTYPE
670 0 : && dstRefreshAddrInfo.memType == RefreshAddrInfo::HCCL_INPUT_MEMTYPE) {
671 : // srcRank一定是current rank
672 0 : const uint32_t srcRank = srcRefreshAddrInfo.rankId;
673 0 : CHK_PRT_RET(
674 : srcRank != curRank,
675 : HCCL_ERROR(
676 : "[OpUnfoldCacheEntry][UpdateRefreshAddrInfoForAlltoallv] srcRank[%u] != curRank[%u], "
677 : "rankSize[%u]",
678 : srcRank, curRank, rankSize),
679 : HCCL_E_INTERNAL);
680 :
681 : // 注意: 对于alltoallv算子, 原始dstRefreshAddrInfo.rankId在MemcpySqeArray时被刷新
682 : // 因为alltoallvMetadata中的hcclOffsetDstRanksIdxMap要等第一次算子Orchestrate结束后才会被设置
683 : // 而根据范围dstAddr落在local hccl input范围内, 因此当时刷新的rankId一定为current rank
684 0 : CHK_PRT_RET(
685 : dstRefreshAddrInfo.rankId != curRank,
686 : HCCL_ERROR(
687 : "[OpUnfoldCacheEntry][UpdateRefreshAddrInfoForAlltoallv] original dstRank[%u] != curRank[%u]",
688 : dstRefreshAddrInfo.rankId, curRank),
689 : HCCL_E_INTERNAL);
690 :
691 : // 根据SQE类型获得dst addr
692 0 : uint64_t sqeDstAddr = 0;
693 0 : if (sqeType == SqeType::MEMCPY_ASYNC_SQE) {
694 0 : const rtStarsMemcpyAsyncSqe_t* memcpyAsyncSqePtr
695 : = reinterpret_cast<const rtStarsMemcpyAsyncSqe_t*>(sqePtr);
696 0 : CombineUint32ToUint64(
697 0 : sqeDstAddr, memcpyAsyncSqePtr->dst_addr_high, memcpyAsyncSqePtr->dst_addr_low);
698 : } else { // cache-memcpy placeholder
699 0 : const rtStarsPlaceHolderSqe_t* placeholderSqePtr
700 : = reinterpret_cast<const rtStarsPlaceHolderSqe_t*>(sqePtr);
701 0 : const struct rtCacheMemcpyTaskTag_t& cacheMemcpyTaskInfo
702 : = placeholderSqePtr->u.cache_memcpy_task_info;
703 0 : CombineUint32ToUint64(
704 0 : sqeDstAddr, cacheMemcpyTaskInfo.dst_addr_high, cacheMemcpyTaskInfo.dst_addr_low);
705 : }
706 :
707 : // 本函数只会在第一次cache miss后处理时调用, dst addr一定落在local hccl input范围内
708 0 : bool isInRange = false;
709 0 : const OpUnfoldMemRange& localHcclInputMemRange = alltoallvMetadata.hcclInputMemRanges[curRank];
710 0 : CHK_RET(localHcclInputMemRange.InRange(sqeDstAddr, isInRange));
711 0 : if (UNLIKELY(!isInRange)) {
712 0 : uint64_t localHcclInputEndAddr = 0;
713 0 : CHK_RET(localHcclInputMemRange.GetEndAddr(localHcclInputEndAddr));
714 0 : HCCL_ERROR(
715 : "[OpUnfoldCacheEntry][UpdateRefreshAddrInfoForAlltoallv] sqeDstAddr[0x%016llx] not "
716 : "in localHcclInputMemRange[0x%016llx -- 0x%016llx]",
717 : sqeDstAddr, localHcclInputMemRange.baseAddr, localHcclInputEndAddr);
718 0 : return HCCL_E_INTERNAL;
719 : }
720 :
721 : // 参考alltoallv_direct_fullmesh.cc中的UpdateCurrRankSendInfo
722 :
723 : // 计算local hccl input buffer下的offset
724 : // 注意: 前面isInRange已经校验过必定为true, 即sqeDstAddr一定 >= localHcclInputBaseAddr
725 0 : uint64_t localHcclInputBaseAddr = localHcclInputMemRange.baseAddr;
726 0 : uint64_t hcclOffset = sqeDstAddr - localHcclInputBaseAddr;
727 :
728 : // 根据hcclOffset-dstRanksInfo mapping获得对应的dst rank
729 : std::unordered_map<uint64_t, RanksIdx>::iterator mapIter
730 0 : = alltoallvMetadata.hcclOffsetDstRanksIdxMap.find(hcclOffset);
731 0 : if (mapIter == alltoallvMetadata.hcclOffsetDstRanksIdxMap.end()) {
732 0 : HCCL_ERROR(
733 : "[OpUnfoldCacheEntry][UpdateRefreshAddrInfoForAlltoallv] hcclOffset[%u] is not found",
734 : hcclOffset);
735 0 : return HCCL_E_INTERNAL;
736 : }
737 0 : const std::vector<uint32_t>& dstRanks = mapIter->second.first;
738 0 : uint32_t& curIdx = mapIter->second.second;
739 0 : const uint32_t dstRank = dstRanks[curIdx % dstRanks.size()];
740 0 : HCCL_INFO(
741 : "[OpUnfoldCacheEntry][UpdateRefreshAddrInfoForAlltoallv] hcclOffset[%llu] dstRanks.size[%u] "
742 : "curIdx[%u] dstRank[%u]",
743 : hcclOffset, dstRanks.size(), curIdx, dstRank);
744 :
745 : // 新的dstRank一定是某个remoteRank, 即不等于curRank
746 0 : CHK_PRT_RET(
747 : dstRank == curRank,
748 : HCCL_ERROR(
749 : "[OpUnfoldCacheEntry][UpdateRefreshAddrInfoForAlltoallv] dstRank[%u] = curRank[%u]", dstRank,
750 : curRank),
751 : HCCL_E_INTERNAL);
752 :
753 : // Prepare for next-round dstRank (if any)
754 0 : curIdx = (curIdx + 1) % dstRanks.size();
755 :
756 : // 更新dst rank
757 0 : dstRefreshAddrInfo.rankId = dstRank;
758 0 : HCCL_INFO(
759 : "[OpUnfoldCacheEntry][UpdateRefreshAddrInfoForAlltoallv]"
760 : "dstRefreshAddrInfoArrays_[%u][%u].rankId[%u -> %u]",
761 : arrayIdx, sqeIdx, curRank, dstRefreshAddrInfo.rankId);
762 : } // PrepareIntraData in alltoallv
763 :
764 : // 刷新memcpy / cache-memcpy placeholder SQE后, 校验src/dst地址字段对应的memType
765 0 : CHK_RET(CheckMemTypeForAlltoallv(sqePtr, sqeType, srcRefreshAddrInfo, dstRefreshAddrInfo));
766 : } // sqeIdx
767 : } // arrayIdx
768 :
769 0 : return HCCL_SUCCESS;
770 : }
771 :
772 0 : HcclResult OpUnfoldCacheEntry::UpdateAndGetSqeArray(
773 : const size_t arrayIdx, const std::vector<OpUnfoldMemRange>& curUserInputMemRanges,
774 : const std::vector<OpUnfoldMemRange>& curUserOutputMemRanges, Stream& mainStream, std::vector<Stream>& slaveStreams,
775 : const uint32_t opRingBufferIdx, size_t& sqeCount, uint8_t** sqeArrayPtr, uint8_t** sqeTypeArrayPtr,
776 : AicpuDfxInfo** sqeDfxInfoArrayPtr, Stream** streamPtrPtr, std::vector<FlipInfo>& flipInfos, const bool profL1Enable,
777 : std::vector<uint64_t>& profTimestamps, const bool isAlltoallv, const AlltoallvMetadata& alltoallvMetadata,
778 : const AlltoallvSendRecvInfo& alltoallvSendRecvInfo)
779 : {
780 0 : HCCL_INFO(
781 : "[OpUnfoldCacheEntry][UpdateAndGetSqeArray] update and get SQEs from %uth SQE array; isAlltoallv[%u]", arrayIdx,
782 : isAlltoallv);
783 :
784 : // 检验入参
785 0 : CHK_PRT_RET(
786 : arrayIdx >= sqeArrays_.size(),
787 : HCCL_ERROR(
788 : "[OpUnfoldCacheEntry][MemcpySqeArray] arrayIdx %u is out of range [0, %u)", arrayIdx, sqeArrays_.size()),
789 : HCCL_E_INTERNAL);
790 0 : CHK_PRT_RET(
791 : arrayIdx >= streamSeqIdxes_.size(),
792 : HCCL_ERROR(
793 : "[OpUnfoldCacheEntry][MemcpySqeArray] arrayIdx %u is out of range [0, %u)", arrayIdx,
794 : streamSeqIdxes_.size()),
795 : HCCL_E_INTERNAL);
796 : // 检查指针, arrayPtr不应该是null, 但*arrayPtr应该是null
797 0 : CHK_PTRPTR_NULL(sqeArrayPtr);
798 0 : CHK_PTRPTR_NULL(sqeTypeArrayPtr);
799 0 : CHK_PTRPTR_NULL(sqeDfxInfoArrayPtr);
800 0 : CHK_PTRPTR_NULL(streamPtrPtr);
801 : // Double-check alltoallv相关入参
802 0 : if (isAlltoallv) {
803 0 : CHK_RET(alltoallvMetadata.Check(true));
804 0 : CHK_RET(alltoallvSendRecvInfo.Check());
805 : }
806 :
807 : // 设置入参
808 0 : sqeCount = srcRefreshAddrInfoArrays_[arrayIdx].size();
809 0 : *sqeArrayPtr = sqeArrays_[arrayIdx];
810 0 : *sqeTypeArrayPtr = sqeTypeArrays_[arrayIdx];
811 0 : *sqeDfxInfoArrayPtr = sqeDfxInfoArrays_[arrayIdx];
812 0 : flipInfos.clear();
813 0 : if (profL1Enable) {
814 0 : profTimestamps.clear();
815 0 : profTimestamps.reserve(
816 : sqeCount); // 需要额外flip placeholder是小概率事件, 所以只reserve cached SQE个数 (即非flip placeholder类SQE)
817 : }
818 :
819 : // 设置入参的stream pointer
820 0 : const uint32_t streamSeqIdx = streamSeqIdxes_[arrayIdx];
821 0 : if (streamSeqIdx == 0) {
822 0 : *streamPtrPtr = &mainStream;
823 : } else {
824 0 : CHK_PRT_RET(
825 : streamSeqIdx > slaveStreams.size(),
826 : HCCL_ERROR(
827 : "[OpUnfoldCacheEntry][UpdateAndGetSqeArray] invalid streamSeqIdx %u > slaveStreams.size() %u",
828 : streamSeqIdx, slaveStreams.size()),
829 : HCCL_E_MEMORY);
830 0 : *streamPtrPtr = &(slaveStreams[streamSeqIdx - 1]); // 0 < streamSeqIdx <= slaveStreams.size())
831 : }
832 :
833 : // 从stream中获取SQE刷新需要的当前task id
834 0 : HcclSqeContext* sqeContext = (*streamPtrPtr)->GetSqeContextPtr();
835 0 : CHK_PTR_NULL(sqeContext);
836 0 : SqeRingBuffer* sqeContextBuffer = &(sqeContext->buffer);
837 0 : CHK_PTR_NULL(sqeContextBuffer);
838 0 : uint16_t& curTaskId = sqeContextBuffer->tailSqeTaskId;
839 0 : uint16_t& curFlipNum = sqeContextBuffer->filpNum;
840 0 : HCCL_INFO(
841 : "[OpUnfoldCacheEntry][UpdateAndGetSqeArray] curTaskId[%u] curFlipNum[%u] from streamId %u for %u cached SQEs",
842 : curTaskId, curFlipNum, (*streamPtrPtr)->GetHcclStreamInfo().actualStreamId, sqeCount);
843 :
844 : // 执行SQE刷新
845 : // 注意: curUserInputMemRanges/curUserOutputMemRanges为当前算子执行时各rank输入输出的user memory range,
846 : // userInputMemRanges_/userOutputMemRanges_为算子缓存时各rank输入输出的user memory range
847 0 : const std::vector<RefreshAddrInfo>& srcRefreshAddrInfoArray = srcRefreshAddrInfoArrays_[arrayIdx];
848 0 : const std::vector<RefreshAddrInfo>& dstRefreshAddrInfoArray = dstRefreshAddrInfoArrays_[arrayIdx];
849 0 : uint64_t sqeSrcAddr = 0;
850 0 : uint64_t sqeDstAddr = 0;
851 0 : uint8_t* sqePtr = (*sqeArrayPtr);
852 0 : for (size_t sqeIdx = 0; sqeIdx < sqeCount; ++sqeIdx) {
853 : // 获取当前SQE的信息
854 0 : uint8_t& sqeType = (*sqeTypeArrayPtr)[sqeIdx];
855 0 : const RefreshAddrInfo& srcRefreshAddrInfo = srcRefreshAddrInfoArray[sqeIdx];
856 0 : const RefreshAddrInfo& dstRefreshAddrInfo = dstRefreshAddrInfoArray[sqeIdx];
857 0 : HCCL_INFO(
858 : "[OpUnfoldCacheEntry][UpdateAndGetSqeArray] update %uth cached SQE with sqeType[%u] "
859 : "srcRefreshAddrInfo[rankid[%u], memType[%u]] dstRefreshAddrInfo[rankid[%u], memType[%u]] curTaskId[%u]",
860 : sqeIdx, sqeType, srcRefreshAddrInfo.rankId, srcRefreshAddrInfo.memType, dstRefreshAddrInfo.rankId,
861 : dstRefreshAddrInfo.memType, curTaskId);
862 :
863 : // 根据SQE type进行对应刷新 (task id始终要刷新; addr相关字段有条件刷新)
864 0 : switch (sqeType) {
865 0 : case SqeType::NOTIFY_SQE: {
866 0 : rtStarsNotifySqeV1_t* notifySqePtr = reinterpret_cast<rtStarsNotifySqeV1_t*>(sqePtr);
867 0 : if (isAlltoallv && notifySqePtr->header.type == RT_STARS_SQE_TYPE_NOTIFY_WAIT) {
868 : // 针对alltoallv算子, 动态刷新Notify SQE / 将Notify SQE生成为CacheNotifyPlaceholder SQE
869 : // 注意: NotifyRecord SQE不需要针对alltoallv类算子做特殊刷新
870 0 : CHK_RET(UpdateSyncSqeForAlltoallv(
871 : sqePtr, &sqeType, curTaskId, srcRefreshAddrInfo, dstRefreshAddrInfo, alltoallvMetadata,
872 : alltoallvSendRecvInfo));
873 0 : } else {
874 0 : notifySqePtr->header.taskId = curTaskId;
875 : }
876 0 : break;
877 : }
878 0 : case SqeType::WRITE_VALUE_SQE:
879 : case SqeType::RDMA_DB_SEND_SQE: {
880 0 : if (isAlltoallv) { // 针对alltoallv算子, 动态刷新WriteValue SQE / 将WriteValue
881 : // SQE生成为CacheWriteValuePlaceholder SQE
882 0 : CHK_RET(UpdateSyncSqeForAlltoallv(
883 : sqePtr, &sqeType, curTaskId, srcRefreshAddrInfo, dstRefreshAddrInfo, alltoallvMetadata,
884 : alltoallvSendRecvInfo));
885 : } else {
886 0 : rtStarsWriteValueSqe_t* writeValueSqePtr = reinterpret_cast<rtStarsWriteValueSqe_t*>(sqePtr);
887 0 : writeValueSqePtr->header.taskId = curTaskId;
888 :
889 0 : if (dstRefreshAddrInfo.memType != RefreshAddrInfo::INVALID_MEMTYPE) { // 需要刷新地址
890 0 : CombineUint32ToUint64(
891 0 : sqeDstAddr, writeValueSqePtr->write_addr_high, writeValueSqePtr->write_addr_low);
892 0 : if (dstRefreshAddrInfo.memType == RefreshAddrInfo::USER_OUTPUT_MEMTYPE) { // user output
893 0 : CHK_RET(RefreshSqeAddr(
894 : sqeDstAddr, dstRefreshAddrInfo.rankId, userOutputMemRanges_, curUserOutputMemRanges,
895 : false, 0));
896 0 : } else if (dstRefreshAddrInfo.memType == RefreshAddrInfo::USER_INPUT_MEMTYPE) { // user input
897 0 : CHK_RET(RefreshSqeAddr(
898 : sqeDstAddr, dstRefreshAddrInfo.rankId, userInputMemRanges_, curUserInputMemRanges,
899 : false, 0));
900 : } else { // hccl input (alltoallv算子不需要对WRITE_VALUE_SQE刷新hccl地址)
901 0 : HCCL_ERROR(
902 : "[OpUnfoldCacheEntry][UpdateAndGetSqeArray] isAlltoallv[%u] sqeType[%u]; memType "
903 : "should be user input/output",
904 : isAlltoallv, sqeType);
905 0 : HCCL_ERROR(
906 : "[OpUnfoldCacheEntry][UpdateAndGetSqeArray] dstRefreshAddrInfo: rankId[%u] memType[%u]",
907 : dstRefreshAddrInfo.rankId, dstRefreshAddrInfo.memType);
908 0 : return HCCL_E_INTERNAL;
909 : }
910 :
911 : // Bit-field member不能直接传引用
912 0 : uint32_t tmpHighAddr = 0;
913 0 : SplitUint64ToUint32(sqeDstAddr, tmpHighAddr, writeValueSqePtr->write_addr_low);
914 0 : writeValueSqePtr->write_addr_high = tmpHighAddr;
915 : }
916 : }
917 0 : break;
918 : }
919 0 : case SqeType::EVENT_SQE: {
920 0 : rtStarsEventSqe_t* eventSqePtr = reinterpret_cast<rtStarsEventSqe_t*>(sqePtr);
921 0 : eventSqePtr->header.taskId = curTaskId;
922 0 : break;
923 : }
924 0 : case SqeType::MEMCPY_ASYNC_SQE: {
925 0 : if (isAlltoallv) { // 针对alltoallv算子
926 0 : if (srcRefreshAddrInfo.memType == RefreshAddrInfo::INVALID_MEMTYPE
927 0 : && dstRefreshAddrInfo.memType == RefreshAddrInfo::INVALID_MEMTYPE) { // MemcpyRecord SQE
928 : // 动态刷新MemcpyRecord SQE / 将MemcpyRecord SQE生成为CacheMemcpyRecordPlaceholder SQE
929 0 : CHK_RET(UpdateSyncSqeForAlltoallv(
930 : sqePtr, &sqeType, curTaskId, srcRefreshAddrInfo, dstRefreshAddrInfo, alltoallvMetadata,
931 : alltoallvSendRecvInfo));
932 0 : } else { // Memcpy SQE
933 : // 动态刷新Memcpy SQE / 将Memcpy SQE生成为CacheMemcpyPlaceholder SQE
934 0 : CHK_RET(UpdateTransferSqeForAlltoallv(
935 : sqePtr, &sqeType, curTaskId, srcRefreshAddrInfo, dstRefreshAddrInfo, curUserInputMemRanges,
936 : curUserOutputMemRanges, alltoallvMetadata, alltoallvSendRecvInfo));
937 : }
938 : } else { // 非V类算子, 动态刷新Memcpy SQE
939 0 : rtStarsMemcpyAsyncSqe_t* memcpyAsyncSqePtr = reinterpret_cast<rtStarsMemcpyAsyncSqe_t*>(sqePtr);
940 0 : memcpyAsyncSqePtr->header.taskId = curTaskId;
941 :
942 0 : if (srcRefreshAddrInfo.memType != RefreshAddrInfo::INVALID_MEMTYPE) { // 需要刷新src addr
943 0 : CombineUint32ToUint64(
944 : sqeSrcAddr, memcpyAsyncSqePtr->src_addr_high, memcpyAsyncSqePtr->src_addr_low);
945 0 : if (srcRefreshAddrInfo.memType == RefreshAddrInfo::USER_OUTPUT_MEMTYPE) { // user output
946 0 : CHK_RET(RefreshSqeAddr(
947 : sqeSrcAddr, srcRefreshAddrInfo.rankId, userOutputMemRanges_, curUserOutputMemRanges,
948 : false, 0));
949 0 : } else if (srcRefreshAddrInfo.memType == RefreshAddrInfo::USER_INPUT_MEMTYPE) { // user input
950 0 : CHK_RET(RefreshSqeAddr(
951 : sqeSrcAddr, srcRefreshAddrInfo.rankId, userInputMemRanges_, curUserInputMemRanges,
952 : false, 0));
953 : } else { // hccl input (非V类算子不需要对MEMCPY_ASYNC_SQE刷新hccl地址)
954 0 : HCCL_ERROR(
955 : "[OpUnfoldCacheEntry][UpdateAndGetSqeArray] isAlltoallv[%u] sqeType[%u]; memType "
956 : "should be user input/output",
957 : isAlltoallv, sqeType);
958 0 : HCCL_ERROR(
959 : "[OpUnfoldCacheEntry][UpdateAndGetSqeArray] srcRefreshAddrInfo: rankId[%u] memType[%u]",
960 : srcRefreshAddrInfo.rankId, srcRefreshAddrInfo.memType);
961 0 : return HCCL_E_INTERNAL;
962 : }
963 0 : SplitUint64ToUint32(
964 0 : sqeSrcAddr, memcpyAsyncSqePtr->src_addr_high, memcpyAsyncSqePtr->src_addr_low);
965 : }
966 :
967 0 : if (dstRefreshAddrInfo.memType != RefreshAddrInfo::INVALID_MEMTYPE) { // 需要刷新地址
968 0 : CombineUint32ToUint64(
969 : sqeDstAddr, memcpyAsyncSqePtr->dst_addr_high, memcpyAsyncSqePtr->dst_addr_low);
970 0 : if (dstRefreshAddrInfo.memType == RefreshAddrInfo::USER_OUTPUT_MEMTYPE) { // user output
971 0 : CHK_RET(RefreshSqeAddr(
972 : sqeDstAddr, dstRefreshAddrInfo.rankId, userOutputMemRanges_, curUserOutputMemRanges,
973 : false, 0));
974 0 : } else if (dstRefreshAddrInfo.memType == RefreshAddrInfo::USER_INPUT_MEMTYPE) { // user input
975 0 : CHK_RET(RefreshSqeAddr(
976 : sqeDstAddr, dstRefreshAddrInfo.rankId, userInputMemRanges_, curUserInputMemRanges,
977 : false, 0));
978 : } else { // hccl input (非V类算子不需要对MEMCPY_ASYNC_SQE刷新hccl地址)
979 0 : HCCL_ERROR(
980 : "[OpUnfoldCacheEntry][UpdateAndGetSqeArray] isAlltoallv[%u] sqeType[%u]; memType "
981 : "should be user input/output",
982 : isAlltoallv, sqeType);
983 0 : HCCL_ERROR(
984 : "[OpUnfoldCacheEntry][UpdateAndGetSqeArray] dstRefreshAddrInfo: rankId[%u] memType[%u]",
985 : dstRefreshAddrInfo.rankId, dstRefreshAddrInfo.memType);
986 0 : return HCCL_E_INTERNAL;
987 : }
988 0 : SplitUint64ToUint32(
989 0 : sqeDstAddr, memcpyAsyncSqePtr->dst_addr_high, memcpyAsyncSqePtr->dst_addr_low);
990 : }
991 : }
992 :
993 0 : break;
994 : }
995 0 : case SqeType::CCORE_WAIT_START_SQE: {
996 0 : HCCL_ERROR(
997 : "[OpUnfoldCacheEntry][UpdateAndGetSqeArray] SqeType::CCORE_WAIT_START_SQE is not supported in A3");
998 0 : return HCCL_E_NOT_SUPPORT;
999 : }
1000 0 : case SqeType::CCORE_WRITE_VALUE_SQE: {
1001 0 : HCCL_ERROR(
1002 : "[OpUnfoldCacheEntry][UpdateAndGetSqeArray] SqeType::CCORE_WRITE_VALUE_SQE is not supported in A3");
1003 0 : return HCCL_E_NOT_SUPPORT;
1004 : }
1005 0 : case SqeType::NOTIFY_SQE_V2: {
1006 0 : HCCL_ERROR("[OpUnfoldCacheEntry][UpdateAndGetSqeArray] SqeType::NOTIFY_SQE_V2 is not supported in A3");
1007 0 : return HCCL_E_NOT_SUPPORT;
1008 : }
1009 0 : case SqeType::WRITE_VALUE_SQE_V2: {
1010 0 : HCCL_ERROR(
1011 : "[OpUnfoldCacheEntry][UpdateAndGetSqeArray] SqeType::WRITE_VALUE_SQE_V2 is not supported in A3");
1012 0 : return HCCL_E_NOT_SUPPORT;
1013 : }
1014 0 : case SqeType::EVENT_SQE_V2: {
1015 0 : HCCL_ERROR("[OpUnfoldCacheEntry][UpdateAndGetSqeArray] SqeType::EVENT_SQE_V2 is not supported in A3");
1016 0 : return HCCL_E_NOT_SUPPORT;
1017 : }
1018 0 : case SqeType::MEMCPY_ASYNC_SQE_V2: {
1019 0 : HCCL_ERROR(
1020 : "[OpUnfoldCacheEntry][UpdateAndGetSqeArray] SqeType::MEMCPY_ASYNC_SQE_V2 is not supported in A3");
1021 0 : return HCCL_E_NOT_SUPPORT;
1022 : }
1023 0 : case SqeType::FLIP_PLACEHOLDER_SQE: {
1024 0 : HCCL_ERROR(
1025 : "[OpUnfoldCacheEntry][UpdateAndGetSqeArray] placeholder should not be cached, sqeType[%u] "
1026 : "sqeIdx[%u]",
1027 : sqeType, sqeIdx);
1028 0 : return HCCL_E_INTERNAL;
1029 : }
1030 0 : case SqeType::CACHE_MEMCPY_PLACEHOLDER_SQE: {
1031 : // 非V类算子, 不应该出现CacheMemcpyPlaceholder SQE
1032 0 : CHK_PRT_RET(
1033 : !isAlltoallv,
1034 : HCCL_ERROR(
1035 : "[OpUnfoldCacheEntry][UpdateAndGetSqeArray] cache-memcpy placeholder"
1036 : "should be not cached for non-alltoallv op, sqeType[%u] sqeIdx[%u]",
1037 : sqeType, sqeIdx),
1038 : HCCL_E_INTERNAL);
1039 :
1040 : // 针对alltoallv类算子, 动态刷新CacheMemcpyPlaceholder SQE / 将CacheMemcpyPlaceholder SQE生成为Memcpy
1041 : // SQE
1042 0 : CHK_RET(UpdateTransferSqeForAlltoallv(
1043 : sqePtr, &sqeType, curTaskId, srcRefreshAddrInfo, dstRefreshAddrInfo, curUserInputMemRanges,
1044 : curUserOutputMemRanges, alltoallvMetadata, alltoallvSendRecvInfo));
1045 :
1046 0 : break;
1047 : }
1048 0 : case SqeType::CACHE_NOTIFY_PLACEHOLDER_SQE:
1049 : case SqeType::CACHE_WRITE_VALUE_PLACEHOLDER_SQE:
1050 : case SqeType::CACHE_MEMCPY_RECORD_PLACEHOLDER_SQE: {
1051 : // 非V类算子, 不应该出现CacheNotify/Write/MemcpyRecordPlaceholder SQE
1052 0 : CHK_PRT_RET(
1053 : !isAlltoallv,
1054 : HCCL_ERROR(
1055 : "[OpUnfoldCacheEntry][UpdateAndGetSqeArray] cache-notify/write/memcpy-record placeholder"
1056 : "should be not cached for non-alltoallv op, sqeType[%u] sqeIdx[%u]",
1057 : sqeType, sqeIdx),
1058 : HCCL_E_INTERNAL);
1059 :
1060 : // 针对alltoallv类算子, 动态刷新CacheNotify/WritePlaceholder SQE / 将CacheNotify/WritePlaceholder
1061 : // SQE生成为Notify/Write SQE
1062 0 : CHK_RET(UpdateSyncSqeForAlltoallv(
1063 : sqePtr, &sqeType, curTaskId, srcRefreshAddrInfo, dstRefreshAddrInfo, alltoallvMetadata,
1064 : alltoallvSendRecvInfo));
1065 :
1066 0 : break;
1067 : }
1068 0 : default: {
1069 0 : HCCL_WARNING(
1070 : "[OpUnfoldCacheEntry][UpdateAndGetSqeArray] sqeType %u is unsupported (sqeIdx[%u] curTaskId[%u])",
1071 : sqeType, sqeIdx, curTaskId);
1072 0 : return HCCL_E_NOT_SUPPORT;
1073 : }
1074 : }
1075 :
1076 : // 记录SQE刷新时间用于profiling
1077 0 : if (profL1Enable) {
1078 0 : const uint64_t curTime = ProfGetCurCpuTimestamp();
1079 0 : profTimestamps.push_back(curTime);
1080 : }
1081 :
1082 : // 刷新taskId和flipNum
1083 0 : if (curTaskId == UINT16_MAX) { // 更新flipNum和taskId
1084 : // 参考stream.cc中的GetNextSqeBufferAddr
1085 0 : curFlipNum += 1;
1086 0 : curTaskId = 0;
1087 0 : } else if (curTaskId == 0 && curFlipNum != 0) { // 更新flipInfos和taskId
1088 : // 参考dispatcher_aicpu.cc中的GetStreamSqeBufferAddr
1089 0 : flipInfos.push_back(FlipInfo(sqeIdx, curFlipNum));
1090 :
1091 : // 为placeholder SQE预留task id = 0
1092 0 : curTaskId = 1;
1093 :
1094 : // Flip placeholder SQE在外侧dispatcher aicpu中生成, 这里记录当前时间作为flip placeholder SQE的生成时间
1095 0 : if (profL1Enable) {
1096 0 : const uint64_t curTime = ProfGetCurCpuTimestamp();
1097 0 : profTimestamps.push_back(curTime);
1098 : }
1099 0 : } else { // 只更新taskid
1100 0 : curTaskId += 1;
1101 : }
1102 :
1103 0 : sqePtr += HCCL_SQE_SIZE;
1104 : }
1105 :
1106 : // 更新每个SQE的DfxInfo中的opRingBufferIdx
1107 0 : HCCL_INFO(
1108 : "[OpUnfoldCacheEntry][UpdateAndGetSqeArray] update opRingBufferIndx in DfxInfoArray as %u", opRingBufferIdx);
1109 0 : for (size_t sqeIdx = 0; sqeIdx < sqeCount; ++sqeIdx) {
1110 0 : (*sqeDfxInfoArrayPtr)[sqeIdx].opRingBufferIdx = opRingBufferIdx;
1111 : }
1112 :
1113 0 : HCCL_INFO(
1114 : "[OpUnfoldCacheEntry][UpdateAndGetSqeArray] update and get %uth SQE array with %u SQEs, streamId[%u] and %u "
1115 : "flipInfos",
1116 : arrayIdx, sqeCount, (*streamPtrPtr)->GetHcclStreamInfo().actualStreamId, flipInfos.size());
1117 :
1118 0 : return HCCL_SUCCESS;
1119 : }
1120 :
1121 0 : HcclResult OpUnfoldCacheEntry::SetInputOutputMemRanges(
1122 : const std::vector<OpUnfoldMemRange>& curUserInputMemRanges,
1123 : const std::vector<OpUnfoldMemRange>& curUserOutputMemRanges)
1124 : {
1125 0 : CHK_PRT_RET(
1126 : userInputMemRanges_.size() != curUserInputMemRanges.size(),
1127 : HCCL_ERROR(
1128 : "[OpUnfoldCacheEntry][SetInputOutputMemRanges] original rankSize %u != new rankSize %u",
1129 : userInputMemRanges_.size(), curUserInputMemRanges.size()),
1130 : HCCL_E_INTERNAL);
1131 :
1132 0 : userInputMemRanges_ = curUserInputMemRanges;
1133 0 : userOutputMemRanges_ = curUserOutputMemRanges;
1134 :
1135 0 : return HCCL_SUCCESS;
1136 : }
1137 :
1138 0 : HcclResult OpUnfoldCacheEntry::CheckAndPrepareRefreshAddrInfo(
1139 : const uint64_t sqeAddr, RefreshAddrInfo& refreshAddrInfo, const bool isAlltoallv,
1140 : const AlltoallvMetadata* alltoallvMetadataPtr)
1141 : {
1142 : // 遍历per-rank user input memory range
1143 0 : for (size_t rankId = 0; rankId < userInputMemRanges_.size(); ++rankId) {
1144 0 : bool isInRange = false;
1145 0 : CHK_RET(userInputMemRanges_[rankId].InRange(sqeAddr, isInRange));
1146 0 : if (isInRange) {
1147 0 : refreshAddrInfo.rankId = rankId;
1148 0 : refreshAddrInfo.memType = RefreshAddrInfo::USER_INPUT_MEMTYPE;
1149 0 : return HCCL_SUCCESS; // 确实是某rank下的user input mem, 则无需继续搜索output mem
1150 : }
1151 : }
1152 :
1153 : // 遍历per-rank user input memory range
1154 0 : for (size_t rankId = 0; rankId < userOutputMemRanges_.size(); ++rankId) {
1155 0 : bool isInRange = false;
1156 0 : CHK_RET(userOutputMemRanges_[rankId].InRange(sqeAddr, isInRange));
1157 0 : if (isInRange) {
1158 0 : refreshAddrInfo.rankId = rankId;
1159 0 : refreshAddrInfo.memType = RefreshAddrInfo::USER_OUTPUT_MEMTYPE;
1160 0 : return HCCL_SUCCESS;
1161 : }
1162 : }
1163 :
1164 : // 针对alltoallv算子, 遍历HCCL input buffer, 确定rank id
1165 0 : if (isAlltoallv) {
1166 0 : CHK_PTR_NULL(alltoallvMetadataPtr);
1167 0 : CHK_RET(alltoallvMetadataPtr->Check(false));
1168 :
1169 0 : const std::vector<OpUnfoldMemRange>& hcclInputMemRanges = alltoallvMetadataPtr->hcclInputMemRanges;
1170 0 : for (size_t rankId = 0; rankId < hcclInputMemRanges.size(); ++rankId) {
1171 0 : bool isInRange = false;
1172 0 : CHK_RET(hcclInputMemRanges[rankId].InRange(sqeAddr, isInRange));
1173 0 : if (isInRange) {
1174 0 : refreshAddrInfo.rankId = rankId;
1175 0 : refreshAddrInfo.memType = RefreshAddrInfo::HCCL_INPUT_MEMTYPE;
1176 0 : return HCCL_SUCCESS;
1177 : }
1178 : }
1179 : }
1180 :
1181 0 : return HCCL_SUCCESS;
1182 : }
1183 :
1184 0 : HcclResult OpUnfoldCacheEntry::CheckMemTypeForAlltoallv(
1185 : const uint8_t* sqePtr, const uint8_t sqeType, const RefreshAddrInfo& srcRefreshAddrInfo,
1186 : const RefreshAddrInfo& dstRefreshAddrInfo) const
1187 : {
1188 : // 对于所有算子, SQE第一次admit时, 在MemcpySqeArray中, 遍历memory ranges获取每个地址字段对应的rankId和memType
1189 : // 对于alltoallv算子, cache miss后处理时, 在UpdateRefreshAddrInfoForAlltoallv中, 进一步更新memType和dstRank
1190 :
1191 : // 如果src在user memory内, 说明是LocalCopy或者PrepareIntraData, 一定是local user input -> local user output / local
1192 : // hccl input 如果dst在user memory内, 说明是LocalCopy或者RemoteCopy, 一定是local user input / remote hccl input ->
1193 : // local user output
1194 :
1195 : // 获取task id和stream id
1196 0 : uint16_t taskId = 0;
1197 0 : uint16_t streamId = 0;
1198 0 : if (sqeType == SqeType::MEMCPY_ASYNC_SQE) {
1199 0 : const rtStarsMemcpyAsyncSqe_t* memcpyAsyncSqePtr = reinterpret_cast<const rtStarsMemcpyAsyncSqe_t*>(sqePtr);
1200 0 : taskId = memcpyAsyncSqePtr->header.taskId;
1201 0 : streamId = memcpyAsyncSqePtr->header.rtStreamId;
1202 0 : } else if (sqeType == SqeType::CACHE_MEMCPY_PLACEHOLDER_SQE) {
1203 0 : const rtStarsPlaceHolderSqe_t* placeholderSqePtr = reinterpret_cast<const rtStarsPlaceHolderSqe_t*>(sqePtr);
1204 0 : taskId = placeholderSqePtr->header.taskId;
1205 0 : streamId = placeholderSqePtr->header.rtStreamId;
1206 : } else {
1207 0 : HCCL_ERROR("[OpUnfoldCacheEntry][CheckMemTypeForAlltoallv] invalid sqeType[%u]", sqeType);
1208 0 : return HCCL_E_INTERNAL;
1209 : }
1210 :
1211 : // memcpy / cache-memcpy placeholder SQE的src/dst memType一定不是invalid
1212 : // 注意: memcpy-record SQE的src/dst一定是invalid, 但不会进入本函数
1213 0 : const uint8_t srcMemType = srcRefreshAddrInfo.memType;
1214 0 : const uint8_t dstMemType = dstRefreshAddrInfo.memType;
1215 0 : CHK_PRT_RET(
1216 : srcMemType == RefreshAddrInfo::INVALID_MEMTYPE || dstMemType == RefreshAddrInfo::INVALID_MEMTYPE,
1217 : HCCL_ERROR(
1218 : "[OpUnfoldCacheEntry][CheckMemTypeForAlltoallv] sqeType[%u] streamId[%u] taskId[%u] srcMemType[%u] "
1219 : "dstMemType[%u]",
1220 : sqeType, streamId, taskId, srcMemType, dstMemType),
1221 : HCCL_E_INTERNAL);
1222 :
1223 : // src一定不是user output
1224 0 : CHK_PRT_RET(
1225 : srcMemType == RefreshAddrInfo::USER_OUTPUT_MEMTYPE,
1226 : HCCL_ERROR(
1227 : "[OpUnfoldCacheEntry][CheckMemTypeForAlltoallv] sqeType[%u] streamId[%u] taskId[%u] srcMemType[%u]",
1228 : sqeType, streamId, taskId, srcMemType),
1229 : HCCL_E_INTERNAL);
1230 :
1231 : // src是user input时, dst一定是user output / hccl input
1232 : // src是hccl input时, dst一定是user output
1233 0 : if (srcMemType == RefreshAddrInfo::USER_INPUT_MEMTYPE) {
1234 0 : CHK_PRT_RET(
1235 : dstMemType != RefreshAddrInfo::USER_OUTPUT_MEMTYPE && dstMemType != RefreshAddrInfo::HCCL_INPUT_MEMTYPE,
1236 : HCCL_ERROR(
1237 : "[OpUnfoldCacheEntry][CheckMemTypeForAlltoallv] sqeType[%u] streamId[%u] taskId[%u] dstMemType[%u]",
1238 : sqeType, streamId, taskId, dstMemType),
1239 : HCCL_E_INTERNAL);
1240 0 : } else if (srcMemType == RefreshAddrInfo::HCCL_INPUT_MEMTYPE) {
1241 0 : CHK_PRT_RET(
1242 : dstMemType != RefreshAddrInfo::USER_OUTPUT_MEMTYPE,
1243 : HCCL_ERROR(
1244 : "[OpUnfoldCacheEntry][CheckMemTypeForAlltoallv] sqeType[%u] streamId[%u] taskId[%u] dstMemType[%u]",
1245 : sqeType, streamId, taskId, dstMemType),
1246 : HCCL_E_INTERNAL);
1247 : }
1248 :
1249 : // dst一定不是user input
1250 0 : CHK_PRT_RET(
1251 : dstMemType == RefreshAddrInfo::USER_INPUT_MEMTYPE,
1252 : HCCL_ERROR(
1253 : "[OpUnfoldCacheEntry][CheckMemTypeForAlltoallv] sqeType[%u] streamId[%u] taskId[%u] dstMemType[%u]",
1254 : sqeType, streamId, taskId, dstMemType),
1255 : HCCL_E_INTERNAL);
1256 :
1257 : // dst是user output时, src一定是user input / hccl input
1258 : // dst是hccl input时, src一定是user input
1259 0 : if (dstMemType == RefreshAddrInfo::USER_OUTPUT_MEMTYPE) {
1260 0 : CHK_PRT_RET(
1261 : srcMemType != RefreshAddrInfo::USER_INPUT_MEMTYPE && srcMemType != RefreshAddrInfo::HCCL_INPUT_MEMTYPE,
1262 : HCCL_ERROR(
1263 : "[OpUnfoldCacheEntry][CheckMemTypeForAlltoallv] sqeType[%u] streamId[%u] taskId[%u] srcMemType[%u]",
1264 : sqeType, streamId, taskId, srcMemType),
1265 : HCCL_E_INTERNAL);
1266 0 : } else if (dstMemType == RefreshAddrInfo::HCCL_INPUT_MEMTYPE) {
1267 0 : CHK_PRT_RET(
1268 : srcMemType != RefreshAddrInfo::USER_INPUT_MEMTYPE,
1269 : HCCL_ERROR(
1270 : "[OpUnfoldCacheEntry][CheckMemTypeForAlltoallv] sqeType[%u] streamId[%u] taskId[%u] srcMemType[%u]",
1271 : sqeType, streamId, taskId, srcMemType),
1272 : HCCL_E_INTERNAL);
1273 : }
1274 :
1275 0 : return HCCL_SUCCESS;
1276 : }
1277 :
1278 0 : HcclResult OpUnfoldCacheEntry::UpdateTransferSqeForAlltoallv(
1279 : uint8_t* sqePtr, uint8_t* sqeTypePtr, const uint16_t curTaskId, const RefreshAddrInfo& srcRefreshAddrInfo,
1280 : const RefreshAddrInfo& dstRefreshAddrInfo, const std::vector<OpUnfoldMemRange>& curUserInputMemRanges,
1281 : const std::vector<OpUnfoldMemRange>& curUserOutputMemRanges, const AlltoallvMetadata& alltoallvMetadata,
1282 : const AlltoallvSendRecvInfo& alltoallvSendRecvInfo)
1283 : {
1284 0 : HCCL_INFO(
1285 : "[OpUnfoldCacheEntry][UpdateTransferSqeForAlltoallv] curTaskId[%u] srcMemType[%u] dstMemType[%u]", curTaskId,
1286 : srcRefreshAddrInfo.memType, dstRefreshAddrInfo.memType);
1287 :
1288 : // 先判断是否需要刷新此memcpy SQE
1289 0 : if (srcRefreshAddrInfo.memType != RefreshAddrInfo::USER_INPUT_MEMTYPE
1290 0 : && dstRefreshAddrInfo.memType != RefreshAddrInfo::USER_OUTPUT_MEMTYPE) {
1291 : // 注意: 由于alltoallv direct fullmesh不会调用InlineReduceAsync和基于memcpy的SignalRecord
1292 : // 理论上所有的memcpy / cache-memcpy SQE都需要地址刷新, 不会进入当前code block
1293 :
1294 : // 更新task id
1295 0 : if ((*sqeTypePtr) == SqeType::MEMCPY_ASYNC_SQE) {
1296 0 : rtStarsMemcpyAsyncSqe_t* memcpyAsyncSqePtr = reinterpret_cast<rtStarsMemcpyAsyncSqe_t*>(sqePtr);
1297 0 : memcpyAsyncSqePtr->header.taskId = curTaskId;
1298 0 : } else if ((*sqeTypePtr) == SqeType::CACHE_MEMCPY_PLACEHOLDER_SQE) {
1299 0 : rtStarsPlaceHolderSqe_t* placeholderSqePtr = reinterpret_cast<rtStarsPlaceHolderSqe_t*>(sqePtr);
1300 0 : placeholderSqePtr->header.taskId = curTaskId;
1301 : } else {
1302 0 : HCCL_ERROR("[OpUnfoldCacheEntry][UpdateTransferSqeForAlltoallv] invalid sqeType[%u]", *sqeTypePtr);
1303 0 : return HCCL_E_INTERNAL;
1304 : }
1305 :
1306 0 : return HCCL_SUCCESS;
1307 : }
1308 : // 注意: 从这里开始, srcAddr为userInput和dstAddr为userOutput至少有一个条件满足
1309 :
1310 : // 校验alltoallv相关参数
1311 0 : CHK_RET(alltoallvMetadata.Check(true));
1312 0 : CHK_RET(alltoallvSendRecvInfo.Check());
1313 0 : const uint32_t rankSize = alltoallvSendRecvInfo.sendOffsets.size();
1314 0 : CHK_PRT_RET(
1315 : rankSize != alltoallvMetadata.hcclInputMemRanges.size(),
1316 : HCCL_ERROR(
1317 : "[OpUnfoldCacheEntry][UpdateTransferSqeForAlltoallv] hcclInputMemRanges.size[%u] != rankSize[%u]",
1318 : alltoallvMetadata.hcclInputMemRanges.size(), rankSize),
1319 : HCCL_E_INTERNAL);
1320 :
1321 : // 获取当前memcpy类SQE对应的count和size
1322 0 : uint64_t count = 0;
1323 0 : uint64_t size = 0; // send/recv bytes
1324 0 : CHK_RET(GetTransferCountForAlltoallv(
1325 : count, size, srcRefreshAddrInfo, dstRefreshAddrInfo, alltoallvMetadata, alltoallvSendRecvInfo));
1326 :
1327 : // 更新/生成相应SQE
1328 0 : if ((*sqeTypePtr) == SqeType::MEMCPY_ASYNC_SQE) {
1329 0 : CHK_RET(UpdateMemcpySqeForAlltoallv(
1330 : sqePtr, sqeTypePtr, curTaskId, srcRefreshAddrInfo, dstRefreshAddrInfo, curUserInputMemRanges,
1331 : curUserOutputMemRanges, alltoallvMetadata, alltoallvSendRecvInfo, count, size));
1332 0 : } else if ((*sqeTypePtr) == SqeType::CACHE_MEMCPY_PLACEHOLDER_SQE) {
1333 0 : CHK_RET(UpdateMemcpyPlaceholderSqeForAlltoallv(
1334 : sqePtr, sqeTypePtr, curTaskId, srcRefreshAddrInfo, dstRefreshAddrInfo, curUserInputMemRanges,
1335 : curUserOutputMemRanges, alltoallvMetadata, alltoallvSendRecvInfo, count, size));
1336 : } else {
1337 0 : HCCL_ERROR("[OpUnfoldCacheEntry][UpdateTransferSqeForAlltoallv] invalid sqeType[%u]", *sqeTypePtr);
1338 0 : return HCCL_E_INTERNAL;
1339 : }
1340 :
1341 0 : return HCCL_SUCCESS;
1342 : }
1343 :
1344 0 : HcclResult OpUnfoldCacheEntry::GetTransferCountForAlltoallv(
1345 : uint64_t& count, uint64_t& size, const RefreshAddrInfo& srcRefreshAddrInfo,
1346 : const RefreshAddrInfo& dstRefreshAddrInfo, const AlltoallvMetadata& alltoallvMetadata,
1347 : const AlltoallvSendRecvInfo& alltoallvSendRecvInfo) const
1348 : {
1349 : // 注意: 如果对应send count, 一定是LocalCopy或者PrepareIntraData, 即local user input -> local user output / hccl
1350 : // input 注意: 如果对应recv count, 一定是RemoteCopy, 即remote hccl input -> local user output
1351 0 : const uint32_t rankSize = alltoallvSendRecvInfo.sendOffsets.size();
1352 0 : if (srcRefreshAddrInfo.memType == RefreshAddrInfo::USER_INPUT_MEMTYPE) { // LocalCopy/PrepareIntraData
1353 : // 获得local rank
1354 0 : uint32_t localRank = srcRefreshAddrInfo.rankId;
1355 0 : CHK_PRT_RET(
1356 : localRank >= rankSize,
1357 : HCCL_ERROR(
1358 : "[OpUnfoldCacheEntry][GetTransferCountForAlltoallv] localRank[%u] >= rankSize[%u]", localRank,
1359 : rankSize),
1360 : HCCL_E_INTERNAL);
1361 :
1362 : // 获得dst rank
1363 0 : uint32_t dstRank = 0;
1364 0 : if (dstRefreshAddrInfo.memType == RefreshAddrInfo::USER_OUTPUT_MEMTYPE) { // LocalCopy
1365 0 : dstRank = dstRefreshAddrInfo.rankId; // dstRank = localRank
1366 0 : CHK_PRT_RET(
1367 : dstRank != localRank,
1368 : HCCL_ERROR(
1369 : "[OpUnfoldCacheEntry][GetTransferCountForAlltoallv] dstRank[%u] != localRank[%u]", dstRank,
1370 : localRank),
1371 : HCCL_E_INTERNAL);
1372 0 : } else if (dstRefreshAddrInfo.memType == RefreshAddrInfo::HCCL_INPUT_MEMTYPE) { // PrepareIntraData
1373 : // dstRank在第一次cache miss后处理时, 被UpdateRefreshAddrInfoForAlltoallv更新, 一定不等于localRank
1374 0 : dstRank = dstRefreshAddrInfo.rankId;
1375 0 : CHK_PRT_RET(
1376 : dstRank == localRank,
1377 : HCCL_ERROR(
1378 : "[OpUnfoldCacheEntry][GetTransferCountForAlltoallv] dstRank[%u] = localRank[%u]", dstRank,
1379 : localRank),
1380 : HCCL_E_INTERNAL);
1381 : } else {
1382 0 : HCCL_ERROR(
1383 : "[OpUnfoldCacheEntry][GetTransferCountForAlltoallv] invalid dstMemType[%u]",
1384 : dstRefreshAddrInfo.memType);
1385 0 : return HCCL_E_INTERNAL;
1386 : }
1387 :
1388 : // 获得dst rank对应的send count/size
1389 0 : CHK_PRT_RET(
1390 : dstRank >= rankSize,
1391 : HCCL_ERROR(
1392 : "[OpUnfoldCacheEntry][GetTransferCountForAlltoallv] dstRank[%u] >= rankSize[%u]", dstRank, rankSize),
1393 : HCCL_E_INTERNAL);
1394 0 : count = alltoallvSendRecvInfo.sendCounts[dstRank];
1395 0 : size = count * SIZE_TABLE[alltoallvSendRecvInfo.sendType];
1396 :
1397 0 : HCCL_INFO(
1398 : "[OpUnfoldCacheEntry][GetTransferCountForAlltoallv] sendCount[%llu] sendSize[%llu] dstRank[%u]", count,
1399 : size, dstRank);
1400 0 : } else if (dstRefreshAddrInfo.memType == RefreshAddrInfo::USER_OUTPUT_MEMTYPE) { // srcAddr不是user input,
1401 : // 但dstAddr是user output (remote
1402 : // copy: remote hccl input -> local
1403 : // user output)
1404 0 : CHK_PRT_RET(
1405 : srcRefreshAddrInfo.memType != RefreshAddrInfo::HCCL_INPUT_MEMTYPE,
1406 : HCCL_ERROR(
1407 : "[OpUnfoldCacheEntry][GetTransferCountForAlltoallv] invalid srcMemType[%u] for remote copy",
1408 : srcRefreshAddrInfo.memType),
1409 : HCCL_E_INTERNAL);
1410 :
1411 : // 获得src rank对应的recv count/size
1412 0 : uint32_t srcRank = srcRefreshAddrInfo.rankId;
1413 0 : CHK_PRT_RET(
1414 : srcRank >= rankSize,
1415 : HCCL_ERROR(
1416 : "[OpUnfoldCacheEntry][GetTransferCountForAlltoallv] srcRank[%u] >= rankSize[%u]", srcRank, rankSize),
1417 : HCCL_E_INTERNAL);
1418 0 : count = alltoallvSendRecvInfo.recvCounts[srcRank];
1419 0 : size = count * SIZE_TABLE[alltoallvSendRecvInfo.recvType];
1420 :
1421 0 : HCCL_INFO(
1422 : "[OpUnfoldCacheEntry][GetTransferCountForAlltoallv] recvCount[%llu] recvSize[%llu] srcRank[%u]", count,
1423 : size, srcRank);
1424 : } else {
1425 0 : HCCL_ERROR(
1426 : "[OpUnfoldCacheEntry][GetTransferCountForAlltoallv] invalid srcMemType[%u] and dstMemType[%u]",
1427 : srcRefreshAddrInfo.memType, dstRefreshAddrInfo.memType);
1428 0 : return HCCL_E_INTERNAL;
1429 : }
1430 :
1431 : // 一定不是大数据量的alltoallv, 否则会在aicpu communicator侧被拦截, 不会进入cache
1432 0 : CHK_PRT_RET(
1433 : size > alltoallvMetadata.sdmaDataBlockSize || size > HCCL_SDMA_MAX_COUNT_4GB,
1434 : HCCL_ERROR(
1435 : "[OpUnfoldCacheEntry][GetTransferCountForAlltoallv] invalid size[%u] sdmaDataBlockSize[%u] 4GB[%u]", size,
1436 : alltoallvMetadata.sdmaDataBlockSize, HCCL_SDMA_MAX_COUNT_4GB),
1437 : HCCL_E_INTERNAL);
1438 :
1439 0 : return HCCL_SUCCESS;
1440 : }
1441 :
1442 0 : HcclResult OpUnfoldCacheEntry::UpdateMemcpySqeForAlltoallv(
1443 : uint8_t* sqePtr, uint8_t* sqeTypePtr, const uint16_t curTaskId, const RefreshAddrInfo& srcRefreshAddrInfo,
1444 : const RefreshAddrInfo& dstRefreshAddrInfo, const std::vector<OpUnfoldMemRange>& curUserInputMemRanges,
1445 : const std::vector<OpUnfoldMemRange>& curUserOutputMemRanges,
1446 : [[maybe_unused]] const AlltoallvMetadata& alltoallvMetadata, const AlltoallvSendRecvInfo& alltoallvSendRecvInfo,
1447 : const uint64_t count, const uint64_t size)
1448 : {
1449 : // 获取src/dst memType
1450 0 : const uint8_t srcMemType = srcRefreshAddrInfo.memType;
1451 0 : const uint8_t dstMemType = dstRefreshAddrInfo.memType;
1452 :
1453 : // 更新SQE (count > 0)或者生成SQE (count = 0)
1454 0 : const uint32_t rankSize = alltoallvSendRecvInfo.sendOffsets.size();
1455 0 : rtStarsMemcpyAsyncSqe_t* memcpyAsyncSqePtr = reinterpret_cast<rtStarsMemcpyAsyncSqe_t*>(sqePtr);
1456 0 : if (count > 0) { // Case 1: memcpy SQE -> memcpy SQE
1457 0 : HCCL_DEBUG(
1458 : "[OpUnfoldCacheEntry][UpdateMemcpySqeForAlltoallv] case 1: memcpy -> memcpy; curTaskId[%u]", curTaskId);
1459 :
1460 : // 校验length
1461 0 : CHK_PRT_RET(
1462 : size == 0,
1463 : HCCL_ERROR("[OpUnfoldCacheEntry][UpdateMemcpySqeForAlltoallv] size[%u] for positive count", size),
1464 : HCCL_E_INTERNAL);
1465 :
1466 : // 更新task id
1467 0 : memcpyAsyncSqePtr->header.taskId = curTaskId;
1468 :
1469 : // 更新length
1470 0 : memcpyAsyncSqePtr->length = static_cast<uint32_t>(size);
1471 :
1472 : // 更新src/dst addr
1473 0 : if (srcMemType == RefreshAddrInfo::USER_INPUT_MEMTYPE) { // LocalCopy/PrepareIntraData
1474 : // 获取send offset
1475 0 : const uint32_t dstRank
1476 : = dstRefreshAddrInfo.rankId; // LocalCopy下是localRank, PrepareIntraData下是remoteRank
1477 0 : CHK_PRT_RET(
1478 : dstRank >= rankSize,
1479 : HCCL_ERROR(
1480 : "[OpUnfoldCacheEntry][UpdateMemcpySqeForAlltoallv] dstRank[%u] >= rankSize[%u]", dstRank, rankSize),
1481 : HCCL_E_INTERNAL);
1482 0 : const uint64_t sendOffset = alltoallvSendRecvInfo.sendOffsets[dstRank];
1483 :
1484 : // 更新src addr (local user input)
1485 0 : uint64_t sqeSrcAddr = 0;
1486 0 : CombineUint32ToUint64(sqeSrcAddr, memcpyAsyncSqePtr->src_addr_high, memcpyAsyncSqePtr->src_addr_low);
1487 0 : CHK_RET(RefreshSqeAddr(
1488 : sqeSrcAddr, srcRefreshAddrInfo.rankId, userInputMemRanges_, curUserInputMemRanges, true, sendOffset));
1489 0 : SplitUint64ToUint32(sqeSrcAddr, memcpyAsyncSqePtr->src_addr_high, memcpyAsyncSqePtr->src_addr_low);
1490 :
1491 : // 只有LocalCopy才需要更新dst addr (hccl addr不用刷新)
1492 0 : if (dstMemType
1493 : == RefreshAddrInfo::USER_OUTPUT_MEMTYPE) { // LocalCopy (local user input -> local user output)
1494 : // 获取recv offset
1495 0 : const uint64_t recvOffset = alltoallvSendRecvInfo.recvOffsets[dstRank];
1496 :
1497 : // 更新dst addr (local user output)
1498 0 : uint64_t sqeDstAddr = 0;
1499 0 : CombineUint32ToUint64(sqeDstAddr, memcpyAsyncSqePtr->dst_addr_high, memcpyAsyncSqePtr->dst_addr_low);
1500 0 : CHK_RET(RefreshSqeAddr(
1501 : sqeDstAddr, dstRank, userOutputMemRanges_, curUserOutputMemRanges, true, recvOffset));
1502 0 : SplitUint64ToUint32(sqeDstAddr, memcpyAsyncSqePtr->dst_addr_high, memcpyAsyncSqePtr->dst_addr_low);
1503 : }
1504 : } else { // RemoteCopy
1505 : // 获取recv offset
1506 0 : uint32_t srcRank = srcRefreshAddrInfo.rankId;
1507 0 : CHK_PRT_RET(
1508 : srcRank >= rankSize,
1509 : HCCL_ERROR(
1510 : "[OpUnfoldCacheEntry][UpdateMemcpySqeForAlltoallv] srcRank[%u] >= rankSize[%u]", srcRank, rankSize),
1511 : HCCL_E_INTERNAL);
1512 0 : const uint64_t recvOffset = alltoallvSendRecvInfo.recvOffsets[srcRank];
1513 :
1514 : // 更新dst addr (local user output)
1515 0 : uint64_t sqeDstAddr = 0;
1516 0 : CombineUint32ToUint64(sqeDstAddr, memcpyAsyncSqePtr->dst_addr_high, memcpyAsyncSqePtr->dst_addr_low);
1517 0 : CHK_RET(RefreshSqeAddr(
1518 : sqeDstAddr, dstRefreshAddrInfo.rankId, userOutputMemRanges_, curUserOutputMemRanges, true, recvOffset));
1519 0 : SplitUint64ToUint32(sqeDstAddr, memcpyAsyncSqePtr->dst_addr_high, memcpyAsyncSqePtr->dst_addr_low);
1520 : }
1521 : } else { // Case 2: memcpy SQE -> placeholder SQE
1522 0 : HCCL_DEBUG(
1523 : "[OpUnfoldCacheEntry][UpdateMemcpySqeForAlltoallv] case 2: memcpy -> placeholder; curTaskId[%u]",
1524 : curTaskId);
1525 :
1526 : // 校验length
1527 0 : CHK_PRT_RET(
1528 : size != 0, HCCL_ERROR("[OpUnfoldCacheEntry][UpdateMemcpySqeForAlltoallv] size[%u] for zero count", size),
1529 : HCCL_E_INTERNAL);
1530 :
1531 : // 保留original SQE中的相关信息
1532 0 : const uint16_t streamId = memcpyAsyncSqePtr->header.rtStreamId;
1533 0 : const uint8_t kernelCredit = memcpyAsyncSqePtr->kernel_credit;
1534 0 : const uint8_t linkType = memcpyAsyncSqePtr->linkType;
1535 0 : const uint32_t qos = memcpyAsyncSqePtr->qos;
1536 0 : uint32_t dstAddrHigh = 0;
1537 0 : uint32_t dstAddrLow = 0;
1538 0 : bool saveDstAddr = false;
1539 0 : if (srcMemType == RefreshAddrInfo::USER_INPUT_MEMTYPE
1540 0 : && dstMemType == RefreshAddrInfo::HCCL_INPUT_MEMTYPE) { // PrepareIntraData
1541 : // 保留dst addr (local hccl input)
1542 : // 注意: 非PrepraeIntraData case下, dst addr为local user output, 会动态计算, 无需保留在placeholder中
1543 0 : dstAddrHigh = memcpyAsyncSqePtr->dst_addr_high;
1544 0 : dstAddrLow = memcpyAsyncSqePtr->dst_addr_low;
1545 0 : saveDstAddr = true;
1546 : }
1547 0 : uint32_t srcAddrHigh = 0;
1548 0 : uint32_t srcAddrLow = 0;
1549 0 : bool saveSrcAddr = false;
1550 0 : if (srcMemType == RefreshAddrInfo::HCCL_INPUT_MEMTYPE
1551 0 : && dstMemType == RefreshAddrInfo::USER_OUTPUT_MEMTYPE) { // RemoteCopy
1552 : // 保留src addr (remote hccl intput)
1553 : // 注意: 非RemoteCopy case下, src addr为local user input, 会动态计算, 无需保留在placeholder中
1554 0 : srcAddrHigh = memcpyAsyncSqePtr->src_addr_high;
1555 0 : srcAddrLow = memcpyAsyncSqePtr->src_addr_low;
1556 0 : saveSrcAddr = true;
1557 : }
1558 :
1559 : // 清空original SQE
1560 0 : CHK_SAFETY_FUNC_RET(memset_s(static_cast<void*>(sqePtr), HCCL_SQE_SIZE, 0, HCCL_SQE_SIZE));
1561 :
1562 : // 生成placeholder SQE
1563 : // 参考aicpu_hccl_sqcqv1.cc中的AddOneCacheMemcpyPlaceholderSqeV1
1564 0 : *sqeTypePtr = SqeType::CACHE_MEMCPY_PLACEHOLDER_SQE;
1565 0 : SetCachePlaceholderHeaderForAlltoallv(streamId, curTaskId, sqePtr);
1566 :
1567 : // 注意: 正常情况下无需设置placeholder SQE中的src/dst addr
1568 : // (只用于第一次算子执行发生cache miss时, 根据memory range获得对应的RefreshAddrInfo, 后续不再使用)
1569 0 : rtStarsPlaceHolderSqe_t* const placeholderSqePtr = (rtStarsPlaceHolderSqe_t*)sqePtr;
1570 0 : if (saveDstAddr) { // PrepareIntraData case下需要保留dst addr (local hccl input)
1571 0 : placeholderSqePtr->u.cache_memcpy_task_info.dst_addr_high = dstAddrHigh;
1572 0 : placeholderSqePtr->u.cache_memcpy_task_info.dst_addr_low = dstAddrLow;
1573 : }
1574 0 : if (saveSrcAddr) { // RemoteCopy case下需要保留src addr (remote hccl input)
1575 0 : placeholderSqePtr->u.cache_memcpy_task_info.src_addr_high = srcAddrHigh;
1576 0 : placeholderSqePtr->u.cache_memcpy_task_info.src_addr_low = srcAddrLow;
1577 : }
1578 0 : placeholderSqePtr->u.cache_memcpy_task_info.kernel_credit = kernelCredit;
1579 0 : placeholderSqePtr->u.cache_memcpy_task_info.linkType = linkType;
1580 0 : placeholderSqePtr->u.cache_memcpy_task_info.qos = qos;
1581 : }
1582 :
1583 0 : return HCCL_SUCCESS;
1584 : }
1585 :
1586 0 : HcclResult OpUnfoldCacheEntry::UpdateMemcpyPlaceholderSqeForAlltoallv(
1587 : uint8_t* sqePtr, uint8_t* sqeTypePtr, const uint16_t curTaskId, const RefreshAddrInfo& srcRefreshAddrInfo,
1588 : const RefreshAddrInfo& dstRefreshAddrInfo, const std::vector<OpUnfoldMemRange>& curUserInputMemRanges,
1589 : const std::vector<OpUnfoldMemRange>& curUserOutputMemRanges,
1590 : [[maybe_unused]] const AlltoallvMetadata& alltoallvMetadata, const AlltoallvSendRecvInfo& alltoallvSendRecvInfo,
1591 : const uint64_t count, const uint64_t size)
1592 : {
1593 : // 更新SQE (count = 0)或者生成SQE (count > 0)
1594 0 : const uint32_t rankSize = alltoallvSendRecvInfo.sendOffsets.size();
1595 0 : rtStarsPlaceHolderSqe_t* placeholderSqePtr = reinterpret_cast<rtStarsPlaceHolderSqe_t*>(sqePtr);
1596 0 : if (count == 0) { // Case 3: placeholder SQE -> placeholder SQE
1597 0 : HCCL_DEBUG(
1598 : "[OpUnfoldCacheEntry][UpdateMemcpyPlaceholderSqeForAlltoallv] case 3: placeholder -> placeholder; "
1599 : "curTaskId[%u]",
1600 : curTaskId);
1601 :
1602 : // 校验length
1603 0 : CHK_PRT_RET(
1604 : size != 0,
1605 : HCCL_ERROR("[OpUnfoldCacheEntry][UpdateMemcpyPlaceholderSqeForAlltoallv] size[%u] for zero count", size),
1606 : HCCL_E_INTERNAL);
1607 :
1608 : // 更新task id
1609 0 : placeholderSqePtr->header.taskId = curTaskId;
1610 :
1611 : // 注意: 无需更新placeholder SQE中的src/dst addr (只用于第一次算子执行发生cache miss时, 根据memory
1612 : // range获得对应的RefreshAddrInfo, 后续不再使用)
1613 : } else { // Case 4: placeholder SQE -> memcpy SQE
1614 0 : HCCL_DEBUG(
1615 : "[OpUnfoldCacheEntry][UpdateMemcpyPlaceholderSqeForAlltoallv] case 4: placeholder -> memcpy; curTaskId[%u]",
1616 : curTaskId);
1617 :
1618 : // 校验length
1619 0 : CHK_PRT_RET(
1620 : size == 0,
1621 : HCCL_ERROR(
1622 : "[OpUnfoldCacheEntry][UpdateMemcpyPlaceholderSqeForAlltoallv] size[%u] for positive count", size),
1623 : HCCL_E_INTERNAL);
1624 :
1625 : // 保留original SQE中的相关信息
1626 0 : const uint16_t streamId = placeholderSqePtr->header.rtStreamId;
1627 0 : const uint8_t kernelCredit = placeholderSqePtr->u.cache_memcpy_task_info.kernel_credit;
1628 0 : const uint8_t linkType = placeholderSqePtr->u.cache_memcpy_task_info.linkType;
1629 0 : const uint32_t qos = placeholderSqePtr->u.cache_memcpy_task_info.qos;
1630 :
1631 : // 准备src/dst addr
1632 0 : uint64_t sqeSrcAddr = 0;
1633 0 : uint64_t sqeDstAddr = 0;
1634 0 : if (srcRefreshAddrInfo.memType == RefreshAddrInfo::USER_INPUT_MEMTYPE) { // LocalCopy/PrepareIntraData
1635 : // 获取send offset
1636 0 : const uint32_t dstRank
1637 : = dstRefreshAddrInfo.rankId; // LocalCopy下是localRank, PrepareIntraData下是remoteRank
1638 0 : CHK_PRT_RET(
1639 : dstRank >= rankSize,
1640 : HCCL_ERROR(
1641 : "[OpUnfoldCacheEntry][UpdateMemcpyPlaceholderSqeForAlltoallv] dstRank[%u] >= rankSize[%u]", dstRank,
1642 : rankSize),
1643 : HCCL_E_INTERNAL);
1644 0 : const uint64_t sendOffset = alltoallvSendRecvInfo.sendOffsets[dstRank];
1645 :
1646 : // 获得src addr (local user input)
1647 0 : const uint32_t localRank = srcRefreshAddrInfo.rankId;
1648 0 : CHK_PRT_RET(
1649 : localRank >= rankSize,
1650 : HCCL_ERROR(
1651 : "[OpUnfoldCacheEntry][UpdateMemcpyPlaceholderSqeForAlltoallv] localRank[%u] >= rankSize[%u]",
1652 : localRank, rankSize),
1653 : HCCL_E_INTERNAL);
1654 0 : const uint64_t localUserInputBaseAddr = curUserInputMemRanges[localRank].baseAddr;
1655 0 : sqeSrcAddr = localUserInputBaseAddr + sendOffset;
1656 :
1657 0 : if (dstRefreshAddrInfo.memType == RefreshAddrInfo::USER_OUTPUT_MEMTYPE) { // LocalCopy
1658 : // 获取recv offset
1659 0 : const uint64_t recvOffset = alltoallvSendRecvInfo.recvOffsets[dstRank]; // dstRank = localRank
1660 :
1661 : // 获得dst addr (local user output)
1662 0 : const uint64_t localUserOutputBaseAddr = curUserOutputMemRanges[dstRank].baseAddr;
1663 0 : sqeDstAddr = localUserOutputBaseAddr + recvOffset;
1664 0 : } else if (dstRefreshAddrInfo.memType == RefreshAddrInfo::HCCL_INPUT_MEMTYPE) { // PrepareIntraData
1665 : // 从placeholder中获取dst addr (local hccl input)
1666 0 : const uint32_t dstAddrHigh = placeholderSqePtr->u.cache_memcpy_task_info.dst_addr_high;
1667 0 : const uint32_t dstAddrLow = placeholderSqePtr->u.cache_memcpy_task_info.dst_addr_low;
1668 0 : CombineUint32ToUint64(sqeDstAddr, dstAddrHigh, dstAddrLow);
1669 : } else {
1670 0 : HCCL_ERROR(
1671 : "[OpUnfoldCacheEntry][UpdateMemcpyPlaceholderSqeForAlltoallv] invalid srcMemType[%u] "
1672 : "dstMemType[%u]",
1673 : srcRefreshAddrInfo.memType, dstRefreshAddrInfo.memType);
1674 0 : return HCCL_E_INTERNAL;
1675 : }
1676 : } else { // RemoteCopy
1677 : // 从placeholder中获取src addr (remote hccl input)
1678 0 : const uint32_t srcAddrHigh = placeholderSqePtr->u.cache_memcpy_task_info.src_addr_high;
1679 0 : const uint32_t srcAddrLow = placeholderSqePtr->u.cache_memcpy_task_info.src_addr_low;
1680 0 : CombineUint32ToUint64(sqeSrcAddr, srcAddrHigh, srcAddrLow);
1681 :
1682 : // 获得remote src rank
1683 0 : const uint32_t srcRank = srcRefreshAddrInfo.rankId;
1684 0 : CHK_PRT_RET(
1685 : srcRank >= rankSize,
1686 : HCCL_ERROR(
1687 : "[OpUnfoldCacheEntry][UpdateMemcpyPlaceholderSqeForAlltoallv] srcRank[%u] >= rankSize[%u]", srcRank,
1688 : rankSize),
1689 : HCCL_E_INTERNAL);
1690 :
1691 : // 获得recv offset
1692 0 : const uint64_t recvOffset = alltoallvSendRecvInfo.recvOffsets[srcRank]; // srcRank = remoteRank
1693 :
1694 : // 获得dst addr
1695 0 : const uint32_t localRank = dstRefreshAddrInfo.rankId;
1696 0 : CHK_PRT_RET(
1697 : localRank >= rankSize,
1698 : HCCL_ERROR(
1699 : "[OpUnfoldCacheEntry][UpdateMemcpyPlaceholderSqeForAlltoallv] localRank[%u] >= rankSize[%u]",
1700 : localRank, rankSize),
1701 : HCCL_E_INTERNAL);
1702 0 : const uint64_t localUserOutputBaseAddr = curUserOutputMemRanges[localRank].baseAddr;
1703 0 : sqeDstAddr = localUserOutputBaseAddr + recvOffset;
1704 : }
1705 :
1706 : // 清空original SQE
1707 0 : CHK_SAFETY_FUNC_RET(memset_s(static_cast<void*>(sqePtr), HCCL_SQE_SIZE, 0, HCCL_SQE_SIZE));
1708 :
1709 : // 生成memcpy SQE
1710 : // 参考aicpu_hccl_sqcqv1.cc中的AddOneMemcpySqeV1
1711 0 : *sqeTypePtr = SqeType::MEMCPY_ASYNC_SQE;
1712 0 : rtStarsMemcpyAsyncSqe_t* memcpySqePtr = reinterpret_cast<rtStarsMemcpyAsyncSqe_t*>(sqePtr);
1713 0 : memcpySqePtr->header.type = RT_STARS_SQE_TYPE_SDMA;
1714 0 : memcpySqePtr->header.rtStreamId = streamId;
1715 0 : memcpySqePtr->header.taskId = curTaskId;
1716 0 : memcpySqePtr->kernel_credit = kernelCredit;
1717 0 : memcpySqePtr->opcode = 0U;
1718 0 : memcpySqePtr->length = static_cast<uint32_t>(size);
1719 0 : SplitUint64ToUint32(sqeSrcAddr, memcpySqePtr->src_addr_high, memcpySqePtr->src_addr_low);
1720 0 : SplitUint64ToUint32(sqeDstAddr, memcpySqePtr->dst_addr_high, memcpySqePtr->dst_addr_low);
1721 0 : memcpySqePtr->sssv = 1U;
1722 0 : memcpySqePtr->dssv = 1U;
1723 0 : memcpySqePtr->sns = 1U;
1724 0 : memcpySqePtr->dns = 1U;
1725 0 : memcpySqePtr->qos = 6; // 6 is HCCL QoS
1726 0 : const uint32_t partId = 0; // 参考dispatcher_aicpu.cc中addOneMemcpySqe_的partId传参始终为0
1727 0 : memcpySqePtr->partid = partId;
1728 0 : memcpySqePtr->linkType = linkType;
1729 0 : memcpySqePtr->qos = qos;
1730 : }
1731 :
1732 0 : return HCCL_SUCCESS;
1733 : }
1734 :
1735 0 : HcclResult OpUnfoldCacheEntry::RefreshSqeAddr(
1736 : uint64_t& sqeAddr, const uint32_t rankId, const std::vector<OpUnfoldMemRange>& cachedMemRanges,
1737 : const std::vector<OpUnfoldMemRange>& curMemRanges, const bool isAlltoallv, const uint64_t offset) const
1738 : {
1739 0 : CHK_PRT_RET(
1740 : rankId == INVALID_VALUE_RANKID, HCCL_ERROR("[OpUnfoldCacheEntry][RefreshSqeAddr] invalid rankId"),
1741 : HCCL_E_INTERNAL);
1742 0 : CHK_PRT_RET(
1743 : rankId >= cachedMemRanges.size(),
1744 : HCCL_ERROR(
1745 : "[OpUnfoldCacheEntry][RefreshSqeAddr] rankId %u exceeds rankSize %u", rankId, cachedMemRanges.size()),
1746 : HCCL_E_INTERNAL);
1747 :
1748 : // 获取缓存的和当前的memory ranges
1749 0 : const OpUnfoldMemRange& cachedMemRange = cachedMemRanges[rankId];
1750 0 : const OpUnfoldMemRange& curMemRange = curMemRanges[rankId];
1751 0 : HCCL_DEBUG(
1752 : "[OpUnfoldCacheEntry][RefreshSqeAddr] cachedMemRange: isValid[%u] baseAddr[0x%016llx] memSize[%llu]; "
1753 : "curMemRange: isValid[%u] baseAddr[0x%016llx] memSize[%llu] isAlltoallv[%u]",
1754 : cachedMemRange.isValid, cachedMemRange.baseAddr, cachedMemRange.memSize, curMemRange.isValid,
1755 : curMemRange.baseAddr, curMemRange.memSize, isAlltoallv);
1756 :
1757 : // 刷新前地址校验
1758 : // (i) user memory: 非V类/alltoallv, SQE addr字段一定在user memory range内, 才需要调用本函数刷新addr
1759 : // (ii) hccl memory: alltoallv, SQE addr一定在hccl memory range内, 才需要调用本函数生成addr
1760 0 : bool isInRange = false;
1761 0 : CHK_RET(cachedMemRange.InRange(sqeAddr, isInRange));
1762 0 : CHK_PRT_RET(
1763 : !isInRange,
1764 : HCCL_ERROR(
1765 : "[OpUnfoldCacheEntry][RefreshSqeAddr] sqeAddr[0x%016llx] not in the range of cachedMemRange[0x%016llx, "
1766 : "0x%016llx)",
1767 : sqeAddr, cachedMemRange.baseAddr, cachedMemRange.baseAddr + cachedMemRange.memSize),
1768 : HCCL_E_INTERNAL);
1769 :
1770 : // 刷新SQE addr
1771 0 : uint64_t curOffset = 0;
1772 0 : if (isAlltoallv) { // alltoallv
1773 0 : curOffset = offset; // 使用给定的offset
1774 : } else { // 非V类算子
1775 0 : curOffset = sqeAddr - cachedMemRange.baseAddr; // 计算缓存的sqe addr相对于缓存的base addr的offset
1776 : }
1777 0 : sqeAddr = curMemRange.baseAddr + curOffset; // 用当前的base addr更新sqe addr
1778 :
1779 : // 刷新后地址校验: 一定在当前的memory range内
1780 0 : isInRange = false;
1781 0 : CHK_RET(curMemRange.InRange(sqeAddr, isInRange));
1782 0 : CHK_PRT_RET(
1783 : !isInRange,
1784 : HCCL_ERROR(
1785 : "[OpUnfoldCacheEntry][RefreshSqeAddr] sqeAddr[0x%016llx] not in the range of curMemRange[0x%016llx, "
1786 : "0x%016llx)",
1787 : sqeAddr, curMemRange.baseAddr, curMemRange.baseAddr + curMemRange.memSize),
1788 : HCCL_E_INTERNAL);
1789 :
1790 0 : return HCCL_SUCCESS;
1791 : }
1792 :
1793 0 : HcclResult OpUnfoldCacheEntry::UpdateSyncSqeForAlltoallv(
1794 : uint8_t* sqePtr, uint8_t* sqeTypePtr, const uint16_t curTaskId, const RefreshAddrInfo& srcRefreshAddrInfo,
1795 : const RefreshAddrInfo& dstRefreshAddrInfo, const AlltoallvMetadata& alltoallvMetadata,
1796 : const AlltoallvSendRecvInfo& alltoallvSendRecvInfo)
1797 : {
1798 0 : HCCL_INFO("[OpUnfoldCacheEntry][UpdateSyncSqeForAlltoallv] curTaskId[%u]", curTaskId);
1799 :
1800 : // 校验alltoallv相关参数
1801 0 : CHK_RET(alltoallvMetadata.Check(true));
1802 0 : CHK_RET(alltoallvSendRecvInfo.Check());
1803 0 : const uint32_t rankSize = alltoallvSendRecvInfo.sendOffsets.size();
1804 0 : CHK_PRT_RET(
1805 : rankSize != alltoallvMetadata.hcclInputMemRanges.size(),
1806 : HCCL_ERROR(
1807 : "[OpUnfoldCacheEntry][UpdateSyncSqeForAlltoallv] hcclInputMemRanges.size[%u] != rankSize[%u]",
1808 : alltoallvMetadata.hcclInputMemRanges.size(), rankSize),
1809 : HCCL_E_INTERNAL);
1810 :
1811 : // 对于NotifyWait, 如果不在notifyid-rank map中, 则无需根据count进行处理, 只需要更新taskid
1812 : // 对于WriteValue/MemcpyRecord, 如果不在signaladdr-rank map中, 则无需根据count进行处理, 只需要更新taskid
1813 0 : if ((*sqeTypePtr) == SqeType::NOTIFY_SQE) {
1814 0 : rtStarsNotifySqeV1_t* notifySqePtr = reinterpret_cast<rtStarsNotifySqeV1_t*>(sqePtr);
1815 0 : const uint32_t notifyId = notifySqePtr->notify_id;
1816 0 : if (alltoallvMetadata.notifyIdRankRflagMap.find(notifyId) == alltoallvMetadata.notifyIdRankRflagMap.cend()) {
1817 0 : notifySqePtr->header.taskId = curTaskId;
1818 0 : return HCCL_SUCCESS;
1819 : }
1820 0 : } else if ((*sqeTypePtr) == SqeType::WRITE_VALUE_SQE) {
1821 0 : rtStarsWriteValueSqe_t* writeValueSqePtr = reinterpret_cast<rtStarsWriteValueSqe_t*>(sqePtr);
1822 0 : uint64_t signalAddr = 0;
1823 0 : const uint32_t lowAddr = writeValueSqePtr->write_addr_low;
1824 0 : const uint32_t highAddr = writeValueSqePtr->write_addr_high;
1825 0 : CombineUint32ToUint64(signalAddr, highAddr, lowAddr);
1826 0 : if (alltoallvMetadata.signalAddrRankRflagMap.find(signalAddr)
1827 0 : == alltoallvMetadata.signalAddrRankRflagMap.cend()) {
1828 0 : writeValueSqePtr->header.taskId = curTaskId;
1829 0 : return HCCL_SUCCESS;
1830 : }
1831 0 : } else if ((*sqeTypePtr) == SqeType::MEMCPY_ASYNC_SQE) {
1832 : // Memcpy-record SQE的src/dst memtype一定是invalid
1833 0 : rtStarsMemcpyAsyncSqe_t* memcpySqePtr = reinterpret_cast<rtStarsMemcpyAsyncSqe_t*>(sqePtr);
1834 0 : CHK_PRT_RET(
1835 : srcRefreshAddrInfo.memType != RefreshAddrInfo::INVALID_MEMTYPE
1836 : || dstRefreshAddrInfo.memType != RefreshAddrInfo::INVALID_MEMTYPE,
1837 : HCCL_ERROR(
1838 : "[OpUnfoldCacheEntry][UpdateTransferSqeForAlltoallv] memcpy-record SQE: "
1839 : "streamId[%u] taskId[%u] curTaskId[%u] srcMemType[%u] dstMemType[%u]",
1840 : memcpySqePtr->header.rtStreamId, memcpySqePtr->header.taskId, curTaskId, srcRefreshAddrInfo.memType,
1841 : dstRefreshAddrInfo.memType),
1842 : HCCL_E_INTERNAL);
1843 :
1844 0 : uint64_t dstSignalAddr = 0;
1845 0 : const uint32_t lowAddr = memcpySqePtr->dst_addr_low;
1846 0 : const uint32_t highAddr = memcpySqePtr->dst_addr_high;
1847 0 : CombineUint32ToUint64(dstSignalAddr, highAddr, lowAddr);
1848 0 : if (alltoallvMetadata.signalAddrRankRflagMap.find(dstSignalAddr)
1849 0 : == alltoallvMetadata.signalAddrRankRflagMap.cend()) {
1850 0 : memcpySqePtr->header.taskId = curTaskId;
1851 0 : return HCCL_SUCCESS;
1852 : }
1853 : }
1854 :
1855 : // 获取当前sync类SQE对应的count和size
1856 0 : uint64_t count = 0;
1857 0 : uint64_t size = 0; // send/recv bytes
1858 0 : CHK_RET(GetTransferCountForAlltoallv(count, size, sqePtr, sqeTypePtr, alltoallvMetadata, alltoallvSendRecvInfo));
1859 :
1860 : // 更新/生成相应SQE
1861 0 : if ((*sqeTypePtr) == SqeType::CACHE_NOTIFY_PLACEHOLDER_SQE) {
1862 0 : CHK_RET(UpdateNotifyPlaceholderSqeForAlltoallv(
1863 : sqePtr, sqeTypePtr, curTaskId, alltoallvMetadata, alltoallvSendRecvInfo, count, size));
1864 0 : } else if ((*sqeTypePtr) == SqeType::CACHE_WRITE_VALUE_PLACEHOLDER_SQE) {
1865 0 : CHK_RET(UpdateWritePlaceholderSqeForAlltoallv(
1866 : sqePtr, sqeTypePtr, curTaskId, alltoallvMetadata, alltoallvSendRecvInfo, count, size));
1867 0 : } else if ((*sqeTypePtr) == SqeType::NOTIFY_SQE) {
1868 0 : CHK_RET(UpdateNotifySqeForAlltoallv(
1869 : sqePtr, sqeTypePtr, curTaskId, alltoallvMetadata, alltoallvSendRecvInfo, count, size));
1870 0 : } else if ((*sqeTypePtr) == SqeType::WRITE_VALUE_SQE) {
1871 0 : CHK_RET(UpdateWriteValueSqeForAlltoallv(
1872 : sqePtr, sqeTypePtr, curTaskId, alltoallvMetadata, alltoallvSendRecvInfo, count, size));
1873 0 : } else if ((*sqeTypePtr) == SqeType::MEMCPY_ASYNC_SQE) { // MemcpyRecord SQE
1874 0 : CHK_RET(UpdateMemcpyRecordSqeForAlltoallv(
1875 : sqePtr, sqeTypePtr, curTaskId, alltoallvMetadata, alltoallvSendRecvInfo, count, size));
1876 0 : } else if ((*sqeTypePtr) == SqeType::CACHE_MEMCPY_RECORD_PLACEHOLDER_SQE) {
1877 0 : CHK_RET(UpdateMemcpyRecordPlaceholderSqeForAlltoallv(
1878 : sqePtr, sqeTypePtr, curTaskId, alltoallvMetadata, alltoallvSendRecvInfo, count, size));
1879 : } else {
1880 0 : HCCL_ERROR("[OpUnfoldCacheEntry][UpdateTransferSqeForAlltoallv] invalid sqeType[%u]", *sqeTypePtr);
1881 0 : return HCCL_E_INTERNAL;
1882 : }
1883 :
1884 0 : return HCCL_SUCCESS;
1885 : }
1886 :
1887 0 : HcclResult OpUnfoldCacheEntry::GetTransferCountForAlltoallv(
1888 : uint64_t& count, uint64_t& size, const uint8_t* sqePtr, const uint8_t* sqeTypePtr,
1889 : const AlltoallvMetadata& alltoallvMetadata, const AlltoallvSendRecvInfo& alltoallvSendRecvInfo) const
1890 : {
1891 : // 获得notifyId或者signalAddr
1892 0 : bool isNotify = false;
1893 0 : uint32_t notifyId = 0;
1894 0 : uint64_t signalAddr = 0;
1895 0 : if ((*sqeTypePtr) == SqeType::NOTIFY_SQE) {
1896 : // 注意: 目前只有NotifyWait可能会生成cache-notify placeholder
1897 0 : const rtStarsNotifySqeV1_t* notifySqePtr = reinterpret_cast<const rtStarsNotifySqeV1_t*>(sqePtr);
1898 0 : CHK_PRT_RET(
1899 : notifySqePtr->header.type != RT_STARS_SQE_TYPE_NOTIFY_WAIT,
1900 : HCCL_ERROR("[OpUnfoldCacheEntry][GetTransferCountForAlltoallv] header.type[%u]", notifySqePtr->header.type),
1901 : HCCL_E_INTERNAL);
1902 :
1903 0 : notifyId = notifySqePtr->notify_id;
1904 0 : isNotify = true;
1905 0 : } else if ((*sqeTypePtr) == SqeType::CACHE_NOTIFY_PLACEHOLDER_SQE) {
1906 : // 注意: 目前只会存在对应NotifyWait的placeholder
1907 0 : const rtStarsPlaceHolderSqe_t* placeholderSqePtr = reinterpret_cast<const rtStarsPlaceHolderSqe_t*>(sqePtr);
1908 0 : CHK_PRT_RET(
1909 : placeholderSqePtr->u.cache_notify_task_info.is_wait != 1,
1910 : HCCL_ERROR(
1911 : "[OpUnfoldCacheEntry][GetTransferCountForAlltoallv] is_wait[%u]",
1912 : placeholderSqePtr->u.cache_notify_task_info.is_wait),
1913 : HCCL_E_INTERNAL);
1914 :
1915 0 : notifyId = placeholderSqePtr->u.cache_notify_task_info.notify_id;
1916 0 : isNotify = true;
1917 0 : } else if ((*sqeTypePtr) == SqeType::WRITE_VALUE_SQE) {
1918 0 : const rtStarsWriteValueSqe_t* writeValueSqePtr = reinterpret_cast<const rtStarsWriteValueSqe_t*>(sqePtr);
1919 0 : const uint32_t lowAddr = writeValueSqePtr->write_addr_low;
1920 0 : const uint32_t highAddr = writeValueSqePtr->write_addr_high;
1921 0 : CombineUint32ToUint64(signalAddr, highAddr, lowAddr);
1922 0 : isNotify = false;
1923 0 : } else if ((*sqeTypePtr) == SqeType::CACHE_WRITE_VALUE_PLACEHOLDER_SQE) {
1924 0 : const rtStarsPlaceHolderSqe_t* placeholderSqePtr = reinterpret_cast<const rtStarsPlaceHolderSqe_t*>(sqePtr);
1925 0 : const uint32_t lowAddr = placeholderSqePtr->u.cache_write_value_task_info.write_addr_low;
1926 0 : const uint32_t highAddr = placeholderSqePtr->u.cache_write_value_task_info.write_addr_high;
1927 0 : CombineUint32ToUint64(signalAddr, highAddr, lowAddr);
1928 0 : isNotify = false;
1929 0 : } else if ((*sqeTypePtr) == SqeType::MEMCPY_ASYNC_SQE) { // MemcpyRecord SQE
1930 0 : const rtStarsMemcpyAsyncSqe_t* memcpySqePtr = reinterpret_cast<const rtStarsMemcpyAsyncSqe_t*>(sqePtr);
1931 0 : const uint32_t lowAddr = memcpySqePtr->dst_addr_low;
1932 0 : const uint32_t highAddr = memcpySqePtr->dst_addr_high;
1933 0 : CombineUint32ToUint64(signalAddr, highAddr, lowAddr);
1934 0 : isNotify = false;
1935 0 : } else if ((*sqeTypePtr) == SqeType::CACHE_MEMCPY_RECORD_PLACEHOLDER_SQE) {
1936 0 : const rtStarsPlaceHolderSqe_t* placeholderSqePtr = reinterpret_cast<const rtStarsPlaceHolderSqe_t*>(sqePtr);
1937 0 : const uint32_t lowAddr = placeholderSqePtr->u.cache_memcpy_record_task_info.dst_addr_low;
1938 0 : const uint32_t highAddr = placeholderSqePtr->u.cache_memcpy_record_task_info.dst_addr_high;
1939 0 : CombineUint32ToUint64(signalAddr, highAddr, lowAddr);
1940 0 : isNotify = false;
1941 : } else {
1942 0 : HCCL_ERROR("[OpUnfoldCacheEntry][GetTransferCountForAlltoallv] invalid sqeType[%u]", *sqeTypePtr);
1943 0 : return HCCL_E_INTERNAL;
1944 : }
1945 0 : HCCL_INFO(
1946 : "[OpUnfoldCacheEntry][GetTransferCountForAlltoallv] notifyId[%u] signalAddr[0x%016llx]", notifyId, signalAddr);
1947 :
1948 : // 获得remoteRank
1949 0 : uint32_t remoteRank = 0;
1950 0 : bool recvFlag = false;
1951 0 : if (isNotify) {
1952 : // 校验notifyId一定在notifyid-rank map中
1953 : std::unordered_map<uint32_t, RankRflag>::const_iterator mapIter
1954 0 : = alltoallvMetadata.notifyIdRankRflagMap.find(notifyId);
1955 0 : CHK_PRT_RET(
1956 : mapIter == alltoallvMetadata.notifyIdRankRflagMap.cend(),
1957 : HCCL_ERROR(
1958 : "[OpUnfoldCacheEntry][GetTransferCountForAlltoallv] notifyId[%u] not in alltoallvMetadata", notifyId),
1959 : HCCL_E_INTERNAL);
1960 :
1961 : // 根据notifyId获得remoteRank
1962 0 : remoteRank = mapIter->second.first;
1963 0 : recvFlag = mapIter->second.second;
1964 : } else {
1965 : // 校验signalAddr一定在signalAddr-rank map中
1966 : std::unordered_map<uint64_t, RankRflag>::const_iterator mapIter
1967 0 : = alltoallvMetadata.signalAddrRankRflagMap.find(signalAddr);
1968 0 : CHK_PRT_RET(
1969 : mapIter == alltoallvMetadata.signalAddrRankRflagMap.cend(),
1970 : HCCL_ERROR(
1971 : "[OpUnfoldCacheEntry][GetTransferCountForAlltoallv] signalAddr[%llu] not in alltoallvMetadata",
1972 : signalAddr),
1973 : HCCL_E_INTERNAL);
1974 :
1975 : // 根据signalAddr获得remoteRank
1976 0 : remoteRank = mapIter->second.first;
1977 0 : recvFlag = mapIter->second.second;
1978 : }
1979 0 : HCCL_INFO("[OpUnfoldCacheEntry][GetTransferCountForAlltoallv] remoteRank[%u] recvFlag[%u]", remoteRank, recvFlag);
1980 :
1981 : // 校验remoteRank (理论上一定不是localRank)
1982 0 : const uint32_t rankSize = alltoallvSendRecvInfo.sendOffsets.size();
1983 0 : CHK_PRT_RET(
1984 : remoteRank >= rankSize,
1985 : HCCL_ERROR(
1986 : "[OpUnfoldCacheEntry][GetTransferCountForAlltoallv] remoteRank[%u] >= rankSize[%u]", remoteRank, rankSize),
1987 : HCCL_E_INTERNAL);
1988 :
1989 : // 根据remoteRank获得count和size
1990 0 : if (recvFlag) {
1991 0 : count = alltoallvSendRecvInfo.recvCounts[remoteRank];
1992 0 : size = count * SIZE_TABLE[alltoallvSendRecvInfo.recvType];
1993 0 : HCCL_INFO("[OpUnfoldCacheEntry][GetTransferCountForAlltoallv] recvCount[%llu] recvSize[%llu]", count, size);
1994 : } else {
1995 0 : count = alltoallvSendRecvInfo.sendCounts[remoteRank];
1996 0 : size = count * SIZE_TABLE[alltoallvSendRecvInfo.sendType];
1997 0 : HCCL_INFO("[OpUnfoldCacheEntry][GetTransferCountForAlltoallv] sendCount[%llu] sendSize[%llu]", count, size);
1998 : }
1999 :
2000 0 : return HCCL_SUCCESS;
2001 : }
2002 :
2003 0 : HcclResult OpUnfoldCacheEntry::UpdateNotifyPlaceholderSqeForAlltoallv(
2004 : uint8_t* sqePtr, uint8_t* sqeTypePtr, const uint16_t curTaskId,
2005 : [[maybe_unused]] const AlltoallvMetadata& alltoallvMetadata,
2006 : [[maybe_unused]] const AlltoallvSendRecvInfo& alltoallvSendRecvInfo, const uint64_t count, const uint64_t size)
2007 : {
2008 : UNUSED_PARAM(size);
2009 :
2010 0 : rtStarsPlaceHolderSqe_t* placeholderSqePtr = reinterpret_cast<rtStarsPlaceHolderSqe_t*>(sqePtr);
2011 0 : if (count == 0) { // 只需要更新task id
2012 0 : placeholderSqePtr->header.taskId = curTaskId;
2013 : } else { // 需要将cache-notify placeholder转变成NotifyWait
2014 : // 保留original SQE中的相关信息
2015 0 : const uint16_t streamId = placeholderSqePtr->header.rtStreamId;
2016 0 : const uint8_t kernel_credit = placeholderSqePtr->u.cache_notify_task_info.kernel_credit;
2017 0 : const uint32_t timeout = placeholderSqePtr->u.cache_notify_task_info.timeout;
2018 0 : const uint32_t notifyId = placeholderSqePtr->u.cache_notify_task_info.notify_id;
2019 :
2020 : // 清空original SQE
2021 0 : CHK_SAFETY_FUNC_RET(memset_s(static_cast<void*>(sqePtr), HCCL_SQE_SIZE, 0, HCCL_SQE_SIZE));
2022 :
2023 : // 生成NotifyWait
2024 : // 参考aicpu_hccl_sqcqv1.cc中的AddOneNotifyWaitSqeV1
2025 0 : *sqeTypePtr = SqeType::NOTIFY_SQE;
2026 0 : rtStarsNotifySqeV1_t* const notifySqePtr = (rtStarsNotifySqeV1_t*)sqePtr;
2027 0 : notifySqePtr->header.type = RT_STARS_SQE_TYPE_NOTIFY_WAIT;
2028 0 : notifySqePtr->kernel_credit = kernel_credit;
2029 0 : notifySqePtr->timeout = timeout;
2030 0 : notifySqePtr->header.rtStreamId = streamId;
2031 0 : notifySqePtr->notify_id = notifyId;
2032 0 : notifySqePtr->header.taskId = curTaskId;
2033 : }
2034 :
2035 0 : return HCCL_SUCCESS;
2036 : }
2037 :
2038 0 : HcclResult OpUnfoldCacheEntry::UpdateWritePlaceholderSqeForAlltoallv(
2039 : uint8_t* sqePtr, uint8_t* sqeTypePtr, const uint16_t curTaskId,
2040 : [[maybe_unused]] const AlltoallvMetadata& alltoallvMetadata,
2041 : [[maybe_unused]] const AlltoallvSendRecvInfo& alltoallvSendRecvInfo, const uint64_t count, const uint64_t size)
2042 : {
2043 : UNUSED_PARAM(size);
2044 :
2045 0 : rtStarsPlaceHolderSqe_t* placeholderSqePtr = reinterpret_cast<rtStarsPlaceHolderSqe_t*>(sqePtr);
2046 0 : if (count == 0) { // 只需要更新task id
2047 0 : placeholderSqePtr->header.taskId = curTaskId;
2048 : } else { // 需要将cache-write placeholder转变成WriteValue
2049 : // 保留original SQE中的相关信息
2050 0 : const uint16_t streamId = placeholderSqePtr->header.rtStreamId;
2051 0 : const uint32_t lowAddr = placeholderSqePtr->u.cache_write_value_task_info.write_addr_low;
2052 0 : const uint32_t highAddr = placeholderSqePtr->u.cache_write_value_task_info.write_addr_high;
2053 :
2054 : // 清空original SQE
2055 0 : CHK_SAFETY_FUNC_RET(memset_s(static_cast<void*>(sqePtr), HCCL_SQE_SIZE, 0, HCCL_SQE_SIZE));
2056 :
2057 : // 生成WriteValue
2058 : // 参考aicpu_hccl_sqcqv1.cc中的AddOneWriteValueRecordSqeV1
2059 0 : *sqeTypePtr = SqeType::WRITE_VALUE_SQE;
2060 0 : rtStarsWriteValueSqe_t* const writeValueSqePtr = (rtStarsWriteValueSqe_t*)sqePtr;
2061 0 : writeValueSqePtr->header.type = RT_STARS_SQE_TYPE_WRITE_VALUE;
2062 0 : writeValueSqePtr->header.rtStreamId = streamId;
2063 0 : writeValueSqePtr->header.taskId = curTaskId;
2064 0 : writeValueSqePtr->kernel_credit = RT_STARS_DEFAULT_KERNEL_CREDIT;
2065 0 : writeValueSqePtr->awsize = RT_STARS_WRITE_VALUE_SIZE_TYPE_32BIT;
2066 0 : writeValueSqePtr->write_value_part0 = 1U;
2067 0 : writeValueSqePtr->sub_type = RT_STARS_WRITE_VALUE_SUB_TYPE_NOTIFY_RECORD_IPC_NO_PCIE;
2068 0 : writeValueSqePtr->write_addr_low = lowAddr;
2069 0 : writeValueSqePtr->write_addr_high = highAddr;
2070 : }
2071 :
2072 0 : return HCCL_SUCCESS;
2073 : }
2074 :
2075 0 : HcclResult OpUnfoldCacheEntry::UpdateNotifySqeForAlltoallv(
2076 : uint8_t* sqePtr, uint8_t* sqeTypePtr, const uint16_t curTaskId,
2077 : [[maybe_unused]] const AlltoallvMetadata& alltoallvMetadata,
2078 : [[maybe_unused]] const AlltoallvSendRecvInfo& alltoallvSendRecvInfo, const uint64_t count, const uint64_t size)
2079 : {
2080 : UNUSED_PARAM(size);
2081 :
2082 0 : rtStarsNotifySqeV1_t* notifySqePtr = reinterpret_cast<rtStarsNotifySqeV1_t*>(sqePtr);
2083 0 : if (count > 0) { // 只需要更新task id
2084 0 : notifySqePtr->header.taskId = curTaskId;
2085 : } else { // 需要将NotifyWait转变成cache-notify placeholder
2086 : // 保留original SQE中的相关信息
2087 0 : const uint16_t streamId = notifySqePtr->header.rtStreamId;
2088 0 : const uint8_t kernel_credit = notifySqePtr->kernel_credit;
2089 0 : const uint32_t timeout = notifySqePtr->timeout;
2090 0 : const uint32_t notifyId = notifySqePtr->notify_id;
2091 :
2092 : // 清空original SQE
2093 0 : CHK_SAFETY_FUNC_RET(memset_s(static_cast<void*>(sqePtr), HCCL_SQE_SIZE, 0, HCCL_SQE_SIZE));
2094 :
2095 : // 生成cache-notify placeholder
2096 : // 参考aicpu_hccl_sqcqv1.cc中的AddOneCacheNotifyWaitPlaceholderSqeV1
2097 0 : *sqeTypePtr = SqeType::CACHE_NOTIFY_PLACEHOLDER_SQE;
2098 0 : SetCachePlaceholderHeaderForAlltoallv(streamId, curTaskId, sqePtr);
2099 0 : rtStarsPlaceHolderSqe_t* const placeholderSqePtr = (rtStarsPlaceHolderSqe_t*)sqePtr;
2100 0 : placeholderSqePtr->u.cache_notify_task_info.is_wait = 1; // NotifyWait
2101 0 : placeholderSqePtr->u.cache_notify_task_info.kernel_credit = kernel_credit;
2102 0 : placeholderSqePtr->u.cache_notify_task_info.timeout = timeout;
2103 0 : placeholderSqePtr->u.cache_notify_task_info.notify_id = notifyId;
2104 : }
2105 :
2106 0 : return HCCL_SUCCESS;
2107 : }
2108 :
2109 0 : HcclResult OpUnfoldCacheEntry::UpdateWriteValueSqeForAlltoallv(
2110 : uint8_t* sqePtr, uint8_t* sqeTypePtr, const uint16_t curTaskId,
2111 : [[maybe_unused]] const AlltoallvMetadata& alltoallvMetadata,
2112 : [[maybe_unused]] const AlltoallvSendRecvInfo& alltoallvSendRecvInfo, const uint64_t count, const uint64_t size)
2113 : {
2114 : UNUSED_PARAM(size);
2115 :
2116 0 : rtStarsWriteValueSqe_t* writeValueSqePtr = reinterpret_cast<rtStarsWriteValueSqe_t*>(sqePtr);
2117 0 : if (count > 0) { // 只需要更新task id
2118 0 : writeValueSqePtr->header.taskId = curTaskId;
2119 : } else { // 需要将WriteValue转变成cache-write placeholder
2120 : // 保留original SQE中的相关信息
2121 0 : const uint16_t streamId = writeValueSqePtr->header.rtStreamId;
2122 0 : const uint32_t lowAddr = writeValueSqePtr->write_addr_low;
2123 0 : const uint32_t highAddr = writeValueSqePtr->write_addr_high;
2124 :
2125 : // 清空original SQE
2126 0 : CHK_SAFETY_FUNC_RET(memset_s(static_cast<void*>(sqePtr), HCCL_SQE_SIZE, 0, HCCL_SQE_SIZE));
2127 :
2128 : // 生成cache-write placeholder
2129 : // 参考aicpu_hccl_sqcqv1.cc中的AddOneCacheWriteValuePlaceholderSqeV1
2130 0 : *sqeTypePtr = SqeType::CACHE_WRITE_VALUE_PLACEHOLDER_SQE;
2131 0 : SetCachePlaceholderHeaderForAlltoallv(streamId, curTaskId, sqePtr);
2132 0 : rtStarsPlaceHolderSqe_t* const placeholderSqePtr = (rtStarsPlaceHolderSqe_t*)sqePtr;
2133 0 : placeholderSqePtr->u.cache_write_value_task_info.write_addr_low = lowAddr;
2134 0 : placeholderSqePtr->u.cache_write_value_task_info.write_addr_high = highAddr;
2135 : }
2136 :
2137 0 : return HCCL_SUCCESS;
2138 : }
2139 :
2140 0 : HcclResult OpUnfoldCacheEntry::UpdateMemcpyRecordSqeForAlltoallv(
2141 : uint8_t* sqePtr, uint8_t* sqeTypePtr, const uint16_t curTaskId,
2142 : [[maybe_unused]] const AlltoallvMetadata& alltoallvMetadata,
2143 : [[maybe_unused]] const AlltoallvSendRecvInfo& alltoallvSendRecvInfo, const uint64_t count, const uint64_t size)
2144 : {
2145 : UNUSED_PARAM(size);
2146 :
2147 0 : rtStarsMemcpyAsyncSqe_t* memcpySqePtr = reinterpret_cast<rtStarsMemcpyAsyncSqe_t*>(sqePtr);
2148 0 : if (count > 0) { // 只需要更新task id
2149 0 : memcpySqePtr->header.taskId = curTaskId;
2150 : } else { // 需要将MemcpyRecord转变成cache-memcpy-record placeholder
2151 : // 保留original SQE中的相关信息
2152 0 : const uint16_t streamId = memcpySqePtr->header.rtStreamId;
2153 0 : const uint8_t kernelCredit = memcpySqePtr->kernel_credit;
2154 0 : const uint32_t opCode = memcpySqePtr->opcode;
2155 0 : const uint32_t length = memcpySqePtr->length;
2156 0 : const uint32_t srcAddrLow = memcpySqePtr->src_addr_low;
2157 0 : const uint32_t srcAddrHigh = memcpySqePtr->src_addr_high;
2158 0 : const uint32_t dstAddrLow = memcpySqePtr->dst_addr_low;
2159 0 : const uint32_t dstAddrHigh = memcpySqePtr->dst_addr_high;
2160 0 : const uint32_t partId = memcpySqePtr->partid;
2161 0 : const uint8_t linkType = memcpySqePtr->linkType;
2162 0 : const uint32_t qos = memcpySqePtr->qos;
2163 :
2164 : // 清空original SQE
2165 0 : CHK_SAFETY_FUNC_RET(memset_s(static_cast<void*>(sqePtr), HCCL_SQE_SIZE, 0, HCCL_SQE_SIZE));
2166 :
2167 : // 生成cache-memcpy-record placeholder
2168 : // 参考aicpu_hccl_sqcqv1.cc中的AddOneCacheMemcpyRecordPlaceholderSqeV1
2169 0 : *sqeTypePtr = SqeType::CACHE_MEMCPY_RECORD_PLACEHOLDER_SQE;
2170 0 : SetCachePlaceholderHeaderForAlltoallv(streamId, curTaskId, sqePtr);
2171 0 : rtStarsPlaceHolderSqe_t* const placeholderSqePtr = (rtStarsPlaceHolderSqe_t*)sqePtr;
2172 0 : placeholderSqePtr->u.cache_memcpy_record_task_info.kernel_credit = kernelCredit;
2173 0 : placeholderSqePtr->u.cache_memcpy_record_task_info.opcode = opCode;
2174 0 : placeholderSqePtr->u.cache_memcpy_record_task_info.length = length;
2175 0 : placeholderSqePtr->u.cache_memcpy_record_task_info.src_addr_low = srcAddrLow;
2176 0 : placeholderSqePtr->u.cache_memcpy_record_task_info.src_addr_high = srcAddrHigh;
2177 0 : placeholderSqePtr->u.cache_memcpy_record_task_info.dst_addr_low = dstAddrLow;
2178 0 : placeholderSqePtr->u.cache_memcpy_record_task_info.dst_addr_high = dstAddrHigh;
2179 0 : placeholderSqePtr->u.cache_memcpy_record_task_info.partid = partId;
2180 0 : placeholderSqePtr->u.cache_memcpy_record_task_info.linkType = linkType;
2181 0 : placeholderSqePtr->u.cache_memcpy_record_task_info.qos = qos;
2182 : }
2183 :
2184 0 : return HCCL_SUCCESS;
2185 : }
2186 :
2187 0 : HcclResult OpUnfoldCacheEntry::UpdateMemcpyRecordPlaceholderSqeForAlltoallv(
2188 : uint8_t* sqePtr, uint8_t* sqeTypePtr, const uint16_t curTaskId,
2189 : [[maybe_unused]] const AlltoallvMetadata& alltoallvMetadata,
2190 : [[maybe_unused]] const AlltoallvSendRecvInfo& alltoallvSendRecvInfo, const uint64_t count, const uint64_t size)
2191 : {
2192 : UNUSED_PARAM(size);
2193 :
2194 0 : rtStarsPlaceHolderSqe_t* placeholderSqePtr = reinterpret_cast<rtStarsPlaceHolderSqe_t*>(sqePtr);
2195 0 : if (count == 0) { // 只需要更新task id
2196 0 : placeholderSqePtr->header.taskId = curTaskId;
2197 : } else { // 需要将cache-memcpy-record placeholder转变成MemcpyRecord
2198 : // 保留original SQE中的相关信息
2199 0 : const uint16_t streamId = placeholderSqePtr->header.rtStreamId;
2200 0 : const uint8_t kernelCredit = placeholderSqePtr->u.cache_memcpy_record_task_info.kernel_credit;
2201 0 : const uint32_t opCode = placeholderSqePtr->u.cache_memcpy_record_task_info.opcode;
2202 0 : const uint32_t length = placeholderSqePtr->u.cache_memcpy_record_task_info.length;
2203 0 : const uint32_t srcAddrLow = placeholderSqePtr->u.cache_memcpy_record_task_info.src_addr_low;
2204 0 : const uint32_t srcAddrHigh = placeholderSqePtr->u.cache_memcpy_record_task_info.src_addr_high;
2205 0 : const uint32_t dstAddrLow = placeholderSqePtr->u.cache_memcpy_record_task_info.dst_addr_low;
2206 0 : const uint32_t dstAddrHigh = placeholderSqePtr->u.cache_memcpy_record_task_info.dst_addr_high;
2207 0 : const uint32_t partId = placeholderSqePtr->u.cache_memcpy_record_task_info.partid;
2208 0 : const uint32_t linkType = placeholderSqePtr->u.cache_memcpy_record_task_info.linkType;
2209 0 : const uint32_t qos = placeholderSqePtr->u.cache_memcpy_record_task_info.qos;
2210 :
2211 : // 清空original SQE
2212 0 : CHK_SAFETY_FUNC_RET(memset_s(static_cast<void*>(sqePtr), HCCL_SQE_SIZE, 0, HCCL_SQE_SIZE));
2213 :
2214 : // 生成MemcpyRecord
2215 : // 参考aicpu_hccl_sqcqv1.cc中的AddOneMemcpySqeV1
2216 0 : *sqeTypePtr = SqeType::MEMCPY_ASYNC_SQE;
2217 0 : rtStarsMemcpyAsyncSqe_t* const memcpySqePtr = (rtStarsMemcpyAsyncSqe_t*)sqePtr;
2218 0 : memcpySqePtr->header.type = RT_STARS_SQE_TYPE_SDMA;
2219 0 : memcpySqePtr->header.rtStreamId = streamId;
2220 0 : memcpySqePtr->header.taskId = curTaskId;
2221 0 : memcpySqePtr->kernel_credit = kernelCredit;
2222 0 : memcpySqePtr->opcode = opCode;
2223 0 : memcpySqePtr->length = length;
2224 0 : memcpySqePtr->src_addr_low = srcAddrLow;
2225 0 : memcpySqePtr->src_addr_high = srcAddrHigh;
2226 0 : memcpySqePtr->dst_addr_low = dstAddrLow;
2227 0 : memcpySqePtr->dst_addr_high = dstAddrHigh;
2228 0 : memcpySqePtr->sssv = 1U;
2229 0 : memcpySqePtr->dssv = 1U;
2230 0 : memcpySqePtr->sns = 1U;
2231 0 : memcpySqePtr->dns = 1U;
2232 0 : memcpySqePtr->qos = 6; // 6 is HCCL QoS
2233 0 : memcpySqePtr->partid = partId;
2234 0 : memcpySqePtr->linkType = linkType;
2235 0 : memcpySqePtr->qos = qos;
2236 : }
2237 :
2238 0 : return HCCL_SUCCESS;
2239 : }
2240 :
2241 0 : void OpUnfoldCacheEntry::SetCachePlaceholderHeaderForAlltoallv(
2242 : const uint16_t streamId, const uint16_t taskId, uint8_t* sqePtr)
2243 : {
2244 : // 参考aicpu_hccl_sqcqv1.cc中的SetCachePlaceholderHeaderV1
2245 : // 注意: 不直接调用SetCachePlaceholderHeaderV1, 避免libhccl_plf对platform/task/rtsq_interact产生依赖
2246 : // 目前rtsq_interact只编译到ccl_kernel_plf与ccl_kernel_plf_a中
2247 0 : rtStarsPlaceHolderSqe_t* placeholderSqePtr = (rtStarsPlaceHolderSqe_t*)sqePtr;
2248 0 : placeholderSqePtr->header.type = RT_STARS_SQE_TYPE_PLACE_HOLDER;
2249 0 : placeholderSqePtr->header.ie = 0U;
2250 0 : placeholderSqePtr->header.preP = 0U; // 不需要STARS_FW参与任何预处理
2251 0 : placeholderSqePtr->header.postP = 0U;
2252 0 : placeholderSqePtr->header.wrCqe = 0U;
2253 0 : placeholderSqePtr->header.reserved = 0U;
2254 : // NOTE: task type在preP阶段被TASK_FW使用, 而此placeholder无preP阶段, 设置为RT_TASK_TYPE_FLIP不影响功能
2255 0 : placeholderSqePtr->header.blockDim = RT_TASK_TYPE_FLIP;
2256 0 : placeholderSqePtr->header.rtStreamId = streamId;
2257 0 : placeholderSqePtr->header.taskId = taskId;
2258 0 : placeholderSqePtr->kernel_credit = RT_STARS_DEFAULT_KERNEL_CREDIT;
2259 0 : return;
2260 : }
2261 :
2262 : }; // namespace hccl
|
