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 451 : DispatcherAiCpu::DispatcherAiCpu(const u32 devPhyId) // deprecated
60 451 : : DispatcherPub(INVALID_INT)
61 : {
62 451 : aicpuInfo_.devId = devPhyId;
63 451 : }
64 :
65 902 : DispatcherAiCpu::~DispatcherAiCpu() {}
66 :
67 450 : HcclResult DispatcherAiCpu::Init()
68 : {
69 450 : CHK_RET(DlHalFunction::GetInstance().DlHalFunctionInit());
70 450 : CHK_RET(hrtHalGetDeviceType(aicpuInfo_.devId, aicpuInfo_.devType));
71 450 : 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 450 : 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 450 : InitTimeOutConfig();
120 450 : HCCL_INFO(
121 : "%s success, devId:%u, devType:%d, chipId:%lld", __func__, aicpuInfo_.devId, aicpuInfo_.devType,
122 : aicpuInfo_.chipId);
123 450 : 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 2762276 : while (((tail < head ? streamInfo.sqDepth : 0U) + tail - head + cnt >= streamInfo.sqDepth)
728 2762276 : && (tail != head)) { // 判断剩余sqe空间是否足够下发
729 : // 需要放在while循环进来后第一个执行
730 2762276 : CHK_RET(QuerySqStatusByType(aicpuInfo_.devId, streamInfo.sqId, DRV_SQCQ_PROP_SQ_HEAD, head));
731 :
732 : // 非阻塞下发场景，rtsq队列空间不足时直接返回
733 2762274 : if (isBlockLaunch == false) {
734 0 : return HCCL_SUCCESS;
735 : }
736 :
737 : // 当前流无法下发，把其他流都launch一遍，避免等待的其他流没有launch
738 2762274 : std::unique_lock<std::recursive_mutex> lock(streamMapMutex_);
739 2762274 : std::vector<s32> invalidStreamIds;
740 4196647 : for (auto it = streamMap_.begin(); it != streamMap_.end(); ++it) {
741 1434373 : if (it->second.IsInvalid()) { // 跳过已销毁的stream
742 0 : invalidStreamIds.push_back(it->first);
743 0 : HCCL_WARNING("[DispatcherAiCpu][LaunchTask] skip invalid stream in streamMap, streamId:%d", it->first);
744 0 : continue;
745 : }
746 1434373 : if (it->first != streamInfo.actualStreamId) {
747 1 : CHK_RET(LaunchTask(it->second, false));
748 : }
749 : }
750 2762274 : for (auto id : invalidStreamIds) {
751 0 : streamMap_.erase(id);
752 : }
753 2762274 : lock.unlock();
754 2762274 : u64 curUsec = GetCurAicpuTimestamp();
755 2762274 : if (dfxTimeOutConfig_.sqFullWaitTimeOut != 0
756 2762273 : && (curUsec - startUsec > NANOSECOND_TO_SECOND * dfxTimeOutConfig_.sqFullWaitTimeOut)) {
757 2 : HCCL_ERROR(
758 : "Rtsq full, timeout %lus. curhead:%u, sqId:%d", dfxTimeOutConfig_.sqFullWaitTimeOut, head,
759 : streamInfo.sqId);
760 2 : return HCCL_E_AGAIN;
761 : }
762 :
763 : // 等待下发阶段，每隔30s打印一次状态
764 2762272 : if (curUsec - lastUsec > NANOSECOND_TO_SECOND * dfx::kPrintSqInterval) {
765 0 : lastUsec = curUsec;
766 0 : HCCL_RUN_INFO(
767 : "[LaunchTask][WaitLaunchWhileLoop]Current state. sqid:%d, head:%u, tail:%u, cnt:%u", streamInfo.sqId,
768 : head, tail, cnt);
769 : }
770 :
771 : // 下发过程中出现cqe异常
772 2762272 : if (checkOpExecStatusCallback_ != nullptr) {
773 0 : HcclResult opExecStatus = checkOpExecStatusCallback_();
774 0 : CHK_PRT_RET(
775 : opExecStatus != HCCL_SUCCESS,
776 : HCCL_ERROR("hccl aicpu stop launch for task exception or stop command, ret:%d", opExecStatus),
777 : opExecStatus);
778 : }
779 2762276 : }
780 :
781 2 : uint32_t left = streamInfo.sqDepth - tail; // sqeAddr 剩余空间
782 2 : const auto tailSqeIdx = sqeContextBuffer->tailSqeIdx;
783 2 : HCCL_INFO("cpy sqe, left:%u, tailSqeId:%u, cnt:%u, streamId:%u", left, tailSqeIdx, cnt, stream.id());
784 2 : if (cnt <= left) { // 剩余buffer放得下新增sqe
785 1 : CHK_SAFETY_FUNC_RET(memcpy_s(
786 : reinterpret_cast<uint8_t*>(streamInfo.sqBaseAddr) + tail * HCCL_SQE_SIZE, left * HCCL_SQE_SIZE,
787 : sqeContextBuffer->localBuff + (tailSqeIdx - cnt) * HCCL_SQE_SIZE, cnt * HCCL_SQE_SIZE));
788 :
789 1 : CHK_SAFETY_FUNC_RET(memcpy_s(
790 : sqeContextBuffer->rtsMirrorBuffer + tail * HCCL_SQE_SIZE, left * HCCL_SQE_SIZE,
791 : sqeContextBuffer->localBuff + (tailSqeIdx - cnt) * HCCL_SQE_SIZE, cnt * HCCL_SQE_SIZE));
792 :
793 1 : CHK_SAFETY_FUNC_RET(
794 : memcpy_s(sqeContextBuffer->rtsqSqeType + tail, left, sqeContextBuffer->sqeType + (tailSqeIdx - cnt), cnt));
795 1 : CHK_SAFETY_FUNC_RET(memcpy_s(
796 : sqeContextBuffer->rtsDfxInfo + tail, left * sizeof(AicpuDfxInfo),
797 : sqeContextBuffer->dfxInfo + (tailSqeIdx - cnt), cnt * sizeof(AicpuDfxInfo)));
798 : } else {
799 1 : CHK_SAFETY_FUNC_RET(memcpy_s(
800 : reinterpret_cast<uint8_t*>(streamInfo.sqBaseAddr) + tail * HCCL_SQE_SIZE, left * HCCL_SQE_SIZE,
801 : sqeContextBuffer->localBuff + (tailSqeIdx - cnt) * HCCL_SQE_SIZE, left * HCCL_SQE_SIZE));
802 :
803 1 : CHK_SAFETY_FUNC_RET(memcpy_s(
804 : reinterpret_cast<uint8_t*>(streamInfo.sqBaseAddr), streamInfo.sqDepth * HCCL_SQE_SIZE,
805 : sqeContextBuffer->localBuff + (tailSqeIdx - cnt + left) * HCCL_SQE_SIZE, (cnt - left) * HCCL_SQE_SIZE));
806 :
807 1 : CHK_SAFETY_FUNC_RET(memcpy_s(
808 : sqeContextBuffer->rtsMirrorBuffer + tail * HCCL_SQE_SIZE, left * HCCL_SQE_SIZE,
809 : sqeContextBuffer->localBuff + (tailSqeIdx - cnt) * HCCL_SQE_SIZE, left * HCCL_SQE_SIZE));
810 :
811 1 : CHK_SAFETY_FUNC_RET(memcpy_s(
812 : sqeContextBuffer->rtsMirrorBuffer, streamInfo.sqDepth * HCCL_SQE_SIZE,
813 : sqeContextBuffer->localBuff + (tailSqeIdx - cnt + left) * HCCL_SQE_SIZE, (cnt - left) * HCCL_SQE_SIZE));
814 :
815 1 : CHK_SAFETY_FUNC_RET(
816 : memcpy_s(sqeContextBuffer->rtsqSqeType + tail, left, sqeContextBuffer->sqeType + (tailSqeIdx - cnt), left));
817 1 : CHK_SAFETY_FUNC_RET(memcpy_s(
818 : sqeContextBuffer->rtsqSqeType + 0, streamInfo.sqDepth,
819 : sqeContextBuffer->sqeType + (tailSqeIdx - cnt + left), (cnt - left)));
820 1 : CHK_SAFETY_FUNC_RET(memcpy_s(
821 : sqeContextBuffer->rtsDfxInfo + tail, left * sizeof(AicpuDfxInfo),
822 : sqeContextBuffer->dfxInfo + (tailSqeIdx - cnt), left * sizeof(AicpuDfxInfo)));
823 1 : CHK_SAFETY_FUNC_RET(memcpy_s(
824 : sqeContextBuffer->rtsDfxInfo + 0, streamInfo.sqDepth * sizeof(AicpuDfxInfo),
825 : sqeContextBuffer->dfxInfo + (tailSqeIdx - cnt + left), (cnt - left) * sizeof(AicpuDfxInfo)));
826 : }
827 : // 打印算子展开下发的SQE内容for debug
828 : // 设置HCCL_DEBUG_CONFIG="task", 或者设置ASCEND_GLOBAL_LOG_LEVEL=0
829 2 : if ((UNLIKELY(GetExternalInputDebugConfig() & PLF_TASK)) || UNLIKELY(HcclCheckLogLevel(HCCL_LOG_DEBUG))) {
830 2 : const int32_t streamId = stream.GetHcclStreamInfo().actualStreamId;
831 2 : PLF_CONFIG_DEBUG(
832 : PLF_TASK, "[DispatcherAicpu][LaunchTask] dump content of %u dispatched SQEs with stream id %u", cnt,
833 : streamId);
834 :
835 2 : uint8_t* sqeArray = sqeContextBuffer->localBuff + (tailSqeIdx - cnt) * HCCL_SQE_SIZE;
836 2 : uint8_t* sqeTypeArray = sqeContextBuffer->sqeType + (tailSqeIdx - cnt);
837 2 : AicpuDfxInfo* sqeDfxInfoArray = sqeContextBuffer->dfxInfo + (tailSqeIdx - cnt);
838 14 : for (size_t sqeIdx = 0; sqeIdx < cnt; ++sqeIdx) {
839 12 : uint8_t* sqePtr = sqeArray + sqeIdx * HCCL_SQE_SIZE;
840 12 : const uint8_t sqeType = sqeTypeArray[sqeIdx];
841 12 : if (sqeType == SqeType::FLIP_PLACEHOLDER_SQE) {
842 0 : const rtStarsPlaceHolderSqe_t* placeholderSqePtr
843 0 : = reinterpret_cast<const rtStarsPlaceHolderSqe_t*>(sqeArray + sqeIdx * HCCL_SQE_SIZE);
844 0 : PLF_CONFIG_DEBUG(
845 : PLF_TASK,
846 : "[DispatcherAicpu][LaunchTask] %uth dispatched SQE (placeholder) header.type[%u] taskid[%u]",
847 : sqeIdx, placeholderSqePtr->header.type, placeholderSqePtr->header.taskId);
848 : } else {
849 12 : PLF_CONFIG_DEBUG(PLF_TASK, "[DispatcherAicpu][LaunchTask] %uth dispatched SQE", sqeIdx);
850 : }
851 :
852 12 : CHK_RET(OpUnfoldCache::DumpSqeContent(sqePtr, sqeType));
853 :
854 12 : const AicpuDfxInfo& dfxinfo = sqeDfxInfoArray[sqeIdx];
855 12 : PLF_CONFIG_DEBUG(
856 : PLF_TASK, "[DispatcherAicpu][LaunchTask] AicpuDfxInfo: remoteRank[%u] opRingBufferIdx[%u] notifyId[%u]",
857 : dfxinfo.remoteRank, dfxinfo.opRingBufferIdx, dfxinfo.notifyId);
858 : }
859 : }
860 :
861 : // 当前算子展开的SQE需要被动态缓存
862 2 : if (needAddSqe_) {
863 0 : CHK_PTR_NULL(cachePtr_);
864 :
865 : // 查找key对应的cache entry, 如果不存在 (即当前算子第一次LaunchTask), 创建新的cache entry
866 0 : OpUnfoldCacheEntry* entryPtr = nullptr;
867 0 : CHK_RET(cachePtr_->FindEntry(key_, &entryPtr));
868 0 : if (entryPtr == nullptr) {
869 0 : CHK_RET(cachePtr_->AddEntry(key_, userInputMemRanges_, userOutputMemRanges_, &entryPtr));
870 : }
871 0 : CHK_PTR_NULL(entryPtr);
872 :
873 : // 准备SQE相关信息的数组基地址
874 0 : uint8_t* sqeArray = sqeContextBuffer->localBuff + (tailSqeIdx - cnt) * HCCL_SQE_SIZE;
875 0 : uint8_t* sqeTypeArray = sqeContextBuffer->sqeType + (tailSqeIdx - cnt);
876 0 : AicpuDfxInfo* sqeDfxInfoArray = sqeContextBuffer->dfxInfo + (tailSqeIdx - cnt);
877 :
878 : // 遍历sqeType找到placeholder的位置
879 0 : std::vector<size_t> placeholderIdxes;
880 0 : uint8_t* curSqeTypePtr = sqeTypeArray;
881 0 : for (size_t sqeTypeIdx = 0; sqeTypeIdx < cnt; ++sqeTypeIdx) {
882 0 : if (*curSqeTypePtr == SqeType::FLIP_PLACEHOLDER_SQE) {
883 0 : placeholderIdxes.emplace_back(sqeTypeIdx);
884 : }
885 0 : ++curSqeTypePtr;
886 : }
887 :
888 : // 在动态缓存中分配实际需要的SQE数组
889 0 : const size_t cacheableSqeCount = cnt - placeholderIdxes.size();
890 0 : const int32_t streamId = stream.GetHcclStreamInfo().actualStreamId;
891 0 : size_t arrayIdx = 0;
892 0 : CHK_RET(entryPtr->AllocSqeArray(cacheableSqeCount, streamId, arrayIdx));
893 :
894 : // 分段拷贝SQE相关信息到cache entry中
895 0 : size_t cacheableSqeStartIdx = 0; // SQE start index (在动态缓存对应SQE数组中的索引)
896 0 : size_t bufferSqeStartIdx = 0; // SQE start index (在SQE ring buffer中的索引)
897 0 : for (size_t i = 0; i < placeholderIdxes.size(); ++i) {
898 : // [bufferSqeStartIdx, curPlaceholderIdx) -> [cacheableSqeStartIdx, cacheableSqeStartIdx + curPlaceholderIdx
899 : // - bufferSqeStartIdx)
900 0 : const size_t curPlaceholderIdx = placeholderIdxes[i];
901 0 : HCCL_INFO(
902 : "[DispatcherAicpu][LaunchTask] %uth placeholder copy dispatchedSqeArray[%u:%u) into "
903 : "cachedSqeArrays[%u][%u:%u)",
904 : i, bufferSqeStartIdx, curPlaceholderIdx, arrayIdx, cacheableSqeStartIdx,
905 : cacheableSqeStartIdx + curPlaceholderIdx - bufferSqeStartIdx);
906 0 : if (curPlaceholderIdx <= bufferSqeStartIdx) { // NO non-placeholder dispatched SQE to admit
907 : // NOTE: NO need to change cacheableSqeStartIdx
908 0 : bufferSqeStartIdx = curPlaceholderIdx + 1;
909 : } else {
910 0 : const size_t curSqeCount = curPlaceholderIdx - bufferSqeStartIdx;
911 0 : CHK_RET(entryPtr->MemcpySqeArray(
912 : arrayIdx, cacheableSqeStartIdx, curSqeCount, sqeArray + bufferSqeStartIdx * HCCL_SQE_SIZE,
913 : sqeTypeArray + bufferSqeStartIdx, sqeDfxInfoArray + bufferSqeStartIdx, isAlltoallv_,
914 : alltoallvMetadataPtr_));
915 0 : cacheableSqeStartIdx += curSqeCount;
916 0 : bufferSqeStartIdx = curPlaceholderIdx + 1;
917 : }
918 : }
919 :
920 : // 存在剩余SQE, 即最后一个SQE不是placeholder
921 0 : if (LIKELY(bufferSqeStartIdx < cnt)) {
922 : // [bufferSqeStartIdx, cnt - 1] -> [cacheableSqeStartIdx, cacheableSqeStartIdx + cnt - bufferSqeStartIdx)
923 0 : const size_t curSqeCount = cnt - bufferSqeStartIdx;
924 0 : CHK_RET(entryPtr->MemcpySqeArray(
925 : arrayIdx, cacheableSqeStartIdx, curSqeCount, sqeArray + bufferSqeStartIdx * HCCL_SQE_SIZE,
926 : sqeTypeArray + bufferSqeStartIdx, sqeDfxInfoArray + bufferSqeStartIdx, isAlltoallv_,
927 : alltoallvMetadataPtr_));
928 : }
929 0 : }
930 :
931 2 : CHK_RET(ConfigSqStatusByType(aicpuInfo_.devId, streamInfo.sqId, DRV_SQCQ_PROP_SQ_TAIL, newTail));
932 2 : tail = newTail;
933 2 : PLF_CONFIG_INFO(
934 : PLF_TASK, "%s success, sqid:%d, sqe_num:%u, curHead:%u, curtail:%u", __func__, streamInfo.sqId, cnt, head,
935 : tail);
936 2 : sqeContextBuffer->sqeCnt = 0;
937 2 : return HCCL_SUCCESS;
938 : }
939 :
940 8 : HcclResult DispatcherAiCpu::LaunchTasksEx(hccl::Stream& stream, std::vector<Stream>& subStreams)
941 : {
942 : /* 两阶段模式，主流待正式执行时再下 */
943 : /* 一阶段第一次，可以先下主流 */
944 8 : HcclResult ret = LaunchTask(stream, true);
945 8 : if (ret != HCCL_SUCCESS) {
946 0 : HCCL_ERROR(
947 : "[DispatcherAiCpu][LaunchTasksEx] "
948 : "launch task failed, sqid:%u, ret:%u",
949 : stream.sqId(), ret);
950 0 : return ret;
951 : }
952 :
953 8 : for (u32 index = 0; index < subStreams.size(); index++) {
954 0 : ret = LaunchTask(subStreams[index], true);
955 0 : if (ret != HCCL_SUCCESS) {
956 0 : HCCL_ERROR(
957 : "[DispatcherAiCpu][LaunchTasksEx] "
958 : "launch task failed, sqid:%u, ret:%u",
959 : subStreams[index].sqId(), ret);
960 0 : return ret;
961 : }
962 : }
963 :
964 8 : return HCCL_SUCCESS;
965 : }
966 :
967 3 : HcclResult DispatcherAiCpu::LaunchAllTasks()
968 : {
969 3 : std::lock_guard<std::recursive_mutex> lock(streamMapMutex_);
970 3 : std::vector<s32> invalidStreamIds;
971 5 : for (auto it = streamMap_.begin(); it != streamMap_.end(); ++it) {
972 3 : if (it->second.IsInvalid()) { // 跳过已销毁的stream
973 0 : invalidStreamIds.push_back(it->first);
974 0 : HCCL_WARNING("[DispatcherAiCpu][LaunchAllTasks] skip invalid stream in streamMap, streamId:%d", it->first);
975 0 : continue;
976 : }
977 3 : HcclResult ret = LaunchTask(it->second, true);
978 3 : if (ret != HCCL_SUCCESS) {
979 1 : HCCL_ERROR("DispatcherAiCpu][LaunchAllTasks] launch task failed, sqid:%u, ret:%u", it->second.sqId(), ret);
980 1 : return ret;
981 : }
982 : }
983 2 : for (auto id : invalidStreamIds) {
984 0 : streamMap_.erase(id);
985 : }
986 :
987 2 : return HCCL_SUCCESS;
988 3 : }
989 :
990 1 : HcclResult DispatcherAiCpu::ReduceAsync(
991 : const void* src, void* dst, u64 dataCount, const HcclDataType datatype, HcclReduceOp redOp, Stream& stream,
992 : HcclReduceType reduceType)
993 : {
994 1 : return (reduceType == HcclReduceType::HCCL_INLINE_REDUCE) ?
995 1 : InlineReduceAsync(src, dataCount, datatype, redOp, stream, dst) :
996 1 : TbeReduceAsync(src, dst, dataCount, datatype, redOp, stream, dst);
997 : }
998 :
999 2 : HcclResult DispatcherAiCpu::InlineReduceAsync(
1000 : const void* src, u64 dataCount, const HcclDataType datatype, HcclReduceOp redOp, hccl::Stream& stream, void* dst,
1001 : u32 remoteUserRank, hccl::LinkType inLinkType)
1002 : {
1003 : // 参数有效性检查
1004 2 : CHK_PTR_NULL(stream.ptr());
1005 2 : if (dataCount == 0) {
1006 0 : HCCL_INFO("%s src memory size is 0, not need inline reduce.", __func__);
1007 0 : return HCCL_SUCCESS;
1008 : }
1009 2 : const HcclComStreamInfo& streamInfo = stream.GetHcclStreamInfo();
1010 :
1011 2 : aclDataType runtimeDataType = DT_MAP_TABLE[datatype];
1012 2 : aclrtReduceKind rtReduceOp = RK_MAP_TABLE[redOp];
1013 :
1014 : // 将数据按4GB切分循环处理
1015 2 : uint64_t spiltLoop = 0;
1016 2 : uint64_t addr_offset = 0;
1017 2 : uint64_t countSplit = 0;
1018 2 : uint64_t countSize = dataCount * SIZE_TABLE[datatype];
1019 2 : uint8_t* sqeBuffer = nullptr;
1020 2 : uint8_t* sqeTypeAddr = nullptr;
1021 2 : uint8_t* sqeDfxInfoAddr = nullptr;
1022 2 : uint16_t taskId = 0U;
1023 :
1024 2 : if (countSize > HCCL_SDMA_MAX_COUNT_4GB) {
1025 0 : spiltLoop = (countSize % HCCL_SDMA_MAX_COUNT_4GB) ? (countSize / HCCL_SDMA_MAX_COUNT_4GB) :
1026 0 : ((countSize / HCCL_SDMA_MAX_COUNT_4GB) - 1);
1027 0 : HCCL_INFO(
1028 : "%s InlineReduceAsync SDMA task countSize is bigger than 4GB"
1029 : " and do segmentation splitloop:%llu",
1030 : __func__, spiltLoop);
1031 : }
1032 2 : uint8_t linkType = static_cast<uint8_t>(inLinkType);
1033 4 : for (uint64_t index = 0; index <= spiltLoop; index++) {
1034 2 : addr_offset = index * HCCL_SDMA_MAX_COUNT_4GB;
1035 2 : countSplit = (index == spiltLoop) ? (countSize - index * HCCL_SDMA_MAX_COUNT_4GB) : (HCCL_SDMA_MAX_COUNT_4GB);
1036 2 : void* srcSplit = static_cast<void*>(static_cast<char*>(const_cast<void*>(src)) + addr_offset);
1037 2 : void* dstSplit = static_cast<void*>(static_cast<char*>(dst) + addr_offset);
1038 :
1039 2 : CHK_RET(GetStreamSqeBufferAddr(stream, sqeBuffer, sqeTypeAddr, sqeDfxInfoAddr, taskId));
1040 2 : AicpuDfxInfo* const dfxInfo = (AicpuDfxInfo* const)sqeDfxInfoAddr;
1041 2 : dfxInfo->opRingBufferIdx = opRingBufferIdx_;
1042 2 : dfxInfo->remoteRank = remoteUserRank;
1043 2 : dfxInfo->notifyId = INVALID_VALUE_RANKID;
1044 2 : addOneMemcpySqe_(
1045 2 : streamInfo.actualStreamId, taskId, srcSplit, countSplit, runtimeDataType, rtReduceOp, dstSplit, 0,
1046 : aicpuInfo_.ssid, aicpuInfo_.devId, aicpuInfo_.overflowAddr, linkType, sqeBuffer, sqeTypeAddr, hcclQos_);
1047 :
1048 2 : PLF_CONFIG_INFO(
1049 : PLF_TASK,
1050 : "%s para: linkType[%u] srcSplit[%p] dstSplit[%p] countSplit[%llu] taskId[%u] streamId[%u] remoteRank[%u] "
1051 : "rtDatatType[%d] rtReduceOp[%d]",
1052 : __func__, linkType, srcSplit, dstSplit, countSplit, taskId, streamInfo.actualStreamId, remoteUserRank,
1053 : runtimeDataType, rtReduceOp);
1054 : }
1055 :
1056 2 : return HCCL_SUCCESS;
1057 : }
1058 :
1059 1 : HcclResult DispatcherAiCpu::TbeReduceAsync(
1060 : [[maybe_unused]] const void* src1, [[maybe_unused]] const void* src2, [[maybe_unused]] u64 count,
1061 : [[maybe_unused]] const HcclDataType datatype, [[maybe_unused]] HcclReduceOp redOp, [[maybe_unused]] Stream& stream,
1062 : [[maybe_unused]] const void* dst)
1063 : {
1064 1 : HCCL_ERROR("[DispatcherAiCpu][TbeReduceAsync] aicpu do not support the tbe reduce");
1065 1 : return HCCL_E_NOT_SUPPORT;
1066 : }
1067 :
1068 5 : HcclResult DispatcherAiCpu::RdmaSend(u32 dbindex, u64 dbinfo, hccl::Stream& stream, RdmaTaskInfo& taskInfo)
1069 : {
1070 5 : const HcclComStreamInfo& streamInfo = stream.GetHcclStreamInfo();
1071 :
1072 5 : uint8_t* sqeBuffer = nullptr;
1073 5 : uint8_t* sqeTypeAddr = nullptr;
1074 5 : uint8_t* sqeDfxInfoAddr = nullptr;
1075 5 : uint16_t taskId = 0U;
1076 :
1077 5 : CHK_RET(GetStreamSqeBufferAddr(stream, sqeBuffer, sqeTypeAddr, sqeDfxInfoAddr, taskId));
1078 5 : AicpuDfxInfo* const dfxInfo = (AicpuDfxInfo* const)sqeDfxInfoAddr;
1079 5 : dfxInfo->opRingBufferIdx = opRingBufferIdx_;
1080 5 : dfxInfo->remoteRank = taskInfo.remoteRank;
1081 5 : dfxInfo->notifyId = INVALID_UINT; // 多个wr只敲一次doorbell的情况下，一般只会有一个notify
1082 :
1083 5 : uint32_t wrLen = 0; // 统计wr的总数据量
1084 5 : for (const WrInformation& wr : taskInfo.wrInfos) {
1085 0 : wrLen += wr.wrData.memList.len;
1086 0 : dfxInfo->notifyId = (wr.notifyId != INVALID_UINT) ? wr.notifyId : dfxInfo->notifyId;
1087 : }
1088 :
1089 5 : u64 dbAddr = CalcDbAddr(dbindex);
1090 5 : addOneRdmaDbSendSqe_(
1091 5 : streamInfo.actualStreamId, taskId, dbinfo, dbAddr, wrLen, static_cast<uint8_t>(taskInfo.rdmaType), sqeBuffer,
1092 : sqeTypeAddr);
1093 :
1094 5 : PLF_CONFIG_INFO(
1095 : PLF_TASK, "%s para: streamId[%u] taskId[%u] remoteRank[%u] RdmaType[%d] wrLen[%u] notifyId[%u]", __func__,
1096 : streamInfo.actualStreamId, taskId, taskInfo.remoteRank, taskInfo.rdmaType, wrLen, dfxInfo->notifyId);
1097 :
1098 5 : return HCCL_SUCCESS;
1099 : }
1100 :
1101 0 : HcclResult DispatcherAiCpu::RdmaRecord(
1102 : [[maybe_unused]] u32 dbindex, [[maybe_unused]] u64 dbinfo, [[maybe_unused]] const struct SendWr& wr,
1103 : [[maybe_unused]] hccl::Stream& stream, [[maybe_unused]] RdmaType rdmaType, [[maybe_unused]] u32 userRank,
1104 : [[maybe_unused]] u64 offset, [[maybe_unused]] u32 notifyId)
1105 : {
1106 0 : return HCCL_SUCCESS;
1107 : }
1108 :
1109 13 : HcclResult DispatcherAiCpu::GetStreamSqeBufferAddr(
1110 : hccl::Stream& stream, uint8_t*& sqeBufferAddr, uint8_t*& sqeTypeAddr, uint8_t*& sqeDfxInfoAddr, uint16_t& taskId)
1111 : {
1112 13 : SaveStreamInfo(stream);
1113 13 : HcclSqeContext* sqeContext = stream.GetSqeContextPtr();
1114 13 : CHK_PTR_NULL(sqeContext);
1115 13 : if (UNLIKELY(sqeContext->buffer.sqeCnt >= HCCL_PER_LAUNCH_SQE_CNT)) {
1116 0 : HCCL_INFO("GetStreamSqeBufferAddr tailSqeIdx[%u], try to launchTask", sqeContext->buffer.tailSqeIdx);
1117 0 : CHK_RET(LaunchTask(stream, true));
1118 : }
1119 13 : if (UNLIKELY(sqeContext->buffer.tailSqeIdx >= HCCL_SQE_MAX_CNT)) {
1120 0 : CHK_RET(LaunchTask(stream, true));
1121 :
1122 0 : if (callback_ != nullptr) {
1123 0 : hccl::AiCPUStreamTasks para(stream.id(), reinterpret_cast<void*>(sqeContext));
1124 0 : hccl::TaskPara taskPara(TaskType::TASK_BATCH_REPORT, para);
1125 0 : callback_(callBackUserPtr_, (void*)&taskPara, sizeof(struct TaskPara));
1126 0 : }
1127 : }
1128 13 : SqeRingBuffer* sqeContextBuffer = &(sqeContext->buffer);
1129 13 : uint16_t flipNum = sqeContextBuffer->filpNum;
1130 13 : uint16_t nextTaskId = sqeContextBuffer->tailSqeTaskId;
1131 : // nextTaskId=0的时候下发PlaceHolder
1132 13 : if (UNLIKELY(nextTaskId == 0 && flipNum != 0)) {
1133 0 : CHK_RET(AddFlipTask(stream));
1134 : }
1135 13 : if (UNLIKELY(sqeContext->buffer.tailSqeIdx >= HCCL_SQE_MAX_CNT)) {
1136 0 : CHK_RET(LaunchTask(stream, true));
1137 :
1138 0 : if (callback_ != nullptr) {
1139 0 : hccl::AiCPUStreamTasks para(stream.id(), reinterpret_cast<void*>(sqeContext));
1140 0 : hccl::TaskPara taskPara(TaskType::TASK_BATCH_REPORT, para);
1141 0 : callback_(callBackUserPtr_, (void*)&taskPara, sizeof(struct TaskPara));
1142 0 : }
1143 : }
1144 13 : CHK_RET(stream.GetNextSqeBufferAddr(sqeBufferAddr, sqeTypeAddr, sqeDfxInfoAddr, taskId));
1145 13 : return HCCL_SUCCESS;
1146 : }
1147 :
1148 2 : HcclResult DispatcherAiCpu::WaitRtsq(Stream& stream, const size_t& sqeCount, const bool isBlockLaunch)
1149 : {
1150 : // 注意: 目前WaitRtsq不会被递归调用, 所以isBlockLaunch永远为true; 为防止以后LaunchTask递归使用WaitRtsq,
1151 : // 编码时考虑isBlockLaunch为false的情况
1152 :
1153 : // 检验入参
1154 2 : const HcclComStreamInfo& streamInfo = stream.GetHcclStreamInfo();
1155 2 : if (sqeCount == 0) {
1156 0 : CHK_PRT_CONT(
1157 : isBlockLaunch, HCCL_DEBUG(
1158 : "[DispatcherAiCpu][WaitRtsq] no sqe, streamId:%d, sqId:%u", streamInfo.actualStreamId,
1159 : streamInfo.sqId));
1160 0 : return HCCL_SUCCESS;
1161 2 : } else if (sqeCount > streamInfo.sqDepth) {
1162 0 : HCCL_ERROR(
1163 : "[DispatcherAiCpu][WaitRtsq] sqeCount %u should be smaller than sqDepth %u]", sqeCount, streamInfo.sqDepth);
1164 0 : return HCCL_E_PTR;
1165 : }
1166 :
1167 : // Get head and tail of RTSQ ring buffer
1168 2 : HcclSqeContext* sqeContext = stream.GetSqeContextPtr();
1169 2 : CHK_PTR_NULL(sqeContext);
1170 2 : SqeRingBuffer* sqeContextBuffer = &(sqeContext->buffer);
1171 2 : CHK_PTR_NULL(sqeContextBuffer);
1172 2 : uint32_t& head = sqeContextBuffer->sqHead;
1173 2 : uint32_t& tail = sqeContextBuffer->sqTail;
1174 :
1175 : // Dump debug information
1176 2 : const uint32_t newTail = (tail + sqeCount) % streamInfo.sqDepth;
1177 : // 仅在阻塞下发场景打印，避免非阻塞场景调用时刷屏
1178 2 : CHK_PRT_CONT(
1179 : isBlockLaunch, HCCL_INFO(
1180 : "[DispatcherAicpu][WaitRtsq] sqid:%d sqeCount:%u head:%u curtail:%u newTail:%u",
1181 : streamInfo.sqId, sqeCount, head, tail, newTail));
1182 :
1183 : // 轮询RTSQ直至获得足够大的剩余空间
1184 2 : u64 startUsec = GetCurAicpuTimestamp();
1185 2 : u64 lastUsec = startUsec;
1186 3 : while (((tail < head ? streamInfo.sqDepth : 0U) + tail - head + sqeCount >= streamInfo.sqDepth)
1187 3 : && (tail != head)) { // 判断RTSQ中剩余sqe空间是否足够下发
1188 : // 需要放在while循环进来后第一个执行 (获取最新的RTSQ head, 查看RTSQ的消费进度)
1189 3 : CHK_RET(QuerySqStatusByType(aicpuInfo_.devId, streamInfo.sqId, DRV_SQCQ_PROP_SQ_HEAD, head));
1190 :
1191 : // 非阻塞下发场景，rtsq队列空间不足时直接返回
1192 2 : if (isBlockLaunch == false) {
1193 1 : return HCCL_SUCCESS;
1194 : }
1195 :
1196 : // 当前流无法下发，把其他流都launch一遍，避免等待的其他流没有launch
1197 1 : std::unique_lock<std::recursive_mutex> lock(streamMapMutex_);
1198 1 : std::vector<s32> invalidStreamIds;
1199 2 : for (auto it = streamMap_.begin(); it != streamMap_.end(); ++it) {
1200 1 : if (it->second.IsInvalid()) { // 跳过已销毁的stream
1201 0 : invalidStreamIds.push_back(it->first);
1202 0 : HCCL_WARNING("[DispatcherAiCpu][WaitRtsq] skip invalid stream in streamMap, streamId:%d", it->first);
1203 0 : continue;
1204 : }
1205 1 : if (it->first != streamInfo.actualStreamId) { // 不是当前stream
1206 1 : CHK_RET(LaunchTask(it->second, false)); // 非阻塞launch
1207 : }
1208 : }
1209 1 : for (auto id : invalidStreamIds) {
1210 0 : streamMap_.erase(id);
1211 : }
1212 1 : lock.unlock();
1213 :
1214 : // 等待超时
1215 1 : u64 curUsec = GetCurAicpuTimestamp();
1216 1 : if (dfxTimeOutConfig_.sqFullWaitTimeOut != 0
1217 0 : && (curUsec - startUsec > NANOSECOND_TO_SECOND * dfxTimeOutConfig_.sqFullWaitTimeOut)) {
1218 0 : HCCL_ERROR(
1219 : "[DispatcherAicpu][WaitRtsq] Rtsq full, timeout %lus. curhead:%u, sqId:%d",
1220 : dfxTimeOutConfig_.sqFullWaitTimeOut, head, streamInfo.sqId);
1221 0 : return HCCL_E_AGAIN;
1222 : }
1223 :
1224 : // 等待下发阶段，每隔30s打印一次状态
1225 1 : if (curUsec - lastUsec > NANOSECOND_TO_SECOND * dfx::kPrintSqInterval) {
1226 0 : lastUsec = curUsec;
1227 0 : HCCL_RUN_INFO(
1228 : "[DispatcherAicpu][WaitRtsq] Current state. sqid:%d, head:%u, tail:%u, sqeCount:%u", streamInfo.sqId,
1229 : head, tail, sqeCount);
1230 : }
1231 :
1232 : // 等待下发过程中出现cqe异常, 需要终止当前算子SQE的下发过程
1233 1 : if (checkOpExecStatusCallback_ != nullptr) {
1234 0 : HcclResult opExecStatus = checkOpExecStatusCallback_();
1235 0 : CHK_PRT_RET(
1236 : opExecStatus != HCCL_SUCCESS,
1237 : HCCL_ERROR(
1238 : "[DispatcherAicpu][WaitRtsq] hccl aicpu stop launch for task exception or stop command, ret:%d",
1239 : opExecStatus),
1240 : opExecStatus);
1241 : }
1242 1 : }
1243 :
1244 1 : return HCCL_SUCCESS;
1245 : }
1246 :
1247 0 : HcclResult DispatcherAiCpu::MemcpyRtsq(
1248 : Stream& stream, const size_t sqeCount, const uint8_t* sqeArray, const uint8_t* sqeTypeArray,
1249 : const AicpuDfxInfo* sqeDfxInfoArray, const bool profL1Enable, const std::vector<uint64_t>& profTimestamps,
1250 : const size_t profTimestampStartIdx)
1251 : {
1252 : // 检验入参
1253 0 : const HcclComStreamInfo& streamInfo = stream.GetHcclStreamInfo();
1254 0 : if (sqeCount == 0) {
1255 0 : HCCL_DEBUG(
1256 : "[DispatcherAiCpu][MemcpyRtsq] no sqe, streamId:%d, sqId:%u", streamInfo.actualStreamId, streamInfo.sqId);
1257 0 : return HCCL_SUCCESS;
1258 0 : } else if (sqeCount > streamInfo.sqDepth) {
1259 0 : HCCL_ERROR(
1260 : "[DispatcherAiCpu][MemcpyRtsq] sqeCount %u should be smaller than sqDepth %u]", sqeCount,
1261 : streamInfo.sqDepth);
1262 0 : return HCCL_E_PTR;
1263 : }
1264 0 : CHK_PTR_NULL(sqeArray);
1265 0 : CHK_PTR_NULL(sqeTypeArray);
1266 0 : CHK_PTR_NULL(sqeDfxInfoArray);
1267 0 : if (profL1Enable) {
1268 : // 会访问profTimestamps[profTimestampStartIdx, profTimestampStartIdx + sqeCount - 1]
1269 0 : CHK_PRT_RET(
1270 : profTimestamps.size() == 0, HCCL_ERROR("[DispatcherAiCpu][MemcpyRtsq] empty profTimestamps"),
1271 : HCCL_E_INTERNAL);
1272 0 : CHK_PRT_RET(
1273 : profTimestampStartIdx >= profTimestamps.size(),
1274 : HCCL_ERROR(
1275 : "[DispatcherAiCpu][MemcpyRtsq] profTimestampStartIdx[%u] >= profTimestamps.size[%u]",
1276 : profTimestampStartIdx, profTimestamps.size()),
1277 : HCCL_E_INTERNAL);
1278 0 : CHK_PRT_RET(
1279 : (profTimestampStartIdx + sqeCount - 1) >= profTimestamps.size(),
1280 : HCCL_ERROR(
1281 : "[DispatcherAiCpu][MemcpyRtsq] profTimestampStartIdx[%u] + sqeCount[%u] - 1 >= profTimestamps.size[%u]",
1282 : profTimestampStartIdx, sqeCount, profTimestamps.size()),
1283 : HCCL_E_INTERNAL);
1284 : }
1285 :
1286 : // 获得RTSQ的head和tail
1287 0 : HcclSqeContext* sqeContext = stream.GetSqeContextPtr();
1288 0 : CHK_PTR_NULL(sqeContext);
1289 0 : SqeRingBuffer* sqeContextBuffer = &(sqeContext->buffer);
1290 0 : CHK_PTR_NULL(sqeContextBuffer);
1291 0 : uint32_t& head = sqeContextBuffer->sqHead;
1292 0 : uint32_t& tail = sqeContextBuffer->sqTail;
1293 :
1294 : // Dump debug information
1295 0 : const uint32_t newTail = (tail + sqeCount) % streamInfo.sqDepth;
1296 0 : HCCL_INFO(
1297 : "[DispatcherAicpu][MemcpyRtsq] before memcpy, sqid:%d sqeCount:%u head:%u curtail:%u newTail:%u",
1298 : streamInfo.sqId, sqeCount, head, tail, newTail);
1299 :
1300 : // 准备memcpy中目的末端基地址 (RTSQ从tail开始拷贝, [head, tail)为待执行SQE)
1301 0 : uint8_t* rtsqSqeTailBaseAddr = reinterpret_cast<uint8_t*>(streamInfo.sqBaseAddr) + tail * HCCL_SQE_SIZE;
1302 0 : uint8_t* mirrorRtsqSqeTailBaseAddr = sqeContextBuffer->rtsMirrorBuffer + tail * HCCL_SQE_SIZE;
1303 0 : uint8_t* rtsqSqeTypeTailBaseAddr = sqeContextBuffer->rtsqSqeType + tail;
1304 0 : AicpuDfxInfo* rtsqDfxInfoTailBaseAddr = sqeContextBuffer->rtsDfxInfo + tail;
1305 :
1306 0 : uint32_t tailLeft = streamInfo.sqDepth - tail; // RTSQ tail到buffer末端的剩余空间 (不包括buffer前端到head的剩余空间)
1307 0 : HCCL_INFO(
1308 : "[DispatcherAicpu][MemcpyRtsq] cpy sqe, tailLeft:%u, sqeCount:%u, streamId:%u", tailLeft, sqeCount,
1309 : stream.id());
1310 0 : if (sqeCount <= tailLeft) { // buffer末端剩余空间放得下新增sqe
1311 : // 向buffer末端拷贝sqeCount个SQE信息
1312 :
1313 : // 拷贝SQE内容到RTSQ
1314 0 : CHK_SAFETY_FUNC_RET(
1315 : memcpy_s(rtsqSqeTailBaseAddr, tailLeft * HCCL_SQE_SIZE, sqeArray, sqeCount * HCCL_SQE_SIZE));
1316 :
1317 : // 拷贝SQE内容到RTSQ mirror
1318 0 : CHK_SAFETY_FUNC_RET(
1319 : memcpy_s(mirrorRtsqSqeTailBaseAddr, tailLeft * HCCL_SQE_SIZE, sqeArray, sqeCount * HCCL_SQE_SIZE));
1320 :
1321 : // 拷贝SQE类型
1322 0 : CHK_SAFETY_FUNC_RET(memcpy_s(rtsqSqeTypeTailBaseAddr, tailLeft, sqeTypeArray, sqeCount));
1323 :
1324 : // 拷贝SQE DfxInfo
1325 0 : CHK_SAFETY_FUNC_RET(memcpy_s(
1326 : rtsqDfxInfoTailBaseAddr, tailLeft * sizeof(AicpuDfxInfo), sqeDfxInfoArray,
1327 : sqeCount * sizeof(AicpuDfxInfo)));
1328 : } else { // 需要buffer末端和首端的剩余空间
1329 : // 先向buffer末端拷贝tailLeft个SQE信息, 再向buffer首端拷贝sqeCount-tailLeft个SQE信息
1330 :
1331 : // 拷贝SQE内容到RTSQ
1332 0 : CHK_SAFETY_FUNC_RET(
1333 : memcpy_s(rtsqSqeTailBaseAddr, tailLeft * HCCL_SQE_SIZE, sqeArray, tailLeft * HCCL_SQE_SIZE));
1334 0 : CHK_SAFETY_FUNC_RET(memcpy_s(
1335 : reinterpret_cast<uint8_t*>(streamInfo.sqBaseAddr), streamInfo.sqDepth * HCCL_SQE_SIZE,
1336 : sqeArray + tailLeft * HCCL_SQE_SIZE, (sqeCount - tailLeft) * HCCL_SQE_SIZE));
1337 :
1338 : // 拷贝SQE内容到RTSQ mirror
1339 0 : CHK_SAFETY_FUNC_RET(
1340 : memcpy_s(mirrorRtsqSqeTailBaseAddr, tailLeft * HCCL_SQE_SIZE, sqeArray, tailLeft * HCCL_SQE_SIZE));
1341 0 : CHK_SAFETY_FUNC_RET(memcpy_s(
1342 : sqeContextBuffer->rtsMirrorBuffer, streamInfo.sqDepth * HCCL_SQE_SIZE, sqeArray + tailLeft * HCCL_SQE_SIZE,
1343 : (sqeCount - tailLeft) * HCCL_SQE_SIZE));
1344 :
1345 : // 拷贝SQE type
1346 0 : CHK_SAFETY_FUNC_RET(memcpy_s(rtsqSqeTypeTailBaseAddr, tailLeft, sqeTypeArray, tailLeft));
1347 0 : CHK_SAFETY_FUNC_RET(memcpy_s(
1348 : sqeContextBuffer->rtsqSqeType, streamInfo.sqDepth, sqeTypeArray + tailLeft, (sqeCount - tailLeft)));
1349 :
1350 : // 拷贝SQE DfxInfo
1351 0 : CHK_SAFETY_FUNC_RET(memcpy_s(
1352 : rtsqDfxInfoTailBaseAddr, tailLeft * sizeof(AicpuDfxInfo), sqeDfxInfoArray,
1353 : tailLeft * sizeof(AicpuDfxInfo)));
1354 0 : CHK_SAFETY_FUNC_RET(memcpy_s(
1355 : sqeContextBuffer->rtsDfxInfo, streamInfo.sqDepth * sizeof(AicpuDfxInfo), sqeDfxInfoArray + tailLeft,
1356 : (sqeCount - tailLeft) * sizeof(AicpuDfxInfo)));
1357 : }
1358 :
1359 : // 更新RTSQ ring buffer的tail
1360 0 : CHK_RET(ConfigSqStatusByType(aicpuInfo_.devId, streamInfo.sqId, DRV_SQCQ_PROP_SQ_TAIL, newTail));
1361 0 : tail = newTail;
1362 0 : PLF_CONFIG_INFO(
1363 : PLF_TASK, "%s success, sqid:%d, sqe_num:%u, curHead:%u, curtail:%u", __func__, streamInfo.sqId, sqeCount, head,
1364 : tail);
1365 :
1366 : // 上报profiling信息
1367 0 : if (profL1Enable) {
1368 : // Cache hit下SQE ring buffer中[tail-cnt, tail)为待下发SQE, 其数量一定为0 (否则更新tailSqeIdx后,
1369 : // 会导致待下发SQE未下发, 而已下发SQE重复下发)
1370 0 : CHK_PRT_RET(
1371 : sqeContextBuffer->sqeCnt != 0,
1372 : HCCL_ERROR(
1373 : "[DispatcherAicpu][MemcpyRtsq] sqeContextBuffer->sqeCnt[%u] should be zero!", sqeContextBuffer->sqeCnt),
1374 : HCCL_E_INTERNAL);
1375 :
1376 : // 上报flip placeholder的profiling信息
1377 0 : if ((*sqeTypeArray) == SqeType::FLIP_PLACEHOLDER_SQE) {
1378 0 : CHK_PRT_RET(
1379 : sqeCount != 1, HCCL_ERROR("[DispatcherAicpu][MemcpyRtsq] sqeCount[%u] should be 1!", sqeCount),
1380 : HCCL_E_INTERNAL);
1381 :
1382 : // 注意: 参考AddFlipTask, 先上报flip task (提醒profiling翻转taskid), 之后再拷贝到SQE ring buffer
1383 : // (捕捉placeholder SQE相关的profiling信息) 注意:
1384 : // ProfilingManager::TaskProfilingCallBack->ReportFilpTask不会扫描SQE ring buffer,
1385 : // 也不会更新其中的streamToSqeIdxMap_
1386 0 : if (callback_ != nullptr) {
1387 0 : const rtStarsPlaceHolderSqe_t* placeholderSqePtr
1388 : = reinterpret_cast<const rtStarsPlaceHolderSqe_t*>(sqeArray);
1389 : hccl::FlipTaskPara para(
1390 0 : stream.id(), placeholderSqePtr->header.taskId, placeholderSqePtr->u.flip_task_info.flipNumReport);
1391 0 : hccl::TaskPara taskPara(TaskType::TASK_FLIP, para);
1392 0 : callback_(callBackUserPtr_, (void*)&taskPara, sizeof(struct TaskPara));
1393 0 : }
1394 : }
1395 :
1396 : // 循环拷贝cached SQE到SQE ring buffer中
1397 0 : size_t reportSqeCount = 0;
1398 0 : while (reportSqeCount < sqeCount) {
1399 0 : CHK_PRT_RET(
1400 : sqeContextBuffer->tailSqeIdx > HCCL_SQE_MAX_CNT,
1401 : HCCL_ERROR(
1402 : "[DispatcherAicpu][MemcpyRtsq] tailSqeIdx[%u] > HCCL_SQE_MAX_CNT[%u]", sqeContextBuffer->tailSqeIdx,
1403 : HCCL_SQE_MAX_CNT),
1404 : HCCL_E_INTERNAL);
1405 0 : const size_t sqeTailLeft = HCCL_SQE_MAX_CNT - sqeContextBuffer->tailSqeIdx;
1406 0 : if (sqeTailLeft > 0) {
1407 : // 准备profiling上报的目的末端基地址 (SQE ring buffer从tail开始拷贝)
1408 0 : uint8_t* sqeLocalBuffTailBaseAddr
1409 0 : = sqeContextBuffer->localBuff + sqeContextBuffer->tailSqeIdx * HCCL_SQE_SIZE;
1410 0 : uint8_t* sqeTypeTailBaseAddr = sqeContextBuffer->sqeType + sqeContextBuffer->tailSqeIdx;
1411 0 : AicpuDfxInfo* dfxInfoTailBaseAddr = sqeContextBuffer->dfxInfo + sqeContextBuffer->tailSqeIdx;
1412 0 : uint64_t* profTimestapTailBaseAddr = sqeContextBuffer->profTimestap + sqeContextBuffer->tailSqeIdx;
1413 :
1414 : // 向SQE ring buffer末端拷贝SQE信息[reportSqeCount, reportSqeCount + tmpSqeCount - 1]
1415 : // (只用于profiling上报, 不会下发)
1416 0 : const size_t tmpSqeCount = std::min(sqeCount - reportSqeCount, sqeTailLeft);
1417 0 : HCCL_INFO(
1418 : "[DispatcherAicpu][MemcpyRtsq] report sqe profiling, sqeCount[%u] reportSqeCount[%u] "
1419 : "tailSqeIdx[%u] sqeTailLeft[%u] tmpSqeCount[%u]",
1420 : sqeCount, reportSqeCount, sqeContextBuffer->tailSqeIdx, sqeTailLeft, tmpSqeCount);
1421 :
1422 : // 拷贝SQE内容
1423 0 : CHK_SAFETY_FUNC_RET(memcpy_s(
1424 : sqeLocalBuffTailBaseAddr, sqeTailLeft * HCCL_SQE_SIZE, sqeArray + reportSqeCount * HCCL_SQE_SIZE,
1425 : tmpSqeCount * HCCL_SQE_SIZE));
1426 :
1427 : // 拷贝SQE类型
1428 0 : CHK_SAFETY_FUNC_RET(
1429 : memcpy_s(sqeTypeTailBaseAddr, sqeTailLeft, sqeTypeArray + reportSqeCount, tmpSqeCount));
1430 :
1431 : // 拷贝SQE DfxInfo
1432 0 : CHK_SAFETY_FUNC_RET(memcpy_s(
1433 : dfxInfoTailBaseAddr, sqeTailLeft * sizeof(AicpuDfxInfo), sqeDfxInfoArray + reportSqeCount,
1434 : tmpSqeCount * sizeof(AicpuDfxInfo)));
1435 :
1436 : // 拷贝SQE timestamp
1437 0 : CHK_SAFETY_FUNC_RET(memcpy_s(
1438 : profTimestapTailBaseAddr, sqeTailLeft * sizeof(uint64_t),
1439 : profTimestamps.data() + profTimestampStartIdx + reportSqeCount, tmpSqeCount * sizeof(uint64_t)));
1440 :
1441 : // 注意: sqeContextBuffer->sqeCnt不更新, 仍然为0 (即拷贝的SQE信息为已下发待上报), 避免SQE重复下发
1442 0 : sqeContextBuffer->tailSqeIdx += static_cast<uint16_t>(tmpSqeCount);
1443 0 : reportSqeCount += tmpSqeCount;
1444 : } else {
1445 : // 注意: SQE ring buffer中待下发SQE数量一定为0, 不需要调用LaunchTask将待下发变成已下发待上报,
1446 : // 可以直接上报profiling将已下发待上报变成已上报 注意: 调用后,
1447 : // ProfilingManager::StartReportSqeIdx为HCCL_SQE_MAX_CNT
1448 0 : if (callback_ != nullptr) {
1449 0 : hccl::AiCPUStreamTasks para(stream.id(), reinterpret_cast<void*>(sqeContext));
1450 0 : hccl::TaskPara taskPara(TaskType::TASK_BATCH_REPORT, para);
1451 0 : callback_(callBackUserPtr_, (void*)&taskPara, sizeof(struct TaskPara));
1452 0 : }
1453 :
1454 : // SQE ring buffer中所有SQE均为已上报 -> 清理SQE ring buffer
1455 0 : HCCL_INFO("[DispatcherAicpu][MemcpyRtsq] Sqe index to %u, need clear", HCCL_SQE_MAX_CNT);
1456 0 : CHK_PRT_RET(
1457 : sqeContextBuffer->sqeCnt != 0,
1458 : HCCL_ERROR(
1459 : "[DispatcherAicpu][MemcpyRtsq] Sqe index to %u, but sqeCnt[%u] is not 0", HCCL_SQE_MAX_CNT,
1460 : sqeContextBuffer->sqeCnt),
1461 : HCCL_E_INTERNAL);
1462 0 : CHK_RET(stream.ClearLocalBuff()); // 会将stream.sqeContextBuffer中的sqeCnt和tailSqeIdx设置为0
1463 0 : CHK_PRT_RET(
1464 : sqeContextBuffer->sqeCnt != 0,
1465 : HCCL_ERROR(
1466 : "[DispatcherAicpu][MemcpyRtsq] sqeCnt[%u] should be 0 after clear", sqeContextBuffer->sqeCnt),
1467 : HCCL_E_INTERNAL);
1468 0 : CHK_PRT_RET(
1469 : sqeContextBuffer->tailSqeIdx != 0,
1470 : HCCL_ERROR(
1471 : "[DispatcherAicpu][MemcpyRtsq] tailSqeIdx[%u] should be 0 after clear",
1472 : sqeContextBuffer->tailSqeIdx),
1473 : HCCL_E_INTERNAL);
1474 :
1475 : // 参考HcclCommAicpu::ClearLocalBuff, 调用Stream::ClearLocalBuff后,
1476 : // 应该调用ProfilingManager::UpdateStartReportSqeIdx手动将ProfilingManager::StartReportSqeIdx设置为0
1477 : // 注意: 由于platform暂未将UpdateStartReportSqeIdx作为回调函数传入, 无法直接调用此framework函数 ->
1478 : // 通过callback_ (即ProfilingManager::TaskProfilingCallBack)
1479 : // 间接将ProfilingManager::StartReportSqeIdx设置为0 注意:
1480 : // 由于调用前ProfilingManager::StartReportSqeIdx为HCCL_SQE_MAX_CNT, tailSqeIdx为0,
1481 : // 即从startIdx=HCCL_SQE_MAX_CNT到endIdx=0, ProfilingManager不会进入profiling上报代码,
1482 : // 而是只会调用UpdateStartReportSqeIdx设置StartReportSqeIdx为0
1483 0 : if (callback_ != nullptr) {
1484 0 : HCCL_INFO("[DispatcherAicpu][MemcpyRtsq] re-invoke callback_ to reset StartReportSqeIdx as 0 in "
1485 : "ProfilingManager");
1486 :
1487 0 : hccl::AiCPUStreamTasks para(stream.id(), reinterpret_cast<void*>(sqeContext));
1488 0 : hccl::TaskPara taskPara(TaskType::TASK_BATCH_REPORT, para);
1489 0 : callback_(callBackUserPtr_, (void*)&taskPara, sizeof(struct TaskPara));
1490 0 : }
1491 : }
1492 : }
1493 : }
1494 :
1495 0 : return HCCL_SUCCESS;
1496 : }
1497 :
1498 0 : HcclResult DispatcherAiCpu::AddFlipTask(Stream& stream)
1499 : {
1500 0 : HcclSqeContext* sqeContext = stream.GetSqeContextPtr();
1501 0 : CHK_PTR_NULL(sqeContext);
1502 0 : SqeRingBuffer* sqeContextBuffer = &(sqeContext->buffer);
1503 0 : CHK_PTR_NULL(sqeContextBuffer);
1504 0 : uint16_t flipNum = sqeContextBuffer->filpNum;
1505 0 : uint16_t taskId = sqeContextBuffer->tailSqeTaskId;
1506 :
1507 0 : if (callback_ != nullptr) {
1508 0 : hccl::FlipTaskPara para(stream.id(), taskId, flipNum);
1509 0 : hccl::TaskPara taskPara(TaskType::TASK_FLIP, para);
1510 0 : callback_(callBackUserPtr_, (void*)&taskPara, sizeof(struct TaskPara));
1511 0 : }
1512 :
1513 0 : const HcclComStreamInfo& streamInfo = stream.GetHcclStreamInfo();
1514 :
1515 0 : uint8_t* sqeBufferAddr = nullptr;
1516 0 : uint8_t* sqeTypeAddr = nullptr;
1517 0 : uint8_t* sqeDfxInfoAddr = nullptr;
1518 0 : CHK_RET(stream.GetNextSqeBufferAddr(sqeBufferAddr, sqeTypeAddr, sqeDfxInfoAddr, taskId));
1519 :
1520 0 : AicpuDfxInfo* const dfxInfo = (AicpuDfxInfo* const)sqeDfxInfoAddr;
1521 0 : dfxInfo->opRingBufferIdx = opRingBufferIdx_;
1522 0 : dfxInfo->remoteRank = INVALID_VALUE_RANKID;
1523 0 : dfxInfo->notifyId = INVALID_VALUE_RANKID;
1524 0 : addOneFlipPlaceHolderSqe_(streamInfo.actualStreamId, flipNum, taskId, sqeBufferAddr, sqeTypeAddr);
1525 :
1526 0 : PLF_CONFIG_INFO(
1527 : PLF_TASK, "%s para: taskId[%u] streamId[%u] flipNum[%u]", __func__, taskId, streamInfo.actualStreamId, flipNum);
1528 0 : return HCCL_SUCCESS;
1529 : }
1530 :
1531 0 : HcclResult DispatcherAiCpu::AddRetryPreamble(Stream& stream) { return AddFlipTask(stream); }
1532 :
1533 22 : void DispatcherAiCpu::SaveStreamInfo(hccl::Stream& stream)
1534 : {
1535 22 : const HcclComStreamInfo& streamInfo = stream.GetHcclStreamInfo();
1536 22 : std::lock_guard<std::recursive_mutex> lock(streamMapMutex_);
1537 22 : auto it = streamMap_.find(streamInfo.actualStreamId);
1538 22 : if (it == streamMap_.end()) {
1539 17 : streamMap_.insert({streamInfo.actualStreamId, stream});
1540 17 : HCCL_INFO("[DispatcherAiCpu][SaveStreamInfo] stream id[%d]", streamInfo.actualStreamId);
1541 5 : } else if (it->second.IsInvalid()) {
1542 : // stream id复用: 旧stream已销毁(IsInvalid=true), 用新stream覆盖, 否则dispatcher会一直跳过该id
1543 0 : it->second = stream;
1544 0 : HCCL_INFO(
1545 : "[DispatcherAiCpu][SaveStreamInfo] refresh invalidated stream id stream id[%d]", streamInfo.actualStreamId);
1546 : }
1547 44 : return;
1548 22 : }
1549 :
1550 2 : HcclResult DispatcherAiCpu::StreamSync(Stream& stream)
1551 : {
1552 2 : uint32_t head = 0;
1553 2 : uint32_t tail = 0;
1554 : const HcclComStreamInfo* streamInfo;
1555 2 : u64 startUsec = GetCurAicpuTimestamp();
1556 2 : u64 lastUsec = startUsec;
1557 2 : CHK_RET(stream.GetStreamInfo(streamInfo));
1558 :
1559 2 : CHK_RET(QuerySqStatusByType(aicpuInfo_.devId, streamInfo->sqId, DRV_SQCQ_PROP_SQ_TAIL, tail));
1560 2 : HCCL_INFO("StreamSync aicpu stream sqid[%d] tail[%u]", streamInfo->sqId, tail);
1561 : do {
1562 2 : CHK_RET(QuerySqStatusByType(aicpuInfo_.devId, streamInfo->sqId, DRV_SQCQ_PROP_SQ_HEAD, head));
1563 2 : u64 curUsec = GetCurAicpuTimestamp();
1564 2 : if (curUsec - startUsec > NANOSECOND_TO_SECOND * dfxTimeOutConfig_.sqeTimeOutTimeOut) {
1565 1 : HCCL_ERROR(
1566 : "stream sync timeout %lus. curhead:%u, curtail:%u, sqId:%d", dfxTimeOutConfig_.sqeTimeOutTimeOut, head,
1567 : tail, streamInfo->sqId);
1568 1 : return HCCL_E_TIMEOUT;
1569 : }
1570 :
1571 : // 等待下发阶段，每隔30s打印一次状态
1572 1 : if (curUsec - lastUsec > NANOSECOND_TO_SECOND * dfx::kPrintSqInterval) {
1573 0 : lastUsec = curUsec;
1574 0 : HCCL_RUN_INFO("[StreamSync]Current state. sqid:%d, head:%u, tail:%u", streamInfo->sqId, head, tail);
1575 : }
1576 1 : } while (head != tail);
1577 :
1578 1 : return HCCL_SUCCESS;
1579 : }
1580 :
1581 5 : u64 DispatcherAiCpu::CalcDbAddr(u32 dbindex)
1582 : {
1583 5 : u64 dbAddr = 0;
1584 5 : if (aicpuInfo_.devType == DevType::DEV_TYPE_910_93) {
1585 : // 910_93 HCCS_SW 组网
1586 2 : constexpr u64 roceBaseAddr = 0x202000000000ULL;
1587 2 : constexpr u64 roceVfDbCfg0Reg = 0x230ULL;
1588 2 : constexpr u64 chipAddrOffset = 0x20000000000ULL;
1589 2 : constexpr u64 dieAddrOffset = 0x10000000000ULL;
1590 2 : constexpr u32 dbDieIdMask = 0x00ff0000;
1591 2 : constexpr u32 dbDieIdShift = 16; // 16 is dbDieIdShift
1592 2 : dbAddr = roceBaseAddr + roceVfDbCfg0Reg + chipAddrOffset * aicpuInfo_.chipId
1593 2 : + dieAddrOffset * ((dbindex & dbDieIdMask) >> dbDieIdShift);
1594 : } else {
1595 3 : constexpr u64 roceBaseAddr = 0x2000000000ULL;
1596 3 : constexpr u64 roceVfDbCfg0Reg = 0x230ULL;
1597 3 : constexpr u64 chipAddrOffset = 0x80000000000ULL;
1598 3 : constexpr u64 dieAddrOffset = 0x10000000000ULL;
1599 3 : constexpr u32 dbDieIdMask = 0x00ff0000;
1600 3 : constexpr u32 dbDieIdShift = 16; // 16 is dbDieIdShift
1601 3 : dbAddr = roceBaseAddr + roceVfDbCfg0Reg + chipAddrOffset * aicpuInfo_.chipId
1602 3 : + dieAddrOffset * ((dbindex & dbDieIdMask) >> dbDieIdShift);
1603 : }
1604 :
1605 5 : HCCL_DEBUG(
1606 : "%s dbindex:%u, devType:%u, chipId:%lld, dbAddr:%llu", __func__, dbindex, aicpuInfo_.devType, aicpuInfo_.chipId,
1607 : dbAddr);
1608 5 : return dbAddr;
1609 : }
1610 :
1611 450 : void DispatcherAiCpu::InitTimeOutConfig()
1612 : {
1613 450 : dfxTimeOutConfig_.useCredit = false;
1614 450 : dfxTimeOutConfig_.sqeTimeOutTimeOut = GetMaxNotifyWaitTime();
1615 450 : dfxTimeOutConfig_.sqeCreditTimeOut = RT_STARS_NEVER_TIMEOUT_KERNEL_CREDIT;
1616 450 : dfxTimeOutConfig_.sqeWaitTimeOut = dfx::kKfcTimeOut;
1617 450 : dfxTimeOutConfig_.sqFullWaitTimeOut = dfx::kSqFullWaitTimeOut;
1618 450 : HCCL_INFO(
1619 : "[DispatcherAiCpu][InitTimeOutConfig]DFX timeout config init successfully with details: [%s]",
1620 : dfxTimeOutConfig_.ToString().c_str());
1621 450 : }
1622 : } // namespace hccl
|
