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