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