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 4 : taskParam.taskPara.DMA.jettyHandle = GetJettyHandle();
459 4 : taskParam.taskPara.DMA.jettyId = GetJettyId();
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 = DfxLinkTypeVal::LINK_UB;
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 2 : if (!IsReportTask()) {
497 0 : return;
498 : }
499 :
500 2 : TaskParam taskParam{};
501 2 : taskParam.taskType = TaskParamType::TASK_UB_REDUCE_INLINE;
502 2 : taskParam.beginTime = ProfGetCurCpuTimestamp();
503 2 : FillTaskParamReducePub(taskParam, src, dst, size, reduceIn);
504 2 : taskParam.taskPara.Reduce.notifyID = INVALID_VALUE_NOTIFYID;
505 :
506 2 : AddTaskCallback(stream, taskId, taskParam);
507 2 : DfxTaskInfo* slot = stream.NextTaskSlot();
508 2 : slot->taskType = TaskParamTypeVal::TASK_UB_REDUCE_INLINE;
509 2 : FillSlotReduceInfo(
510 : slot, stream, taskId, reinterpret_cast<u64>(src), reinterpret_cast<u64>(dst), size, INVALID_U32,
511 2 : static_cast<u8>(ConvertReduceOpToHcclReduceOp(reduceIn.reduceOp)));
512 2 : }
513 :
514 0 : void UbTransportLiteImpl::WriteWithNotifyProfilingProcess(
515 : void* src, void* dst, u64 size, const StreamLite& stream, u32 taskId, u64 notifyId)
516 : {
517 0 : if (!IsReportTask()) {
518 0 : return;
519 : }
520 :
521 0 : TaskParam taskParam{};
522 0 : taskParam.taskType = TaskParamType::TASK_WRITE_WITH_NOTIFY;
523 0 : taskParam.beginTime = ProfGetCurCpuTimestamp();
524 0 : FillTaskParamDmaPub(taskParam, dst, size, DmaOp::HCCL_DMA_WRITE);
525 0 : taskParam.taskPara.DMA.src = src;
526 0 : taskParam.taskPara.DMA.notifyID = notifyId;
527 0 : taskParam.taskPara.DMA.notifyValue = 1;
528 0 : taskParam.taskPara.DMA.jettyHandle = GetJettyHandle();
529 0 : taskParam.taskPara.DMA.jettyId = GetJettyId();
530 :
531 0 : AddTaskCallback(stream, taskId, taskParam);
532 0 : DfxTaskInfo* slot = stream.NextTaskSlot();
533 0 : slot->taskType = TaskParamTypeVal::TASK_WRITE_WITH_NOTIFY;
534 0 : FillSlotUbDmaInfo(
535 : slot, stream, taskId, reinterpret_cast<u64>(src), reinterpret_cast<u64>(dst), size, static_cast<u32>(notifyId));
536 0 : }
537 :
538 0 : void UbTransportLiteImpl::WriteReduceWithNotifyProfilingProcess(
539 : void* src, void* dst, u64 size, const ReduceIn& reduceIn, const StreamLite& stream, u32 taskId, u64 notifyId)
540 : {
541 0 : if (!IsReportTask()) {
542 0 : return;
543 : }
544 :
545 0 : TaskParam taskParam{};
546 0 : taskParam.taskType = TaskParamType::TASK_WRITE_REDUCE_WITH_NOTIFY;
547 0 : taskParam.beginTime = ProfGetCurCpuTimestamp();
548 0 : FillTaskParamReducePub(taskParam, src, dst, size, reduceIn);
549 0 : taskParam.taskPara.Reduce.notifyID = notifyId;
550 :
551 0 : AddTaskCallback(stream, taskId, taskParam);
552 0 : DfxTaskInfo* slot = stream.NextTaskSlot();
553 0 : slot->taskType = TaskParamTypeVal::TASK_WRITE_REDUCE_WITH_NOTIFY;
554 0 : FillSlotReduceInfo(
555 : slot, stream, taskId, reinterpret_cast<u64>(src), reinterpret_cast<u64>(dst), size, static_cast<u32>(notifyId),
556 0 : static_cast<u8>(ConvertReduceOpToHcclReduceOp(reduceIn.reduceOp)));
557 0 : }
558 :
559 0 : void UbTransportLiteImpl::NotifyRecordProfilingProcess(
560 : void* dst, u64 size, const StreamLite& stream, u32 taskId, u64 notifyId)
561 : {
562 0 : if (!IsReportTask()) {
563 0 : return;
564 : }
565 :
566 0 : TaskParam taskParam{};
567 0 : taskParam.taskType = TaskParamType::TASK_UB_INLINE_WRITE;
568 0 : taskParam.beginTime = ProfGetCurCpuTimestamp();
569 0 : FillTaskParamDmaPub(taskParam, dst, size, DmaOp::HCCL_DMA_WRITE);
570 0 : taskParam.taskPara.DMA.notifyID = notifyId;
571 0 : taskParam.taskPara.DMA.notifyValue = 1;
572 0 : taskParam.taskPara.DMA.jettyHandle = GetJettyHandle();
573 0 : taskParam.taskPara.DMA.jettyId = GetJettyId();
574 :
575 0 : AddTaskCallback(stream, taskId, taskParam);
576 0 : DfxTaskInfo* slot = stream.NextTaskSlot();
577 0 : slot->taskType = TaskParamTypeVal::TASK_UB_INLINE_WRITE;
578 0 : FillSlotUbDmaInfo(slot, stream, taskId, 0, reinterpret_cast<u64>(dst), size, static_cast<u32>(notifyId));
579 0 : }
580 :
581 5 : void UbTransportLiteImpl::FillSlotUbDmaInfo(
582 : DfxTaskInfo* slot, const StreamLite& stream, u32 taskId, u64 srcAddr, u64 dstAddr, u64 size, u32 notifyId)
583 : {
584 5 : slot->sqId = stream.GetSqId();
585 5 : slot->taskId = taskId;
586 5 : const void* opInfo = stream.GetLatestDfxOpInfo();
587 5 : slot->dfxOpInfo = (opInfo != nullptr) ? reinterpret_cast<u64>(opInfo) : INVALID_U64;
588 5 : slot->linkType = DfxLinkTypeVal::LINK_UB;
589 5 : slot->transportType = static_cast<u8>(DfxTransportType::DFX_TRANSPORT_TYPE_UB);
590 5 : slot->channelHandle = reinterpret_cast<u64>(this);
591 5 : slot->taskPara.ubDma.sqeAddr = stream.GetRtsq()->GetSqeAddr();
592 5 : slot->taskPara.ubDma.srcAddr = srcAddr;
593 5 : slot->taskPara.ubDma.dstAddr = dstAddr;
594 5 : slot->taskPara.ubDma.size = size;
595 5 : slot->taskPara.ubDma.notifyId = notifyId;
596 5 : }
597 :
598 2 : void UbTransportLiteImpl::FillSlotReduceInfo(
599 : DfxTaskInfo* slot, const StreamLite& stream, u32 taskId, u64 srcAddr, u64 dstAddr, u64 size, u32 notifyId,
600 : u8 reduceOp)
601 : {
602 2 : slot->sqId = stream.GetSqId();
603 2 : slot->taskId = taskId;
604 2 : const void* opInfo = stream.GetLatestDfxOpInfo();
605 2 : slot->dfxOpInfo = (opInfo != nullptr) ? reinterpret_cast<u64>(opInfo) : INVALID_U64;
606 2 : slot->linkType = DfxLinkTypeVal::LINK_UB;
607 2 : slot->transportType = static_cast<u8>(DfxTransportType::DFX_TRANSPORT_TYPE_UB);
608 2 : slot->channelHandle = reinterpret_cast<u64>(this);
609 2 : slot->taskPara.Reduce.sqeAddr = stream.GetRtsq()->GetSqeAddr();
610 2 : slot->taskPara.Reduce.srcAddr = srcAddr;
611 2 : slot->taskPara.Reduce.dstAddr = dstAddr;
612 2 : slot->taskPara.Reduce.size = size;
613 2 : slot->taskPara.Reduce.notifyId = notifyId;
614 2 : slot->taskPara.Reduce.reduceOp = reduceOp;
615 2 : }
616 :
617 1 : void UbTransportLiteImpl::Read(const RmaBufferLite& loc, const Buffer& rmt, const StreamLite& stream)
618 : {
619 1 : SqeConfigLite cfg;
620 1 : SetFenceConfig(cfg);
621 :
622 1 : auto taskId = stream.GetRtsq()->GetTaskId();
623 :
624 : // 当前使用1个connection,下标为0
625 1 : RmaConnLite* conn = connVec[0];
626 :
627 : // 展开下发WQE前, 按需设置cache context
628 1 : UbConnLite* ubConnLitePtr = nullptr;
629 1 : bool needCacheTask = false;
630 1 : PreLaunchWqe(ubConnLitePtr, needCacheTask, conn);
631 : // 下发DbSqe前, 备份相关信息
632 1 : const uint32_t pendingSqeCnt = needCacheTask ? stream.GetRtsq()->GetPendingSqeCnt() : 0;
633 :
634 : // 展开下发WQE
635 1 : auto locRmaBufSlicelite = GetRmaBufSlicelite(loc);
636 1 : auto rmtRmaBufSlicelite = GetRmtRmaBufSliceLite(rmt);
637 1 : conn->Read(locRmaBufSlicelite, rmtRmaBufSlicelite, cfg, stream, connOut);
638 :
639 : // 展开下发WQE后, 展开下发DbSqe前, 按需缓存wqe及DbSqeIdx
640 : // 注意: pendingSqeCnt在下发DbSqe前已备份
641 : // 注意: 一定要在展开下发DbSqe前调用PostLaunchWqe, 否则如果下发DbSqe触发LaunchTask,
642 : // 而尚未调用PostLaunchWqe插入当前WQE数组，
643 : // 会导致AicpuTaskCache找不到当前WQE数组, 无法正确更新对应的DbSqeLocation
644 1 : const bool isReportTask = IsReportTask();
645 1 : DbSqeProfInfo dbSqeProfInfo;
646 1 : if (needCacheTask && isReportTask) {
647 0 : BuildDbSqeProfInfoForProfilingProcess(
648 0 : reinterpret_cast<void*>(locRmaBufSlicelite.GetAddr()),
649 0 : reinterpret_cast<void*>(rmtRmaBufSlicelite.GetAddr()), locRmaBufSlicelite.GetSize(), DmaOp::HCCL_DMA_READ,
650 : dbSqeProfInfo);
651 : }
652 1 : PostLaunchWqe(stream, ubConnLitePtr, needCacheTask, pendingSqeCnt, isReportTask, dbSqeProfInfo);
653 :
654 1 : BuildUbDbSendTask(stream, conn->GetUbJettyLiteId(), connOut.pi);
655 :
656 2 : ProfilingProcess(
657 1 : reinterpret_cast<void*>(locRmaBufSlicelite.GetAddr()), reinterpret_cast<void*>(rmtRmaBufSlicelite.GetAddr()),
658 : locRmaBufSlicelite.GetSize(), stream, DmaOp::HCCL_DMA_READ, taskId);
659 1 : }
660 :
661 1 : void UbTransportLiteImpl::Write(const RmaBufferLite& loc, const Buffer& rmt, const StreamLite& stream)
662 : {
663 1 : SqeConfigLite cfg;
664 1 : SetFenceConfig(cfg);
665 :
666 1 : auto taskId = stream.GetRtsq()->GetTaskId();
667 :
668 : // 当前使用1个connection，下标为0
669 1 : RmaConnLite* conn = connVec[0];
670 :
671 : // 展开下发WQE前, 按需设置cache context
672 1 : UbConnLite* ubConnLitePtr = nullptr;
673 1 : bool needCacheTask = false;
674 1 : PreLaunchWqe(ubConnLitePtr, needCacheTask, conn);
675 : // 下发DbSqe前, 备份相关信息
676 1 : const uint32_t pendingSqeCnt = needCacheTask ? stream.GetRtsq()->GetPendingSqeCnt() : 0;
677 :
678 : // 展开下发WQE
679 1 : auto locRmaBufSlicelite = GetRmaBufSlicelite(loc);
680 1 : auto rmtRmaBufSlicelite = GetRmtRmaBufSliceLite(rmt);
681 1 : conn->Write(locRmaBufSlicelite, rmtRmaBufSlicelite, cfg, stream, connOut);
682 :
683 : // 展开下发WQE后, 展开下发DbSqe前, 按需缓存wqe及DbSqeIdx
684 : // 注意: pendingSqeCnt在下发DbSqe前已备份
685 : // 注意: 一定要在展开下发DbSqe前调用PostLaunchWqe, 否则如果下发DbSqe触发LaunchTask,
686 : // 而尚未调用PostLaunchWqe插入当前WQE数组，
687 : // 会导致AicpuTaskCache找不到当前WQE数组, 无法正确更新对应的DbSqeLocation
688 1 : const bool isReportTask = IsReportTask();
689 1 : DbSqeProfInfo dbSqeProfInfo;
690 1 : if (needCacheTask && isReportTask) {
691 0 : BuildDbSqeProfInfoForProfilingProcess(
692 0 : reinterpret_cast<void*>(locRmaBufSlicelite.GetAddr()),
693 0 : reinterpret_cast<void*>(rmtRmaBufSlicelite.GetAddr()), locRmaBufSlicelite.GetSize(), DmaOp::HCCL_DMA_WRITE,
694 : dbSqeProfInfo);
695 : }
696 1 : PostLaunchWqe(stream, ubConnLitePtr, needCacheTask, pendingSqeCnt, isReportTask, dbSqeProfInfo);
697 :
698 1 : BuildUbDbSendTask(stream, conn->GetUbJettyLiteId(), connOut.pi);
699 :
700 2 : ProfilingProcess(
701 1 : reinterpret_cast<void*>(locRmaBufSlicelite.GetAddr()), reinterpret_cast<void*>(rmtRmaBufSlicelite.GetAddr()),
702 : locRmaBufSlicelite.GetSize(), stream, DmaOp::HCCL_DMA_WRITE, taskId);
703 1 : }
704 :
705 1 : void UbTransportLiteImpl::ReadReduce(
706 : const RmaBufferLite& loc, const Buffer& rmt, const ReduceIn& reduceIn, const StreamLite& stream)
707 : {
708 1 : SqeConfigLite cfg;
709 1 : SetFenceConfig(cfg);
710 :
711 1 : auto taskId = stream.GetRtsq()->GetTaskId();
712 :
713 : // 当前使用1个connection，下标为0
714 1 : RmaConnLite* conn = connVec[0];
715 :
716 : // 展开下发WQE前, 按需设置cache context
717 1 : UbConnLite* ubConnLitePtr = nullptr;
718 1 : bool needCacheTask = false;
719 1 : PreLaunchWqe(ubConnLitePtr, needCacheTask, conn);
720 : // 下发DbSqe前, 备份相关信息
721 1 : const uint32_t pendingSqeCnt = needCacheTask ? stream.GetRtsq()->GetPendingSqeCnt() : 0;
722 :
723 : // 展开下发WQE
724 1 : auto locRmaBufSlicelite = GetRmaBufSlicelite(loc);
725 1 : auto rmtRmaBufSlicelite = GetRmtRmaBufSliceLite(rmt);
726 1 : conn->ReadReduce(reduceIn, locRmaBufSlicelite, rmtRmaBufSlicelite, stream, cfg, connOut);
727 :
728 : // 展开下发WQE后, 展开下发DbSqe前, 按需缓存wqe及DbSqeIdx
729 : // 注意: pendingSqeCnt在下发DbSqe前已备份
730 : // 注意: 一定要在展开下发DbSqe前调用PostLaunchWqe, 否则如果下发DbSqe触发LaunchTask,
731 : // 而尚未调用PostLaunchWqe插入当前WQE数组，
732 : // 会导致AicpuTaskCache找不到当前WQE数组, 无法正确更新对应的DbSqeLocation
733 1 : const bool isReportTask = IsReportTask();
734 1 : DbSqeProfInfo dbSqeProfInfo;
735 1 : if (needCacheTask && isReportTask) {
736 0 : BuildDbSqeProfInfoForReduceProfilingProcess(
737 0 : reinterpret_cast<void*>(locRmaBufSlicelite.GetAddr()),
738 0 : reinterpret_cast<void*>(rmtRmaBufSlicelite.GetAddr()), locRmaBufSlicelite.GetSize(), reduceIn,
739 : dbSqeProfInfo);
740 : }
741 1 : PostLaunchWqe(stream, ubConnLitePtr, needCacheTask, pendingSqeCnt, isReportTask, dbSqeProfInfo);
742 :
743 1 : BuildUbDbSendTask(stream, conn->GetUbJettyLiteId(), connOut.pi);
744 :
745 2 : ReduceProfilingProcess(
746 1 : reinterpret_cast<void*>(locRmaBufSlicelite.GetAddr()), reinterpret_cast<void*>(rmtRmaBufSlicelite.GetAddr()),
747 : locRmaBufSlicelite.GetSize(), reduceIn, stream, taskId);
748 1 : }
749 :
750 1 : void UbTransportLiteImpl::WriteReduce(
751 : const RmaBufferLite& loc, const Buffer& rmt, const ReduceIn& reduceIn, const StreamLite& stream)
752 : {
753 1 : SqeConfigLite cfg;
754 1 : SetFenceConfig(cfg);
755 :
756 1 : auto taskId = stream.GetRtsq()->GetTaskId();
757 :
758 : // 当前使用1个connection，下标为0
759 1 : RmaConnLite* conn = connVec[0];
760 :
761 : // 展开下发WQE前, 按需设置cache context
762 1 : UbConnLite* ubConnLitePtr = nullptr;
763 1 : bool needCacheTask = false;
764 1 : PreLaunchWqe(ubConnLitePtr, needCacheTask, conn);
765 : // 下发DbSqe前, 备份相关信息
766 1 : const uint32_t pendingSqeCnt = needCacheTask ? stream.GetRtsq()->GetPendingSqeCnt() : 0;
767 :
768 : // 展开下发WQE
769 1 : auto locRmaBufSlicelite = GetRmaBufSlicelite(loc);
770 1 : auto rmtRmaBufSlicelite = GetRmtRmaBufSliceLite(rmt);
771 1 : conn->WriteReduce(
772 1 : reduceIn.dataType, reduceIn.reduceOp, locRmaBufSlicelite, stream, rmtRmaBufSlicelite, cfg, connOut);
773 :
774 : // 展开下发WQE后, 展开下发DbSqe前, 按需缓存wqe及DbSqeIdx
775 : // 注意: pendingSqeCnt在下发DbSqe前已备份
776 : // 注意: 一定要在展开下发DbSqe前调用PostLaunchWqe, 否则如果下发DbSqe触发LaunchTask,
777 : // 而尚未调用PostLaunchWqe插入当前WQE数组，
778 : // 会导致AicpuTaskCache找不到当前WQE数组, 无法正确更新对应的DbSqeLocation
779 1 : const bool isReportTask = IsReportTask();
780 1 : DbSqeProfInfo dbSqeProfInfo;
781 1 : if (needCacheTask && isReportTask) {
782 0 : BuildDbSqeProfInfoForReduceProfilingProcess(
783 0 : reinterpret_cast<void*>(locRmaBufSlicelite.GetAddr()),
784 0 : reinterpret_cast<void*>(rmtRmaBufSlicelite.GetAddr()), locRmaBufSlicelite.GetSize(), reduceIn,
785 : dbSqeProfInfo);
786 : }
787 1 : PostLaunchWqe(stream, ubConnLitePtr, needCacheTask, pendingSqeCnt, isReportTask, dbSqeProfInfo);
788 :
789 1 : BuildUbDbSendTask(stream, conn->GetUbJettyLiteId(), connOut.pi);
790 :
791 2 : ReduceProfilingProcess(
792 1 : reinterpret_cast<void*>(locRmaBufSlicelite.GetAddr()), reinterpret_cast<void*>(rmtRmaBufSlicelite.GetAddr()),
793 : locRmaBufSlicelite.GetSize(), reduceIn, stream, taskId);
794 1 : }
795 :
796 1 : void UbTransportLiteImpl::ExecProfiling(
797 : const RmaBufferLite& loc, const Buffer& rmt, const u64 totalSize,
798 : const BaseTransportLiteImpl::TransferOp& transferOp, const StreamLite& stream, u32 taskId)
799 : {
800 1 : if (transferOp.reduceIn.reduceOp == ReduceOp::INVALID) {
801 1 : DmaOp dmaOp = DmaOp::HCCL_DMA_WRITE;
802 1 : if (transferOp.transType == TransferType::READ) {
803 1 : dmaOp = DmaOp::HCCL_DMA_READ;
804 : }
805 1 : ProfilingProcess(
806 1 : reinterpret_cast<void*>(GetRmaBufSlicelite(loc).GetAddr()),
807 2 : reinterpret_cast<void*>(GetRmtRmaBufSliceLite(rmt).GetAddr()), totalSize, stream, dmaOp, taskId);
808 : } else {
809 0 : ReduceProfilingProcess(
810 0 : reinterpret_cast<void*>(GetRmaBufSlicelite(loc).GetAddr()),
811 0 : reinterpret_cast<void*>(GetRmtRmaBufSliceLite(rmt).GetAddr()), totalSize, transferOp.reduceIn, stream,
812 : taskId);
813 : }
814 1 : }
815 :
816 0 : void UbTransportLiteImpl::ExecProfilingAll(
817 : const RmaBufferLite& loc, const Buffer& rmt, const u64 totalSize,
818 : const BaseTransportLiteImpl::TransferOp& transferOp, const StreamLite& stream, u32 taskId, const uint32_t notifyId)
819 : {
820 0 : if (transferOp.transType == TransferType::READ) {
821 0 : ProfilingProcess(
822 0 : reinterpret_cast<void*>(GetRmaBufSlicelite(loc).GetAddr()),
823 0 : reinterpret_cast<void*>(GetRmtRmaBufSliceLite(rmt).GetAddr()), totalSize, stream, DmaOp::HCCL_DMA_READ,
824 : taskId);
825 0 : } else if (transferOp.transType == TransferType::WRITE) {
826 0 : ProfilingProcess(
827 0 : reinterpret_cast<void*>(GetRmaBufSlicelite(loc).GetAddr()),
828 0 : reinterpret_cast<void*>(GetRmtRmaBufSliceLite(rmt).GetAddr()), totalSize, stream, DmaOp::HCCL_DMA_WRITE,
829 : taskId);
830 0 : } else if (transferOp.transType == TransferType::READ_REDUCE) {
831 0 : ReduceProfilingProcess(
832 0 : reinterpret_cast<void*>(GetRmaBufSlicelite(loc).GetAddr()),
833 0 : reinterpret_cast<void*>(GetRmtRmaBufSliceLite(rmt).GetAddr()), totalSize, transferOp.reduceIn, stream,
834 : taskId);
835 0 : } else if (transferOp.transType == TransferType::WRITE_REDUCE) {
836 0 : ReduceProfilingProcess(
837 0 : reinterpret_cast<void*>(GetRmaBufSlicelite(loc).GetAddr()),
838 0 : reinterpret_cast<void*>(GetRmtRmaBufSliceLite(rmt).GetAddr()), totalSize, transferOp.reduceIn, stream,
839 : taskId);
840 0 : } else if (transferOp.transType == TransferType::WRITE_WITH_NOTIFY) {
841 0 : WriteWithNotifyProfilingProcess(
842 0 : reinterpret_cast<void*>(GetRmaBufSlicelite(loc).GetAddr()),
843 0 : reinterpret_cast<void*>(GetRmtRmaBufSliceLite(rmt).GetAddr()), totalSize, stream, taskId,
844 0 : GetRmtNotifySliceLite(notifyId).GetAddr());
845 0 : } else if (transferOp.transType == TransferType::WRITE_REDUCE_WITH_NOTIFY) {
846 0 : WriteReduceWithNotifyProfilingProcess(
847 0 : reinterpret_cast<void*>(GetRmaBufSlicelite(loc).GetAddr()),
848 0 : reinterpret_cast<void*>(GetRmtRmaBufSliceLite(rmt).GetAddr()), totalSize, transferOp.reduceIn, stream,
849 0 : taskId, GetRmtNotifySliceLite(notifyId).GetAddr());
850 0 : } else if (transferOp.transType == TransferType::NOTIFY_RECORD) {
851 0 : NotifyRecordProfilingProcess(
852 0 : reinterpret_cast<void*>(GetRmtNotifySliceLite(notifyId).GetAddr()),
853 0 : GetRmtNotifySliceLite(notifyId).GetSize(), stream, taskId, GetRmtNotifySliceLite(notifyId).GetAddr());
854 : }
855 0 : }
856 :
857 1 : void UbTransportLiteImpl::BatchTransfer(
858 : const std::vector<RmaBufferLite>& loc, const std::vector<Buffer>& rmt,
859 : const std::vector<BaseTransportLiteImpl::TransferOp>& transferOp, const StreamLite& stream)
860 : {
861 1 : if (UNLIKELY(loc.empty())) {
862 0 : return;
863 : }
864 1 : SqeConfigLite cfg;
865 1 : SetFenceConfig(cfg);
866 :
867 1 : auto taskId = stream.GetRtsq()->GetTaskId();
868 :
869 : // 当前使用1个connection，下标为0 (当前只有一个connection，对应一个jetty)
870 1 : RmaConnLite* conn = connVec[0];
871 :
872 : // 展开下发WQE前, 按需设置cache context
873 1 : UbConnLite* ubConnLitePtr = nullptr;
874 1 : bool needCacheTask = false;
875 1 : PreLaunchWqe(ubConnLitePtr, needCacheTask, conn);
876 : // 下发DbSqe前, 备份相关信息
877 1 : const uint32_t pendingSqeCnt = needCacheTask ? stream.GetRtsq()->GetPendingSqeCnt() : 0;
878 :
879 1 : u32 insNum = loc.size();
880 2 : for (u32 i = 0; i < insNum; i++) {
881 1 : cfg.cqeEn = (i == insNum - 1) ? true : false; // 返回最后一个sqe的cqe
882 1 : cfg.placeOdr = UB_RELAX_ORDER;
883 1 : cfg.compOrder = UB_NO_COMPLETION;
884 1 : cfg.userConfig = true;
885 :
886 1 : auto localBuffer = GetRmaBufSlicelite(loc[i]);
887 1 : auto remoteBuffer = GetRmtRmaBufSliceLite(rmt[i]);
888 1 : if (transferOp[i].transType == TransferType::WRITE) {
889 0 : conn->Write(localBuffer, remoteBuffer, cfg, stream, connOut); // 当前只有一个connection，对应一个jetty
890 1 : } else if (transferOp[i].transType == TransferType::WRITE_REDUCE) { // write reduce
891 0 : conn->WriteReduce(
892 0 : transferOp[i].reduceIn.dataType, transferOp[i].reduceIn.reduceOp, localBuffer, stream, remoteBuffer,
893 0 : cfg, connOut);
894 1 : } else if (transferOp[i].transType == TransferType::READ) {
895 1 : conn->Read(localBuffer, remoteBuffer, cfg, stream, connOut); // 当前只有一个connection，对应一个jetty
896 0 : } else if (transferOp[i].transType == TransferType::READ_REDUCE) { // read reduce
897 0 : conn->ReadReduce(transferOp[i].reduceIn, localBuffer, remoteBuffer, stream, cfg, connOut);
898 : }
899 : }
900 :
901 : // 按需计算totalSize
902 1 : const bool isReportTask = IsReportTask();
903 1 : u64 totalSize = 0;
904 1 : if (isReportTask) {
905 2 : for (u32 i = 0; i < insNum; i++) {
906 1 : totalSize += GetRmaBufSlicelite(loc[i]).GetSize();
907 : }
908 : }
909 :
910 : // 展开下发WQE后, 展开下发DbSqe前, 按需缓存wqe及DbSqeIdx
911 : // 注意: pendingSqeCnt在下发DbSqe前已备份
912 : // 注意: 一定要在展开下发DbSqe前调用PostLaunchWqe, 否则如果下发DbSqe触发LaunchTask,
913 : // 而尚未调用PostLaunchWqe插入当前WQE数组，
914 : // 会导致AicpuTaskCache找不到当前WQE数组, 无法正确更新对应的DbSqeLocation
915 1 : DbSqeProfInfo dbSqeProfInfo;
916 1 : if (needCacheTask && isReportTask) {
917 0 : BuildDbSqeProfInfoForExecProfiling(
918 0 : loc[insNum - 1], rmt[insNum - 1], totalSize, transferOp[insNum - 1], dbSqeProfInfo);
919 : }
920 1 : PostLaunchWqe(stream, ubConnLitePtr, needCacheTask, pendingSqeCnt, isReportTask, dbSqeProfInfo);
921 :
922 1 : BuildUbDbSendTask(stream, conn->GetUbJettyLiteId(), connOut.pi);
923 :
924 1 : ExecProfiling(loc[insNum - 1], rmt[insNum - 1], totalSize, transferOp[insNum - 1], stream, taskId);
925 : }
926 :
927 : // Convert hccl::HcommDataType => Hccl::DataType, hccl::HcommReduceOp => Hccl::ReduceOp
928 : static const std::unordered_map<HcommReduceOp, Hccl::ReduceOp> mapHcommReduceOpA5
929 : = {{HcommReduceOp::HCOMM_REDUCE_SUM, Hccl::ReduceOp::SUM},
930 : {HcommReduceOp::HCOMM_REDUCE_PROD, Hccl::ReduceOp::PROD},
931 : {HcommReduceOp::HCOMM_REDUCE_MAX, Hccl::ReduceOp::MAX},
932 : {HcommReduceOp::HCOMM_REDUCE_MIN, Hccl::ReduceOp::MIN},
933 : {HcommReduceOp::HCOMM_REDUCE_RESERVED, Hccl::ReduceOp::INVALID}};
934 :
935 : static const std::unordered_map<HcommDataType, Hccl::DataType> mapHcommDataTypeA5 = {
936 : #ifndef OPEN_BUILD_PROJECT
937 : {HcommDataType::HCOMM_DATA_TYPE_HIF8, Hccl::DataType::HIF8},
938 : {HcommDataType::HCOMM_DATA_TYPE_FP8E4M3, Hccl::DataType::FP8E4M3},
939 : {HcommDataType::HCOMM_DATA_TYPE_FP8E5M2, Hccl::DataType::FP8E5M2},
940 : {HcommDataType::HCOMM_DATA_TYPE_FP8E8M0, Hccl::DataType::FP8E8M0},
941 : #endif
942 : {HcommDataType::HCOMM_DATA_TYPE_INT8, Hccl::DataType::INT8},
943 : {HcommDataType::HCOMM_DATA_TYPE_INT16, Hccl::DataType::INT16},
944 : {HcommDataType::HCOMM_DATA_TYPE_INT32, Hccl::DataType::INT32},
945 : {HcommDataType::HCOMM_DATA_TYPE_INT64, Hccl::DataType::INT64},
946 : {HcommDataType::HCOMM_DATA_TYPE_INT128, Hccl::DataType::INT128},
947 : {HcommDataType::HCOMM_DATA_TYPE_UINT8, Hccl::DataType::UINT8},
948 : {HcommDataType::HCOMM_DATA_TYPE_UINT16, Hccl::DataType::UINT16},
949 : {HcommDataType::HCOMM_DATA_TYPE_UINT32, Hccl::DataType::UINT32},
950 : {HcommDataType::HCOMM_DATA_TYPE_UINT64, Hccl::DataType::UINT64},
951 : {HcommDataType::HCOMM_DATA_TYPE_FP16, Hccl::DataType::FP16},
952 : {HcommDataType::HCOMM_DATA_TYPE_FP32, Hccl::DataType::FP32},
953 : {HcommDataType::HCOMM_DATA_TYPE_FP64, Hccl::DataType::FP64},
954 : {HcommDataType::HCOMM_DATA_TYPE_BFP16, Hccl::DataType::BFP16},
955 : {HcommDataType::HCOMM_DATA_TYPE_RESERVED, Hccl::DataType::INVALID}};
956 :
957 13 : static HcclResult CheckReduceHcommDataTypeAndHcommReduceOp(HcommDataType dataType, HcommReduceOp reduceOp)
958 : {
959 13 : auto dataTypeIt = mapHcommDataTypeA5.find(dataType); // reduce类型，dataType不能是RESERVED
960 13 : if (dataTypeIt == mapHcommDataTypeA5.end() || dataTypeIt->first == HcommDataType::HCOMM_DATA_TYPE_RESERVED) {
961 0 : HCCL_ERROR("[%s] type[%u] is not supported.", __func__, dataType);
962 0 : return HCCL_E_PARA;
963 : }
964 :
965 13 : auto reduceOpIt = mapHcommReduceOpA5.find(reduceOp); // reduce类型，reduceOp不能是RESERVED
966 13 : if (reduceOpIt == mapHcommReduceOpA5.end() || reduceOpIt->first == HcommReduceOp::HCOMM_REDUCE_RESERVED) {
967 0 : HCCL_ERROR("[%s] op[%u] is not supported.", __func__, reduceOp);
968 0 : return HCCL_E_PARA;
969 : }
970 :
971 13 : return HCCL_SUCCESS;
972 : }
973 :
974 : constexpr u32 SIZE_TABLE[HCCL_DATA_TYPE_RESERVED]
975 : = {sizeof(s8),
976 : sizeof(s16),
977 : sizeof(s32),
978 : 2,
979 : sizeof(float),
980 : sizeof(s64),
981 : sizeof(u64),
982 : sizeof(u8),
983 : sizeof(u16),
984 : sizeof(u32),
985 : 8,
986 : 2,
987 : 16,
988 : 2,
989 : 1,
990 : 1,
991 : 1,
992 : 1};
993 :
994 8 : static HcclResult ParasReduceData(
995 : const HcommBatchTransferDesc& transferDesc, uint64_t& len, HcommDataType& dataType, HcommReduceOp& reduceOp)
996 : {
997 8 : len = transferDesc.transferInfo.reduce.count;
998 8 : dataType = transferDesc.transferInfo.reduce.dataType;
999 8 : reduceOp = transferDesc.transferInfo.reduce.reduceOp;
1000 8 : auto ret = CheckReduceHcommDataTypeAndHcommReduceOp(dataType, reduceOp);
1001 8 : CHK_PRT_RET(
1002 : ret != HCCL_SUCCESS,
1003 : HCCL_ERROR("FAIL at CheckReduceHcommDataTypeAndHcommReduceOp dataType[%d], reduceOp[%d].", dataType, reduceOp),
1004 : ret);
1005 8 : return HCCL_SUCCESS;
1006 : }
1007 :
1008 34 : static HcclResult ParseData(
1009 : const HcommBatchTransferDesc& transferDesc, void*& rmt, void*& loc, uint64_t& len, Hccl::TransferType& tfType,
1010 : HcommDataType& dataType, HcommReduceOp& reduceOp, uint32_t& notifyIdx)
1011 : {
1012 34 : if (transferDesc.transType == HCOMM_TRANSFER_TYPE_WRITE) {
1013 7 : rmt = transferDesc.transferInfo.write.dst; // write操作，dst是远端地址
1014 7 : loc = transferDesc.transferInfo.write.src; // src是本端地址
1015 7 : len = transferDesc.transferInfo.write.len;
1016 7 : tfType = Hccl::TransferType::WRITE;
1017 27 : } else if (transferDesc.transType == HCOMM_TRANSFER_TYPE_READ) {
1018 5 : rmt = transferDesc.transferInfo.read.src; // read操作，src是远端地址
1019 5 : loc = transferDesc.transferInfo.read.dst; // dst是本端地址
1020 5 : len = transferDesc.transferInfo.read.len;
1021 5 : tfType = Hccl::TransferType::READ;
1022 22 : } else if (transferDesc.transType == HCOMM_TRANSFER_TYPE_WRITE_REDUCE) {
1023 4 : rmt = transferDesc.transferInfo.reduce.dst;
1024 4 : loc = transferDesc.transferInfo.reduce.src;
1025 4 : tfType = Hccl::TransferType::WRITE_REDUCE;
1026 4 : CHK_RET(ParasReduceData(transferDesc, len, dataType, reduceOp));
1027 18 : } else if (transferDesc.transType == HCOMM_TRANSFER_TYPE_READ_REDUCE) {
1028 4 : rmt = transferDesc.transferInfo.reduce.src;
1029 4 : loc = transferDesc.transferInfo.reduce.dst;
1030 4 : tfType = Hccl::TransferType::READ_REDUCE;
1031 4 : CHK_RET(ParasReduceData(transferDesc, len, dataType, reduceOp));
1032 14 : } else if (transferDesc.transType == HCOMM_TRANSFER_TYPE_WRITE_WITH_NOTIFY) {
1033 4 : rmt = transferDesc.transferInfo.writeWithNotify.dst; // write操作，dst是远端地址
1034 4 : loc = transferDesc.transferInfo.writeWithNotify.src; // src是本端地址
1035 4 : len = transferDesc.transferInfo.writeWithNotify.len;
1036 4 : notifyIdx = transferDesc.transferInfo.writeWithNotify.notifyIdx;
1037 4 : tfType = Hccl::TransferType::WRITE_WITH_NOTIFY;
1038 10 : } else if (transferDesc.transType == HCOMM_TRANSFER_TYPE_WRITE_REDUCE_WITH_NOTIFY) {
1039 5 : rmt = transferDesc.transferInfo.writeReduceWithNotify.dst;
1040 5 : loc = transferDesc.transferInfo.writeReduceWithNotify.src;
1041 5 : len = transferDesc.transferInfo.writeReduceWithNotify.count;
1042 5 : dataType = transferDesc.transferInfo.writeReduceWithNotify.dataType;
1043 5 : reduceOp = transferDesc.transferInfo.writeReduceWithNotify.reduceOp;
1044 5 : notifyIdx = transferDesc.transferInfo.writeReduceWithNotify.notifyIdx;
1045 5 : tfType = Hccl::TransferType::WRITE_REDUCE_WITH_NOTIFY;
1046 5 : CHK_RET(CheckReduceHcommDataTypeAndHcommReduceOp(dataType, reduceOp));
1047 5 : } else if (transferDesc.transType == HCOMM_TRANSFER_TYPE_NOTIFY_RECORD) {
1048 4 : notifyIdx = transferDesc.transferInfo.notifyRecord.notifyIdx;
1049 4 : tfType = Hccl::TransferType::NOTIFY_RECORD;
1050 : } else {
1051 1 : HCCL_ERROR("[%s] unsupported transType[%d]", __func__, transferDesc.transType);
1052 1 : return HCCL_E_NOT_SUPPORT;
1053 : }
1054 33 : if (reduceOp != HcommReduceOp::HCOMM_REDUCE_RESERVED) { // 对于规约类型, size = count * sizeof(datatype)
1055 13 : len = len * SIZE_TABLE[dataType];
1056 : }
1057 33 : return HCCL_SUCCESS;
1058 : }
1059 : constexpr uint32_t NOTIFYIDX_INVALID_VALUE = 0xFFFFFFFF; // NOTIFY idex非法值
1060 12 : HcclResult UbTransportLiteImpl::ExecuteBatchTransfer(
1061 : StreamLite* streamLitePtr, const HcommBatchTransferDesc* transferDescs, uint32_t transferDescNum)
1062 : {
1063 12 : std::vector<Hccl::RmaBufferLite> locSlices;
1064 12 : std::vector<Hccl::Buffer> rmtSlices;
1065 12 : std::vector<Hccl::BaseTransportLiteImpl::TransferOp> transferOps;
1066 12 : std::vector<uint32_t> notifyIdxs;
1067 :
1068 12 : locSlices.reserve(transferDescNum);
1069 12 : rmtSlices.reserve(transferDescNum);
1070 12 : transferOps.reserve(transferDescNum);
1071 12 : notifyIdxs.reserve(transferDescNum);
1072 :
1073 42 : for (uint32_t i = 0; i < transferDescNum; i++) {
1074 34 : Hccl::RmaBufferLite locRmaBuf;
1075 34 : void* rmt = nullptr;
1076 34 : void* loc = nullptr;
1077 34 : uint64_t len = 0;
1078 34 : Hccl::TransferType tfType;
1079 34 : HcommDataType dataType{HcommDataType::HCOMM_DATA_TYPE_RESERVED};
1080 34 : HcommReduceOp reduceOp{HcommReduceOp::HCOMM_REDUCE_RESERVED};
1081 34 : uint32_t notifyIdx = NOTIFYIDX_INVALID_VALUE;
1082 37 : CHK_RET(ParseData(transferDescs[i], rmt, loc, len, tfType, dataType, reduceOp, notifyIdx));
1083 33 : if (tfType != Hccl::TransferType::NOTIFY_RECORD) { // NOTIFY_RECORD时没有地址字段
1084 29 : CHK_PTR_NULL(rmt);
1085 28 : CHK_PTR_NULL(loc);
1086 27 : HcclResult ret = BuildLocRmaBufferLite(reinterpret_cast<uintptr_t>(loc), len, locRmaBuf);
1087 27 : CHK_PRT_RET(
1088 : ret != HCCL_SUCCESS,
1089 : HCCL_ERROR(
1090 : "[%s] FAIL at BuildLocRmaBufferLite for index %u. rmt[%p], loc[%p], len[0x%llx], tfType[%u], "
1091 : "dataType[%d], reduceOp[%d].",
1092 : __func__, i, rmt, loc, len, tfType, dataType, reduceOp),
1093 : ret);
1094 : }
1095 58 : if (tfType == Hccl::TransferType::NOTIFY_RECORD || tfType == Hccl::TransferType::WRITE_WITH_NOTIFY
1096 58 : || tfType == Hccl::TransferType::WRITE_REDUCE_WITH_NOTIFY) {
1097 13 : CHK_PRT_RET(
1098 : notifyIdx == NOTIFYIDX_INVALID_VALUE,
1099 : HCCL_ERROR(
1100 : "[%s] FAIL at ParseData for index %u. tfType[%u], notifyIdx[%u].", __func__, i, tfType, notifyIdx),
1101 : HCCL_E_PARA);
1102 : }
1103 30 : notifyIdxs.push_back(notifyIdx);
1104 30 : locSlices.push_back(locRmaBuf);
1105 :
1106 30 : const Hccl::Buffer rmtBuf{reinterpret_cast<uintptr_t>(rmt), len};
1107 30 : rmtSlices.push_back(rmtBuf);
1108 :
1109 30 : Hccl::ReduceIn reduceIn{mapHcommDataTypeA5.at(dataType), mapHcommReduceOpA5.at(reduceOp)};
1110 :
1111 30 : transferOps.push_back(Hccl::BaseTransportLiteImpl::TransferOp{tfType, reduceIn});
1112 :
1113 30 : HCCL_DEBUG(
1114 : "[%s] Prepared transfer op for index %u. rmt[%p], loc[%p], len[0x%llx], tfType[%u], dataType[%d], "
1115 : "reduceOp[%d].",
1116 : __func__, i, rmt, loc, len, tfType, dataType, reduceOp);
1117 30 : }
1118 8 : EXCEPTION_CATCH(
1119 : BatchTransferAll(locSlices, rmtSlices, transferOps, notifyIdxs, *streamLitePtr), return HCCL_E_INTERNAL);
1120 8 : return HCCL_SUCCESS;
1121 12 : }
1122 :
1123 0 : void UbTransportLiteImpl::BatchTransferAll(
1124 : const std::vector<RmaBufferLite>& loc, const std::vector<Buffer>& rmt,
1125 : const std::vector<BaseTransportLiteImpl::TransferOp>& transferOp, const std::vector<uint32_t>& notifyIdxs,
1126 : const StreamLite& stream)
1127 : {
1128 0 : if (UNLIKELY(loc.empty())) {
1129 0 : return;
1130 : }
1131 :
1132 0 : auto taskId = stream.GetRtsq()->GetTaskId();
1133 :
1134 : // 当前使用1个connection，下标为0 (当前只有一个connection，对应一个jetty)
1135 0 : RmaConnLite* conn = connVec[0];
1136 :
1137 : // 展开下发WQE前, 按需设置cache context
1138 0 : UbConnLite* ubConnLitePtr = nullptr;
1139 0 : bool needCacheTask = false;
1140 0 : PreLaunchWqe(ubConnLitePtr, needCacheTask, conn);
1141 : // 下发DbSqe前, 备份相关信息
1142 0 : const uint32_t pendingSqeCnt = needCacheTask ? stream.GetRtsq()->GetPendingSqeCnt() : 0;
1143 :
1144 : // 批量展开下发WQE
1145 0 : u32 insNum = loc.size();
1146 0 : u64 totalSize = 0;
1147 0 : BatchTransferAllWqe_(loc, rmt, transferOp, notifyIdxs, stream, conn, totalSize);
1148 :
1149 0 : const bool isReportTask = IsReportTask();
1150 : // 展开下发WQE后, 展开下发DbSqe前, 按需缓存wqe及DbSqeIdx
1151 : // 注意: pendingSqeCnt在下发DbSqe前已备份
1152 : // 注意: 一定要在展开下发DbSqe前调用PostLaunchWqe, 否则如果下发DbSqe触发LaunchTask,
1153 : // 而尚未调用PostLaunchWqe插入当前WQE数组，
1154 : // 会导致AicpuTaskCache找不到当前WQE数组, 无法正确更新对应的DbSqeLocation
1155 0 : DbSqeProfInfo dbSqeProfInfo;
1156 0 : if (needCacheTask && isReportTask) {
1157 0 : BuildDbSqeProfInfoForExecProfilingAll(
1158 0 : loc[insNum - 1], rmt[insNum - 1], totalSize, transferOp[insNum - 1], notifyIdxs[insNum - 1], dbSqeProfInfo);
1159 : }
1160 0 : PostLaunchWqe(stream, ubConnLitePtr, needCacheTask, pendingSqeCnt, isReportTask, dbSqeProfInfo);
1161 :
1162 0 : BuildUbDbSendTask(stream, conn->GetUbJettyLiteId(), connOut.pi); // 约束使用一批wqe的个数不会导致反压
1163 :
1164 0 : ExecProfilingAll(
1165 0 : loc[insNum - 1], rmt[insNum - 1], totalSize, transferOp[insNum - 1], stream, taskId, notifyIdxs[insNum - 1]);
1166 : }
1167 :
1168 0 : inline void UbTransportLiteImpl::BatchTransferAllWqe_(
1169 : const std::vector<RmaBufferLite>& loc, const std::vector<Buffer>& rmt,
1170 : const std::vector<BaseTransportLiteImpl::TransferOp>& transferOp, const std::vector<uint32_t>& notifyIdxs,
1171 : const StreamLite& stream, RmaConnLite* conn, u64& totalSize)
1172 : {
1173 0 : u64 notifyData = 1; // 普通notify，固定1，用于writeWithNotify与writeReduceWithNotify
1174 0 : SqeConfigLite cfg;
1175 0 : SetFenceConfig(cfg);
1176 0 : u32 insNum = loc.size();
1177 0 : const bool isReportTask = IsReportTask();
1178 :
1179 0 : for (u32 i = 0; i < insNum; i++) {
1180 0 : cfg.cqeEn = (i == insNum - 1) ? true : false; // 返回最后一个sqe的cqe
1181 0 : cfg.placeOdr = (i == insNum - 1) ? UB_STRONG_ORDER : UB_RELAX_ORDER; // 最后一个要求保序
1182 0 : cfg.compOrder = (i == insNum - 1) ? UB_COMPLETION : UB_NO_COMPLETION;
1183 0 : cfg.userConfig = true;
1184 :
1185 0 : if (transferOp[i].transType == TransferType::NOTIFY_RECORD) { // notifyRecord操作没有loc/rmt，因此单独处理
1186 0 : if (notifyIdxs[i] == 1) { // PostFin场景
1187 0 : cfg.cqeEn = true;
1188 0 : cfg.placeOdr = UB_STRONG_ORDER;
1189 0 : cfg.compOrder = UB_COMPLETION;
1190 0 : cfg.userConfig = true;
1191 : }
1192 0 : u32 inlineData = 1;
1193 : // 当前使用1个connection，下标为0 构建sqe
1194 0 : conn->InlineWrite(
1195 0 : reinterpret_cast<u8*>(&inlineData), UB_INLINE_WRITE_SIZE, GetRmtNotifySliceLite(notifyIdxs[i]), cfg,
1196 0 : stream, connOut);
1197 : } else {
1198 0 : auto localBuffer = GetRmaBufSlicelite(loc[i]);
1199 0 : auto remoteBuffer = GetRmtRmaBufSliceLite(rmt[i]);
1200 0 : if (transferOp[i].transType == TransferType::WRITE) {
1201 0 : conn->Write(localBuffer, remoteBuffer, cfg, stream, connOut);
1202 0 : } else if (transferOp[i].transType == TransferType::WRITE_REDUCE) {
1203 0 : conn->WriteReduce(
1204 0 : transferOp[i].reduceIn.dataType, transferOp[i].reduceIn.reduceOp, localBuffer, stream, remoteBuffer,
1205 0 : cfg, connOut);
1206 0 : } else if (transferOp[i].transType == TransferType::READ) {
1207 0 : conn->Read(localBuffer, remoteBuffer, cfg, stream, connOut);
1208 0 : } else if (transferOp[i].transType == TransferType::READ_REDUCE) {
1209 0 : conn->ReadReduce(transferOp[i].reduceIn, localBuffer, remoteBuffer, stream, cfg, connOut);
1210 0 : } else if (transferOp[i].transType == TransferType::WRITE_WITH_NOTIFY) {
1211 0 : conn->WriteWithNotify(
1212 0 : localBuffer, remoteBuffer, cfg, connOut, GetRmtNotifySliceLite(notifyIdxs[i]), stream,
1213 : notifyData); // 当前使用1个connection，下标为0
1214 0 : } else if (transferOp[i].transType == TransferType::WRITE_REDUCE_WITH_NOTIFY) {
1215 0 : conn->WriteReduceWithNotify(
1216 0 : transferOp[i].reduceIn.dataType, transferOp[i].reduceIn.reduceOp, localBuffer, remoteBuffer, cfg,
1217 0 : stream, connOut, GetRmtNotifySliceLite(notifyIdxs[i]),
1218 : notifyData); // 当前使用1个connection，下标为0
1219 : }
1220 : }
1221 0 : if (isReportTask) {
1222 0 : totalSize += GetRmaBufSlicelite(loc[i]).GetSize();
1223 : }
1224 : }
1225 0 : }
1226 :
1227 0 : void UbTransportLiteImpl::Drain(const StreamLite& stream)
1228 : {
1229 0 : std::lock_guard<std::mutex> lock(drainMtx_);
1230 0 : if (drainNotify_.size == 0 || rmtDrainBuffer_.size == 0) {
1231 0 : HCCL_WARNING("[UbTransportLiteImpl::%s] drain resource is null skip", __func__);
1232 0 : return;
1233 : }
1234 :
1235 0 : SqeConfigLite cfg;
1236 0 : Fence();
1237 0 : SetFenceConfig(cfg);
1238 :
1239 : // 当前使用1个connection，下标为0 (当前只有一个connection，对应一个jetty)
1240 0 : RmaConnLite* conn = connVec[0];
1241 :
1242 : // 展开下发WQE前, 按需设置cache context
1243 0 : UbConnLite* ubConnLitePtr = nullptr;
1244 0 : bool needCacheTask = false;
1245 0 : PreLaunchWqe(ubConnLitePtr, needCacheTask, conn);
1246 : // 下发DbSqe前, 备份相关信息
1247 0 : const uint32_t pendingSqeCnt = needCacheTask ? stream.GetRtsq()->GetPendingSqeCnt() : 0;
1248 :
1249 : // 展开下发WQE
1250 0 : auto drainNotifyBufSlice = RmaBufSliceLite(drainNotify_.addr, drainNotify_.size, 0, drainNotify_.tokenId);
1251 : auto drainConstBufSlice = RmtRmaBufSliceLite(
1252 0 : rmtDrainBuffer_.addr, rmtDrainBuffer_.size, 0, rmtDrainBuffer_.tokenId, rmtDrainBuffer_.tokenValue, UINT32_MAX);
1253 0 : conn->Read(drainNotifyBufSlice, drainConstBufSlice, cfg, stream, connOut);
1254 :
1255 0 : const bool isReportTask = false; // Drain操作当前不构造TaskParam
1256 : // 展开下发WQE后, 展开下发DbSqe前, 按需缓存wqe及DbSqeIdx
1257 : // 注意: pendingSqeCnt在下发DbSqe前已备份
1258 : // 注意: 一定要在展开下发DbSqe前调用PostLaunchWqe, 否则如果下发DbSqe触发LaunchTask,
1259 : // 而尚未调用PostLaunchWqe插入当前WQE数组，
1260 : // 会导致AicpuTaskCache找不到当前WQE数组, 无法正确更新对应的DbSqeLocation
1261 0 : DbSqeProfInfo dbSqeProfInfo;
1262 0 : PostLaunchWqe(stream, ubConnLitePtr, needCacheTask, pendingSqeCnt, isReportTask, dbSqeProfInfo);
1263 :
1264 0 : BuildUbDbSendTask(stream, conn->GetUbJettyLiteId(), connOut.pi);
1265 :
1266 0 : BuildNotifyWaitTask(stream, drainNotify_.notifyId);
1267 0 : }
1268 :
1269 1 : void UbTransportLiteImpl::ReportWriteWithNotifyTask(
1270 : const RmaBufSliceLite& locSlice, const RmtRmaBufSliceLite& rmtSlice, const RmtRmaBufSliceLite& rmtNotifySlice,
1271 : const StreamLite& stream, u32 taskId)
1272 : {
1273 1 : if (!IsReportTask()) {
1274 0 : return;
1275 : }
1276 :
1277 1 : TaskParam taskParam{};
1278 1 : taskParam.taskType = TaskParamType::TASK_WRITE_WITH_NOTIFY;
1279 1 : taskParam.beginTime = ProfGetCurCpuTimestamp();
1280 2 : FillTaskParamDmaPub(
1281 1 : taskParam, reinterpret_cast<void*>(rmtSlice.GetAddr()), locSlice.GetSize(), DmaOp::HCCL_DMA_WRITE);
1282 1 : taskParam.taskPara.DMA.src = reinterpret_cast<void*>(locSlice.GetAddr());
1283 1 : taskParam.taskPara.DMA.notifyID = rmtNotifySlice.GetAddr();
1284 1 : taskParam.taskPara.DMA.notifyValue = 1;
1285 :
1286 1 : AddTaskCallback(stream, taskId, taskParam);
1287 1 : DfxTaskInfo* slot = stream.NextTaskSlot();
1288 1 : slot->taskType = TaskParamTypeVal::TASK_WRITE_WITH_NOTIFY;
1289 1 : FillSlotUbDmaInfo(
1290 : slot, stream, taskId, locSlice.GetAddr(), rmtSlice.GetAddr(), locSlice.GetSize(), rmtNotifySlice.GetNotifyId());
1291 1 : }
1292 :
1293 0 : void UbTransportLiteImpl::ReportWriteReduceWithNotifyTask(
1294 : const RmaBufSliceLite& locSlice, const RmtRmaBufSliceLite& rmtSlice, const RmtRmaBufSliceLite& rmtNotifySlice,
1295 : const ReduceIn& reduceIn, const StreamLite& stream, u32 taskId)
1296 : {
1297 0 : if (!IsReportTask()) {
1298 0 : return;
1299 : }
1300 :
1301 0 : TaskParam taskParam{};
1302 0 : taskParam.taskType = TaskParamType::TASK_WRITE_REDUCE_WITH_NOTIFY;
1303 0 : taskParam.beginTime = ProfGetCurCpuTimestamp();
1304 0 : FillTaskParamReducePub(
1305 0 : taskParam, reinterpret_cast<void*>(locSlice.GetAddr()), reinterpret_cast<void*>(rmtSlice.GetAddr()),
1306 : locSlice.GetSize(), reduceIn);
1307 0 : taskParam.taskPara.Reduce.notifyID = rmtNotifySlice.GetAddr();
1308 :
1309 0 : AddTaskCallback(stream, taskId, taskParam);
1310 0 : DfxTaskInfo* slot = stream.NextTaskSlot();
1311 0 : slot->taskType = TaskParamTypeVal::TASK_WRITE_REDUCE_WITH_NOTIFY;
1312 0 : FillSlotReduceInfo(
1313 : slot, stream, taskId, locSlice.GetAddr(), rmtSlice.GetAddr(), locSlice.GetSize(), rmtNotifySlice.GetNotifyId(),
1314 0 : static_cast<u8>(ConvertReduceOpToHcclReduceOp(reduceIn.reduceOp)));
1315 0 : }
1316 :
1317 1 : void UbTransportLiteImpl::WriteWithNotify(
1318 : const RmaBufferLite& loc, const Buffer& rmt, const WithNotifyIn& withNotify, const StreamLite& stream)
1319 : {
1320 1 : SqeConfigLite cfg;
1321 1 : SetFenceConfig(cfg);
1322 1 : u64 notifyData = 1; // 普通notify，固定1
1323 :
1324 1 : auto taskId = stream.GetRtsq()->GetTaskId();
1325 :
1326 : // 当前使用1个connection，下标为0
1327 1 : RmaConnLite* conn = connVec[0];
1328 :
1329 : // 展开下发WQE前, 按需设置cache context
1330 1 : UbConnLite* ubConnLitePtr = nullptr;
1331 1 : bool needCacheTask = false;
1332 1 : PreLaunchWqe(ubConnLitePtr, needCacheTask, conn);
1333 : // 下发DbSqe前, 备份相关信息
1334 1 : const uint32_t pendingSqeCnt = needCacheTask ? stream.GetRtsq()->GetPendingSqeCnt() : 0;
1335 :
1336 : // 展开下发WQE
1337 1 : auto locRmaBufSlicelite = GetRmaBufSlicelite(loc);
1338 1 : auto rmtRmaBufSlicelite = GetRmtRmaBufSliceLite(rmt);
1339 1 : auto rmtNotifySliceLite = GetRmtNotifySliceLite(withNotify.index_);
1340 1 : conn->WriteWithNotify(locRmaBufSlicelite, rmtRmaBufSlicelite, cfg, connOut, rmtNotifySliceLite, stream, notifyData);
1341 :
1342 : // 展开下发WQE后, 展开下发DbSqe前, 按需缓存wqe及DbSqeIdx
1343 : // 注意: pendingSqeCnt在下发DbSqe前已备份
1344 : // 注意: 一定要在展开下发DbSqe前调用PostLaunchWqe, 否则如果下发DbSqe触发LaunchTask,
1345 : // 而尚未调用PostLaunchWqe插入当前WQE数组，
1346 : // 会导致AicpuTaskCache找不到当前WQE数组, 无法正确更新对应的DbSqeLocation
1347 1 : const bool isReportTask = IsReportTask();
1348 1 : DbSqeProfInfo dbSqeProfInfo;
1349 1 : if (needCacheTask && isReportTask) {
1350 0 : BuildDbSqeProfInfoForWriteWithNotify(
1351 0 : reinterpret_cast<void*>(locRmaBufSlicelite.GetAddr()),
1352 0 : reinterpret_cast<void*>(rmtRmaBufSlicelite.GetAddr()), locRmaBufSlicelite.GetSize(),
1353 : rmtNotifySliceLite.GetAddr(), dbSqeProfInfo);
1354 : }
1355 1 : PostLaunchWqe(stream, ubConnLitePtr, needCacheTask, pendingSqeCnt, isReportTask, dbSqeProfInfo);
1356 :
1357 1 : BuildUbDbSendTask(stream, conn->GetUbJettyLiteId(), connOut.pi);
1358 :
1359 1 : ReportWriteWithNotifyTask(locRmaBufSlicelite, rmtRmaBufSlicelite, rmtNotifySliceLite, stream, taskId);
1360 1 : }
1361 :
1362 0 : void UbTransportLiteImpl::WriteReduceWithNotify(
1363 : const RmaBufferLite& loc, const Buffer& rmt, const ReduceIn& reduceIn, const WithNotifyIn& withNotify,
1364 : const StreamLite& stream)
1365 : {
1366 0 : SqeConfigLite cfg;
1367 0 : SetFenceConfig(cfg);
1368 0 : u64 notifyData = 1; // 普通notify，固定1
1369 :
1370 0 : auto taskId = stream.GetRtsq()->GetTaskId();
1371 :
1372 : // 当前使用1个connection，下标为0
1373 0 : RmaConnLite* conn = connVec[0];
1374 :
1375 : // 展开下发WQE前, 按需设置cache context
1376 0 : UbConnLite* ubConnLitePtr = nullptr;
1377 0 : bool needCacheTask = false;
1378 0 : PreLaunchWqe(ubConnLitePtr, needCacheTask, conn);
1379 : // 下发DbSqe前, 备份相关信息
1380 0 : const uint32_t pendingSqeCnt = needCacheTask ? stream.GetRtsq()->GetPendingSqeCnt() : 0;
1381 :
1382 : // 展开下发WQE
1383 0 : auto locRmaBufSlicelite = GetRmaBufSlicelite(loc);
1384 0 : auto rmtRmaBufSlicelite = GetRmtRmaBufSliceLite(rmt);
1385 0 : auto rmtNotifySliceLite = GetRmtNotifySliceLite(withNotify.index_);
1386 0 : conn->WriteReduceWithNotify(
1387 0 : reduceIn.dataType, reduceIn.reduceOp, locRmaBufSlicelite, rmtRmaBufSlicelite, cfg, stream, connOut,
1388 : rmtNotifySliceLite, notifyData);
1389 :
1390 : // 展开下发WQE后, 展开下发DbSqe前, 按需缓存wqe及DbSqeIdx
1391 : // 注意: pendingSqeCnt在下发DbSqe前已备份
1392 : // 注意: 一定要在展开下发DbSqe前调用PostLaunchWqe, 否则如果下发DbSqe触发LaunchTask,
1393 : // 而尚未调用PostLaunchWqe插入当前WQE数组，
1394 : // 会导致AicpuTaskCache找不到当前WQE数组, 无法正确更新对应的DbSqeLocation
1395 0 : const bool isReportTask = IsReportTask();
1396 0 : DbSqeProfInfo dbSqeProfInfo;
1397 0 : if (needCacheTask && isReportTask) {
1398 0 : BuildDbSqeProfInfoForWriteReduceWithNotify(
1399 0 : reinterpret_cast<void*>(locRmaBufSlicelite.GetAddr()),
1400 0 : reinterpret_cast<void*>(rmtRmaBufSlicelite.GetAddr()), locRmaBufSlicelite.GetSize(), reduceIn,
1401 : rmtNotifySliceLite.GetAddr(), dbSqeProfInfo);
1402 : }
1403 0 : PostLaunchWqe(stream, ubConnLitePtr, needCacheTask, pendingSqeCnt, isReportTask, dbSqeProfInfo);
1404 :
1405 0 : BuildUbDbSendTask(stream, conn->GetUbJettyLiteId(), connOut.pi);
1406 :
1407 0 : ReportWriteReduceWithNotifyTask(
1408 : locRmaBufSlicelite, rmtRmaBufSlicelite, rmtNotifySliceLite, reduceIn, stream, taskId);
1409 0 : }
1410 :
1411 1 : void UbTransportLiteImpl::BatchOneSidedRead(
1412 : const vector<RmaBufSliceLite>& loc, const vector<RmtRmaBufSliceLite>& rmt, const StreamLite& stream)
1413 : {
1414 1 : SqeConfigLite cfg;
1415 1 : SetFenceConfig(cfg);
1416 :
1417 : // 当前使用1个connection，下标为0
1418 1 : RmaConnLite* conn = connVec[0];
1419 :
1420 : // 展开下发WQE前, 按需设置cache context
1421 1 : UbConnLite* ubConnLitePtr = nullptr;
1422 1 : bool needCacheTask = false;
1423 1 : PreLaunchWqe(ubConnLitePtr, needCacheTask, conn);
1424 : // 下发DbSqe前, 备份相关信息
1425 1 : const uint32_t pendingSqeCnt = needCacheTask ? stream.GetRtsq()->GetPendingSqeCnt() : 0;
1426 :
1427 : // 展开下发WQE
1428 1 : conn->BatchOneSidedRead(loc, rmt, cfg, stream, connOut);
1429 :
1430 : // 展开下发WQE后, 展开下发DbSqe前, 按需缓存wqe及DbSqeIdx
1431 : // 注意: pendingSqeCnt在下发DbSqe前已备份
1432 : // 注意: 一定要在展开下发DbSqe前调用PostLaunchWqe, 否则如果下发DbSqe触发LaunchTask,
1433 : // 而尚未调用PostLaunchWqe插入当前WQE数组，
1434 : // 会导致AicpuTaskCache找不到当前WQE数组, 无法正确更新对应的DbSqeLocation
1435 1 : PostLaunchWqe(stream, ubConnLitePtr, needCacheTask, pendingSqeCnt, false, DbSqeProfInfo());
1436 :
1437 1 : BuildUbDbSendTask(stream, conn->GetUbJettyLiteId(), connOut.pi);
1438 1 : }
1439 :
1440 1 : void UbTransportLiteImpl::BatchOneSidedWrite(
1441 : const vector<RmaBufSliceLite>& loc, const vector<RmtRmaBufSliceLite>& rmt, const StreamLite& stream)
1442 : {
1443 1 : SqeConfigLite cfg;
1444 1 : SetFenceConfig(cfg);
1445 :
1446 : // 当前使用1个connection，下标为0
1447 1 : RmaConnLite* conn = connVec[0];
1448 :
1449 : // 展开下发WQE前, 按需设置cache context
1450 1 : UbConnLite* ubConnLitePtr = nullptr;
1451 1 : bool needCacheTask = false;
1452 1 : PreLaunchWqe(ubConnLitePtr, needCacheTask, conn);
1453 : // 下发DbSqe前, 备份相关信息
1454 1 : const uint32_t pendingSqeCnt = needCacheTask ? stream.GetRtsq()->GetPendingSqeCnt() : 0;
1455 :
1456 : // 展开下发WQE
1457 1 : conn->BatchOneSidedWrite(loc, rmt, cfg, stream, connOut);
1458 :
1459 : // 展开下发WQE后, 展开下发DbSqe前, 按需缓存wqe及DbSqeIdx
1460 : // 注意: pendingSqeCnt在下发DbSqe前已备份
1461 : // 注意: 一定要在展开下发DbSqe前调用PostLaunchWqe, 否则如果下发DbSqe触发LaunchTask,
1462 : // 而尚未调用PostLaunchWqe插入当前WQE数组，
1463 : // 会导致AicpuTaskCache找不到当前WQE数组, 无法正确更新对应的DbSqeLocation
1464 1 : PostLaunchWqe(stream, ubConnLitePtr, needCacheTask, pendingSqeCnt, false, DbSqeProfInfo());
1465 :
1466 1 : BuildUbDbSendTask(stream, conn->GetUbJettyLiteId(), connOut.pi);
1467 1 : }
1468 :
1469 8 : Eid UbTransportLiteImpl::GetLocEid() const { return connVec[0]->GetLocEid(); }
1470 :
1471 8 : Eid UbTransportLiteImpl::GetRmtEid() const { return connVec[0]->GetRmtEid(); }
1472 :
1473 11 : uint64_t UbTransportLiteImpl::GetJettyHandle() const { return connVec[0]->GetJettyHandle(); }
1474 :
1475 11 : uint32_t UbTransportLiteImpl::GetJettyId() const { return connVec[0]->GetJettyId(); }
1476 :
1477 0 : HcclResult UbTransportLiteImpl::Clean()
1478 : {
1479 0 : locNotifyVec.clear();
1480 0 : rmtNotifyVec.clear();
1481 0 : locBufferMap.clear();
1482 0 : rmtBufferVec.clear();
1483 0 : rmtBufferMap.clear();
1484 :
1485 : // 清理connVec，connLite由UbConnLiteMgr管理
1486 0 : for (auto& it : connUniqueIdVec) {
1487 0 : DECTOR_TRY_CATCH("UbTransportLiteImpl", UbConnLiteMgr::GetInstance().Clear(it));
1488 : }
1489 0 : connUniqueIdVec.clear();
1490 0 : connVec.clear();
1491 :
1492 0 : return HCCL_SUCCESS;
1493 : }
1494 :
1495 0 : HcclResult UbTransportLiteImpl::Resume(std::vector<char>& uniqueId)
1496 : {
1497 0 : Init(uniqueId);
1498 0 : return HCCL_SUCCESS;
1499 : }
1500 :
1501 1 : HcclResult UbTransportLiteImpl::Fence()
1502 : {
1503 1 : fence_ = true;
1504 1 : HCCL_INFO("[%s] SUCCESS. fence[%d]", __func__, fence_);
1505 1 : return HCCL_SUCCESS;
1506 : }
1507 :
1508 8 : void UbTransportLiteImpl::SetFenceConfig(SqeConfigLite& cfg)
1509 : {
1510 8 : if (fence_) {
1511 0 : cfg.fence = UB_FENCE_ENABLED;
1512 0 : cfg.placeOdr = UB_STRONG_ORDER;
1513 0 : cfg.compOrder = UB_COMPLETION;
1514 0 : cfg.userConfig = true;
1515 : }
1516 8 : fence_ = false;
1517 8 : }
1518 :
1519 17 : bool UbTransportLiteImpl::IsReportTask()
1520 : {
1521 17 : return taskExceptionEnable_ || DfxProfilingHandlerLite::GetInstance().GetProfL1State();
1522 : }
1523 : } // namespace Hccl
|
