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 <memory>
12 : #include "hccl_common.h"
13 : #include "stream_pub.h"
14 : #include "aicpu/aicpu_hccl_sqcqv1.h"
15 : #include "aicpu/aicpu_hccl_sqcqv2.h"
16 : #include "config_plf_log.h"
17 : #include "dlhal_function.h"
18 : #include "adapter_hal_pub.h"
19 : #include "dispatcher_aicpu.h"
20 :
21 : namespace hccl {
22 : constexpr uint64_t NANOSECOND_TO_SECOND = 1000000000U;
23 : constexpr uint16_t TURN_LEFT_SHIFT_BIT = 16;
24 : constexpr uint32_t HCCL_PER_LAUNCH_SQE_CNT = 128U; // 每编排N个SQE，做一次launchtask
25 :
26 : #ifdef __cplusplus
27 : extern "C" {
28 : #endif // __cplusplus
29 11 : HcclResult HcclDispatcherAicpuInit(HcclDispatcher *dispatcher, const u32 devPhyId, uint32_t hcclQos, DispatcherType type)
30 : {
31 11 : CHK_PTR_NULL(dispatcher);
32 11 : DispatcherPub *pDispatcher = nullptr;
33 11 : if (type == DispatcherType::DISPATCHER_AICPU) {
34 11 : pDispatcher = new (std::nothrow) DispatcherAiCpu(devPhyId);
35 11 : pDispatcher->SetHcclQos(hcclQos);
36 11 : pDispatcher->SetMpamid(0);
37 11 : HCCL_INFO("HcclDispatcherAicpuInit hcclQos = %u", pDispatcher->GetHcclQos());
38 : } else {
39 0 : HCCL_ERROR("[HcclCommAicpu][HcclDispatcherInit] Not support the dispatcher type[%d]", type);
40 0 : return HCCL_E_NOT_SUPPORT;
41 : }
42 11 : CHK_PTR_NULL(pDispatcher);
43 11 : HcclResult ret = pDispatcher->Init();
44 11 : if (ret != HCCL_SUCCESS) {
45 0 : HCCL_ERROR("[HcclCommAicpu][HcclDispatcherInit] Dispatcher init failed, type[%d]", type);
46 0 : delete pDispatcher;
47 0 : pDispatcher = nullptr;
48 0 : return ret;
49 : }
50 11 : *dispatcher = pDispatcher;
51 11 : return HCCL_SUCCESS;
52 : }
53 : #ifdef __cplusplus
54 : }
55 : #endif // __cplusplus
56 :
57 445 : DispatcherAiCpu::DispatcherAiCpu(const u32 devPhyId) // deprecated
58 445 : : DispatcherPub(INVALID_INT)
59 : {
60 445 : aicpuInfo_.devId = devPhyId;
61 445 : }
62 :
63 890 : DispatcherAiCpu::~DispatcherAiCpu() {}
64 :
65 444 : HcclResult DispatcherAiCpu::Init()
66 : {
67 444 : CHK_RET(DlHalFunction::GetInstance().DlHalFunctionInit());
68 444 : CHK_RET(hrtHalGetDeviceType(aicpuInfo_.devId, aicpuInfo_.devType));
69 444 : CHK_RET(hrtHalGetDeviceInfo(aicpuInfo_.devId, MODULE_TYPE_SYSTEM, INFO_TYPE_PHY_CHIP_ID,
70 : reinterpret_cast<int64_t*>(&aicpuInfo_.chipId)));
71 :
72 443 : if (aicpuInfo_.devType == DevType::DEV_TYPE_310P1 || aicpuInfo_.devType == DevType::DEV_TYPE_310P3) {
73 0 : addOneNotifyWaitSqe_ = AddOneNotifyWaitSqeV2;
74 0 : addOneRecordSqe_ = AddOneRecordSqeV2;
75 0 : addOneWriteValueRecordSqe_ = AddOneWriteValueRecordSqeV2;
76 0 : addOneMemcpySqe_ = AddOneMemcpySqeV2;
77 0 : addOneEventResetSqe_ = AddOneEventResetSqeV2;
78 0 : addOneEventRecordSqe_ = AddOneEventRecordSqeV2;
79 0 : addOneEventWaitSqe_ = AddOneEventWaitSqeV2;
80 : } else {
81 443 : addOneNotifyWaitSqe_ = AddOneNotifyWaitSqeV1;
82 443 : addOneRecordSqe_ = AddOneRecordSqeV1;
83 443 : addOneWriteValueRecordSqe_ = AddOneWriteValueRecordSqeV1;
84 443 : addOneMemcpySqe_ = AddOneMemcpySqeV1;
85 443 : addOneEventResetSqe_ = AddOneEventResetSqeV1;
86 443 : addOneEventRecordSqe_ = AddOneEventRecordSqeV1;
87 443 : addOneEventWaitSqe_ = AddOneEventWaitSqeV1;
88 443 : addOneRdmaDbSendSqe_ = AddOneRdmaDbSendSqeV1;
89 443 : addOneFlipPlaceHolderSqe_ = AddOneFlipPlaceHolderSqeV1;
90 443 : addOneCacheMemcpyPlaceHolderSqe_ = AddOneCacheMemcpyPlaceHolderSqeV1;
91 443 : addOneCacheNotifyWaitPlaceholderSqe_ = AddOneCacheNotifyWaitPlaceholderSqeV1;
92 443 : addOneCacheNotifyRecordPlaceholderSqe_ = AddOneCacheNotifyRecordPlaceholderSqeV1;
93 443 : addOneCacheWriteValuePlaceholderSqe_ = AddOneCacheWriteValuePlaceholderSqeV1;
94 443 : addOneCacheMemcpyRecordPlaceholderSqe_ = AddOneCacheMemcpyRecordPlaceholderSqeV1;
95 443 : CHK_PTR_NULL(addOneRdmaDbSendSqe_);
96 443 : CHK_PTR_NULL(addOneFlipPlaceHolderSqe_);
97 443 : CHK_PTR_NULL(addOneCacheMemcpyPlaceHolderSqe_);
98 443 : CHK_PTR_NULL(addOneCacheNotifyWaitPlaceholderSqe_);
99 443 : CHK_PTR_NULL(addOneCacheNotifyRecordPlaceholderSqe_);
100 443 : CHK_PTR_NULL(addOneCacheWriteValuePlaceholderSqe_);
101 443 : CHK_PTR_NULL(addOneCacheMemcpyRecordPlaceholderSqe_);
102 : }
103 :
104 443 : CHK_PTR_NULL(addOneNotifyWaitSqe_);
105 443 : CHK_PTR_NULL(addOneRecordSqe_);
106 443 : CHK_PTR_NULL(addOneWriteValueRecordSqe_);
107 443 : CHK_PTR_NULL(addOneMemcpySqe_);
108 443 : CHK_PTR_NULL(addOneEventResetSqe_);
109 443 : CHK_PTR_NULL(addOneEventRecordSqe_);
110 443 : CHK_PTR_NULL(addOneEventWaitSqe_);
111 :
112 443 : CHK_RET(GetNotifyMaxWaitTime());
113 444 : notifySize_ = (aicpuInfo_.devType == DevType::DEV_TYPE_910B || aicpuInfo_.devType == DevType::DEV_TYPE_910_93) ?
114 : 4 : 8; // 和hrtGetNotifySize接口保持一致，910B和910_93的notify寄存器大小为4，其他芯片为8
115 444 : InitTimeOutConfig();
116 444 : HCCL_INFO("%s success, devId:%u, devType:%d, chipId:%lld",
117 : __func__, aicpuInfo_.devId, aicpuInfo_.devType, aicpuInfo_.chipId);
118 444 : return HCCL_SUCCESS;
119 : }
120 :
121 1 : HcclResult DispatcherAiCpu::WaitValue(hccl::Stream &stream, u64 waitAddr, u64 valueAddr, bool reset)
122 : {
123 1 : u32 turnNum= *(reinterpret_cast<u32*>(static_cast<uintptr_t>(valueAddr)));
124 1 : HCCL_DEBUG("[DispatcherAiCpu][WaitValue] turnNum %u", turnNum);
125 1 : uint8_t *sqeBuffer = nullptr;
126 1 : uint8_t *sqeTypeAddr = nullptr;
127 1 : uint8_t *sqeDfxInfoAddr = nullptr;
128 1 : uint16_t taskId = 0U;
129 1 : CHK_RET(stream.GetNextSqeBufferAddr(sqeBuffer, sqeTypeAddr, sqeDfxInfoAddr, taskId));
130 : const HcclComStreamInfo *streamInfo;
131 1 : CHK_RET(stream.GetStreamInfo(streamInfo));
132 1 : AddOneWaitStartSqe(streamInfo->actualStreamId, taskId, waitAddr, valueAddr,
133 : reset, reinterpret_cast<rtStarsCcoreWaitStartSqe_t *>(sqeBuffer), sqeTypeAddr);
134 1 : HcclSqeContext *sqeCtx = stream.GetSqeContextPtr();
135 1 : if (sqeCtx == nullptr) {
136 0 : HCCL_ERROR("[DispatcherAiCpu][WaitValue] AddCcoreWait sqeCtx is nullptr");
137 0 : return HCCL_E_INTERNAL;
138 : }
139 1 : sqeCtx->buffer.addInfo[taskId % hccl::HCCL_SQE_MAX_CNT] =
140 1 : ((turnNum << TURN_LEFT_SHIFT_BIT) + static_cast<uint32_t>(reset));
141 1 : return HCCL_SUCCESS;
142 : }
143 :
144 1 : HcclResult DispatcherAiCpu::WriteValue(hccl::Stream &stream, u64 writeAddr, u64 valueAddr)
145 : {
146 1 : u32 turnNum= *(reinterpret_cast<u32*>(static_cast<uintptr_t>(valueAddr)));
147 1 : HCCL_DEBUG("[DispatcherAiCpu][WriteValue] turnNum %u", turnNum);
148 1 : uint8_t *sqeBuffer = nullptr;
149 1 : uint8_t *sqeTypeAddr = nullptr;
150 1 : uint8_t *sqeDfxInfoAddr = nullptr;
151 1 : uint16_t taskId = 0U;
152 1 : CHK_RET(stream.GetNextSqeBufferAddr(sqeBuffer, sqeTypeAddr, sqeDfxInfoAddr, taskId));
153 : const HcclComStreamInfo *streamInfo;
154 1 : CHK_RET(stream.GetStreamInfo(streamInfo));
155 1 : AddOneWriteValueStartSqe(streamInfo->actualStreamId, taskId, writeAddr, valueAddr, reinterpret_cast<rtStarsCcoreWriteValueSqe_t *>(sqeBuffer),
156 : sqeTypeAddr);
157 1 : HcclSqeContext *sqeCtx = stream.GetSqeContextPtr();
158 1 : if (sqeCtx == nullptr) {
159 0 : HCCL_ERROR("[DispatcherAiCpu][WriteValue] AddCcoreNotify sqeCtx is nullptr");
160 0 : return HCCL_E_INTERNAL;
161 : }
162 1 : sqeCtx->buffer.addInfo[taskId % hccl::HCCL_SQE_MAX_CNT] = turnNum;
163 1 : return HCCL_SUCCESS;
164 : }
165 :
166 2 : HcclResult DispatcherAiCpu::SignalRecord(HcclRtNotify signal, hccl::Stream &stream, u32 userRank, u64 offset, s32 stage,
167 : bool inchip, u64 signalAddr, u32 notifyId)
168 : {
169 2 : const HcclComStreamInfo &streamInfo = stream.GetHcclStreamInfo();
170 2 : uint8_t *sqeBuffer = nullptr;
171 2 : uint8_t *sqeTypeAddr = nullptr;
172 2 : uint8_t *sqeDfxInfoAddr = nullptr;
173 2 : uint16_t taskId = 0U;
174 2 : CHK_RET(GetStreamSqeBufferAddr(stream, sqeBuffer, sqeTypeAddr, sqeDfxInfoAddr, taskId));
175 2 : AicpuDfxInfo * const dfxInfo = (AicpuDfxInfo * const)sqeDfxInfoAddr;
176 2 : dfxInfo->opRingBufferIdx = opRingBufferIdx_;
177 2 : dfxInfo->remoteRank = userRank;
178 2 : dfxInfo->notifyId = notifyId;
179 :
180 : // 检查是否需要生成cache-write placeholder
181 2 : if (isPlaceholder_) {
182 : // 只有使能alltoallv类算子的aicpu cache才需要cache-write placeholder
183 0 : CHK_PRT_RET(!(needAddSqe_ && isAlltoallv_), HCCL_ERROR("[DispatcherAiCpu][SignalRecord] isPlaceholder_[%u]"
184 : "needAddSqe_[%u] isAlltoallv_[%u]", isPlaceholder_, needAddSqe_, isAlltoallv_), HCCL_E_INTERNAL);
185 :
186 : // 只有p2p的SignalRecord才需要cache-write placeholder
187 : // 不需要为inchip场景下LocalNotify::Post的SignalRecord调用addOneCacheNotifyRecordPlaceholderSqe_生成cache-notify placeholder
188 0 : CHK_PRT_RET(inchip, HCCL_ERROR("[DispatcherAiCpu][SignalRecord] isPlaceholder_[%u], inchip[%u]",
189 : isPlaceholder_, inchip), HCCL_E_INTERNAL);
190 :
191 0 : addOneCacheWriteValuePlaceholderSqe_(streamInfo.actualStreamId, taskId, signalAddr, sqeBuffer, sqeTypeAddr);
192 0 : HCCL_INFO("%s generate cache-write placeholder for signal record", __func__);
193 : } else { // 正常生成SQE
194 2 : if (inchip) {
195 1 : addOneRecordSqe_(streamInfo.actualStreamId, taskId, notifyId, sqeBuffer, sqeTypeAddr);
196 : } else {
197 1 : addOneWriteValueRecordSqe_(streamInfo.actualStreamId, taskId, signalAddr, sqeBuffer, sqeTypeAddr);
198 : }
199 : }
200 :
201 2 : PLF_CONFIG_INFO(PLF_TASK,
202 : "%s para: streamId[%d] remoteRank[%u] inchip[%d] devType[%d] notifyId[%u]",
203 : __func__, streamInfo.actualStreamId, userRank, inchip, aicpuInfo_.devType, notifyId);
204 2 : return HCCL_SUCCESS;
205 : }
206 :
207 1 : HcclResult DispatcherAiCpu::SignalRecord(hccl::DeviceMem &dst, hccl::DeviceMem &src, hccl::Stream &stream,
208 : u32 remoteUserRank, hccl::LinkType inLinkType, u32 notifyId)
209 : {
210 : // 参数有效性检查
211 1 : CHK_PRT_RET(src.size() == 0, HCCL_INFO("%s src size is 0, not need copy", __func__), HCCL_SUCCESS);
212 1 : CHK_PRT_RET(src == dst, HCCL_INFO("%s src and dst is same, not need copy", __func__), HCCL_SUCCESS);
213 1 : CHK_PRT_RET(dst.size() < src.size(),
214 : HCCL_ERROR("%s The size of dst is smaller than that of src. dst addr:%p, dst size:%llu, src addr:%p, "
215 : "src size:%llu", __func__, dst.ptr(), dst.size(), src.ptr(), src.size()),
216 : HCCL_E_PARA);
217 1 : CHK_PRT_RET(src.size() != notifySize_,
218 : HCCL_ERROR("%s src size[%llu] should be %llu in devType[%d]",
219 : __func__, src.size(), notifySize_, aicpuInfo_.devType), HCCL_E_PARA);
220 :
221 1 : const HcclComStreamInfo &streamInfo = stream.GetHcclStreamInfo();
222 1 : uint8_t *sqeBuffer = nullptr;
223 1 : uint8_t *sqeTypeAddr = nullptr;
224 1 : uint8_t *sqeDfxInfoAddr = nullptr;
225 1 : uint16_t taskId = 0U;
226 1 : aclrtReduceKind rtReduceOp = RK_MAP_TABLE[HCCL_REDUCE_RESERVED];
227 1 : uint8_t linkType = static_cast<uint8_t>(inLinkType);
228 :
229 1 : CHK_RET(GetStreamSqeBufferAddr(stream, sqeBuffer, sqeTypeAddr, sqeDfxInfoAddr, taskId));
230 1 : AicpuDfxInfo * const dfxInfo = reinterpret_cast<AicpuDfxInfo * const>(sqeDfxInfoAddr);
231 1 : dfxInfo->opRingBufferIdx = opRingBufferIdx_;
232 1 : dfxInfo->remoteRank = remoteUserRank;
233 1 : dfxInfo->notifyId = notifyId;
234 :
235 : // 检查是否需要生成cache-memcpy-record placeholder
236 1 : if (isPlaceholder_) {
237 : // 只有使能alltoallv类算子的aicpu cache才需要cache-memcpy-record placeholder
238 0 : CHK_PRT_RET(!(needAddSqe_ && isAlltoallv_), HCCL_ERROR("[DispatcherAiCpu][SignalRecord] isPlaceholder_[%u]"
239 : "needAddSqe_[%u] isAlltoallv_[%u]", isPlaceholder_, needAddSqe_, isAlltoallv_), HCCL_E_INTERNAL);
240 :
241 0 : addOneCacheMemcpyRecordPlaceholderSqe_(streamInfo.actualStreamId, taskId, src.ptr(), src.size() , ACL_FLOAT, rtReduceOp,
242 0 : dst.ptr(), 0, aicpuInfo_.ssid, aicpuInfo_.devId, aicpuInfo_.overflowAddr, linkType, sqeBuffer, sqeTypeAddr, SDMA_QOS_DEFAULT);
243 0 : HCCL_INFO("%s generate cache-memcpy-record placeholder for signal record", __func__);
244 : } else { // 正常生成memcpy record SQE
245 1 : addOneMemcpySqe_(streamInfo.actualStreamId, taskId, src.ptr(), src.size() , ACL_FLOAT, rtReduceOp,
246 1 : dst.ptr(), 0, aicpuInfo_.ssid, aicpuInfo_.devId, aicpuInfo_.overflowAddr, linkType, sqeBuffer, sqeTypeAddr, SDMA_QOS_DEFAULT);
247 : }
248 :
249 1 : PLF_CONFIG_INFO(PLF_TASK,
250 : "%s para: linkType[%u] srcPtr[%p] srcSize[%llu] dstPtr[%p] taskId[%u] streamId[%u] remoteRank[%u] notifyId[%u] hcclQos[%u]",
251 : __func__, linkType, src.ptr(), src.size(), dst.ptr(), taskId, streamInfo.actualStreamId, remoteUserRank, notifyId, SDMA_QOS_DEFAULT);
252 1 : return HCCL_SUCCESS;
253 : }
254 :
255 3 : HcclResult DispatcherAiCpu::SignalWait(HcclRtNotify signal, Stream &stream, u32 userRank, u32 remoteUserRank, s32 stage,
256 : bool inchip, u32 notifyId, u32 timeOut)
257 : {
258 3 : const HcclComStreamInfo &streamInfo = stream.GetHcclStreamInfo();
259 3 : uint8_t *sqeBuffer = nullptr;
260 3 : uint8_t *sqeTypeAddr = nullptr;
261 3 : uint8_t *sqeDfxInfoAddr = nullptr;
262 3 : uint16_t taskId = 0U;
263 3 : CHK_RET(GetStreamSqeBufferAddr(stream, sqeBuffer, sqeTypeAddr, sqeDfxInfoAddr, taskId));
264 3 : AicpuDfxInfo * const dfxInfo = (AicpuDfxInfo * const)sqeDfxInfoAddr;
265 3 : dfxInfo->opRingBufferIdx = opRingBufferIdx_;
266 3 : dfxInfo->remoteRank = remoteUserRank;
267 3 : dfxInfo->notifyId = notifyId;
268 :
269 3 : dfxTimeOutConfig_.sqeTimeOutTimeOut = timeOut < notifyMaxWaitTime_ ? timeOut : dfxTimeOutConfig_.sqeTimeOutTimeOut;
270 : // 检查是否需要生成cache-notify placeholder
271 3 : if (isPlaceholder_) {
272 : // 只有使能alltoallv类算子的aicpu cache才需要cache-notify placeholder
273 0 : CHK_PRT_RET(!(needAddSqe_ && isAlltoallv_), HCCL_ERROR("[DispatcherAiCpu][SignalWait] isPlaceholder_[%u]"
274 : "needAddSqe_[%u] isAlltoallv_[%u]", isPlaceholder_, needAddSqe_, isAlltoallv_), HCCL_E_INTERNAL);
275 :
276 : // 只有A3下才会使能alltoallv aicpu cache并需要cache-notify placeholder
277 0 : CHK_PRT_RET(!(inchip || (aicpuInfo_.devType != DevType::DEV_TYPE_310P1 && aicpuInfo_.devType != DevType::DEV_TYPE_310P3)),
278 : HCCL_ERROR("[DispatcherAiCpu][SignalWait] isPlaceholder_[%u] inchip[%u] devType[%u]",
279 : isPlaceholder_, inchip, aicpuInfo_.devType),
280 : HCCL_E_INTERNAL);
281 :
282 0 : addOneCacheNotifyWaitPlaceholderSqe_(streamInfo.actualStreamId, taskId, notifyId, sqeBuffer, sqeTypeAddr, dfxTimeOutConfig_);
283 0 : HCCL_INFO("%s generate cache-notify placeholder for signal wait", __func__);
284 : } else { // 正常生成SQE
285 3 : if (inchip || (aicpuInfo_.devType != DevType::DEV_TYPE_310P1 && aicpuInfo_.devType != DevType::DEV_TYPE_310P3)) {
286 3 : addOneNotifyWaitSqe_(streamInfo.actualStreamId, taskId, notifyId, sqeBuffer, sqeTypeAddr, dfxTimeOutConfig_);
287 : } else {
288 0 : u32 notifyRevisedOffset = 15U; // eventid偏移15位后为1
289 0 : u32 notifyGetEventId = 0x3FFU; // 取低15位
290 0 : if ((notifyId >> notifyRevisedOffset) != 0) {
291 0 : addOneEventWaitSqe_(streamInfo.actualStreamId,
292 0 : (notifyId & notifyGetEventId), taskId, sqeBuffer, sqeTypeAddr);
293 :
294 0 : uint8_t *sqeBuffer1 = nullptr;
295 0 : uint8_t *sqeTypeAddr1 = nullptr;
296 0 : uint8_t *sqeDfxInfoAddr1 = nullptr;
297 0 : CHK_RET(GetStreamSqeBufferAddr(stream, sqeBuffer1, sqeTypeAddr1, sqeDfxInfoAddr1, taskId));
298 0 : AicpuDfxInfo * const dfxInfo = (AicpuDfxInfo * const)sqeDfxInfoAddr1;
299 0 : dfxInfo->opRingBufferIdx = opRingBufferIdx_;
300 0 : dfxInfo->remoteRank = INVALID_VALUE_RANKID;
301 :
302 0 : u64 addr = 0;
303 0 : addOneEventResetSqe_(streamInfo.actualStreamId,
304 0 : (notifyId & notifyGetEventId), taskId, aicpuInfo_.devId, 0,
305 : addr, sqeBuffer1, sqeTypeAddr1);
306 : } else {
307 0 : HCCL_WARNING("%s SignalWait id is not event, please check %d", __func__, notifyId);
308 : }
309 : }
310 : }
311 :
312 3 : PLF_CONFIG_INFO(PLF_TASK,
313 : "%s para: streamId[%u] userRank[%u] remoteRank[%u] inchip[%d] devType[%d] notifyId[%u] sqeTimeOutTimeOut[%llu]",
314 : __func__, streamInfo.actualStreamId, userRank, remoteUserRank, inchip, aicpuInfo_.devType, notifyId,
315 : dfxTimeOutConfig_.sqeTimeOutTimeOut);
316 3 : return HCCL_SUCCESS;
317 : }
318 :
319 3 : HcclResult DispatcherAiCpu::MemcpyAsync(hccl::DeviceMem &dst, const hccl::DeviceMem &src, hccl::Stream &stream,
320 : u32 remoteUserRank, hccl::LinkType inLinkType)
321 : {
322 : // 检查是否需要生成cache-memcpy placeholder
323 3 : if (isPlaceholder_) {
324 : // 只有零长拷贝才需要cache-memcpy placeholder
325 0 : CHK_PRT_RET(src.size() != 0, HCCL_ERROR("[DispatcherAiCpu][MemcpyAsync] isPlaceholder_[%u] src.size[%llu]",
326 : isPlaceholder_, src.size()), HCCL_E_INTERNAL);
327 :
328 : // 只有使能alltoallv类算子的aicpu cache才需要cache-memcpy placeholder
329 0 : CHK_PRT_RET(!(needAddSqe_ && isAlltoallv_), HCCL_ERROR("[DispatcherAiCpu][MemcpyAsync] isPlaceholder_[%u]"
330 : "needAddSqe_[%u] isAlltoallv_[%u]", isPlaceholder_, needAddSqe_, isAlltoallv_), HCCL_E_INTERNAL);
331 :
332 : // 准备SQE, sqeType, dfxInfo
333 0 : uint8_t *sqeBuffer = nullptr;
334 0 : uint8_t *sqeTypeAddr = nullptr;
335 0 : uint8_t *sqeDfxInfoAddr = nullptr;
336 0 : uint16_t taskId = 0U;
337 0 : CHK_RET(GetStreamSqeBufferAddr(stream, sqeBuffer, sqeTypeAddr, sqeDfxInfoAddr, taskId));
338 :
339 : // 生成cache-memcpy placeholder SQE, 并设置sqeType
340 0 : CHK_PTR_NULL(addOneCacheMemcpyPlaceHolderSqe_);
341 0 : const HcclComStreamInfo &streamInfo = stream.GetHcclStreamInfo();
342 0 : addOneCacheMemcpyPlaceHolderSqe_(streamInfo.actualStreamId, taskId, src.ptr(), dst.ptr(),
343 : static_cast<uint8_t>(inLinkType), sqeBuffer, sqeTypeAddr, hcclQos_);
344 :
345 : // 设置dfxInfo
346 0 : AicpuDfxInfo * const dfxInfo = (AicpuDfxInfo * const)sqeDfxInfoAddr;
347 0 : dfxInfo->opRingBufferIdx = opRingBufferIdx_;
348 0 : dfxInfo->remoteRank = remoteUserRank;
349 0 : dfxInfo->notifyId = INVALID_VALUE_RANKID;
350 :
351 0 : HCCL_INFO("%s capture zero-len memcpy for alltoallv, generate cache-memcpy placeholder sqe", __func__);
352 :
353 0 : return HCCL_SUCCESS;
354 : }
355 :
356 : // 参数有效性检查
357 3 : if (src.size() == 0) {
358 0 : HCCL_INFO("%s src memory size is 0, not need copy.", __func__);
359 0 : return HCCL_SUCCESS;
360 : }
361 :
362 3 : if (src == dst) {
363 3 : HCCL_INFO("%s src memory and dst memory is same, not need copy.", __func__);
364 3 : return HCCL_SUCCESS;
365 : }
366 :
367 0 : if (dst.size() < src.size()) {
368 0 : HCCL_ERROR("%s The size of dst is smaller than that of src. dst addr[%p], dst size[%llu], "\
369 : "src addr[%p], src size[%llu]", __func__, dst.ptr(), dst.size(), src.ptr(), src.size());
370 0 : return HCCL_E_PTR;
371 : }
372 0 : const HcclComStreamInfo &streamInfo = stream.GetHcclStreamInfo();
373 :
374 : // 将数据按4GB切分循环处理
375 0 : uint64_t spiltLoop = 0;
376 0 : uint64_t addrOffset = 0;
377 0 : uint64_t countSplit = 0;
378 0 : uint64_t countSize = src.size();
379 0 : uint8_t *sqeBuffer = nullptr;
380 0 : uint8_t *sqeTypeAddr = nullptr;
381 0 : uint8_t *sqeDfxInfoAddr = nullptr;
382 0 : uint16_t taskId = 0U;
383 0 : HcclReduceOp redOp = HCCL_REDUCE_RESERVED;
384 0 : aclrtReduceKind rtReduceOp = RK_MAP_TABLE[redOp];
385 :
386 0 : if (countSize > HCCL_SDMA_MAX_COUNT_4GB) {
387 0 : spiltLoop = (countSize % HCCL_SDMA_MAX_COUNT_4GB) ? (countSize / HCCL_SDMA_MAX_COUNT_4GB) :
388 0 : ((countSize / HCCL_SDMA_MAX_COUNT_4GB) - 1);
389 0 : HCCL_INFO("%s MemcpyAsync SDMA task countSize is bigger than 4GB"
390 : " and do segmentation splitloop[%llu]", __func__, spiltLoop);
391 : }
392 0 : uint8_t linkType = static_cast<uint8_t>(inLinkType);
393 0 : for (uint64_t index = 0; index <= spiltLoop; index++) {
394 0 : addrOffset = index * HCCL_SDMA_MAX_COUNT_4GB;
395 0 : countSplit = (index == spiltLoop) ? (countSize - index * HCCL_SDMA_MAX_COUNT_4GB) : (HCCL_SDMA_MAX_COUNT_4GB);
396 0 : void *srcSplit = static_cast<void *>(static_cast<char *>(const_cast<void *>(src.ptr())) + addrOffset);
397 0 : void *dstSplit = static_cast<void *>(static_cast<char *>(dst.ptr()) + addrOffset);
398 :
399 0 : CHK_RET(GetStreamSqeBufferAddr(stream, sqeBuffer, sqeTypeAddr, sqeDfxInfoAddr, taskId));
400 0 : AicpuDfxInfo * const dfxInfo = (AicpuDfxInfo * const)sqeDfxInfoAddr;
401 0 : dfxInfo->opRingBufferIdx = opRingBufferIdx_;
402 0 : dfxInfo->remoteRank = remoteUserRank;
403 0 : dfxInfo->notifyId = INVALID_VALUE_RANKID;
404 0 : addOneMemcpySqe_(streamInfo.actualStreamId, taskId, srcSplit, countSplit , ACL_FLOAT, rtReduceOp,
405 : dstSplit, 0, aicpuInfo_.ssid, aicpuInfo_.devId, aicpuInfo_.overflowAddr, linkType, sqeBuffer, sqeTypeAddr, hcclQos_);
406 :
407 0 : PLF_CONFIG_INFO(PLF_TASK,
408 : "%s para: linkType[%u] srcSplit[%p] dstSplit[%p] countSplit[%llu] taskId[%u] streamId[%u] remoteRank[%u]",
409 : __func__, linkType, srcSplit, dstSplit, countSplit , taskId, streamInfo.actualStreamId, remoteUserRank);
410 : }
411 0 : return HCCL_SUCCESS;
412 : }
413 :
414 0 : HcclResult DispatcherAiCpu::ClearLaunchContext()
415 : {
416 0 : HCCL_INFO("[DispatcherAiCpu][ClearLaunchContext] clear launch context");
417 :
418 0 : key_ = OpUnfoldKey();
419 0 : cachePtr_ = nullptr;
420 0 : userInputMemRanges_.clear();
421 0 : userOutputMemRanges_.clear();
422 0 : isAlltoallv_ = false;
423 0 : alltoallvMetadataPtr_ = nullptr;
424 0 : needAddSqe_ = false;
425 :
426 0 : return HCCL_SUCCESS;
427 : }
428 :
429 0 : HcclResult DispatcherAiCpu::SetLaunchContext(const OpUnfoldKey& key, OpUnfoldCache *cachePtr, const std::vector<OpUnfoldMemRange>& userInputMemRanges, const std::vector<OpUnfoldMemRange>& userOutputMemRanges, const bool isAlltoallv, const AlltoallvMetadata* alltoallvMetadataPtr)
430 : {
431 0 : CHK_PTR_NULL(cachePtr);
432 :
433 0 : HCCL_INFO("[DispatcherAiCpu][SetLaunchContext] set launch context for key %s", key.GetKeyString().c_str());
434 :
435 0 : if (isAlltoallv) {
436 0 : CHK_PTR_NULL(alltoallvMetadataPtr);
437 0 : CHK_RET(alltoallvMetadataPtr->Check(false));
438 : }
439 :
440 0 : key_ = key;
441 0 : cachePtr_ = cachePtr;
442 0 : userInputMemRanges_ = userInputMemRanges;
443 0 : userOutputMemRanges_ = userOutputMemRanges;
444 0 : isAlltoallv_ = isAlltoallv;
445 0 : alltoallvMetadataPtr_ = alltoallvMetadataPtr;
446 0 : needAddSqe_ = true;
447 :
448 0 : return HCCL_SUCCESS;
449 : }
450 :
451 0 : HcclResult DispatcherAiCpu::LaunchNewTask(OpUnfoldCacheEntry *entryPtr, const std::vector<OpUnfoldMemRange>& userInputMemRanges, const std::vector<OpUnfoldMemRange>& userOutputMemRanges, Stream& mainStream, std::vector<Stream> &slaveStreams, const bool profL1Enable, const bool isAlltoallv, const AlltoallvMetadata& alltoallvMetadata, const AlltoallvSendRecvInfo& alltoallvSendRecvInfo)
452 : {
453 : // 校验入参
454 0 : CHK_PTR_NULL(entryPtr);
455 0 : if (isAlltoallv) { // 注意: LaunchNewTask是在缓存命中时调用, 此时无launch context, 所以不能直接通过key_.opType来判断是否为alltoallv算子, 需要框架侧传入
456 0 : CHK_RET(alltoallvMetadata.Check(true));
457 0 : CHK_RET(alltoallvSendRecvInfo.Check());
458 : }
459 :
460 : // 准备SQE刷新需要的变量
461 0 : size_t sqeCount = 0;
462 0 : uint8_t *sqeArray = nullptr;
463 0 : uint8_t *sqeTypeArray = nullptr;
464 0 : AicpuDfxInfo *sqeDfxInfoArray = nullptr;
465 0 : Stream *streamPtr = nullptr;
466 0 : std::vector<FlipInfo> flipInfos; // taskid==0且flipnum!=0的SQE索引, 即它们前面需要添加placeholder
467 0 : std::vector<uint64_t> profTimestamps; // 只有当profiling L1 enable时, 才需要记录各SQE的刷新时间
468 :
469 : // 准备placeholder需要的变量
470 : uint8_t placeholderSqe[HCCL_SQE_SIZE]; // placeholder SQE
471 : uint8_t placeholderSqeType; // placeholder SQE type
472 0 : AicpuDfxInfo placeholderSqeDfxInfo; // placeholder DfxInfo
473 0 : placeholderSqeDfxInfo.opRingBufferIdx = opRingBufferIdx_;
474 0 : placeholderSqeDfxInfo.remoteRank = INVALID_VALUE_RANKID;
475 0 : placeholderSqeDfxInfo.notifyId = INVALID_VALUE_RANKID;
476 0 : constexpr uint16_t flipPlaceholderTaskId = 0; // FlipPlaceholder的taskId一定为0
477 :
478 : // 下发多段SQE数组，SQE刷新与下发异步执行
479 0 : size_t sqeArrayCount = 0;
480 0 : CHK_RET(entryPtr->GetSqeArrayCount(sqeArrayCount));
481 0 : HCCL_INFO("[DispatcherAiCpu][LaunchNewTask] launch new task for sqeArrayCount[%u] in the cache entry at 0x%016llx", sqeArrayCount, entryPtr);
482 0 : for (size_t arrayIdx = 0; arrayIdx < sqeArrayCount; ++arrayIdx) {
483 : // 刷新并获得对应信息 (之前下发到RTSQ的SQE正在异步被消费)
484 0 : CHK_RET(entryPtr->UpdateAndGetSqeArray(arrayIdx, userInputMemRanges, userOutputMemRanges, mainStream, slaveStreams,
485 : opRingBufferIdx_, sqeCount, &sqeArray, &sqeTypeArray, &sqeDfxInfoArray, &streamPtr, flipInfos,
486 : profL1Enable, profTimestamps, isAlltoallv, alltoallvMetadata, alltoallvSendRecvInfo));
487 :
488 : // Profiling timestamp的个数应该等于缓存的SQE个数 + flip placeholder个数
489 0 : CHK_PRT_RET((!profL1Enable) && (profTimestamps.size() != 0),
490 : HCCL_ERROR("[DispatcherAiCpu][LaunchNewTask] profL1Enable[%u] profTimestamps.size[%u]",
491 : profL1Enable, profTimestamps.size()),
492 : HCCL_E_INTERNAL);
493 0 : CHK_PRT_RET(profL1Enable && (profTimestamps.size() != (sqeCount + flipInfos.size())),
494 : HCCL_ERROR("[DispatcherAiCpu][LaunchNewTask] profL1Enable[%u] profTimestamps.size[%u] sqeCount[%u] flipInfos.size[%u]",
495 : profL1Enable, profTimestamps.size(), sqeCount, flipInfos.size()),
496 : HCCL_E_INTERNAL);
497 :
498 : // 打印缓存并下发的SQE内容for debug
499 : // 设置HCCL_DEBUG_CONFIG="task", 或者设置ASCEND_GLOBAL_LOG_LEVEL=0
500 0 : int32_t streamId = streamPtr->GetHcclStreamInfo().actualStreamId;
501 0 : if ((UNLIKELY(GetExternalInputDebugConfig() & PLF_TASK)) || UNLIKELY(HcclCheckLogLevel(HCCL_LOG_DEBUG))) {
502 0 : PLF_CONFIG_DEBUG(PLF_TASK,
503 : "[DispatcherAicpu][LaunchNewTask] dump content of %uth cached SQE array with %u cached SQEs and stream id %u",
504 : arrayIdx, sqeCount, streamId);
505 0 : for (size_t sqeIdx = 0; sqeIdx < sqeCount; ++sqeIdx) {
506 0 : uint8_t *sqePtr = sqeArray + sqeIdx * HCCL_SQE_SIZE;
507 0 : const uint8_t sqeType = sqeTypeArray[sqeIdx];
508 0 : PLF_CONFIG_DEBUG(PLF_TASK, "[DispatcherAicpu][LaunchNewTask] %uth cached SQE", sqeIdx);
509 0 : CHK_RET(OpUnfoldCache::DumpSqeContent(sqePtr, sqeType));
510 :
511 0 : const AicpuDfxInfo& dfxinfo = sqeDfxInfoArray[sqeIdx];
512 0 : PLF_CONFIG_DEBUG(PLF_TASK, "[DispatcherAicpu][LaunchNewTask] AicpuDfxInfo: remoteRank[%u] opRingBufferIdx[%u] notifyId[%u]",
513 : dfxinfo.remoteRank, dfxinfo.opRingBufferIdx, dfxinfo.notifyId);
514 : }
515 : }
516 :
517 0 : HCCL_INFO("[DispatcherAiCpu][LaunchNewTask] arrayIdx[%u] flipInfos.size[%u]", arrayIdx, flipInfos.size());
518 :
519 : // 分段下发
520 0 : size_t sqeStartIdx = 0; // 要拷贝的SQE在sqeArray中的起始索引
521 0 : size_t profTimestampStartIdx = 0; // 要拷贝的profiling timestamp在profTimestamps中的起始索引
522 0 : for (size_t i = 0; i < flipInfos.size(); ++i) {
523 : // Copy [sqeStartIdx, curZeroTaskidSqeIdx) + placeholder into RTSQ
524 0 : const size_t curZeroTaskidSqeIdx = flipInfos[i].first;
525 0 : const size_t curSqeCount = curZeroTaskidSqeIdx - sqeStartIdx + 1;
526 0 : CHK_PRT_RET(curZeroTaskidSqeIdx < sqeStartIdx,
527 : HCCL_ERROR("[DispatcherAiCpu][LaunchNewTask] curZeroTaskidSqeIdx[%u] < sqeStartIdx[%u]",
528 : curZeroTaskidSqeIdx, sqeStartIdx),
529 : HCCL_E_INTERNAL);
530 :
531 : // Wait RTSQ for curSqeCount SQE (including placeholder) space
532 0 : HCCL_INFO("[DispatcherAiCpu][LaunchNewTask] wait rtsq for %u sqe space", curSqeCount);
533 0 : CHK_RET(WaitRtsq(*streamPtr, curSqeCount, true));
534 :
535 : // 下发sqeArray[sqeStartIdx, curZeroTaskidSqeIdx)到RTSQ中 (excluding placeholder)
536 0 : if (curZeroTaskidSqeIdx > sqeStartIdx) { // 需要下发的cached SQE数量 > 0
537 0 : HCCL_INFO("[DispatcherAiCpu][LaunchNewTask] launch %uth sqeArray[%u:%u)",
538 : arrayIdx, sqeStartIdx, curZeroTaskidSqeIdx);
539 0 : CHK_RET(MemcpyRtsq(*streamPtr, curSqeCount - 1,
540 : sqeArray + sqeStartIdx * HCCL_SQE_SIZE,
541 : sqeTypeArray + sqeStartIdx,
542 : sqeDfxInfoArray + sqeStartIdx,
543 : profL1Enable, profTimestamps, profTimestampStartIdx));
544 0 : if (profL1Enable) {
545 0 : profTimestampStartIdx += (curSqeCount - 1); // Cached SQEs
546 : }
547 : }
548 :
549 : // 根据具体SQE下发信息更新placeholder
550 : // 参考AddFlipTask, 设置placeholder SQE (streamId和stream相关, flipNum和SQE下发相关)
551 : // 注意: 由于flipPlaceholder DfxInfo在当前算子下不变, 提前设置, 后续只需要刷新placeholder即可
552 0 : const uint16_t curFlipNum = flipInfos[i].second;
553 0 : CHK_PRT_RET(curFlipNum == 0,
554 : HCCL_ERROR("[DispatcherAiCpu][LaunchNewTask] invalid flipNum[%u]: flipInfoIdx[%u] arrayIdx[%u] zeroTaskidSqeIdx[%u] streamId[%u]",
555 : curFlipNum, i, arrayIdx, curZeroTaskidSqeIdx, streamId),
556 : HCCL_E_INTERNAL);
557 0 : CHK_PTR_NULL(addOneFlipPlaceHolderSqe_);
558 0 : addOneFlipPlaceHolderSqe_(streamId, curFlipNum, flipPlaceholderTaskId, placeholderSqe, &placeholderSqeType);
559 0 : HCCL_INFO("[DispatcherAiCpu][LaunchNewTask] flip placeholder SQE with flipnum[%u] and streamid[%u]",
560 : curFlipNum, streamId);
561 :
562 : // 下发placeholder SQE
563 0 : HCCL_INFO("[DispatcherAiCpu][LaunchNewTask] launch placeholder SQE after %uth sqeArray[%u:%u)",
564 : arrayIdx, sqeStartIdx, curZeroTaskidSqeIdx);
565 0 : CHK_RET(MemcpyRtsq(*streamPtr, 1, placeholderSqe, &placeholderSqeType, &placeholderSqeDfxInfo,
566 : profL1Enable, profTimestamps, profTimestampStartIdx));
567 0 : if (profL1Enable) {
568 0 : profTimestampStartIdx += 1; // Flip placeholder
569 : }
570 :
571 0 : sqeStartIdx = curZeroTaskidSqeIdx;
572 : }
573 :
574 : // 按需下发剩余SQE
575 0 : if (sqeStartIdx < sqeCount) {
576 : // Copy [sqeStartIdx, sqeCount - 1] into RTSQ
577 0 : const size_t remainSqeCount = sqeCount - sqeStartIdx;
578 :
579 : // Wait RTSQ for remainSqeCount SQE space
580 0 : HCCL_INFO("[DispatcherAiCpu][LaunchNewTask] wait rtsq for %u sqe space", remainSqeCount);
581 0 : CHK_RET(WaitRtsq(*streamPtr, remainSqeCount, true));
582 :
583 : // 下发sqeArray[sqeStartIdx, sqeCount - 1]到RTSQ中
584 0 : HCCL_INFO("[DispatcherAiCpu][LaunchNewTask] launch %uth sqeArray[%u:%u]", arrayIdx, sqeStartIdx, sqeCount - 1);
585 0 : CHK_RET(MemcpyRtsq(*streamPtr, remainSqeCount,
586 : sqeArray + sqeStartIdx * HCCL_SQE_SIZE,
587 : sqeTypeArray + sqeStartIdx,
588 : sqeDfxInfoArray + sqeStartIdx,
589 : profL1Enable, profTimestamps, profTimestampStartIdx));
590 0 : if (profL1Enable) {
591 0 : profTimestampStartIdx += remainSqeCount; // Remaining cached SQEs
592 : }
593 : }
594 :
595 : // 为下一段SQE数组的刷新清理变量
596 0 : sqeCount = 0;
597 0 : sqeArray = nullptr;
598 0 : sqeTypeArray = nullptr;
599 0 : sqeDfxInfoArray = nullptr;
600 0 : streamPtr = nullptr;
601 0 : flipInfos.clear();
602 : }
603 :
604 : // 下发完当前cache entry中所有SQE数组后, 更新input/output memory ranges, 与SQE中in-place update的addr-related fields保持一直
605 0 : CHK_RET(entryPtr->SetInputOutputMemRanges(userInputMemRanges, userOutputMemRanges));
606 :
607 0 : return HCCL_SUCCESS;
608 0 : }
609 :
610 11 : HcclResult DispatcherAiCpu::LaunchTask(Stream &stream, bool isBlockLaunch)
611 : {
612 11 : const HcclComStreamInfo &streamInfo = stream.GetHcclStreamInfo();
613 11 : HcclSqeContext *sqeContext = stream.GetSqeContextPtr();
614 11 : CHK_PTR_NULL(sqeContext);
615 11 : SqeRingBuffer *sqeContextBuffer = &(sqeContext->buffer);
616 11 : CHK_PTR_NULL(sqeContextBuffer);
617 11 : const auto cnt = sqeContextBuffer->sqeCnt;
618 11 : if (cnt == 0) {
619 6 : CHK_PRT_CONT(isBlockLaunch,
620 : HCCL_DEBUG("no sqe, streamId:%d, sqId:%u", streamInfo.actualStreamId, streamInfo.sqId));
621 6 : return HCCL_SUCCESS;
622 5 : } else if (cnt > streamInfo.sqDepth) {
623 1 : HCCL_ERROR("LaunchTask fail, cnt:%u should be less than sqDepth:%u", cnt, streamInfo.sqDepth);
624 1 : return HCCL_E_PTR;
625 : }
626 :
627 4 : auto &head = sqeContextBuffer->sqHead;
628 4 : auto &tail = sqeContextBuffer->sqTail;
629 4 : u32 newTail = (tail + cnt) % streamInfo.sqDepth;
630 : // 仅在阻塞下发场景打印，避免非阻塞场景调用时刷屏
631 4 : CHK_PRT_CONT(isBlockLaunch,
632 : HCCL_INFO("Before send sqid:%d cnt:%u head:%u curtail:%u newTail:%u",
633 : streamInfo.sqId, cnt, head, tail, newTail));
634 :
635 4 : u64 startUsec = GetCurAicpuTimestamp();
636 4 : u64 lastUsec = startUsec;
637 3166944 : while (((tail < head ? streamInfo.sqDepth : 0U) + tail - head + cnt >= streamInfo.sqDepth) && (tail != head)) { // 判断剩余sqe空间是否足够下发
638 : // 需要放在while循环进来后第一个执行
639 3166942 : CHK_RET(QuerySqStatusByType(aicpuInfo_.devId, streamInfo.sqId, DRV_SQCQ_PROP_SQ_HEAD, head));
640 :
641 : // 非阻塞下发场景，rtsq队列空间不足时直接返回
642 3166942 : if (isBlockLaunch == false) {
643 0 : return HCCL_SUCCESS;
644 : }
645 :
646 : // 当前流无法下发，把其他流都launch一遍，避免等待的其他流没有launch
647 4731095 : for (auto it = streamMap_.begin(); it != streamMap_.end(); ++it) {
648 1564153 : if (it->first != streamInfo.actualStreamId) {
649 1 : CHK_RET(LaunchTask(it->second, false));
650 : }
651 : }
652 3166942 : u64 curUsec = GetCurAicpuTimestamp();
653 3166942 : if (dfxTimeOutConfig_.sqFullWaitTimeOut != 0 &&
654 3166941 : (curUsec - startUsec > NANOSECOND_TO_SECOND * dfxTimeOutConfig_.sqFullWaitTimeOut)) {
655 2 : HCCL_ERROR("Rtsq full, timeout %lus. curhead:%u, sqId:%d", dfxTimeOutConfig_.sqFullWaitTimeOut, head,
656 : streamInfo.sqId);
657 2 : return HCCL_E_AGAIN;
658 : }
659 :
660 : // 等待下发阶段，每隔30s打印一次状态
661 3166940 : if (curUsec - lastUsec > NANOSECOND_TO_SECOND * dfx::kPrintSqInterval) {
662 0 : lastUsec = curUsec;
663 0 : HCCL_RUN_INFO("[LaunchTask][WaitLaunchWhileLoop]Current state. sqid:%d, head:%u, tail:%u, cnt:%u",
664 : streamInfo.sqId, head, tail, cnt);
665 : }
666 :
667 : // 下发过程中出现cqe异常
668 3166940 : if (checkOpExecStatusCallback_ != nullptr) {
669 0 : HcclResult opExecStatus = checkOpExecStatusCallback_();
670 0 : CHK_PRT_RET(opExecStatus != HCCL_SUCCESS,
671 : HCCL_ERROR("hccl aicpu stop launch for task exception or stop command, ret:%d", opExecStatus),
672 : opExecStatus);
673 : }
674 : }
675 :
676 2 : uint32_t left = streamInfo.sqDepth - tail; // sqeAddr 剩余空间
677 2 : const auto tailSqeIdx = sqeContextBuffer->tailSqeIdx;
678 2 : HCCL_INFO("cpy sqe, left:%u, tailSqeId:%u, cnt:%u, streamId:%u", left, tailSqeIdx, cnt, stream.id());
679 2 : if (cnt <= left) { // 剩余buffer放得下新增sqe
680 1 : CHK_SAFETY_FUNC_RET(memcpy_s(
681 : reinterpret_cast<uint8_t *>(streamInfo.sqBaseAddr) + tail * HCCL_SQE_SIZE,
682 : left * HCCL_SQE_SIZE,
683 : sqeContextBuffer->localBuff + (tailSqeIdx - cnt) * HCCL_SQE_SIZE,
684 : cnt * HCCL_SQE_SIZE));
685 :
686 1 : CHK_SAFETY_FUNC_RET(memcpy_s(sqeContextBuffer->rtsMirrorBuffer + tail * HCCL_SQE_SIZE,
687 : left * HCCL_SQE_SIZE,
688 : sqeContextBuffer->localBuff + (tailSqeIdx - cnt) * HCCL_SQE_SIZE,
689 : cnt * HCCL_SQE_SIZE));
690 :
691 1 : CHK_SAFETY_FUNC_RET(memcpy_s(sqeContextBuffer->rtsqSqeType + tail, left,
692 : sqeContextBuffer->sqeType + (tailSqeIdx - cnt), cnt));
693 1 : CHK_SAFETY_FUNC_RET(memcpy_s(sqeContextBuffer->rtsDfxInfo + tail, left * sizeof(AicpuDfxInfo),
694 : sqeContextBuffer->dfxInfo + (tailSqeIdx - cnt), cnt * sizeof(AicpuDfxInfo)));
695 : } else {
696 1 : CHK_SAFETY_FUNC_RET(memcpy_s(reinterpret_cast<uint8_t *>(streamInfo.sqBaseAddr) + tail * HCCL_SQE_SIZE,
697 : left * HCCL_SQE_SIZE,
698 : sqeContextBuffer->localBuff + (tailSqeIdx - cnt) * HCCL_SQE_SIZE,
699 : left * HCCL_SQE_SIZE));
700 :
701 1 : CHK_SAFETY_FUNC_RET(memcpy_s(reinterpret_cast<uint8_t *>(streamInfo.sqBaseAddr),
702 : streamInfo.sqDepth * HCCL_SQE_SIZE,
703 : sqeContextBuffer->localBuff + (tailSqeIdx - cnt + left) * HCCL_SQE_SIZE,
704 : (cnt - left) * HCCL_SQE_SIZE));
705 :
706 1 : CHK_SAFETY_FUNC_RET(memcpy_s(sqeContextBuffer->rtsMirrorBuffer + tail * HCCL_SQE_SIZE,
707 : left * HCCL_SQE_SIZE,
708 : sqeContextBuffer->localBuff + (tailSqeIdx - cnt) * HCCL_SQE_SIZE,
709 : left * HCCL_SQE_SIZE));
710 :
711 1 : CHK_SAFETY_FUNC_RET(memcpy_s(sqeContextBuffer->rtsMirrorBuffer,
712 : streamInfo.sqDepth * HCCL_SQE_SIZE,
713 : sqeContextBuffer->localBuff + (tailSqeIdx - cnt + left) * HCCL_SQE_SIZE,
714 : (cnt - left) * HCCL_SQE_SIZE));
715 :
716 1 : CHK_SAFETY_FUNC_RET(memcpy_s(sqeContextBuffer->rtsqSqeType + tail,
717 : left, sqeContextBuffer->sqeType + (tailSqeIdx - cnt), left));
718 1 : CHK_SAFETY_FUNC_RET(memcpy_s(sqeContextBuffer->rtsqSqeType + 0, streamInfo.sqDepth,
719 : sqeContextBuffer->sqeType + (tailSqeIdx - cnt + left), (cnt - left)));
720 1 : CHK_SAFETY_FUNC_RET(memcpy_s(sqeContextBuffer->rtsDfxInfo + tail,
721 : left * sizeof(AicpuDfxInfo), sqeContextBuffer->dfxInfo + (tailSqeIdx - cnt), left * sizeof(AicpuDfxInfo)));
722 1 : CHK_SAFETY_FUNC_RET(memcpy_s(sqeContextBuffer->rtsDfxInfo + 0, streamInfo.sqDepth * sizeof(AicpuDfxInfo),
723 : sqeContextBuffer->dfxInfo + (tailSqeIdx - cnt + left), (cnt - left) * sizeof(AicpuDfxInfo)));
724 : }
725 : // 打印算子展开下发的SQE内容for debug
726 : // 设置HCCL_DEBUG_CONFIG="task", 或者设置ASCEND_GLOBAL_LOG_LEVEL=0
727 2 : if ((UNLIKELY(GetExternalInputDebugConfig() & PLF_TASK)) || UNLIKELY(HcclCheckLogLevel(HCCL_LOG_DEBUG))) {
728 2 : const int32_t streamId = stream.GetHcclStreamInfo().actualStreamId;
729 2 : PLF_CONFIG_DEBUG(PLF_TASK, "[DispatcherAicpu][LaunchTask] dump content of %u dispatched SQEs with stream id %u", cnt, streamId);
730 :
731 2 : uint8_t *sqeArray = sqeContextBuffer->localBuff + (tailSqeIdx - cnt) * HCCL_SQE_SIZE;
732 2 : uint8_t *sqeTypeArray = sqeContextBuffer->sqeType + (tailSqeIdx - cnt);
733 2 : AicpuDfxInfo *sqeDfxInfoArray = sqeContextBuffer->dfxInfo + (tailSqeIdx - cnt);
734 14 : for (size_t sqeIdx = 0; sqeIdx < cnt; ++sqeIdx) {
735 12 : uint8_t *sqePtr = sqeArray + sqeIdx * HCCL_SQE_SIZE;
736 12 : const uint8_t sqeType = sqeTypeArray[sqeIdx];
737 12 : if (sqeType == SqeType::FLIP_PLACEHOLDER_SQE) {
738 0 : const rtStarsPlaceHolderSqe_t *placeholderSqePtr = reinterpret_cast<const rtStarsPlaceHolderSqe_t *>(sqeArray + sqeIdx * HCCL_SQE_SIZE);
739 0 : PLF_CONFIG_DEBUG(PLF_TASK, "[DispatcherAicpu][LaunchTask] %uth dispatched SQE (placeholder) header.type[%u] taskid[%u]", sqeIdx, placeholderSqePtr->header.type, placeholderSqePtr->header.taskId);
740 : } else {
741 12 : PLF_CONFIG_DEBUG(PLF_TASK, "[DispatcherAicpu][LaunchTask] %uth dispatched SQE", sqeIdx);
742 : }
743 :
744 12 : CHK_RET(OpUnfoldCache::DumpSqeContent(sqePtr, sqeType));
745 :
746 12 : const AicpuDfxInfo& dfxinfo = sqeDfxInfoArray[sqeIdx];
747 12 : PLF_CONFIG_DEBUG(PLF_TASK, "[DispatcherAicpu][LaunchTask] AicpuDfxInfo: remoteRank[%u] opRingBufferIdx[%u] notifyId[%u]",
748 : dfxinfo.remoteRank, dfxinfo.opRingBufferIdx, dfxinfo.notifyId);
749 : }
750 : }
751 :
752 : // 当前算子展开的SQE需要被动态缓存
753 2 : if (needAddSqe_) {
754 0 : CHK_PTR_NULL(cachePtr_);
755 :
756 : // 查找key对应的cache entry, 如果不存在 (即当前算子第一次LaunchTask), 创建新的cache entry
757 0 : OpUnfoldCacheEntry *entryPtr = nullptr;
758 0 : CHK_RET(cachePtr_->FindEntry(key_, &entryPtr));
759 0 : if (entryPtr == nullptr) {
760 0 : CHK_RET(cachePtr_->AddEntry(key_, userInputMemRanges_, userOutputMemRanges_, &entryPtr));
761 : }
762 0 : CHK_PTR_NULL(entryPtr);
763 :
764 : // 准备SQE相关信息的数组基地址
765 0 : uint8_t *sqeArray = sqeContextBuffer->localBuff + (tailSqeIdx - cnt) * HCCL_SQE_SIZE;
766 0 : uint8_t *sqeTypeArray = sqeContextBuffer->sqeType + (tailSqeIdx - cnt);
767 0 : AicpuDfxInfo *sqeDfxInfoArray = sqeContextBuffer->dfxInfo + (tailSqeIdx - cnt);
768 :
769 : // 遍历sqeType找到placeholder的位置
770 0 : std::vector<size_t> placeholderIdxes;
771 0 : uint8_t *curSqeTypePtr = sqeTypeArray;
772 0 : for (size_t sqeTypeIdx = 0; sqeTypeIdx < cnt; ++sqeTypeIdx) {
773 0 : if (*curSqeTypePtr == SqeType::FLIP_PLACEHOLDER_SQE) {
774 0 : placeholderIdxes.emplace_back(sqeTypeIdx);
775 : }
776 0 : ++curSqeTypePtr;
777 : }
778 :
779 : // 在动态缓存中分配实际需要的SQE数组
780 0 : const size_t cacheableSqeCount = cnt - placeholderIdxes.size();
781 0 : const int32_t streamId = stream.GetHcclStreamInfo().actualStreamId;
782 0 : size_t arrayIdx = 0;
783 0 : CHK_RET(entryPtr->AllocSqeArray(cacheableSqeCount, streamId, arrayIdx));
784 :
785 : // 分段拷贝SQE相关信息到cache entry中
786 0 : size_t cacheableSqeStartIdx = 0; // SQE start index (在动态缓存对应SQE数组中的索引)
787 0 : size_t bufferSqeStartIdx = 0; // SQE start index (在SQE ring buffer中的索引)
788 0 : for (size_t i = 0; i < placeholderIdxes.size(); ++i) {
789 : // [bufferSqeStartIdx, curPlaceholderIdx) -> [cacheableSqeStartIdx, cacheableSqeStartIdx + curPlaceholderIdx - bufferSqeStartIdx)
790 0 : const size_t curPlaceholderIdx = placeholderIdxes[i];
791 0 : HCCL_INFO("[DispatcherAicpu][LaunchTask] %uth placeholder copy dispatchedSqeArray[%u:%u) into cachedSqeArrays[%u][%u:%u)", i, bufferSqeStartIdx, curPlaceholderIdx, arrayIdx, cacheableSqeStartIdx, cacheableSqeStartIdx + curPlaceholderIdx - bufferSqeStartIdx);
792 0 : if (curPlaceholderIdx <= bufferSqeStartIdx) { // NO non-placeholder dispatched SQE to admit
793 : // NOTE: NO need to change cacheableSqeStartIdx
794 0 : bufferSqeStartIdx = curPlaceholderIdx + 1;
795 : } else {
796 0 : const size_t curSqeCount = curPlaceholderIdx - bufferSqeStartIdx;
797 0 : CHK_RET(entryPtr->MemcpySqeArray(arrayIdx, cacheableSqeStartIdx, curSqeCount,
798 : sqeArray + bufferSqeStartIdx * HCCL_SQE_SIZE,
799 : sqeTypeArray + bufferSqeStartIdx,
800 : sqeDfxInfoArray + bufferSqeStartIdx,
801 : isAlltoallv_, alltoallvMetadataPtr_
802 : ));
803 0 : cacheableSqeStartIdx += curSqeCount;
804 0 : bufferSqeStartIdx = curPlaceholderIdx + 1;
805 : }
806 : }
807 :
808 : // 存在剩余SQE, 即最后一个SQE不是placeholder
809 0 : if (LIKELY(bufferSqeStartIdx < cnt)) {
810 : // [bufferSqeStartIdx, cnt - 1] -> [cacheableSqeStartIdx, cacheableSqeStartIdx + cnt - bufferSqeStartIdx)
811 0 : const size_t curSqeCount = cnt - bufferSqeStartIdx;
812 0 : CHK_RET(entryPtr->MemcpySqeArray(arrayIdx, cacheableSqeStartIdx, curSqeCount,
813 : sqeArray + bufferSqeStartIdx * HCCL_SQE_SIZE,
814 : sqeTypeArray + bufferSqeStartIdx,
815 : sqeDfxInfoArray + bufferSqeStartIdx,
816 : isAlltoallv_, alltoallvMetadataPtr_
817 : ));
818 : }
819 0 : }
820 :
821 2 : CHK_RET(ConfigSqStatusByType(aicpuInfo_.devId, streamInfo.sqId, DRV_SQCQ_PROP_SQ_TAIL, newTail));
822 2 : tail = newTail;
823 2 : PLF_CONFIG_INFO(PLF_TASK,
824 : "%s success, sqid:%d, sqe_num:%u, curHead:%u, curtail:%u", __func__, streamInfo.sqId, cnt, head, tail);
825 2 : sqeContextBuffer->sqeCnt = 0;
826 2 : return HCCL_SUCCESS;
827 : }
828 :
829 8 : HcclResult DispatcherAiCpu::LaunchTasksEx(hccl::Stream &stream, std::vector<Stream> &subStreams)
830 : {
831 : /* 两阶段模式，主流待正式执行时再下 */
832 : /* 一阶段第一次，可以先下主流 */
833 8 : HcclResult ret = LaunchTask(stream, true);
834 8 : if (ret != HCCL_SUCCESS) {
835 0 : HCCL_ERROR("[DispatcherAiCpu][LaunchTasksEx] "\
836 : "launch task failed, sqid:%u, ret:%u", stream.sqId(), ret);
837 0 : return ret;
838 : }
839 :
840 8 : for (u32 index = 0; index < subStreams.size(); index++) {
841 0 : ret = LaunchTask(subStreams[index], true);
842 0 : if (ret != HCCL_SUCCESS) {
843 0 : HCCL_ERROR("[DispatcherAiCpu][LaunchTasksEx] "\
844 : "launch task failed, sqid:%u, ret:%u", subStreams[index].sqId(), ret);
845 0 : return ret;
846 : }
847 : }
848 :
849 8 : return HCCL_SUCCESS;
850 : }
851 :
852 3 : HcclResult DispatcherAiCpu::LaunchAllTasks()
853 : {
854 5 : for (auto it = streamMap_.begin(); it != streamMap_.end(); ++it) {
855 3 : HcclResult ret = LaunchTask(it->second, true);
856 3 : if (ret != HCCL_SUCCESS) {
857 1 : HCCL_ERROR("DispatcherAiCpu][LaunchAllTasks] launch task failed, sqid:%u, ret:%u", it->second.sqId(), ret);
858 1 : return ret;
859 : }
860 : }
861 2 : return HCCL_SUCCESS;
862 : }
863 :
864 1 : HcclResult DispatcherAiCpu::ReduceAsync(const void *src, void *dst, u64 dataCount, const HcclDataType datatype,
865 : HcclReduceOp redOp, Stream &stream, HcclReduceType reduceType)
866 : {
867 1 : return (reduceType == HcclReduceType::HCCL_INLINE_REDUCE) ?
868 1 : InlineReduceAsync(src, dataCount, datatype, redOp, stream, dst) :
869 1 : TbeReduceAsync(src, dst, dataCount, datatype, redOp, stream, dst);
870 : }
871 :
872 2 : HcclResult DispatcherAiCpu::InlineReduceAsync(const void *src, u64 dataCount, const HcclDataType datatype,
873 : HcclReduceOp redOp, hccl::Stream &stream, void *dst, u32 remoteUserRank, hccl::LinkType inLinkType)
874 : {
875 : // 参数有效性检查
876 2 : CHK_PTR_NULL(stream.ptr());
877 2 : if (dataCount == 0) {
878 0 : HCCL_INFO("%s src memory size is 0, not need inline reduce.", __func__);
879 0 : return HCCL_SUCCESS;
880 : }
881 2 : const HcclComStreamInfo &streamInfo = stream.GetHcclStreamInfo();
882 :
883 2 : aclDataType runtimeDataType = DT_MAP_TABLE[datatype];
884 2 : aclrtReduceKind rtReduceOp = RK_MAP_TABLE[redOp];
885 :
886 : // 将数据按4GB切分循环处理
887 2 : uint64_t spiltLoop = 0;
888 2 : uint64_t addr_offset = 0;
889 2 : uint64_t countSplit = 0;
890 2 : uint64_t countSize = dataCount * SIZE_TABLE[datatype];
891 2 : uint8_t *sqeBuffer = nullptr;
892 2 : uint8_t *sqeTypeAddr = nullptr;
893 2 : uint8_t *sqeDfxInfoAddr = nullptr;
894 2 : uint16_t taskId = 0U;
895 :
896 2 : if (countSize > HCCL_SDMA_MAX_COUNT_4GB) {
897 0 : spiltLoop = (countSize % HCCL_SDMA_MAX_COUNT_4GB) ? (countSize / HCCL_SDMA_MAX_COUNT_4GB) :
898 0 : ((countSize / HCCL_SDMA_MAX_COUNT_4GB) - 1);
899 0 : HCCL_INFO("%s InlineReduceAsync SDMA task countSize is bigger than 4GB"
900 : " and do segmentation splitloop:%llu", __func__, spiltLoop);
901 : }
902 2 : uint8_t linkType = static_cast<uint8_t>(inLinkType);
903 4 : for (uint64_t index = 0; index <= spiltLoop; index++) {
904 2 : addr_offset = index * HCCL_SDMA_MAX_COUNT_4GB;
905 2 : countSplit = (index == spiltLoop) ? (countSize - index * HCCL_SDMA_MAX_COUNT_4GB) : (HCCL_SDMA_MAX_COUNT_4GB);
906 2 : void *srcSplit = static_cast<void *>(static_cast<char *>(const_cast<void *>(src)) + addr_offset);
907 2 : void *dstSplit = static_cast<void *>(static_cast<char *>(dst) + addr_offset);
908 :
909 2 : CHK_RET(GetStreamSqeBufferAddr(stream, sqeBuffer, sqeTypeAddr, sqeDfxInfoAddr, taskId));
910 2 : AicpuDfxInfo * const dfxInfo = (AicpuDfxInfo * const)sqeDfxInfoAddr;
911 2 : dfxInfo->opRingBufferIdx = opRingBufferIdx_;
912 2 : dfxInfo->remoteRank = remoteUserRank;
913 2 : dfxInfo->notifyId = INVALID_VALUE_RANKID;
914 2 : addOneMemcpySqe_(streamInfo.actualStreamId, taskId, srcSplit, countSplit, runtimeDataType, rtReduceOp, dstSplit, 0,
915 : aicpuInfo_.ssid, aicpuInfo_.devId, aicpuInfo_.overflowAddr, linkType, sqeBuffer, sqeTypeAddr, hcclQos_);
916 :
917 2 : PLF_CONFIG_INFO(PLF_TASK,
918 : "%s para: linkType[%u] srcSplit[%p] dstSplit[%p] countSplit[%llu] taskId[%u] streamId[%u] remoteRank[%u] "\
919 : "rtDatatType[%d] rtReduceOp[%d]", __func__, linkType, srcSplit, dstSplit, countSplit, taskId,
920 : streamInfo.actualStreamId, remoteUserRank, runtimeDataType, rtReduceOp);
921 : }
922 :
923 2 : return HCCL_SUCCESS;
924 : }
925 :
926 1 : HcclResult DispatcherAiCpu::TbeReduceAsync(const void *src1, const void *src2, u64 count, const HcclDataType datatype,
927 : HcclReduceOp redOp, Stream &stream, const void *dst)
928 : {
929 1 : HCCL_ERROR("[DispatcherAiCpu][TbeReduceAsync] aicpu do not support the tbe reduce");
930 1 : return HCCL_E_NOT_SUPPORT;
931 : }
932 :
933 5 : HcclResult DispatcherAiCpu::RdmaSend(u32 dbindex, u64 dbinfo, hccl::Stream &stream, RdmaTaskInfo &taskInfo)
934 : {
935 5 : const HcclComStreamInfo &streamInfo = stream.GetHcclStreamInfo();
936 :
937 5 : uint8_t *sqeBuffer = nullptr;
938 5 : uint8_t *sqeTypeAddr = nullptr;
939 5 : uint8_t *sqeDfxInfoAddr = nullptr;
940 5 : uint16_t taskId = 0U;
941 :
942 5 : CHK_RET(GetStreamSqeBufferAddr(stream, sqeBuffer, sqeTypeAddr, sqeDfxInfoAddr, taskId));
943 5 : AicpuDfxInfo * const dfxInfo = (AicpuDfxInfo * const)sqeDfxInfoAddr;
944 5 : dfxInfo->opRingBufferIdx = opRingBufferIdx_;
945 5 : dfxInfo->remoteRank = taskInfo.remoteRank;
946 5 : dfxInfo->notifyId = INVALID_UINT; // 多个wr只敲一次doorbell的情况下，一般只会有一个notify
947 :
948 5 : uint32_t wrLen = 0; // 统计wr的总数据量
949 5 : for (const WrInformation& wr : taskInfo.wrInfos) {
950 0 : wrLen += wr.wrData.memList.len;
951 0 : dfxInfo->notifyId = (wr.notifyId != INVALID_UINT) ? wr.notifyId : dfxInfo->notifyId;
952 : }
953 :
954 5 : u64 dbAddr = CalcDbAddr(dbindex);
955 5 : addOneRdmaDbSendSqe_(streamInfo.actualStreamId, taskId, dbinfo, dbAddr, wrLen,
956 5 : static_cast<uint8_t>(taskInfo.rdmaType), sqeBuffer, sqeTypeAddr);
957 :
958 5 : PLF_CONFIG_INFO(PLF_TASK,
959 : "%s para: streamId[%u] taskId[%u] remoteRank[%u] RdmaType[%d] wrLen[%u] notifyId[%u]",
960 : __func__, streamInfo.actualStreamId, taskId, taskInfo.remoteRank, taskInfo.rdmaType, wrLen, dfxInfo->notifyId);
961 :
962 5 : return HCCL_SUCCESS;
963 : }
964 :
965 0 : HcclResult DispatcherAiCpu::RdmaRecord(u32 dbindex, u64 dbinfo, const struct SendWr &wr, hccl::Stream &stream,
966 : RdmaType rdmaType, u32 userRank, u64 offset, u32 notifyId)
967 : {
968 0 : return HCCL_SUCCESS;
969 : }
970 :
971 13 : HcclResult DispatcherAiCpu::GetStreamSqeBufferAddr(hccl::Stream &stream, uint8_t *&sqeBufferAddr, uint8_t *&sqeTypeAddr,
972 : uint8_t *&sqeDfxInfoAddr, uint16_t &taskId)
973 : {
974 13 : SaveStreamInfo(stream);
975 13 : HcclSqeContext* sqeContext = stream.GetSqeContextPtr();
976 13 : CHK_PTR_NULL(sqeContext);
977 13 : if (UNLIKELY(sqeContext->buffer.sqeCnt >= HCCL_PER_LAUNCH_SQE_CNT)) {
978 0 : HCCL_INFO("GetStreamSqeBufferAddr tailSqeIdx[%u], try to launchTask", sqeContext->buffer.tailSqeIdx);
979 0 : CHK_RET(LaunchTask(stream, true));
980 : }
981 13 : if (UNLIKELY(sqeContext->buffer.tailSqeIdx >= HCCL_SQE_MAX_CNT)) {
982 0 : CHK_RET(LaunchTask(stream, true));
983 :
984 0 : if (callback_ != nullptr) {
985 0 : hccl::AiCPUStreamTasks para(stream.id(), reinterpret_cast<void*>(sqeContext));
986 0 : hccl::TaskPara taskPara(TaskType::TASK_BATCH_REPORT, para);
987 0 : callback_(callBackUserPtr_, (void *)&taskPara, sizeof(struct TaskPara));
988 0 : }
989 : }
990 13 : SqeRingBuffer *sqeContextBuffer = &(sqeContext->buffer);
991 13 : uint16_t flipNum = sqeContextBuffer->filpNum;
992 13 : uint16_t nextTaskId = sqeContextBuffer->tailSqeTaskId;
993 : // nextTaskId=0的时候下发PlaceHolder
994 13 : if (UNLIKELY(nextTaskId == 0 && flipNum != 0)) {
995 0 : CHK_RET(AddFlipTask(stream));
996 : }
997 13 : if (UNLIKELY(sqeContext->buffer.tailSqeIdx >= HCCL_SQE_MAX_CNT)) {
998 0 : CHK_RET(LaunchTask(stream, true));
999 :
1000 0 : if (callback_ != nullptr) {
1001 0 : hccl::AiCPUStreamTasks para(stream.id(), reinterpret_cast<void*>(sqeContext));
1002 0 : hccl::TaskPara taskPara(TaskType::TASK_BATCH_REPORT, para);
1003 0 : callback_(callBackUserPtr_, (void *)&taskPara, sizeof(struct TaskPara));
1004 0 : }
1005 : }
1006 13 : CHK_RET(stream.GetNextSqeBufferAddr(sqeBufferAddr, sqeTypeAddr, sqeDfxInfoAddr, taskId));
1007 13 : return HCCL_SUCCESS;
1008 : }
1009 :
1010 2 : HcclResult DispatcherAiCpu::WaitRtsq(Stream& stream, const size_t& sqeCount, const bool isBlockLaunch) {
1011 : // 注意: 目前WaitRtsq不会被递归调用, 所以isBlockLaunch永远为true; 为防止以后LaunchTask递归使用WaitRtsq, 编码时考虑isBlockLaunch为false的情况
1012 :
1013 : // 检验入参
1014 2 : const HcclComStreamInfo &streamInfo = stream.GetHcclStreamInfo();
1015 2 : if (sqeCount == 0) {
1016 0 : CHK_PRT_CONT(isBlockLaunch,
1017 : HCCL_DEBUG("[DispatcherAiCpu][WaitRtsq] no sqe, streamId:%d, sqId:%u", streamInfo.actualStreamId, streamInfo.sqId));
1018 0 : return HCCL_SUCCESS;
1019 2 : } else if (sqeCount > streamInfo.sqDepth) {
1020 0 : HCCL_ERROR("[DispatcherAiCpu][WaitRtsq] sqeCount %u should be smaller than sqDepth %u]", sqeCount, streamInfo.sqDepth);
1021 0 : return HCCL_E_PTR;
1022 : }
1023 :
1024 : // Get head and tail of RTSQ ring buffer
1025 2 : HcclSqeContext *sqeContext = stream.GetSqeContextPtr();
1026 2 : CHK_PTR_NULL(sqeContext);
1027 2 : SqeRingBuffer *sqeContextBuffer = &(sqeContext->buffer);
1028 2 : CHK_PTR_NULL(sqeContextBuffer);
1029 2 : uint32_t& head = sqeContextBuffer->sqHead;
1030 2 : uint32_t& tail = sqeContextBuffer->sqTail;
1031 :
1032 : // Dump debug information
1033 2 : const uint32_t newTail = (tail + sqeCount) % streamInfo.sqDepth;
1034 : // 仅在阻塞下发场景打印，避免非阻塞场景调用时刷屏
1035 2 : CHK_PRT_CONT(isBlockLaunch,
1036 : HCCL_INFO("[DispatcherAicpu][WaitRtsq] sqid:%d sqeCount:%u head:%u curtail:%u newTail:%u", streamInfo.sqId, sqeCount, head, tail, newTail));
1037 :
1038 : // 轮询RTSQ直至获得足够大的剩余空间
1039 2 : u64 startUsec = GetCurAicpuTimestamp();
1040 2 : u64 lastUsec = startUsec;
1041 3 : while (((tail < head ? streamInfo.sqDepth : 0U) + tail - head + sqeCount >= streamInfo.sqDepth) && (tail != head)) { // 判断RTSQ中剩余sqe空间是否足够下发
1042 : // 需要放在while循环进来后第一个执行 (获取最新的RTSQ head, 查看RTSQ的消费进度)
1043 2 : CHK_RET(QuerySqStatusByType(aicpuInfo_.devId, streamInfo.sqId, DRV_SQCQ_PROP_SQ_HEAD, head));
1044 :
1045 : // 非阻塞下发场景，rtsq队列空间不足时直接返回
1046 2 : if (isBlockLaunch == false) {
1047 1 : return HCCL_SUCCESS;
1048 : }
1049 :
1050 : // 当前流无法下发，把其他流都launch一遍，避免等待的其他流没有launch
1051 2 : for (auto it = streamMap_.begin(); it != streamMap_.end(); ++it) {
1052 1 : if (it->first != streamInfo.actualStreamId) { // 不是当前stream
1053 1 : CHK_RET(LaunchTask(it->second, false)); // 非阻塞launch
1054 : }
1055 : }
1056 :
1057 : // 等待超时
1058 1 : u64 curUsec = GetCurAicpuTimestamp();
1059 1 : if (dfxTimeOutConfig_.sqFullWaitTimeOut != 0 &&
1060 0 : (curUsec - startUsec > NANOSECOND_TO_SECOND * dfxTimeOutConfig_.sqFullWaitTimeOut)) {
1061 0 : HCCL_ERROR("[DispatcherAicpu][WaitRtsq] Rtsq full, timeout %lus. curhead:%u, sqId:%d", dfxTimeOutConfig_.sqFullWaitTimeOut, head, streamInfo.sqId);
1062 0 : return HCCL_E_AGAIN;
1063 : }
1064 :
1065 : // 等待下发阶段，每隔30s打印一次状态
1066 1 : if (curUsec - lastUsec > NANOSECOND_TO_SECOND * dfx::kPrintSqInterval) {
1067 0 : lastUsec = curUsec;
1068 0 : HCCL_RUN_INFO("[DispatcherAicpu][WaitRtsq] Current state. sqid:%d, head:%u, tail:%u, sqeCount:%u",
1069 : streamInfo.sqId, head, tail, sqeCount);
1070 : }
1071 :
1072 : // 等待下发过程中出现cqe异常, 需要终止当前算子SQE的下发过程
1073 1 : if (checkOpExecStatusCallback_ != nullptr) {
1074 0 : HcclResult opExecStatus = checkOpExecStatusCallback_();
1075 0 : CHK_PRT_RET(opExecStatus != HCCL_SUCCESS,
1076 : HCCL_ERROR("[DispatcherAicpu][WaitRtsq] hccl aicpu stop launch for task exception or stop command, ret:%d", opExecStatus),
1077 : opExecStatus);
1078 : }
1079 : }
1080 :
1081 1 : return HCCL_SUCCESS;
1082 : }
1083 :
1084 0 : HcclResult DispatcherAiCpu::MemcpyRtsq(Stream& stream, const size_t sqeCount, const uint8_t *sqeArray, const uint8_t *sqeTypeArray, const AicpuDfxInfo *sqeDfxInfoArray, const bool profL1Enable, const std::vector<uint64_t>& profTimestamps, const size_t profTimestampStartIdx) {
1085 : // 检验入参
1086 0 : const HcclComStreamInfo &streamInfo = stream.GetHcclStreamInfo();
1087 0 : if (sqeCount == 0) {
1088 0 : HCCL_DEBUG("[DispatcherAiCpu][MemcpyRtsq] no sqe, streamId:%d, sqId:%u", streamInfo.actualStreamId, streamInfo.sqId);
1089 0 : return HCCL_SUCCESS;
1090 0 : } else if (sqeCount > streamInfo.sqDepth) {
1091 0 : HCCL_ERROR("[DispatcherAiCpu][MemcpyRtsq] sqeCount %u should be smaller than sqDepth %u]", sqeCount, streamInfo.sqDepth);
1092 0 : return HCCL_E_PTR;
1093 : }
1094 0 : CHK_PTR_NULL(sqeArray);
1095 0 : CHK_PTR_NULL(sqeTypeArray);
1096 0 : CHK_PTR_NULL(sqeDfxInfoArray);
1097 0 : if (profL1Enable) {
1098 : // 会访问profTimestamps[profTimestampStartIdx, profTimestampStartIdx + sqeCount - 1]
1099 0 : CHK_PRT_RET(profTimestamps.size() == 0, HCCL_ERROR("[DispatcherAiCpu][MemcpyRtsq] empty profTimestamps"), HCCL_E_INTERNAL);
1100 0 : CHK_PRT_RET(profTimestampStartIdx >= profTimestamps.size(), HCCL_ERROR("[DispatcherAiCpu][MemcpyRtsq] profTimestampStartIdx[%u] >= profTimestamps.size[%u]", profTimestampStartIdx, profTimestamps.size()), HCCL_E_INTERNAL);
1101 0 : CHK_PRT_RET((profTimestampStartIdx + sqeCount - 1) >= profTimestamps.size(), HCCL_ERROR("[DispatcherAiCpu][MemcpyRtsq] profTimestampStartIdx[%u] + sqeCount[%u] - 1 >= profTimestamps.size[%u]", profTimestampStartIdx, sqeCount, profTimestamps.size()), HCCL_E_INTERNAL);
1102 : }
1103 :
1104 : // 获得RTSQ的head和tail
1105 0 : HcclSqeContext *sqeContext = stream.GetSqeContextPtr();
1106 0 : CHK_PTR_NULL(sqeContext);
1107 0 : SqeRingBuffer *sqeContextBuffer = &(sqeContext->buffer);
1108 0 : CHK_PTR_NULL(sqeContextBuffer);
1109 0 : uint32_t& head = sqeContextBuffer->sqHead;
1110 0 : uint32_t& tail = sqeContextBuffer->sqTail;
1111 :
1112 : // Dump debug information
1113 0 : const uint32_t newTail = (tail + sqeCount) % streamInfo.sqDepth;
1114 0 : HCCL_INFO("[DispatcherAicpu][MemcpyRtsq] before memcpy, sqid:%d sqeCount:%u head:%u curtail:%u newTail:%u", streamInfo.sqId, sqeCount, head, tail, newTail);
1115 :
1116 : // 准备memcpy中目的末端基地址 (RTSQ从tail开始拷贝, [head, tail)为待执行SQE)
1117 0 : uint8_t *rtsqSqeTailBaseAddr = reinterpret_cast<uint8_t *>(streamInfo.sqBaseAddr) + tail * HCCL_SQE_SIZE;
1118 0 : uint8_t *mirrorRtsqSqeTailBaseAddr = sqeContextBuffer->rtsMirrorBuffer + tail * HCCL_SQE_SIZE;
1119 0 : uint8_t *rtsqSqeTypeTailBaseAddr = sqeContextBuffer->rtsqSqeType + tail;
1120 0 : AicpuDfxInfo *rtsqDfxInfoTailBaseAddr = sqeContextBuffer->rtsDfxInfo + tail;
1121 :
1122 0 : uint32_t tailLeft = streamInfo.sqDepth - tail; // RTSQ tail到buffer末端的剩余空间 (不包括buffer前端到head的剩余空间)
1123 0 : HCCL_INFO("[DispatcherAicpu][MemcpyRtsq] cpy sqe, tailLeft:%u, sqeCount:%u, streamId:%u", tailLeft, sqeCount, stream.id());
1124 0 : if (sqeCount <= tailLeft) { // buffer末端剩余空间放得下新增sqe
1125 : // 向buffer末端拷贝sqeCount个SQE信息
1126 :
1127 : // 拷贝SQE内容到RTSQ
1128 0 : CHK_SAFETY_FUNC_RET(memcpy_s(rtsqSqeTailBaseAddr, tailLeft * HCCL_SQE_SIZE, sqeArray, sqeCount * HCCL_SQE_SIZE));
1129 :
1130 : // 拷贝SQE内容到RTSQ mirror
1131 0 : CHK_SAFETY_FUNC_RET(memcpy_s(mirrorRtsqSqeTailBaseAddr, tailLeft * HCCL_SQE_SIZE, sqeArray, sqeCount * HCCL_SQE_SIZE));
1132 :
1133 : // 拷贝SQE类型
1134 0 : CHK_SAFETY_FUNC_RET(memcpy_s(rtsqSqeTypeTailBaseAddr, tailLeft, sqeTypeArray, sqeCount));
1135 :
1136 : // 拷贝SQE DfxInfo
1137 0 : CHK_SAFETY_FUNC_RET(memcpy_s(rtsqDfxInfoTailBaseAddr, tailLeft * sizeof(AicpuDfxInfo), sqeDfxInfoArray, sqeCount * sizeof(AicpuDfxInfo)));
1138 : } else { // 需要buffer末端和首端的剩余空间
1139 : // 先向buffer末端拷贝tailLeft个SQE信息, 再向buffer首端拷贝sqeCount-tailLeft个SQE信息
1140 :
1141 : // 拷贝SQE内容到RTSQ
1142 0 : CHK_SAFETY_FUNC_RET(memcpy_s(rtsqSqeTailBaseAddr, tailLeft * HCCL_SQE_SIZE, sqeArray, tailLeft * HCCL_SQE_SIZE));
1143 0 : CHK_SAFETY_FUNC_RET(memcpy_s(reinterpret_cast<uint8_t *>(streamInfo.sqBaseAddr),
1144 : streamInfo.sqDepth * HCCL_SQE_SIZE,
1145 : sqeArray + tailLeft * HCCL_SQE_SIZE,
1146 : (sqeCount - tailLeft) * HCCL_SQE_SIZE));
1147 :
1148 : // 拷贝SQE内容到RTSQ mirror
1149 0 : CHK_SAFETY_FUNC_RET(memcpy_s(mirrorRtsqSqeTailBaseAddr, tailLeft * HCCL_SQE_SIZE, sqeArray, tailLeft * HCCL_SQE_SIZE));
1150 0 : CHK_SAFETY_FUNC_RET(memcpy_s(sqeContextBuffer->rtsMirrorBuffer,
1151 : streamInfo.sqDepth * HCCL_SQE_SIZE,
1152 : sqeArray + tailLeft * HCCL_SQE_SIZE,
1153 : (sqeCount - tailLeft) * HCCL_SQE_SIZE));
1154 :
1155 : // 拷贝SQE type
1156 0 : CHK_SAFETY_FUNC_RET(memcpy_s(rtsqSqeTypeTailBaseAddr, tailLeft, sqeTypeArray, tailLeft));
1157 0 : CHK_SAFETY_FUNC_RET(memcpy_s(sqeContextBuffer->rtsqSqeType, streamInfo.sqDepth, sqeTypeArray + tailLeft, (sqeCount - tailLeft)));
1158 :
1159 : // 拷贝SQE DfxInfo
1160 0 : CHK_SAFETY_FUNC_RET(memcpy_s(rtsqDfxInfoTailBaseAddr,
1161 : tailLeft * sizeof(AicpuDfxInfo), sqeDfxInfoArray, tailLeft * sizeof(AicpuDfxInfo)));
1162 0 : CHK_SAFETY_FUNC_RET(memcpy_s(sqeContextBuffer->rtsDfxInfo, streamInfo.sqDepth * sizeof(AicpuDfxInfo),
1163 : sqeDfxInfoArray + tailLeft, (sqeCount - tailLeft) * sizeof(AicpuDfxInfo)));
1164 : }
1165 :
1166 : // 更新RTSQ ring buffer的tail
1167 0 : CHK_RET(ConfigSqStatusByType(aicpuInfo_.devId, streamInfo.sqId, DRV_SQCQ_PROP_SQ_TAIL, newTail));
1168 0 : tail = newTail;
1169 0 : PLF_CONFIG_INFO(PLF_TASK,
1170 : "%s success, sqid:%d, sqe_num:%u, curHead:%u, curtail:%u", __func__, streamInfo.sqId, sqeCount, head, tail);
1171 :
1172 : // 上报profiling信息
1173 0 : if (profL1Enable) {
1174 : // Cache hit下SQE ring buffer中[tail-cnt, tail)为待下发SQE, 其数量一定为0 (否则更新tailSqeIdx后, 会导致待下发SQE未下发, 而已下发SQE重复下发)
1175 0 : CHK_PRT_RET(sqeContextBuffer->sqeCnt != 0, HCCL_ERROR("[DispatcherAicpu][MemcpyRtsq] sqeContextBuffer->sqeCnt[%u] should be zero!", sqeContextBuffer->sqeCnt), HCCL_E_INTERNAL);
1176 :
1177 : // 上报flip placeholder的profiling信息
1178 0 : if ((*sqeTypeArray) == SqeType::FLIP_PLACEHOLDER_SQE) {
1179 0 : CHK_PRT_RET(sqeCount != 1, HCCL_ERROR("[DispatcherAicpu][MemcpyRtsq] sqeCount[%u] should be 1!", sqeCount), HCCL_E_INTERNAL);
1180 :
1181 : // 注意: 参考AddFlipTask, 先上报flip task (提醒profiling翻转taskid), 之后再拷贝到SQE ring buffer (捕捉placeholder SQE相关的profiling信息)
1182 : // 注意: ProfilingManager::TaskProfilingCallBack->ReportFilpTask不会扫描SQE ring buffer, 也不会更新其中的streamToSqeIdxMap_
1183 0 : if (callback_ != nullptr) {
1184 0 : const rtStarsPlaceHolderSqe_t *placeholderSqePtr = reinterpret_cast<const rtStarsPlaceHolderSqe_t *>(sqeArray);
1185 0 : hccl::FlipTaskPara para(stream.id(), placeholderSqePtr->header.taskId, placeholderSqePtr->u.flip_task_info.flipNumReport);
1186 0 : hccl::TaskPara taskPara(TaskType::TASK_FLIP, para);
1187 0 : callback_(callBackUserPtr_, (void *)&taskPara, sizeof(struct TaskPara));
1188 0 : }
1189 : }
1190 :
1191 : // 循环拷贝cached SQE到SQE ring buffer中
1192 0 : size_t reportSqeCount = 0;
1193 0 : while (reportSqeCount < sqeCount) {
1194 0 : CHK_PRT_RET(sqeContextBuffer->tailSqeIdx > HCCL_SQE_MAX_CNT, HCCL_ERROR("[DispatcherAicpu][MemcpyRtsq] tailSqeIdx[%u] > HCCL_SQE_MAX_CNT[%u]", sqeContextBuffer->tailSqeIdx, HCCL_SQE_MAX_CNT), HCCL_E_INTERNAL);
1195 0 : const size_t sqeTailLeft = HCCL_SQE_MAX_CNT - sqeContextBuffer->tailSqeIdx;
1196 0 : if (sqeTailLeft > 0) {
1197 : // 准备profiling上报的目的末端基地址 (SQE ring buffer从tail开始拷贝)
1198 0 : uint8_t *sqeLocalBuffTailBaseAddr = sqeContextBuffer->localBuff + sqeContextBuffer->tailSqeIdx * HCCL_SQE_SIZE;
1199 0 : uint8_t *sqeTypeTailBaseAddr = sqeContextBuffer->sqeType + sqeContextBuffer->tailSqeIdx;
1200 0 : AicpuDfxInfo *dfxInfoTailBaseAddr = sqeContextBuffer->dfxInfo + sqeContextBuffer->tailSqeIdx;
1201 0 : uint64_t *profTimestapTailBaseAddr = sqeContextBuffer->profTimestap + sqeContextBuffer->tailSqeIdx;
1202 :
1203 : // 向SQE ring buffer末端拷贝SQE信息[reportSqeCount, reportSqeCount + tmpSqeCount - 1] (只用于profiling上报, 不会下发)
1204 0 : const size_t tmpSqeCount = std::min(sqeCount - reportSqeCount, sqeTailLeft);
1205 0 : HCCL_INFO("[DispatcherAicpu][MemcpyRtsq] report sqe profiling, sqeCount[%u] reportSqeCount[%u] tailSqeIdx[%u] sqeTailLeft[%u] tmpSqeCount[%u]", sqeCount, reportSqeCount, sqeContextBuffer->tailSqeIdx, sqeTailLeft, tmpSqeCount);
1206 :
1207 : // 拷贝SQE内容
1208 0 : CHK_SAFETY_FUNC_RET(memcpy_s(sqeLocalBuffTailBaseAddr, sqeTailLeft * HCCL_SQE_SIZE, sqeArray + reportSqeCount * HCCL_SQE_SIZE, tmpSqeCount * HCCL_SQE_SIZE));
1209 :
1210 : // 拷贝SQE类型
1211 0 : CHK_SAFETY_FUNC_RET(memcpy_s(sqeTypeTailBaseAddr, sqeTailLeft, sqeTypeArray + reportSqeCount, tmpSqeCount));
1212 :
1213 : // 拷贝SQE DfxInfo
1214 0 : CHK_SAFETY_FUNC_RET(memcpy_s(dfxInfoTailBaseAddr, sqeTailLeft * sizeof(AicpuDfxInfo), sqeDfxInfoArray + reportSqeCount, tmpSqeCount * sizeof(AicpuDfxInfo)));
1215 :
1216 : // 拷贝SQE timestamp
1217 0 : CHK_SAFETY_FUNC_RET(memcpy_s(profTimestapTailBaseAddr, sqeTailLeft * sizeof(uint64_t), profTimestamps.data() + profTimestampStartIdx + reportSqeCount, tmpSqeCount * sizeof(uint64_t)));
1218 :
1219 : // 注意: sqeContextBuffer->sqeCnt不更新, 仍然为0 (即拷贝的SQE信息为已下发待上报), 避免SQE重复下发
1220 0 : sqeContextBuffer->tailSqeIdx += static_cast<uint16_t>(tmpSqeCount);
1221 0 : reportSqeCount += tmpSqeCount;
1222 : } else {
1223 : // 注意: SQE ring buffer中待下发SQE数量一定为0, 不需要调用LaunchTask将待下发变成已下发待上报, 可以直接上报profiling将已下发待上报变成已上报
1224 : // 注意: 调用后, ProfilingManager::StartReportSqeIdx为HCCL_SQE_MAX_CNT
1225 0 : if (callback_ != nullptr) {
1226 0 : hccl::AiCPUStreamTasks para(stream.id(), reinterpret_cast<void*>(sqeContext));
1227 0 : hccl::TaskPara taskPara(TaskType::TASK_BATCH_REPORT, para);
1228 0 : callback_(callBackUserPtr_, (void *)&taskPara, sizeof(struct TaskPara));
1229 0 : }
1230 :
1231 : // SQE ring buffer中所有SQE均为已上报 -> 清理SQE ring buffer
1232 0 : HCCL_INFO("[DispatcherAicpu][MemcpyRtsq] Sqe index to %u, need clear", HCCL_SQE_MAX_CNT);
1233 0 : CHK_PRT_RET(sqeContextBuffer->sqeCnt != 0, HCCL_ERROR("[DispatcherAicpu][MemcpyRtsq] Sqe index to %u, but sqeCnt[%u] is not 0", HCCL_SQE_MAX_CNT, sqeContextBuffer->sqeCnt), HCCL_E_INTERNAL);
1234 0 : CHK_RET(stream.ClearLocalBuff()); // 会将stream.sqeContextBuffer中的sqeCnt和tailSqeIdx设置为0
1235 0 : CHK_PRT_RET(sqeContextBuffer->sqeCnt != 0, HCCL_ERROR("[DispatcherAicpu][MemcpyRtsq] sqeCnt[%u] should be 0 after clear", sqeContextBuffer->sqeCnt), HCCL_E_INTERNAL);
1236 0 : CHK_PRT_RET(sqeContextBuffer->tailSqeIdx != 0, HCCL_ERROR("[DispatcherAicpu][MemcpyRtsq] tailSqeIdx[%u] should be 0 after clear", sqeContextBuffer->tailSqeIdx), HCCL_E_INTERNAL);
1237 :
1238 : // 参考HcclCommAicpu::ClearLocalBuff, 调用Stream::ClearLocalBuff后, 应该调用ProfilingManager::UpdateStartReportSqeIdx手动将ProfilingManager::StartReportSqeIdx设置为0
1239 : // 注意: 由于platform暂未将UpdateStartReportSqeIdx作为回调函数传入, 无法直接调用此framework函数 -> 通过callback_ (即ProfilingManager::TaskProfilingCallBack) 间接将ProfilingManager::StartReportSqeIdx设置为0
1240 : // 注意: 由于调用前ProfilingManager::StartReportSqeIdx为HCCL_SQE_MAX_CNT, tailSqeIdx为0, 即从startIdx=HCCL_SQE_MAX_CNT到endIdx=0, ProfilingManager不会进入profiling上报代码, 而是只会调用UpdateStartReportSqeIdx设置StartReportSqeIdx为0
1241 0 : if (callback_ != nullptr) {
1242 0 : HCCL_INFO("[DispatcherAicpu][MemcpyRtsq] re-invoke callback_ to reset StartReportSqeIdx as 0 in ProfilingManager");
1243 :
1244 0 : hccl::AiCPUStreamTasks para(stream.id(), reinterpret_cast<void*>(sqeContext));
1245 0 : hccl::TaskPara taskPara(TaskType::TASK_BATCH_REPORT, para);
1246 0 : callback_(callBackUserPtr_, (void *)&taskPara, sizeof(struct TaskPara));
1247 0 : }
1248 : }
1249 : }
1250 : }
1251 :
1252 0 : return HCCL_SUCCESS;
1253 : }
1254 :
1255 0 : HcclResult DispatcherAiCpu::AddFlipTask(Stream &stream)
1256 : {
1257 0 : HcclSqeContext *sqeContext = stream.GetSqeContextPtr();
1258 0 : CHK_PTR_NULL(sqeContext);
1259 0 : SqeRingBuffer *sqeContextBuffer = &(sqeContext->buffer);
1260 0 : CHK_PTR_NULL(sqeContextBuffer);
1261 0 : uint16_t flipNum = sqeContextBuffer->filpNum;
1262 0 : uint16_t taskId = sqeContextBuffer->tailSqeTaskId;
1263 :
1264 0 : if (callback_ != nullptr) {
1265 0 : hccl::FlipTaskPara para(stream.id(), taskId, flipNum);
1266 0 : hccl::TaskPara taskPara(TaskType::TASK_FLIP, para);
1267 0 : callback_(callBackUserPtr_, (void *)&taskPara, sizeof(struct TaskPara));
1268 0 : }
1269 :
1270 0 : const HcclComStreamInfo &streamInfo = stream.GetHcclStreamInfo();
1271 :
1272 0 : uint8_t *sqeBufferAddr = nullptr;
1273 0 : uint8_t *sqeTypeAddr = nullptr;
1274 0 : uint8_t *sqeDfxInfoAddr = nullptr;
1275 0 : CHK_RET(stream.GetNextSqeBufferAddr(sqeBufferAddr, sqeTypeAddr, sqeDfxInfoAddr, taskId));
1276 :
1277 0 : AicpuDfxInfo * const dfxInfo = (AicpuDfxInfo * const)sqeDfxInfoAddr;
1278 0 : dfxInfo->opRingBufferIdx = opRingBufferIdx_;
1279 0 : dfxInfo->remoteRank = INVALID_VALUE_RANKID;
1280 0 : dfxInfo->notifyId = INVALID_VALUE_RANKID;
1281 0 : addOneFlipPlaceHolderSqe_(streamInfo.actualStreamId, flipNum, taskId, sqeBufferAddr, sqeTypeAddr);
1282 :
1283 0 : PLF_CONFIG_INFO(PLF_TASK,
1284 : "%s para: taskId[%u] streamId[%u] flipNum[%u]", __func__, taskId, streamInfo.actualStreamId, flipNum);
1285 0 : return HCCL_SUCCESS;
1286 : }
1287 :
1288 0 : HcclResult DispatcherAiCpu::AddRetryPreamble(Stream &stream)
1289 : {
1290 0 : return AddFlipTask(stream);
1291 : }
1292 :
1293 22 : void DispatcherAiCpu::SaveStreamInfo(hccl::Stream &stream)
1294 : {
1295 22 : const HcclComStreamInfo &streamInfo = stream.GetHcclStreamInfo();
1296 22 : if (streamMap_.find(streamInfo.actualStreamId) == streamMap_.end()) {
1297 17 : streamMap_.insert({streamInfo.actualStreamId, stream});
1298 17 : HCCL_INFO("[DispatcherAiCpu][SaveStreamInfo] stream id[%d]", streamInfo.actualStreamId);
1299 : }
1300 22 : return;
1301 : }
1302 :
1303 2 : HcclResult DispatcherAiCpu::StreamSync(Stream &stream)
1304 : {
1305 2 : uint32_t head = 0;
1306 2 : uint32_t tail = 0;
1307 : const HcclComStreamInfo *streamInfo;
1308 2 : u64 startUsec = GetCurAicpuTimestamp();
1309 2 : u64 lastUsec = startUsec;
1310 2 : CHK_RET(stream.GetStreamInfo(streamInfo));
1311 :
1312 2 : CHK_RET(QuerySqStatusByType(aicpuInfo_.devId, streamInfo->sqId, DRV_SQCQ_PROP_SQ_TAIL, tail));
1313 2 : HCCL_INFO("StreamSync aicpu stream sqid[%d] tail[%u]", streamInfo->sqId, tail);
1314 : do {
1315 2 : CHK_RET(QuerySqStatusByType(aicpuInfo_.devId, streamInfo->sqId, DRV_SQCQ_PROP_SQ_HEAD, head));
1316 2 : u64 curUsec = GetCurAicpuTimestamp();
1317 2 : if (curUsec - startUsec > NANOSECOND_TO_SECOND * dfxTimeOutConfig_.sqeTimeOutTimeOut) {
1318 1 : HCCL_ERROR("stream sync timeout %lus. curhead:%u, curtall:%u, sqId:%d",
1319 : dfxTimeOutConfig_.sqeTimeOutTimeOut, head, tail, streamInfo->sqId);
1320 1 : return HCCL_E_TIMEOUT;
1321 : }
1322 :
1323 : // 等待下发阶段，每隔30s打印一次状态
1324 1 : if (curUsec - lastUsec > NANOSECOND_TO_SECOND * dfx::kPrintSqInterval) {
1325 0 : lastUsec = curUsec;
1326 0 : HCCL_RUN_INFO("[StreamSync]Current state. sqid:%d, head:%u, tail:%u",
1327 : streamInfo->sqId, head, tail);
1328 : }
1329 1 : } while (head != tail);
1330 :
1331 1 : return HCCL_SUCCESS;
1332 : }
1333 :
1334 5 : u64 DispatcherAiCpu::CalcDbAddr(u32 dbindex)
1335 : {
1336 5 : u64 dbAddr = 0;
1337 5 : if (aicpuInfo_.devType == DevType::DEV_TYPE_910_93) {
1338 : // 910_93 HCCS_SW 组网
1339 2 : constexpr u64 roceBaseAddr = 0x202000000000ULL;
1340 2 : constexpr u64 roceVfDbCfg0Reg = 0x230ULL;
1341 2 : constexpr u64 chipAddrOffset = 0x20000000000ULL;
1342 2 : constexpr u64 dieAddrOffset = 0x10000000000ULL;
1343 2 : constexpr u32 dbDieIdMask = 0x00ff0000;
1344 2 : constexpr u32 dbDieIdShift = 16; // 16 is dbDieIdShift
1345 2 : dbAddr = roceBaseAddr + roceVfDbCfg0Reg + chipAddrOffset * aicpuInfo_.chipId +
1346 2 : dieAddrOffset * ((dbindex & dbDieIdMask) >> dbDieIdShift);
1347 : } else {
1348 3 : constexpr u64 roceBaseAddr = 0x2000000000ULL;
1349 3 : constexpr u64 roceVfDbCfg0Reg = 0x230ULL;
1350 3 : constexpr u64 chipAddrOffset = 0x80000000000ULL;
1351 3 : constexpr u64 dieAddrOffset = 0x10000000000ULL;
1352 3 : constexpr u32 dbDieIdMask = 0x00ff0000;
1353 3 : constexpr u32 dbDieIdShift = 16; // 16 is dbDieIdShift
1354 3 : dbAddr = roceBaseAddr + roceVfDbCfg0Reg + chipAddrOffset * aicpuInfo_.chipId +
1355 3 : dieAddrOffset * ((dbindex & dbDieIdMask) >> dbDieIdShift);
1356 : }
1357 :
1358 5 : HCCL_DEBUG("%s dbindex:%u, devType:%u, chipId:%lld, dbAddr:%llu",
1359 : __func__, dbindex, aicpuInfo_.devType, aicpuInfo_.chipId, dbAddr);
1360 5 : return dbAddr;
1361 : }
1362 :
1363 444 : void DispatcherAiCpu::InitTimeOutConfig()
1364 : {
1365 444 : dfxTimeOutConfig_.useCredit = false;
1366 444 : dfxTimeOutConfig_.sqeTimeOutTimeOut = GetMaxNotifyWaitTime();
1367 444 : dfxTimeOutConfig_.sqeCreditTimeOut = RT_STARS_NEVER_TIMEOUT_KERNEL_CREDIT;
1368 444 : dfxTimeOutConfig_.sqeWaitTimeOut = dfx::kKfcTimeOut;
1369 444 : dfxTimeOutConfig_.sqFullWaitTimeOut = dfx::kSqFullWaitTimeOut;
1370 444 : HCCL_INFO("[DispatcherAiCpu][InitTimeOutConfig]DFX timeout config init successfully with details: [%s]",
1371 : dfxTimeOutConfig_.ToString().c_str());
1372 444 : }
1373 : } // namespace hccl
|
