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 : #include <chrono>
11 : #include <unordered_map>
12 : #include "rtsq_a5.h"
13 : #include "log.h"
14 : #include "exception_util.h"
15 : #include "internal_exception.h"
16 : #include "sqe_build_a5.h"
17 : #include "sqe.h"
18 : #ifdef CCL_KERNEL_AICPU
19 : #include "aicpu_ts_primitives_c_adpt.h"
20 : #endif
21 : #include "aicpu_task_utils.h"
22 :
23 : namespace Hccl {
24 : using namespace std;
25 : constexpr u32 RTSQ_A5_PART_ID = 0;
26 : constexpr u32 PRINT_INTERVAL = 30;
27 :
28 288 : RtsqA5::RtsqA5(u32 devPhyId, u32 streamId, u32 sqId) : RtsqBase(devPhyId, streamId, sqId) { SetTaskIdBySqeId(); }
29 :
30 0 : RtsqA5::RtsqA5(u32 devPhyId, u32 streamId, u32 sqId, bool launchFlag) : RtsqBase(devPhyId, streamId, sqId)
31 : {
32 0 : SetTaskIdBySqeId();
33 0 : launchFlag_ = launchFlag;
34 0 : }
35 :
36 1 : void RtsqA5::Reset()
37 : {
38 1 : RtsqBase::Reset();
39 1 : pendingSqeCnt = 0;
40 1 : s32 sRet = memset_s(locBuf, RTSQ_SQE_SIZE * PER_LAUNCH_SQE_CNT, 0, RTSQ_SQE_SIZE * PER_LAUNCH_SQE_CNT);
41 1 : if (UNLIKELY(sRet != EOK)) {
42 0 : auto msg = StringFormat("[RtsqA5][Reset] locBuf memset fail. errorno[%d]", sRet);
43 0 : THROW<InternalException>(msg);
44 0 : }
45 3 : HCCL_INFO("[NsRecovery]RtsqA5::%s success", __func__);
46 1 : }
47 :
48 : // 计算head和tail之间的距离
49 11 : u32 RtsqA5::GetTailToHeadDist() const
50 : {
51 11 : if (UNLIKELY(sqHead_ == sqTail_)) { // 头尾相同，则距离大小为sq深度
52 8 : return sqDepth_;
53 : }
54 3 : return (sqTail_ < sqHead_) ? (sqHead_ - sqTail_) : (sqDepth_ - (sqTail_ - sqHead_));
55 : }
56 :
57 7 : void RtsqA5::MakeSureAvailableSpace()
58 : {
59 7 : u32 availableSpace = GetTailToHeadDist();
60 7 : auto startTime = std::chrono::steady_clock::now();
61 : #ifdef CCL_KERNEL_AICPU
62 : sqFullTimeout_ = GetSqFullTimeOut();
63 : #endif
64 7 : const std::chrono::seconds printInterval(PRINT_INTERVAL); // 打印间隔30s
65 7 : auto lastPrintTime = std::chrono::steady_clock::now() - printInterval;
66 21 : HCCL_INFO(
67 : "[%s]sqId:%u, sqFullTimeout_: %u s, sqHead:%u, sqTail:%u, pendingSqeCnt:%u", __func__, sqId_, sqFullTimeout_,
68 : sqHead_, sqTail_, pendingSqeCnt);
69 :
70 7 : while (availableSpace <= pendingSqeCnt) {
71 0 : sqHead_ = QuerySqHead();
72 0 : availableSpace = GetTailToHeadDist();
73 0 : if (availableSpace > pendingSqeCnt) {
74 0 : break; // 避免head没更新导致假反压
75 : }
76 :
77 0 : auto curTime = std::chrono::steady_clock::now();
78 0 : if (UNLIKELY(curTime - lastPrintTime >= printInterval)) {
79 0 : HCCL_RUN_INFO(
80 : "[%s]while loop, sqId:%u, sqHead:%u, sqTail:%u, availableSpace:%u, pendingSqeCnt:%u, "
81 : "sqFullTimeout_:%u s",
82 : __func__, sqId_, sqHead_, sqTail_, availableSpace, pendingSqeCnt, sqFullTimeout_);
83 0 : lastPrintTime = curTime;
84 : }
85 :
86 0 : CheckLaunchTaskStatus(startTime, curTime);
87 : #ifdef CCL_KERNEL_AICPU
88 : HcclResult ret = HandleDispatchAllStreams();
89 : if (UNLIKELY(ret != HCCL_SUCCESS)) {
90 : auto msg
91 : = StringFormat("RtsqA5::%s HandleDispatchAllStreams failed, ret = %d, sqId:%u, ", __func__, ret, sqId_);
92 : HCCL_ERROR("%s", msg.c_str());
93 : THROW<InternalException>(msg);
94 : }
95 : #endif
96 0 : if (checkOpExecStatusCallback_ != nullptr) {
97 0 : checkOpExecStatusCallback_();
98 : }
99 : }
100 7 : }
101 :
102 5 : void RtsqA5::CheckLaunchTaskStatus(
103 : const std::chrono::steady_clock::time_point& startTime, const std::chrono::steady_clock::time_point& curTime)
104 : {
105 5 : bool isTimeout = (sqFullTimeout_ == 0) ? false : ((curTime - startTime) >= std::chrono::seconds(sqFullTimeout_));
106 : // step1 检测是否launch超时，如果超时打印rtsq full的ERROR日志
107 5 : if (UNLIKELY(isTimeout)) {
108 6 : HCCL_ERROR(
109 : "Rtsq full, sqFullTimeout_:[%u s]. sqId:[%u], sqHead:[%u], sqTail:[%u], pendingSqeCnt:[%u]", sqFullTimeout_,
110 : sqId_, sqHead_, sqTail_, pendingSqeCnt);
111 : }
112 :
113 5 : HcclResult checkRet = (checkExecStatusCallback_ != nullptr) ? checkExecStatusCallback_(isTimeout) : HCCL_SUCCESS;
114 : // step2 通信域状态为HCCL_COMM_STATUS_SUSPENDING状态，则终止launch不抛异
115 5 : if (UNLIKELY(checkRet == HCCL_E_SUSPENDING)) {
116 3 : pendingSqeCnt = 0;
117 3 : return;
118 : }
119 : // step3 调用回调检查执行状态：1、如果超时，打印taskException；2、如果通信域不可用，终止launch
120 2 : if (UNLIKELY(isTimeout || checkRet != HCCL_SUCCESS)) {
121 2 : THROW<InternalException>(
122 4 : StringFormat("[%s]stop launch Task, isTimeout[%d], checkRet[%d]", __func__, isTimeout, checkRet));
123 : }
124 : }
125 :
126 11 : void RtsqA5::CopySqeBufToSq(u8* sqeBuf)
127 : {
128 11 : u8* sqCurrAddr = reinterpret_cast<u8*>(sqBaseAddr_) + sqTail_ * RTSQ_SQE_SIZE;
129 11 : if (sqTail_ >= sqHead_) {
130 10 : u32 depthLeft = sqDepth_ - sqTail_;
131 10 : if (pendingSqeCnt <= depthLeft) { // 没有回绕
132 21 : HCCL_INFO(
133 : "RtsqA5::%s copy sqe from sqe buffer, sqId_: %u, streamId_: %u, cur head: %u, cur tail: %u, size: %u, "
134 : "depth remain: %u",
135 : __func__, sqId_, streamId_, sqHead_, sqTail_, pendingSqeCnt, depthLeft);
136 7 : int ret = memcpy_sp(sqCurrAddr, pendingSqeCnt * AC_SQE_SIZE, sqeBuf, pendingSqeCnt * RTSQ_SQE_SIZE);
137 7 : if (UNLIKELY(ret != 0)) {
138 4 : THROW<InternalException>(StringFormat("RtsqA5::%s sqe memcpy_sp failed, ret = %d", __func__, ret));
139 : }
140 : } else {
141 9 : HCCL_INFO(
142 : "RtsqA5::%s copy sqe twice, sqId_: %u, streamId_: %u, cur head: %u, cur tail: %u, cnt: %u, depth "
143 : "remain: %u",
144 : __func__, sqId_, streamId_, sqHead_, sqTail_, pendingSqeCnt, depthLeft);
145 : // 先拷贝rtsq里剩余空间大小
146 3 : int ret = memcpy_sp(sqCurrAddr, depthLeft * AC_SQE_SIZE, sqeBuf, depthLeft * RTSQ_SQE_SIZE);
147 3 : if (ret != 0) {
148 1 : THROW<InternalException>(
149 3 : StringFormat("RtsqA5::%s rtsq remaining space memcpy_sp failed, ret = %d", __func__, ret));
150 : }
151 : // 拷贝剩余sqe
152 2 : ret = memcpy_sp(
153 : reinterpret_cast<u8*>(sqBaseAddr_), sqHead_ * RTSQ_SQE_SIZE, sqeBuf + depthLeft * RTSQ_SQE_SIZE,
154 : (pendingSqeCnt - depthLeft) * AC_SQE_SIZE);
155 2 : if (UNLIKELY(ret != 0)) {
156 0 : THROW<InternalException>(
157 0 : StringFormat("RtsqA5::%s remaining sqe memcpy_sp failed, ret = %d", __func__, ret));
158 : }
159 : }
160 : } else {
161 3 : HCCL_INFO(
162 : "RtsqA5::%s copy sqe from sqe buffer, tail < head, sqId_: %u, streamId_: %u, cur head: %u, cur tail: %u, "
163 : "size: %u",
164 : __func__, sqId_, streamId_, sqHead_, sqTail_, pendingSqeCnt);
165 1 : int ret = memcpy_sp(sqCurrAddr, pendingSqeCnt * AC_SQE_SIZE, sqeBuf, pendingSqeCnt * RTSQ_SQE_SIZE);
166 1 : if (UNLIKELY(ret != 0)) {
167 0 : THROW<InternalException>(StringFormat("RtsqA5::%s sqe memcpy_sp failed, ret = %d", __func__, ret));
168 : }
169 : }
170 8 : }
171 :
172 4 : void RtsqA5::PreLaunchSqeForCache(bool& needCacheTask)
173 : {
174 : // 校验needCacheTaskCallback_
175 : // 注意: A5新流程下needCacheTaskCallback_一定非空; 但A5老流程下不支持aicpu task cache, needCacheTaskCallback_为空;
176 : // 为避免A5老流程报错, 这里为空时跳过执行而非报错
177 4 : needCacheTask = false;
178 4 : if (UNLIKELY(needCacheTaskCallback_ == nullptr)) {
179 12 : HCCL_WARNING("[RtsqA5][PreLaunchSqeForCache] needCacheTaskCallback_ is null, keep needCacheTask as false");
180 : } else {
181 0 : needCacheTask = needCacheTaskCallback_();
182 : }
183 4 : }
184 :
185 0 : void RtsqA5::PostLaunchSqeForCache()
186 : {
187 : // 注意: 只有needCacheTask为true时才调用PostLaunchSqeForCache, 此时一定是A5新流程, 因此addSqeArrayCallback_一定非空
188 0 : if (UNLIKELY(aicpuTsThreadPtr_ == nullptr)) {
189 0 : THROW<InternalException>("[RtsqA5][PostLaunchSqeForCache] aicpuTsThreadPtr_ is null");
190 : }
191 0 : if (UNLIKELY(addSqeArrayCallback_ == nullptr)) {
192 0 : THROW<InternalException>("[RtsqA5][PostLaunchSqeForCache] addSqeArrayCallback_ is null");
193 : }
194 0 : HcclResult ret = addSqeArrayCallback_(this, aicpuTsThreadPtr_, pendingSqeCnt, locBuf, streamId_);
195 0 : if (UNLIKELY(ret != HCCL_SUCCESS)) {
196 0 : THROW<InternalException>("[RtsqA5][PostLaunchSqeForCache] addSqeArrayCallback_ failed, ret %d", ret);
197 : }
198 0 : }
199 :
200 : // 向芯片RTSQ VA中写入 SQE，并触发芯片执行
201 5 : void RtsqA5::LaunchTask()
202 : {
203 15 : HCCL_INFO("RtsqA5::%s: START, pendingSqeCnt[%u]", __func__, pendingSqeCnt);
204 5 : if (pendingSqeCnt == 0) { // 没有SQE ，直接返回
205 6 : HCCL_INFO("RtsqA5::%s: pendingSqeCnt is %u, return", __func__, pendingSqeCnt);
206 2 : return;
207 : }
208 : // 确保 rtsq 有足够空间放pending SQE
209 3 : MakeSureAvailableSpace();
210 :
211 3 : if (pendingSqeCnt == 0) {
212 0 : return;
213 : }
214 :
215 3 : bool needCacheTask = false;
216 3 : PreLaunchSqeForCache(needCacheTask);
217 : // localBuffer拷贝到 RTSQ
218 3 : CopySqeBufToSq(locBuf);
219 :
220 : // 正常展开按需打印SQE
221 3 : if ((UNLIKELY(GetPlfDebugConfigValue() & PLF_TASK)) || UNLIKELY(HcclCheckLogLevel(HCCL_LOG_DEBUG))) {
222 7 : PLF_CONFIG_DEBUG(
223 : PLF_TASK, "[RtsqA5][LaunchTask] dump %llu generated SQEs in stream[%u]", pendingSqeCnt, streamId_);
224 :
225 3 : int ret = HCCL_SUCCESS;
226 3 : uint8_t* sqePtr = locBuf;
227 9 : for (size_t sqeIdx = 0; sqeIdx < pendingSqeCnt; sqeIdx++) {
228 12 : PLF_CONFIG_DEBUG(PLF_TASK, "[RtsqA5][LaunchTask] %uth generated SQE in stream[%u]", sqeIdx, streamId_);
229 6 : ret = hcomm::AicpuTaskUtils::DumpSqeContent(sqePtr);
230 6 : if (UNLIKELY(ret != HCCL_SUCCESS)) {
231 0 : THROW<InternalException>(StringFormat("RtsqA5::%s DumpSqeContent failed, ret = %d", __func__, ret));
232 : }
233 :
234 6 : sqePtr += RTSQ_SQE_SIZE;
235 : }
236 : }
237 :
238 : // 更新tail，触发芯片执行
239 3 : u32 newTail = (sqTail_ + pendingSqeCnt) % sqDepth_;
240 3 : ConfigSqTail(newTail);
241 3 : sqTail_ = newTail;
242 :
243 : // 缓存sqe
244 3 : if (needCacheTask) {
245 0 : PostLaunchSqeForCache();
246 : }
247 : // 清空本地的locBuffer和sqeCnt数目
248 9 : HCCL_INFO(
249 : "RtsqA5::%s: END, pendingSqeCnt[%u], streamId_[%u] sqHead_[%u] sqTail_[%u]", __func__, pendingSqeCnt, streamId_,
250 : sqHead_, sqTail_);
251 3 : pendingSqeCnt = 0;
252 3 : (void)memset_s(locBuf, RTSQ_SQE_SIZE * PER_LAUNCH_SQE_CNT, 0, RTSQ_SQE_SIZE * PER_LAUNCH_SQE_CNT); // locBuffer清零
253 : }
254 :
255 16 : void RtsqA5::RefreshSqeHeaderTaskField(Rt91095StarsSqeHeader* sqeHeaderPtr)
256 : {
257 16 : SetSqeHeaderTaskFields(sqeHeaderPtr, taskId_);
258 16 : SetTaskIdBySqeId();
259 16 : }
260 :
261 : // 向芯片RTSQ VA中写入aicpu task cache SQE，并触发芯片执行
262 5 : void RtsqA5::LaunchNewTask(uint8_t* sqeArray, uint32_t sqeCount)
263 : {
264 : // 注意: cache命中时才会调用LaunchNewTask, 此时一定不存在pending SQE
265 5 : if (UNLIKELY(pendingSqeCnt > 0)) {
266 2 : THROW<InternalException>(StringFormat(
267 : "RtsqA5::%s: pendingSqeCnt[%u] should be 0 when aicpu task cache hits!", __func__, pendingSqeCnt));
268 : }
269 :
270 : // 临时设置pendingSqeCnt, 用于MakeSureAvailableSpace
271 4 : pendingSqeCnt = sqeCount;
272 :
273 : // 确保 rtsq 有足够空间放pending SQE
274 4 : MakeSureAvailableSpace();
275 :
276 : // sqeArray拷贝到 RTSQ
277 4 : CopySqeBufToSq(sqeArray);
278 :
279 : // 更新tail，触发芯片执行
280 4 : u32 newTail = (sqTail_ + pendingSqeCnt) % sqDepth_;
281 4 : ConfigSqTail(newTail);
282 4 : sqTail_ = newTail;
283 :
284 12 : HCCL_INFO(
285 : "RtsqA5::%s: END, pendingSqeCnt[%u], streamId_[%u] sqHead_[%u] sqTail_[%u]", __func__, pendingSqeCnt, streamId_,
286 : sqHead_, sqTail_);
287 :
288 : // 重置pendingSqeCnt
289 4 : pendingSqeCnt = 0;
290 4 : }
291 :
292 2 : void RtsqA5::TryLaunchTask()
293 : {
294 2 : if (pendingSqeCnt == 0) {
295 1 : return;
296 : }
297 :
298 1 : sqHead_ = QuerySqHead();
299 1 : u32 availableSpace = GetTailToHeadDist();
300 1 : if (availableSpace <= pendingSqeCnt) {
301 0 : return;
302 : }
303 :
304 1 : bool needCacheTask = false;
305 1 : PreLaunchSqeForCache(needCacheTask);
306 :
307 1 : CopySqeBufToSq(locBuf);
308 :
309 1 : u32 newTail = (sqTail_ + pendingSqeCnt) % sqDepth_;
310 1 : ConfigSqTail(newTail);
311 1 : sqTail_ = newTail;
312 :
313 : // 缓存sqe
314 1 : if (needCacheTask) {
315 0 : PostLaunchSqeForCache();
316 : }
317 1 : pendingSqeCnt = 0;
318 1 : (void)memset_s(locBuf, RTSQ_SQE_SIZE * PER_LAUNCH_SQE_CNT, 0, RTSQ_SQE_SIZE * PER_LAUNCH_SQE_CNT);
319 3 : HCCL_INFO(
320 : "RtsqA5::%s: END, pendingSqeCnt[%u], streamId_[%u] sqHead_[%u] sqTail_[%u]", __func__, pendingSqeCnt, streamId_,
321 : sqHead_, sqTail_);
322 : }
323 :
324 36 : u8* RtsqA5::GetCurrSqeBuffer()
325 : {
326 36 : lastSqeAddr_ = sqBaseAddr_ + static_cast<u64>((sqTail_ + pendingSqeCnt) % sqDepth_) * RTSQ_SQE_SIZE;
327 36 : return locBuf + pendingSqeCnt * RTSQ_SQE_SIZE;
328 : }
329 :
330 21 : u64 RtsqA5::GetSqeAddr() const { return lastSqeAddr_; }
331 :
332 33 : void RtsqA5::RefreshInfo()
333 : {
334 33 : SetTaskIdBySqeId();
335 33 : pendingSqeCnt++;
336 :
337 : #ifdef CCL_KERNEL_AICPU
338 : if (launchFlag_ && !IsBatchLaunchMode()) {
339 : LaunchTask();
340 : return;
341 : }
342 : #endif
343 :
344 33 : if (pendingSqeCnt != PER_LAUNCH_SQE_CNT) {
345 33 : return;
346 : }
347 : // 挂起的sqe数量为128个，则需要向芯片RTSQ中写入task
348 0 : LaunchTask();
349 : }
350 :
351 2 : void RtsqA5::NotifyWait(u32 notifyId) { NotifyWait(notifyId, GetKernelExecTimeoutFromEnvConfig()); }
352 :
353 11 : void RtsqA5::NotifyWait(u32 notifyId, u32 timeout)
354 : {
355 11 : BuildA5SqeNotifyWait(streamId_, taskId_, notifyId, timeout, GetCurrSqeBuffer());
356 33 : HCCL_INFO(
357 : "RtsqA5::NotifyWait: streamId %u, taskId %u, notifyId %u, timeout[%u ms]", streamId_, taskId_, notifyId,
358 : timeout);
359 11 : RefreshInfo();
360 11 : }
361 :
362 1 : void RtsqA5::NotifyRecordLoc(u32 notifyId)
363 : {
364 1 : BuildA5SqeNotifyRecord(streamId_, taskId_, notifyId, GetCurrSqeBuffer());
365 3 : HCCL_INFO("RtsqA5::NotifyRecordLoc: streamId %u, taskId %u, notifyId %u", streamId_, taskId_, notifyId);
366 1 : RefreshInfo();
367 1 : }
368 :
369 1 : void RtsqA5::Cnt1toNNotifyWait(u32 notifyId, u32 value)
370 : {
371 1 : BuildA5SqeCnt1toNNotifyWait(streamId_, taskId_, notifyId, value, GetCurrSqeBuffer());
372 3 : HCCL_INFO("RtsqA5::Cnt1toNNotifyWait: streamId %u, taskId %u, notifyId %u", streamId_, taskId_, notifyId);
373 1 : RefreshInfo();
374 1 : }
375 :
376 1 : void RtsqA5::Cnt1toNNotifyRecord(u32 notifyId, u32 value)
377 : {
378 1 : BuildA5SqeCnt1toNNotifyRecord(streamId_, taskId_, notifyId, value, GetCurrSqeBuffer());
379 3 : HCCL_INFO("RtsqA5::Cnt1toNNotifyRecord: streamId %u, taskId %u, notifyId %u", streamId_, taskId_, notifyId);
380 1 : RefreshInfo();
381 1 : }
382 :
383 1 : void RtsqA5::CntNto1NotifyWait(u32 notifyId, u32 value)
384 : {
385 1 : BuildA5SqeCntNto1NotifyWait(streamId_, taskId_, notifyId, value, GetCurrSqeBuffer());
386 3 : HCCL_INFO("RtsqA5::CntNto1NotifyWait: streamId %u, taskId %u, notifyId %u", streamId_, taskId_, notifyId);
387 1 : RefreshInfo();
388 1 : }
389 :
390 1 : void RtsqA5::CntNto1NotifyRecord(u32 notifyId, u32 value)
391 : {
392 1 : BuildA5SqeCntNto1NotifyRecord(streamId_, taskId_, notifyId, value, GetCurrSqeBuffer());
393 3 : HCCL_INFO("RtsqA5::CntNto1NotifyRecord: streamId %u, taskId %u, notifyId %u", streamId_, taskId_, notifyId);
394 1 : RefreshInfo();
395 1 : }
396 :
397 2 : void RtsqA5::SdmaCopy(u64 srcAddr, u64 dstAddr, u32 size, u32 partId)
398 : {
399 : // 不带reduce的拷贝，opcode填0
400 : (void)partId;
401 2 : BuildA5SqeSdmaCopy(streamId_, taskId_, dstAddr, srcAddr, size, RTSQ_A5_PART_ID, 0, GetCurrSqeBuffer());
402 6 : HCCL_INFO(
403 : "RtsqA5::SdmaCopy: streamId %u, taskId %u, srcAddr 0x%llx, dstAddr 0x%llx, size %u", streamId_, taskId_,
404 : srcAddr, dstAddr, size);
405 2 : RefreshInfo();
406 2 : }
407 :
408 : const std::unordered_map<ReduceOp, RtStarsMemcpyAsyncOperationKind, EnumClassHash> ReduceOpToStarsOpKindMap
409 : = {{ReduceOp::SUM, RtStarsMemcpyAsyncOperationKind::RT_STARS_MEMCPY_ASYNC_OP_KIND_ADD},
410 : {ReduceOp::MAX, RtStarsMemcpyAsyncOperationKind::RT_STARS_MEMCPY_ASYNC_OP_KIND_MAX},
411 : {ReduceOp::MIN, RtStarsMemcpyAsyncOperationKind::RT_STARS_MEMCPY_ASYNC_OP_KIND_MIN},
412 : {ReduceOp::EQUAL, RtStarsMemcpyAsyncOperationKind::RT_STARS_MEMCPY_ASYNC_OP_KIND_EQUAL}};
413 :
414 : const std::unordered_map<DataType, RtStarsMemcpyAsyncDataType, EnumClassHash> DataTypeToStarsDataTypeMap
415 : = {{DataType::INT8, RtStarsMemcpyAsyncDataType::RT_STARS_MEMCPY_ASYNC_DATA_TYPE_INT8},
416 : {DataType::INT16, RtStarsMemcpyAsyncDataType::RT_STARS_MEMCPY_ASYNC_DATA_TYPE_INT16},
417 : {DataType::INT32, RtStarsMemcpyAsyncDataType::RT_STARS_MEMCPY_ASYNC_DATA_TYPE_INT32},
418 : {DataType::FP16, RtStarsMemcpyAsyncDataType::RT_STARS_MEMCPY_ASYNC_DATA_TYPE_FP16},
419 : {DataType::FP32, RtStarsMemcpyAsyncDataType::RT_STARS_MEMCPY_ASYNC_DATA_TYPE_FP32},
420 : {DataType::BFP16, RtStarsMemcpyAsyncDataType::RT_STARS_MEMCPY_ASYNC_DATA_TYPE_BFP16}};
421 :
422 3 : void RtsqA5::SdmaReduce(u64 srcAddr, u64 dstAddr, u32 size, u32 partId, const ReduceIn& reduceIn)
423 : {
424 : (void)partId;
425 3 : if (UNLIKELY(
426 : ReduceOpToStarsOpKindMap.find(reduceIn.reduceOp) == ReduceOpToStarsOpKindMap.end()
427 : || DataTypeToStarsDataTypeMap.find(reduceIn.dataType) == DataTypeToStarsDataTypeMap.end())) {
428 3 : THROW<InternalException>(StringFormat(
429 3 : "Sdma does not support reduceOp %s dataType %s", reduceIn.reduceOp.Describe().c_str(),
430 3 : reduceIn.dataType.Describe().c_str()));
431 : }
432 :
433 2 : u8 op = static_cast<u8>(ReduceOpToStarsOpKindMap.at(reduceIn.reduceOp));
434 2 : u8 type = static_cast<u8>(DataTypeToStarsDataTypeMap.at(reduceIn.dataType));
435 :
436 2 : BuildA5SqeSdmaCopy(streamId_, taskId_, dstAddr, srcAddr, size, RTSQ_A5_PART_ID, (op | type), GetCurrSqeBuffer());
437 6 : HCCL_INFO(
438 : "RtsqA5::SdmaReduce: streamId %u, taskId %u, srcAddr 0x%llx, dstAddr 0x%llx, size %u", streamId_, taskId_,
439 : srcAddr, dstAddr, size);
440 2 : RefreshInfo();
441 2 : }
442 :
443 2 : bool RtsqA5::IsRtsqQueueSpaceSufficient()
444 : {
445 : // 判断逻辑与rtsq内部保持一致，rtsq剩余空间需要大于（rtsq挂起的任务数量+本次任务）
446 2 : u32 availableSpace = GetTailToHeadDist();
447 2 : if (availableSpace > pendingSqeCnt + 1) {
448 1 : return true;
449 : }
450 :
451 : // 否则的话，需要再次查询一次head，确认是否是因为head没有更新导致空间不足，如果查询后空间仍然不足，则返回false
452 1 : sqHead_ = QuerySqHead();
453 1 : availableSpace = GetTailToHeadDist();
454 :
455 1 : return (availableSpace > pendingSqeCnt + 1);
456 : }
457 :
458 3 : HcclResult RtsqA5::SetPreStreamSyncReady()
459 : {
460 3 : isPreStreamSync = true;
461 3 : return HCCL_SUCCESS;
462 : }
463 :
464 4 : HcclResult RtsqA5::SetPreStreamSyncFin()
465 : {
466 4 : isPreStreamSync = false;
467 4 : return HCCL_SUCCESS;
468 : }
469 :
470 89 : bool RtsqA5::GetPreStreamSyncStatus() { return isPreStreamSync; }
471 :
472 2 : void RtsqA5::UbDbSend(const UbJettyLiteId& jettyLiteId, u16 piValue)
473 : {
474 : // piValue需要使用u16数据类型，保证自然增长，用于判断是否翻转
475 2 : BuildA5SqeUbDbSend(streamId_, taskId_, jettyLiteId, piValue, GetCurrSqeBuffer());
476 6 : HCCL_INFO(
477 : "RtsqA5::UbDbSend: streamId %u, taskId %u, piValue(UbPi):%u, SqTail(Rtsq Pi):%u", streamId_, taskId_, piValue,
478 : sqTail_);
479 2 : RefreshInfo();
480 2 : }
481 :
482 6 : void RtsqA5::RdmaDbSend(const uint64_t& dbAddr, const uint64_t& dbValue)
483 : {
484 6 : BuildA5SqeRdmaDbSend(streamId_, taskId_, dbAddr, dbValue, GetCurrSqeBuffer());
485 18 : HCCL_INFO(
486 : "RtsqA5::RdmaDbSend: RdmaDbSend streamId %u, taskId %u, Sqe: %s, dbAddr:0x%llx, dbValue:0x%llx, SqTail(Rtsq "
487 : "Pi):%u",
488 : streamId_, taskId_, Bytes2hex(GetCurrSqeBuffer(), RTSQ_SQE_SIZE).c_str(), dbAddr, dbValue, sqTail_);
489 6 : RefreshInfo();
490 6 : }
491 :
492 1 : void RtsqA5::CCoreNotifyWait(u64 waitAddr, u64 curTurnCntAddr, bool last)
493 : {
494 1 : BuildA5SqeCCoreNotifyWait(streamId_, taskId_, waitAddr, curTurnCntAddr, last, GetCurrSqeBuffer());
495 3 : HCCL_INFO(
496 : "RtsqA5::CCoreNotifyWait: streamId %u, taskId %u, waitAddr %llu, curTurnCntAddr %llu, last %d", streamId_,
497 : taskId_, waitAddr, curTurnCntAddr, last);
498 1 : RefreshInfo();
499 1 : }
500 :
501 1 : void RtsqA5::CCoreNotifyRecord(u64 recordAddr, u64 curTurnCntAddr)
502 : {
503 1 : BuildA5SqeCCoreNotifyRecord(streamId_, taskId_, recordAddr, curTurnCntAddr, GetCurrSqeBuffer());
504 3 : HCCL_INFO(
505 : "RtsqA5::CCoreNotifyRecord: streamId %u, taskId %u, recordAddr %llu, curTurnCntAddr %llu", streamId_, taskId_,
506 : recordAddr, curTurnCntAddr);
507 1 : RefreshInfo();
508 1 : }
509 :
510 0 : void RtsqA5::P2PWriteValue(u64 remoteAddr, u32 writeValue)
511 : {
512 0 : BuildA5SqeP2pWriteValue(streamId_, taskId_, remoteAddr, writeValue, GetCurrSqeBuffer());
513 0 : HCCL_INFO(
514 : "RtsqA5::P2PWriteValue: streamId %u, taskId %u, remoteAddr %llu, writeValue %u", streamId_, taskId_, remoteAddr,
515 : writeValue);
516 0 : RefreshInfo();
517 0 : }
518 : } // namespace Hccl
|
