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 : }
375 1 : u32 inlineData = 1;
376 :
377 1 : auto taskId = stream.GetRtsq()->GetTaskId();
378 : // 当前使用1个connection，下标为0 构建sqe
379 1 : auto rmtBuffSliceLite = GetRmtNotifySliceLite(index);
380 1 : connVec[0]->InlineWrite(reinterpret_cast<u8 *>(&inlineData), UB_INLINE_WRITE_SIZE, rmtBuffSliceLite,
381 1 : cfg, stream, connOut);
382 : // 构建rts 的 sqe
383 1 : BuildUbDbSendTask(stream, connVec[0]->GetUbJettyLiteId(), connOut.pi);
384 :
385 3 : HCCL_INFO("UbTransportLiteImpl::Post notifyId[0x%llx], pi=%u", rmtBuffSliceLite.GetAddr(), connOut.pi);
386 :
387 1 : if (!IsReportTask()) {
388 0 : return;
389 : }
390 :
391 1 : TaskParam taskParam{};
392 1 : taskParam.taskType = TaskParamType::TASK_UB_INLINE_WRITE;
393 1 : taskParam.beginTime = ProfGetCurCpuTimestamp();
394 1 : taskParam.taskPara.DMA.dst = reinterpret_cast<void*>(rmtBuffSliceLite.GetAddr());
395 1 : taskParam.taskPara.DMA.size = rmtBuffSliceLite.GetSize();
396 1 : taskParam.taskPara.DMA.notifyID = rmtBuffSliceLite.GetNotifyId();
397 1 : taskParam.taskPara.DMA.notifyValue = 1;
398 1 : taskParam.taskPara.DMA.linkType = DfxLinkType::UB;
399 1 : taskParam.taskPara.DMA.dmaOp = DmaOp::HCCL_DMA_WRITE;
400 1 : taskParam.taskPara.DMA.locEid = GetLocEid();
401 1 : taskParam.taskPara.DMA.rmtEid = GetRmtEid();
402 :
403 3 : HCCL_INFO("[UbTransportLiteImpl::%s] locEid[%s], rmtEid[%s]", __func__, GetLocEid().Describe().c_str(), GetRmtEid().Describe().c_str());
404 :
405 1 : AddTaskCallback(stream, taskId, taskParam);
406 1 : }
407 :
408 1 : void UbTransportLiteImpl::Wait(u32 index, const StreamLite &stream)
409 : {
410 1 : WaitWithTimeout(index, stream, CommunicatorImplLiteMgr::GetInstance().GetEnvConfig().hcclExecTimeout);
411 1 : }
412 :
413 4 : void UbTransportLiteImpl::WaitWithTimeout(u32 index, const StreamLite &stream, u32 timeout)
414 : {
415 4 : auto taskId = stream.GetRtsq()->GetTaskId();
416 4 : auto notifyId = locNotifyVec[index]->GetId();
417 4 : stream.GetRtsq()->NotifyWait(notifyId, timeout);
418 :
419 4 : if (!IsReportTask()) {
420 0 : return;
421 : }
422 :
423 4 : TaskParam taskParam{};
424 4 : taskParam.taskType = TaskParamType::TASK_NOTIFY_WAIT;
425 4 : taskParam.beginTime = ProfGetCurCpuTimestamp();
426 4 : taskParam.taskPara.Notify.notifyID = notifyId;
427 4 : taskParam.taskPara.Notify.value = 1;
428 :
429 4 : AddTaskCallback(stream, taskId, taskParam);
430 4 : }
431 :
432 3 : void UbTransportLiteImpl::ProfilingProcess(void *src, void *dst, u64 size, const StreamLite &stream,
433 : DmaOp dmaOp, u32 taskId)
434 : {
435 3 : if (!IsReportTask()) {
436 0 : return;
437 : }
438 :
439 3 : TaskParam taskParam{};
440 3 : taskParam.taskType = TaskParamType::TASK_UB;
441 3 : taskParam.beginTime = ProfGetCurCpuTimestamp();
442 3 : FillTaskParamDmaPub(taskParam, dst, size, dmaOp);
443 3 : taskParam.taskPara.DMA.src = src;
444 :
445 3 : AddTaskCallback(stream, taskId, taskParam);
446 3 : }
447 :
448 2 : void UbTransportLiteImpl::ReduceProfilingProcess(void *src, void *dst, u64 size,
449 : const ReduceIn &reduceIn, const StreamLite &stream, u32 taskId)
450 : {
451 2 : if (!IsReportTask()) {
452 0 : return;
453 : }
454 :
455 2 : TaskParam taskParam {};
456 2 : taskParam.taskType = TaskParamType::TASK_UB_REDUCE_INLINE;
457 2 : taskParam.beginTime = ProfGetCurCpuTimestamp();
458 2 : taskParam.taskPara.Reduce.src = src;
459 2 : taskParam.taskPara.Reduce.dst = dst;
460 2 : taskParam.taskPara.Reduce.size = size;
461 2 : taskParam.taskPara.Reduce.notifyID = INVALID_VALUE_NOTIFYID;
462 2 : taskParam.taskPara.Reduce.notifyValue = 1;
463 2 : taskParam.taskPara.Reduce.linkType = DfxLinkType::UB;
464 2 : taskParam.taskPara.Reduce.reduceOp = ConvertReduceOpToHcclReduceOp(reduceIn.reduceOp);
465 2 : taskParam.taskPara.Reduce.dataType = DataTypeToHcclDataType(reduceIn.dataType);
466 2 : taskParam.taskPara.Reduce.locEid = GetLocEid();
467 2 : taskParam.taskPara.Reduce.rmtEid = GetRmtEid();
468 :
469 2 : AddTaskCallback(stream, taskId, taskParam);
470 2 : }
471 :
472 0 : void UbTransportLiteImpl::WriteWithNotifyProfilingProcess(void *src, void *dst, u64 size, const StreamLite &stream,
473 : u32 taskId, u64 notifyId)
474 : {
475 0 : if (!IsReportTask()) {
476 0 : return;
477 : }
478 :
479 0 : TaskParam taskParam{};
480 0 : taskParam.taskType = TaskParamType::TASK_WRITE_WITH_NOTIFY;
481 0 : taskParam.beginTime = ProfGetCurCpuTimestamp();
482 0 : FillTaskParamDmaPub(taskParam, dst, size, DmaOp::HCCL_DMA_WRITE);
483 0 : taskParam.taskPara.DMA.src = src;
484 0 : taskParam.taskPara.DMA.notifyID = notifyId;
485 0 : taskParam.taskPara.DMA.notifyValue = 1;
486 :
487 0 : AddTaskCallback(stream, taskId, taskParam);
488 0 : }
489 :
490 0 : void UbTransportLiteImpl::WriteReduceWithNotifyProfilingProcess(void *src, void *dst, u64 size,
491 : const ReduceIn &reduceIn, const StreamLite &stream, u32 taskId, u64 notifyId)
492 : {
493 0 : if (!IsReportTask()) {
494 0 : return;
495 : }
496 :
497 0 : TaskParam taskParam {};
498 0 : taskParam.taskType = TaskParamType::TASK_WRITE_REDUCE_WITH_NOTIFY;
499 0 : taskParam.beginTime = ProfGetCurCpuTimestamp();
500 0 : taskParam.taskPara.Reduce.src = src;
501 0 : taskParam.taskPara.Reduce.dst = dst;
502 0 : taskParam.taskPara.Reduce.size = size;
503 0 : taskParam.taskPara.Reduce.notifyID = notifyId;
504 0 : taskParam.taskPara.Reduce.notifyValue = 1;
505 0 : taskParam.taskPara.Reduce.linkType = DfxLinkType::UB;
506 0 : taskParam.taskPara.Reduce.reduceOp = ConvertReduceOpToHcclReduceOp(reduceIn.reduceOp);
507 0 : taskParam.taskPara.Reduce.dataType = DataTypeToHcclDataType(reduceIn.dataType);
508 0 : taskParam.taskPara.Reduce.locEid = GetLocEid();
509 0 : taskParam.taskPara.Reduce.rmtEid = GetRmtEid();
510 :
511 0 : AddTaskCallback(stream, taskId, taskParam);
512 0 : }
513 :
514 0 : void UbTransportLiteImpl::NotifyRecordProfilingProcess(void *dst, u64 size,
515 : 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_UB_INLINE_WRITE;
523 0 : taskParam.beginTime = ProfGetCurCpuTimestamp();
524 0 : FillTaskParamDmaPub(taskParam, dst, size, DmaOp::HCCL_DMA_WRITE);
525 0 : taskParam.taskPara.DMA.notifyID = notifyId;
526 0 : taskParam.taskPara.DMA.notifyValue = 1;
527 :
528 0 : AddTaskCallback(stream, taskId, taskParam);
529 0 : }
530 :
531 1 : void UbTransportLiteImpl::Read(const RmaBufferLite &loc, const Buffer &rmt, const StreamLite &stream)
532 : {
533 1 : SqeConfigLite cfg;
534 1 : SetFenceConfig(cfg);
535 1 : auto taskId = stream.GetRtsq()->GetTaskId();
536 :
537 : // 当前使用1个connection,下标为0
538 1 : auto locRmaBufSlicelite = GetRmaBufSlicelite(loc);
539 1 : auto rmtRmaBufSlicelite = GetRmtRmaBufSliceLite(rmt);
540 1 : connVec[0]->Read(locRmaBufSlicelite, rmtRmaBufSlicelite, cfg, stream, connOut);
541 1 : BuildUbDbSendTask(stream, connVec[0]->GetUbJettyLiteId(), connOut.pi);
542 :
543 2 : ProfilingProcess(reinterpret_cast<void *>(locRmaBufSlicelite.GetAddr()),
544 1 : reinterpret_cast<void *>(rmtRmaBufSlicelite.GetAddr()),
545 : locRmaBufSlicelite.GetSize(), stream, DmaOp::HCCL_DMA_READ, taskId);
546 1 : }
547 :
548 1 : void UbTransportLiteImpl::Write(const RmaBufferLite &loc, const Buffer &rmt, const StreamLite &stream)
549 : {
550 1 : SqeConfigLite cfg;
551 1 : SetFenceConfig(cfg);
552 1 : auto taskId = stream.GetRtsq()->GetTaskId();
553 :
554 : // 当前使用1个connection，下标为0
555 1 : auto locRmaBufSlicelite = GetRmaBufSlicelite(loc);
556 1 : auto rmtRmaBufSlicelite = GetRmtRmaBufSliceLite(rmt);
557 1 : connVec[0]->Write(locRmaBufSlicelite, rmtRmaBufSlicelite, cfg, stream, connOut);
558 1 : BuildUbDbSendTask(stream, connVec[0]->GetUbJettyLiteId(), connOut.pi);
559 :
560 2 : ProfilingProcess(reinterpret_cast<void *>(locRmaBufSlicelite.GetAddr()),
561 1 : reinterpret_cast<void *>(rmtRmaBufSlicelite.GetAddr()),
562 : locRmaBufSlicelite.GetSize(), stream, DmaOp::HCCL_DMA_WRITE, taskId);
563 1 : }
564 :
565 1 : void UbTransportLiteImpl::ReadReduce(const RmaBufferLite &loc, const Buffer &rmt, const ReduceIn &reduceIn,
566 : const StreamLite &stream)
567 : {
568 1 : SqeConfigLite cfg;
569 1 : SetFenceConfig(cfg);
570 1 : auto taskId = stream.GetRtsq()->GetTaskId();
571 :
572 : // 当前使用1个connection，下标为0
573 1 : auto locRmaBufSlicelite = GetRmaBufSlicelite(loc);
574 1 : auto rmtRmaBufSlicelite = GetRmtRmaBufSliceLite(rmt);
575 1 : connVec[0]->ReadReduce(reduceIn, locRmaBufSlicelite, rmtRmaBufSlicelite, stream, cfg, connOut);
576 1 : BuildUbDbSendTask(stream, connVec[0]->GetUbJettyLiteId(), connOut.pi);
577 :
578 2 : ReduceProfilingProcess(reinterpret_cast<void *>(locRmaBufSlicelite.GetAddr()),
579 1 : reinterpret_cast<void *>(rmtRmaBufSlicelite.GetAddr()),
580 : locRmaBufSlicelite.GetSize(), reduceIn, stream, taskId);
581 1 : }
582 :
583 1 : void UbTransportLiteImpl::WriteReduce(const RmaBufferLite &loc, const Buffer &rmt, const ReduceIn &reduceIn,
584 : const StreamLite &stream)
585 : {
586 1 : SqeConfigLite cfg;
587 1 : SetFenceConfig(cfg);
588 1 : auto taskId = stream.GetRtsq()->GetTaskId();
589 :
590 : // 当前使用1个connection，下标为0
591 1 : auto locRmaBufSlicelite = GetRmaBufSlicelite(loc);
592 1 : auto rmtRmaBufSlicelite = GetRmtRmaBufSliceLite(rmt);
593 1 : connVec[0]->WriteReduce(reduceIn.dataType, reduceIn.reduceOp, locRmaBufSlicelite, stream,
594 1 : rmtRmaBufSlicelite, cfg, connOut);
595 1 : BuildUbDbSendTask(stream, connVec[0]->GetUbJettyLiteId(), connOut.pi);
596 :
597 2 : ReduceProfilingProcess(reinterpret_cast<void *>(locRmaBufSlicelite.GetAddr()),
598 1 : reinterpret_cast<void *>(rmtRmaBufSlicelite.GetAddr()),
599 : locRmaBufSlicelite.GetSize(), reduceIn, stream, taskId);
600 1 : }
601 :
602 1 : void UbTransportLiteImpl::ExecProfiling(const std::vector<RmaBufferLite> &loc, const std::vector<Buffer> &rmt,
603 : const std::vector<BaseTransportLiteImpl::TransferOp> &transferOp, const StreamLite &stream, u32 taskId)
604 : {
605 1 : u32 insNum = loc.size();
606 1 : u64 totalSize = 0;
607 2 : for (u32 i = 0; i < insNum; i++) {
608 1 : totalSize += GetRmaBufSlicelite(loc[i]).GetSize();
609 : }
610 1 : if (transferOp[insNum - 1].reduceIn.reduceOp == ReduceOp::INVALID) {
611 1 : DmaOp dmaOp = DmaOp::HCCL_DMA_WRITE;
612 1 : if (transferOp[insNum - 1].transType == TransferType::READ) {
613 1 : dmaOp = DmaOp::HCCL_DMA_READ;
614 : }
615 1 : ProfilingProcess(reinterpret_cast<void *>(GetRmaBufSlicelite(loc[insNum - 1]).GetAddr()),
616 2 : reinterpret_cast<void *>(GetRmtRmaBufSliceLite(rmt[insNum - 1]).GetAddr()),
617 : totalSize, stream, dmaOp, taskId);
618 : } else {
619 0 : ReduceProfilingProcess(reinterpret_cast<void *>(GetRmaBufSlicelite(loc[insNum - 1]).GetAddr()),
620 0 : reinterpret_cast<void *>(GetRmtRmaBufSliceLite(rmt[insNum - 1]).GetAddr()),
621 0 : totalSize, transferOp[insNum - 1].reduceIn, stream, taskId);
622 : }
623 1 : }
624 :
625 0 : void UbTransportLiteImpl::ExecProfilingAll(const std::vector<RmaBufferLite> &loc, const std::vector<Buffer> &rmt,
626 : const std::vector<BaseTransportLiteImpl::TransferOp> &transferOp, const StreamLite &stream, u32 taskId,
627 : const std::vector<uint32_t> ¬ifyIdxs)
628 : {
629 0 : u32 insNum = loc.size();
630 0 : u64 totalSize = 0;
631 0 : for (u32 i = 0; i < insNum; i++) {
632 0 : totalSize += GetRmaBufSlicelite(loc[i]).GetSize();
633 : }
634 :
635 0 : if (transferOp[insNum - 1].transType == TransferType::READ) {
636 0 : ProfilingProcess(reinterpret_cast<void *>(GetRmaBufSlicelite(loc[insNum - 1]).GetAddr()),
637 0 : reinterpret_cast<void *>(GetRmtRmaBufSliceLite(rmt[insNum - 1]).GetAddr()),
638 : totalSize, stream, DmaOp::HCCL_DMA_READ, taskId);
639 0 : } else if (transferOp[insNum - 1].transType == TransferType::WRITE) {
640 0 : ProfilingProcess(reinterpret_cast<void *>(GetRmaBufSlicelite(loc[insNum - 1]).GetAddr()),
641 0 : reinterpret_cast<void *>(GetRmtRmaBufSliceLite(rmt[insNum - 1]).GetAddr()),
642 : totalSize, stream, DmaOp::HCCL_DMA_WRITE, taskId);
643 0 : } else if (transferOp[insNum - 1].transType == TransferType::READ_REDUCE) {
644 0 : ReduceProfilingProcess(reinterpret_cast<void *>(GetRmaBufSlicelite(loc[insNum - 1]).GetAddr()),
645 0 : reinterpret_cast<void *>(GetRmtRmaBufSliceLite(rmt[insNum - 1]).GetAddr()),
646 0 : totalSize, transferOp[insNum - 1].reduceIn, stream, taskId);
647 0 : } else if (transferOp[insNum - 1].transType == TransferType::WRITE_REDUCE) {
648 0 : ReduceProfilingProcess(reinterpret_cast<void *>(GetRmaBufSlicelite(loc[insNum - 1]).GetAddr()),
649 0 : reinterpret_cast<void *>(GetRmtRmaBufSliceLite(rmt[insNum - 1]).GetAddr()),
650 0 : totalSize, transferOp[insNum - 1].reduceIn, stream, taskId);
651 0 : } else if (transferOp[insNum - 1].transType == TransferType::WRITE_WITH_NOTIFY) {
652 0 : WriteWithNotifyProfilingProcess(reinterpret_cast<void *>(GetRmaBufSlicelite(loc[insNum - 1]).GetAddr()),
653 0 : reinterpret_cast<void *>(GetRmtRmaBufSliceLite(rmt[insNum - 1]).GetAddr()),
654 0 : totalSize, stream, taskId, GetRmtNotifySliceLite(notifyIdxs[insNum - 1]).GetAddr());
655 0 : } else if (transferOp[insNum - 1].transType == TransferType::WRITE_REDUCE_WITH_NOTIFY) {
656 0 : WriteReduceWithNotifyProfilingProcess(reinterpret_cast<void *>(GetRmaBufSlicelite(loc[insNum - 1]).GetAddr()),
657 0 : reinterpret_cast<void *>(GetRmtRmaBufSliceLite(rmt[insNum - 1]).GetAddr()),
658 0 : totalSize, transferOp[insNum - 1].reduceIn, stream, taskId, GetRmtNotifySliceLite(notifyIdxs[insNum - 1]).GetAddr());
659 0 : } else if (transferOp[insNum - 1].transType == TransferType::NOTIFY_RECORD) {
660 0 : NotifyRecordProfilingProcess(reinterpret_cast<void *>(GetRmtNotifySliceLite(notifyIdxs[insNum - 1]).GetAddr()),
661 0 : GetRmtNotifySliceLite(notifyIdxs[insNum - 1]).GetSize(), stream, taskId, GetRmtNotifySliceLite(notifyIdxs[insNum - 1]).GetAddr());
662 : }
663 0 : }
664 :
665 1 : void UbTransportLiteImpl::BatchTransfer(const std::vector<RmaBufferLite> &loc, const std::vector<Buffer> &rmt,
666 : const std::vector<BaseTransportLiteImpl::TransferOp> &transferOp, const StreamLite &stream)
667 : {
668 1 : if (UNLIKELY(loc.empty())) {
669 0 : return;
670 : }
671 1 : SqeConfigLite cfg;
672 1 : SetFenceConfig(cfg);
673 1 : auto taskId = stream.GetRtsq()->GetTaskId();
674 1 : u32 insNum = loc.size();
675 2 : for (u32 i = 0; i < insNum; i++) {
676 1 : cfg.cqeEn = (i == insNum - 1) ? true : false; // 返回最后一个sqe的cqe
677 1 : cfg.placeOdr = UB_RELAX_ORDER;
678 1 : cfg.compOrder = UB_NO_COMPLETION;
679 :
680 1 : auto localBuffer = GetRmaBufSlicelite(loc[i]);
681 1 : auto remoteBuffer = GetRmtRmaBufSliceLite(rmt[i]);
682 1 : if (transferOp[i].transType == TransferType::WRITE) {
683 0 : connVec[0]->Write(localBuffer, remoteBuffer, cfg, stream, connOut); // 当前只有一个connection，对应一个jetty
684 1 : } else if (transferOp[i].transType == TransferType::WRITE_REDUCE) { // write reduce
685 0 : connVec[0]->WriteReduce(transferOp[i].reduceIn.dataType, transferOp[i].reduceIn.reduceOp, localBuffer,
686 0 : stream, remoteBuffer, cfg, connOut);
687 1 : } else if (transferOp[i].transType == TransferType::READ) {
688 1 : connVec[0]->Read(localBuffer, remoteBuffer, cfg, stream, connOut); // 当前只有一个connection，对应一个jetty
689 0 : } else if (transferOp[i].transType == TransferType::READ_REDUCE) { // read reduce
690 0 : connVec[0]->ReadReduce(transferOp[i].reduceIn, localBuffer, remoteBuffer, stream, cfg, connOut);
691 : }
692 : }
693 1 : BuildUbDbSendTask(stream, connVec[0]->GetUbJettyLiteId(), connOut.pi);
694 :
695 1 : ExecProfiling(loc, rmt, transferOp, stream, taskId);
696 : }
697 :
698 : // Convert hccl::HcommDataType => Hccl::DataType, hccl::HcommReduceOp => Hccl::ReduceOp
699 : static const std::unordered_map<HcommReduceOp, Hccl::ReduceOp> mapHcommReduceOpA5 = {
700 : {HcommReduceOp::HCOMM_REDUCE_SUM, Hccl::ReduceOp::SUM},
701 : {HcommReduceOp::HCOMM_REDUCE_PROD, Hccl::ReduceOp::PROD},
702 : {HcommReduceOp::HCOMM_REDUCE_MAX, Hccl::ReduceOp::MAX},
703 : {HcommReduceOp::HCOMM_REDUCE_MIN, Hccl::ReduceOp::MIN},
704 : {HcommReduceOp::HCOMM_REDUCE_RESERVED, Hccl::ReduceOp::INVALID}
705 : };
706 :
707 : static const std::unordered_map<HcommDataType, Hccl::DataType> mapHcommDataTypeA5 = {
708 : #ifndef OPEN_BUILD_PROJECT
709 : {HcommDataType::HCOMM_DATA_TYPE_HIF8, Hccl::DataType::HIF8},
710 : {HcommDataType::HCOMM_DATA_TYPE_FP8E4M3, Hccl::DataType::FP8E4M3},
711 : {HcommDataType::HCOMM_DATA_TYPE_FP8E5M2, Hccl::DataType::FP8E5M2},
712 : {HcommDataType::HCOMM_DATA_TYPE_FP8E8M0, Hccl::DataType::FP8E8M0},
713 : #endif
714 : {HcommDataType::HCOMM_DATA_TYPE_INT8, Hccl::DataType::INT8},
715 : {HcommDataType::HCOMM_DATA_TYPE_INT16, Hccl::DataType::INT16},
716 : {HcommDataType::HCOMM_DATA_TYPE_INT32, Hccl::DataType::INT32},
717 : {HcommDataType::HCOMM_DATA_TYPE_INT64, Hccl::DataType::INT64},
718 : {HcommDataType::HCOMM_DATA_TYPE_INT128, Hccl::DataType::INT128},
719 : {HcommDataType::HCOMM_DATA_TYPE_UINT8, Hccl::DataType::UINT8},
720 : {HcommDataType::HCOMM_DATA_TYPE_UINT16, Hccl::DataType::UINT16},
721 : {HcommDataType::HCOMM_DATA_TYPE_UINT32, Hccl::DataType::UINT32},
722 : {HcommDataType::HCOMM_DATA_TYPE_UINT64, Hccl::DataType::UINT64},
723 : {HcommDataType::HCOMM_DATA_TYPE_FP16, Hccl::DataType::FP16},
724 : {HcommDataType::HCOMM_DATA_TYPE_FP32, Hccl::DataType::FP32},
725 : {HcommDataType::HCOMM_DATA_TYPE_FP64, Hccl::DataType::FP64},
726 : {HcommDataType::HCOMM_DATA_TYPE_BFP16, Hccl::DataType::BFP16},
727 : {HcommDataType::HCOMM_DATA_TYPE_RESERVED, Hccl::DataType::INVALID}
728 : };
729 :
730 13 : static HcclResult CheckReduceHcommDataTypeAndHcommReduceOp(HcommDataType dataType, HcommReduceOp reduceOp)
731 : {
732 13 : auto dataTypeIt = mapHcommDataTypeA5.find(dataType); // reduce类型，dataType不能是RESERVED
733 13 : if (dataTypeIt == mapHcommDataTypeA5.end() || dataTypeIt->first == HcommDataType::HCOMM_DATA_TYPE_RESERVED) {
734 0 : HCCL_ERROR("[%s] type[%u] is not supported.", __func__, dataType);
735 0 : return HCCL_E_PARA;
736 : }
737 :
738 13 : auto reduceOpIt = mapHcommReduceOpA5.find(reduceOp); // reduce类型，reduceOp不能是RESERVED
739 13 : if (reduceOpIt == mapHcommReduceOpA5.end() || reduceOpIt->first == HcommReduceOp::HCOMM_REDUCE_RESERVED) {
740 0 : HCCL_ERROR("[%s] op[%u] is not supported.", __func__, reduceOp);
741 0 : return HCCL_E_PARA;
742 : }
743 :
744 13 : return HCCL_SUCCESS;
745 : }
746 :
747 : constexpr u32 SIZE_TABLE[HCCL_DATA_TYPE_RESERVED] = {sizeof(s8), sizeof(s16), sizeof(s32),
748 : 2, sizeof(float), sizeof(s64), sizeof(u64), sizeof(u8), sizeof(u16), sizeof(u32),
749 : 8, 2, 16, 2, 1, 1, 1, 1};
750 :
751 8 : static HcclResult ParasReduceData(const HcommBatchTransferDesc &transferDesc, uint64_t &len,
752 : HcommDataType &dataType, HcommReduceOp &reduceOp)
753 : {
754 8 : len = transferDesc.transferInfo.reduce.count;
755 8 : dataType = transferDesc.transferInfo.reduce.dataType;
756 8 : reduceOp = transferDesc.transferInfo.reduce.reduceOp;
757 8 : auto ret = CheckReduceHcommDataTypeAndHcommReduceOp(dataType, reduceOp);
758 8 : CHK_PRT_RET(ret != HCCL_SUCCESS,
759 : HCCL_ERROR("FAIL at CheckReduceHcommDataTypeAndHcommReduceOp dataType[%d], reduceOp[%d].", dataType, reduceOp), ret);
760 8 : return HCCL_SUCCESS;
761 : }
762 :
763 34 : static HcclResult ParseData(const HcommBatchTransferDesc &transferDesc, void* &rmt, void* &loc,
764 : uint64_t &len, Hccl::TransferType &tfType, HcommDataType &dataType, HcommReduceOp &reduceOp, uint32_t ¬ifyIdx)
765 : {
766 34 : if (transferDesc.transType == HCOMM_TRANSFER_TYPE_WRITE) {
767 7 : rmt = transferDesc.transferInfo.write.dst; // write操作，dst是远端地址
768 7 : loc = transferDesc.transferInfo.write.src; // src是本端地址
769 7 : len = transferDesc.transferInfo.write.len;
770 7 : tfType = Hccl::TransferType::WRITE;
771 27 : } else if (transferDesc.transType == HCOMM_TRANSFER_TYPE_READ) {
772 5 : rmt = transferDesc.transferInfo.read.src; // read操作，src是远端地址
773 5 : loc = transferDesc.transferInfo.read.dst; // dst是本端地址
774 5 : len = transferDesc.transferInfo.read.len;
775 5 : tfType = Hccl::TransferType::READ;
776 22 : } else if (transferDesc.transType == HCOMM_TRANSFER_TYPE_WRITE_REDUCE) {
777 4 : rmt = transferDesc.transferInfo.reduce.dst;
778 4 : loc = transferDesc.transferInfo.reduce.src;
779 4 : tfType = Hccl::TransferType::WRITE_REDUCE;
780 4 : CHK_RET(ParasReduceData(transferDesc, len, dataType, reduceOp));
781 18 : } else if (transferDesc.transType == HCOMM_TRANSFER_TYPE_READ_REDUCE) {
782 4 : rmt = transferDesc.transferInfo.reduce.src;
783 4 : loc = transferDesc.transferInfo.reduce.dst;
784 4 : tfType = Hccl::TransferType::READ_REDUCE;
785 4 : CHK_RET(ParasReduceData(transferDesc, len, dataType, reduceOp));
786 14 : } else if (transferDesc.transType == HCOMM_TRANSFER_TYPE_WRITE_WITH_NOTIFY) {
787 4 : rmt = transferDesc.transferInfo.writeWithNotify.dst; // write操作，dst是远端地址
788 4 : loc = transferDesc.transferInfo.writeWithNotify.src; // src是本端地址
789 4 : len = transferDesc.transferInfo.writeWithNotify.len;
790 4 : notifyIdx = transferDesc.transferInfo.writeWithNotify.notifyIdx;
791 4 : tfType = Hccl::TransferType::WRITE_WITH_NOTIFY;
792 10 : } else if (transferDesc.transType == HCOMM_TRANSFER_TYPE_WRITE_REDUCE_WITH_NOTIFY) {
793 5 : rmt = transferDesc.transferInfo.writeReduceWithNotify.dst;
794 5 : loc = transferDesc.transferInfo.writeReduceWithNotify.src;
795 5 : len = transferDesc.transferInfo.writeReduceWithNotify.count;
796 5 : dataType = transferDesc.transferInfo.writeReduceWithNotify.dataType;
797 5 : reduceOp = transferDesc.transferInfo.writeReduceWithNotify.reduceOp;
798 5 : notifyIdx = transferDesc.transferInfo.writeReduceWithNotify.notifyIdx;
799 5 : tfType = Hccl::TransferType::WRITE_REDUCE_WITH_NOTIFY;
800 5 : CHK_RET(CheckReduceHcommDataTypeAndHcommReduceOp(dataType, reduceOp));
801 5 : } else if (transferDesc.transType == HCOMM_TRANSFER_TYPE_NOTIFY_RECORD) {
802 4 : notifyIdx = transferDesc.transferInfo.notifyRecord.notifyIdx;
803 4 : tfType = Hccl::TransferType::NOTIFY_RECORD;
804 : } else {
805 1 : HCCL_ERROR("[%s] unsupported transType[%d]", __func__, transferDesc.transType);
806 1 : return HCCL_E_NOT_SUPPORT;
807 : }
808 33 : if (reduceOp != HcommReduceOp::HCOMM_REDUCE_RESERVED) { // 对于规约类型, size = count * sizeof(datatype)
809 13 : len = len * SIZE_TABLE[dataType];
810 : }
811 33 : return HCCL_SUCCESS;
812 : }
813 : constexpr uint32_t NOTIFYIDX_INVALID_VALUE = 0xFFFFFFFF; // NOTIFY idex非法值
814 12 : HcclResult UbTransportLiteImpl::ExecuteBatchTransfer(StreamLite *streamLitePtr,
815 : const HcommBatchTransferDesc *transferDescs, uint32_t transferDescNum)
816 : {
817 12 : std::vector<Hccl::RmaBufferLite> locSlices;
818 12 : std::vector<Hccl::Buffer> rmtSlices;
819 12 : std::vector<Hccl::BaseTransportLiteImpl::TransferOp> transferOps;
820 12 : std::vector<uint32_t> notifyIdxs;
821 :
822 12 : locSlices.reserve(transferDescNum);
823 12 : rmtSlices.reserve(transferDescNum);
824 12 : transferOps.reserve(transferDescNum);
825 12 : notifyIdxs.reserve(transferDescNum);
826 :
827 42 : for (uint32_t i = 0; i < transferDescNum; i++) {
828 34 : Hccl::RmaBufferLite locRmaBuf;
829 34 : void *rmt = nullptr;
830 34 : void *loc = nullptr;
831 34 : uint64_t len = 0;
832 34 : Hccl::TransferType tfType;
833 34 : HcommDataType dataType{HcommDataType::HCOMM_DATA_TYPE_RESERVED};
834 34 : HcommReduceOp reduceOp{HcommReduceOp::HCOMM_REDUCE_RESERVED};
835 34 : uint32_t notifyIdx = NOTIFYIDX_INVALID_VALUE;
836 37 : CHK_RET(ParseData(transferDescs[i], rmt, loc, len, tfType, dataType, reduceOp, notifyIdx));
837 33 : if (tfType != Hccl::TransferType::NOTIFY_RECORD) { // NOTIFY_RECORD时没有地址字段
838 29 : CHK_PTR_NULL(rmt);
839 28 : CHK_PTR_NULL(loc);
840 27 : HcclResult ret = BuildLocRmaBufferLite(reinterpret_cast<uintptr_t>(loc), len, locRmaBuf);
841 27 : CHK_PRT_RET(ret != HCCL_SUCCESS,
842 : HCCL_ERROR("[%s] FAIL at BuildLocRmaBufferLite for index %u. rmt[%p], loc[%p], len[0x%llx], tfType[%u], dataType[%d], reduceOp[%d].",
843 : __func__, i, rmt, loc, len, tfType, dataType, reduceOp), ret);
844 : }
845 58 : if (tfType == Hccl::TransferType::NOTIFY_RECORD || tfType == Hccl::TransferType::WRITE_WITH_NOTIFY
846 58 : || tfType == Hccl::TransferType::WRITE_REDUCE_WITH_NOTIFY) {
847 13 : CHK_PRT_RET(notifyIdx == NOTIFYIDX_INVALID_VALUE,
848 : HCCL_ERROR("[%s] FAIL at ParseData for index %u. tfType[%u], notifyIdx[%u].",
849 : __func__, i, tfType, notifyIdx), HCCL_E_PARA);
850 : }
851 30 : notifyIdxs.push_back(notifyIdx);
852 30 : locSlices.push_back(locRmaBuf);
853 :
854 30 : const Hccl::Buffer rmtBuf{reinterpret_cast<uintptr_t>(rmt), len};
855 30 : rmtSlices.push_back(rmtBuf);
856 :
857 30 : Hccl::ReduceIn reduceIn{mapHcommDataTypeA5.at(dataType), mapHcommReduceOpA5.at(reduceOp)};
858 :
859 30 : transferOps.push_back(Hccl::BaseTransportLiteImpl::TransferOp{tfType, reduceIn});
860 :
861 30 : HCCL_DEBUG("[%s] Prepared transfer op for index %u. rmt[%p], loc[%p], len[0x%llx], tfType[%u], dataType[%d], reduceOp[%d].",
862 : __func__, i, rmt, loc, len, tfType, dataType, reduceOp);
863 30 : }
864 8 : EXCEPTION_CATCH(BatchTransferAll(locSlices, rmtSlices, transferOps, notifyIdxs, *streamLitePtr), return HCCL_E_INTERNAL);
865 8 : return HCCL_SUCCESS;
866 12 : }
867 :
868 0 : void UbTransportLiteImpl::BatchTransferAll(const std::vector<RmaBufferLite> &loc, const std::vector<Buffer> &rmt,
869 : const std::vector<BaseTransportLiteImpl::TransferOp> &transferOp, const std::vector<uint32_t> ¬ifyIdxs, const StreamLite &stream)
870 : {
871 0 : if (UNLIKELY(loc.empty())) {
872 0 : return;
873 : }
874 0 : auto taskId = stream.GetRtsq()->GetTaskId();
875 0 : u64 notifyData = 1; // 普通notify，固定1，用于writeWithNotify与writeReduceWithNotify
876 0 : SqeConfigLite cfg;
877 0 : SetFenceConfig(cfg);
878 0 : u32 insNum = loc.size();
879 0 : for (u32 i = 0; i < insNum; i++) {
880 0 : cfg.cqeEn = (i == insNum - 1) ? true : false; // 返回最后一个sqe的cqe
881 0 : cfg.placeOdr = (i == insNum - 1) ? UB_STRONG_ORDER : UB_RELAX_ORDER; // 最后一个要求保序
882 0 : cfg.compOrder = (i == insNum - 1) ? UB_COMPLETION : UB_NO_COMPLETION;
883 :
884 0 : if (transferOp[i].transType == TransferType::NOTIFY_RECORD) { // notifyRecord操作没有loc/rmt，因此单独处理
885 0 : if (notifyIdxs[i] == 1) { // PostFin场景
886 0 : cfg.cqeEn = true;
887 0 : cfg.placeOdr = UB_STRONG_ORDER;
888 0 : cfg.compOrder = UB_COMPLETION;
889 : }
890 0 : u32 inlineData = 1;
891 : // 当前使用1个connection，下标为0 构建sqe
892 0 : connVec[0]->InlineWrite(reinterpret_cast<u8 *>(&inlineData), UB_INLINE_WRITE_SIZE, GetRmtNotifySliceLite(notifyIdxs[i]),
893 0 : cfg, stream, connOut);
894 : } else {
895 0 : auto localBuffer = GetRmaBufSlicelite(loc[i]);
896 0 : auto remoteBuffer = GetRmtRmaBufSliceLite(rmt[i]);
897 :
898 0 : if (transferOp[i].transType == TransferType::WRITE) {
899 0 : connVec[0]->Write(localBuffer, remoteBuffer, cfg, stream, connOut);
900 0 : } else if (transferOp[i].transType == TransferType::WRITE_REDUCE) {
901 0 : connVec[0]->WriteReduce(transferOp[i].reduceIn.dataType, transferOp[i].reduceIn.reduceOp, localBuffer, stream, remoteBuffer, cfg, connOut);
902 0 : } else if (transferOp[i].transType == TransferType::READ) {
903 0 : connVec[0]->Read(localBuffer, remoteBuffer, cfg, stream, connOut);
904 0 : } else if (transferOp[i].transType == TransferType::READ_REDUCE) {
905 0 : connVec[0]->ReadReduce(transferOp[i].reduceIn, localBuffer, remoteBuffer, stream, cfg, connOut);
906 0 : } else if (transferOp[i].transType == TransferType::WRITE_WITH_NOTIFY) {
907 0 : connVec[0]->WriteWithNotify(localBuffer, remoteBuffer, cfg, connOut, GetRmtNotifySliceLite(notifyIdxs[i]), stream, notifyData); // 当前使用1个connection，下标为0
908 0 : } else if (transferOp[i].transType == TransferType::WRITE_REDUCE_WITH_NOTIFY) {
909 0 : connVec[0]->WriteReduceWithNotify(transferOp[i].reduceIn.dataType, transferOp[i].reduceIn.reduceOp, localBuffer,
910 0 : remoteBuffer, cfg, stream, connOut, GetRmtNotifySliceLite(notifyIdxs[i]), notifyData); // 当前使用1个connection，下标为0
911 : }
912 : }
913 : }
914 0 : BuildUbDbSendTask(stream, connVec[0]->GetUbJettyLiteId(), connOut.pi); // 约束使用一批wqe的个数不会导致反压
915 :
916 0 : ExecProfilingAll(loc, rmt, transferOp, stream, taskId, notifyIdxs);
917 : }
918 :
919 0 : void UbTransportLiteImpl::Drain(const StreamLite &stream)
920 : {
921 0 : std::lock_guard<std::mutex> lock(drainMtx_);
922 0 : if (drainNotify_.size == 0 || rmtDrainBuffer_.size == 0) {
923 0 : HCCL_WARNING("[UbTransportLiteImpl::%s] drain resource is null skip", __func__);
924 0 : return;
925 : }
926 :
927 0 : SqeConfigLite cfg;
928 0 : Fence();
929 0 : SetFenceConfig(cfg);
930 :
931 : // 当前使用1个connection,下标为0
932 : auto drainNotifyBufSlice = RmaBufSliceLite(drainNotify_.addr, drainNotify_.size, 0,
933 0 : drainNotify_.tokenId);
934 : auto drainConstBufSlice = RmtRmaBufSliceLite(rmtDrainBuffer_.addr, rmtDrainBuffer_.size, 0,
935 0 : rmtDrainBuffer_.tokenId, rmtDrainBuffer_.tokenValue, UINT32_MAX);
936 0 : connVec[0]->Read(drainNotifyBufSlice, drainConstBufSlice, cfg, stream, connOut);
937 0 : BuildUbDbSendTask(stream, connVec[0]->GetUbJettyLiteId(), connOut.pi);
938 :
939 0 : BuildNotifyWaitTask(stream, drainNotify_.notifyId);
940 0 : }
941 :
942 1 : void UbTransportLiteImpl::WriteWithNotify(const RmaBufferLite &loc, const Buffer &rmt, const WithNotifyIn &withNotify,
943 : const StreamLite &stream)
944 : {
945 1 : SqeConfigLite cfg;
946 1 : SetFenceConfig(cfg);
947 1 : u64 notifyData = 1; // 普通notify，固定1
948 1 : auto taskId = stream.GetRtsq()->GetTaskId();
949 :
950 : // 当前使用1个connection，下标为0
951 1 : auto locRmaBufSlicelite = GetRmaBufSlicelite(loc);
952 1 : auto rmtRmaBufSlicelite = GetRmtRmaBufSliceLite(rmt);
953 1 : auto rmtNotifySliceLite = GetRmtNotifySliceLite(withNotify.index_);
954 1 : connVec[0]->WriteWithNotify(locRmaBufSlicelite, rmtRmaBufSlicelite, cfg, connOut,
955 : rmtNotifySliceLite, stream, notifyData);
956 1 : BuildUbDbSendTask(stream, connVec[0]->GetUbJettyLiteId(), connOut.pi);
957 :
958 1 : if (!IsReportTask()) {
959 0 : return;
960 : }
961 :
962 1 : TaskParam taskParam{};
963 1 : taskParam.taskType = TaskParamType::TASK_WRITE_WITH_NOTIFY;
964 1 : taskParam.beginTime = ProfGetCurCpuTimestamp();
965 1 : taskParam.taskPara.DMA.src = reinterpret_cast<void *>(locRmaBufSlicelite.GetAddr());
966 1 : taskParam.taskPara.DMA.dst = reinterpret_cast<void *>(rmtRmaBufSlicelite.GetAddr());
967 1 : taskParam.taskPara.DMA.size = locRmaBufSlicelite.GetSize();
968 1 : taskParam.taskPara.DMA.notifyID = rmtNotifySliceLite.GetNotifyId();
969 1 : taskParam.taskPara.DMA.notifyValue = 1;
970 1 : taskParam.taskPara.DMA.linkType = DfxLinkType::UB;
971 1 : taskParam.taskPara.DMA.dmaOp = DmaOp::HCCL_DMA_WRITE;
972 1 : taskParam.taskPara.DMA.locEid = GetLocEid();
973 1 : taskParam.taskPara.DMA.rmtEid = GetRmtEid();
974 :
975 1 : AddTaskCallback(stream, taskId, taskParam);
976 1 : }
977 :
978 0 : void UbTransportLiteImpl::WriteReduceWithNotify(const RmaBufferLite &loc, const Buffer &rmt, const ReduceIn &reduceIn,
979 : const WithNotifyIn &withNotify, const StreamLite &stream)
980 : {
981 0 : SqeConfigLite cfg;
982 0 : SetFenceConfig(cfg);
983 0 : u64 notifyData = 1; // 普通notify，固定1
984 0 : auto taskId = stream.GetRtsq()->GetTaskId();
985 :
986 : // 当前使用1个connection，下标为0
987 0 : auto locRmaBufSlicelite = GetRmaBufSlicelite(loc);
988 0 : auto rmtRmaBufSlicelite = GetRmtRmaBufSliceLite(rmt);
989 0 : auto rmtNotifySliceLite = GetRmtNotifySliceLite(withNotify.index_);
990 0 : connVec[0]->WriteReduceWithNotify(reduceIn.dataType, reduceIn.reduceOp, locRmaBufSlicelite,
991 0 : rmtRmaBufSlicelite, cfg, stream, connOut, rmtNotifySliceLite,
992 : notifyData);
993 0 : BuildUbDbSendTask(stream, connVec[0]->GetUbJettyLiteId(), connOut.pi);
994 0 : if (!IsReportTask()) {
995 0 : return;
996 : }
997 :
998 0 : TaskParam taskParam{};
999 0 : taskParam.taskType = TaskParamType::TASK_WRITE_REDUCE_WITH_NOTIFY;
1000 0 : taskParam.beginTime = ProfGetCurCpuTimestamp();
1001 0 : taskParam.taskPara.Reduce.src = reinterpret_cast<void *>(locRmaBufSlicelite.GetAddr());
1002 0 : taskParam.taskPara.Reduce.dst = reinterpret_cast<void *>(rmtRmaBufSlicelite.GetAddr());
1003 0 : taskParam.taskPara.Reduce.size = locRmaBufSlicelite.GetSize();
1004 0 : taskParam.taskPara.Reduce.notifyID = rmtNotifySliceLite.GetNotifyId();
1005 0 : taskParam.taskPara.Reduce.notifyValue = 1;
1006 0 : taskParam.taskPara.Reduce.linkType = DfxLinkType::UB;
1007 0 : taskParam.taskPara.Reduce.reduceOp = ConvertReduceOpToHcclReduceOp(reduceIn.reduceOp);
1008 0 : taskParam.taskPara.Reduce.dataType = DataTypeToHcclDataType(reduceIn.dataType);
1009 0 : taskParam.taskPara.Reduce.locEid = GetLocEid();
1010 0 : taskParam.taskPara.Reduce.rmtEid = GetRmtEid();
1011 :
1012 0 : AddTaskCallback(stream, taskId, taskParam);
1013 0 : }
1014 :
1015 1 : void UbTransportLiteImpl::BatchOneSidedRead(const vector<RmaBufSliceLite> &loc, const vector<RmtRmaBufSliceLite> &rmt,
1016 : const StreamLite &stream)
1017 : {
1018 1 : SqeConfigLite cfg;
1019 1 : SetFenceConfig(cfg);
1020 :
1021 : // 当前使用1个connection，下标为0
1022 1 : connVec[0]->BatchOneSidedRead(loc, rmt, cfg, stream, connOut);
1023 1 : BuildUbDbSendTask(stream, connVec[0]->GetUbJettyLiteId(), connOut.pi);
1024 1 : }
1025 :
1026 1 : void UbTransportLiteImpl::BatchOneSidedWrite(const vector<RmaBufSliceLite> &loc, const vector<RmtRmaBufSliceLite> &rmt,
1027 : const StreamLite &stream)
1028 : {
1029 1 : SqeConfigLite cfg;
1030 1 : SetFenceConfig(cfg);
1031 :
1032 : // 当前使用1个connection，下标为0
1033 1 : connVec[0]->BatchOneSidedWrite(loc, rmt, cfg, stream, connOut);
1034 1 : BuildUbDbSendTask(stream, connVec[0]->GetUbJettyLiteId(), connOut.pi);
1035 1 : }
1036 :
1037 :
1038 8 : Eid UbTransportLiteImpl::GetLocEid() const
1039 : {
1040 8 : return connVec[0]->GetLocEid();
1041 : }
1042 :
1043 8 : Eid UbTransportLiteImpl::GetRmtEid() const
1044 : {
1045 8 : return connVec[0]->GetRmtEid();
1046 : }
1047 :
1048 0 : HcclResult UbTransportLiteImpl::Clean()
1049 : {
1050 0 : locNotifyVec.clear();
1051 0 : rmtNotifyVec.clear();
1052 0 : locBufferMap.clear();
1053 0 : rmtBufferVec.clear();
1054 0 : rmtBufferMap.clear();
1055 :
1056 : // 清理connVec，connLite由UbConnLiteMgr管理
1057 0 : for (auto &it : connUniqueIdVec) {
1058 0 : DECTOR_TRY_CATCH("UbTransportLiteImpl", UbConnLiteMgr::GetInstance().Clear(it));
1059 : }
1060 0 : connUniqueIdVec.clear();
1061 0 : connVec.clear();
1062 :
1063 0 : return HCCL_SUCCESS;
1064 : }
1065 :
1066 0 : HcclResult UbTransportLiteImpl::Resume(std::vector<char> &uniqueId)
1067 : {
1068 0 : Init(uniqueId);
1069 0 : return HCCL_SUCCESS;
1070 : }
1071 :
1072 1 : HcclResult UbTransportLiteImpl::Fence()
1073 : {
1074 1 : fence_ = true;
1075 1 : HCCL_INFO("[%s] SUCCESS. fence[%d]", __func__, fence_);
1076 1 : return HCCL_SUCCESS;
1077 : }
1078 :
1079 8 : void UbTransportLiteImpl::SetFenceConfig(SqeConfigLite &cfg)
1080 : {
1081 8 : if (fence_) {
1082 0 : cfg.fence = UB_FENCE_ENABLED;
1083 0 : cfg.placeOdr = UB_STRONG_ORDER;
1084 0 : cfg.compOrder = UB_COMPLETION;
1085 : }
1086 8 : fence_ = false;
1087 8 : }
1088 :
1089 11 : bool UbTransportLiteImpl::IsReportTask()
1090 : {
1091 22 : return (taskExceptionEnable_ || ProfilingHandlerLite::GetInstance().GetProfL1State()) &&
1092 22 : (callback_ != nullptr || newCallback_ != nullptr);
1093 : }
1094 : } // namespace Hccl
|
