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 : #ifndef UB_MEM_TRANSPORT_LITE_H
12 : #define UB_MEM_TRANSPORT_LITE_H
13 :
14 : #include <vector>
15 : #include <map>
16 : #include <memory>
17 : #include <unordered_map>
18 : #include <mutex>
19 : #include "base_transport_lite_impl.h"
20 : #include "notify_lite.h"
21 : #include "task_param.h"
22 : #include "rmt_rma_buf_slice_lite.h"
23 : #include "rma_conn_lite.h"
24 : #include "kernel_param_lite.h"
25 : #include "hcomm_primitives.h"
26 : #include "ub_conn_lite.h"
27 : #include "rtsq_a5.h"
28 : #include "aicpu_task_utils.h"
29 :
30 : namespace hcomm {
31 : class AicpuTaskCacheEntry;
32 : }
33 :
34 : namespace Hccl {
35 :
36 : class UbTransportLiteImpl : public BaseTransportLiteImpl {
37 : public:
38 : explicit UbTransportLiteImpl(
39 : std::vector<char>& uniqueId,
40 : std::function<void(u32 streamId, u32 taskId, const TaskParam& taskParam)> callback);
41 :
42 : UbTransportLiteImpl(std::vector<char>& uniqueId);
43 : void Init(std::vector<char>& uniqueId);
44 :
45 : ~UbTransportLiteImpl() override;
46 :
47 : std::string Describe() const override;
48 :
49 : Buffer GetRmtBuffer(u32 index) override;
50 :
51 : Eid GetLocEid() const;
52 : Eid GetRmtEid() const;
53 : uint64_t GetJettyHandle() const;
54 : uint32_t GetJettyId() const;
55 :
56 : void Post(u32 index, const StreamLite& stream) override;
57 :
58 : void Wait(u32 index, const StreamLite& stream) override;
59 :
60 : void WaitWithTimeout(u32 index, const StreamLite& stream, u32 timeout) override;
61 :
62 : void Read(const RmaBufferLite& loc, const Buffer& rmt, const StreamLite& stream) override;
63 :
64 : void Write(const RmaBufferLite& loc, const Buffer& rmt, const StreamLite& stream) override;
65 :
66 : void ReadReduce(
67 : const RmaBufferLite& loc, const Buffer& rmt, const ReduceIn& reduceIn, const StreamLite& stream) override;
68 :
69 : void WriteReduce(
70 : const RmaBufferLite& loc, const Buffer& rmt, const ReduceIn& reduceIn, const StreamLite& stream) override;
71 :
72 : void WriteWithNotify(
73 : const RmaBufferLite& loc, const Buffer& rmt, const WithNotifyIn& withNotify, const StreamLite& stream) override;
74 :
75 : void WriteReduceWithNotify(
76 : const RmaBufferLite& loc, const Buffer& rmt, const ReduceIn& reduceIn, const WithNotifyIn& withNotify,
77 : const StreamLite& stream) override;
78 :
79 : void BatchOneSidedWrite(
80 : const vector<RmaBufSliceLite>& loc, const vector<RmtRmaBufSliceLite>& rmt, const StreamLite& stream) override;
81 :
82 : void BatchOneSidedRead(
83 : const vector<RmaBufSliceLite>& loc, const vector<RmtRmaBufSliceLite>& rmt, const StreamLite& stream) override;
84 :
85 : void BatchTransfer(
86 : const std::vector<RmaBufferLite>& loc, const std::vector<Buffer>& rmt,
87 : const std::vector<TransferOp>& transferOp, const StreamLite& stream) override;
88 : // 子类独有方法，支持所有操作类型，用于aicpu场景批量下发任务
89 : void BatchTransferAll(
90 : const std::vector<RmaBufferLite>& loc, const std::vector<Buffer>& rmt,
91 : const std::vector<TransferOp>& transferOp, const std::vector<uint32_t>& notifyIdxs, const StreamLite& stream);
92 :
93 : inline void BatchTransferAllWqe_(
94 : const std::vector<RmaBufferLite>& loc, const std::vector<Buffer>& rmt,
95 : const std::vector<TransferOp>& transferOp, const std::vector<uint32_t>& notifyIdxs, const StreamLite& stream,
96 : RmaConnLite* conn, u64& totalSize);
97 :
98 : void Drain(const StreamLite& stream) override;
99 :
100 : HcclResult BuildLocRmaBufferLite(const uintptr_t addr, const size_t size, RmaBufferLite& rmaBufferLite) override;
101 : HcclResult Fence() override;
102 :
103 : HcclResult Clean();
104 : HcclResult Resume(std::vector<char>& uniqueId);
105 0 : void SetTaskExceptionEnable(bool flag) { taskExceptionEnable_ = flag; }
106 :
107 : HcclResult ExecuteBatchTransfer(
108 : StreamLite* streamLitePtr, const HcommBatchTransferDesc* transferDescs, uint32_t transferDescNum);
109 :
110 : // 用于aicpu task cache
111 0 : inline HcclResult SetNeedCacheTaskCallback(std::function<bool()> callback)
112 : {
113 0 : CHK_PTR_NULL(callback);
114 0 : needCacheTaskCallback_ = callback;
115 0 : return HCCL_SUCCESS;
116 : }
117 0 : inline HcclResult SetAddWqeArrayCallback(std::function<HcclResult(
118 : UbConnLite*, UbTransportLiteImpl*, const std::vector<WqeTask>&,
119 : const uint32_t, const uint32_t, const bool, const DbSqeProfInfo&)>
120 : callback)
121 : {
122 0 : CHK_PTR_NULL(callback);
123 0 : addWqeArrayCallback_ = callback;
124 0 : return HCCL_SUCCESS;
125 : }
126 :
127 : std::function<void(u32, u32, const TaskParam&)> GetCallback() { return callback_; }
128 :
129 : friend class hcomm::AicpuTaskCacheEntry;
130 :
131 : private:
132 : u32 notifyNum{0};
133 : u32 bufferNum{0};
134 : u32 rmtbufferNum{0};
135 : u32 connNum{0};
136 : bool fence_{false};
137 : bool taskExceptionEnable_{true};
138 :
139 : struct RmtUbBufLite {
140 : u64 addr;
141 : u64 size;
142 : u32 tokenId;
143 : u32 tokenValue;
144 : u32 notifyId;
145 31 : std::string Describe() const
146 : {
147 31 : return StringFormat("RmtUbBufLite[addr=0x%llx, size=%llu, notifyId=%u]", addr, size, notifyId);
148 : }
149 : };
150 :
151 : struct LocUbBufLite {
152 : u64 addr;
153 : u64 size;
154 : u32 tokenId;
155 : u32 tokenValue;
156 3 : std::string Describe() const { return StringFormat("LocUbBufLite[addr=0x%llx, size=%llu]", addr, size); }
157 : };
158 :
159 : struct DrainNotify {
160 : u64 addr;
161 : u64 size;
162 : u32 tokenId;
163 : u32 tokenValue;
164 : u32 notifyId;
165 0 : std::string Describe() const
166 : {
167 0 : return StringFormat("DrainNotify[addr=0x%llx, size=0x%llx, notifyId=%u]", addr, size, notifyId);
168 : }
169 : };
170 :
171 : std::vector<char> wqeData; // connection返回的WQE内容
172 : ConnLiteOperationOut connOut; // connection的输出
173 :
174 : void ClearConnOut();
175 :
176 : using RmtUbBufLiteVec = std::vector<RmtUbBufLite>;
177 : using RmtUbBufLiteMap = std::map<uintptr_t, RmtUbBufLite>;
178 : using LocUbBufLiteMap = std::map<uintptr_t, LocUbBufLite>;
179 925 : MAKE_ENUM(RmaUbBufType, NOTIFY, BUFFER)
180 : RmtUbBufLiteVec rmtNotifyVec;
181 : RmtUbBufLiteVec rmtBufferVec;
182 : RmtUbBufLiteMap rmtBufferMap; // 性能优化使用
183 : LocUbBufLiteMap locBufferMap;
184 :
185 : RmtRmaBufSliceLite GetRmtNotifySliceLite(u32 index);
186 : RmtRmaBufSliceLite GetRmtRmaBufSliceLite(const Buffer& rmtBuf);
187 :
188 : RmaBufSliceLite GetRmaBufSlicelite(const RmaBufferLite& lite) const;
189 : RmtRmaBufSliceLite GetRmtRmaBufSliceLite(const RmaBufferLite& lite) const;
190 :
191 : std::vector<std::unique_ptr<NotifyLite>> locNotifyVec;
192 :
193 : std::mutex drainMtx_;
194 : DrainNotify drainNotify_{};
195 : RmtUbBufLite rmtDrainBuffer_{};
196 :
197 : // N秒快恢需要清理的两个资源
198 : std::vector<std::vector<char>> connUniqueIdVec;
199 : std::vector<RmaConnLite*> connVec;
200 :
201 : std::function<void(u32 streamId, u32 taskId, const TaskParam& taskParam)> callback_{nullptr};
202 :
203 : void ProfilingProcess(void* src, void* dst, u64 size, const StreamLite& stream, DmaOp dmaOp, u32 taskId);
204 :
205 : inline void
206 0 : BuildDbSqeProfInfoForProfilingProcess(void* src, void* dst, u64 size, DmaOp dmaOp, DbSqeProfInfo& dbSqeProfInfo)
207 : {
208 0 : FillDbSqeProfInfoDmaPub(dst, size, dmaOp, dbSqeProfInfo);
209 :
210 : // 构造DbSqeProfInfo (注意: 其他字段已在FillDbSqeProfInfo设置)
211 0 : dbSqeProfInfo.taskParamType = TaskParamType::TASK_UB;
212 0 : dbSqeProfInfo.srcAddr = reinterpret_cast<uint64_t>(src);
213 0 : }
214 :
215 : void ReduceProfilingProcess(
216 : void* src, void* dst, u64 size, const ReduceIn& reduceIn, const StreamLite& stream, u32 taskId);
217 :
218 0 : inline void BuildDbSqeProfInfoForReduceProfilingProcess(
219 : void* src, void* dst, u64 size, const ReduceIn& reduceIn, DbSqeProfInfo& dbSqeProfInfo)
220 : {
221 : // 构造DbSqeProfInfo
222 0 : dbSqeProfInfo.isValid = true;
223 0 : dbSqeProfInfo.taskParamType = TaskParamType::TASK_UB_REDUCE_INLINE;
224 0 : FillDbSqeProfInfoReducePub(src, dst, size, reduceIn, dbSqeProfInfo);
225 0 : }
226 :
227 : void ParseLocNotifyVec(std::vector<char>& data);
228 :
229 : void ParseRmtBufferVec(std::vector<char>& data, RmaUbBufType rmtType);
230 :
231 : void ParseLocBufferMap(std::vector<char>& data);
232 :
233 : void ParseDrainResource(std::vector<char>& data);
234 :
235 : void ParseConnVec(std::vector<char>& data);
236 :
237 : void BuildUbDbSendTask(const StreamLite& stream, const UbJettyLiteId& jettyLiteId, u32 pi);
238 :
239 : void BuildNotifyWaitTask(const StreamLite& stream, u32 notifyId);
240 :
241 : void CheckConnVec(const std::string& desc);
242 :
243 : void SetFenceConfig(SqeConfigLite& cfg);
244 :
245 : bool IsReportTask();
246 :
247 : void ExecProfiling(
248 : const RmaBufferLite& loc, const Buffer& rmt, const u64 totalSize,
249 : const BaseTransportLiteImpl::TransferOp& transferOp, const StreamLite& stream, u32 taskId);
250 :
251 0 : inline void BuildDbSqeProfInfoForExecProfiling(
252 : const RmaBufferLite& loc, const Buffer& rmt, const u64 totalSize,
253 : const BaseTransportLiteImpl::TransferOp& transferOp, DbSqeProfInfo& dbSqeProfInfo)
254 : {
255 0 : if (transferOp.reduceIn.reduceOp == ReduceOp::INVALID) {
256 0 : DmaOp dmaOp = DmaOp::HCCL_DMA_WRITE;
257 0 : if (transferOp.transType == TransferType::READ) {
258 0 : dmaOp = DmaOp::HCCL_DMA_READ;
259 : }
260 0 : BuildDbSqeProfInfoForProfilingProcess(
261 0 : reinterpret_cast<void*>(GetRmaBufSlicelite(loc).GetAddr()),
262 0 : reinterpret_cast<void*>(GetRmtRmaBufSliceLite(rmt).GetAddr()), totalSize, dmaOp, dbSqeProfInfo);
263 : } else {
264 0 : BuildDbSqeProfInfoForReduceProfilingProcess(
265 0 : reinterpret_cast<void*>(GetRmaBufSlicelite(loc).GetAddr()),
266 0 : reinterpret_cast<void*>(GetRmtRmaBufSliceLite(rmt).GetAddr()), totalSize, transferOp.reduceIn,
267 : dbSqeProfInfo);
268 : }
269 0 : }
270 :
271 11 : inline void AddTaskCallback(const StreamLite& stream, u32 taskId, const TaskParam& taskParam)
272 : {
273 11 : if (callback_ != nullptr) {
274 11 : callback_(stream.GetSqId(), taskId, taskParam);
275 : }
276 11 : }
277 :
278 5 : inline void FillTaskParamDmaPub(TaskParam& taskParam, void* dst, u64 size, DmaOp dmaOp) const
279 : {
280 5 : taskParam.taskPara.DMA.dst = dst;
281 5 : taskParam.taskPara.DMA.size = size;
282 5 : taskParam.taskPara.DMA.notifyID = INVALID_VALUE_NOTIFYID;
283 5 : taskParam.taskPara.DMA.notifyValue = 0xffffffff;
284 5 : taskParam.taskPara.DMA.linkType = DfxLinkType::UB;
285 5 : taskParam.taskPara.DMA.dmaOp = dmaOp;
286 5 : taskParam.taskPara.DMA.locEid = GetLocEid();
287 5 : taskParam.taskPara.DMA.rmtEid = GetRmtEid();
288 5 : }
289 :
290 0 : inline void FillDbSqeProfInfoDmaPub(void* dst, u64 size, DmaOp dmaOp, DbSqeProfInfo& dbSqeProfInfo) const
291 : {
292 : // 构造DbSqeProfInfo
293 0 : dbSqeProfInfo.isValid = true;
294 0 : dbSqeProfInfo.dstAddr = reinterpret_cast<uint64_t>(dst);
295 0 : dbSqeProfInfo.size = size;
296 0 : dbSqeProfInfo.dmaOp = dmaOp;
297 0 : dbSqeProfInfo.locEid = GetLocEid();
298 0 : dbSqeProfInfo.rmtEid = GetRmtEid();
299 0 : dbSqeProfInfo.jettyHandle = GetJettyHandle();
300 0 : dbSqeProfInfo.jettyId = GetJettyId();
301 0 : }
302 :
303 : inline void
304 2 : FillTaskParamReducePub(TaskParam& taskParam, void* src, void* dst, u64 size, const ReduceIn& reduceIn) const
305 : {
306 2 : taskParam.taskPara.Reduce.src = src;
307 2 : taskParam.taskPara.Reduce.dst = dst;
308 2 : taskParam.taskPara.Reduce.size = size;
309 2 : taskParam.taskPara.Reduce.notifyValue = 1;
310 2 : taskParam.taskPara.Reduce.linkType = DfxLinkType::UB;
311 2 : taskParam.taskPara.Reduce.reduceOp = ConvertReduceOpToHcclReduceOp(reduceIn.reduceOp);
312 2 : taskParam.taskPara.Reduce.dataType = DataTypeToHcclDataType(reduceIn.dataType);
313 2 : taskParam.taskPara.Reduce.locEid = GetLocEid();
314 2 : taskParam.taskPara.Reduce.rmtEid = GetRmtEid();
315 2 : }
316 :
317 0 : inline void FillDbSqeProfInfoReducePub(
318 : void* src, void* dst, u64 size, const ReduceIn& reduceIn, DbSqeProfInfo& dbSqeProfInfo) const
319 : {
320 0 : dbSqeProfInfo.srcAddr = reinterpret_cast<uint64_t>(src);
321 0 : dbSqeProfInfo.dstAddr = reinterpret_cast<uint64_t>(dst);
322 0 : dbSqeProfInfo.size = size;
323 0 : dbSqeProfInfo.locEid = GetLocEid();
324 0 : dbSqeProfInfo.rmtEid = GetRmtEid();
325 0 : dbSqeProfInfo.reduceOp = ConvertReduceOpToHcclReduceOp(reduceIn.reduceOp);
326 0 : dbSqeProfInfo.dataType = DataTypeToHcclDataType(reduceIn.dataType);
327 0 : dbSqeProfInfo.jettyHandle = GetJettyHandle();
328 0 : dbSqeProfInfo.jettyId = GetJettyId();
329 0 : }
330 :
331 : void ExecProfilingAll(
332 : const RmaBufferLite& loc, const Buffer& rmt, const u64 totalSize,
333 : const BaseTransportLiteImpl::TransferOp& transferOp, const StreamLite& stream, u32 taskId,
334 : const uint32_t notifyId);
335 :
336 0 : inline void BuildDbSqeProfInfoForExecProfilingAll(
337 : const RmaBufferLite& loc, const Buffer& rmt, const u64 totalSize,
338 : const BaseTransportLiteImpl::TransferOp& transferOp, const uint32_t notifyId, DbSqeProfInfo& dbSqeProfInfo)
339 : {
340 0 : if (transferOp.transType == TransferType::READ) {
341 0 : BuildDbSqeProfInfoForProfilingProcess(
342 0 : reinterpret_cast<void*>(GetRmaBufSlicelite(loc).GetAddr()),
343 0 : reinterpret_cast<void*>(GetRmtRmaBufSliceLite(rmt).GetAddr()), totalSize, DmaOp::HCCL_DMA_READ,
344 : dbSqeProfInfo);
345 0 : } else if (transferOp.transType == TransferType::WRITE) {
346 0 : BuildDbSqeProfInfoForProfilingProcess(
347 0 : reinterpret_cast<void*>(GetRmaBufSlicelite(loc).GetAddr()),
348 0 : reinterpret_cast<void*>(GetRmtRmaBufSliceLite(rmt).GetAddr()), totalSize, DmaOp::HCCL_DMA_WRITE,
349 : dbSqeProfInfo);
350 0 : } else if (transferOp.transType == TransferType::READ_REDUCE) {
351 0 : BuildDbSqeProfInfoForReduceProfilingProcess(
352 0 : reinterpret_cast<void*>(GetRmaBufSlicelite(loc).GetAddr()),
353 0 : reinterpret_cast<void*>(GetRmtRmaBufSliceLite(rmt).GetAddr()), totalSize, transferOp.reduceIn,
354 : dbSqeProfInfo);
355 0 : } else if (transferOp.transType == TransferType::WRITE_REDUCE) {
356 0 : BuildDbSqeProfInfoForReduceProfilingProcess(
357 0 : reinterpret_cast<void*>(GetRmaBufSlicelite(loc).GetAddr()),
358 0 : reinterpret_cast<void*>(GetRmtRmaBufSliceLite(rmt).GetAddr()), totalSize, transferOp.reduceIn,
359 : dbSqeProfInfo);
360 0 : } else if (transferOp.transType == TransferType::WRITE_WITH_NOTIFY) {
361 0 : BuildDbSqeProfInfoForWriteWithNotify(
362 0 : reinterpret_cast<void*>(GetRmaBufSlicelite(loc).GetAddr()),
363 0 : reinterpret_cast<void*>(GetRmtRmaBufSliceLite(rmt).GetAddr()), totalSize,
364 0 : GetRmtNotifySliceLite(notifyId).GetAddr(), dbSqeProfInfo);
365 0 : } else if (transferOp.transType == TransferType::WRITE_REDUCE_WITH_NOTIFY) {
366 0 : BuildDbSqeProfInfoForWriteReduceWithNotify(
367 0 : reinterpret_cast<void*>(GetRmaBufSlicelite(loc).GetAddr()),
368 0 : reinterpret_cast<void*>(GetRmtRmaBufSliceLite(rmt).GetAddr()), totalSize, transferOp.reduceIn,
369 0 : GetRmtNotifySliceLite(notifyId).GetAddr(), dbSqeProfInfo);
370 0 : } else if (transferOp.transType == TransferType::NOTIFY_RECORD) {
371 0 : BuildDbSqeProfInfoForNotifyRecord(
372 0 : reinterpret_cast<void*>(GetRmtNotifySliceLite(notifyId).GetAddr()),
373 0 : GetRmtNotifySliceLite(notifyId).GetSize(), GetRmtNotifySliceLite(notifyId).GetAddr(), dbSqeProfInfo);
374 : }
375 0 : }
376 :
377 : void
378 : WriteWithNotifyProfilingProcess(void* src, void* dst, u64 size, const StreamLite& stream, u32 taskId, u64 notifyId);
379 :
380 : inline void
381 0 : BuildDbSqeProfInfoForWriteWithNotify(void* src, void* dst, u64 size, u64 notifyId, DbSqeProfInfo& dbSqeProfInfo)
382 : {
383 0 : FillDbSqeProfInfoDmaPub(dst, size, DmaOp::HCCL_DMA_WRITE, dbSqeProfInfo);
384 :
385 : // 构造DbSqeProfInfo (注意: 其他字段已在FillDbSqeProfInfo设置)
386 0 : dbSqeProfInfo.taskParamType = TaskParamType::TASK_WRITE_WITH_NOTIFY;
387 0 : dbSqeProfInfo.srcAddr = reinterpret_cast<uint64_t>(src);
388 0 : dbSqeProfInfo.notifyId = notifyId;
389 0 : }
390 :
391 : void WriteReduceWithNotifyProfilingProcess(
392 : void* src, void* dst, u64 size, const ReduceIn& reduceIn, const StreamLite& stream, u32 taskId, u64 notifyId);
393 :
394 0 : inline void BuildDbSqeProfInfoForWriteReduceWithNotify(
395 : void* src, void* dst, u64 size, const ReduceIn& reduceIn, u64 notifyId, DbSqeProfInfo& dbSqeProfInfo)
396 : {
397 : // 构造DbSqeProfInfo
398 0 : dbSqeProfInfo.isValid = true;
399 0 : dbSqeProfInfo.taskParamType = TaskParamType::TASK_WRITE_REDUCE_WITH_NOTIFY;
400 0 : FillDbSqeProfInfoReducePub(src, dst, size, reduceIn, dbSqeProfInfo);
401 0 : dbSqeProfInfo.notifyId = notifyId;
402 0 : }
403 :
404 : void NotifyRecordProfilingProcess(void* dst, u64 size, const StreamLite& stream, u32 taskId, u64 notifyId);
405 :
406 0 : inline void BuildDbSqeProfInfoForNotifyRecord(void* dst, u64 size, u64 notifyId, DbSqeProfInfo& dbSqeProfInfo)
407 : {
408 0 : FillDbSqeProfInfoDmaPub(dst, size, DmaOp::HCCL_DMA_WRITE, dbSqeProfInfo);
409 :
410 : // 构造DbSqeProfInfo (注意: 其他字段已在FillDbSqeProfInfo设置)
411 0 : dbSqeProfInfo.taskParamType = TaskParamType::TASK_UB_INLINE_WRITE;
412 0 : dbSqeProfInfo.notifyId = notifyId;
413 0 : }
414 :
415 : // 用于aicpu task cache
416 : std::function<bool()> needCacheTaskCallback_{nullptr};
417 : std::function<HcclResult(
418 : UbConnLite*, UbTransportLiteImpl*, const std::vector<WqeTask>&, const uint32_t, const uint32_t, const bool,
419 : const DbSqeProfInfo& dbSqeProfInfo)>
420 : addWqeArrayCallback_{nullptr};
421 :
422 : // 展开下发WQE前，按需设置wqe tasks
423 0 : inline void PreLaunchWqe(UbConnLite*& ubConnLitePtr, bool& needCacheTask, RmaConnLite* connPtr)
424 : {
425 : // 校验needCacheTaskCallback_
426 : // 注意: A5新流程下needCacheTaskCallback_一定非空; 但A5老流程下不支持aicpu task cache,
427 : // needCacheTaskCallback_为空;
428 : // 为避免A5老流程报错, 这里为空时跳过执行而非报错
429 0 : needCacheTask = false;
430 0 : if (UNLIKELY(needCacheTaskCallback_ == nullptr)) {
431 0 : HCCL_WARNING(
432 : "[UbTransportLiteImpl][PreLaunchWqe] needCacheTaskCallback_ is null, keep needCacheTask as false");
433 : } else {
434 0 : needCacheTask = needCacheTaskCallback_();
435 : }
436 :
437 : // 校验是否需要打印WQE
438 0 : bool needDumpWqe = false;
439 0 : if ((UNLIKELY(GetPlfDebugConfigValue() & PLF_TASK)) || UNLIKELY(HcclCheckLogLevel(HCCL_LOG_DEBUG))) {
440 0 : needDumpWqe = true;
441 : }
442 :
443 : // 如果需要缓存WQE 或者 打印WQE
444 0 : if (needCacheTask || UNLIKELY(needDumpWqe)) {
445 : // 校验connPtr
446 0 : if (UNLIKELY(connPtr == nullptr)) {
447 0 : THROW<InternalException>("[UbTransportLiteImpl][PreLaunchWqe] connPtr is null");
448 : }
449 :
450 : // 转换ubConnLitePtr并校验
451 0 : ubConnLitePtr = dynamic_cast<UbConnLite*>(connPtr);
452 0 : if (UNLIKELY(ubConnLitePtr == nullptr)) {
453 0 : THROW<InternalException>("[UbTransportLiteImpl][PreLaunchWqe] ubConnLitePtr is null");
454 : }
455 :
456 0 : HcclResult ret = ubConnLitePtr->EnableWqeTasks();
457 0 : if (UNLIKELY(ret != HCCL_SUCCESS)) {
458 0 : THROW<InternalException>(
459 : "[UbTransportLiteImpl][PreLaunchWqe] "
460 : "ubConnLitePtr->EnableWqeTasks failed, ret %d",
461 : ret);
462 : }
463 : }
464 0 : }
465 :
466 : // 展开下发WQE后, 展开下发DbSqe前, 按需缓存wqe及DbSqeIdx 或者 打印正常展开的WQE
467 0 : inline void PostLaunchWqe(
468 : const StreamLite& stream, UbConnLite* ubConnLitePtr, bool needCacheTask, const uint32_t pendingSqeCnt,
469 : const bool isReportTask, const DbSqeProfInfo& dbSqeProfInfo)
470 : {
471 : // 校验是否需要打印WQE
472 0 : bool needDumpWqe = false;
473 0 : if ((UNLIKELY(GetPlfDebugConfigValue() & PLF_TASK)) || UNLIKELY(HcclCheckLogLevel(HCCL_LOG_DEBUG))) {
474 0 : needDumpWqe = true;
475 : }
476 :
477 : // 如果需要缓存WQE 或者 打印WQE
478 0 : if (needCacheTask || UNLIKELY(needDumpWqe)) {
479 : // 校验ubConnLitePtr
480 0 : if (UNLIKELY(ubConnLitePtr == nullptr)) {
481 0 : THROW<InternalException>("[UbTransportLiteImpl][PostLaunchWqe] ubConnLitePtr is null");
482 : }
483 :
484 0 : HcclResult ret = HCCL_SUCCESS;
485 :
486 : // 按需缓存WQE
487 0 : if (needCacheTask) {
488 : // 校验addWqeArrayCallback_
489 : // 注意: 如果needCacheTask为true, 一定是A5新流程, 所以addWqeArrayCallback_一定非空
490 0 : if (UNLIKELY(addWqeArrayCallback_ == nullptr)) {
491 0 : THROW<InternalException>("[UbTransportLiteImpl][PostLaunchWqe] addWqeArrayCallback_ is null");
492 : }
493 :
494 : // 调用addWqeArrayCallback_函数, 缓存wqe
495 : // 注意: pendingSqeCnt即下发DbSqe前, SqeRingBuffer的tailSqeIdx
496 0 : ret = addWqeArrayCallback_(
497 : ubConnLitePtr, this, ubConnLitePtr->GetWqeTasks(), stream.GetId(), pendingSqeCnt, isReportTask,
498 : dbSqeProfInfo);
499 0 : if (UNLIKELY(ret != HCCL_SUCCESS)) {
500 0 : THROW<InternalException>(
501 : "[UbTransportLiteImpl][PostLaunchWqe] "
502 : "addWqeArrayCallback_ failed, ret %d",
503 : ret);
504 : }
505 : }
506 :
507 : // 按需打印WQE
508 0 : if (UNLIKELY(needDumpWqe)) {
509 0 : const std::vector<WqeTask>& wqeTasks = ubConnLitePtr->GetWqeTasks();
510 0 : const uint64_t wqeCount = wqeTasks.size();
511 0 : PLF_CONFIG_DEBUG(
512 : PLF_TASK,
513 : "[UbTransportLiteImpl][PostLaunchWqe] dump %llu generated WQEs "
514 : "in jetty[%u, %u, %u]",
515 : wqeCount, ubConnLitePtr->GetUbJettyLiteId().GetDieId(),
516 : ubConnLitePtr->GetUbJettyLiteId().GetFuncId(), ubConnLitePtr->GetUbJettyLiteId().GetJettyId());
517 0 : for (size_t wqeIdx = 0; wqeIdx < wqeCount; wqeIdx++) {
518 0 : PLF_CONFIG_DEBUG(
519 : PLF_TASK,
520 : "[UbTransportLiteImpl][PostLaunchWqe] %uth generated WQE "
521 : "in jetty[%u, %u, %u]",
522 : wqeIdx, ubConnLitePtr->GetUbJettyLiteId().GetDieId(),
523 : ubConnLitePtr->GetUbJettyLiteId().GetFuncId(), ubConnLitePtr->GetUbJettyLiteId().GetJettyId());
524 0 : ret = hcomm::AicpuTaskUtils::DumpWqeContent(reinterpret_cast<const uint8_t*>(&wqeTasks[wqeIdx]));
525 0 : if (UNLIKELY(ret != HCCL_SUCCESS)) {
526 0 : THROW<InternalException>(
527 : "[UbTransportLiteImpl][PostLaunchWqe] "
528 : "AicpuTaskUtils::DumpWqeContent failed, ret %d",
529 : ret);
530 : }
531 : }
532 : }
533 :
534 : // 缓存或者打印后清理wqe tasks
535 0 : ret = ubConnLitePtr->DisableWqeTasks();
536 0 : if (UNLIKELY(ret != HCCL_SUCCESS)) {
537 0 : THROW<InternalException>(
538 : "[UbTransportLiteImpl][PostLaunchWqe] "
539 : "ubConnLitePtr->DisableWqeTasks failed, ret %d",
540 : ret);
541 : }
542 : }
543 0 : }
544 :
545 : void FillSlotUbDmaInfo(
546 : DfxTaskInfo* slot, const StreamLite& stream, u32 taskId, u64 srcAddr, u64 dstAddr, u64 size, u32 notifyId);
547 : void FillSlotReduceInfo(
548 : DfxTaskInfo* slot, const StreamLite& stream, u32 taskId, u64 srcAddr, u64 dstAddr, u64 size, u32 notifyId,
549 : u8 reduceOp);
550 : void ReportWriteWithNotifyTask(
551 : const RmaBufSliceLite& locSlice, const RmtRmaBufSliceLite& rmtSlice, const RmtRmaBufSliceLite& rmtNotifySlice,
552 : const StreamLite& stream, u32 taskId);
553 : void ReportWriteReduceWithNotifyTask(
554 : const RmaBufSliceLite& locSlice, const RmtRmaBufSliceLite& rmtSlice, const RmtRmaBufSliceLite& rmtNotifySlice,
555 : const ReduceIn& reduceIn, const StreamLite& stream, u32 taskId);
556 : };
557 :
558 : } // namespace Hccl
559 : #endif
|
