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