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