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