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 "ub_transport_lite_impl.h"
12 : #include "binary_stream.h"
13 : #include "ub_conn_lite_mgr.h"
14 : #include "exception_util.h"
15 : #include "internal_exception.h"
16 : #include "communicator_impl_lite_manager.h"
17 : #include "dfx_profiling_handler_lite.h"
18 :
19 : namespace Hccl {
20 : constexpr u32 UB_WQE_MAX_SIZE = 128; // 针对WriteWithNotify类型WQE，最大是128Byte
21 : constexpr u32 UB_INLINE_WRITE_SIZE = 4;
22 : constexpr u32 UB_RELAX_ORDER = 0X01; // Relax Order表示当前SQE与后续Strong Order SQE有保序要求
23 : constexpr u32 UB_STRONG_ORDER = 0X02; // Strong Order表示当前SQE有保序要求，该SQE不能超越前面的Relax Order SQE
24 : constexpr u32 UB_NO_COMPLETION = 0; // 表示当前报文和前面报文没有completion序要求，报文对应的CQE可以乱序上报
25 : constexpr u32 UB_COMPLETION = 1; // 表示当前报文和前面报文有completion序要求，报文对应的CQE需要保序上报
26 : constexpr u8 UB_FENCE_ENABLED = 1; // fence使能
27 54 : UbTransportLiteImpl::UbTransportLiteImpl(
28 54 : std::vector<char>& uniqueId, std::function<void(u32 streamId, u32 taskId, const TaskParam& taskParam)> callback)
29 : {
30 54 : callback_ = callback;
31 : // [header...][notifyUniqueId...][rmtNotifyUniqueId...][rmtBufferUniqueIds...]
32 54 : BinaryStream binaryStream(uniqueId);
33 : u32 theType;
34 54 : binaryStream >> theType;
35 54 : binaryStream >> notifyNum;
36 54 : binaryStream >> bufferNum;
37 54 : binaryStream >> rmtbufferNum;
38 54 : binaryStream >> connNum;
39 54 : linkType_ = (theType == static_cast<u32>(TransportType::UB)) ? DfxLinkType::UB : DfxLinkType::UBoE;
40 :
41 54 : std::vector<char> notifyUniqueIds;
42 54 : binaryStream >> notifyUniqueIds;
43 54 : ParseLocNotifyVec(notifyUniqueIds);
44 :
45 54 : std::vector<char> rmtNotifyUniqueIds;
46 54 : binaryStream >> rmtNotifyUniqueIds;
47 54 : ParseRmtBufferVec(rmtNotifyUniqueIds, RmaUbBufType::NOTIFY);
48 :
49 54 : std::vector<char> rmtBufferUniqueIds;
50 54 : binaryStream >> rmtBufferUniqueIds;
51 54 : ParseRmtBufferVec(rmtBufferUniqueIds, RmaUbBufType::BUFFER);
52 :
53 54 : std::vector<char> connUniqueIds;
54 54 : binaryStream >> connUniqueIds;
55 54 : ParseConnVec(connUniqueIds);
56 54 : }
57 3 : UbTransportLiteImpl::UbTransportLiteImpl(std::vector<char>& uniqueId) { Init(uniqueId); }
58 :
59 3 : void UbTransportLiteImpl::Init(std::vector<char>& uniqueId)
60 : {
61 3 : BinaryStream binaryStream(uniqueId);
62 : u32 theType;
63 3 : binaryStream >> theType;
64 3 : binaryStream >> notifyNum;
65 3 : binaryStream >> bufferNum;
66 3 : binaryStream >> rmtbufferNum;
67 3 : binaryStream >> connNum;
68 3 : linkType_ = (theType == static_cast<u32>(TransportType::UB)) ? DfxLinkType::UB : DfxLinkType::UBoE;
69 :
70 3 : std::vector<char> notifyUniqueIds;
71 3 : binaryStream >> notifyUniqueIds;
72 3 : ParseLocNotifyVec(notifyUniqueIds);
73 :
74 3 : std::vector<char> rmtNotifyUniqueIds;
75 3 : binaryStream >> rmtNotifyUniqueIds;
76 3 : ParseRmtBufferVec(rmtNotifyUniqueIds, RmaUbBufType::NOTIFY);
77 :
78 3 : std::vector<char> locBufferUniqueIds;
79 3 : binaryStream >> locBufferUniqueIds;
80 3 : ParseLocBufferMap(locBufferUniqueIds);
81 :
82 3 : std::vector<char> rmtBufferUniqueIds;
83 3 : binaryStream >> rmtBufferUniqueIds;
84 3 : ParseRmtBufferVec(rmtBufferUniqueIds, RmaUbBufType::BUFFER);
85 :
86 : // 解析drain相关的资源信息
87 3 : std::vector<char> drainBufferUniqueIds;
88 3 : binaryStream >> drainBufferUniqueIds;
89 3 : ParseDrainResource(drainBufferUniqueIds);
90 :
91 3 : std::vector<char> connUniqueIds;
92 3 : binaryStream >> connUniqueIds;
93 3 : ParseConnVec(connUniqueIds);
94 3 : }
95 :
96 112 : UbTransportLiteImpl::~UbTransportLiteImpl()
97 : {
98 62 : for (auto& it : connUniqueIdVec) {
99 5 : DECTOR_TRY_CATCH("UbTransportLiteImpl", UbConnLiteMgr::GetInstance().Clear(it));
100 : }
101 112 : }
102 :
103 1 : std::string UbTransportLiteImpl::Describe() const
104 : {
105 1 : std::string desc = "UbTransportLiteImpl[";
106 :
107 1 : u32 idx = 0;
108 1 : desc += "locNotifyVec=[";
109 3 : for (auto& it : locNotifyVec) {
110 2 : desc += StringFormat("idx=%u, %s;", idx, it->Describe().c_str());
111 2 : idx++;
112 : }
113 :
114 1 : idx = 0;
115 1 : desc += "], rmtNotifyVec=[";
116 3 : for (auto& it : rmtNotifyVec) {
117 2 : desc += StringFormat("idx=%u, %s;", idx, it.Describe().c_str());
118 2 : idx++;
119 : }
120 :
121 1 : idx = 0;
122 1 : desc += "], rmtBufferVec=[";
123 3 : for (auto& it : rmtBufferVec) {
124 2 : desc += StringFormat("idx=%u, %s;", idx, it.Describe().c_str());
125 2 : idx++;
126 : }
127 :
128 1 : idx = 0;
129 1 : desc += "], connVec=[";
130 2 : for (auto& it : connVec) {
131 1 : desc += StringFormat("idx=%u, %s;", idx, it->Describe().c_str());
132 1 : idx++;
133 : }
134 :
135 1 : desc += "]]";
136 1 : return desc;
137 0 : }
138 :
139 57 : void UbTransportLiteImpl::ParseLocNotifyVec(std::vector<char>& data)
140 : {
141 57 : if (notifyNum == 0) {
142 152 : HCCL_WARNING("UbTransportLiteImpl::ParseLocNotifyVec num is 0");
143 52 : return;
144 : }
145 5 : u32 notifySizePerDto = data.size() / notifyNum;
146 :
147 15 : for (u32 idx = 0; idx < notifyNum; idx++) {
148 10 : auto start = data.begin() + idx * notifySizePerDto;
149 10 : auto end = start + notifySizePerDto;
150 10 : std::vector<char> dto(start, end);
151 10 : locNotifyVec.push_back(std::make_unique<NotifyLite>(dto));
152 26 : HCCL_INFO("locNotify idx=%u, %s", idx, locNotifyVec.back()->Describe().c_str());
153 10 : }
154 : }
155 :
156 114 : void UbTransportLiteImpl::ParseRmtBufferVec(std::vector<char>& data, RmaUbBufType rmtType)
157 : {
158 114 : u32 num = 0;
159 114 : if (rmtType == RmaUbBufType::NOTIFY) {
160 57 : num = notifyNum;
161 : } else {
162 57 : num = rmtbufferNum;
163 : }
164 :
165 114 : if (num == 0) {
166 304 : HCCL_WARNING("UbTransportLiteImpl::ParseRmtBufferVec %s num is 0", rmtType.Describe().c_str());
167 104 : return;
168 : }
169 :
170 10 : u32 rmtBufferSizePerDto = data.size() / num;
171 26 : HCCL_INFO("Parse %s num=%u, sizePerDto=%u", rmtType.Describe().c_str(), num, rmtBufferSizePerDto);
172 10 : BinaryStream binaryStream(data);
173 :
174 33 : for (u32 idx = 0; idx < num; idx++) {
175 : RmtUbBufLite ubBufLite;
176 23 : binaryStream >> ubBufLite.addr;
177 23 : binaryStream >> ubBufLite.size;
178 23 : binaryStream >> ubBufLite.tokenId;
179 23 : binaryStream >> ubBufLite.tokenValue;
180 23 : binaryStream >> ubBufLite.notifyId;
181 55 : HCCL_INFO("idx=%u, %s %s", idx, rmtType.Describe().c_str(), ubBufLite.Describe().c_str());
182 23 : if (rmtType == RmaUbBufType::NOTIFY) {
183 10 : rmtNotifyVec.push_back(ubBufLite);
184 : } else {
185 13 : rmtBufferMap[static_cast<uintptr_t>(ubBufLite.addr)] = ubBufLite;
186 13 : rmtBufferVec.push_back(ubBufLite);
187 : }
188 : }
189 10 : }
190 :
191 3 : void UbTransportLiteImpl::ParseLocBufferMap(std::vector<char>& data)
192 : {
193 3 : u32 num = bufferNum;
194 :
195 3 : if (num == 0) {
196 2 : HCCL_WARNING("UbTransportLiteImpl::ParseLocBufferMap num is 0");
197 2 : return;
198 : }
199 :
200 1 : u32 rmtBufferSizePerDto = data.size() / num;
201 1 : HCCL_INFO("ParseLocBufferMap num=%u, sizePerDto=%u", num, rmtBufferSizePerDto);
202 1 : BinaryStream binaryStream(data);
203 :
204 4 : for (u32 idx = 0; idx < num; idx++) {
205 : LocUbBufLite ubBufLite;
206 3 : binaryStream >> ubBufLite.addr;
207 3 : binaryStream >> ubBufLite.size;
208 3 : binaryStream >> ubBufLite.tokenId;
209 3 : binaryStream >> ubBufLite.tokenValue;
210 3 : HCCL_INFO("idx=%u, LocBuffer %s", idx, ubBufLite.Describe().c_str());
211 3 : locBufferMap[static_cast<uintptr_t>(ubBufLite.addr)] = ubBufLite;
212 : }
213 1 : }
214 :
215 3 : void UbTransportLiteImpl::ParseDrainResource(std::vector<char>& data)
216 : {
217 3 : if (data.size() == 0) {
218 3 : HCCL_WARNING("UbTransportLiteImpl::ParseDrainResource is null");
219 3 : return;
220 : }
221 :
222 0 : BinaryStream binaryStream(data);
223 0 : binaryStream >> drainNotify_.addr;
224 0 : binaryStream >> drainNotify_.size;
225 0 : binaryStream >> drainNotify_.tokenId;
226 0 : binaryStream >> drainNotify_.tokenValue;
227 0 : binaryStream >> drainNotify_.notifyId;
228 0 : HCCL_INFO("drain notify %s", drainNotify_.Describe().c_str());
229 :
230 0 : binaryStream >> rmtDrainBuffer_.addr;
231 0 : binaryStream >> rmtDrainBuffer_.size;
232 0 : binaryStream >> rmtDrainBuffer_.tokenId;
233 0 : binaryStream >> rmtDrainBuffer_.tokenValue;
234 0 : binaryStream >> rmtDrainBuffer_.notifyId;
235 0 : HCCL_INFO("drain remote buffer %s", rmtDrainBuffer_.Describe().c_str());
236 0 : }
237 :
238 57 : void UbTransportLiteImpl::ParseConnVec(std::vector<char>& data)
239 : {
240 57 : if (connNum == 0) {
241 152 : HCCL_WARNING("UbTransportLiteImpl::ParseConnVec num is 0");
242 52 : return;
243 : }
244 5 : u32 connSizePerDto = data.size() / connNum;
245 13 : HCCL_INFO("Parse ConnVec num=%u, connSizePerDto=%u", connNum, connSizePerDto);
246 10 : for (u32 idx = 0; idx < connNum; idx++) {
247 5 : auto start = data.begin() + idx * connSizePerDto;
248 5 : auto end = start + connSizePerDto;
249 5 : std::vector<char> connUniqueId(start, end);
250 5 : connUniqueIdVec.push_back(connUniqueId);
251 : // connLite的复用由 ubConnLiteMgr管理
252 5 : auto lite = UbConnLiteMgr::GetInstance().Get(connUniqueId);
253 5 : connVec.push_back(lite);
254 13 : HCCL_INFO("[%s]idx=%u, %s", __func__, idx, lite->Describe().c_str());
255 5 : }
256 10 : CheckConnVec("after ParseConnVec");
257 : }
258 :
259 0 : void UbTransportLiteImpl::BuildUbDbSendTask(const StreamLite& stream, const UbJettyLiteId& jettyLiteId, u32 pi)
260 : {
261 0 : stream.GetRtsq()->UbDbSend(jettyLiteId, pi);
262 0 : }
263 :
264 0 : void UbTransportLiteImpl::BuildNotifyWaitTask(const StreamLite& stream, u32 notifyId)
265 : {
266 0 : stream.GetRtsq()->NotifyWait(notifyId);
267 0 : }
268 :
269 1 : Buffer UbTransportLiteImpl::GetRmtBuffer(u32 index)
270 : {
271 1 : if (UNLIKELY(index >= rmtBufferVec.size())) {
272 0 : THROW<InternalException>(StringFormat(
273 : "UbTransportLiteImpl::GetRmtBuffer out-of-bounds. index=%u, size=%u", index, rmtBufferVec.size()));
274 : }
275 1 : return Buffer(rmtBufferVec[index].addr, rmtBufferVec[index].size);
276 : }
277 :
278 2 : RmtRmaBufSliceLite UbTransportLiteImpl::GetRmtNotifySliceLite(u32 index)
279 : {
280 2 : RmtUbBufLite& lite = rmtNotifyVec[index];
281 : // ub conn lite 不关心rkey , rkey 设定为0
282 2 : return RmtRmaBufSliceLite(lite.addr, lite.size, 0, lite.tokenId, lite.tokenValue, lite.notifyId);
283 : }
284 :
285 7 : RmtRmaBufSliceLite UbTransportLiteImpl::GetRmtRmaBufSliceLite(const Buffer& rmtBuf)
286 : {
287 7 : auto it = rmtBufferMap.upper_bound(rmtBuf.GetAddr());
288 :
289 7 : while (it != rmtBufferMap.begin()) {
290 7 : --it;
291 7 : Buffer iterBuf(it->second.addr, it->second.size);
292 7 : if (iterBuf.Contains(rmtBuf.GetAddr(), rmtBuf.GetSize())) {
293 : return RmtRmaBufSliceLite(
294 14 : rmtBuf.GetAddr(), rmtBuf.GetSize(), 0, it->second.tokenId, it->second.tokenValue, UINT32_MAX);
295 : }
296 7 : }
297 0 : MACRO_THROW(InternalException, StringFormat("%s is not in current transport", rmtBuf.Describe().c_str()));
298 : }
299 :
300 0 : RmtRmaBufSliceLite UbTransportLiteImpl::GetRmtRmaBufSliceLite(const RmaBufferLite& lite) const
301 : {
302 0 : return RmtRmaBufSliceLite(lite.GetAddr(), lite.GetSize(), 0, lite.GetTokenId(), lite.GetTokenValue(), UINT32_MAX);
303 : }
304 :
305 : HcclResult
306 0 : UbTransportLiteImpl::BuildLocRmaBufferLite(const uintptr_t addr, const size_t size, RmaBufferLite& rmaBufferLite)
307 : {
308 0 : HCCL_INFO(
309 : "[UbTransportLiteImpl::%s] start to find addr[0x%llx], size[0x%llx] in locBufferMap, whose size is %zu. ",
310 : __func__, addr, size, locBufferMap.size());
311 0 : if (locBufferMap.empty()) {
312 0 : HCCL_ERROR("[UbTransportLiteImpl::%s] locBufferMap is empty.", __func__);
313 0 : return HCCL_E_INTERNAL;
314 : }
315 :
316 0 : bool isAddrInRange = false;
317 0 : auto it = locBufferMap.upper_bound(addr);
318 :
319 0 : while (it != locBufferMap.begin()) {
320 0 : --it;
321 0 : Buffer iterBuf(it->second.addr, it->second.size);
322 0 : if (iterBuf.Contains(addr, size)) {
323 0 : rmaBufferLite = RmaBufferLite(addr, size, it->second.tokenId, it->second.tokenValue);
324 0 : isAddrInRange = true;
325 0 : break;
326 : }
327 0 : }
328 :
329 0 : if (!isAddrInRange) {
330 0 : HCCL_WARNING(
331 : "[UbTransportLiteImpl::%s] addr[0x%llx], size[0x%llx] not in any range of locBufferMap, use the first in "
332 : "map addr[0x%llx] size[0x%llx]",
333 : __func__, addr, size, it->second.addr, it->second.size);
334 0 : rmaBufferLite = RmaBufferLite(addr, size, it->second.tokenId, it->second.tokenValue);
335 : }
336 :
337 0 : return HCCL_SUCCESS;
338 : }
339 :
340 0 : void UbTransportLiteImpl::ClearConnOut()
341 : {
342 0 : wqeData.clear();
343 0 : wqeData.resize(UB_WQE_MAX_SIZE);
344 0 : connOut.data = (u8*)wqeData.data();
345 0 : connOut.dataSize = sizeof(wqeData);
346 0 : }
347 :
348 : // 检查connection不能为空
349 5 : void UbTransportLiteImpl::CheckConnVec(const std::string& desc)
350 : {
351 5 : if (UNLIKELY(connVec.size() == 0)) {
352 0 : THROW<InternalException>(StringFormat("connVec size is 0 %s", desc.c_str()));
353 : }
354 :
355 5 : u32 idx = 0;
356 10 : for (auto& it : connVec) {
357 5 : if (UNLIKELY(it == nullptr)) {
358 0 : THROW<InternalException>(StringFormat("connVec[%u] is null %s", idx, desc.c_str()));
359 : }
360 5 : idx++;
361 : }
362 5 : }
363 :
364 8 : RmaBufSliceLite UbTransportLiteImpl::GetRmaBufSlicelite(const RmaBufferLite& lite) const
365 : {
366 : // ub conn lite 不关心rkey , rkey 设定为0
367 8 : return RmaBufSliceLite(lite.GetAddr(), lite.GetSize(), 0, lite.GetTokenId());
368 : }
369 :
370 1 : void UbTransportLiteImpl::Post(u32 index, const StreamLite& stream)
371 : {
372 1 : SqeConfigLite cfg;
373 1 : if (index == 1) { // PostFin场景
374 0 : cfg.cqeEn = true;
375 0 : cfg.placeOdr = UB_STRONG_ORDER;
376 0 : cfg.compOrder = UB_COMPLETION;
377 0 : cfg.userConfig = true;
378 : }
379 1 : u32 inlineData = 1;
380 :
381 1 : auto taskId = stream.GetRtsq()->GetTaskId();
382 :
383 : // 当前使用1个connection，下标为0 构建sqe
384 1 : RmaConnLite* conn = connVec[0];
385 :
386 : // 展开下发WQE前, 按需设置cache context
387 1 : UbConnLite* ubConnLitePtr = nullptr;
388 1 : bool needCacheTask = false;
389 1 : PreLaunchWqe(ubConnLitePtr, needCacheTask, conn);
390 : // 下发DbSqe前, 备份相关信息
391 1 : const uint32_t pendingSqeCnt = needCacheTask ? stream.GetRtsq()->GetPendingSqeCnt() : 0;
392 :
393 : // 展开下发WQE
394 1 : auto rmtBuffSliceLite = GetRmtNotifySliceLite(index);
395 1 : conn->InlineWrite(reinterpret_cast<u8*>(&inlineData), UB_INLINE_WRITE_SIZE, rmtBuffSliceLite, cfg, stream, connOut);
396 :
397 : // 展开下发WQE后, 展开下发DbSqe前, 按需缓存wqe及DbSqeIdx
398 : // 注意: pendingSqeCnt在下发DbSqe前已备份
399 : // 注意: 一定要在展开下发DbSqe前调用PostLaunchWqe, 否则如果下发DbSqe触发LaunchTask,
400 : // 而尚未调用PostLaunchWqe插入当前WQE数组，
401 : // 会导致AicpuTaskCache找不到当前WQE数组, 无法正确更新对应的DbSqeLocation
402 1 : const bool isReportTask = IsReportTask();
403 1 : DbSqeProfInfo dbSqeProfInfo;
404 1 : if (needCacheTask && isReportTask) { // 构造DbSqeProfInfo
405 0 : dbSqeProfInfo.isValid = true;
406 0 : dbSqeProfInfo.taskParamType = TaskParamType::TASK_UB_INLINE_WRITE;
407 0 : FillDbSqeProfInfoDmaPub(
408 0 : reinterpret_cast<void*>(rmtBuffSliceLite.GetAddr()), rmtBuffSliceLite.GetSize(), DmaOp::HCCL_DMA_WRITE,
409 : dbSqeProfInfo);
410 0 : dbSqeProfInfo.notifyId = rmtBuffSliceLite.GetAddr();
411 : }
412 1 : PostLaunchWqe(stream, ubConnLitePtr, needCacheTask, pendingSqeCnt, isReportTask, dbSqeProfInfo);
413 :
414 : // 构建rts 的 sqe
415 1 : BuildUbDbSendTask(stream, conn->GetUbJettyLiteId(), connOut.pi);
416 :
417 3 : HCCL_INFO("UbTransportLiteImpl::Post notifyId[0x%llx], pi=%u", rmtBuffSliceLite.GetAddr(), connOut.pi);
418 1 : if (isReportTask) {
419 1 : TaskParam taskParam{};
420 1 : taskParam.taskType = TaskParamType::TASK_UB_INLINE_WRITE;
421 1 : taskParam.beginTime = ProfGetCurCpuTimestamp();
422 2 : FillTaskParamDmaPub(
423 1 : taskParam, reinterpret_cast<void*>(rmtBuffSliceLite.GetAddr()), rmtBuffSliceLite.GetSize(),
424 : DmaOp::HCCL_DMA_WRITE);
425 1 : taskParam.taskPara.DMA.notifyID = rmtBuffSliceLite.GetAddr();
426 1 : taskParam.taskPara.DMA.notifyValue = 1;
427 :
428 3 : HCCL_INFO(
429 : "[UbTransportLiteImpl::%s] locEid[%s], rmtEid[%s]", __func__, GetLocEid().Describe().c_str(),
430 : GetRmtEid().Describe().c_str());
431 1 : AddTaskCallback(stream, taskId, taskParam);
432 1 : DfxTaskInfo* slot = stream.NextTaskSlot();
433 1 : slot->taskType = TaskParamTypeVal::TASK_UB_INLINE_WRITE;
434 1 : FillSlotUbDmaInfo(
435 : slot, stream, taskId, 0, rmtBuffSliceLite.GetAddr(), rmtBuffSliceLite.GetSize(),
436 : rmtBuffSliceLite.GetNotifyId());
437 1 : }
438 1 : }
439 :
440 1 : void UbTransportLiteImpl::Wait(u32 index, const StreamLite& stream)
441 : {
442 1 : WaitWithTimeout(index, stream, CommunicatorImplLiteMgr::GetInstance().GetEnvConfig().hcclExecTimeout);
443 1 : }
444 :
445 4 : void UbTransportLiteImpl::WaitWithTimeout(u32 index, const StreamLite& stream, u32 timeout)
446 : {
447 4 : auto taskId = stream.GetRtsq()->GetTaskId();
448 4 : auto notifyId = locNotifyVec[index]->GetId();
449 4 : stream.GetRtsq()->NotifyWait(notifyId, timeout);
450 :
451 4 : if (!IsReportTask()) {
452 0 : return;
453 : }
454 :
455 4 : TaskParam taskParam{};
456 4 : taskParam.taskType = TaskParamType::TASK_NOTIFY_WAIT;
457 4 : taskParam.beginTime = ProfGetCurCpuTimestamp();
458 4 : taskParam.taskPara.Notify.notifyID = notifyId;
459 4 : taskParam.taskPara.Notify.value = 1;
460 :
461 4 : AddTaskCallback(stream, taskId, taskParam);
462 4 : DfxTaskInfo* slot = stream.NextTaskSlot();
463 4 : slot->taskType = TaskParamTypeVal::TASK_NOTIFY_WAIT;
464 4 : slot->sqId = stream.GetSqId();
465 4 : slot->taskId = taskId;
466 4 : const void* opInfo = stream.GetLatestDfxOpInfo();
467 4 : slot->dfxOpInfo = (opInfo != nullptr) ? reinterpret_cast<u64>(opInfo) : INVALID_U64;
468 4 : slot->linkType = (linkType_ == DfxLinkType::UB) ? DfxLinkTypeVal::LINK_UB : DfxLinkTypeVal::LINK_UBoE;
469 4 : slot->transportType = static_cast<u8>(DfxTransportType::DFX_TRANSPORT_TYPE_UB);
470 4 : slot->channelHandle = reinterpret_cast<u64>(this);
471 4 : slot->taskPara.Notify.sqeAddr = stream.GetRtsq()->GetSqeAddr();
472 4 : }
473 :
474 3 : void UbTransportLiteImpl::ProfilingProcess(
475 : void* src, void* dst, u64 size, const StreamLite& stream, DmaOp dmaOp, u32 taskId)
476 : {
477 3 : if (!IsReportTask()) {
478 0 : return;
479 : }
480 :
481 3 : TaskParam taskParam{};
482 3 : taskParam.taskType = TaskParamType::TASK_UB;
483 3 : taskParam.beginTime = ProfGetCurCpuTimestamp();
484 3 : FillTaskParamDmaPub(taskParam, dst, size, dmaOp);
485 3 : taskParam.taskPara.DMA.src = src;
486 :
487 3 : AddTaskCallback(stream, taskId, taskParam);
488 3 : DfxTaskInfo* slot = stream.NextTaskSlot();
489 3 : slot->taskType = TaskParamTypeVal::TASK_UB;
490 3 : FillSlotUbDmaInfo(slot, stream, taskId, reinterpret_cast<u64>(src), reinterpret_cast<u64>(dst), size, INVALID_U32);
491 3 : }
492 :
493 2 : void UbTransportLiteImpl::ReduceProfilingProcess(
494 : void* src, void* dst, u64 size, const ReduceIn& reduceIn, const StreamLite& stream, u32 taskId)
495 :
496 : {
497 2 : if (!IsReportTask()) {
498 0 : return;
499 : }
500 :
501 2 : TaskParam taskParam{};
502 2 : taskParam.taskType = TaskParamType::TASK_UB_REDUCE_INLINE;
503 2 : taskParam.beginTime = ProfGetCurCpuTimestamp();
504 2 : FillTaskParamReducePub(taskParam, src, dst, size, reduceIn);
505 2 : taskParam.taskPara.Reduce.notifyID = INVALID_VALUE_NOTIFYID;
506 :
507 2 : AddTaskCallback(stream, taskId, taskParam);
508 2 : DfxTaskInfo* slot = stream.NextTaskSlot();
509 2 : slot->taskType = TaskParamTypeVal::TASK_UB_REDUCE_INLINE;
510 2 : FillSlotReduceInfo(
511 : slot, stream, taskId, reinterpret_cast<u64>(src), reinterpret_cast<u64>(dst), size, INVALID_U32,
512 2 : static_cast<u8>(ConvertReduceOpToHcclReduceOp(reduceIn.reduceOp)));
513 2 : }
514 :
515 0 : void UbTransportLiteImpl::WriteWithNotifyProfilingProcess(
516 : void* src, void* dst, u64 size, const StreamLite& stream, u32 taskId, u64 notifyId)
517 : {
518 0 : if (!IsReportTask()) {
519 0 : return;
520 : }
521 :
522 0 : TaskParam taskParam{};
523 0 : taskParam.taskType = TaskParamType::TASK_WRITE_WITH_NOTIFY;
524 0 : taskParam.beginTime = ProfGetCurCpuTimestamp();
525 0 : FillTaskParamDmaPub(taskParam, dst, size, DmaOp::HCCL_DMA_WRITE);
526 0 : taskParam.taskPara.DMA.src = src;
527 0 : taskParam.taskPara.DMA.notifyID = notifyId;
528 0 : taskParam.taskPara.DMA.notifyValue = 1;
529 :
530 0 : AddTaskCallback(stream, taskId, taskParam);
531 0 : DfxTaskInfo* slot = stream.NextTaskSlot();
532 0 : slot->taskType = TaskParamTypeVal::TASK_WRITE_WITH_NOTIFY;
533 0 : FillSlotUbDmaInfo(
534 : slot, stream, taskId, reinterpret_cast<u64>(src), reinterpret_cast<u64>(dst), size, static_cast<u32>(notifyId));
535 0 : }
536 :
537 0 : void UbTransportLiteImpl::WriteReduceWithNotifyProfilingProcess(
538 : void* src, void* dst, u64 size, const ReduceIn& reduceIn, const StreamLite& stream, u32 taskId, u64 notifyId)
539 : {
540 0 : if (!IsReportTask()) {
541 0 : return;
542 : }
543 :
544 0 : TaskParam taskParam{};
545 0 : taskParam.taskType = TaskParamType::TASK_WRITE_REDUCE_WITH_NOTIFY;
546 0 : taskParam.beginTime = ProfGetCurCpuTimestamp();
547 0 : FillTaskParamReducePub(taskParam, src, dst, size, reduceIn);
548 0 : taskParam.taskPara.Reduce.notifyID = notifyId;
549 :
550 0 : AddTaskCallback(stream, taskId, taskParam);
551 0 : DfxTaskInfo* slot = stream.NextTaskSlot();
552 0 : slot->taskType = TaskParamTypeVal::TASK_WRITE_REDUCE_WITH_NOTIFY;
553 0 : FillSlotReduceInfo(
554 : slot, stream, taskId, reinterpret_cast<u64>(src), reinterpret_cast<u64>(dst), size, static_cast<u32>(notifyId),
555 0 : static_cast<u8>(ConvertReduceOpToHcclReduceOp(reduceIn.reduceOp)));
556 0 : }
557 :
558 0 : void UbTransportLiteImpl::NotifyRecordProfilingProcess(
559 : void* dst, u64 size, const StreamLite& stream, u32 taskId, u64 notifyId)
560 : {
561 0 : if (!IsReportTask()) {
562 0 : return;
563 : }
564 :
565 0 : TaskParam taskParam{};
566 0 : taskParam.taskType = TaskParamType::TASK_UB_INLINE_WRITE;
567 0 : taskParam.beginTime = ProfGetCurCpuTimestamp();
568 0 : FillTaskParamDmaPub(taskParam, dst, size, DmaOp::HCCL_DMA_WRITE);
569 0 : taskParam.taskPara.DMA.notifyID = notifyId;
570 0 : taskParam.taskPara.DMA.notifyValue = 1;
571 :
572 0 : AddTaskCallback(stream, taskId, taskParam);
573 0 : DfxTaskInfo* slot = stream.NextTaskSlot();
574 0 : slot->taskType = TaskParamTypeVal::TASK_UB_INLINE_WRITE;
575 0 : FillSlotUbDmaInfo(slot, stream, taskId, 0, reinterpret_cast<u64>(dst), size, static_cast<u32>(notifyId));
576 0 : }
577 :
578 5 : void UbTransportLiteImpl::FillSlotUbDmaInfo(
579 : DfxTaskInfo* slot, const StreamLite& stream, u32 taskId, u64 srcAddr, u64 dstAddr, u64 size, u32 notifyId)
580 : {
581 5 : slot->sqId = stream.GetSqId();
582 5 : slot->taskId = taskId;
583 5 : const void* opInfo = stream.GetLatestDfxOpInfo();
584 5 : slot->dfxOpInfo = (opInfo != nullptr) ? reinterpret_cast<u64>(opInfo) : INVALID_U64;
585 5 : slot->linkType = (linkType_ == DfxLinkType::UB) ? DfxLinkTypeVal::LINK_UB : DfxLinkTypeVal::LINK_UBoE;
586 5 : slot->transportType = static_cast<u8>(DfxTransportType::DFX_TRANSPORT_TYPE_UB);
587 5 : slot->channelHandle = reinterpret_cast<u64>(this);
588 5 : slot->taskPara.ubDma.sqeAddr = stream.GetRtsq()->GetSqeAddr();
589 5 : slot->taskPara.ubDma.srcAddr = srcAddr;
590 5 : slot->taskPara.ubDma.dstAddr = dstAddr;
591 5 : slot->taskPara.ubDma.size = size;
592 5 : slot->taskPara.ubDma.notifyId = notifyId;
593 5 : slot->taskPara.ubDma.jettyHandle = GetJettyHandle();
594 5 : slot->taskPara.ubDma.jettyId = GetJettyId();
595 5 : }
596 :
597 2 : void UbTransportLiteImpl::FillSlotReduceInfo(
598 : DfxTaskInfo* slot, const StreamLite& stream, u32 taskId, u64 srcAddr, u64 dstAddr, u64 size, u32 notifyId,
599 : u8 reduceOp)
600 : {
601 2 : slot->sqId = stream.GetSqId();
602 2 : slot->taskId = taskId;
603 2 : const void* opInfo = stream.GetLatestDfxOpInfo();
604 2 : slot->dfxOpInfo = (opInfo != nullptr) ? reinterpret_cast<u64>(opInfo) : INVALID_U64;
605 2 : slot->linkType = (linkType_ == DfxLinkType::UB) ? DfxLinkTypeVal::LINK_UB : DfxLinkTypeVal::LINK_UBoE;
606 2 : slot->transportType = static_cast<u8>(DfxTransportType::DFX_TRANSPORT_TYPE_UB);
607 2 : slot->channelHandle = reinterpret_cast<u64>(this);
608 2 : slot->taskPara.Reduce.sqeAddr = stream.GetRtsq()->GetSqeAddr();
609 2 : slot->taskPara.Reduce.srcAddr = srcAddr;
610 2 : slot->taskPara.Reduce.dstAddr = dstAddr;
611 2 : slot->taskPara.Reduce.size = size;
612 2 : slot->taskPara.Reduce.notifyId = notifyId;
613 2 : slot->taskPara.Reduce.reduceOp = reduceOp;
614 2 : slot->taskPara.Reduce.jettyHandle = GetJettyHandle();
615 2 : slot->taskPara.Reduce.jettyId = GetJettyId();
616 2 : }
617 :
618 1 : void UbTransportLiteImpl::Read(const RmaBufferLite& loc, const Buffer& rmt, const StreamLite& stream)
619 : {
620 1 : SqeConfigLite cfg;
621 1 : SetFenceConfig(cfg);
622 :
623 1 : auto taskId = stream.GetRtsq()->GetTaskId();
624 :
625 : // 当前使用1个connection,下标为0
626 1 : RmaConnLite* conn = connVec[0];
627 :
628 : // 展开下发WQE前, 按需设置cache context
629 1 : UbConnLite* ubConnLitePtr = nullptr;
630 1 : bool needCacheTask = false;
631 1 : PreLaunchWqe(ubConnLitePtr, needCacheTask, conn);
632 : // 下发DbSqe前, 备份相关信息
633 1 : const uint32_t pendingSqeCnt = needCacheTask ? stream.GetRtsq()->GetPendingSqeCnt() : 0;
634 :
635 : // 展开下发WQE
636 1 : auto locRmaBufSlicelite = GetRmaBufSlicelite(loc);
637 1 : auto rmtRmaBufSlicelite = GetRmtRmaBufSliceLite(rmt);
638 1 : conn->Read(locRmaBufSlicelite, rmtRmaBufSlicelite, cfg, stream, connOut);
639 :
640 : // 展开下发WQE后, 展开下发DbSqe前, 按需缓存wqe及DbSqeIdx
641 : // 注意: pendingSqeCnt在下发DbSqe前已备份
642 : // 注意: 一定要在展开下发DbSqe前调用PostLaunchWqe, 否则如果下发DbSqe触发LaunchTask,
643 : // 而尚未调用PostLaunchWqe插入当前WQE数组，
644 : // 会导致AicpuTaskCache找不到当前WQE数组, 无法正确更新对应的DbSqeLocation
645 1 : const bool isReportTask = IsReportTask();
646 1 : DbSqeProfInfo dbSqeProfInfo;
647 1 : if (needCacheTask && isReportTask) {
648 0 : BuildDbSqeProfInfoForProfilingProcess(
649 0 : reinterpret_cast<void*>(locRmaBufSlicelite.GetAddr()),
650 0 : reinterpret_cast<void*>(rmtRmaBufSlicelite.GetAddr()), locRmaBufSlicelite.GetSize(), DmaOp::HCCL_DMA_READ,
651 : dbSqeProfInfo);
652 : }
653 1 : PostLaunchWqe(stream, ubConnLitePtr, needCacheTask, pendingSqeCnt, isReportTask, dbSqeProfInfo);
654 :
655 1 : BuildUbDbSendTask(stream, conn->GetUbJettyLiteId(), connOut.pi);
656 :
657 2 : ProfilingProcess(
658 1 : reinterpret_cast<void*>(locRmaBufSlicelite.GetAddr()), reinterpret_cast<void*>(rmtRmaBufSlicelite.GetAddr()),
659 : locRmaBufSlicelite.GetSize(), stream, DmaOp::HCCL_DMA_READ, taskId);
660 1 : }
661 :
662 1 : void UbTransportLiteImpl::Write(const RmaBufferLite& loc, const Buffer& rmt, const StreamLite& stream)
663 : {
664 1 : SqeConfigLite cfg;
665 1 : SetFenceConfig(cfg);
666 :
667 1 : auto taskId = stream.GetRtsq()->GetTaskId();
668 :
669 : // 当前使用1个connection，下标为0
670 1 : RmaConnLite* conn = connVec[0];
671 :
672 : // 展开下发WQE前, 按需设置cache context
673 1 : UbConnLite* ubConnLitePtr = nullptr;
674 1 : bool needCacheTask = false;
675 1 : PreLaunchWqe(ubConnLitePtr, needCacheTask, conn);
676 : // 下发DbSqe前, 备份相关信息
677 1 : const uint32_t pendingSqeCnt = needCacheTask ? stream.GetRtsq()->GetPendingSqeCnt() : 0;
678 :
679 : // 展开下发WQE
680 1 : auto locRmaBufSlicelite = GetRmaBufSlicelite(loc);
681 1 : auto rmtRmaBufSlicelite = GetRmtRmaBufSliceLite(rmt);
682 1 : conn->Write(locRmaBufSlicelite, rmtRmaBufSlicelite, cfg, stream, connOut);
683 :
684 : // 展开下发WQE后, 展开下发DbSqe前, 按需缓存wqe及DbSqeIdx
685 : // 注意: pendingSqeCnt在下发DbSqe前已备份
686 : // 注意: 一定要在展开下发DbSqe前调用PostLaunchWqe, 否则如果下发DbSqe触发LaunchTask,
687 : // 而尚未调用PostLaunchWqe插入当前WQE数组，
688 : // 会导致AicpuTaskCache找不到当前WQE数组, 无法正确更新对应的DbSqeLocation
689 1 : const bool isReportTask = IsReportTask();
690 1 : DbSqeProfInfo dbSqeProfInfo;
691 1 : if (needCacheTask && isReportTask) {
692 0 : BuildDbSqeProfInfoForProfilingProcess(
693 0 : reinterpret_cast<void*>(locRmaBufSlicelite.GetAddr()),
694 0 : reinterpret_cast<void*>(rmtRmaBufSlicelite.GetAddr()), locRmaBufSlicelite.GetSize(), DmaOp::HCCL_DMA_WRITE,
695 : dbSqeProfInfo);
696 : }
697 1 : PostLaunchWqe(stream, ubConnLitePtr, needCacheTask, pendingSqeCnt, isReportTask, dbSqeProfInfo);
698 :
699 1 : BuildUbDbSendTask(stream, conn->GetUbJettyLiteId(), connOut.pi);
700 :
701 2 : ProfilingProcess(
702 1 : reinterpret_cast<void*>(locRmaBufSlicelite.GetAddr()), reinterpret_cast<void*>(rmtRmaBufSlicelite.GetAddr()),
703 : locRmaBufSlicelite.GetSize(), stream, DmaOp::HCCL_DMA_WRITE, taskId);
704 1 : }
705 :
706 1 : void UbTransportLiteImpl::ReadReduce(
707 : const RmaBufferLite& loc, const Buffer& rmt, const ReduceIn& reduceIn, const StreamLite& stream)
708 : {
709 1 : SqeConfigLite cfg;
710 1 : SetFenceConfig(cfg);
711 :
712 1 : auto taskId = stream.GetRtsq()->GetTaskId();
713 :
714 : // 当前使用1个connection，下标为0
715 1 : RmaConnLite* conn = connVec[0];
716 :
717 : // 展开下发WQE前, 按需设置cache context
718 1 : UbConnLite* ubConnLitePtr = nullptr;
719 1 : bool needCacheTask = false;
720 1 : PreLaunchWqe(ubConnLitePtr, needCacheTask, conn);
721 : // 下发DbSqe前, 备份相关信息
722 1 : const uint32_t pendingSqeCnt = needCacheTask ? stream.GetRtsq()->GetPendingSqeCnt() : 0;
723 :
724 : // 展开下发WQE
725 1 : auto locRmaBufSlicelite = GetRmaBufSlicelite(loc);
726 1 : auto rmtRmaBufSlicelite = GetRmtRmaBufSliceLite(rmt);
727 1 : conn->ReadReduce(reduceIn, locRmaBufSlicelite, rmtRmaBufSlicelite, stream, cfg, connOut);
728 :
729 : // 展开下发WQE后, 展开下发DbSqe前, 按需缓存wqe及DbSqeIdx
730 : // 注意: pendingSqeCnt在下发DbSqe前已备份
731 : // 注意: 一定要在展开下发DbSqe前调用PostLaunchWqe, 否则如果下发DbSqe触发LaunchTask,
732 : // 而尚未调用PostLaunchWqe插入当前WQE数组，
733 : // 会导致AicpuTaskCache找不到当前WQE数组, 无法正确更新对应的DbSqeLocation
734 1 : const bool isReportTask = IsReportTask();
735 1 : DbSqeProfInfo dbSqeProfInfo;
736 1 : if (needCacheTask && isReportTask) {
737 0 : BuildDbSqeProfInfoForReduceProfilingProcess(
738 0 : reinterpret_cast<void*>(locRmaBufSlicelite.GetAddr()),
739 0 : reinterpret_cast<void*>(rmtRmaBufSlicelite.GetAddr()), locRmaBufSlicelite.GetSize(), reduceIn,
740 : dbSqeProfInfo);
741 : }
742 1 : PostLaunchWqe(stream, ubConnLitePtr, needCacheTask, pendingSqeCnt, isReportTask, dbSqeProfInfo);
743 :
744 1 : BuildUbDbSendTask(stream, conn->GetUbJettyLiteId(), connOut.pi);
745 :
746 2 : ReduceProfilingProcess(
747 1 : reinterpret_cast<void*>(locRmaBufSlicelite.GetAddr()), reinterpret_cast<void*>(rmtRmaBufSlicelite.GetAddr()),
748 : locRmaBufSlicelite.GetSize(), reduceIn, stream, taskId);
749 1 : }
750 :
751 1 : void UbTransportLiteImpl::WriteReduce(
752 : const RmaBufferLite& loc, const Buffer& rmt, const ReduceIn& reduceIn, const StreamLite& stream)
753 : {
754 1 : SqeConfigLite cfg;
755 1 : SetFenceConfig(cfg);
756 :
757 1 : auto taskId = stream.GetRtsq()->GetTaskId();
758 :
759 : // 当前使用1个connection，下标为0
760 1 : RmaConnLite* conn = connVec[0];
761 :
762 : // 展开下发WQE前, 按需设置cache context
763 1 : UbConnLite* ubConnLitePtr = nullptr;
764 1 : bool needCacheTask = false;
765 1 : PreLaunchWqe(ubConnLitePtr, needCacheTask, conn);
766 : // 下发DbSqe前, 备份相关信息
767 1 : const uint32_t pendingSqeCnt = needCacheTask ? stream.GetRtsq()->GetPendingSqeCnt() : 0;
768 :
769 : // 展开下发WQE
770 1 : auto locRmaBufSlicelite = GetRmaBufSlicelite(loc);
771 1 : auto rmtRmaBufSlicelite = GetRmtRmaBufSliceLite(rmt);
772 1 : conn->WriteReduce(
773 1 : reduceIn.dataType, reduceIn.reduceOp, locRmaBufSlicelite, stream, rmtRmaBufSlicelite, cfg, connOut);
774 :
775 : // 展开下发WQE后, 展开下发DbSqe前, 按需缓存wqe及DbSqeIdx
776 : // 注意: pendingSqeCnt在下发DbSqe前已备份
777 : // 注意: 一定要在展开下发DbSqe前调用PostLaunchWqe, 否则如果下发DbSqe触发LaunchTask,
778 : // 而尚未调用PostLaunchWqe插入当前WQE数组，
779 : // 会导致AicpuTaskCache找不到当前WQE数组, 无法正确更新对应的DbSqeLocation
780 1 : const bool isReportTask = IsReportTask();
781 1 : DbSqeProfInfo dbSqeProfInfo;
782 1 : if (needCacheTask && isReportTask) {
783 0 : BuildDbSqeProfInfoForReduceProfilingProcess(
784 0 : reinterpret_cast<void*>(locRmaBufSlicelite.GetAddr()),
785 0 : reinterpret_cast<void*>(rmtRmaBufSlicelite.GetAddr()), locRmaBufSlicelite.GetSize(), reduceIn,
786 : dbSqeProfInfo);
787 : }
788 1 : PostLaunchWqe(stream, ubConnLitePtr, needCacheTask, pendingSqeCnt, isReportTask, dbSqeProfInfo);
789 :
790 1 : BuildUbDbSendTask(stream, conn->GetUbJettyLiteId(), connOut.pi);
791 :
792 2 : ReduceProfilingProcess(
793 1 : reinterpret_cast<void*>(locRmaBufSlicelite.GetAddr()), reinterpret_cast<void*>(rmtRmaBufSlicelite.GetAddr()),
794 : locRmaBufSlicelite.GetSize(), reduceIn, stream, taskId);
795 1 : }
796 :
797 1 : void UbTransportLiteImpl::ExecProfiling(
798 : const RmaBufferLite& loc, const Buffer& rmt, const u64 totalSize,
799 : const BaseTransportLiteImpl::TransferOp& transferOp, const StreamLite& stream, u32 taskId)
800 : {
801 1 : if (transferOp.reduceIn.reduceOp == ReduceOp::INVALID) {
802 1 : DmaOp dmaOp = DmaOp::HCCL_DMA_WRITE;
803 1 : if (transferOp.transType == TransferType::READ) {
804 1 : dmaOp = DmaOp::HCCL_DMA_READ;
805 : }
806 1 : ProfilingProcess(
807 1 : reinterpret_cast<void*>(GetRmaBufSlicelite(loc).GetAddr()),
808 2 : reinterpret_cast<void*>(GetRmtRmaBufSliceLite(rmt).GetAddr()), totalSize, stream, dmaOp, taskId);
809 : } else {
810 0 : ReduceProfilingProcess(
811 0 : reinterpret_cast<void*>(GetRmaBufSlicelite(loc).GetAddr()),
812 0 : reinterpret_cast<void*>(GetRmtRmaBufSliceLite(rmt).GetAddr()), totalSize, transferOp.reduceIn, stream,
813 : taskId);
814 : }
815 1 : }
816 :
817 0 : void UbTransportLiteImpl::ExecProfilingAll(
818 : const RmaBufferLite& loc, const Buffer& rmt, const u64 totalSize,
819 : const BaseTransportLiteImpl::TransferOp& transferOp, const StreamLite& stream, u32 taskId, const uint32_t notifyId)
820 : {
821 0 : if (transferOp.transType == TransferType::READ) {
822 0 : ProfilingProcess(
823 0 : reinterpret_cast<void*>(GetRmaBufSlicelite(loc).GetAddr()),
824 0 : reinterpret_cast<void*>(GetRmtRmaBufSliceLite(rmt).GetAddr()), totalSize, stream, DmaOp::HCCL_DMA_READ,
825 : taskId);
826 0 : } else if (transferOp.transType == TransferType::WRITE) {
827 0 : ProfilingProcess(
828 0 : reinterpret_cast<void*>(GetRmaBufSlicelite(loc).GetAddr()),
829 0 : reinterpret_cast<void*>(GetRmtRmaBufSliceLite(rmt).GetAddr()), totalSize, stream, DmaOp::HCCL_DMA_WRITE,
830 : taskId);
831 0 : } else if (transferOp.transType == TransferType::READ_REDUCE) {
832 0 : ReduceProfilingProcess(
833 0 : reinterpret_cast<void*>(GetRmaBufSlicelite(loc).GetAddr()),
834 0 : reinterpret_cast<void*>(GetRmtRmaBufSliceLite(rmt).GetAddr()), totalSize, transferOp.reduceIn, stream,
835 : taskId);
836 0 : } else if (transferOp.transType == TransferType::WRITE_REDUCE) {
837 0 : ReduceProfilingProcess(
838 0 : reinterpret_cast<void*>(GetRmaBufSlicelite(loc).GetAddr()),
839 0 : reinterpret_cast<void*>(GetRmtRmaBufSliceLite(rmt).GetAddr()), totalSize, transferOp.reduceIn, stream,
840 : taskId);
841 0 : } else if (transferOp.transType == TransferType::WRITE_WITH_NOTIFY) {
842 0 : WriteWithNotifyProfilingProcess(
843 0 : reinterpret_cast<void*>(GetRmaBufSlicelite(loc).GetAddr()),
844 0 : reinterpret_cast<void*>(GetRmtRmaBufSliceLite(rmt).GetAddr()), totalSize, stream, taskId,
845 0 : GetRmtNotifySliceLite(notifyId).GetAddr());
846 0 : } else if (transferOp.transType == TransferType::WRITE_REDUCE_WITH_NOTIFY) {
847 0 : WriteReduceWithNotifyProfilingProcess(
848 0 : reinterpret_cast<void*>(GetRmaBufSlicelite(loc).GetAddr()),
849 0 : reinterpret_cast<void*>(GetRmtRmaBufSliceLite(rmt).GetAddr()), totalSize, transferOp.reduceIn, stream,
850 0 : taskId, GetRmtNotifySliceLite(notifyId).GetAddr());
851 0 : } else if (transferOp.transType == TransferType::NOTIFY_RECORD) {
852 0 : NotifyRecordProfilingProcess(
853 0 : reinterpret_cast<void*>(GetRmtNotifySliceLite(notifyId).GetAddr()),
854 0 : GetRmtNotifySliceLite(notifyId).GetSize(), stream, taskId, GetRmtNotifySliceLite(notifyId).GetAddr());
855 : }
856 0 : }
857 :
858 1 : void UbTransportLiteImpl::BatchTransfer(
859 : const std::vector<RmaBufferLite>& loc, const std::vector<Buffer>& rmt,
860 : const std::vector<BaseTransportLiteImpl::TransferOp>& transferOp, const StreamLite& stream)
861 : {
862 1 : if (UNLIKELY(loc.empty())) {
863 0 : return;
864 : }
865 1 : SqeConfigLite cfg;
866 1 : SetFenceConfig(cfg);
867 :
868 1 : auto taskId = stream.GetRtsq()->GetTaskId();
869 :
870 : // 当前使用1个connection，下标为0 (当前只有一个connection，对应一个jetty)
871 1 : RmaConnLite* conn = connVec[0];
872 :
873 : // 展开下发WQE前, 按需设置cache context
874 1 : UbConnLite* ubConnLitePtr = nullptr;
875 1 : bool needCacheTask = false;
876 1 : PreLaunchWqe(ubConnLitePtr, needCacheTask, conn);
877 : // 下发DbSqe前, 备份相关信息
878 1 : const uint32_t pendingSqeCnt = needCacheTask ? stream.GetRtsq()->GetPendingSqeCnt() : 0;
879 :
880 1 : u32 insNum = loc.size();
881 2 : for (u32 i = 0; i < insNum; i++) {
882 1 : cfg.cqeEn = (i == insNum - 1) ? true : false; // 返回最后一个sqe的cqe
883 1 : cfg.placeOdr = UB_RELAX_ORDER;
884 1 : cfg.compOrder = UB_NO_COMPLETION;
885 1 : cfg.userConfig = true;
886 :
887 1 : auto localBuffer = GetRmaBufSlicelite(loc[i]);
888 1 : auto remoteBuffer = GetRmtRmaBufSliceLite(rmt[i]);
889 1 : if (transferOp[i].transType == TransferType::WRITE) {
890 0 : conn->Write(localBuffer, remoteBuffer, cfg, stream, connOut); // 当前只有一个connection，对应一个jetty
891 1 : } else if (transferOp[i].transType == TransferType::WRITE_REDUCE) { // write reduce
892 0 : conn->WriteReduce(
893 0 : transferOp[i].reduceIn.dataType, transferOp[i].reduceIn.reduceOp, localBuffer, stream, remoteBuffer,
894 0 : cfg, connOut);
895 1 : } else if (transferOp[i].transType == TransferType::READ) {
896 1 : conn->Read(localBuffer, remoteBuffer, cfg, stream, connOut); // 当前只有一个connection，对应一个jetty
897 0 : } else if (transferOp[i].transType == TransferType::READ_REDUCE) { // read reduce
898 0 : conn->ReadReduce(transferOp[i].reduceIn, localBuffer, remoteBuffer, stream, cfg, connOut);
899 : }
900 : }
901 :
902 : // 按需计算totalSize
903 1 : const bool isReportTask = IsReportTask();
904 1 : u64 totalSize = 0;
905 1 : if (isReportTask) {
906 2 : for (u32 i = 0; i < insNum; i++) {
907 1 : totalSize += GetRmaBufSlicelite(loc[i]).GetSize();
908 : }
909 : }
910 :
911 : // 展开下发WQE后, 展开下发DbSqe前, 按需缓存wqe及DbSqeIdx
912 : // 注意: pendingSqeCnt在下发DbSqe前已备份
913 : // 注意: 一定要在展开下发DbSqe前调用PostLaunchWqe, 否则如果下发DbSqe触发LaunchTask,
914 : // 而尚未调用PostLaunchWqe插入当前WQE数组，
915 : // 会导致AicpuTaskCache找不到当前WQE数组, 无法正确更新对应的DbSqeLocation
916 1 : DbSqeProfInfo dbSqeProfInfo;
917 1 : if (needCacheTask && isReportTask) {
918 0 : BuildDbSqeProfInfoForExecProfiling(
919 0 : loc[insNum - 1], rmt[insNum - 1], totalSize, transferOp[insNum - 1], dbSqeProfInfo);
920 : }
921 1 : PostLaunchWqe(stream, ubConnLitePtr, needCacheTask, pendingSqeCnt, isReportTask, dbSqeProfInfo);
922 :
923 1 : BuildUbDbSendTask(stream, conn->GetUbJettyLiteId(), connOut.pi);
924 :
925 1 : ExecProfiling(loc[insNum - 1], rmt[insNum - 1], totalSize, transferOp[insNum - 1], stream, taskId);
926 : }
927 :
928 : // Convert hccl::HcommDataType => Hccl::DataType, hccl::HcommReduceOp => Hccl::ReduceOp
929 : static const std::unordered_map<HcommReduceOp, Hccl::ReduceOp> mapHcommReduceOpA5
930 : = {{HcommReduceOp::HCOMM_REDUCE_SUM, Hccl::ReduceOp::SUM},
931 : {HcommReduceOp::HCOMM_REDUCE_PROD, Hccl::ReduceOp::PROD},
932 : {HcommReduceOp::HCOMM_REDUCE_MAX, Hccl::ReduceOp::MAX},
933 : {HcommReduceOp::HCOMM_REDUCE_MIN, Hccl::ReduceOp::MIN},
934 : {HcommReduceOp::HCOMM_REDUCE_RESERVED, Hccl::ReduceOp::INVALID}};
935 :
936 : static const std::unordered_map<HcommDataType, Hccl::DataType> mapHcommDataTypeA5 = {
937 : #ifndef OPEN_BUILD_PROJECT
938 : {HcommDataType::HCOMM_DATA_TYPE_HIF8, Hccl::DataType::HIF8},
939 : {HcommDataType::HCOMM_DATA_TYPE_FP8E4M3, Hccl::DataType::FP8E4M3},
940 : {HcommDataType::HCOMM_DATA_TYPE_FP8E5M2, Hccl::DataType::FP8E5M2},
941 : {HcommDataType::HCOMM_DATA_TYPE_FP8E8M0, Hccl::DataType::FP8E8M0},
942 : #endif
943 : {HcommDataType::HCOMM_DATA_TYPE_INT8, Hccl::DataType::INT8},
944 : {HcommDataType::HCOMM_DATA_TYPE_INT16, Hccl::DataType::INT16},
945 : {HcommDataType::HCOMM_DATA_TYPE_INT32, Hccl::DataType::INT32},
946 : {HcommDataType::HCOMM_DATA_TYPE_INT64, Hccl::DataType::INT64},
947 : {HcommDataType::HCOMM_DATA_TYPE_INT128, Hccl::DataType::INT128},
948 : {HcommDataType::HCOMM_DATA_TYPE_UINT8, Hccl::DataType::UINT8},
949 : {HcommDataType::HCOMM_DATA_TYPE_UINT16, Hccl::DataType::UINT16},
950 : {HcommDataType::HCOMM_DATA_TYPE_UINT32, Hccl::DataType::UINT32},
951 : {HcommDataType::HCOMM_DATA_TYPE_UINT64, Hccl::DataType::UINT64},
952 : {HcommDataType::HCOMM_DATA_TYPE_FP16, Hccl::DataType::FP16},
953 : {HcommDataType::HCOMM_DATA_TYPE_FP32, Hccl::DataType::FP32},
954 : {HcommDataType::HCOMM_DATA_TYPE_FP64, Hccl::DataType::FP64},
955 : {HcommDataType::HCOMM_DATA_TYPE_BFP16, Hccl::DataType::BFP16},
956 : {HcommDataType::HCOMM_DATA_TYPE_RESERVED, Hccl::DataType::INVALID}};
957 :
958 13 : static HcclResult CheckReduceHcommDataTypeAndHcommReduceOp(HcommDataType dataType, HcommReduceOp reduceOp)
959 : {
960 13 : auto dataTypeIt = mapHcommDataTypeA5.find(dataType); // reduce类型，dataType不能是RESERVED
961 13 : if (dataTypeIt == mapHcommDataTypeA5.end() || dataTypeIt->first == HcommDataType::HCOMM_DATA_TYPE_RESERVED) {
962 0 : HCCL_ERROR("[%s] type[%u] is not supported.", __func__, dataType);
963 0 : return HCCL_E_PARA;
964 : }
965 :
966 13 : auto reduceOpIt = mapHcommReduceOpA5.find(reduceOp); // reduce类型，reduceOp不能是RESERVED
967 13 : if (reduceOpIt == mapHcommReduceOpA5.end() || reduceOpIt->first == HcommReduceOp::HCOMM_REDUCE_RESERVED) {
968 0 : HCCL_ERROR("[%s] op[%u] is not supported.", __func__, reduceOp);
969 0 : return HCCL_E_PARA;
970 : }
971 :
972 13 : return HCCL_SUCCESS;
973 : }
974 :
975 : constexpr u32 SIZE_TABLE[HCCL_DATA_TYPE_RESERVED]
976 : = {sizeof(s8),
977 : sizeof(s16),
978 : sizeof(s32),
979 : 2,
980 : sizeof(float),
981 : sizeof(s64),
982 : sizeof(u64),
983 : sizeof(u8),
984 : sizeof(u16),
985 : sizeof(u32),
986 : 8,
987 : 2,
988 : 16,
989 : 2,
990 : 1,
991 : 1,
992 : 1,
993 : 1};
994 :
995 8 : static HcclResult ParasReduceData(
996 : const HcommBatchTransferDesc& transferDesc, uint64_t& len, HcommDataType& dataType, HcommReduceOp& reduceOp)
997 : {
998 8 : len = transferDesc.transferInfo.reduce.count;
999 8 : dataType = transferDesc.transferInfo.reduce.dataType;
1000 8 : reduceOp = transferDesc.transferInfo.reduce.reduceOp;
1001 8 : auto ret = CheckReduceHcommDataTypeAndHcommReduceOp(dataType, reduceOp);
1002 8 : CHK_PRT_RET(
1003 : ret != HCCL_SUCCESS,
1004 : HCCL_ERROR("FAIL at CheckReduceHcommDataTypeAndHcommReduceOp dataType[%d], reduceOp[%d].", dataType, reduceOp),
1005 : ret);
1006 8 : return HCCL_SUCCESS;
1007 : }
1008 :
1009 34 : static HcclResult ParseData(
1010 : const HcommBatchTransferDesc& transferDesc, void*& rmt, void*& loc, uint64_t& len, Hccl::TransferType& tfType,
1011 : HcommDataType& dataType, HcommReduceOp& reduceOp, uint32_t& notifyIdx)
1012 : {
1013 34 : if (transferDesc.transType == HCOMM_TRANSFER_TYPE_WRITE) {
1014 7 : rmt = transferDesc.transferInfo.write.dst; // write操作，dst是远端地址
1015 7 : loc = transferDesc.transferInfo.write.src; // src是本端地址
1016 7 : len = transferDesc.transferInfo.write.len;
1017 7 : tfType = Hccl::TransferType::WRITE;
1018 27 : } else if (transferDesc.transType == HCOMM_TRANSFER_TYPE_READ) {
1019 5 : rmt = transferDesc.transferInfo.read.src; // read操作，src是远端地址
1020 5 : loc = transferDesc.transferInfo.read.dst; // dst是本端地址
1021 5 : len = transferDesc.transferInfo.read.len;
1022 5 : tfType = Hccl::TransferType::READ;
1023 22 : } else if (transferDesc.transType == HCOMM_TRANSFER_TYPE_WRITE_REDUCE) {
1024 4 : rmt = transferDesc.transferInfo.reduce.dst;
1025 4 : loc = transferDesc.transferInfo.reduce.src;
1026 4 : tfType = Hccl::TransferType::WRITE_REDUCE;
1027 4 : CHK_RET(ParasReduceData(transferDesc, len, dataType, reduceOp));
1028 18 : } else if (transferDesc.transType == HCOMM_TRANSFER_TYPE_READ_REDUCE) {
1029 4 : rmt = transferDesc.transferInfo.reduce.src;
1030 4 : loc = transferDesc.transferInfo.reduce.dst;
1031 4 : tfType = Hccl::TransferType::READ_REDUCE;
1032 4 : CHK_RET(ParasReduceData(transferDesc, len, dataType, reduceOp));
1033 14 : } else if (transferDesc.transType == HCOMM_TRANSFER_TYPE_WRITE_WITH_NOTIFY) {
1034 4 : rmt = transferDesc.transferInfo.writeWithNotify.dst; // write操作，dst是远端地址
1035 4 : loc = transferDesc.transferInfo.writeWithNotify.src; // src是本端地址
1036 4 : len = transferDesc.transferInfo.writeWithNotify.len;
1037 4 : notifyIdx = transferDesc.transferInfo.writeWithNotify.notifyIdx;
1038 4 : tfType = Hccl::TransferType::WRITE_WITH_NOTIFY;
1039 10 : } else if (transferDesc.transType == HCOMM_TRANSFER_TYPE_WRITE_REDUCE_WITH_NOTIFY) {
1040 5 : rmt = transferDesc.transferInfo.writeReduceWithNotify.dst;
1041 5 : loc = transferDesc.transferInfo.writeReduceWithNotify.src;
1042 5 : len = transferDesc.transferInfo.writeReduceWithNotify.count;
1043 5 : dataType = transferDesc.transferInfo.writeReduceWithNotify.dataType;
1044 5 : reduceOp = transferDesc.transferInfo.writeReduceWithNotify.reduceOp;
1045 5 : notifyIdx = transferDesc.transferInfo.writeReduceWithNotify.notifyIdx;
1046 5 : tfType = Hccl::TransferType::WRITE_REDUCE_WITH_NOTIFY;
1047 5 : CHK_RET(CheckReduceHcommDataTypeAndHcommReduceOp(dataType, reduceOp));
1048 5 : } else if (transferDesc.transType == HCOMM_TRANSFER_TYPE_NOTIFY_RECORD) {
1049 4 : notifyIdx = transferDesc.transferInfo.notifyRecord.notifyIdx;
1050 4 : tfType = Hccl::TransferType::NOTIFY_RECORD;
1051 : } else {
1052 1 : HCCL_ERROR("[%s] unsupported transType[%d]", __func__, transferDesc.transType);
1053 1 : return HCCL_E_NOT_SUPPORT;
1054 : }
1055 33 : if (reduceOp != HcommReduceOp::HCOMM_REDUCE_RESERVED) { // 对于规约类型, size = count * sizeof(datatype)
1056 13 : len = len * SIZE_TABLE[dataType];
1057 : }
1058 33 : return HCCL_SUCCESS;
1059 : }
1060 : constexpr uint32_t NOTIFYIDX_INVALID_VALUE = 0xFFFFFFFF; // NOTIFY idex非法值
1061 12 : HcclResult UbTransportLiteImpl::ExecuteBatchTransfer(
1062 : StreamLite* streamLitePtr, const HcommBatchTransferDesc* transferDescs, uint32_t transferDescNum)
1063 : {
1064 12 : std::vector<Hccl::RmaBufferLite> locSlices;
1065 12 : std::vector<Hccl::Buffer> rmtSlices;
1066 12 : std::vector<Hccl::BaseTransportLiteImpl::TransferOp> transferOps;
1067 12 : std::vector<uint32_t> notifyIdxs;
1068 :
1069 12 : locSlices.reserve(transferDescNum);
1070 12 : rmtSlices.reserve(transferDescNum);
1071 12 : transferOps.reserve(transferDescNum);
1072 12 : notifyIdxs.reserve(transferDescNum);
1073 :
1074 42 : for (uint32_t i = 0; i < transferDescNum; i++) {
1075 34 : Hccl::RmaBufferLite locRmaBuf;
1076 34 : void* rmt = nullptr;
1077 34 : void* loc = nullptr;
1078 34 : uint64_t len = 0;
1079 34 : Hccl::TransferType tfType;
1080 34 : HcommDataType dataType{HcommDataType::HCOMM_DATA_TYPE_RESERVED};
1081 34 : HcommReduceOp reduceOp{HcommReduceOp::HCOMM_REDUCE_RESERVED};
1082 34 : uint32_t notifyIdx = NOTIFYIDX_INVALID_VALUE;
1083 37 : CHK_RET(ParseData(transferDescs[i], rmt, loc, len, tfType, dataType, reduceOp, notifyIdx));
1084 33 : if (tfType != Hccl::TransferType::NOTIFY_RECORD) { // NOTIFY_RECORD时没有地址字段
1085 29 : CHK_PTR_NULL(rmt);
1086 28 : CHK_PTR_NULL(loc);
1087 27 : HcclResult ret = BuildLocRmaBufferLite(reinterpret_cast<uintptr_t>(loc), len, locRmaBuf);
1088 27 : CHK_PRT_RET(
1089 : ret != HCCL_SUCCESS,
1090 : HCCL_ERROR(
1091 : "[%s] FAIL at BuildLocRmaBufferLite for index %u. rmt[%p], loc[%p], len[0x%llx], tfType[%u], "
1092 : "dataType[%d], reduceOp[%d].",
1093 : __func__, i, rmt, loc, len, tfType, dataType, reduceOp),
1094 : ret);
1095 : }
1096 58 : if (tfType == Hccl::TransferType::NOTIFY_RECORD || tfType == Hccl::TransferType::WRITE_WITH_NOTIFY
1097 58 : || tfType == Hccl::TransferType::WRITE_REDUCE_WITH_NOTIFY) {
1098 13 : CHK_PRT_RET(
1099 : notifyIdx == NOTIFYIDX_INVALID_VALUE,
1100 : HCCL_ERROR(
1101 : "[%s] FAIL at ParseData for index %u. tfType[%u], notifyIdx[%u].", __func__, i, tfType, notifyIdx),
1102 : HCCL_E_PARA);
1103 : }
1104 30 : notifyIdxs.push_back(notifyIdx);
1105 30 : locSlices.push_back(locRmaBuf);
1106 :
1107 30 : const Hccl::Buffer rmtBuf{reinterpret_cast<uintptr_t>(rmt), len};
1108 30 : rmtSlices.push_back(rmtBuf);
1109 :
1110 30 : Hccl::ReduceIn reduceIn{mapHcommDataTypeA5.at(dataType), mapHcommReduceOpA5.at(reduceOp)};
1111 :
1112 30 : transferOps.push_back(Hccl::BaseTransportLiteImpl::TransferOp{tfType, reduceIn});
1113 :
1114 30 : HCCL_DEBUG(
1115 : "[%s] Prepared transfer op for index %u. rmt[%p], loc[%p], len[0x%llx], tfType[%u], dataType[%d], "
1116 : "reduceOp[%d].",
1117 : __func__, i, rmt, loc, len, tfType, dataType, reduceOp);
1118 30 : }
1119 8 : EXCEPTION_CATCH(
1120 : BatchTransferAll(locSlices, rmtSlices, transferOps, notifyIdxs, *streamLitePtr), return HCCL_E_INTERNAL);
1121 8 : return HCCL_SUCCESS;
1122 12 : }
1123 :
1124 0 : void UbTransportLiteImpl::BatchTransferAll(
1125 : const std::vector<RmaBufferLite>& loc, const std::vector<Buffer>& rmt,
1126 : const std::vector<BaseTransportLiteImpl::TransferOp>& transferOp, const std::vector<uint32_t>& notifyIdxs,
1127 : const StreamLite& stream)
1128 : {
1129 0 : if (UNLIKELY(loc.empty())) {
1130 0 : return;
1131 : }
1132 :
1133 0 : auto taskId = stream.GetRtsq()->GetTaskId();
1134 :
1135 : // 当前使用1个connection，下标为0 (当前只有一个connection，对应一个jetty)
1136 0 : RmaConnLite* conn = connVec[0];
1137 :
1138 : // 展开下发WQE前, 按需设置cache context
1139 0 : UbConnLite* ubConnLitePtr = nullptr;
1140 0 : bool needCacheTask = false;
1141 0 : PreLaunchWqe(ubConnLitePtr, needCacheTask, conn);
1142 : // 下发DbSqe前, 备份相关信息
1143 0 : const uint32_t pendingSqeCnt = needCacheTask ? stream.GetRtsq()->GetPendingSqeCnt() : 0;
1144 :
1145 : // 批量展开下发WQE
1146 0 : u32 insNum = loc.size();
1147 0 : u64 totalSize = 0;
1148 0 : BatchTransferAllWqe_(loc, rmt, transferOp, notifyIdxs, stream, conn, totalSize);
1149 :
1150 0 : const bool isReportTask = IsReportTask();
1151 : // 展开下发WQE后, 展开下发DbSqe前, 按需缓存wqe及DbSqeIdx
1152 : // 注意: pendingSqeCnt在下发DbSqe前已备份
1153 : // 注意: 一定要在展开下发DbSqe前调用PostLaunchWqe, 否则如果下发DbSqe触发LaunchTask,
1154 : // 而尚未调用PostLaunchWqe插入当前WQE数组，
1155 : // 会导致AicpuTaskCache找不到当前WQE数组, 无法正确更新对应的DbSqeLocation
1156 0 : DbSqeProfInfo dbSqeProfInfo;
1157 0 : if (needCacheTask && isReportTask) {
1158 0 : BuildDbSqeProfInfoForExecProfilingAll(
1159 0 : loc[insNum - 1], rmt[insNum - 1], totalSize, transferOp[insNum - 1], notifyIdxs[insNum - 1], dbSqeProfInfo);
1160 : }
1161 0 : PostLaunchWqe(stream, ubConnLitePtr, needCacheTask, pendingSqeCnt, isReportTask, dbSqeProfInfo);
1162 :
1163 0 : BuildUbDbSendTask(stream, conn->GetUbJettyLiteId(), connOut.pi); // 约束使用一批wqe的个数不会导致反压
1164 :
1165 0 : ExecProfilingAll(
1166 0 : loc[insNum - 1], rmt[insNum - 1], totalSize, transferOp[insNum - 1], stream, taskId, notifyIdxs[insNum - 1]);
1167 : }
1168 :
1169 0 : inline void UbTransportLiteImpl::BatchTransferAllWqe_(
1170 : const std::vector<RmaBufferLite>& loc, const std::vector<Buffer>& rmt,
1171 : const std::vector<BaseTransportLiteImpl::TransferOp>& transferOp, const std::vector<uint32_t>& notifyIdxs,
1172 : const StreamLite& stream, RmaConnLite* conn, u64& totalSize)
1173 : {
1174 0 : u64 notifyData = 1; // 普通notify，固定1，用于writeWithNotify与writeReduceWithNotify
1175 0 : SqeConfigLite cfg;
1176 0 : SetFenceConfig(cfg);
1177 0 : u32 insNum = loc.size();
1178 0 : const bool isReportTask = IsReportTask();
1179 :
1180 0 : for (u32 i = 0; i < insNum; i++) {
1181 0 : cfg.cqeEn = (i == insNum - 1) ? true : false; // 返回最后一个sqe的cqe
1182 0 : cfg.placeOdr = (i == insNum - 1) ? UB_STRONG_ORDER : UB_RELAX_ORDER; // 最后一个要求保序
1183 0 : cfg.compOrder = (i == insNum - 1) ? UB_COMPLETION : UB_NO_COMPLETION;
1184 0 : cfg.userConfig = true;
1185 :
1186 0 : if (transferOp[i].transType == TransferType::NOTIFY_RECORD) { // notifyRecord操作没有loc/rmt，因此单独处理
1187 0 : if (notifyIdxs[i] == 1) { // PostFin场景
1188 0 : cfg.cqeEn = true;
1189 0 : cfg.placeOdr = UB_STRONG_ORDER;
1190 0 : cfg.compOrder = UB_COMPLETION;
1191 0 : cfg.userConfig = true;
1192 : }
1193 0 : u32 inlineData = 1;
1194 : // 当前使用1个connection，下标为0 构建sqe
1195 0 : conn->InlineWrite(
1196 0 : reinterpret_cast<u8*>(&inlineData), UB_INLINE_WRITE_SIZE, GetRmtNotifySliceLite(notifyIdxs[i]), cfg,
1197 0 : stream, connOut);
1198 : } else {
1199 0 : auto localBuffer = GetRmaBufSlicelite(loc[i]);
1200 0 : auto remoteBuffer = GetRmtRmaBufSliceLite(rmt[i]);
1201 0 : if (transferOp[i].transType == TransferType::WRITE) {
1202 0 : conn->Write(localBuffer, remoteBuffer, cfg, stream, connOut);
1203 0 : } else if (transferOp[i].transType == TransferType::WRITE_REDUCE) {
1204 0 : conn->WriteReduce(
1205 0 : transferOp[i].reduceIn.dataType, transferOp[i].reduceIn.reduceOp, localBuffer, stream, remoteBuffer,
1206 0 : cfg, connOut);
1207 0 : } else if (transferOp[i].transType == TransferType::READ) {
1208 0 : conn->Read(localBuffer, remoteBuffer, cfg, stream, connOut);
1209 0 : } else if (transferOp[i].transType == TransferType::READ_REDUCE) {
1210 0 : conn->ReadReduce(transferOp[i].reduceIn, localBuffer, remoteBuffer, stream, cfg, connOut);
1211 0 : } else if (transferOp[i].transType == TransferType::WRITE_WITH_NOTIFY) {
1212 0 : conn->WriteWithNotify(
1213 0 : localBuffer, remoteBuffer, cfg, connOut, GetRmtNotifySliceLite(notifyIdxs[i]), stream,
1214 : notifyData); // 当前使用1个connection，下标为0
1215 0 : } else if (transferOp[i].transType == TransferType::WRITE_REDUCE_WITH_NOTIFY) {
1216 0 : conn->WriteReduceWithNotify(
1217 0 : transferOp[i].reduceIn.dataType, transferOp[i].reduceIn.reduceOp, localBuffer, remoteBuffer, cfg,
1218 0 : stream, connOut, GetRmtNotifySliceLite(notifyIdxs[i]),
1219 : notifyData); // 当前使用1个connection，下标为0
1220 : }
1221 : }
1222 0 : if (isReportTask) {
1223 0 : totalSize += GetRmaBufSlicelite(loc[i]).GetSize();
1224 : }
1225 : }
1226 0 : }
1227 :
1228 0 : void UbTransportLiteImpl::Drain(const StreamLite& stream)
1229 : {
1230 0 : std::lock_guard<std::mutex> lock(drainMtx_);
1231 0 : if (drainNotify_.size == 0 || rmtDrainBuffer_.size == 0) {
1232 0 : HCCL_WARNING("[UbTransportLiteImpl::%s] drain resource is null skip", __func__);
1233 0 : return;
1234 : }
1235 :
1236 0 : SqeConfigLite cfg;
1237 0 : Fence();
1238 0 : SetFenceConfig(cfg);
1239 :
1240 : // 当前使用1个connection，下标为0 (当前只有一个connection，对应一个jetty)
1241 0 : RmaConnLite* conn = connVec[0];
1242 :
1243 : // 展开下发WQE前, 按需设置cache context
1244 0 : UbConnLite* ubConnLitePtr = nullptr;
1245 0 : bool needCacheTask = false;
1246 0 : PreLaunchWqe(ubConnLitePtr, needCacheTask, conn);
1247 : // 下发DbSqe前, 备份相关信息
1248 0 : const uint32_t pendingSqeCnt = needCacheTask ? stream.GetRtsq()->GetPendingSqeCnt() : 0;
1249 :
1250 : // 展开下发WQE
1251 0 : auto drainNotifyBufSlice = RmaBufSliceLite(drainNotify_.addr, drainNotify_.size, 0, drainNotify_.tokenId);
1252 : auto drainConstBufSlice = RmtRmaBufSliceLite(
1253 0 : rmtDrainBuffer_.addr, rmtDrainBuffer_.size, 0, rmtDrainBuffer_.tokenId, rmtDrainBuffer_.tokenValue, UINT32_MAX);
1254 0 : conn->Read(drainNotifyBufSlice, drainConstBufSlice, cfg, stream, connOut);
1255 :
1256 0 : const bool isReportTask = false; // Drain操作当前不构造TaskParam
1257 : // 展开下发WQE后, 展开下发DbSqe前, 按需缓存wqe及DbSqeIdx
1258 : // 注意: pendingSqeCnt在下发DbSqe前已备份
1259 : // 注意: 一定要在展开下发DbSqe前调用PostLaunchWqe, 否则如果下发DbSqe触发LaunchTask,
1260 : // 而尚未调用PostLaunchWqe插入当前WQE数组，
1261 : // 会导致AicpuTaskCache找不到当前WQE数组, 无法正确更新对应的DbSqeLocation
1262 0 : DbSqeProfInfo dbSqeProfInfo;
1263 0 : PostLaunchWqe(stream, ubConnLitePtr, needCacheTask, pendingSqeCnt, isReportTask, dbSqeProfInfo);
1264 :
1265 0 : BuildUbDbSendTask(stream, conn->GetUbJettyLiteId(), connOut.pi);
1266 :
1267 0 : BuildNotifyWaitTask(stream, drainNotify_.notifyId);
1268 0 : }
1269 :
1270 1 : void UbTransportLiteImpl::ReportWriteWithNotifyTask(
1271 : const RmaBufSliceLite& locSlice, const RmtRmaBufSliceLite& rmtSlice, const RmtRmaBufSliceLite& rmtNotifySlice,
1272 : const StreamLite& stream, u32 taskId)
1273 : {
1274 1 : if (!IsReportTask()) {
1275 0 : return;
1276 : }
1277 :
1278 1 : TaskParam taskParam{};
1279 1 : taskParam.taskType = TaskParamType::TASK_WRITE_WITH_NOTIFY;
1280 1 : taskParam.beginTime = ProfGetCurCpuTimestamp();
1281 2 : FillTaskParamDmaPub(
1282 1 : taskParam, reinterpret_cast<void*>(rmtSlice.GetAddr()), locSlice.GetSize(), DmaOp::HCCL_DMA_WRITE);
1283 1 : taskParam.taskPara.DMA.src = reinterpret_cast<void*>(locSlice.GetAddr());
1284 1 : taskParam.taskPara.DMA.notifyID = rmtNotifySlice.GetAddr();
1285 1 : taskParam.taskPara.DMA.notifyValue = 1;
1286 :
1287 1 : AddTaskCallback(stream, taskId, taskParam);
1288 1 : DfxTaskInfo* slot = stream.NextTaskSlot();
1289 1 : slot->taskType = TaskParamTypeVal::TASK_WRITE_WITH_NOTIFY;
1290 1 : FillSlotUbDmaInfo(
1291 : slot, stream, taskId, locSlice.GetAddr(), rmtSlice.GetAddr(), locSlice.GetSize(), rmtNotifySlice.GetNotifyId());
1292 1 : }
1293 :
1294 0 : void UbTransportLiteImpl::ReportWriteReduceWithNotifyTask(
1295 : const RmaBufSliceLite& locSlice, const RmtRmaBufSliceLite& rmtSlice, const RmtRmaBufSliceLite& rmtNotifySlice,
1296 : const ReduceIn& reduceIn, const StreamLite& stream, u32 taskId)
1297 : {
1298 0 : if (!IsReportTask()) {
1299 0 : return;
1300 : }
1301 :
1302 0 : TaskParam taskParam{};
1303 0 : taskParam.taskType = TaskParamType::TASK_WRITE_REDUCE_WITH_NOTIFY;
1304 0 : taskParam.beginTime = ProfGetCurCpuTimestamp();
1305 0 : FillTaskParamReducePub(
1306 0 : taskParam, reinterpret_cast<void*>(locSlice.GetAddr()), reinterpret_cast<void*>(rmtSlice.GetAddr()),
1307 : locSlice.GetSize(), reduceIn);
1308 0 : taskParam.taskPara.Reduce.notifyID = rmtNotifySlice.GetAddr();
1309 :
1310 0 : AddTaskCallback(stream, taskId, taskParam);
1311 0 : DfxTaskInfo* slot = stream.NextTaskSlot();
1312 0 : slot->taskType = TaskParamTypeVal::TASK_WRITE_REDUCE_WITH_NOTIFY;
1313 0 : FillSlotReduceInfo(
1314 : slot, stream, taskId, locSlice.GetAddr(), rmtSlice.GetAddr(), locSlice.GetSize(), rmtNotifySlice.GetNotifyId(),
1315 0 : static_cast<u8>(ConvertReduceOpToHcclReduceOp(reduceIn.reduceOp)));
1316 0 : }
1317 :
1318 1 : void UbTransportLiteImpl::WriteWithNotify(
1319 : const RmaBufferLite& loc, const Buffer& rmt, const WithNotifyIn& withNotify, const StreamLite& stream)
1320 : {
1321 1 : SqeConfigLite cfg;
1322 1 : SetFenceConfig(cfg);
1323 1 : u64 notifyData = 1; // 普通notify，固定1
1324 :
1325 1 : auto taskId = stream.GetRtsq()->GetTaskId();
1326 :
1327 : // 当前使用1个connection，下标为0
1328 1 : RmaConnLite* conn = connVec[0];
1329 :
1330 : // 展开下发WQE前, 按需设置cache context
1331 1 : UbConnLite* ubConnLitePtr = nullptr;
1332 1 : bool needCacheTask = false;
1333 1 : PreLaunchWqe(ubConnLitePtr, needCacheTask, conn);
1334 : // 下发DbSqe前, 备份相关信息
1335 1 : const uint32_t pendingSqeCnt = needCacheTask ? stream.GetRtsq()->GetPendingSqeCnt() : 0;
1336 :
1337 : // 展开下发WQE
1338 1 : auto locRmaBufSlicelite = GetRmaBufSlicelite(loc);
1339 1 : auto rmtRmaBufSlicelite = GetRmtRmaBufSliceLite(rmt);
1340 1 : auto rmtNotifySliceLite = GetRmtNotifySliceLite(withNotify.index_);
1341 1 : conn->WriteWithNotify(locRmaBufSlicelite, rmtRmaBufSlicelite, cfg, connOut, rmtNotifySliceLite, stream, notifyData);
1342 :
1343 : // 展开下发WQE后, 展开下发DbSqe前, 按需缓存wqe及DbSqeIdx
1344 : // 注意: pendingSqeCnt在下发DbSqe前已备份
1345 : // 注意: 一定要在展开下发DbSqe前调用PostLaunchWqe, 否则如果下发DbSqe触发LaunchTask,
1346 : // 而尚未调用PostLaunchWqe插入当前WQE数组，
1347 : // 会导致AicpuTaskCache找不到当前WQE数组, 无法正确更新对应的DbSqeLocation
1348 1 : const bool isReportTask = IsReportTask();
1349 1 : DbSqeProfInfo dbSqeProfInfo;
1350 1 : if (needCacheTask && isReportTask) {
1351 0 : BuildDbSqeProfInfoForWriteWithNotify(
1352 0 : reinterpret_cast<void*>(locRmaBufSlicelite.GetAddr()),
1353 0 : reinterpret_cast<void*>(rmtRmaBufSlicelite.GetAddr()), locRmaBufSlicelite.GetSize(),
1354 : rmtNotifySliceLite.GetAddr(), dbSqeProfInfo);
1355 : }
1356 1 : PostLaunchWqe(stream, ubConnLitePtr, needCacheTask, pendingSqeCnt, isReportTask, dbSqeProfInfo);
1357 :
1358 1 : BuildUbDbSendTask(stream, conn->GetUbJettyLiteId(), connOut.pi);
1359 :
1360 1 : ReportWriteWithNotifyTask(locRmaBufSlicelite, rmtRmaBufSlicelite, rmtNotifySliceLite, stream, taskId);
1361 1 : }
1362 :
1363 0 : void UbTransportLiteImpl::WriteReduceWithNotify(
1364 : const RmaBufferLite& loc, const Buffer& rmt, const ReduceIn& reduceIn, const WithNotifyIn& withNotify,
1365 : const StreamLite& stream)
1366 : {
1367 0 : SqeConfigLite cfg;
1368 0 : SetFenceConfig(cfg);
1369 0 : u64 notifyData = 1; // 普通notify，固定1
1370 :
1371 0 : auto taskId = stream.GetRtsq()->GetTaskId();
1372 :
1373 : // 当前使用1个connection，下标为0
1374 0 : RmaConnLite* conn = connVec[0];
1375 :
1376 : // 展开下发WQE前, 按需设置cache context
1377 0 : UbConnLite* ubConnLitePtr = nullptr;
1378 0 : bool needCacheTask = false;
1379 0 : PreLaunchWqe(ubConnLitePtr, needCacheTask, conn);
1380 : // 下发DbSqe前, 备份相关信息
1381 0 : const uint32_t pendingSqeCnt = needCacheTask ? stream.GetRtsq()->GetPendingSqeCnt() : 0;
1382 :
1383 : // 展开下发WQE
1384 0 : auto locRmaBufSlicelite = GetRmaBufSlicelite(loc);
1385 0 : auto rmtRmaBufSlicelite = GetRmtRmaBufSliceLite(rmt);
1386 0 : auto rmtNotifySliceLite = GetRmtNotifySliceLite(withNotify.index_);
1387 0 : conn->WriteReduceWithNotify(
1388 0 : reduceIn.dataType, reduceIn.reduceOp, locRmaBufSlicelite, rmtRmaBufSlicelite, cfg, stream, connOut,
1389 : rmtNotifySliceLite, notifyData);
1390 :
1391 : // 展开下发WQE后, 展开下发DbSqe前, 按需缓存wqe及DbSqeIdx
1392 : // 注意: pendingSqeCnt在下发DbSqe前已备份
1393 : // 注意: 一定要在展开下发DbSqe前调用PostLaunchWqe, 否则如果下发DbSqe触发LaunchTask,
1394 : // 而尚未调用PostLaunchWqe插入当前WQE数组，
1395 : // 会导致AicpuTaskCache找不到当前WQE数组, 无法正确更新对应的DbSqeLocation
1396 0 : const bool isReportTask = IsReportTask();
1397 0 : DbSqeProfInfo dbSqeProfInfo;
1398 0 : if (needCacheTask && isReportTask) {
1399 0 : BuildDbSqeProfInfoForWriteReduceWithNotify(
1400 0 : reinterpret_cast<void*>(locRmaBufSlicelite.GetAddr()),
1401 0 : reinterpret_cast<void*>(rmtRmaBufSlicelite.GetAddr()), locRmaBufSlicelite.GetSize(), reduceIn,
1402 : rmtNotifySliceLite.GetAddr(), dbSqeProfInfo);
1403 : }
1404 0 : PostLaunchWqe(stream, ubConnLitePtr, needCacheTask, pendingSqeCnt, isReportTask, dbSqeProfInfo);
1405 :
1406 0 : BuildUbDbSendTask(stream, conn->GetUbJettyLiteId(), connOut.pi);
1407 :
1408 0 : ReportWriteReduceWithNotifyTask(
1409 : locRmaBufSlicelite, rmtRmaBufSlicelite, rmtNotifySliceLite, reduceIn, stream, taskId);
1410 0 : }
1411 :
1412 1 : void UbTransportLiteImpl::BatchOneSidedRead(
1413 : const vector<RmaBufSliceLite>& loc, const vector<RmtRmaBufSliceLite>& rmt, const StreamLite& stream)
1414 : {
1415 1 : SqeConfigLite cfg;
1416 1 : SetFenceConfig(cfg);
1417 :
1418 : // 当前使用1个connection，下标为0
1419 1 : RmaConnLite* conn = connVec[0];
1420 :
1421 : // 展开下发WQE前, 按需设置cache context
1422 1 : UbConnLite* ubConnLitePtr = nullptr;
1423 1 : bool needCacheTask = false;
1424 1 : PreLaunchWqe(ubConnLitePtr, needCacheTask, conn);
1425 : // 下发DbSqe前, 备份相关信息
1426 1 : const uint32_t pendingSqeCnt = needCacheTask ? stream.GetRtsq()->GetPendingSqeCnt() : 0;
1427 :
1428 : // 展开下发WQE
1429 1 : conn->BatchOneSidedRead(loc, rmt, cfg, stream, connOut);
1430 :
1431 : // 展开下发WQE后, 展开下发DbSqe前, 按需缓存wqe及DbSqeIdx
1432 : // 注意: pendingSqeCnt在下发DbSqe前已备份
1433 : // 注意: 一定要在展开下发DbSqe前调用PostLaunchWqe, 否则如果下发DbSqe触发LaunchTask,
1434 : // 而尚未调用PostLaunchWqe插入当前WQE数组，
1435 : // 会导致AicpuTaskCache找不到当前WQE数组, 无法正确更新对应的DbSqeLocation
1436 1 : PostLaunchWqe(stream, ubConnLitePtr, needCacheTask, pendingSqeCnt, false, DbSqeProfInfo());
1437 :
1438 1 : BuildUbDbSendTask(stream, conn->GetUbJettyLiteId(), connOut.pi);
1439 1 : }
1440 :
1441 1 : void UbTransportLiteImpl::BatchOneSidedWrite(
1442 : const vector<RmaBufSliceLite>& loc, const vector<RmtRmaBufSliceLite>& rmt, const StreamLite& stream)
1443 : {
1444 1 : SqeConfigLite cfg;
1445 1 : SetFenceConfig(cfg);
1446 :
1447 : // 当前使用1个connection，下标为0
1448 1 : RmaConnLite* conn = connVec[0];
1449 :
1450 : // 展开下发WQE前, 按需设置cache context
1451 1 : UbConnLite* ubConnLitePtr = nullptr;
1452 1 : bool needCacheTask = false;
1453 1 : PreLaunchWqe(ubConnLitePtr, needCacheTask, conn);
1454 : // 下发DbSqe前, 备份相关信息
1455 1 : const uint32_t pendingSqeCnt = needCacheTask ? stream.GetRtsq()->GetPendingSqeCnt() : 0;
1456 :
1457 : // 展开下发WQE
1458 1 : conn->BatchOneSidedWrite(loc, rmt, cfg, stream, connOut);
1459 :
1460 : // 展开下发WQE后, 展开下发DbSqe前, 按需缓存wqe及DbSqeIdx
1461 : // 注意: pendingSqeCnt在下发DbSqe前已备份
1462 : // 注意: 一定要在展开下发DbSqe前调用PostLaunchWqe, 否则如果下发DbSqe触发LaunchTask,
1463 : // 而尚未调用PostLaunchWqe插入当前WQE数组，
1464 : // 会导致AicpuTaskCache找不到当前WQE数组, 无法正确更新对应的DbSqeLocation
1465 1 : PostLaunchWqe(stream, ubConnLitePtr, needCacheTask, pendingSqeCnt, false, DbSqeProfInfo());
1466 :
1467 1 : BuildUbDbSendTask(stream, conn->GetUbJettyLiteId(), connOut.pi);
1468 1 : }
1469 :
1470 8 : Eid UbTransportLiteImpl::GetLocEid() const { return connVec[0]->GetLocEid(); }
1471 :
1472 8 : Eid UbTransportLiteImpl::GetRmtEid() const { return connVec[0]->GetRmtEid(); }
1473 :
1474 7 : uint64_t UbTransportLiteImpl::GetJettyHandle() const { return connVec[0]->GetJettyHandle(); }
1475 :
1476 7 : uint32_t UbTransportLiteImpl::GetJettyId() const { return connVec[0]->GetJettyId(); }
1477 :
1478 0 : HcclResult UbTransportLiteImpl::Clean()
1479 : {
1480 0 : locNotifyVec.clear();
1481 0 : rmtNotifyVec.clear();
1482 0 : locBufferMap.clear();
1483 0 : rmtBufferVec.clear();
1484 0 : rmtBufferMap.clear();
1485 :
1486 : // 清理connVec，connLite由UbConnLiteMgr管理
1487 0 : for (auto& it : connUniqueIdVec) {
1488 0 : DECTOR_TRY_CATCH("UbTransportLiteImpl", UbConnLiteMgr::GetInstance().Clear(it));
1489 : }
1490 0 : connUniqueIdVec.clear();
1491 0 : connVec.clear();
1492 :
1493 0 : return HCCL_SUCCESS;
1494 : }
1495 :
1496 0 : HcclResult UbTransportLiteImpl::Resume(std::vector<char>& uniqueId)
1497 : {
1498 0 : Init(uniqueId);
1499 0 : return HCCL_SUCCESS;
1500 : }
1501 :
1502 1 : HcclResult UbTransportLiteImpl::Fence()
1503 : {
1504 1 : fence_ = true;
1505 1 : HCCL_INFO("[%s] SUCCESS. fence[%d]", __func__, fence_);
1506 1 : return HCCL_SUCCESS;
1507 : }
1508 :
1509 8 : void UbTransportLiteImpl::SetFenceConfig(SqeConfigLite& cfg)
1510 : {
1511 8 : if (fence_) {
1512 0 : cfg.fence = UB_FENCE_ENABLED;
1513 0 : cfg.placeOdr = UB_STRONG_ORDER;
1514 0 : cfg.compOrder = UB_COMPLETION;
1515 0 : cfg.userConfig = true;
1516 : }
1517 8 : fence_ = false;
1518 8 : }
1519 :
1520 17 : bool UbTransportLiteImpl::IsReportTask()
1521 : {
1522 17 : return taskExceptionEnable_ || DfxProfilingHandlerLite::GetInstance().GetProfL1State();
1523 : }
1524 : } // namespace Hccl
|
