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