Line data Source code
1 : /**
2 : * Copyright (c) 2026 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 "roce_transport_lite_impl.h"
12 : #include "binary_stream.h"
13 : #include "log.h"
14 : #include "dfx_profiling_handler_lite.h"
15 : #include "sal.h"
16 :
17 : namespace Hccl {
18 :
19 29 : RoceTransportLiteImpl::RoceTransportLiteImpl(std::vector<char>& uniqueId) { Init(uniqueId); }
20 :
21 31 : RoceTransportLiteImpl::~RoceTransportLiteImpl() {}
22 :
23 29 : void RoceTransportLiteImpl::Init(std::vector<char>& uniqueId)
24 : {
25 29 : BinaryStream binaryStream(uniqueId);
26 : u32 type;
27 29 : binaryStream >> type;
28 29 : binaryStream >> notifyNum_;
29 29 : binaryStream >> bufferNum_;
30 29 : binaryStream >> connNum_;
31 :
32 29 : std::vector<char> locNotifyUniqueIds;
33 29 : binaryStream >> locNotifyUniqueIds;
34 29 : ParseLocNotifyVec(locNotifyUniqueIds);
35 :
36 29 : std::vector<char> rmtNotifyUniqueIds;
37 29 : binaryStream >> rmtNotifyUniqueIds;
38 29 : ParseRmtNotifyVec(rmtNotifyUniqueIds);
39 :
40 29 : std::vector<char> notifyValueBufferUniqueIds;
41 29 : binaryStream >> notifyValueBufferUniqueIds;
42 29 : ParseNotifyValueBuffer(notifyValueBufferUniqueIds);
43 :
44 29 : std::vector<char> locBufferUniqueIds;
45 29 : binaryStream >> locBufferUniqueIds;
46 29 : ParseLocBufferVec(locBufferUniqueIds);
47 :
48 29 : std::vector<char> rmtBufferUniqueIds;
49 29 : binaryStream >> rmtBufferUniqueIds;
50 29 : ParseRmtBufferVec(rmtBufferUniqueIds);
51 :
52 29 : std::vector<char> connUniqueIds;
53 29 : binaryStream >> connUniqueIds;
54 29 : ParseConnVec(connUniqueIds);
55 29 : }
56 :
57 29 : void RoceTransportLiteImpl::ParseLocNotifyVec(std::vector<char>& data)
58 : {
59 29 : if (notifyNum_ == 0) {
60 0 : HCCL_WARNING("[RoceTransportLiteImpl::%s] notifyNum is 0", __func__);
61 0 : return;
62 : }
63 :
64 29 : u32 notifySizePerDto = data.size() / notifyNum_;
65 :
66 87 : for (u32 idx = 0; idx < notifyNum_; idx++) {
67 58 : auto start = data.begin() + idx * notifySizePerDto;
68 58 : auto end = start + notifySizePerDto;
69 58 : std::vector<char> dto(start, end);
70 58 : localNotifies_.push_back(std::make_unique<NotifyLite>(dto));
71 166 : HCCL_INFO("locNotify idx=%u, %s", idx, localNotifies_.back()->Describe().c_str());
72 58 : }
73 : }
74 :
75 29 : void RoceTransportLiteImpl::ParseRmtNotifyVec(std::vector<char>& data)
76 : {
77 29 : if (notifyNum_ == 0) {
78 0 : HCCL_WARNING("[RoceTransportLiteImpl::%s] notifyNum is 0", __func__);
79 0 : return;
80 : }
81 :
82 29 : u32 rmtBufferSizePerDto = data.size() / notifyNum_;
83 83 : HCCL_INFO(
84 : "[RoceTransportLiteImpl::%s] Parse remote notify num=%u, sizePerDto=%u", __func__, notifyNum_,
85 : rmtBufferSizePerDto);
86 :
87 29 : BinaryStream binaryStream(data);
88 29 : remoteNotifies_.clear();
89 : u64 addr;
90 : u64 size;
91 : u32 rkey;
92 87 : for (u32 idx = 0; idx < notifyNum_; idx++) {
93 58 : binaryStream >> addr;
94 58 : binaryStream >> size;
95 58 : binaryStream >> rkey;
96 58 : RmtRmaBufferLite rdmaBufLite(addr, size, rkey);
97 166 : HCCL_INFO("idx=%u, %s", idx, rdmaBufLite.Describe().c_str());
98 58 : remoteNotifies_.emplace_back(rdmaBufLite);
99 : }
100 29 : }
101 :
102 29 : void RoceTransportLiteImpl::ParseNotifyValueBuffer(std::vector<char>& data)
103 : {
104 83 : HCCL_INFO("[RoceTransportLiteImpl::%s] Parse notify value buffer", __func__);
105 :
106 29 : BinaryStream binaryStream(data);
107 : u64 addr;
108 : u64 size;
109 : u32 lkey;
110 29 : binaryStream >> addr;
111 29 : binaryStream >> size;
112 29 : binaryStream >> lkey;
113 29 : notifyValueBuffer_ = std::make_unique<RmaBufferLite>(addr, size, lkey);
114 29 : }
115 :
116 29 : void RoceTransportLiteImpl::ParseLocBufferVec(std::vector<char>& data)
117 : {
118 29 : if (bufferNum_ == 0) {
119 0 : HCCL_WARNING("[RoceTransportLiteImpl::%s] bufferNum is 0", __func__);
120 0 : return;
121 : }
122 :
123 29 : u32 locBufferSizePerDto = data.size() / bufferNum_;
124 83 : HCCL_INFO(
125 : "[RoceTransportLiteImpl::%s] Parse local buffer num=%u, sizePerDto=%u", __func__, bufferNum_,
126 : locBufferSizePerDto);
127 :
128 29 : BinaryStream binaryStream(data);
129 29 : locBufferVec_.clear();
130 : u64 addr;
131 : u64 size;
132 : u32 lkey;
133 58 : for (u32 idx = 0; idx < bufferNum_; idx++) {
134 29 : binaryStream >> addr;
135 29 : binaryStream >> size;
136 29 : binaryStream >> lkey;
137 29 : RmaBufferLite rdmaBufLite(addr, size, lkey);
138 83 : HCCL_INFO("idx=%u, %s", idx, rdmaBufLite.Describe().c_str());
139 29 : locBufferVec_.emplace_back(rdmaBufLite);
140 : }
141 29 : }
142 :
143 29 : void RoceTransportLiteImpl::ParseRmtBufferVec(std::vector<char>& data)
144 : {
145 29 : if (bufferNum_ == 0) {
146 0 : HCCL_WARNING("[RoceTransportLiteImpl::%s] bufferNum is 0", __func__);
147 0 : return;
148 : }
149 :
150 29 : u32 rmtBufferSizePerDto = data.size() / bufferNum_;
151 83 : HCCL_INFO(
152 : "[RoceTransportLiteImpl::%s] Parse remote buffer num=%u, sizePerDto=%u", __func__, bufferNum_,
153 : rmtBufferSizePerDto);
154 :
155 29 : BinaryStream binaryStream(data);
156 29 : rmtBufferVec_.clear();
157 : u64 addr;
158 : u64 size;
159 : u32 rkey;
160 58 : for (u32 idx = 0; idx < bufferNum_; idx++) {
161 29 : binaryStream >> addr;
162 29 : binaryStream >> size;
163 29 : binaryStream >> rkey;
164 29 : RmtRmaBufferLite rdmaBufLite(addr, size, rkey);
165 83 : HCCL_INFO("idx=%u, %s", idx, rdmaBufLite.Describe().c_str());
166 29 : rmtBufferVec_.emplace_back(rdmaBufLite);
167 : }
168 29 : }
169 :
170 29 : void RoceTransportLiteImpl::ParseConnVec(std::vector<char>& data)
171 : {
172 29 : if (connNum_ == 0) {
173 0 : HCCL_WARNING("[RoceTransportLiteImpl::%s] connNum is 0", __func__);
174 0 : return;
175 : }
176 :
177 29 : u32 connSizePerDto = data.size() / connNum_;
178 83 : HCCL_INFO("[RoceTransportLiteImpl::%s] Parse conn num=%u, sizePerDto=%u", __func__, connNum_, connSizePerDto);
179 58 : for (u32 idx = 0; idx < connNum_; idx++) {
180 29 : auto start = data.begin() + idx * connSizePerDto;
181 29 : auto end = start + connSizePerDto;
182 29 : std::vector<char> connUniqueId(start, end);
183 29 : connUniqueIdVec_.emplace_back(connUniqueId);
184 29 : std::unique_ptr<RdmaConnLiteV2> connLite;
185 29 : connLite = std::make_unique<RdmaConnLiteV2>(connUniqueId);
186 83 : HCCL_INFO("[RoceTransportLiteImpl::%s] idx=%u, %s", __func__, idx, connLite->Describe().c_str());
187 29 : connVec_.emplace_back(std::move(connLite));
188 29 : }
189 : }
190 :
191 5 : RmaBufSliceLite RoceTransportLiteImpl::GetRmaBufSlicelite(const RmaBufferLite& lite) const
192 : {
193 5 : return RmaBufSliceLite(lite.GetAddr(), lite.GetSize(), lite.GetLkey(), 0);
194 : }
195 :
196 3 : RmaBufSliceLite RoceTransportLiteImpl::GetNotifySlicelite(u32 index) const
197 : {
198 : (void)index;
199 : return RmaBufSliceLite(
200 3 : notifyValueBuffer_->GetAddr(), notifyValueBuffer_->GetSize(), notifyValueBuffer_->GetLkey(), 0);
201 : }
202 :
203 5 : RmtRmaBufSliceLite RoceTransportLiteImpl::GetRmtRmaBufSliceLite(const Buffer& rmtBuf) const
204 : {
205 5 : for (auto& it : rmtBufferVec_) {
206 5 : Buffer buf(it.GetAddr(), it.GetSize());
207 5 : if (buf.Contains(rmtBuf.GetAddr(), rmtBuf.GetSize())) {
208 10 : return RmtRmaBufSliceLite(rmtBuf.GetAddr(), rmtBuf.GetSize(), it.GetRkey(), 0, 0, UINT32_MAX);
209 : }
210 5 : }
211 0 : MACRO_THROW(InternalException, StringFormat("%s is not in current transport", rmtBuf.Describe().c_str()));
212 : }
213 :
214 3 : RmtRmaBufSliceLite RoceTransportLiteImpl::GetRmtNotifySliceLite(u32 index) const
215 : {
216 3 : auto& lite = remoteNotifies_[index];
217 3 : return RmtRmaBufSliceLite(lite.GetAddr(), lite.GetSize(), lite.GetRkey(), 0, 0, UINT32_MAX);
218 : }
219 :
220 2 : std::string RoceTransportLiteImpl::Describe() const
221 : {
222 2 : std::string desc = "RoceTransportLiteImpl[";
223 :
224 2 : u32 idx = 0;
225 2 : desc += "localNotifies=[";
226 6 : for (auto& it : localNotifies_) {
227 4 : desc += StringFormat("idx=%u, %s;", idx, it->Describe().c_str());
228 4 : idx++;
229 : }
230 :
231 2 : idx = 0;
232 2 : desc += "], remoteNotifies=[";
233 6 : for (auto& it : remoteNotifies_) {
234 4 : desc += StringFormat("idx=%u, %s;", idx, it.Describe().c_str());
235 4 : idx++;
236 : }
237 :
238 2 : idx = 0;
239 2 : desc += "], locBufferVec=[";
240 4 : for (auto& it : locBufferVec_) {
241 2 : desc += StringFormat("idx=%u, %s;", idx, it.Describe().c_str());
242 2 : idx++;
243 : }
244 :
245 2 : idx = 0;
246 2 : desc += "], rmtBufferVec=[";
247 4 : for (auto& it : rmtBufferVec_) {
248 2 : desc += StringFormat("idx=%u, %s;", idx, it.Describe().c_str());
249 2 : idx++;
250 : }
251 :
252 2 : idx = 0;
253 2 : desc += "], connVec=[";
254 4 : for (auto& it : connVec_) {
255 2 : desc += StringFormat("idx=%u, %s;", idx, it->Describe().c_str());
256 2 : idx++;
257 : }
258 :
259 2 : desc += "]]";
260 2 : return desc;
261 0 : }
262 :
263 : HcclResult
264 0 : RoceTransportLiteImpl::BuildLocRmaBufferLite(const uintptr_t addr, const size_t size, RmaBufferLite& rmaBufferLite)
265 : {
266 0 : HCCL_INFO(
267 : "[RoceTransportLiteImpl::%s] start to find addr[0x%llx], size[0x%llx] in locBufferVec, whose size is %zu. ",
268 : __func__, addr, size, locBufferVec_.size());
269 :
270 0 : if (locBufferVec_.empty()) {
271 0 : HCCL_ERROR("[RoceTransportLiteImpl::%s] locBufferVec is empty.", __func__);
272 0 : return HCCL_E_INTERNAL;
273 : }
274 :
275 0 : bool isAddrInRange = false;
276 0 : for (auto& it : locBufferVec_) {
277 0 : Buffer iterBuf(it.GetAddr(), it.GetSize());
278 0 : if (iterBuf.Contains(addr, size)) {
279 0 : rmaBufferLite = RmaBufferLite(addr, size, it.GetLkey());
280 0 : isAddrInRange = true;
281 0 : break;
282 : }
283 0 : }
284 :
285 0 : if (!isAddrInRange) {
286 0 : HCCL_WARNING(
287 : "[RoceTransportLiteImpl::%s] addr[0x%llx], size[0x%llx] not in any range of locBufferVec. The token of the "
288 : "first locBuffer is used.",
289 : __func__, addr, size);
290 0 : rmaBufferLite = RmaBufferLite(addr, size, locBufferVec_[0].GetLkey());
291 0 : return HCCL_SUCCESS;
292 : }
293 :
294 0 : return HCCL_SUCCESS;
295 : }
296 :
297 1 : void RoceTransportLiteImpl::Read(const RmaBufferLite& loc, const Buffer& rmt, const StreamLite& stream)
298 : {
299 1 : u64 dbAddr = 0;
300 1 : u64 dbValue = 0;
301 : // 获取Profiling任务ID
302 1 : auto taskId = stream.GetRtsq()->GetTaskId();
303 :
304 : // 获取本端和远端Buffer切片
305 1 : SqeConfigLite cfg;
306 1 : SetFenceConfig(cfg);
307 1 : auto locRmaBufSliceLite = GetRmaBufSlicelite(loc);
308 1 : auto rmtRmaBufSliceLite = GetRmtRmaBufSliceLite(rmt);
309 :
310 : // Post Wqe && return dbValue
311 1 : connVec_[0]->Read(locRmaBufSliceLite, rmtRmaBufSliceLite, cfg, dbAddr, dbValue);
312 :
313 : // Ring Doorbell
314 1 : BuildRdmaDbSendTask(stream, dbAddr, dbValue);
315 :
316 : // 上报Profiling任务
317 2 : ReportDmaTask(
318 1 : reinterpret_cast<const void*>(locRmaBufSliceLite.GetAddr()),
319 1 : reinterpret_cast<const void*>(rmtRmaBufSliceLite.GetAddr()), locRmaBufSliceLite.GetSize(), stream, taskId,
320 : TaskParamType::TASK_RDMA, DmaOp::HCCL_DMA_READ, INVALID_VALUE_NOTIFYID, UINT32_MAX, __func__);
321 :
322 : // Poll Cq
323 1 : constexpr int32_t POLL_NUM = 1; // poll cqe num
324 1 : constexpr int32_t POLL_TIMEOUT = 5; // 5 ms
325 1 : std::vector<int32_t> errList = {};
326 1 : connVec_[0]->PollCq(POLL_NUM, POLL_TIMEOUT, errList, dbAddr, dbValue);
327 1 : }
328 :
329 1 : void RoceTransportLiteImpl::Write(const RmaBufferLite& loc, const Buffer& rmt, const StreamLite& stream)
330 : {
331 1 : u64 dbAddr = 0;
332 1 : u64 dbValue = 0;
333 : // 获取Profiling任务ID
334 1 : auto taskId = stream.GetRtsq()->GetTaskId();
335 :
336 : // 获取本端和远端Buffer切片
337 1 : auto locRmaBufSliceLite = GetRmaBufSlicelite(loc);
338 1 : auto rmtRmaBufSliceLite = GetRmtRmaBufSliceLite(rmt);
339 1 : SqeConfigLite cfg;
340 1 : SetFenceConfig(cfg);
341 :
342 : // Post Wqe && return dbValue
343 1 : connVec_[0]->Write(locRmaBufSliceLite, rmtRmaBufSliceLite, cfg, dbAddr, dbValue);
344 :
345 : // Ring Doorbell
346 1 : BuildRdmaDbSendTask(stream, dbAddr, dbValue);
347 :
348 : // 上报Profiling任务
349 2 : ReportDmaTask(
350 1 : reinterpret_cast<const void*>(locRmaBufSliceLite.GetAddr()),
351 1 : reinterpret_cast<const void*>(rmtRmaBufSliceLite.GetAddr()), locRmaBufSliceLite.GetSize(), stream, taskId,
352 : TaskParamType::TASK_RDMA, DmaOp::HCCL_DMA_WRITE, INVALID_VALUE_NOTIFYID, UINT32_MAX, __func__);
353 :
354 : // Poll Cq
355 1 : constexpr int32_t POLL_NUM = 1; // poll cqe num
356 1 : constexpr int32_t POLL_TIMEOUT = 5; // 5 ms
357 1 : std::vector<int32_t> errList = {};
358 1 : connVec_[0]->PollCq(POLL_NUM, POLL_TIMEOUT, errList, dbAddr, dbValue);
359 1 : }
360 :
361 1 : void RoceTransportLiteImpl::WriteReduce(
362 : const RmaBufferLite& loc, const Buffer& rmt, const ReduceIn& reduceIn, const StreamLite& stream)
363 : {
364 1 : u64 dbAddr = 0;
365 1 : u64 dbValue = 0;
366 1 : auto taskId = stream.GetRtsq()->GetTaskId();
367 :
368 1 : SqeConfigLite cfg;
369 1 : SetFenceConfig(cfg);
370 1 : auto locRmaBufSliceLite = GetRmaBufSlicelite(loc);
371 1 : auto rmtRmaBufSliceLite = GetRmtRmaBufSliceLite(rmt);
372 :
373 : // Post Wqe && return dbValue
374 1 : connVec_[0]->WriteReduce(
375 : locRmaBufSliceLite, rmtRmaBufSliceLite, cfg, reduceIn.dataType, reduceIn.reduceOp, dbAddr, dbValue);
376 :
377 : // Ring Doorbell
378 1 : BuildRdmaDbSendTask(stream, dbAddr, dbValue);
379 :
380 : // 上报Profiling任务
381 2 : ReportReduceTask(
382 1 : reinterpret_cast<const void*>(locRmaBufSliceLite.GetAddr()),
383 1 : reinterpret_cast<const void*>(rmtRmaBufSliceLite.GetAddr()), locRmaBufSliceLite.GetSize(), reduceIn, stream,
384 : taskId, TaskParamType::TASK_REDUCE_INLINE, INVALID_VALUE_NOTIFYID, UINT32_MAX, __func__);
385 :
386 : // Poll Cq
387 1 : constexpr int32_t POLL_NUM = 1; // poll cqe num
388 1 : constexpr int32_t POLL_TIMEOUT = 5; // 5 ms
389 1 : std::vector<int32_t> errList = {};
390 1 : connVec_[0]->PollCq(POLL_NUM, POLL_TIMEOUT, errList, dbAddr, dbValue);
391 1 : }
392 :
393 1 : void RoceTransportLiteImpl::WriteWithNotify(
394 : const RmaBufferLite& loc, const Buffer& rmt, const WithNotifyIn& withNotify, const StreamLite& stream)
395 : {
396 1 : auto taskId = stream.GetRtsq()->GetTaskId();
397 1 : u64 dbAddr = 0;
398 1 : u64 dbValue = 0;
399 :
400 1 : SqeConfigLite cfg;
401 1 : SetFenceConfig(cfg);
402 1 : auto locRmaBufSliceLite = GetRmaBufSlicelite(loc);
403 1 : auto rmtRmaBufSliceLite = GetRmtRmaBufSliceLite(rmt);
404 1 : auto locNotifySliceLite = GetNotifySlicelite(withNotify.index_); // 普通Notify
405 1 : auto rmtNotifySliceLite = GetRmtNotifySliceLite(withNotify.index_);
406 :
407 : // Post Wqe && return dbValue
408 1 : connVec_[0]->WriteWithNotify(
409 : locRmaBufSliceLite, rmtRmaBufSliceLite, locNotifySliceLite, rmtNotifySliceLite, cfg, dbAddr, dbValue);
410 :
411 : // Ring Doorbell
412 1 : BuildRdmaDbSendTask(stream, dbAddr, dbValue);
413 :
414 : // 上报Profiling任务
415 2 : ReportDmaTask(
416 1 : reinterpret_cast<const void*>(locRmaBufSliceLite.GetAddr()),
417 1 : reinterpret_cast<const void*>(rmtRmaBufSliceLite.GetAddr()), locRmaBufSliceLite.GetSize(), stream, taskId,
418 1 : TaskParamType::TASK_WRITE_WITH_NOTIFY, DmaOp::HCCL_DMA_WRITE, rmtNotifySliceLite.GetNotifyId(), 1, __func__);
419 :
420 : // Poll Cq
421 1 : constexpr int32_t POLL_NUM = 2; // poll cqe num
422 1 : constexpr int32_t POLL_TIMEOUT = 5; // 5 ms
423 1 : std::vector<int32_t> errList = {};
424 1 : connVec_[0]->PollCq(POLL_NUM, POLL_TIMEOUT, errList, dbAddr, dbValue);
425 1 : }
426 :
427 1 : void RoceTransportLiteImpl::WriteReduceWithNotify(
428 : const RmaBufferLite& loc, const Buffer& rmt, const ReduceIn& reduceIn, const WithNotifyIn& withNotify,
429 : const StreamLite& stream)
430 : {
431 1 : u64 dbAddr = 0;
432 1 : u64 dbValue = 0;
433 1 : auto taskId = stream.GetRtsq()->GetTaskId();
434 :
435 1 : auto locRmaBufSliceLite = GetRmaBufSlicelite(loc);
436 1 : auto rmtRmaBufSliceLite = GetRmtRmaBufSliceLite(rmt);
437 1 : auto locNotifySliceLite = GetNotifySlicelite(withNotify.index_); // 普通Notify
438 1 : auto rmtNotifySliceLite = GetRmtNotifySliceLite(withNotify.index_);
439 1 : SqeConfigLite cfg;
440 1 : SetFenceConfig(cfg);
441 :
442 : // Post Wqe && return dbValue
443 1 : connVec_[0]->WriteReduceWithNotify(
444 : locRmaBufSliceLite, rmtRmaBufSliceLite, locNotifySliceLite, rmtNotifySliceLite, cfg, reduceIn.dataType,
445 : reduceIn.reduceOp, dbAddr, dbValue);
446 :
447 : // Ring Doorbell
448 1 : BuildRdmaDbSendTask(stream, dbAddr, dbValue);
449 :
450 : // 上报Profiling任务
451 2 : ReportReduceTask(
452 1 : reinterpret_cast<const void*>(locRmaBufSliceLite.GetAddr()),
453 1 : reinterpret_cast<const void*>(rmtRmaBufSliceLite.GetAddr()), locRmaBufSliceLite.GetSize(), reduceIn, stream,
454 1 : taskId, TaskParamType::TASK_WRITE_REDUCE_WITH_NOTIFY, rmtNotifySliceLite.GetNotifyId(), 1, __func__);
455 :
456 : // Poll Cq
457 1 : constexpr int32_t POLL_NUM = 2; // poll cqe num
458 1 : constexpr int32_t POLL_TIMEOUT = 5; // 5 ms
459 1 : std::vector<int32_t> errList = {};
460 1 : connVec_[0]->PollCq(POLL_NUM, POLL_TIMEOUT, errList, dbAddr, dbValue);
461 1 : }
462 :
463 1 : HcclResult RoceTransportLiteImpl::Fence()
464 : {
465 1 : fence_ = true;
466 3 : HCCL_INFO("[%s] SUCCESS. fence[%d]", __func__, fence_);
467 1 : return HCCL_SUCCESS;
468 : }
469 :
470 1 : void RoceTransportLiteImpl::Post(u32 index, const StreamLite& stream)
471 : {
472 1 : u64 dbAddr = 0;
473 1 : u64 dbValue = 0;
474 1 : auto taskId = stream.GetRtsq()->GetTaskId();
475 :
476 1 : SqeConfigLite cfg;
477 1 : SetFenceConfig(cfg);
478 1 : auto locNotifySliceLite = GetNotifySlicelite(index);
479 1 : auto rmtNotifySliceLite = GetRmtNotifySliceLite(index);
480 :
481 : // Post Wqe && return dbValue
482 1 : connVec_[0]->Write(locNotifySliceLite, rmtNotifySliceLite, cfg, dbAddr, dbValue);
483 :
484 : // Ring Doorbell
485 1 : BuildRdmaDbSendTask(stream, dbAddr, dbValue);
486 :
487 : // 上报Profiling任务
488 2 : ReportDmaTask(
489 1 : reinterpret_cast<const void*>(locNotifySliceLite.GetAddr()),
490 1 : reinterpret_cast<const void*>(rmtNotifySliceLite.GetAddr()), locNotifySliceLite.GetSize(), stream, taskId,
491 1 : TaskParamType::TASK_RDMA, DmaOp::HCCL_DMA_WRITE, rmtNotifySliceLite.GetNotifyId(), 1, __func__);
492 :
493 : // Poll Cq
494 1 : constexpr int32_t POLL_NUM = 1; // poll cqe num
495 1 : constexpr int32_t POLL_TIMEOUT = 5; // 5 ms
496 1 : std::vector<int32_t> errList = {};
497 1 : connVec_[0]->PollCq(POLL_NUM, POLL_TIMEOUT, errList, dbAddr, dbValue);
498 1 : }
499 :
500 1 : HcclResult RoceTransportLiteImpl::PollCq(int32_t numEntries, int32_t timeOut, std::vector<int32_t>& errList)
501 : {
502 1 : u64 dbAddr = 0;
503 1 : u64 cqDbValue = 0;
504 1 : HcclResult ret = HCCL_SUCCESS;
505 :
506 : // Poll numEntries个Cqe, 返回异常的status, 同时返回cq的db
507 1 : ret = connVec_[0]->PollCq(numEntries, timeOut, errList, dbAddr, cqDbValue);
508 :
509 1 : return ret;
510 : }
511 :
512 1 : void RoceTransportLiteImpl::WaitWithTimeout(u32 index, const StreamLite& stream, u32 timeout)
513 : {
514 1 : auto taskId = stream.GetRtsq()->GetTaskId();
515 1 : auto notifyId = localNotifies_[index]->GetId();
516 1 : BuildNotifyWaitTask(notifyId, stream, timeout);
517 :
518 : // 上报Profiling任务
519 1 : ReportNotifyWaitTask(notifyId, stream, taskId);
520 1 : }
521 :
522 : // 下发Rtsq sqe, 敲DB
523 6 : void RoceTransportLiteImpl::BuildRdmaDbSendTask(const StreamLite& stream, u64 remoteAddr, u64 dbValue) const
524 : {
525 6 : stream.GetRtsq()->RdmaDbSend(remoteAddr, dbValue);
526 6 : }
527 :
528 : // 下发Rtsq sqe, NotifyWait
529 1 : void RoceTransportLiteImpl::BuildNotifyWaitTask(u32 notifyId, const StreamLite& stream, u32 timeout) const
530 : {
531 1 : stream.GetRtsq()->NotifyWait(notifyId, timeout);
532 1 : }
533 :
534 6 : void RoceTransportLiteImpl::SetFenceConfig(SqeConfigLite& cfg)
535 : {
536 6 : cfg.cqeEn = true;
537 6 : cfg.fence = fence_ ? 1 : 0;
538 6 : fence_ = false;
539 6 : }
540 :
541 4 : void RoceTransportLiteImpl::ReportDmaTask(
542 : const void* src, const void* dst, u64 size, const StreamLite& stream, u32 taskId, TaskParamType taskType,
543 : [[maybe_unused]] DmaOp dmaOp, u64 notifyId, u32 notifyValue, const char* funcName)
544 : {
545 : // 未开启任务上报时直接返回
546 4 : if (!IsReportTask()) {
547 0 : return;
548 : }
549 :
550 12 : HCCL_INFO(
551 : "[RoceTransportLiteImpl::%s][ProfilingTaskParam] sqId[%u], taskId[%u], taskType[%s], "
552 : "src[%p], dst[%p], size[%zu], notifyId[%llu], notifyValue[%u]",
553 : funcName, stream.GetSqId(), taskId, taskType.Describe().c_str(), src, dst, size, notifyId, notifyValue);
554 :
555 4 : DfxTaskInfo* slot = stream.NextTaskSlot();
556 4 : slot->taskType = static_cast<u8>(taskType);
557 4 : slot->sqId = stream.GetSqId();
558 4 : slot->taskId = taskId;
559 4 : const void* opInfo = stream.GetLatestDfxOpInfo();
560 4 : slot->dfxOpInfo = (opInfo != nullptr) ? reinterpret_cast<u64>(opInfo) : DFX_INVALID_U64;
561 4 : slot->linkType = DfxLinkTypeVal::LINK_ROCE;
562 4 : slot->transportType = static_cast<u8>(DfxTransportType::DFX_TRANSPORT_TYPE_ROCE);
563 4 : slot->channelHandle = reinterpret_cast<u64>(this);
564 4 : slot->taskPara.Dma.sqeAddr = stream.GetRtsq()->GetSqeAddr();
565 : }
566 :
567 2 : void RoceTransportLiteImpl::ReportReduceTask(
568 : const void* src, const void* dst, u64 size, const ReduceIn& reduceIn, const StreamLite& stream, u32 taskId,
569 : TaskParamType taskType, u64 notifyId, u32 notifyValue, const char* funcName)
570 : {
571 : // 未开启任务上报时直接返回
572 2 : if (!IsReportTask()) {
573 0 : return;
574 : }
575 :
576 6 : HCCL_INFO(
577 : "[RoceTransportLiteImpl::%s][ProfilingTaskParam] sqId[%u], taskId[%u], taskType[%s], "
578 : "src[%p], dst[%p], size[%zu], notifyId[%llu], notifyValue[%u], dataType[%d], reduceOp[%d]",
579 : funcName, stream.GetSqId(), taskId, taskType.Describe().c_str(), src, dst, size, notifyId, notifyValue,
580 : static_cast<int>(reduceIn.dataType), static_cast<int>(reduceIn.reduceOp));
581 :
582 2 : DfxTaskInfo* slot = stream.NextTaskSlot();
583 2 : slot->taskType = static_cast<u8>(taskType);
584 2 : slot->sqId = stream.GetSqId();
585 2 : slot->taskId = taskId;
586 2 : const void* opInfo = stream.GetLatestDfxOpInfo();
587 2 : slot->dfxOpInfo = (opInfo != nullptr) ? reinterpret_cast<u64>(opInfo) : DFX_INVALID_U64;
588 2 : slot->linkType = DfxLinkTypeVal::LINK_ROCE;
589 2 : slot->transportType = static_cast<u8>(DfxTransportType::DFX_TRANSPORT_TYPE_ROCE);
590 2 : slot->channelHandle = reinterpret_cast<u64>(this);
591 2 : slot->taskPara.Reduce.sqeAddr = stream.GetRtsq()->GetSqeAddr();
592 2 : slot->taskPara.Reduce.srcAddr = reinterpret_cast<u64>(src);
593 2 : slot->taskPara.Reduce.dstAddr = reinterpret_cast<u64>(dst);
594 2 : slot->taskPara.Reduce.size = size;
595 2 : slot->taskPara.Reduce.notifyId = static_cast<u32>(notifyId);
596 2 : slot->taskPara.Reduce.reduceOp = static_cast<u8>(ConvertReduceOpToHcclReduceOp(reduceIn.reduceOp));
597 : }
598 :
599 1 : void RoceTransportLiteImpl::ReportNotifyWaitTask(u64 notifyId, const StreamLite& stream, u32 taskId)
600 : {
601 : // 未开启任务上报时直接返回
602 1 : if (!IsReportTask()) {
603 0 : return;
604 : }
605 :
606 3 : HCCL_INFO(
607 : "[RoceTransportLiteImpl::%s][ProfilingTaskParam] sqId[%u], taskId[%u], notifyId[%llu]", __func__,
608 : stream.GetSqId(), taskId, notifyId);
609 :
610 1 : DfxTaskInfo* slot = stream.NextTaskSlot();
611 1 : slot->taskType = static_cast<u8>(TaskParamTypeVal::TASK_NOTIFY_WAIT);
612 1 : slot->sqId = stream.GetSqId();
613 1 : slot->taskId = taskId;
614 1 : const void* opInfo = stream.GetLatestDfxOpInfo();
615 1 : slot->dfxOpInfo = (opInfo != nullptr) ? reinterpret_cast<u64>(opInfo) : DFX_INVALID_U64;
616 1 : slot->linkType = DfxLinkTypeVal::LINK_ROCE;
617 1 : slot->transportType = static_cast<u8>(DfxTransportType::DFX_TRANSPORT_TYPE_ROCE);
618 1 : slot->channelHandle = reinterpret_cast<u64>(this);
619 1 : slot->taskPara.Notify.sqeAddr = stream.GetRtsq()->GetSqeAddr();
620 : }
621 :
622 7 : bool RoceTransportLiteImpl::IsReportTask()
623 : {
624 : // TaskException或Profiling开启时，允许上报
625 7 : return taskExceptionEnable_ || DfxProfilingHandlerLite::GetInstance().GetProfL1State();
626 : }
627 :
628 : } // namespace Hccl
|
