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