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 "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 50 : UbTransportLiteImpl::UbTransportLiteImpl(
28 50 : std::vector<char> &uniqueId, std::function<void(u32 streamId, u32 taskId, const TaskParam &taskParam)> callback)
29 : {
30 50 : callback_ = callback;
31 : // [header...][notifyUniqueId...][rmtNotifyUniqueId...][rmtBufferUniqueIds...]
32 50 : BinaryStream binaryStream(uniqueId);
33 : u32 theType;
34 50 : binaryStream >> theType;
35 50 : binaryStream >> notifyNum;
36 50 : binaryStream >> bufferNum;
37 50 : binaryStream >> rmtbufferNum;
38 50 : binaryStream >> connNum;
39 :
40 50 : std::vector<char> notifyUniqueIds;
41 50 : binaryStream >> notifyUniqueIds;
42 50 : ParseLocNotifyVec(notifyUniqueIds);
43 :
44 50 : std::vector<char> rmtNotifyUniqueIds;
45 50 : binaryStream >> rmtNotifyUniqueIds;
46 50 : ParseRmtBufferVec(rmtNotifyUniqueIds, RmaUbBufType::NOTIFY);
47 :
48 50 : std::vector<char> rmtBufferUniqueIds;
49 50 : binaryStream >> rmtBufferUniqueIds;
50 50 : ParseRmtBufferVec(rmtBufferUniqueIds, RmaUbBufType::BUFFER);
51 :
52 50 : std::vector<char> connUniqueIds;
53 50 : binaryStream >> connUniqueIds;
54 50 : ParseConnVec(connUniqueIds);
55 50 : }
56 2 : UbTransportLiteImpl::UbTransportLiteImpl(std::vector<char> &uniqueId)
57 : {
58 2 : Init(uniqueId);
59 2 : }
60 :
61 2 : void UbTransportLiteImpl::Init(std::vector<char> &uniqueId)
62 : {
63 2 : BinaryStream binaryStream(uniqueId);
64 : u32 theType;
65 2 : binaryStream >> theType;
66 2 : binaryStream >> notifyNum;
67 2 : binaryStream >> bufferNum;
68 2 : binaryStream >> rmtbufferNum;
69 2 : binaryStream >> connNum;
70 :
71 2 : std::vector<char> notifyUniqueIds;
72 2 : binaryStream >> notifyUniqueIds;
73 2 : ParseLocNotifyVec(notifyUniqueIds);
74 :
75 2 : std::vector<char> rmtNotifyUniqueIds;
76 2 : binaryStream >> rmtNotifyUniqueIds;
77 2 : ParseRmtBufferVec(rmtNotifyUniqueIds, RmaUbBufType::NOTIFY);
78 :
79 2 : std::vector<char> locBufferUniqueIds;
80 2 : binaryStream >> locBufferUniqueIds;
81 2 : ParseLocBufferMap(locBufferUniqueIds);
82 :
83 2 : std::vector<char> rmtBufferUniqueIds;
84 2 : binaryStream >> rmtBufferUniqueIds;
85 2 : ParseRmtBufferVec(rmtBufferUniqueIds, RmaUbBufType::BUFFER);
86 :
87 : // 解析drain相关的资源信息
88 2 : std::vector<char> drainBufferUniqueIds;
89 2 : binaryStream >> drainBufferUniqueIds;
90 2 : ParseDrainResource(drainBufferUniqueIds);
91 :
92 2 : std::vector<char> connUniqueIds;
93 2 : binaryStream >> connUniqueIds;
94 2 : ParseConnVec(connUniqueIds);
95 2 : }
96 :
97 102 : UbTransportLiteImpl::~UbTransportLiteImpl()
98 : {
99 57 : for (auto &it : connUniqueIdVec) {
100 5 : DECTOR_TRY_CATCH("UbTransportLiteImpl", UbConnLiteMgr::GetInstance().Clear(it));
101 : }
102 102 : }
103 :
104 1 : std::string UbTransportLiteImpl::Describe() const
105 : {
106 1 : std::string desc = "UbTransportLiteImpl[";
107 :
108 1 : u32 idx = 0;
109 1 : desc += "locNotifyVec=[";
110 3 : for (auto &it : locNotifyVec) {
111 2 : desc += StringFormat("idx=%u, %s;", idx, it->Describe().c_str());
112 2 : idx++;
113 : }
114 :
115 1 : idx = 0;
116 1 : desc += "], rmtNotifyVec=[";
117 3 : for (auto &it : rmtNotifyVec) {
118 2 : desc += StringFormat("idx=%u, %s;", idx, it.Describe().c_str());
119 2 : idx++;
120 : }
121 :
122 1 : idx = 0;
123 1 : desc += "], rmtBufferVec=[";
124 3 : for (auto &it : rmtBufferVec) {
125 2 : desc += StringFormat("idx=%u, %s;", idx, it.Describe().c_str());
126 2 : idx++;
127 : }
128 :
129 1 : idx = 0;
130 1 : desc += "], connVec=[";
131 2 : for (auto &it : connVec) {
132 1 : desc += StringFormat("idx=%u, %s;", idx, it->Describe().c_str());
133 1 : idx++;
134 : }
135 :
136 1 : desc += "]]";
137 1 : return desc;
138 0 : }
139 :
140 52 : void UbTransportLiteImpl::ParseLocNotifyVec(std::vector<char> &data)
141 : {
142 52 : if (notifyNum == 0) {
143 139 : HCCL_WARNING("UbTransportLiteImpl::ParseLocNotifyVec num is 0");
144 47 : return;
145 : }
146 5 : u32 notifySizePerDto = data.size() / notifyNum;
147 :
148 15 : for (u32 idx = 0; idx < notifyNum; idx++) {
149 10 : auto start = data.begin() + idx * notifySizePerDto;
150 10 : auto end = start + notifySizePerDto;
151 10 : std::vector<char> dto(start, end);
152 10 : locNotifyVec.push_back(std::make_unique<NotifyLite>(dto));
153 26 : HCCL_INFO("locNotify idx=%u, %s", idx, locNotifyVec.back()->Describe().c_str());
154 10 : }
155 : }
156 :
157 104 : void UbTransportLiteImpl::ParseRmtBufferVec(std::vector<char> &data, RmaUbBufType rmtType)
158 : {
159 104 : u32 num = 0;
160 104 : if (rmtType == RmaUbBufType::NOTIFY) {
161 52 : num = notifyNum;
162 : } else {
163 52 : num = rmtbufferNum;
164 : }
165 :
166 104 : if (num == 0) {
167 278 : HCCL_WARNING("UbTransportLiteImpl::ParseRmtBufferVec %s num is 0", rmtType.Describe().c_str());
168 94 : return;
169 : }
170 :
171 10 : u32 rmtBufferSizePerDto = data.size() / num;
172 26 : HCCL_INFO("Parse %s num=%u, sizePerDto=%u", rmtType.Describe().c_str(), num, rmtBufferSizePerDto);
173 10 : BinaryStream binaryStream(data);
174 :
175 33 : for (u32 idx = 0; idx < num; idx++) {
176 : RmtUbBufLite ubBufLite;
177 23 : binaryStream >> ubBufLite.addr;
178 23 : binaryStream >> ubBufLite.size;
179 23 : binaryStream >> ubBufLite.tokenId;
180 23 : binaryStream >> ubBufLite.tokenValue;
181 23 : binaryStream >> ubBufLite.notifyId;
182 55 : HCCL_INFO("idx=%u, %s %s", idx, rmtType.Describe().c_str(), ubBufLite.Describe().c_str());
183 23 : if (rmtType == RmaUbBufType::NOTIFY) {
184 10 : rmtNotifyVec.push_back(ubBufLite);
185 : } else {
186 13 : rmtBufferMap[static_cast<uintptr_t>(ubBufLite.addr)] = ubBufLite;
187 13 : rmtBufferVec.push_back(ubBufLite);
188 : }
189 : }
190 10 : }
191 :
192 2 : void UbTransportLiteImpl::ParseLocBufferMap(std::vector<char> &data)
193 : {
194 2 : u32 num = bufferNum;
195 :
196 2 : if (num == 0) {
197 1 : HCCL_WARNING("UbTransportLiteImpl::ParseLocBufferMap num is 0");
198 1 : return;
199 : }
200 :
201 1 : u32 rmtBufferSizePerDto = data.size() / num;
202 1 : HCCL_INFO("ParseLocBufferMap num=%u, sizePerDto=%u", num, rmtBufferSizePerDto);
203 1 : BinaryStream binaryStream(data);
204 :
205 4 : for (u32 idx = 0; idx < num; idx++) {
206 : LocUbBufLite ubBufLite;
207 3 : binaryStream >> ubBufLite.addr;
208 3 : binaryStream >> ubBufLite.size;
209 3 : binaryStream >> ubBufLite.tokenId;
210 3 : binaryStream >> ubBufLite.tokenValue;
211 3 : HCCL_INFO("idx=%u, LocBuffer %s", idx, ubBufLite.Describe().c_str());
212 3 : locBufferMap[static_cast<uintptr_t>(ubBufLite.addr)] = ubBufLite;
213 : }
214 1 : }
215 :
216 2 : void UbTransportLiteImpl::ParseDrainResource(std::vector<char> &data)
217 : {
218 2 : if (data.size() == 0) {
219 2 : HCCL_WARNING("UbTransportLiteImpl::ParseDrainResource is null");
220 2 : return;
221 : }
222 :
223 0 : BinaryStream binaryStream(data);
224 0 : binaryStream >> drainNotify_.addr;
225 0 : binaryStream >> drainNotify_.size;
226 0 : binaryStream >> drainNotify_.tokenId;
227 0 : binaryStream >> drainNotify_.tokenValue;
228 0 : binaryStream >> drainNotify_.notifyId;
229 0 : HCCL_INFO("drain notify %s", drainNotify_.Describe().c_str());
230 :
231 0 : binaryStream >> rmtDrainBuffer_.addr;
232 0 : binaryStream >> rmtDrainBuffer_.size;
233 0 : binaryStream >> rmtDrainBuffer_.tokenId;
234 0 : binaryStream >> rmtDrainBuffer_.tokenValue;
235 0 : binaryStream >> rmtDrainBuffer_.notifyId;
236 0 : HCCL_INFO("drain remote buffer %s", rmtDrainBuffer_.Describe().c_str());
237 0 : }
238 :
239 52 : void UbTransportLiteImpl::ParseConnVec(std::vector<char> &data)
240 : {
241 52 : if (connNum == 0) {
242 139 : HCCL_WARNING("UbTransportLiteImpl::ParseConnVec num is 0");
243 47 : return;
244 : }
245 5 : u32 connSizePerDto = data.size() / connNum;
246 13 : HCCL_INFO("Parse ConnVec num=%u, connSizePerDto=%u", connNum, connSizePerDto);
247 10 : for (u32 idx = 0; idx < connNum; idx++) {
248 5 : auto start = data.begin() + idx * connSizePerDto;
249 5 : auto end = start + connSizePerDto;
250 5 : std::vector<char> connUniqueId(start, end);
251 5 : connUniqueIdVec.push_back(connUniqueId);
252 : // connLite的复用由 ubConnLiteMgr管理
253 5 : auto lite = UbConnLiteMgr::GetInstance().Get(connUniqueId);
254 5 : connVec.push_back(lite);
255 13 : HCCL_INFO("[%s]idx=%u, %s", __func__, idx, lite->Describe().c_str());
256 5 : }
257 10 : CheckConnVec("after ParseConnVec");
258 : }
259 :
260 0 : void UbTransportLiteImpl::BuildUbDbSendTask(const StreamLite &stream, const UbJettyLiteId &jettyLiteId, u32 pi)
261 : {
262 0 : stream.GetRtsq()->UbDbSend(jettyLiteId, pi);
263 0 : }
264 :
265 0 : void UbTransportLiteImpl::BuildNotifyWaitTask(const StreamLite &stream, u32 notifyId)
266 : {
267 0 : stream.GetRtsq()->NotifyWait(notifyId);
268 0 : }
269 :
270 1 : Buffer UbTransportLiteImpl::GetRmtBuffer(u32 index)
271 : {
272 1 : if (UNLIKELY(index >= rmtBufferVec.size())) {
273 0 : THROW<InternalException>(StringFormat("UbTransportLiteImpl::GetRmtBuffer out-of-bounds. index=%u, size=%u",
274 : index, rmtBufferVec.size()));
275 : }
276 1 : return Buffer(rmtBufferVec[index].addr, rmtBufferVec[index].size);
277 : }
278 :
279 2 : RmtRmaBufSliceLite UbTransportLiteImpl::GetRmtNotifySliceLite(u32 index)
280 : {
281 2 : RmtUbBufLite &lite = rmtNotifyVec[index];
282 : // ub conn lite 不关心rkey , rkey 设定为0
283 2 : return RmtRmaBufSliceLite(lite.addr, lite.size, 0, lite.tokenId, lite.tokenValue, lite.notifyId);
284 : }
285 :
286 7 : RmtRmaBufSliceLite UbTransportLiteImpl::GetRmtRmaBufSliceLite(const Buffer &rmtBuf)
287 : {
288 7 : auto it = rmtBufferMap.upper_bound(rmtBuf.GetAddr());
289 :
290 7 : while(it != rmtBufferMap.begin()) {
291 7 : --it;
292 7 : Buffer iterBuf(it->second.addr, it->second.size);
293 7 : if (iterBuf.Contains(rmtBuf.GetAddr(), rmtBuf.GetSize())) {
294 14 : return RmtRmaBufSliceLite(rmtBuf.GetAddr(), rmtBuf.GetSize(), 0, it->second.tokenId, it->second.tokenValue,
295 21 : UINT32_MAX);
296 : }
297 7 : }
298 0 : MACRO_THROW(InternalException, StringFormat("%s is not in current transport", rmtBuf.Describe().c_str()));
299 : }
300 :
301 0 : RmtRmaBufSliceLite UbTransportLiteImpl::GetRmtRmaBufSliceLite(const RmaBufferLite &lite) const
302 : {
303 0 : return RmtRmaBufSliceLite(lite.GetAddr(), lite.GetSize(), 0, lite.GetTokenId() , lite.GetTokenValue(), UINT32_MAX);
304 : }
305 :
306 0 : HcclResult UbTransportLiteImpl::BuildLocRmaBufferLite(const uintptr_t addr, const size_t size, RmaBufferLite &rmaBufferLite)
307 : {
308 0 : HCCL_INFO("[UbTransportLiteImpl::%s] start to find addr[0x%llx], size[0x%llx] in locBufferMap, whose size is %zu. ",
309 : __func__, addr, size, locBufferMap.size());
310 0 : if (locBufferMap.empty()) {
311 0 : HCCL_ERROR("[UbTransportLiteImpl::%s] locBufferMap is empty.", __func__);
312 0 : return HCCL_E_INTERNAL;
313 : }
314 :
315 0 : bool isAddrInRange = false;
316 0 : auto it = locBufferMap.upper_bound(addr);
317 :
318 0 : while(it != locBufferMap.begin()) {
319 0 : --it;
320 0 : Buffer iterBuf(it->second.addr, it->second.size);
321 0 : if (iterBuf.Contains(addr, size)) {
322 0 : rmaBufferLite = RmaBufferLite(addr, size, it->second.tokenId, it->second.tokenValue);
323 0 : isAddrInRange = true;
324 0 : break;
325 : }
326 0 : }
327 :
328 0 : if (!isAddrInRange) {
329 0 : HCCL_WARNING("[UbTransportLiteImpl::%s] addr[0x%llx], size[0x%llx] not in any range of locBufferMap, use the first in map addr[0x%llx] size[0x%llx]",
330 : __func__, addr, size, it->second.addr, it->second.size);
331 0 : rmaBufferLite = RmaBufferLite(addr, size, it->second.tokenId, it->second.tokenValue);
332 : }
333 :
334 0 : return HCCL_SUCCESS;
335 : }
336 :
337 0 : void UbTransportLiteImpl::ClearConnOut()
338 : {
339 0 : wqeData.clear();
340 0 : wqeData.resize(UB_WQE_MAX_SIZE);
341 0 : connOut.data = (u8 *)wqeData.data();
342 0 : connOut.dataSize = sizeof(wqeData);
343 0 : }
344 :
345 : // 检查connection不能为空
346 5 : void UbTransportLiteImpl::CheckConnVec(const std::string &desc)
347 : {
348 5 : if (UNLIKELY(connVec.size() == 0)) {
349 0 : THROW<InternalException>(StringFormat("connVec size is 0 %s", desc.c_str()));
350 : }
351 :
352 5 : u32 idx = 0;
353 10 : for (auto &it : connVec) {
354 5 : if (UNLIKELY(it == nullptr)) {
355 0 : THROW<InternalException>(StringFormat("connVec[%u] is null %s", idx, desc.c_str()));
356 : }
357 5 : idx++;
358 : }
359 5 : }
360 :
361 8 : RmaBufSliceLite UbTransportLiteImpl::GetRmaBufSlicelite(const RmaBufferLite &lite) const
362 : {
363 : // ub conn lite 不关心rkey , rkey 设定为0
364 8 : return RmaBufSliceLite(lite.GetAddr(), lite.GetSize(), 0, lite.GetTokenId());
365 : }
366 :
367 1 : void UbTransportLiteImpl::Post(u32 index, const StreamLite &stream)
368 : {
369 1 : SqeConfigLite cfg;
370 1 : if (index == 1) { // PostFin场景
371 0 : cfg.cqeEn = true;
372 0 : cfg.placeOdr = UB_STRONG_ORDER;
373 0 : cfg.compOrder = UB_COMPLETION;
374 0 : cfg.userConfig = true;
375 : }
376 1 : u32 inlineData = 1;
377 :
378 1 : auto taskId = stream.GetRtsq()->GetTaskId();
379 : // 当前使用1个connection，下标为0 构建sqe
380 1 : auto rmtBuffSliceLite = GetRmtNotifySliceLite(index);
381 1 : connVec[0]->InlineWrite(reinterpret_cast<u8 *>(&inlineData), UB_INLINE_WRITE_SIZE, rmtBuffSliceLite,
382 1 : cfg, stream, connOut);
383 : // 构建rts 的 sqe
384 1 : BuildUbDbSendTask(stream, connVec[0]->GetUbJettyLiteId(), connOut.pi);
385 :
386 3 : HCCL_INFO("UbTransportLiteImpl::Post notifyId[0x%llx], pi=%u", rmtBuffSliceLite.GetAddr(), connOut.pi);
387 :
388 1 : if (!IsReportTask()) {
389 0 : return;
390 : }
391 :
392 1 : TaskParam taskParam{};
393 1 : taskParam.taskType = TaskParamType::TASK_UB_INLINE_WRITE;
394 1 : taskParam.beginTime = ProfGetCurCpuTimestamp();
395 1 : taskParam.taskPara.DMA.dst = reinterpret_cast<void*>(rmtBuffSliceLite.GetAddr());
396 1 : taskParam.taskPara.DMA.size = rmtBuffSliceLite.GetSize();
397 1 : taskParam.taskPara.DMA.notifyID = rmtBuffSliceLite.GetNotifyId();
398 1 : taskParam.taskPara.DMA.notifyValue = 1;
399 1 : taskParam.taskPara.DMA.linkType = DfxLinkType::UB;
400 1 : taskParam.taskPara.DMA.dmaOp = DmaOp::HCCL_DMA_WRITE;
401 1 : taskParam.taskPara.DMA.locEid = GetLocEid();
402 1 : taskParam.taskPara.DMA.rmtEid = GetRmtEid();
403 1 : taskParam.taskPara.DMA.jettyHandle = GetJettyHandle();
404 1 : taskParam.taskPara.DMA.jettyId = GetJettyId();
405 :
406 3 : HCCL_INFO("[UbTransportLiteImpl::%s] locEid[%s], rmtEid[%s]", __func__, GetLocEid().Describe().c_str(), GetRmtEid().Describe().c_str());
407 :
408 1 : AddTaskCallback(stream, taskId, taskParam);
409 1 : }
410 :
411 1 : void UbTransportLiteImpl::Wait(u32 index, const StreamLite &stream)
412 : {
413 1 : WaitWithTimeout(index, stream, CommunicatorImplLiteMgr::GetInstance().GetEnvConfig().hcclExecTimeout);
414 1 : }
415 :
416 4 : void UbTransportLiteImpl::WaitWithTimeout(u32 index, const StreamLite &stream, u32 timeout)
417 : {
418 4 : auto taskId = stream.GetRtsq()->GetTaskId();
419 4 : auto notifyId = locNotifyVec[index]->GetId();
420 4 : stream.GetRtsq()->NotifyWait(notifyId, timeout);
421 :
422 4 : if (!IsReportTask()) {
423 0 : return;
424 : }
425 :
426 4 : TaskParam taskParam{};
427 4 : taskParam.taskType = TaskParamType::TASK_NOTIFY_WAIT;
428 4 : taskParam.beginTime = ProfGetCurCpuTimestamp();
429 4 : taskParam.taskPara.Notify.notifyID = notifyId;
430 4 : taskParam.taskPara.Notify.value = 1;
431 4 : taskParam.taskPara.DMA.jettyHandle = GetJettyHandle();
432 4 : taskParam.taskPara.DMA.jettyId = GetJettyId();
433 :
434 4 : AddTaskCallback(stream, taskId, taskParam);
435 4 : }
436 :
437 3 : void UbTransportLiteImpl::ProfilingProcess(void *src, void *dst, u64 size, const StreamLite &stream,
438 : DmaOp dmaOp, u32 taskId)
439 : {
440 3 : if (!IsReportTask()) {
441 0 : return;
442 : }
443 :
444 3 : TaskParam taskParam{};
445 3 : taskParam.taskType = TaskParamType::TASK_UB;
446 3 : taskParam.beginTime = ProfGetCurCpuTimestamp();
447 3 : FillTaskParamDmaPub(taskParam, dst, size, dmaOp);
448 3 : taskParam.taskPara.DMA.src = src;
449 3 : taskParam.taskPara.DMA.jettyHandle = GetJettyHandle();
450 3 : taskParam.taskPara.DMA.jettyId = GetJettyId();
451 :
452 3 : AddTaskCallback(stream, taskId, taskParam);
453 3 : }
454 :
455 2 : void UbTransportLiteImpl::ReduceProfilingProcess(void *src, void *dst, u64 size,
456 : const ReduceIn &reduceIn, const StreamLite &stream, u32 taskId)
457 : {
458 2 : if (!IsReportTask()) {
459 0 : return;
460 : }
461 :
462 2 : TaskParam taskParam {};
463 2 : taskParam.taskType = TaskParamType::TASK_UB_REDUCE_INLINE;
464 2 : taskParam.beginTime = ProfGetCurCpuTimestamp();
465 2 : taskParam.taskPara.Reduce.src = src;
466 2 : taskParam.taskPara.Reduce.dst = dst;
467 2 : taskParam.taskPara.Reduce.size = size;
468 2 : taskParam.taskPara.Reduce.notifyID = INVALID_VALUE_NOTIFYID;
469 2 : taskParam.taskPara.Reduce.notifyValue = 1;
470 2 : taskParam.taskPara.Reduce.linkType = DfxLinkType::UB;
471 2 : taskParam.taskPara.Reduce.reduceOp = ConvertReduceOpToHcclReduceOp(reduceIn.reduceOp);
472 2 : taskParam.taskPara.Reduce.dataType = DataTypeToHcclDataType(reduceIn.dataType);
473 2 : taskParam.taskPara.Reduce.locEid = GetLocEid();
474 2 : taskParam.taskPara.Reduce.rmtEid = GetRmtEid();
475 2 : taskParam.taskPara.DMA.jettyHandle = GetJettyHandle();
476 2 : taskParam.taskPara.DMA.jettyId = GetJettyId();
477 :
478 2 : AddTaskCallback(stream, taskId, taskParam);
479 2 : }
480 :
481 0 : void UbTransportLiteImpl::WriteWithNotifyProfilingProcess(void *src, void *dst, u64 size, const StreamLite &stream,
482 : u32 taskId, u64 notifyId)
483 : {
484 0 : if (!IsReportTask()) {
485 0 : return;
486 : }
487 :
488 0 : TaskParam taskParam{};
489 0 : taskParam.taskType = TaskParamType::TASK_WRITE_WITH_NOTIFY;
490 0 : taskParam.beginTime = ProfGetCurCpuTimestamp();
491 0 : FillTaskParamDmaPub(taskParam, dst, size, DmaOp::HCCL_DMA_WRITE);
492 0 : taskParam.taskPara.DMA.src = src;
493 0 : taskParam.taskPara.DMA.notifyID = notifyId;
494 0 : taskParam.taskPara.DMA.notifyValue = 1;
495 0 : taskParam.taskPara.DMA.jettyHandle = GetJettyHandle();
496 0 : taskParam.taskPara.DMA.jettyId = GetJettyId();
497 :
498 0 : AddTaskCallback(stream, taskId, taskParam);
499 0 : }
500 :
501 0 : void UbTransportLiteImpl::WriteReduceWithNotifyProfilingProcess(void *src, void *dst, u64 size,
502 : const ReduceIn &reduceIn, const StreamLite &stream, u32 taskId, u64 notifyId)
503 : {
504 0 : if (!IsReportTask()) {
505 0 : return;
506 : }
507 :
508 0 : TaskParam taskParam {};
509 0 : taskParam.taskType = TaskParamType::TASK_WRITE_REDUCE_WITH_NOTIFY;
510 0 : taskParam.beginTime = ProfGetCurCpuTimestamp();
511 0 : taskParam.taskPara.Reduce.src = src;
512 0 : taskParam.taskPara.Reduce.dst = dst;
513 0 : taskParam.taskPara.Reduce.size = size;
514 0 : taskParam.taskPara.Reduce.notifyID = notifyId;
515 0 : taskParam.taskPara.Reduce.notifyValue = 1;
516 0 : taskParam.taskPara.Reduce.linkType = DfxLinkType::UB;
517 0 : taskParam.taskPara.Reduce.reduceOp = ConvertReduceOpToHcclReduceOp(reduceIn.reduceOp);
518 0 : taskParam.taskPara.Reduce.dataType = DataTypeToHcclDataType(reduceIn.dataType);
519 0 : taskParam.taskPara.Reduce.locEid = GetLocEid();
520 0 : taskParam.taskPara.Reduce.rmtEid = GetRmtEid();
521 0 : taskParam.taskPara.DMA.jettyHandle = GetJettyHandle();
522 0 : taskParam.taskPara.DMA.jettyId = GetJettyId();
523 :
524 0 : AddTaskCallback(stream, taskId, taskParam);
525 0 : }
526 :
527 0 : void UbTransportLiteImpl::NotifyRecordProfilingProcess(void *dst, u64 size,
528 : const StreamLite &stream, u32 taskId, u64 notifyId)
529 : {
530 0 : if (!IsReportTask()) {
531 0 : return;
532 : }
533 :
534 0 : TaskParam taskParam {};
535 0 : taskParam.taskType = TaskParamType::TASK_UB_INLINE_WRITE;
536 0 : taskParam.beginTime = ProfGetCurCpuTimestamp();
537 0 : FillTaskParamDmaPub(taskParam, dst, size, DmaOp::HCCL_DMA_WRITE);
538 0 : taskParam.taskPara.DMA.notifyID = notifyId;
539 0 : taskParam.taskPara.DMA.notifyValue = 1;
540 0 : taskParam.taskPara.DMA.jettyHandle = GetJettyHandle();
541 0 : taskParam.taskPara.DMA.jettyId = GetJettyId();
542 :
543 0 : AddTaskCallback(stream, taskId, taskParam);
544 0 : }
545 :
546 1 : void UbTransportLiteImpl::Read(const RmaBufferLite &loc, const Buffer &rmt, const StreamLite &stream)
547 : {
548 1 : SqeConfigLite cfg;
549 1 : SetFenceConfig(cfg);
550 1 : auto taskId = stream.GetRtsq()->GetTaskId();
551 :
552 : // 当前使用1个connection,下标为0
553 1 : auto locRmaBufSlicelite = GetRmaBufSlicelite(loc);
554 1 : auto rmtRmaBufSlicelite = GetRmtRmaBufSliceLite(rmt);
555 1 : connVec[0]->Read(locRmaBufSlicelite, rmtRmaBufSlicelite, cfg, stream, connOut);
556 1 : BuildUbDbSendTask(stream, connVec[0]->GetUbJettyLiteId(), connOut.pi);
557 :
558 2 : ProfilingProcess(reinterpret_cast<void *>(locRmaBufSlicelite.GetAddr()),
559 1 : reinterpret_cast<void *>(rmtRmaBufSlicelite.GetAddr()),
560 : locRmaBufSlicelite.GetSize(), stream, DmaOp::HCCL_DMA_READ, taskId);
561 1 : }
562 :
563 1 : void UbTransportLiteImpl::Write(const RmaBufferLite &loc, const Buffer &rmt, const StreamLite &stream)
564 : {
565 1 : SqeConfigLite cfg;
566 1 : SetFenceConfig(cfg);
567 1 : auto taskId = stream.GetRtsq()->GetTaskId();
568 :
569 : // 当前使用1个connection，下标为0
570 1 : auto locRmaBufSlicelite = GetRmaBufSlicelite(loc);
571 1 : auto rmtRmaBufSlicelite = GetRmtRmaBufSliceLite(rmt);
572 1 : connVec[0]->Write(locRmaBufSlicelite, rmtRmaBufSlicelite, cfg, stream, connOut);
573 1 : BuildUbDbSendTask(stream, connVec[0]->GetUbJettyLiteId(), connOut.pi);
574 :
575 2 : ProfilingProcess(reinterpret_cast<void *>(locRmaBufSlicelite.GetAddr()),
576 1 : reinterpret_cast<void *>(rmtRmaBufSlicelite.GetAddr()),
577 : locRmaBufSlicelite.GetSize(), stream, DmaOp::HCCL_DMA_WRITE, taskId);
578 1 : }
579 :
580 1 : void UbTransportLiteImpl::ReadReduce(const RmaBufferLite &loc, const Buffer &rmt, const ReduceIn &reduceIn,
581 : const StreamLite &stream)
582 : {
583 1 : SqeConfigLite cfg;
584 1 : SetFenceConfig(cfg);
585 1 : auto taskId = stream.GetRtsq()->GetTaskId();
586 :
587 : // 当前使用1个connection，下标为0
588 1 : auto locRmaBufSlicelite = GetRmaBufSlicelite(loc);
589 1 : auto rmtRmaBufSlicelite = GetRmtRmaBufSliceLite(rmt);
590 1 : connVec[0]->ReadReduce(reduceIn, locRmaBufSlicelite, rmtRmaBufSlicelite, stream, cfg, connOut);
591 1 : BuildUbDbSendTask(stream, connVec[0]->GetUbJettyLiteId(), connOut.pi);
592 :
593 2 : ReduceProfilingProcess(reinterpret_cast<void *>(locRmaBufSlicelite.GetAddr()),
594 1 : reinterpret_cast<void *>(rmtRmaBufSlicelite.GetAddr()),
595 : locRmaBufSlicelite.GetSize(), reduceIn, stream, taskId);
596 1 : }
597 :
598 1 : void UbTransportLiteImpl::WriteReduce(const RmaBufferLite &loc, const Buffer &rmt, const ReduceIn &reduceIn,
599 : const StreamLite &stream)
600 : {
601 1 : SqeConfigLite cfg;
602 1 : SetFenceConfig(cfg);
603 1 : auto taskId = stream.GetRtsq()->GetTaskId();
604 :
605 : // 当前使用1个connection，下标为0
606 1 : auto locRmaBufSlicelite = GetRmaBufSlicelite(loc);
607 1 : auto rmtRmaBufSlicelite = GetRmtRmaBufSliceLite(rmt);
608 1 : connVec[0]->WriteReduce(reduceIn.dataType, reduceIn.reduceOp, locRmaBufSlicelite, stream,
609 1 : rmtRmaBufSlicelite, cfg, connOut);
610 1 : BuildUbDbSendTask(stream, connVec[0]->GetUbJettyLiteId(), connOut.pi);
611 :
612 2 : ReduceProfilingProcess(reinterpret_cast<void *>(locRmaBufSlicelite.GetAddr()),
613 1 : reinterpret_cast<void *>(rmtRmaBufSlicelite.GetAddr()),
614 : locRmaBufSlicelite.GetSize(), reduceIn, stream, taskId);
615 1 : }
616 :
617 1 : void UbTransportLiteImpl::ExecProfiling(const std::vector<RmaBufferLite> &loc, const std::vector<Buffer> &rmt,
618 : const std::vector<BaseTransportLiteImpl::TransferOp> &transferOp, const StreamLite &stream, u32 taskId)
619 : {
620 1 : u32 insNum = loc.size();
621 1 : u64 totalSize = 0;
622 2 : for (u32 i = 0; i < insNum; i++) {
623 1 : totalSize += GetRmaBufSlicelite(loc[i]).GetSize();
624 : }
625 1 : if (transferOp[insNum - 1].reduceIn.reduceOp == ReduceOp::INVALID) {
626 1 : DmaOp dmaOp = DmaOp::HCCL_DMA_WRITE;
627 1 : if (transferOp[insNum - 1].transType == TransferType::READ) {
628 1 : dmaOp = DmaOp::HCCL_DMA_READ;
629 : }
630 1 : ProfilingProcess(reinterpret_cast<void *>(GetRmaBufSlicelite(loc[insNum - 1]).GetAddr()),
631 2 : reinterpret_cast<void *>(GetRmtRmaBufSliceLite(rmt[insNum - 1]).GetAddr()),
632 : totalSize, stream, dmaOp, taskId);
633 : } else {
634 0 : ReduceProfilingProcess(reinterpret_cast<void *>(GetRmaBufSlicelite(loc[insNum - 1]).GetAddr()),
635 0 : reinterpret_cast<void *>(GetRmtRmaBufSliceLite(rmt[insNum - 1]).GetAddr()),
636 0 : totalSize, transferOp[insNum - 1].reduceIn, stream, taskId);
637 : }
638 1 : }
639 :
640 0 : void UbTransportLiteImpl::ExecProfilingAll(const std::vector<RmaBufferLite> &loc, const std::vector<Buffer> &rmt,
641 : const std::vector<BaseTransportLiteImpl::TransferOp> &transferOp, const StreamLite &stream, u32 taskId,
642 : const std::vector<uint32_t> ¬ifyIdxs)
643 : {
644 0 : u32 insNum = loc.size();
645 0 : u64 totalSize = 0;
646 0 : for (u32 i = 0; i < insNum; i++) {
647 0 : totalSize += GetRmaBufSlicelite(loc[i]).GetSize();
648 : }
649 :
650 0 : if (transferOp[insNum - 1].transType == TransferType::READ) {
651 0 : ProfilingProcess(reinterpret_cast<void *>(GetRmaBufSlicelite(loc[insNum - 1]).GetAddr()),
652 0 : reinterpret_cast<void *>(GetRmtRmaBufSliceLite(rmt[insNum - 1]).GetAddr()),
653 : totalSize, stream, DmaOp::HCCL_DMA_READ, taskId);
654 0 : } else if (transferOp[insNum - 1].transType == TransferType::WRITE) {
655 0 : ProfilingProcess(reinterpret_cast<void *>(GetRmaBufSlicelite(loc[insNum - 1]).GetAddr()),
656 0 : reinterpret_cast<void *>(GetRmtRmaBufSliceLite(rmt[insNum - 1]).GetAddr()),
657 : totalSize, stream, DmaOp::HCCL_DMA_WRITE, taskId);
658 0 : } else if (transferOp[insNum - 1].transType == TransferType::READ_REDUCE) {
659 0 : ReduceProfilingProcess(reinterpret_cast<void *>(GetRmaBufSlicelite(loc[insNum - 1]).GetAddr()),
660 0 : reinterpret_cast<void *>(GetRmtRmaBufSliceLite(rmt[insNum - 1]).GetAddr()),
661 0 : totalSize, transferOp[insNum - 1].reduceIn, stream, taskId);
662 0 : } else if (transferOp[insNum - 1].transType == TransferType::WRITE_REDUCE) {
663 0 : ReduceProfilingProcess(reinterpret_cast<void *>(GetRmaBufSlicelite(loc[insNum - 1]).GetAddr()),
664 0 : reinterpret_cast<void *>(GetRmtRmaBufSliceLite(rmt[insNum - 1]).GetAddr()),
665 0 : totalSize, transferOp[insNum - 1].reduceIn, stream, taskId);
666 0 : } else if (transferOp[insNum - 1].transType == TransferType::WRITE_WITH_NOTIFY) {
667 0 : WriteWithNotifyProfilingProcess(reinterpret_cast<void *>(GetRmaBufSlicelite(loc[insNum - 1]).GetAddr()),
668 0 : reinterpret_cast<void *>(GetRmtRmaBufSliceLite(rmt[insNum - 1]).GetAddr()),
669 0 : totalSize, stream, taskId, GetRmtNotifySliceLite(notifyIdxs[insNum - 1]).GetAddr());
670 0 : } else if (transferOp[insNum - 1].transType == TransferType::WRITE_REDUCE_WITH_NOTIFY) {
671 0 : WriteReduceWithNotifyProfilingProcess(reinterpret_cast<void *>(GetRmaBufSlicelite(loc[insNum - 1]).GetAddr()),
672 0 : reinterpret_cast<void *>(GetRmtRmaBufSliceLite(rmt[insNum - 1]).GetAddr()),
673 0 : totalSize, transferOp[insNum - 1].reduceIn, stream, taskId, GetRmtNotifySliceLite(notifyIdxs[insNum - 1]).GetAddr());
674 0 : } else if (transferOp[insNum - 1].transType == TransferType::NOTIFY_RECORD) {
675 0 : NotifyRecordProfilingProcess(reinterpret_cast<void *>(GetRmtNotifySliceLite(notifyIdxs[insNum - 1]).GetAddr()),
676 0 : GetRmtNotifySliceLite(notifyIdxs[insNum - 1]).GetSize(), stream, taskId, GetRmtNotifySliceLite(notifyIdxs[insNum - 1]).GetAddr());
677 : }
678 0 : }
679 :
680 1 : void UbTransportLiteImpl::BatchTransfer(const std::vector<RmaBufferLite> &loc, const std::vector<Buffer> &rmt,
681 : const std::vector<BaseTransportLiteImpl::TransferOp> &transferOp, const StreamLite &stream)
682 : {
683 1 : if (UNLIKELY(loc.empty())) {
684 0 : return;
685 : }
686 1 : SqeConfigLite cfg;
687 1 : SetFenceConfig(cfg);
688 1 : auto taskId = stream.GetRtsq()->GetTaskId();
689 1 : u32 insNum = loc.size();
690 2 : for (u32 i = 0; i < insNum; i++) {
691 1 : cfg.cqeEn = (i == insNum - 1) ? true : false; // 返回最后一个sqe的cqe
692 1 : cfg.placeOdr = UB_RELAX_ORDER;
693 1 : cfg.compOrder = UB_NO_COMPLETION;
694 1 : cfg.userConfig = true;
695 :
696 1 : auto localBuffer = GetRmaBufSlicelite(loc[i]);
697 1 : auto remoteBuffer = GetRmtRmaBufSliceLite(rmt[i]);
698 1 : if (transferOp[i].transType == TransferType::WRITE) {
699 0 : connVec[0]->Write(localBuffer, remoteBuffer, cfg, stream, connOut); // 当前只有一个connection，对应一个jetty
700 1 : } else if (transferOp[i].transType == TransferType::WRITE_REDUCE) { // write reduce
701 0 : connVec[0]->WriteReduce(transferOp[i].reduceIn.dataType, transferOp[i].reduceIn.reduceOp, localBuffer,
702 0 : stream, remoteBuffer, cfg, connOut);
703 1 : } else if (transferOp[i].transType == TransferType::READ) {
704 1 : connVec[0]->Read(localBuffer, remoteBuffer, cfg, stream, connOut); // 当前只有一个connection，对应一个jetty
705 0 : } else if (transferOp[i].transType == TransferType::READ_REDUCE) { // read reduce
706 0 : connVec[0]->ReadReduce(transferOp[i].reduceIn, localBuffer, remoteBuffer, stream, cfg, connOut);
707 : }
708 : }
709 1 : BuildUbDbSendTask(stream, connVec[0]->GetUbJettyLiteId(), connOut.pi);
710 :
711 1 : ExecProfiling(loc, rmt, transferOp, stream, taskId);
712 : }
713 :
714 : // Convert hccl::HcommDataType => Hccl::DataType, hccl::HcommReduceOp => Hccl::ReduceOp
715 : static const std::unordered_map<HcommReduceOp, Hccl::ReduceOp> mapHcommReduceOpA5 = {
716 : {HcommReduceOp::HCOMM_REDUCE_SUM, Hccl::ReduceOp::SUM},
717 : {HcommReduceOp::HCOMM_REDUCE_PROD, Hccl::ReduceOp::PROD},
718 : {HcommReduceOp::HCOMM_REDUCE_MAX, Hccl::ReduceOp::MAX},
719 : {HcommReduceOp::HCOMM_REDUCE_MIN, Hccl::ReduceOp::MIN},
720 : {HcommReduceOp::HCOMM_REDUCE_RESERVED, Hccl::ReduceOp::INVALID}
721 : };
722 :
723 : static const std::unordered_map<HcommDataType, Hccl::DataType> mapHcommDataTypeA5 = {
724 : #ifndef OPEN_BUILD_PROJECT
725 : {HcommDataType::HCOMM_DATA_TYPE_HIF8, Hccl::DataType::HIF8},
726 : {HcommDataType::HCOMM_DATA_TYPE_FP8E4M3, Hccl::DataType::FP8E4M3},
727 : {HcommDataType::HCOMM_DATA_TYPE_FP8E5M2, Hccl::DataType::FP8E5M2},
728 : {HcommDataType::HCOMM_DATA_TYPE_FP8E8M0, Hccl::DataType::FP8E8M0},
729 : #endif
730 : {HcommDataType::HCOMM_DATA_TYPE_INT8, Hccl::DataType::INT8},
731 : {HcommDataType::HCOMM_DATA_TYPE_INT16, Hccl::DataType::INT16},
732 : {HcommDataType::HCOMM_DATA_TYPE_INT32, Hccl::DataType::INT32},
733 : {HcommDataType::HCOMM_DATA_TYPE_INT64, Hccl::DataType::INT64},
734 : {HcommDataType::HCOMM_DATA_TYPE_INT128, Hccl::DataType::INT128},
735 : {HcommDataType::HCOMM_DATA_TYPE_UINT8, Hccl::DataType::UINT8},
736 : {HcommDataType::HCOMM_DATA_TYPE_UINT16, Hccl::DataType::UINT16},
737 : {HcommDataType::HCOMM_DATA_TYPE_UINT32, Hccl::DataType::UINT32},
738 : {HcommDataType::HCOMM_DATA_TYPE_UINT64, Hccl::DataType::UINT64},
739 : {HcommDataType::HCOMM_DATA_TYPE_FP16, Hccl::DataType::FP16},
740 : {HcommDataType::HCOMM_DATA_TYPE_FP32, Hccl::DataType::FP32},
741 : {HcommDataType::HCOMM_DATA_TYPE_FP64, Hccl::DataType::FP64},
742 : {HcommDataType::HCOMM_DATA_TYPE_BFP16, Hccl::DataType::BFP16},
743 : {HcommDataType::HCOMM_DATA_TYPE_RESERVED, Hccl::DataType::INVALID}
744 : };
745 :
746 13 : static HcclResult CheckReduceHcommDataTypeAndHcommReduceOp(HcommDataType dataType, HcommReduceOp reduceOp)
747 : {
748 13 : auto dataTypeIt = mapHcommDataTypeA5.find(dataType); // reduce类型，dataType不能是RESERVED
749 13 : if (dataTypeIt == mapHcommDataTypeA5.end() || dataTypeIt->first == HcommDataType::HCOMM_DATA_TYPE_RESERVED) {
750 0 : HCCL_ERROR("[%s] type[%u] is not supported.", __func__, dataType);
751 0 : return HCCL_E_PARA;
752 : }
753 :
754 13 : auto reduceOpIt = mapHcommReduceOpA5.find(reduceOp); // reduce类型，reduceOp不能是RESERVED
755 13 : if (reduceOpIt == mapHcommReduceOpA5.end() || reduceOpIt->first == HcommReduceOp::HCOMM_REDUCE_RESERVED) {
756 0 : HCCL_ERROR("[%s] op[%u] is not supported.", __func__, reduceOp);
757 0 : return HCCL_E_PARA;
758 : }
759 :
760 13 : return HCCL_SUCCESS;
761 : }
762 :
763 : constexpr u32 SIZE_TABLE[HCCL_DATA_TYPE_RESERVED] = {sizeof(s8), sizeof(s16), sizeof(s32),
764 : 2, sizeof(float), sizeof(s64), sizeof(u64), sizeof(u8), sizeof(u16), sizeof(u32),
765 : 8, 2, 16, 2, 1, 1, 1, 1};
766 :
767 8 : static HcclResult ParasReduceData(const HcommBatchTransferDesc &transferDesc, uint64_t &len,
768 : HcommDataType &dataType, HcommReduceOp &reduceOp)
769 : {
770 8 : len = transferDesc.transferInfo.reduce.count;
771 8 : dataType = transferDesc.transferInfo.reduce.dataType;
772 8 : reduceOp = transferDesc.transferInfo.reduce.reduceOp;
773 8 : auto ret = CheckReduceHcommDataTypeAndHcommReduceOp(dataType, reduceOp);
774 8 : CHK_PRT_RET(ret != HCCL_SUCCESS,
775 : HCCL_ERROR("FAIL at CheckReduceHcommDataTypeAndHcommReduceOp dataType[%d], reduceOp[%d].", dataType, reduceOp), ret);
776 8 : return HCCL_SUCCESS;
777 : }
778 :
779 34 : static HcclResult ParseData(const HcommBatchTransferDesc &transferDesc, void* &rmt, void* &loc,
780 : uint64_t &len, Hccl::TransferType &tfType, HcommDataType &dataType, HcommReduceOp &reduceOp, uint32_t ¬ifyIdx)
781 : {
782 34 : if (transferDesc.transType == HCOMM_TRANSFER_TYPE_WRITE) {
783 7 : rmt = transferDesc.transferInfo.write.dst; // write操作，dst是远端地址
784 7 : loc = transferDesc.transferInfo.write.src; // src是本端地址
785 7 : len = transferDesc.transferInfo.write.len;
786 7 : tfType = Hccl::TransferType::WRITE;
787 27 : } else if (transferDesc.transType == HCOMM_TRANSFER_TYPE_READ) {
788 5 : rmt = transferDesc.transferInfo.read.src; // read操作，src是远端地址
789 5 : loc = transferDesc.transferInfo.read.dst; // dst是本端地址
790 5 : len = transferDesc.transferInfo.read.len;
791 5 : tfType = Hccl::TransferType::READ;
792 22 : } else if (transferDesc.transType == HCOMM_TRANSFER_TYPE_WRITE_REDUCE) {
793 4 : rmt = transferDesc.transferInfo.reduce.dst;
794 4 : loc = transferDesc.transferInfo.reduce.src;
795 4 : tfType = Hccl::TransferType::WRITE_REDUCE;
796 4 : CHK_RET(ParasReduceData(transferDesc, len, dataType, reduceOp));
797 18 : } else if (transferDesc.transType == HCOMM_TRANSFER_TYPE_READ_REDUCE) {
798 4 : rmt = transferDesc.transferInfo.reduce.src;
799 4 : loc = transferDesc.transferInfo.reduce.dst;
800 4 : tfType = Hccl::TransferType::READ_REDUCE;
801 4 : CHK_RET(ParasReduceData(transferDesc, len, dataType, reduceOp));
802 14 : } else if (transferDesc.transType == HCOMM_TRANSFER_TYPE_WRITE_WITH_NOTIFY) {
803 4 : rmt = transferDesc.transferInfo.writeWithNotify.dst; // write操作，dst是远端地址
804 4 : loc = transferDesc.transferInfo.writeWithNotify.src; // src是本端地址
805 4 : len = transferDesc.transferInfo.writeWithNotify.len;
806 4 : notifyIdx = transferDesc.transferInfo.writeWithNotify.notifyIdx;
807 4 : tfType = Hccl::TransferType::WRITE_WITH_NOTIFY;
808 10 : } else if (transferDesc.transType == HCOMM_TRANSFER_TYPE_WRITE_REDUCE_WITH_NOTIFY) {
809 5 : rmt = transferDesc.transferInfo.writeReduceWithNotify.dst;
810 5 : loc = transferDesc.transferInfo.writeReduceWithNotify.src;
811 5 : len = transferDesc.transferInfo.writeReduceWithNotify.count;
812 5 : dataType = transferDesc.transferInfo.writeReduceWithNotify.dataType;
813 5 : reduceOp = transferDesc.transferInfo.writeReduceWithNotify.reduceOp;
814 5 : notifyIdx = transferDesc.transferInfo.writeReduceWithNotify.notifyIdx;
815 5 : tfType = Hccl::TransferType::WRITE_REDUCE_WITH_NOTIFY;
816 5 : CHK_RET(CheckReduceHcommDataTypeAndHcommReduceOp(dataType, reduceOp));
817 5 : } else if (transferDesc.transType == HCOMM_TRANSFER_TYPE_NOTIFY_RECORD) {
818 4 : notifyIdx = transferDesc.transferInfo.notifyRecord.notifyIdx;
819 4 : tfType = Hccl::TransferType::NOTIFY_RECORD;
820 : } else {
821 1 : HCCL_ERROR("[%s] unsupported transType[%d]", __func__, transferDesc.transType);
822 1 : return HCCL_E_NOT_SUPPORT;
823 : }
824 33 : if (reduceOp != HcommReduceOp::HCOMM_REDUCE_RESERVED) { // 对于规约类型, size = count * sizeof(datatype)
825 13 : len = len * SIZE_TABLE[dataType];
826 : }
827 33 : return HCCL_SUCCESS;
828 : }
829 : constexpr uint32_t NOTIFYIDX_INVALID_VALUE = 0xFFFFFFFF; // NOTIFY idex非法值
830 12 : HcclResult UbTransportLiteImpl::ExecuteBatchTransfer(StreamLite *streamLitePtr,
831 : const HcommBatchTransferDesc *transferDescs, uint32_t transferDescNum)
832 : {
833 12 : std::vector<Hccl::RmaBufferLite> locSlices;
834 12 : std::vector<Hccl::Buffer> rmtSlices;
835 12 : std::vector<Hccl::BaseTransportLiteImpl::TransferOp> transferOps;
836 12 : std::vector<uint32_t> notifyIdxs;
837 :
838 12 : locSlices.reserve(transferDescNum);
839 12 : rmtSlices.reserve(transferDescNum);
840 12 : transferOps.reserve(transferDescNum);
841 12 : notifyIdxs.reserve(transferDescNum);
842 :
843 42 : for (uint32_t i = 0; i < transferDescNum; i++) {
844 34 : Hccl::RmaBufferLite locRmaBuf;
845 34 : void *rmt = nullptr;
846 34 : void *loc = nullptr;
847 34 : uint64_t len = 0;
848 34 : Hccl::TransferType tfType;
849 34 : HcommDataType dataType{HcommDataType::HCOMM_DATA_TYPE_RESERVED};
850 34 : HcommReduceOp reduceOp{HcommReduceOp::HCOMM_REDUCE_RESERVED};
851 34 : uint32_t notifyIdx = NOTIFYIDX_INVALID_VALUE;
852 37 : CHK_RET(ParseData(transferDescs[i], rmt, loc, len, tfType, dataType, reduceOp, notifyIdx));
853 33 : if (tfType != Hccl::TransferType::NOTIFY_RECORD) { // NOTIFY_RECORD时没有地址字段
854 29 : CHK_PTR_NULL(rmt);
855 28 : CHK_PTR_NULL(loc);
856 27 : HcclResult ret = BuildLocRmaBufferLite(reinterpret_cast<uintptr_t>(loc), len, locRmaBuf);
857 27 : CHK_PRT_RET(ret != HCCL_SUCCESS,
858 : HCCL_ERROR("[%s] FAIL at BuildLocRmaBufferLite for index %u. rmt[%p], loc[%p], len[0x%llx], tfType[%u], dataType[%d], reduceOp[%d].",
859 : __func__, i, rmt, loc, len, tfType, dataType, reduceOp), ret);
860 : }
861 58 : if (tfType == Hccl::TransferType::NOTIFY_RECORD || tfType == Hccl::TransferType::WRITE_WITH_NOTIFY
862 58 : || tfType == Hccl::TransferType::WRITE_REDUCE_WITH_NOTIFY) {
863 13 : CHK_PRT_RET(notifyIdx == NOTIFYIDX_INVALID_VALUE,
864 : HCCL_ERROR("[%s] FAIL at ParseData for index %u. tfType[%u], notifyIdx[%u].",
865 : __func__, i, tfType, notifyIdx), HCCL_E_PARA);
866 : }
867 30 : notifyIdxs.push_back(notifyIdx);
868 30 : locSlices.push_back(locRmaBuf);
869 :
870 30 : const Hccl::Buffer rmtBuf{reinterpret_cast<uintptr_t>(rmt), len};
871 30 : rmtSlices.push_back(rmtBuf);
872 :
873 30 : Hccl::ReduceIn reduceIn{mapHcommDataTypeA5.at(dataType), mapHcommReduceOpA5.at(reduceOp)};
874 :
875 30 : transferOps.push_back(Hccl::BaseTransportLiteImpl::TransferOp{tfType, reduceIn});
876 :
877 30 : HCCL_DEBUG("[%s] Prepared transfer op for index %u. rmt[%p], loc[%p], len[0x%llx], tfType[%u], dataType[%d], reduceOp[%d].",
878 : __func__, i, rmt, loc, len, tfType, dataType, reduceOp);
879 30 : }
880 8 : EXCEPTION_CATCH(BatchTransferAll(locSlices, rmtSlices, transferOps, notifyIdxs, *streamLitePtr), return HCCL_E_INTERNAL);
881 8 : return HCCL_SUCCESS;
882 12 : }
883 :
884 0 : void UbTransportLiteImpl::BatchTransferAll(const std::vector<RmaBufferLite> &loc, const std::vector<Buffer> &rmt,
885 : const std::vector<BaseTransportLiteImpl::TransferOp> &transferOp, const std::vector<uint32_t> ¬ifyIdxs, const StreamLite &stream)
886 : {
887 0 : if (UNLIKELY(loc.empty())) {
888 0 : return;
889 : }
890 0 : auto taskId = stream.GetRtsq()->GetTaskId();
891 0 : u64 notifyData = 1; // 普通notify，固定1，用于writeWithNotify与writeReduceWithNotify
892 0 : SqeConfigLite cfg;
893 0 : SetFenceConfig(cfg);
894 0 : u32 insNum = loc.size();
895 0 : for (u32 i = 0; i < insNum; i++) {
896 0 : cfg.cqeEn = (i == insNum - 1) ? true : false; // 返回最后一个sqe的cqe
897 0 : cfg.placeOdr = (i == insNum - 1) ? UB_STRONG_ORDER : UB_RELAX_ORDER; // 最后一个要求保序
898 0 : cfg.compOrder = (i == insNum - 1) ? UB_COMPLETION : UB_NO_COMPLETION;
899 0 : cfg.userConfig = true;
900 :
901 0 : if (transferOp[i].transType == TransferType::NOTIFY_RECORD) { // notifyRecord操作没有loc/rmt，因此单独处理
902 0 : if (notifyIdxs[i] == 1) { // PostFin场景
903 0 : cfg.cqeEn = true;
904 0 : cfg.placeOdr = UB_STRONG_ORDER;
905 0 : cfg.compOrder = UB_COMPLETION;
906 0 : cfg.userConfig = true;
907 : }
908 0 : u32 inlineData = 1;
909 : // 当前使用1个connection，下标为0 构建sqe
910 0 : connVec[0]->InlineWrite(reinterpret_cast<u8 *>(&inlineData), UB_INLINE_WRITE_SIZE, GetRmtNotifySliceLite(notifyIdxs[i]),
911 0 : cfg, stream, connOut);
912 : } else {
913 0 : auto localBuffer = GetRmaBufSlicelite(loc[i]);
914 0 : auto remoteBuffer = GetRmtRmaBufSliceLite(rmt[i]);
915 :
916 0 : if (transferOp[i].transType == TransferType::WRITE) {
917 0 : connVec[0]->Write(localBuffer, remoteBuffer, cfg, stream, connOut);
918 0 : } else if (transferOp[i].transType == TransferType::WRITE_REDUCE) {
919 0 : connVec[0]->WriteReduce(transferOp[i].reduceIn.dataType, transferOp[i].reduceIn.reduceOp, localBuffer, stream, remoteBuffer, cfg, connOut);
920 0 : } else if (transferOp[i].transType == TransferType::READ) {
921 0 : connVec[0]->Read(localBuffer, remoteBuffer, cfg, stream, connOut);
922 0 : } else if (transferOp[i].transType == TransferType::READ_REDUCE) {
923 0 : connVec[0]->ReadReduce(transferOp[i].reduceIn, localBuffer, remoteBuffer, stream, cfg, connOut);
924 0 : } else if (transferOp[i].transType == TransferType::WRITE_WITH_NOTIFY) {
925 0 : connVec[0]->WriteWithNotify(localBuffer, remoteBuffer, cfg, connOut, GetRmtNotifySliceLite(notifyIdxs[i]), stream, notifyData); // 当前使用1个connection，下标为0
926 0 : } else if (transferOp[i].transType == TransferType::WRITE_REDUCE_WITH_NOTIFY) {
927 0 : connVec[0]->WriteReduceWithNotify(transferOp[i].reduceIn.dataType, transferOp[i].reduceIn.reduceOp, localBuffer,
928 0 : remoteBuffer, cfg, stream, connOut, GetRmtNotifySliceLite(notifyIdxs[i]), notifyData); // 当前使用1个connection，下标为0
929 : }
930 : }
931 : }
932 0 : BuildUbDbSendTask(stream, connVec[0]->GetUbJettyLiteId(), connOut.pi); // 约束使用一批wqe的个数不会导致反压
933 :
934 0 : ExecProfilingAll(loc, rmt, transferOp, stream, taskId, notifyIdxs);
935 : }
936 :
937 0 : void UbTransportLiteImpl::Drain(const StreamLite &stream)
938 : {
939 0 : std::lock_guard<std::mutex> lock(drainMtx_);
940 0 : if (drainNotify_.size == 0 || rmtDrainBuffer_.size == 0) {
941 0 : HCCL_WARNING("[UbTransportLiteImpl::%s] drain resource is null skip", __func__);
942 0 : return;
943 : }
944 :
945 0 : SqeConfigLite cfg;
946 0 : Fence();
947 0 : SetFenceConfig(cfg);
948 :
949 : // 当前使用1个connection,下标为0
950 : auto drainNotifyBufSlice = RmaBufSliceLite(drainNotify_.addr, drainNotify_.size, 0,
951 0 : drainNotify_.tokenId);
952 : auto drainConstBufSlice = RmtRmaBufSliceLite(rmtDrainBuffer_.addr, rmtDrainBuffer_.size, 0,
953 0 : rmtDrainBuffer_.tokenId, rmtDrainBuffer_.tokenValue, UINT32_MAX);
954 0 : connVec[0]->Read(drainNotifyBufSlice, drainConstBufSlice, cfg, stream, connOut);
955 0 : BuildUbDbSendTask(stream, connVec[0]->GetUbJettyLiteId(), connOut.pi);
956 :
957 0 : BuildNotifyWaitTask(stream, drainNotify_.notifyId);
958 0 : }
959 :
960 1 : void UbTransportLiteImpl::WriteWithNotify(const RmaBufferLite &loc, const Buffer &rmt, const WithNotifyIn &withNotify,
961 : const StreamLite &stream)
962 : {
963 1 : SqeConfigLite cfg;
964 1 : SetFenceConfig(cfg);
965 1 : u64 notifyData = 1; // 普通notify，固定1
966 1 : auto taskId = stream.GetRtsq()->GetTaskId();
967 :
968 : // 当前使用1个connection，下标为0
969 1 : auto locRmaBufSlicelite = GetRmaBufSlicelite(loc);
970 1 : auto rmtRmaBufSlicelite = GetRmtRmaBufSliceLite(rmt);
971 1 : auto rmtNotifySliceLite = GetRmtNotifySliceLite(withNotify.index_);
972 1 : connVec[0]->WriteWithNotify(locRmaBufSlicelite, rmtRmaBufSlicelite, cfg, connOut,
973 : rmtNotifySliceLite, stream, notifyData);
974 1 : BuildUbDbSendTask(stream, connVec[0]->GetUbJettyLiteId(), connOut.pi);
975 :
976 1 : if (!IsReportTask()) {
977 0 : return;
978 : }
979 :
980 1 : TaskParam taskParam{};
981 1 : taskParam.taskType = TaskParamType::TASK_WRITE_WITH_NOTIFY;
982 1 : taskParam.beginTime = ProfGetCurCpuTimestamp();
983 1 : taskParam.taskPara.DMA.src = reinterpret_cast<void *>(locRmaBufSlicelite.GetAddr());
984 1 : taskParam.taskPara.DMA.dst = reinterpret_cast<void *>(rmtRmaBufSlicelite.GetAddr());
985 1 : taskParam.taskPara.DMA.size = locRmaBufSlicelite.GetSize();
986 1 : taskParam.taskPara.DMA.notifyID = rmtNotifySliceLite.GetNotifyId();
987 1 : taskParam.taskPara.DMA.notifyValue = 1;
988 1 : taskParam.taskPara.DMA.linkType = DfxLinkType::UB;
989 1 : taskParam.taskPara.DMA.dmaOp = DmaOp::HCCL_DMA_WRITE;
990 1 : taskParam.taskPara.DMA.locEid = GetLocEid();
991 1 : taskParam.taskPara.DMA.rmtEid = GetRmtEid();
992 1 : taskParam.taskPara.DMA.jettyHandle = GetJettyHandle();
993 1 : taskParam.taskPara.DMA.jettyId = GetJettyId();
994 :
995 1 : AddTaskCallback(stream, taskId, taskParam);
996 1 : }
997 :
998 0 : void UbTransportLiteImpl::WriteReduceWithNotify(const RmaBufferLite &loc, const Buffer &rmt, const ReduceIn &reduceIn,
999 : const WithNotifyIn &withNotify, const StreamLite &stream)
1000 : {
1001 0 : SqeConfigLite cfg;
1002 0 : SetFenceConfig(cfg);
1003 0 : u64 notifyData = 1; // 普通notify，固定1
1004 0 : auto taskId = stream.GetRtsq()->GetTaskId();
1005 :
1006 : // 当前使用1个connection，下标为0
1007 0 : auto locRmaBufSlicelite = GetRmaBufSlicelite(loc);
1008 0 : auto rmtRmaBufSlicelite = GetRmtRmaBufSliceLite(rmt);
1009 0 : auto rmtNotifySliceLite = GetRmtNotifySliceLite(withNotify.index_);
1010 0 : connVec[0]->WriteReduceWithNotify(reduceIn.dataType, reduceIn.reduceOp, locRmaBufSlicelite,
1011 0 : rmtRmaBufSlicelite, cfg, stream, connOut, rmtNotifySliceLite,
1012 : notifyData);
1013 0 : BuildUbDbSendTask(stream, connVec[0]->GetUbJettyLiteId(), connOut.pi);
1014 0 : if (!IsReportTask()) {
1015 0 : return;
1016 : }
1017 :
1018 0 : TaskParam taskParam{};
1019 0 : taskParam.taskType = TaskParamType::TASK_WRITE_REDUCE_WITH_NOTIFY;
1020 0 : taskParam.beginTime = ProfGetCurCpuTimestamp();
1021 0 : taskParam.taskPara.Reduce.src = reinterpret_cast<void *>(locRmaBufSlicelite.GetAddr());
1022 0 : taskParam.taskPara.Reduce.dst = reinterpret_cast<void *>(rmtRmaBufSlicelite.GetAddr());
1023 0 : taskParam.taskPara.Reduce.size = locRmaBufSlicelite.GetSize();
1024 0 : taskParam.taskPara.Reduce.notifyID = rmtNotifySliceLite.GetNotifyId();
1025 0 : taskParam.taskPara.Reduce.notifyValue = 1;
1026 0 : taskParam.taskPara.Reduce.linkType = DfxLinkType::UB;
1027 0 : taskParam.taskPara.Reduce.reduceOp = ConvertReduceOpToHcclReduceOp(reduceIn.reduceOp);
1028 0 : taskParam.taskPara.Reduce.dataType = DataTypeToHcclDataType(reduceIn.dataType);
1029 0 : taskParam.taskPara.Reduce.locEid = GetLocEid();
1030 0 : taskParam.taskPara.Reduce.rmtEid = GetRmtEid();
1031 0 : taskParam.taskPara.DMA.jettyHandle = GetJettyHandle();
1032 0 : taskParam.taskPara.DMA.jettyId = GetJettyId();
1033 :
1034 0 : AddTaskCallback(stream, taskId, taskParam);
1035 0 : }
1036 :
1037 1 : void UbTransportLiteImpl::BatchOneSidedRead(const vector<RmaBufSliceLite> &loc, const vector<RmtRmaBufSliceLite> &rmt,
1038 : const StreamLite &stream)
1039 : {
1040 1 : SqeConfigLite cfg;
1041 1 : SetFenceConfig(cfg);
1042 :
1043 : // 当前使用1个connection，下标为0
1044 1 : connVec[0]->BatchOneSidedRead(loc, rmt, cfg, stream, connOut);
1045 1 : BuildUbDbSendTask(stream, connVec[0]->GetUbJettyLiteId(), connOut.pi);
1046 1 : }
1047 :
1048 1 : void UbTransportLiteImpl::BatchOneSidedWrite(const vector<RmaBufSliceLite> &loc, const vector<RmtRmaBufSliceLite> &rmt,
1049 : const StreamLite &stream)
1050 : {
1051 1 : SqeConfigLite cfg;
1052 1 : SetFenceConfig(cfg);
1053 :
1054 : // 当前使用1个connection，下标为0
1055 1 : connVec[0]->BatchOneSidedWrite(loc, rmt, cfg, stream, connOut);
1056 1 : BuildUbDbSendTask(stream, connVec[0]->GetUbJettyLiteId(), connOut.pi);
1057 1 : }
1058 :
1059 :
1060 8 : Eid UbTransportLiteImpl::GetLocEid() const
1061 : {
1062 8 : return connVec[0]->GetLocEid();
1063 : }
1064 :
1065 8 : Eid UbTransportLiteImpl::GetRmtEid() const
1066 : {
1067 8 : return connVec[0]->GetRmtEid();
1068 : }
1069 :
1070 11 : uint64_t UbTransportLiteImpl::GetJettyHandle() const
1071 : {
1072 11 : return connVec[0]->GetJettyHandle();
1073 : }
1074 :
1075 11 : uint32_t UbTransportLiteImpl::GetJettyId() const
1076 : {
1077 11 : return connVec[0]->GetJettyId();
1078 : }
1079 :
1080 0 : HcclResult UbTransportLiteImpl::Clean()
1081 : {
1082 0 : locNotifyVec.clear();
1083 0 : rmtNotifyVec.clear();
1084 0 : locBufferMap.clear();
1085 0 : rmtBufferVec.clear();
1086 0 : rmtBufferMap.clear();
1087 :
1088 : // 清理connVec，connLite由UbConnLiteMgr管理
1089 0 : for (auto &it : connUniqueIdVec) {
1090 0 : DECTOR_TRY_CATCH("UbTransportLiteImpl", UbConnLiteMgr::GetInstance().Clear(it));
1091 : }
1092 0 : connUniqueIdVec.clear();
1093 0 : connVec.clear();
1094 :
1095 0 : return HCCL_SUCCESS;
1096 : }
1097 :
1098 0 : HcclResult UbTransportLiteImpl::Resume(std::vector<char> &uniqueId)
1099 : {
1100 0 : Init(uniqueId);
1101 0 : return HCCL_SUCCESS;
1102 : }
1103 :
1104 1 : HcclResult UbTransportLiteImpl::Fence()
1105 : {
1106 1 : fence_ = true;
1107 1 : HCCL_INFO("[%s] SUCCESS. fence[%d]", __func__, fence_);
1108 1 : return HCCL_SUCCESS;
1109 : }
1110 :
1111 8 : void UbTransportLiteImpl::SetFenceConfig(SqeConfigLite &cfg)
1112 : {
1113 8 : if (fence_) {
1114 0 : cfg.fence = UB_FENCE_ENABLED;
1115 0 : cfg.placeOdr = UB_STRONG_ORDER;
1116 0 : cfg.compOrder = UB_COMPLETION;
1117 0 : cfg.userConfig = true;
1118 : }
1119 8 : fence_ = false;
1120 8 : }
1121 :
1122 11 : bool UbTransportLiteImpl::IsReportTask()
1123 : {
1124 22 : return (taskExceptionEnable_ || ProfilingHandlerLite::GetInstance().GetProfL1State()) &&
1125 22 : (callback_ != nullptr || newCallback_ != nullptr);
1126 : }
1127 : } // namespace Hccl
|
