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 444 : 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 444 : addOneNotifyWaitSqe_ = AddOneNotifyWaitSqeV1;
82 444 : addOneRecordSqe_ = AddOneRecordSqeV1;
83 444 : addOneWriteValueRecordSqe_ = AddOneWriteValueRecordSqeV1;
84 444 : addOneMemcpySqe_ = AddOneMemcpySqeV1;
85 444 : addOneEventResetSqe_ = AddOneEventResetSqeV1;
86 444 : addOneEventRecordSqe_ = AddOneEventRecordSqeV1;
87 444 : addOneEventWaitSqe_ = AddOneEventWaitSqeV1;
88 444 : addOneRdmaDbSendSqe_ = AddOneRdmaDbSendSqeV1;
89 444 : addOneFlipPlaceHolderSqe_ = AddOneFlipPlaceHolderSqeV1;
90 444 : addOneCacheMemcpyPlaceHolderSqe_ = AddOneCacheMemcpyPlaceHolderSqeV1;
91 444 : addOneCacheNotifyWaitPlaceholderSqe_ = AddOneCacheNotifyWaitPlaceholderSqeV1;
92 444 : addOneCacheNotifyRecordPlaceholderSqe_ = AddOneCacheNotifyRecordPlaceholderSqeV1;
93 444 : addOneCacheWriteValuePlaceholderSqe_ = AddOneCacheWriteValuePlaceholderSqeV1;
94 444 : addOneCacheMemcpyRecordPlaceholderSqe_ = AddOneCacheMemcpyRecordPlaceholderSqeV1;
95 444 : CHK_PTR_NULL(addOneRdmaDbSendSqe_);
96 444 : CHK_PTR_NULL(addOneFlipPlaceHolderSqe_);
97 444 : CHK_PTR_NULL(addOneCacheMemcpyPlaceHolderSqe_);
98 444 : CHK_PTR_NULL(addOneCacheNotifyWaitPlaceholderSqe_);
99 444 : CHK_PTR_NULL(addOneCacheNotifyRecordPlaceholderSqe_);
100 444 : CHK_PTR_NULL(addOneCacheWriteValuePlaceholderSqe_);
101 444 : CHK_PTR_NULL(addOneCacheMemcpyRecordPlaceholderSqe_);
102 : }
103 :
104 444 : CHK_PTR_NULL(addOneNotifyWaitSqe_);
105 444 : CHK_PTR_NULL(addOneRecordSqe_);
106 444 : CHK_PTR_NULL(addOneWriteValueRecordSqe_);
107 444 : CHK_PTR_NULL(addOneMemcpySqe_);
108 444 : CHK_PTR_NULL(addOneEventResetSqe_);
109 444 : CHK_PTR_NULL(addOneEventRecordSqe_);
110 444 : CHK_PTR_NULL(addOneEventWaitSqe_);
111 :
112 444 : 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 2967779 : while (((tail < head ? streamInfo.sqDepth : 0U) + tail - head + cnt >= streamInfo.sqDepth) && (tail != head)) { // 判断剩余sqe空间是否足够下发
638 : // 需要放在while循环进来后第一个执行
639 2967777 : CHK_RET(QuerySqStatusByType(aicpuInfo_.devId, streamInfo.sqId, DRV_SQCQ_PROP_SQ_HEAD, head));
640 :
641 : // 非阻塞下发场景，rtsq队列空间不足时直接返回
642 2967777 : if (isBlockLaunch == false) {
643 0 : return HCCL_SUCCESS;
644 : }
645 :
646 : // 当前流无法下发，把其他流都launch一遍，避免等待的其他流没有launch
647 4520132 : for (auto it = streamMap_.begin(); it != streamMap_.end(); ++it) {
648 1552355 : if (it->first != streamInfo.actualStreamId) {
649 1 : if (it->second.IsInvalid()) { // 跳过已销毁的stream (streamMap_只增不删, 原stream销毁后副本通过shared invalid标志感知)
650 0 : HCCL_WARNING("[DispatcherAiCpu][LaunchTask] skip invalid stream in streamMap, streamId:%d", it->first);
651 0 : continue;
652 : }
653 1 : CHK_RET(LaunchTask(it->second, false));
654 : }
655 : }
656 2967777 : u64 curUsec = GetCurAicpuTimestamp();
657 2967777 : if (dfxTimeOutConfig_.sqFullWaitTimeOut != 0 &&
658 2967776 : (curUsec - startUsec > NANOSECOND_TO_SECOND * dfxTimeOutConfig_.sqFullWaitTimeOut)) {
659 2 : HCCL_ERROR("Rtsq full, timeout %lus. curhead:%u, sqId:%d", dfxTimeOutConfig_.sqFullWaitTimeOut, head,
660 : streamInfo.sqId);
661 2 : return HCCL_E_AGAIN;
662 : }
663 :
664 : // 等待下发阶段，每隔30s打印一次状态
665 2967775 : if (curUsec - lastUsec > NANOSECOND_TO_SECOND * dfx::kPrintSqInterval) {
666 0 : lastUsec = curUsec;
667 0 : HCCL_RUN_INFO("[LaunchTask][WaitLaunchWhileLoop]Current state. sqid:%d, head:%u, tail:%u, cnt:%u",
668 : streamInfo.sqId, head, tail, cnt);
669 : }
670 :
671 : // 下发过程中出现cqe异常
672 2967775 : if (checkOpExecStatusCallback_ != nullptr) {
673 0 : HcclResult opExecStatus = checkOpExecStatusCallback_();
674 0 : CHK_PRT_RET(opExecStatus != HCCL_SUCCESS,
675 : HCCL_ERROR("hccl aicpu stop launch for task exception or stop command, ret:%d", opExecStatus),
676 : opExecStatus);
677 : }
678 : }
679 :
680 2 : uint32_t left = streamInfo.sqDepth - tail; // sqeAddr 剩余空间
681 2 : const auto tailSqeIdx = sqeContextBuffer->tailSqeIdx;
682 2 : HCCL_INFO("cpy sqe, left:%u, tailSqeId:%u, cnt:%u, streamId:%u", left, tailSqeIdx, cnt, stream.id());
683 2 : if (cnt <= left) { // 剩余buffer放得下新增sqe
684 1 : CHK_SAFETY_FUNC_RET(memcpy_s(
685 : reinterpret_cast<uint8_t *>(streamInfo.sqBaseAddr) + tail * HCCL_SQE_SIZE,
686 : left * HCCL_SQE_SIZE,
687 : sqeContextBuffer->localBuff + (tailSqeIdx - cnt) * HCCL_SQE_SIZE,
688 : cnt * HCCL_SQE_SIZE));
689 :
690 1 : CHK_SAFETY_FUNC_RET(memcpy_s(sqeContextBuffer->rtsMirrorBuffer + tail * HCCL_SQE_SIZE,
691 : left * HCCL_SQE_SIZE,
692 : sqeContextBuffer->localBuff + (tailSqeIdx - cnt) * HCCL_SQE_SIZE,
693 : cnt * HCCL_SQE_SIZE));
694 :
695 1 : CHK_SAFETY_FUNC_RET(memcpy_s(sqeContextBuffer->rtsqSqeType + tail, left,
696 : sqeContextBuffer->sqeType + (tailSqeIdx - cnt), cnt));
697 1 : CHK_SAFETY_FUNC_RET(memcpy_s(sqeContextBuffer->rtsDfxInfo + tail, left * sizeof(AicpuDfxInfo),
698 : sqeContextBuffer->dfxInfo + (tailSqeIdx - cnt), cnt * sizeof(AicpuDfxInfo)));
699 : } else {
700 1 : CHK_SAFETY_FUNC_RET(memcpy_s(reinterpret_cast<uint8_t *>(streamInfo.sqBaseAddr) + tail * HCCL_SQE_SIZE,
701 : left * HCCL_SQE_SIZE,
702 : sqeContextBuffer->localBuff + (tailSqeIdx - cnt) * HCCL_SQE_SIZE,
703 : left * HCCL_SQE_SIZE));
704 :
705 1 : CHK_SAFETY_FUNC_RET(memcpy_s(reinterpret_cast<uint8_t *>(streamInfo.sqBaseAddr),
706 : streamInfo.sqDepth * HCCL_SQE_SIZE,
707 : sqeContextBuffer->localBuff + (tailSqeIdx - cnt + left) * HCCL_SQE_SIZE,
708 : (cnt - left) * HCCL_SQE_SIZE));
709 :
710 1 : CHK_SAFETY_FUNC_RET(memcpy_s(sqeContextBuffer->rtsMirrorBuffer + tail * HCCL_SQE_SIZE,
711 : left * HCCL_SQE_SIZE,
712 : sqeContextBuffer->localBuff + (tailSqeIdx - cnt) * HCCL_SQE_SIZE,
713 : left * HCCL_SQE_SIZE));
714 :
715 1 : CHK_SAFETY_FUNC_RET(memcpy_s(sqeContextBuffer->rtsMirrorBuffer,
716 : streamInfo.sqDepth * HCCL_SQE_SIZE,
717 : sqeContextBuffer->localBuff + (tailSqeIdx - cnt + left) * HCCL_SQE_SIZE,
718 : (cnt - left) * HCCL_SQE_SIZE));
719 :
720 1 : CHK_SAFETY_FUNC_RET(memcpy_s(sqeContextBuffer->rtsqSqeType + tail,
721 : left, sqeContextBuffer->sqeType + (tailSqeIdx - cnt), left));
722 1 : CHK_SAFETY_FUNC_RET(memcpy_s(sqeContextBuffer->rtsqSqeType + 0, streamInfo.sqDepth,
723 : sqeContextBuffer->sqeType + (tailSqeIdx - cnt + left), (cnt - left)));
724 1 : CHK_SAFETY_FUNC_RET(memcpy_s(sqeContextBuffer->rtsDfxInfo + tail,
725 : left * sizeof(AicpuDfxInfo), sqeContextBuffer->dfxInfo + (tailSqeIdx - cnt), left * sizeof(AicpuDfxInfo)));
726 1 : CHK_SAFETY_FUNC_RET(memcpy_s(sqeContextBuffer->rtsDfxInfo + 0, streamInfo.sqDepth * sizeof(AicpuDfxInfo),
727 : sqeContextBuffer->dfxInfo + (tailSqeIdx - cnt + left), (cnt - left) * sizeof(AicpuDfxInfo)));
728 : }
729 : // 打印算子展开下发的SQE内容for debug
730 : // 设置HCCL_DEBUG_CONFIG="task", 或者设置ASCEND_GLOBAL_LOG_LEVEL=0
731 2 : if ((UNLIKELY(GetExternalInputDebugConfig() & PLF_TASK)) || UNLIKELY(HcclCheckLogLevel(HCCL_LOG_DEBUG))) {
732 2 : const int32_t streamId = stream.GetHcclStreamInfo().actualStreamId;
733 2 : PLF_CONFIG_DEBUG(PLF_TASK, "[DispatcherAicpu][LaunchTask] dump content of %u dispatched SQEs with stream id %u", cnt, streamId);
734 :
735 2 : uint8_t *sqeArray = sqeContextBuffer->localBuff + (tailSqeIdx - cnt) * HCCL_SQE_SIZE;
736 2 : uint8_t *sqeTypeArray = sqeContextBuffer->sqeType + (tailSqeIdx - cnt);
737 2 : AicpuDfxInfo *sqeDfxInfoArray = sqeContextBuffer->dfxInfo + (tailSqeIdx - cnt);
738 14 : for (size_t sqeIdx = 0; sqeIdx < cnt; ++sqeIdx) {
739 12 : uint8_t *sqePtr = sqeArray + sqeIdx * HCCL_SQE_SIZE;
740 12 : const uint8_t sqeType = sqeTypeArray[sqeIdx];
741 12 : if (sqeType == SqeType::FLIP_PLACEHOLDER_SQE) {
742 0 : const rtStarsPlaceHolderSqe_t *placeholderSqePtr = reinterpret_cast<const rtStarsPlaceHolderSqe_t *>(sqeArray + sqeIdx * HCCL_SQE_SIZE);
743 0 : PLF_CONFIG_DEBUG(PLF_TASK, "[DispatcherAicpu][LaunchTask] %uth dispatched SQE (placeholder) header.type[%u] taskid[%u]", sqeIdx, placeholderSqePtr->header.type, placeholderSqePtr->header.taskId);
744 : } else {
745 12 : PLF_CONFIG_DEBUG(PLF_TASK, "[DispatcherAicpu][LaunchTask] %uth dispatched SQE", sqeIdx);
746 : }
747 :
748 12 : CHK_RET(OpUnfoldCache::DumpSqeContent(sqePtr, sqeType));
749 :
750 12 : const AicpuDfxInfo& dfxinfo = sqeDfxInfoArray[sqeIdx];
751 12 : PLF_CONFIG_DEBUG(PLF_TASK, "[DispatcherAicpu][LaunchTask] AicpuDfxInfo: remoteRank[%u] opRingBufferIdx[%u] notifyId[%u]",
752 : dfxinfo.remoteRank, dfxinfo.opRingBufferIdx, dfxinfo.notifyId);
753 : }
754 : }
755 :
756 : // 当前算子展开的SQE需要被动态缓存
757 2 : if (needAddSqe_) {
758 0 : CHK_PTR_NULL(cachePtr_);
759 :
760 : // 查找key对应的cache entry, 如果不存在 (即当前算子第一次LaunchTask), 创建新的cache entry
761 0 : OpUnfoldCacheEntry *entryPtr = nullptr;
762 0 : CHK_RET(cachePtr_->FindEntry(key_, &entryPtr));
763 0 : if (entryPtr == nullptr) {
764 0 : CHK_RET(cachePtr_->AddEntry(key_, userInputMemRanges_, userOutputMemRanges_, &entryPtr));
765 : }
766 0 : CHK_PTR_NULL(entryPtr);
767 :
768 : // 准备SQE相关信息的数组基地址
769 0 : uint8_t *sqeArray = sqeContextBuffer->localBuff + (tailSqeIdx - cnt) * HCCL_SQE_SIZE;
770 0 : uint8_t *sqeTypeArray = sqeContextBuffer->sqeType + (tailSqeIdx - cnt);
771 0 : AicpuDfxInfo *sqeDfxInfoArray = sqeContextBuffer->dfxInfo + (tailSqeIdx - cnt);
772 :
773 : // 遍历sqeType找到placeholder的位置
774 0 : std::vector<size_t> placeholderIdxes;
775 0 : uint8_t *curSqeTypePtr = sqeTypeArray;
776 0 : for (size_t sqeTypeIdx = 0; sqeTypeIdx < cnt; ++sqeTypeIdx) {
777 0 : if (*curSqeTypePtr == SqeType::FLIP_PLACEHOLDER_SQE) {
778 0 : placeholderIdxes.emplace_back(sqeTypeIdx);
779 : }
780 0 : ++curSqeTypePtr;
781 : }
782 :
783 : // 在动态缓存中分配实际需要的SQE数组
784 0 : const size_t cacheableSqeCount = cnt - placeholderIdxes.size();
785 0 : const int32_t streamId = stream.GetHcclStreamInfo().actualStreamId;
786 0 : size_t arrayIdx = 0;
787 0 : CHK_RET(entryPtr->AllocSqeArray(cacheableSqeCount, streamId, arrayIdx));
788 :
789 : // 分段拷贝SQE相关信息到cache entry中
790 0 : size_t cacheableSqeStartIdx = 0; // SQE start index (在动态缓存对应SQE数组中的索引)
791 0 : size_t bufferSqeStartIdx = 0; // SQE start index (在SQE ring buffer中的索引)
792 0 : for (size_t i = 0; i < placeholderIdxes.size(); ++i) {
793 : // [bufferSqeStartIdx, curPlaceholderIdx) -> [cacheableSqeStartIdx, cacheableSqeStartIdx + curPlaceholderIdx - bufferSqeStartIdx)
794 0 : const size_t curPlaceholderIdx = placeholderIdxes[i];
795 0 : HCCL_INFO("[DispatcherAicpu][LaunchTask] %uth placeholder copy dispatchedSqeArray[%u:%u) into cachedSqeArrays[%u][%u:%u)", i, bufferSqeStartIdx, curPlaceholderIdx, arrayIdx, cacheableSqeStartIdx, cacheableSqeStartIdx + curPlaceholderIdx - bufferSqeStartIdx);
796 0 : if (curPlaceholderIdx <= bufferSqeStartIdx) { // NO non-placeholder dispatched SQE to admit
797 : // NOTE: NO need to change cacheableSqeStartIdx
798 0 : bufferSqeStartIdx = curPlaceholderIdx + 1;
799 : } else {
800 0 : const size_t curSqeCount = curPlaceholderIdx - bufferSqeStartIdx;
801 0 : CHK_RET(entryPtr->MemcpySqeArray(arrayIdx, cacheableSqeStartIdx, curSqeCount,
802 : sqeArray + bufferSqeStartIdx * HCCL_SQE_SIZE,
803 : sqeTypeArray + bufferSqeStartIdx,
804 : sqeDfxInfoArray + bufferSqeStartIdx,
805 : isAlltoallv_, alltoallvMetadataPtr_
806 : ));
807 0 : cacheableSqeStartIdx += curSqeCount;
808 0 : bufferSqeStartIdx = curPlaceholderIdx + 1;
809 : }
810 : }
811 :
812 : // 存在剩余SQE, 即最后一个SQE不是placeholder
813 0 : if (LIKELY(bufferSqeStartIdx < cnt)) {
814 : // [bufferSqeStartIdx, cnt - 1] -> [cacheableSqeStartIdx, cacheableSqeStartIdx + cnt - bufferSqeStartIdx)
815 0 : const size_t curSqeCount = cnt - bufferSqeStartIdx;
816 0 : CHK_RET(entryPtr->MemcpySqeArray(arrayIdx, cacheableSqeStartIdx, curSqeCount,
817 : sqeArray + bufferSqeStartIdx * HCCL_SQE_SIZE,
818 : sqeTypeArray + bufferSqeStartIdx,
819 : sqeDfxInfoArray + bufferSqeStartIdx,
820 : isAlltoallv_, alltoallvMetadataPtr_
821 : ));
822 : }
823 0 : }
824 :
825 2 : CHK_RET(ConfigSqStatusByType(aicpuInfo_.devId, streamInfo.sqId, DRV_SQCQ_PROP_SQ_TAIL, newTail));
826 2 : tail = newTail;
827 2 : PLF_CONFIG_INFO(PLF_TASK,
828 : "%s success, sqid:%d, sqe_num:%u, curHead:%u, curtail:%u", __func__, streamInfo.sqId, cnt, head, tail);
829 2 : sqeContextBuffer->sqeCnt = 0;
830 2 : return HCCL_SUCCESS;
831 : }
832 :
833 8 : HcclResult DispatcherAiCpu::LaunchTasksEx(hccl::Stream &stream, std::vector<Stream> &subStreams)
834 : {
835 : /* 两阶段模式，主流待正式执行时再下 */
836 : /* 一阶段第一次，可以先下主流 */
837 8 : HcclResult ret = LaunchTask(stream, true);
838 8 : if (ret != HCCL_SUCCESS) {
839 0 : HCCL_ERROR("[DispatcherAiCpu][LaunchTasksEx] "\
840 : "launch task failed, sqid:%u, ret:%u", stream.sqId(), ret);
841 0 : return ret;
842 : }
843 :
844 8 : for (u32 index = 0; index < subStreams.size(); index++) {
845 0 : ret = LaunchTask(subStreams[index], true);
846 0 : if (ret != HCCL_SUCCESS) {
847 0 : HCCL_ERROR("[DispatcherAiCpu][LaunchTasksEx] "\
848 : "launch task failed, sqid:%u, ret:%u", subStreams[index].sqId(), ret);
849 0 : return ret;
850 : }
851 : }
852 :
853 8 : return HCCL_SUCCESS;
854 : }
855 :
856 3 : HcclResult DispatcherAiCpu::LaunchAllTasks()
857 : {
858 5 : for (auto it = streamMap_.begin(); it != streamMap_.end(); ++it) {
859 3 : if (it->second.IsInvalid()) { // 跳过已销毁的stream (streamMap_只增不删, 原stream销毁后副本通过shared invalid标志感知)
860 0 : HCCL_WARNING("[DispatcherAiCpu][LaunchAllTasks] skip invalid stream in streamMap, streamId:%d", it->first);
861 0 : continue;
862 : }
863 3 : HcclResult ret = LaunchTask(it->second, true);
864 3 : if (ret != HCCL_SUCCESS) {
865 1 : HCCL_ERROR("DispatcherAiCpu][LaunchAllTasks] launch task failed, sqid:%u, ret:%u", it->second.sqId(), ret);
866 1 : return ret;
867 : }
868 : }
869 2 : return HCCL_SUCCESS;
870 : }
871 :
872 1 : HcclResult DispatcherAiCpu::ReduceAsync(const void *src, void *dst, u64 dataCount, const HcclDataType datatype,
873 : HcclReduceOp redOp, Stream &stream, HcclReduceType reduceType)
874 : {
875 1 : return (reduceType == HcclReduceType::HCCL_INLINE_REDUCE) ?
876 1 : InlineReduceAsync(src, dataCount, datatype, redOp, stream, dst) :
877 1 : TbeReduceAsync(src, dst, dataCount, datatype, redOp, stream, dst);
878 : }
879 :
880 2 : HcclResult DispatcherAiCpu::InlineReduceAsync(const void *src, u64 dataCount, const HcclDataType datatype,
881 : HcclReduceOp redOp, hccl::Stream &stream, void *dst, u32 remoteUserRank, hccl::LinkType inLinkType)
882 : {
883 : // 参数有效性检查
884 2 : CHK_PTR_NULL(stream.ptr());
885 2 : if (dataCount == 0) {
886 0 : HCCL_INFO("%s src memory size is 0, not need inline reduce.", __func__);
887 0 : return HCCL_SUCCESS;
888 : }
889 2 : const HcclComStreamInfo &streamInfo = stream.GetHcclStreamInfo();
890 :
891 2 : aclDataType runtimeDataType = DT_MAP_TABLE[datatype];
892 2 : aclrtReduceKind rtReduceOp = RK_MAP_TABLE[redOp];
893 :
894 : // 将数据按4GB切分循环处理
895 2 : uint64_t spiltLoop = 0;
896 2 : uint64_t addr_offset = 0;
897 2 : uint64_t countSplit = 0;
898 2 : uint64_t countSize = dataCount * SIZE_TABLE[datatype];
899 2 : uint8_t *sqeBuffer = nullptr;
900 2 : uint8_t *sqeTypeAddr = nullptr;
901 2 : uint8_t *sqeDfxInfoAddr = nullptr;
902 2 : uint16_t taskId = 0U;
903 :
904 2 : if (countSize > HCCL_SDMA_MAX_COUNT_4GB) {
905 0 : spiltLoop = (countSize % HCCL_SDMA_MAX_COUNT_4GB) ? (countSize / HCCL_SDMA_MAX_COUNT_4GB) :
906 0 : ((countSize / HCCL_SDMA_MAX_COUNT_4GB) - 1);
907 0 : HCCL_INFO("%s InlineReduceAsync SDMA task countSize is bigger than 4GB"
908 : " and do segmentation splitloop:%llu", __func__, spiltLoop);
909 : }
910 2 : uint8_t linkType = static_cast<uint8_t>(inLinkType);
911 4 : for (uint64_t index = 0; index <= spiltLoop; index++) {
912 2 : addr_offset = index * HCCL_SDMA_MAX_COUNT_4GB;
913 2 : countSplit = (index == spiltLoop) ? (countSize - index * HCCL_SDMA_MAX_COUNT_4GB) : (HCCL_SDMA_MAX_COUNT_4GB);
914 2 : void *srcSplit = static_cast<void *>(static_cast<char *>(const_cast<void *>(src)) + addr_offset);
915 2 : void *dstSplit = static_cast<void *>(static_cast<char *>(dst) + addr_offset);
916 :
917 2 : CHK_RET(GetStreamSqeBufferAddr(stream, sqeBuffer, sqeTypeAddr, sqeDfxInfoAddr, taskId));
918 2 : AicpuDfxInfo * const dfxInfo = (AicpuDfxInfo * const)sqeDfxInfoAddr;
919 2 : dfxInfo->opRingBufferIdx = opRingBufferIdx_;
920 2 : dfxInfo->remoteRank = remoteUserRank;
921 2 : dfxInfo->notifyId = INVALID_VALUE_RANKID;
922 2 : addOneMemcpySqe_(streamInfo.actualStreamId, taskId, srcSplit, countSplit, runtimeDataType, rtReduceOp, dstSplit, 0,
923 : aicpuInfo_.ssid, aicpuInfo_.devId, aicpuInfo_.overflowAddr, linkType, sqeBuffer, sqeTypeAddr, hcclQos_);
924 :
925 2 : PLF_CONFIG_INFO(PLF_TASK,
926 : "%s para: linkType[%u] srcSplit[%p] dstSplit[%p] countSplit[%llu] taskId[%u] streamId[%u] remoteRank[%u] "\
927 : "rtDatatType[%d] rtReduceOp[%d]", __func__, linkType, srcSplit, dstSplit, countSplit, taskId,
928 : streamInfo.actualStreamId, remoteUserRank, runtimeDataType, rtReduceOp);
929 : }
930 :
931 2 : return HCCL_SUCCESS;
932 : }
933 :
934 1 : HcclResult DispatcherAiCpu::TbeReduceAsync(const void *src1, const void *src2, u64 count, const HcclDataType datatype,
935 : HcclReduceOp redOp, Stream &stream, const void *dst)
936 : {
937 1 : HCCL_ERROR("[DispatcherAiCpu][TbeReduceAsync] aicpu do not support the tbe reduce");
938 1 : return HCCL_E_NOT_SUPPORT;
939 : }
940 :
941 5 : HcclResult DispatcherAiCpu::RdmaSend(u32 dbindex, u64 dbinfo, hccl::Stream &stream, RdmaTaskInfo &taskInfo)
942 : {
943 5 : const HcclComStreamInfo &streamInfo = stream.GetHcclStreamInfo();
944 :
945 5 : uint8_t *sqeBuffer = nullptr;
946 5 : uint8_t *sqeTypeAddr = nullptr;
947 5 : uint8_t *sqeDfxInfoAddr = nullptr;
948 5 : uint16_t taskId = 0U;
949 :
950 5 : CHK_RET(GetStreamSqeBufferAddr(stream, sqeBuffer, sqeTypeAddr, sqeDfxInfoAddr, taskId));
951 5 : AicpuDfxInfo * const dfxInfo = (AicpuDfxInfo * const)sqeDfxInfoAddr;
952 5 : dfxInfo->opRingBufferIdx = opRingBufferIdx_;
953 5 : dfxInfo->remoteRank = taskInfo.remoteRank;
954 5 : dfxInfo->notifyId = INVALID_UINT; // 多个wr只敲一次doorbell的情况下，一般只会有一个notify
955 :
956 5 : uint32_t wrLen = 0; // 统计wr的总数据量
957 5 : for (const WrInformation& wr : taskInfo.wrInfos) {
958 0 : wrLen += wr.wrData.memList.len;
959 0 : dfxInfo->notifyId = (wr.notifyId != INVALID_UINT) ? wr.notifyId : dfxInfo->notifyId;
960 : }
961 :
962 5 : u64 dbAddr = CalcDbAddr(dbindex);
963 5 : addOneRdmaDbSendSqe_(streamInfo.actualStreamId, taskId, dbinfo, dbAddr, wrLen,
964 5 : static_cast<uint8_t>(taskInfo.rdmaType), sqeBuffer, sqeTypeAddr);
965 :
966 5 : PLF_CONFIG_INFO(PLF_TASK,
967 : "%s para: streamId[%u] taskId[%u] remoteRank[%u] RdmaType[%d] wrLen[%u] notifyId[%u]",
968 : __func__, streamInfo.actualStreamId, taskId, taskInfo.remoteRank, taskInfo.rdmaType, wrLen, dfxInfo->notifyId);
969 :
970 5 : return HCCL_SUCCESS;
971 : }
972 :
973 0 : HcclResult DispatcherAiCpu::RdmaRecord(u32 dbindex, u64 dbinfo, const struct SendWr &wr, hccl::Stream &stream,
974 : RdmaType rdmaType, u32 userRank, u64 offset, u32 notifyId)
975 : {
976 0 : return HCCL_SUCCESS;
977 : }
978 :
979 13 : HcclResult DispatcherAiCpu::GetStreamSqeBufferAddr(hccl::Stream &stream, uint8_t *&sqeBufferAddr, uint8_t *&sqeTypeAddr,
980 : uint8_t *&sqeDfxInfoAddr, uint16_t &taskId)
981 : {
982 13 : SaveStreamInfo(stream);
983 13 : HcclSqeContext* sqeContext = stream.GetSqeContextPtr();
984 13 : CHK_PTR_NULL(sqeContext);
985 13 : if (UNLIKELY(sqeContext->buffer.sqeCnt >= HCCL_PER_LAUNCH_SQE_CNT)) {
986 0 : HCCL_INFO("GetStreamSqeBufferAddr tailSqeIdx[%u], try to launchTask", sqeContext->buffer.tailSqeIdx);
987 0 : CHK_RET(LaunchTask(stream, true));
988 : }
989 13 : if (UNLIKELY(sqeContext->buffer.tailSqeIdx >= HCCL_SQE_MAX_CNT)) {
990 0 : CHK_RET(LaunchTask(stream, true));
991 :
992 0 : if (callback_ != nullptr) {
993 0 : hccl::AiCPUStreamTasks para(stream.id(), reinterpret_cast<void*>(sqeContext));
994 0 : hccl::TaskPara taskPara(TaskType::TASK_BATCH_REPORT, para);
995 0 : callback_(callBackUserPtr_, (void *)&taskPara, sizeof(struct TaskPara));
996 0 : }
997 : }
998 13 : SqeRingBuffer *sqeContextBuffer = &(sqeContext->buffer);
999 13 : uint16_t flipNum = sqeContextBuffer->filpNum;
1000 13 : uint16_t nextTaskId = sqeContextBuffer->tailSqeTaskId;
1001 : // nextTaskId=0的时候下发PlaceHolder
1002 13 : if (UNLIKELY(nextTaskId == 0 && flipNum != 0)) {
1003 0 : CHK_RET(AddFlipTask(stream));
1004 : }
1005 13 : if (UNLIKELY(sqeContext->buffer.tailSqeIdx >= HCCL_SQE_MAX_CNT)) {
1006 0 : CHK_RET(LaunchTask(stream, true));
1007 :
1008 0 : if (callback_ != nullptr) {
1009 0 : hccl::AiCPUStreamTasks para(stream.id(), reinterpret_cast<void*>(sqeContext));
1010 0 : hccl::TaskPara taskPara(TaskType::TASK_BATCH_REPORT, para);
1011 0 : callback_(callBackUserPtr_, (void *)&taskPara, sizeof(struct TaskPara));
1012 0 : }
1013 : }
1014 13 : CHK_RET(stream.GetNextSqeBufferAddr(sqeBufferAddr, sqeTypeAddr, sqeDfxInfoAddr, taskId));
1015 13 : return HCCL_SUCCESS;
1016 : }
1017 :
1018 2 : HcclResult DispatcherAiCpu::WaitRtsq(Stream& stream, const size_t& sqeCount, const bool isBlockLaunch) {
1019 : // 注意: 目前WaitRtsq不会被递归调用, 所以isBlockLaunch永远为true; 为防止以后LaunchTask递归使用WaitRtsq, 编码时考虑isBlockLaunch为false的情况
1020 :
1021 : // 检验入参
1022 2 : const HcclComStreamInfo &streamInfo = stream.GetHcclStreamInfo();
1023 2 : if (sqeCount == 0) {
1024 0 : CHK_PRT_CONT(isBlockLaunch,
1025 : HCCL_DEBUG("[DispatcherAiCpu][WaitRtsq] no sqe, streamId:%d, sqId:%u", streamInfo.actualStreamId, streamInfo.sqId));
1026 0 : return HCCL_SUCCESS;
1027 2 : } else if (sqeCount > streamInfo.sqDepth) {
1028 0 : HCCL_ERROR("[DispatcherAiCpu][WaitRtsq] sqeCount %u should be smaller than sqDepth %u]", sqeCount, streamInfo.sqDepth);
1029 0 : return HCCL_E_PTR;
1030 : }
1031 :
1032 : // Get head and tail of RTSQ ring buffer
1033 2 : HcclSqeContext *sqeContext = stream.GetSqeContextPtr();
1034 2 : CHK_PTR_NULL(sqeContext);
1035 2 : SqeRingBuffer *sqeContextBuffer = &(sqeContext->buffer);
1036 2 : CHK_PTR_NULL(sqeContextBuffer);
1037 2 : uint32_t& head = sqeContextBuffer->sqHead;
1038 2 : uint32_t& tail = sqeContextBuffer->sqTail;
1039 :
1040 : // Dump debug information
1041 2 : const uint32_t newTail = (tail + sqeCount) % streamInfo.sqDepth;
1042 : // 仅在阻塞下发场景打印，避免非阻塞场景调用时刷屏
1043 2 : CHK_PRT_CONT(isBlockLaunch,
1044 : HCCL_INFO("[DispatcherAicpu][WaitRtsq] sqid:%d sqeCount:%u head:%u curtail:%u newTail:%u", streamInfo.sqId, sqeCount, head, tail, newTail));
1045 :
1046 : // 轮询RTSQ直至获得足够大的剩余空间
1047 2 : u64 startUsec = GetCurAicpuTimestamp();
1048 2 : u64 lastUsec = startUsec;
1049 3 : while (((tail < head ? streamInfo.sqDepth : 0U) + tail - head + sqeCount >= streamInfo.sqDepth) && (tail != head)) { // 判断RTSQ中剩余sqe空间是否足够下发
1050 : // 需要放在while循环进来后第一个执行 (获取最新的RTSQ head, 查看RTSQ的消费进度)
1051 2 : CHK_RET(QuerySqStatusByType(aicpuInfo_.devId, streamInfo.sqId, DRV_SQCQ_PROP_SQ_HEAD, head));
1052 :
1053 : // 非阻塞下发场景，rtsq队列空间不足时直接返回
1054 2 : if (isBlockLaunch == false) {
1055 1 : return HCCL_SUCCESS;
1056 : }
1057 :
1058 : // 当前流无法下发，把其他流都launch一遍，避免等待的其他流没有launch
1059 2 : for (auto it = streamMap_.begin(); it != streamMap_.end(); ++it) {
1060 1 : if (it->first != streamInfo.actualStreamId) { // 不是当前stream
1061 1 : if (it->second.IsInvalid()) { // 跳过已销毁的stream (streamMap_只增不删, 原stream销毁后副本通过shared invalid标志感知)
1062 0 : HCCL_WARNING("[DispatcherAiCpu][WaitRtsq] skip invalid stream in streamMap, streamId:%d", it->first);
1063 0 : continue;
1064 : }
1065 1 : CHK_RET(LaunchTask(it->second, false)); // 非阻塞launch
1066 : }
1067 : }
1068 :
1069 : // 等待超时
1070 1 : u64 curUsec = GetCurAicpuTimestamp();
1071 1 : if (dfxTimeOutConfig_.sqFullWaitTimeOut != 0 &&
1072 0 : (curUsec - startUsec > NANOSECOND_TO_SECOND * dfxTimeOutConfig_.sqFullWaitTimeOut)) {
1073 0 : HCCL_ERROR("[DispatcherAicpu][WaitRtsq] Rtsq full, timeout %lus. curhead:%u, sqId:%d", dfxTimeOutConfig_.sqFullWaitTimeOut, head, streamInfo.sqId);
1074 0 : return HCCL_E_AGAIN;
1075 : }
1076 :
1077 : // 等待下发阶段，每隔30s打印一次状态
1078 1 : if (curUsec - lastUsec > NANOSECOND_TO_SECOND * dfx::kPrintSqInterval) {
1079 0 : lastUsec = curUsec;
1080 0 : HCCL_RUN_INFO("[DispatcherAicpu][WaitRtsq] Current state. sqid:%d, head:%u, tail:%u, sqeCount:%u",
1081 : streamInfo.sqId, head, tail, sqeCount);
1082 : }
1083 :
1084 : // 等待下发过程中出现cqe异常, 需要终止当前算子SQE的下发过程
1085 1 : if (checkOpExecStatusCallback_ != nullptr) {
1086 0 : HcclResult opExecStatus = checkOpExecStatusCallback_();
1087 0 : CHK_PRT_RET(opExecStatus != HCCL_SUCCESS,
1088 : HCCL_ERROR("[DispatcherAicpu][WaitRtsq] hccl aicpu stop launch for task exception or stop command, ret:%d", opExecStatus),
1089 : opExecStatus);
1090 : }
1091 : }
1092 :
1093 1 : return HCCL_SUCCESS;
1094 : }
1095 :
1096 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) {
1097 : // 检验入参
1098 0 : const HcclComStreamInfo &streamInfo = stream.GetHcclStreamInfo();
1099 0 : if (sqeCount == 0) {
1100 0 : HCCL_DEBUG("[DispatcherAiCpu][MemcpyRtsq] no sqe, streamId:%d, sqId:%u", streamInfo.actualStreamId, streamInfo.sqId);
1101 0 : return HCCL_SUCCESS;
1102 0 : } else if (sqeCount > streamInfo.sqDepth) {
1103 0 : HCCL_ERROR("[DispatcherAiCpu][MemcpyRtsq] sqeCount %u should be smaller than sqDepth %u]", sqeCount, streamInfo.sqDepth);
1104 0 : return HCCL_E_PTR;
1105 : }
1106 0 : CHK_PTR_NULL(sqeArray);
1107 0 : CHK_PTR_NULL(sqeTypeArray);
1108 0 : CHK_PTR_NULL(sqeDfxInfoArray);
1109 0 : if (profL1Enable) {
1110 : // 会访问profTimestamps[profTimestampStartIdx, profTimestampStartIdx + sqeCount - 1]
1111 0 : CHK_PRT_RET(profTimestamps.size() == 0, HCCL_ERROR("[DispatcherAiCpu][MemcpyRtsq] empty profTimestamps"), HCCL_E_INTERNAL);
1112 0 : CHK_PRT_RET(profTimestampStartIdx >= profTimestamps.size(), HCCL_ERROR("[DispatcherAiCpu][MemcpyRtsq] profTimestampStartIdx[%u] >= profTimestamps.size[%u]", profTimestampStartIdx, profTimestamps.size()), HCCL_E_INTERNAL);
1113 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);
1114 : }
1115 :
1116 : // 获得RTSQ的head和tail
1117 0 : HcclSqeContext *sqeContext = stream.GetSqeContextPtr();
1118 0 : CHK_PTR_NULL(sqeContext);
1119 0 : SqeRingBuffer *sqeContextBuffer = &(sqeContext->buffer);
1120 0 : CHK_PTR_NULL(sqeContextBuffer);
1121 0 : uint32_t& head = sqeContextBuffer->sqHead;
1122 0 : uint32_t& tail = sqeContextBuffer->sqTail;
1123 :
1124 : // Dump debug information
1125 0 : const uint32_t newTail = (tail + sqeCount) % streamInfo.sqDepth;
1126 0 : HCCL_INFO("[DispatcherAicpu][MemcpyRtsq] before memcpy, sqid:%d sqeCount:%u head:%u curtail:%u newTail:%u", streamInfo.sqId, sqeCount, head, tail, newTail);
1127 :
1128 : // 准备memcpy中目的末端基地址 (RTSQ从tail开始拷贝, [head, tail)为待执行SQE)
1129 0 : uint8_t *rtsqSqeTailBaseAddr = reinterpret_cast<uint8_t *>(streamInfo.sqBaseAddr) + tail * HCCL_SQE_SIZE;
1130 0 : uint8_t *mirrorRtsqSqeTailBaseAddr = sqeContextBuffer->rtsMirrorBuffer + tail * HCCL_SQE_SIZE;
1131 0 : uint8_t *rtsqSqeTypeTailBaseAddr = sqeContextBuffer->rtsqSqeType + tail;
1132 0 : AicpuDfxInfo *rtsqDfxInfoTailBaseAddr = sqeContextBuffer->rtsDfxInfo + tail;
1133 :
1134 0 : uint32_t tailLeft = streamInfo.sqDepth - tail; // RTSQ tail到buffer末端的剩余空间 (不包括buffer前端到head的剩余空间)
1135 0 : HCCL_INFO("[DispatcherAicpu][MemcpyRtsq] cpy sqe, tailLeft:%u, sqeCount:%u, streamId:%u", tailLeft, sqeCount, stream.id());
1136 0 : if (sqeCount <= tailLeft) { // buffer末端剩余空间放得下新增sqe
1137 : // 向buffer末端拷贝sqeCount个SQE信息
1138 :
1139 : // 拷贝SQE内容到RTSQ
1140 0 : CHK_SAFETY_FUNC_RET(memcpy_s(rtsqSqeTailBaseAddr, tailLeft * HCCL_SQE_SIZE, sqeArray, sqeCount * HCCL_SQE_SIZE));
1141 :
1142 : // 拷贝SQE内容到RTSQ mirror
1143 0 : CHK_SAFETY_FUNC_RET(memcpy_s(mirrorRtsqSqeTailBaseAddr, tailLeft * HCCL_SQE_SIZE, sqeArray, sqeCount * HCCL_SQE_SIZE));
1144 :
1145 : // 拷贝SQE类型
1146 0 : CHK_SAFETY_FUNC_RET(memcpy_s(rtsqSqeTypeTailBaseAddr, tailLeft, sqeTypeArray, sqeCount));
1147 :
1148 : // 拷贝SQE DfxInfo
1149 0 : CHK_SAFETY_FUNC_RET(memcpy_s(rtsqDfxInfoTailBaseAddr, tailLeft * sizeof(AicpuDfxInfo), sqeDfxInfoArray, sqeCount * sizeof(AicpuDfxInfo)));
1150 : } else { // 需要buffer末端和首端的剩余空间
1151 : // 先向buffer末端拷贝tailLeft个SQE信息, 再向buffer首端拷贝sqeCount-tailLeft个SQE信息
1152 :
1153 : // 拷贝SQE内容到RTSQ
1154 0 : CHK_SAFETY_FUNC_RET(memcpy_s(rtsqSqeTailBaseAddr, tailLeft * HCCL_SQE_SIZE, sqeArray, tailLeft * HCCL_SQE_SIZE));
1155 0 : CHK_SAFETY_FUNC_RET(memcpy_s(reinterpret_cast<uint8_t *>(streamInfo.sqBaseAddr),
1156 : streamInfo.sqDepth * HCCL_SQE_SIZE,
1157 : sqeArray + tailLeft * HCCL_SQE_SIZE,
1158 : (sqeCount - tailLeft) * HCCL_SQE_SIZE));
1159 :
1160 : // 拷贝SQE内容到RTSQ mirror
1161 0 : CHK_SAFETY_FUNC_RET(memcpy_s(mirrorRtsqSqeTailBaseAddr, tailLeft * HCCL_SQE_SIZE, sqeArray, tailLeft * HCCL_SQE_SIZE));
1162 0 : CHK_SAFETY_FUNC_RET(memcpy_s(sqeContextBuffer->rtsMirrorBuffer,
1163 : streamInfo.sqDepth * HCCL_SQE_SIZE,
1164 : sqeArray + tailLeft * HCCL_SQE_SIZE,
1165 : (sqeCount - tailLeft) * HCCL_SQE_SIZE));
1166 :
1167 : // 拷贝SQE type
1168 0 : CHK_SAFETY_FUNC_RET(memcpy_s(rtsqSqeTypeTailBaseAddr, tailLeft, sqeTypeArray, tailLeft));
1169 0 : CHK_SAFETY_FUNC_RET(memcpy_s(sqeContextBuffer->rtsqSqeType, streamInfo.sqDepth, sqeTypeArray + tailLeft, (sqeCount - tailLeft)));
1170 :
1171 : // 拷贝SQE DfxInfo
1172 0 : CHK_SAFETY_FUNC_RET(memcpy_s(rtsqDfxInfoTailBaseAddr,
1173 : tailLeft * sizeof(AicpuDfxInfo), sqeDfxInfoArray, tailLeft * sizeof(AicpuDfxInfo)));
1174 0 : CHK_SAFETY_FUNC_RET(memcpy_s(sqeContextBuffer->rtsDfxInfo, streamInfo.sqDepth * sizeof(AicpuDfxInfo),
1175 : sqeDfxInfoArray + tailLeft, (sqeCount - tailLeft) * sizeof(AicpuDfxInfo)));
1176 : }
1177 :
1178 : // 更新RTSQ ring buffer的tail
1179 0 : CHK_RET(ConfigSqStatusByType(aicpuInfo_.devId, streamInfo.sqId, DRV_SQCQ_PROP_SQ_TAIL, newTail));
1180 0 : tail = newTail;
1181 0 : PLF_CONFIG_INFO(PLF_TASK,
1182 : "%s success, sqid:%d, sqe_num:%u, curHead:%u, curtail:%u", __func__, streamInfo.sqId, sqeCount, head, tail);
1183 :
1184 : // 上报profiling信息
1185 0 : if (profL1Enable) {
1186 : // Cache hit下SQE ring buffer中[tail-cnt, tail)为待下发SQE, 其数量一定为0 (否则更新tailSqeIdx后, 会导致待下发SQE未下发, 而已下发SQE重复下发)
1187 0 : CHK_PRT_RET(sqeContextBuffer->sqeCnt != 0, HCCL_ERROR("[DispatcherAicpu][MemcpyRtsq] sqeContextBuffer->sqeCnt[%u] should be zero!", sqeContextBuffer->sqeCnt), HCCL_E_INTERNAL);
1188 :
1189 : // 上报flip placeholder的profiling信息
1190 0 : if ((*sqeTypeArray) == SqeType::FLIP_PLACEHOLDER_SQE) {
1191 0 : CHK_PRT_RET(sqeCount != 1, HCCL_ERROR("[DispatcherAicpu][MemcpyRtsq] sqeCount[%u] should be 1!", sqeCount), HCCL_E_INTERNAL);
1192 :
1193 : // 注意: 参考AddFlipTask, 先上报flip task (提醒profiling翻转taskid), 之后再拷贝到SQE ring buffer (捕捉placeholder SQE相关的profiling信息)
1194 : // 注意: ProfilingManager::TaskProfilingCallBack->ReportFilpTask不会扫描SQE ring buffer, 也不会更新其中的streamToSqeIdxMap_
1195 0 : if (callback_ != nullptr) {
1196 0 : const rtStarsPlaceHolderSqe_t *placeholderSqePtr = reinterpret_cast<const rtStarsPlaceHolderSqe_t *>(sqeArray);
1197 0 : hccl::FlipTaskPara para(stream.id(), placeholderSqePtr->header.taskId, placeholderSqePtr->u.flip_task_info.flipNumReport);
1198 0 : hccl::TaskPara taskPara(TaskType::TASK_FLIP, para);
1199 0 : callback_(callBackUserPtr_, (void *)&taskPara, sizeof(struct TaskPara));
1200 0 : }
1201 : }
1202 :
1203 : // 循环拷贝cached SQE到SQE ring buffer中
1204 0 : size_t reportSqeCount = 0;
1205 0 : while (reportSqeCount < sqeCount) {
1206 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);
1207 0 : const size_t sqeTailLeft = HCCL_SQE_MAX_CNT - sqeContextBuffer->tailSqeIdx;
1208 0 : if (sqeTailLeft > 0) {
1209 : // 准备profiling上报的目的末端基地址 (SQE ring buffer从tail开始拷贝)
1210 0 : uint8_t *sqeLocalBuffTailBaseAddr = sqeContextBuffer->localBuff + sqeContextBuffer->tailSqeIdx * HCCL_SQE_SIZE;
1211 0 : uint8_t *sqeTypeTailBaseAddr = sqeContextBuffer->sqeType + sqeContextBuffer->tailSqeIdx;
1212 0 : AicpuDfxInfo *dfxInfoTailBaseAddr = sqeContextBuffer->dfxInfo + sqeContextBuffer->tailSqeIdx;
1213 0 : uint64_t *profTimestapTailBaseAddr = sqeContextBuffer->profTimestap + sqeContextBuffer->tailSqeIdx;
1214 :
1215 : // 向SQE ring buffer末端拷贝SQE信息[reportSqeCount, reportSqeCount + tmpSqeCount - 1] (只用于profiling上报, 不会下发)
1216 0 : const size_t tmpSqeCount = std::min(sqeCount - reportSqeCount, sqeTailLeft);
1217 0 : HCCL_INFO("[DispatcherAicpu][MemcpyRtsq] report sqe profiling, sqeCount[%u] reportSqeCount[%u] tailSqeIdx[%u] sqeTailLeft[%u] tmpSqeCount[%u]", sqeCount, reportSqeCount, sqeContextBuffer->tailSqeIdx, sqeTailLeft, tmpSqeCount);
1218 :
1219 : // 拷贝SQE内容
1220 0 : CHK_SAFETY_FUNC_RET(memcpy_s(sqeLocalBuffTailBaseAddr, sqeTailLeft * HCCL_SQE_SIZE, sqeArray + reportSqeCount * HCCL_SQE_SIZE, tmpSqeCount * HCCL_SQE_SIZE));
1221 :
1222 : // 拷贝SQE类型
1223 0 : CHK_SAFETY_FUNC_RET(memcpy_s(sqeTypeTailBaseAddr, sqeTailLeft, sqeTypeArray + reportSqeCount, tmpSqeCount));
1224 :
1225 : // 拷贝SQE DfxInfo
1226 0 : CHK_SAFETY_FUNC_RET(memcpy_s(dfxInfoTailBaseAddr, sqeTailLeft * sizeof(AicpuDfxInfo), sqeDfxInfoArray + reportSqeCount, tmpSqeCount * sizeof(AicpuDfxInfo)));
1227 :
1228 : // 拷贝SQE timestamp
1229 0 : CHK_SAFETY_FUNC_RET(memcpy_s(profTimestapTailBaseAddr, sqeTailLeft * sizeof(uint64_t), profTimestamps.data() + profTimestampStartIdx + reportSqeCount, tmpSqeCount * sizeof(uint64_t)));
1230 :
1231 : // 注意: sqeContextBuffer->sqeCnt不更新, 仍然为0 (即拷贝的SQE信息为已下发待上报), 避免SQE重复下发
1232 0 : sqeContextBuffer->tailSqeIdx += static_cast<uint16_t>(tmpSqeCount);
1233 0 : reportSqeCount += tmpSqeCount;
1234 : } else {
1235 : // 注意: SQE ring buffer中待下发SQE数量一定为0, 不需要调用LaunchTask将待下发变成已下发待上报, 可以直接上报profiling将已下发待上报变成已上报
1236 : // 注意: 调用后, ProfilingManager::StartReportSqeIdx为HCCL_SQE_MAX_CNT
1237 0 : if (callback_ != nullptr) {
1238 0 : hccl::AiCPUStreamTasks para(stream.id(), reinterpret_cast<void*>(sqeContext));
1239 0 : hccl::TaskPara taskPara(TaskType::TASK_BATCH_REPORT, para);
1240 0 : callback_(callBackUserPtr_, (void *)&taskPara, sizeof(struct TaskPara));
1241 0 : }
1242 :
1243 : // SQE ring buffer中所有SQE均为已上报 -> 清理SQE ring buffer
1244 0 : HCCL_INFO("[DispatcherAicpu][MemcpyRtsq] Sqe index to %u, need clear", HCCL_SQE_MAX_CNT);
1245 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);
1246 0 : CHK_RET(stream.ClearLocalBuff()); // 会将stream.sqeContextBuffer中的sqeCnt和tailSqeIdx设置为0
1247 0 : CHK_PRT_RET(sqeContextBuffer->sqeCnt != 0, HCCL_ERROR("[DispatcherAicpu][MemcpyRtsq] sqeCnt[%u] should be 0 after clear", sqeContextBuffer->sqeCnt), HCCL_E_INTERNAL);
1248 0 : CHK_PRT_RET(sqeContextBuffer->tailSqeIdx != 0, HCCL_ERROR("[DispatcherAicpu][MemcpyRtsq] tailSqeIdx[%u] should be 0 after clear", sqeContextBuffer->tailSqeIdx), HCCL_E_INTERNAL);
1249 :
1250 : // 参考HcclCommAicpu::ClearLocalBuff, 调用Stream::ClearLocalBuff后, 应该调用ProfilingManager::UpdateStartReportSqeIdx手动将ProfilingManager::StartReportSqeIdx设置为0
1251 : // 注意: 由于platform暂未将UpdateStartReportSqeIdx作为回调函数传入, 无法直接调用此framework函数 -> 通过callback_ (即ProfilingManager::TaskProfilingCallBack) 间接将ProfilingManager::StartReportSqeIdx设置为0
1252 : // 注意: 由于调用前ProfilingManager::StartReportSqeIdx为HCCL_SQE_MAX_CNT, tailSqeIdx为0, 即从startIdx=HCCL_SQE_MAX_CNT到endIdx=0, ProfilingManager不会进入profiling上报代码, 而是只会调用UpdateStartReportSqeIdx设置StartReportSqeIdx为0
1253 0 : if (callback_ != nullptr) {
1254 0 : HCCL_INFO("[DispatcherAicpu][MemcpyRtsq] re-invoke callback_ to reset StartReportSqeIdx as 0 in ProfilingManager");
1255 :
1256 0 : hccl::AiCPUStreamTasks para(stream.id(), reinterpret_cast<void*>(sqeContext));
1257 0 : hccl::TaskPara taskPara(TaskType::TASK_BATCH_REPORT, para);
1258 0 : callback_(callBackUserPtr_, (void *)&taskPara, sizeof(struct TaskPara));
1259 0 : }
1260 : }
1261 : }
1262 : }
1263 :
1264 0 : return HCCL_SUCCESS;
1265 : }
1266 :
1267 0 : HcclResult DispatcherAiCpu::AddFlipTask(Stream &stream)
1268 : {
1269 0 : HcclSqeContext *sqeContext = stream.GetSqeContextPtr();
1270 0 : CHK_PTR_NULL(sqeContext);
1271 0 : SqeRingBuffer *sqeContextBuffer = &(sqeContext->buffer);
1272 0 : CHK_PTR_NULL(sqeContextBuffer);
1273 0 : uint16_t flipNum = sqeContextBuffer->filpNum;
1274 0 : uint16_t taskId = sqeContextBuffer->tailSqeTaskId;
1275 :
1276 0 : if (callback_ != nullptr) {
1277 0 : hccl::FlipTaskPara para(stream.id(), taskId, flipNum);
1278 0 : hccl::TaskPara taskPara(TaskType::TASK_FLIP, para);
1279 0 : callback_(callBackUserPtr_, (void *)&taskPara, sizeof(struct TaskPara));
1280 0 : }
1281 :
1282 0 : const HcclComStreamInfo &streamInfo = stream.GetHcclStreamInfo();
1283 :
1284 0 : uint8_t *sqeBufferAddr = nullptr;
1285 0 : uint8_t *sqeTypeAddr = nullptr;
1286 0 : uint8_t *sqeDfxInfoAddr = nullptr;
1287 0 : CHK_RET(stream.GetNextSqeBufferAddr(sqeBufferAddr, sqeTypeAddr, sqeDfxInfoAddr, taskId));
1288 :
1289 0 : AicpuDfxInfo * const dfxInfo = (AicpuDfxInfo * const)sqeDfxInfoAddr;
1290 0 : dfxInfo->opRingBufferIdx = opRingBufferIdx_;
1291 0 : dfxInfo->remoteRank = INVALID_VALUE_RANKID;
1292 0 : dfxInfo->notifyId = INVALID_VALUE_RANKID;
1293 0 : addOneFlipPlaceHolderSqe_(streamInfo.actualStreamId, flipNum, taskId, sqeBufferAddr, sqeTypeAddr);
1294 :
1295 0 : PLF_CONFIG_INFO(PLF_TASK,
1296 : "%s para: taskId[%u] streamId[%u] flipNum[%u]", __func__, taskId, streamInfo.actualStreamId, flipNum);
1297 0 : return HCCL_SUCCESS;
1298 : }
1299 :
1300 0 : HcclResult DispatcherAiCpu::AddRetryPreamble(Stream &stream)
1301 : {
1302 0 : return AddFlipTask(stream);
1303 : }
1304 :
1305 22 : void DispatcherAiCpu::SaveStreamInfo(hccl::Stream &stream)
1306 : {
1307 22 : const HcclComStreamInfo &streamInfo = stream.GetHcclStreamInfo();
1308 22 : auto it = streamMap_.find(streamInfo.actualStreamId);
1309 22 : if (it == streamMap_.end()) {
1310 17 : streamMap_.insert({streamInfo.actualStreamId, stream});
1311 17 : HCCL_INFO("[DispatcherAiCpu][SaveStreamInfo] stream id[%d]", streamInfo.actualStreamId);
1312 5 : } else if (it->second.IsInvalid()) {
1313 : // stream id复用: 旧stream已销毁(IsInvalid=true), 用新stream覆盖, 否则dispatcher会一直跳过该id
1314 0 : it->second = stream;
1315 0 : HCCL_INFO("[DispatcherAiCpu][SaveStreamInfo] refresh invalidated stream id stream id[%d]", streamInfo.actualStreamId);
1316 : }
1317 44 : return;
1318 : }
1319 :
1320 2 : HcclResult DispatcherAiCpu::StreamSync(Stream &stream)
1321 : {
1322 2 : uint32_t head = 0;
1323 2 : uint32_t tail = 0;
1324 : const HcclComStreamInfo *streamInfo;
1325 2 : u64 startUsec = GetCurAicpuTimestamp();
1326 2 : u64 lastUsec = startUsec;
1327 2 : CHK_RET(stream.GetStreamInfo(streamInfo));
1328 :
1329 2 : CHK_RET(QuerySqStatusByType(aicpuInfo_.devId, streamInfo->sqId, DRV_SQCQ_PROP_SQ_TAIL, tail));
1330 2 : HCCL_INFO("StreamSync aicpu stream sqid[%d] tail[%u]", streamInfo->sqId, tail);
1331 : do {
1332 2 : CHK_RET(QuerySqStatusByType(aicpuInfo_.devId, streamInfo->sqId, DRV_SQCQ_PROP_SQ_HEAD, head));
1333 2 : u64 curUsec = GetCurAicpuTimestamp();
1334 2 : if (curUsec - startUsec > NANOSECOND_TO_SECOND * dfxTimeOutConfig_.sqeTimeOutTimeOut) {
1335 1 : HCCL_ERROR("stream sync timeout %lus. curhead:%u, curtall:%u, sqId:%d",
1336 : dfxTimeOutConfig_.sqeTimeOutTimeOut, head, tail, streamInfo->sqId);
1337 1 : return HCCL_E_TIMEOUT;
1338 : }
1339 :
1340 : // 等待下发阶段，每隔30s打印一次状态
1341 1 : if (curUsec - lastUsec > NANOSECOND_TO_SECOND * dfx::kPrintSqInterval) {
1342 0 : lastUsec = curUsec;
1343 0 : HCCL_RUN_INFO("[StreamSync]Current state. sqid:%d, head:%u, tail:%u",
1344 : streamInfo->sqId, head, tail);
1345 : }
1346 1 : } while (head != tail);
1347 :
1348 1 : return HCCL_SUCCESS;
1349 : }
1350 :
1351 5 : u64 DispatcherAiCpu::CalcDbAddr(u32 dbindex)
1352 : {
1353 5 : u64 dbAddr = 0;
1354 5 : if (aicpuInfo_.devType == DevType::DEV_TYPE_910_93) {
1355 : // 910_93 HCCS_SW 组网
1356 2 : constexpr u64 roceBaseAddr = 0x202000000000ULL;
1357 2 : constexpr u64 roceVfDbCfg0Reg = 0x230ULL;
1358 2 : constexpr u64 chipAddrOffset = 0x20000000000ULL;
1359 2 : constexpr u64 dieAddrOffset = 0x10000000000ULL;
1360 2 : constexpr u32 dbDieIdMask = 0x00ff0000;
1361 2 : constexpr u32 dbDieIdShift = 16; // 16 is dbDieIdShift
1362 2 : dbAddr = roceBaseAddr + roceVfDbCfg0Reg + chipAddrOffset * aicpuInfo_.chipId +
1363 2 : dieAddrOffset * ((dbindex & dbDieIdMask) >> dbDieIdShift);
1364 : } else {
1365 3 : constexpr u64 roceBaseAddr = 0x2000000000ULL;
1366 3 : constexpr u64 roceVfDbCfg0Reg = 0x230ULL;
1367 3 : constexpr u64 chipAddrOffset = 0x80000000000ULL;
1368 3 : constexpr u64 dieAddrOffset = 0x10000000000ULL;
1369 3 : constexpr u32 dbDieIdMask = 0x00ff0000;
1370 3 : constexpr u32 dbDieIdShift = 16; // 16 is dbDieIdShift
1371 3 : dbAddr = roceBaseAddr + roceVfDbCfg0Reg + chipAddrOffset * aicpuInfo_.chipId +
1372 3 : dieAddrOffset * ((dbindex & dbDieIdMask) >> dbDieIdShift);
1373 : }
1374 :
1375 5 : HCCL_DEBUG("%s dbindex:%u, devType:%u, chipId:%lld, dbAddr:%llu",
1376 : __func__, dbindex, aicpuInfo_.devType, aicpuInfo_.chipId, dbAddr);
1377 5 : return dbAddr;
1378 : }
1379 :
1380 444 : void DispatcherAiCpu::InitTimeOutConfig()
1381 : {
1382 444 : dfxTimeOutConfig_.useCredit = false;
1383 444 : dfxTimeOutConfig_.sqeTimeOutTimeOut = GetMaxNotifyWaitTime();
1384 444 : dfxTimeOutConfig_.sqeCreditTimeOut = RT_STARS_NEVER_TIMEOUT_KERNEL_CREDIT;
1385 444 : dfxTimeOutConfig_.sqeWaitTimeOut = dfx::kKfcTimeOut;
1386 444 : dfxTimeOutConfig_.sqFullWaitTimeOut = dfx::kSqFullWaitTimeOut;
1387 444 : HCCL_INFO("[DispatcherAiCpu][InitTimeOutConfig]DFX timeout config init successfully with details: [%s]",
1388 : dfxTimeOutConfig_.ToString().c_str());
1389 444 : }
1390 : } // namespace hccl
|
