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 <cmath>
12 : #include "p2p_transport_lite_impl.h"
13 : #include "binary_stream.h"
14 : #include "exception_util.h"
15 : #include "communicator_impl_lite_manager.h"
16 : #include "res_pub.h"
17 :
18 : namespace Hccl {
19 : constexpr u32 NOTIFY_RECORD_WRITE_VALUE = 1;
20 3 : P2PTransportLiteImpl::P2PTransportLiteImpl(
21 3 : std::vector<char>& uniqueId, std::function<void(u32 streamId, u32 taskId, const TaskParam& taskParam)> callback)
22 : {
23 3 : callback_ = callback;
24 : // [header...][notifyUniqueId...][rmtNotifyUniqueId...][rmtBufferUniqueIds...]
25 3 : BinaryStream binaryStream(uniqueId);
26 : u32 theType;
27 3 : binaryStream >> theType;
28 3 : binaryStream >> notifyNum;
29 3 : binaryStream >> bufferNum;
30 :
31 3 : std::vector<char> notifyUniqueIds;
32 3 : binaryStream >> notifyUniqueIds;
33 3 : ParseLocNotifyVec(notifyUniqueIds);
34 :
35 3 : std::vector<char> rmtNotifyUniqueIds;
36 3 : binaryStream >> rmtNotifyUniqueIds;
37 3 : ParseRmtNotifyVec(rmtNotifyUniqueIds, rmtNotifyVec);
38 :
39 3 : std::vector<char> rmtBufferUniqueIds;
40 3 : binaryStream >> rmtBufferUniqueIds;
41 3 : ParseRmtBufferVec(rmtBufferUniqueIds, rmtBufferVec);
42 3 : }
43 :
44 0 : P2PTransportLiteImpl::P2PTransportLiteImpl(std::vector<char>& uniqueId) { Init(uniqueId); }
45 :
46 0 : void P2PTransportLiteImpl::Init(std::vector<char>& uniqueId)
47 : {
48 : // [header...][notifyUniqueId...][rmtNotifyUniqueId...][rmtBufferUniqueIds...]
49 0 : BinaryStream binaryStream(uniqueId);
50 : u32 theType;
51 0 : binaryStream >> theType;
52 0 : binaryStream >> notifyNum;
53 0 : binaryStream >> bufferNum;
54 :
55 0 : std::vector<char> notifyUniqueIds;
56 0 : binaryStream >> notifyUniqueIds;
57 0 : ParseLocNotifyVec(notifyUniqueIds);
58 :
59 0 : std::vector<char> rmtNotifyUniqueIds;
60 0 : binaryStream >> rmtNotifyUniqueIds;
61 0 : ParseRmtNotifyVec(rmtNotifyUniqueIds, rmtNotifyVec);
62 :
63 0 : std::vector<char> locBufferUniqueIds;
64 0 : binaryStream >> locBufferUniqueIds;
65 0 : ParseRmtBufferVec(locBufferUniqueIds, locBufferVec);
66 :
67 0 : std::vector<char> rmtBufferUniqueIds;
68 0 : binaryStream >> rmtBufferUniqueIds;
69 0 : ParseRmtBufferVec(rmtBufferUniqueIds, rmtBufferVec);
70 0 : }
71 :
72 3 : P2PTransportLiteImpl::~P2PTransportLiteImpl() {}
73 :
74 0 : std::string P2PTransportLiteImpl::Describe() const
75 : {
76 0 : std::string desc = "P2PTransportLiteImpl[";
77 :
78 0 : u32 idx = 0;
79 0 : desc += "locNotifyVec=[";
80 0 : for (auto& it : locNotifyVec) {
81 0 : desc += StringFormat("idx=%u, %s;", idx, it->Describe().c_str());
82 0 : idx++;
83 : }
84 :
85 0 : idx = 0;
86 0 : desc += "], rmtNotifyVec=[";
87 0 : for (auto& it : rmtNotifyVec) {
88 0 : desc += StringFormat("idx=%u, %s;", idx, it.Describe().c_str());
89 0 : idx++;
90 : }
91 :
92 0 : idx = 0;
93 0 : desc += "], rmtBufferVec=[";
94 0 : for (auto& it : rmtBufferVec) {
95 0 : desc += StringFormat("idx=%u, %s;", idx, it.Describe().c_str());
96 0 : idx++;
97 : }
98 :
99 0 : desc += "]]";
100 0 : return desc;
101 0 : }
102 :
103 3 : void P2PTransportLiteImpl::ParseLocNotifyVec(std::vector<char>& data)
104 : {
105 3 : if (notifyNum == 0) {
106 0 : HCCL_WARNING("P2PTransportLiteImpl::ParseLocNotifyVec num is 0");
107 0 : return;
108 : }
109 3 : u32 notifySizePerDto = data.size() / notifyNum;
110 :
111 9 : for (u32 idx = 0; idx < notifyNum; idx++) {
112 6 : auto start = data.begin() + idx * notifySizePerDto;
113 6 : auto end = start + notifySizePerDto;
114 6 : std::vector<char> dto(start, end);
115 6 : locNotifyVec.push_back(std::make_unique<NotifyLite>(dto));
116 18 : HCCL_INFO(
117 : "[P2PTransportLiteImpl][ParseLocNotifyVec]locNotify idx=%u, %s", idx,
118 : locNotifyVec.back()->Describe().c_str());
119 6 : }
120 : }
121 :
122 3 : void P2PTransportLiteImpl::ParseRmtNotifyVec(std::vector<char>& data, std::vector<RmtP2PNotifyLite>& vec) const
123 : {
124 3 : if (notifyNum == 0) {
125 0 : HCCL_WARNING("P2PTransportLiteImpl::ParseRmtNotifyVec notifyNum is 0");
126 0 : return;
127 : }
128 :
129 3 : u32 rmtBufferSizePerDto = data.size() / notifyNum;
130 9 : HCCL_INFO(
131 : "[P2PTransportLiteImpl][ParseRmtNotifyVec]Parse notifyNum=%u, sizePerDto=%u", notifyNum, rmtBufferSizePerDto);
132 3 : BinaryStream binaryStream(data);
133 :
134 9 : for (u32 idx = 0; idx < notifyNum; idx++) {
135 : RmtP2PNotifyLite p2pNotifyLite;
136 6 : binaryStream >> p2pNotifyLite.addr;
137 6 : binaryStream >> p2pNotifyLite.size;
138 6 : binaryStream >> p2pNotifyLite.id;
139 18 : HCCL_INFO("[P2PTransportLiteImpl][ParseRmtNotifyVec]idx=%u, %s", idx, p2pNotifyLite.Describe().c_str());
140 6 : vec.push_back(p2pNotifyLite);
141 : }
142 3 : }
143 :
144 3 : void P2PTransportLiteImpl::ParseRmtBufferVec(std::vector<char>& data, std::vector<P2PBufLite>& vec) const
145 : {
146 3 : if (bufferNum == 0) {
147 0 : HCCL_WARNING("P2PTransportLiteImpl::ParseRmtBufferVec bufferNum is 0");
148 0 : return;
149 : }
150 :
151 3 : u32 rmtBufferSizePerDto = data.size() / bufferNum;
152 9 : HCCL_INFO(
153 : "[P2PTransportLiteImpl][ParseRmtBufferVec]Parse bufferNum=%u, sizePerDto=%u", bufferNum, rmtBufferSizePerDto);
154 3 : BinaryStream binaryStream(data);
155 :
156 6 : for (u32 idx = 0; idx < bufferNum; idx++) {
157 : P2PBufLite p2pBufLite;
158 3 : binaryStream >> p2pBufLite.addr;
159 3 : binaryStream >> p2pBufLite.size;
160 9 : HCCL_INFO("[P2PTransportLiteImpl][ParseRmtBufferVec]idx=%u, %s", idx, p2pBufLite.Describe().c_str());
161 3 : vec.push_back(p2pBufLite);
162 : }
163 3 : }
164 :
165 0 : Buffer P2PTransportLiteImpl::GetRmtBuffer(u32 index)
166 : {
167 0 : if (UNLIKELY(index >= rmtBufferVec.size())) {
168 0 : THROW<InternalException>(StringFormat(
169 : "P2PTransportLiteImpl::GetRmtBuffer out-of-bounds. index=%u, size=%u", index, rmtBufferVec.size()));
170 : }
171 0 : HCCL_DEBUG(
172 : "[P2PTransportLiteImpl][GetRmtBuffer]buffer index[%u], addr[%llu], size[%llu]", index, rmtBufferVec[index].addr,
173 : rmtBufferVec[index].size);
174 0 : return Buffer(rmtBufferVec[index].addr, rmtBufferVec[index].size);
175 : }
176 :
177 : HcclResult
178 0 : P2PTransportLiteImpl::BuildLocRmaBufferLite(const uintptr_t addr, const size_t size, RmaBufferLite& rmaBufferLite)
179 : {
180 0 : HCCL_INFO(
181 : "[P2PTransportLiteImpl::%s] start to find addr[0x%llx], size[0x%llx] in locBufferVec, whose size is %zu. ",
182 : __func__, addr, size, locBufferVec.size());
183 0 : if (locBufferVec.empty()) {
184 0 : HCCL_ERROR("[P2PTransportLiteImpl::%s] locBufferVec is empty.", __func__);
185 0 : return HCCL_E_INTERNAL;
186 : }
187 :
188 0 : bool isAddrInRange = false;
189 0 : for (auto& it : locBufferVec) {
190 0 : Buffer iterBuf(it.addr, it.size);
191 0 : if (iterBuf.Contains(addr, size)) {
192 0 : rmaBufferLite = RmaBufferLite(addr, size, 0, 0);
193 0 : isAddrInRange = true;
194 0 : break;
195 : }
196 0 : }
197 :
198 0 : if (!isAddrInRange) {
199 0 : HCCL_WARNING(
200 : "[P2PTransportLiteImpl::%s] addr[0x%llx], size[0x%llx] not in any range of locBufferVec.", __func__, addr,
201 : size);
202 0 : rmaBufferLite = RmaBufferLite(addr, size, 0, 0);
203 0 : return HCCL_SUCCESS;
204 : }
205 :
206 0 : return HCCL_SUCCESS;
207 : }
208 :
209 0 : void P2PTransportLiteImpl::BuildNotifyRecordTask(const StreamLite& stream, u64 rmtNotifyAddr)
210 : {
211 : // Post仅需向对端寄存器写入1
212 0 : stream.GetRtsq()->P2PWriteValue(rmtNotifyAddr, NOTIFY_RECORD_WRITE_VALUE);
213 0 : }
214 :
215 0 : void P2PTransportLiteImpl::BuildNotifyWaitTask(const StreamLite& stream, u32 notifyId)
216 : {
217 0 : stream.GetRtsq()->NotifyWait(notifyId);
218 0 : }
219 :
220 0 : void P2PTransportLiteImpl::BuildP2PRead(const StreamLite& stream, const RmaBufferLite& loc, const Buffer& rmt)
221 : {
222 0 : if (UNLIKELY(rmt.GetSize() != loc.GetSize())) {
223 0 : HCCL_ERROR(
224 : "[P2PTransportLiteImpl]%s srcBuffer size[%llu] is not equal to distBuffer size[%llu], return", __func__,
225 : rmt.GetSize(), loc.GetSize());
226 0 : THROW<InternalException>("[P2PTransportLiteImpl]BuildP2PRead srcBuffer is not equal to distBuffer");
227 : return;
228 : }
229 :
230 0 : if (UNLIKELY(rmt.GetSize() == 0)) {
231 0 : HCCL_WARNING("[P2PTransportLiteImpl]%s srcBuffer size is 0, return", __func__);
232 0 : return;
233 : }
234 :
235 0 : HCCL_INFO(
236 : "P2PTransportLiteImpl::Read remoteBuff[%s] localBuff[%s]", rmt.Describe().c_str(), loc.Describe().c_str());
237 : // 传入数据大小不能超过 u32最大值， 需要进行切分
238 0 : u64 u32Max = UINT32_MAX;
239 0 : double countSplitingTimes = static_cast<double>(rmt.GetSize()) / static_cast<double>(u32Max);
240 0 : u64 splitingTimes = static_cast<int>(std::ceil(countSplitingTimes));
241 0 : u64 src = rmt.GetAddr();
242 0 : u64 dst = loc.GetAddr();
243 0 : u64 blockSize = u32Max;
244 0 : u64 offset = u32Max;
245 0 : for (u64 i = 0; i < splitingTimes; i++) {
246 : // 处理尾块数据
247 0 : if (i == splitingTimes - 1) {
248 0 : blockSize = rmt.GetSize() - u32Max * (splitingTimes - 1);
249 0 : offset = blockSize;
250 : }
251 :
252 0 : auto taskId = stream.GetRtsq()->GetTaskId();
253 0 : stream.GetRtsq()->SdmaCopy(src, dst, blockSize, 0);
254 0 : HCCL_INFO(
255 : "P2PTransportLiteImpl::%s, srcA:0x%llx dstA:0x%llx,size=0x%llx, taskId=%u", __func__, src, dst, blockSize,
256 : taskId);
257 :
258 0 : if (callback_) {
259 0 : TaskParam taskParam{};
260 0 : taskParam.taskType = TaskParamType::TASK_SDMA;
261 0 : taskParam.beginTime = ProfGetCurCpuTimestamp();
262 0 : taskParam.taskPara.DMA.src = reinterpret_cast<void*>(src);
263 0 : taskParam.taskPara.DMA.dst = reinterpret_cast<void*>(dst);
264 0 : taskParam.taskPara.DMA.size = blockSize;
265 0 : taskParam.taskPara.DMA.notifyID = INVALID_VALUE_NOTIFYID;
266 0 : taskParam.taskPara.DMA.linkType = DfxLinkType::PCIE;
267 0 : taskParam.taskPara.DMA.dmaOp = DmaOp::HCCL_DMA_READ;
268 0 : callback_(stream.GetSqId(), taskId, taskParam);
269 0 : }
270 0 : DfxTaskInfo* slot = stream.NextTaskSlot();
271 0 : slot->taskType = TaskParamTypeVal::TASK_SDMA;
272 0 : slot->sqId = stream.GetSqId();
273 0 : slot->taskId = taskId;
274 0 : const void* opInfo = stream.GetLatestDfxOpInfo();
275 0 : slot->dfxOpInfo = (opInfo != nullptr) ? reinterpret_cast<u64>(opInfo) : INVALID_U64;
276 0 : slot->linkType = DfxLinkTypeVal::LINK_PCIE;
277 0 : slot->transportType = static_cast<u8>(DfxTransportType::DFX_TRANSPORT_TYPE_SDMA);
278 0 : slot->channelHandle = reinterpret_cast<u64>(this);
279 0 : slot->taskPara.Dma.sqeAddr = stream.GetRtsq()->GetSqeAddr();
280 0 : src += offset;
281 0 : dst += offset;
282 : }
283 : }
284 :
285 0 : void P2PTransportLiteImpl::BuildP2PReadReduce(
286 : const StreamLite& stream, const RmaBufferLite& loc, const Buffer& rmt, const ReduceIn& reduceIn)
287 : {
288 0 : if (UNLIKELY(rmt.GetSize() != loc.GetSize())) {
289 0 : HCCL_ERROR(
290 : "[P2PTransportLiteImpl]%s srcBuffer size[%llu] is not equal to distBuffer size[%llu], return", __func__,
291 : rmt.GetSize(), loc.GetSize());
292 0 : THROW<InternalException>("[P2PTransportLiteImpl]BuildP2PReadReduce srcBuffer is not equal to distBuffer");
293 : return;
294 : }
295 :
296 0 : if (UNLIKELY(rmt.GetSize() == 0)) {
297 0 : HCCL_WARNING("[P2PTransportLiteImpl]%s srcBuffer size is 0, return", __func__);
298 0 : return;
299 : }
300 :
301 0 : HCCL_INFO(
302 : "P2PTransportLiteImpl::ReadReduce remoteBuff[%s] localBuff[%s] reduce[%s]", rmt.Describe().c_str(),
303 : loc.Describe().c_str(), reduceIn.Describe().c_str());
304 : // 传入数据大小不能超过 u32最大值， 需要进行切分
305 0 : u64 u32Max = UINT32_MAX;
306 0 : double countSplitingTimes = static_cast<double>(rmt.GetSize()) / static_cast<double>(u32Max);
307 0 : u64 splitingTimes = static_cast<int>(std::ceil(countSplitingTimes));
308 0 : u64 src = rmt.GetAddr();
309 0 : u64 dst = loc.GetAddr();
310 0 : u64 blockSize = u32Max;
311 0 : u64 offset = u32Max;
312 0 : for (u64 i = 0; i < splitingTimes; i++) {
313 : // 处理尾块数据
314 0 : if (i == splitingTimes - 1) {
315 0 : blockSize = rmt.GetSize() - u32Max * (splitingTimes - 1);
316 0 : offset = blockSize;
317 : }
318 :
319 0 : auto taskId = stream.GetRtsq()->GetTaskId();
320 0 : stream.GetRtsq()->SdmaReduce(src, dst, blockSize, 0, reduceIn);
321 :
322 0 : HCCL_INFO(
323 : "P2PTransportLiteImpl::%s, srcA:0x%llx dstA:0x%llx,size=0x%llx, reduceIn=%s, taskId=%u", __func__, src, dst,
324 : blockSize, reduceIn.Describe(), taskId);
325 :
326 0 : if (callback_) {
327 0 : TaskParam taskParam{};
328 0 : taskParam.taskType = TaskParamType::TASK_REDUCE_INLINE;
329 0 : taskParam.beginTime = ProfGetCurCpuTimestamp();
330 0 : taskParam.taskPara.Reduce.src = reinterpret_cast<void*>(src);
331 0 : taskParam.taskPara.Reduce.dst = reinterpret_cast<void*>(dst);
332 0 : taskParam.taskPara.Reduce.size = blockSize;
333 0 : taskParam.taskPara.Reduce.notifyID = INVALID_VALUE_NOTIFYID;
334 0 : taskParam.taskPara.Reduce.linkType = DfxLinkType::PCIE;
335 0 : taskParam.taskPara.Reduce.reduceOp = ConvertReduceOpToHcclReduceOp(reduceIn.reduceOp);
336 0 : taskParam.taskPara.Reduce.dataType = DataTypeToHcclDataType(reduceIn.dataType);
337 0 : callback_(stream.GetSqId(), taskId, taskParam);
338 0 : }
339 0 : DfxTaskInfo* slot = stream.NextTaskSlot();
340 0 : slot->taskType = TaskParamTypeVal::TASK_REDUCE_INLINE;
341 0 : slot->sqId = stream.GetSqId();
342 0 : slot->taskId = taskId;
343 0 : const void* opInfo = stream.GetLatestDfxOpInfo();
344 0 : slot->dfxOpInfo = (opInfo != nullptr) ? reinterpret_cast<u64>(opInfo) : INVALID_U64;
345 0 : slot->linkType = DfxLinkTypeVal::LINK_PCIE;
346 0 : slot->transportType = static_cast<u8>(DfxTransportType::DFX_TRANSPORT_TYPE_SDMA);
347 0 : slot->channelHandle = reinterpret_cast<u64>(this);
348 0 : slot->taskPara.Reduce.sqeAddr = stream.GetRtsq()->GetSqeAddr();
349 0 : slot->taskPara.Reduce.srcAddr = src;
350 0 : slot->taskPara.Reduce.dstAddr = dst;
351 0 : slot->taskPara.Reduce.size = blockSize;
352 0 : slot->taskPara.Reduce.notifyId = INVALID_U32;
353 0 : slot->taskPara.Reduce.reduceOp = static_cast<u8>(ConvertReduceOpToHcclReduceOp(reduceIn.reduceOp));
354 0 : src += offset;
355 0 : dst += offset;
356 : }
357 : }
358 :
359 0 : void P2PTransportLiteImpl::Post(u32 index, const StreamLite& stream)
360 : {
361 0 : if (UNLIKELY(index >= rmtNotifyVec.size())) {
362 0 : HCCL_ERROR(
363 : "[P2PTransportLiteImpl]%s notify out-of-bounds, notifyNum[%u], index[%u]", __func__, rmtNotifyVec.size(),
364 : index);
365 0 : THROW<InternalException>(
366 : "[P2PTransportLiteImpl]%s notify out-of-bounds, notifyNum[%u], index[%u]", __func__, rmtNotifyVec.size(),
367 : index);
368 : return;
369 : }
370 :
371 0 : auto taskId = stream.GetRtsq()->GetTaskId();
372 0 : auto rmtNotifyAddr = rmtNotifyVec[index].addr;
373 0 : BuildNotifyRecordTask(stream, rmtNotifyAddr);
374 :
375 0 : HCCL_INFO(
376 : "P2PTransportLiteImpl::Post rmtNotifyAddr[0x%llx], notifyId[%u], taskId[%u]", rmtNotifyAddr,
377 : rmtNotifyVec[index].id, taskId);
378 :
379 0 : if (callback_) {
380 0 : TaskParam taskParam{};
381 0 : taskParam.taskType = TaskParamType::TASK_NOTIFY_RECORD;
382 0 : taskParam.beginTime = ProfGetCurCpuTimestamp();
383 0 : taskParam.taskPara.Notify.notifyID = rmtNotifyVec[index].id;
384 0 : taskParam.taskPara.Notify.value = 1;
385 0 : callback_(stream.GetSqId(), taskId, taskParam);
386 0 : }
387 0 : DfxTaskInfo* slot = stream.NextTaskSlot();
388 0 : slot->taskType = TaskParamTypeVal::TASK_NOTIFY_RECORD;
389 0 : slot->sqId = stream.GetSqId();
390 0 : slot->taskId = taskId;
391 0 : const void* opInfo = stream.GetLatestDfxOpInfo();
392 0 : slot->dfxOpInfo = (opInfo != nullptr) ? reinterpret_cast<u64>(opInfo) : INVALID_U64;
393 0 : slot->linkType = DfxLinkTypeVal::LINK_PCIE;
394 0 : slot->transportType = static_cast<u8>(DfxTransportType::DFX_TRANSPORT_TYPE_SDMA);
395 0 : slot->channelHandle = reinterpret_cast<u64>(this);
396 0 : slot->taskPara.Notify.sqeAddr = stream.GetRtsq()->GetSqeAddr();
397 0 : return;
398 : }
399 :
400 0 : void P2PTransportLiteImpl::Wait(u32 index, const StreamLite& stream)
401 : {
402 0 : WaitWithTimeout(index, stream, CommunicatorImplLiteMgr::GetInstance().GetEnvConfig().hcclExecTimeout);
403 0 : }
404 :
405 3 : void P2PTransportLiteImpl::WaitWithTimeout(u32 index, const StreamLite& stream, u32 timeout)
406 : {
407 3 : auto taskId = stream.GetRtsq()->GetTaskId();
408 3 : auto notifyId = locNotifyVec[index]->GetId();
409 3 : stream.GetRtsq()->NotifyWait(notifyId, timeout);
410 :
411 9 : HCCL_INFO(
412 : "P2PTransportLiteImpl::WaitWithTimeout notifyId[%u], taskId[%u], timeout[%u ms]", notifyId, taskId, timeout);
413 3 : if (callback_) {
414 3 : TaskParam taskParam{};
415 3 : taskParam.taskType = TaskParamType::TASK_NOTIFY_WAIT;
416 3 : taskParam.beginTime = ProfGetCurCpuTimestamp();
417 3 : taskParam.taskPara.Notify.notifyID = notifyId;
418 3 : taskParam.taskPara.Notify.value = 1;
419 3 : callback_(stream.GetSqId(), taskId, taskParam);
420 3 : }
421 3 : DfxTaskInfo* slot = stream.NextTaskSlot();
422 3 : slot->taskType = TaskParamTypeVal::TASK_NOTIFY_WAIT;
423 3 : slot->sqId = stream.GetSqId();
424 3 : slot->taskId = taskId;
425 3 : const void* opInfo = stream.GetLatestDfxOpInfo();
426 3 : slot->dfxOpInfo = (opInfo != nullptr) ? reinterpret_cast<u64>(opInfo) : INVALID_U64;
427 3 : slot->linkType = DfxLinkTypeVal::LINK_PCIE;
428 3 : slot->transportType = static_cast<u8>(DfxTransportType::DFX_TRANSPORT_TYPE_SDMA);
429 3 : slot->channelHandle = reinterpret_cast<u64>(this);
430 3 : slot->taskPara.Notify.sqeAddr = stream.GetRtsq()->GetSqeAddr();
431 3 : return;
432 : }
433 :
434 0 : void P2PTransportLiteImpl::Read(const RmaBufferLite& loc, const Buffer& rmt, const StreamLite& stream)
435 : {
436 0 : BuildP2PRead(stream, loc, rmt);
437 0 : }
438 :
439 0 : void P2PTransportLiteImpl::ReadReduce(
440 : const RmaBufferLite& loc, const Buffer& rmt, const ReduceIn& reduceIn, const StreamLite& stream)
441 : {
442 0 : BuildP2PReadReduce(stream, loc, rmt, reduceIn);
443 0 : }
444 :
445 0 : void P2PTransportLiteImpl::BatchTransfer(
446 : const std::vector<RmaBufferLite>& loc, const std::vector<Buffer>& rmt,
447 : const std::vector<BaseTransportLiteImpl::TransferOp>& transferOp, const StreamLite& stream)
448 : {
449 0 : if (UNLIKELY(loc.empty())) {
450 0 : return;
451 : }
452 0 : u32 insNum = loc.size();
453 0 : for (u32 i = 0; i < insNum; i++) {
454 0 : if (transferOp[i].transType == TransferType::WRITE) {
455 0 : HCCL_ERROR("[P2PTransportLiteImpl][BatchTransfer]does not support WRITE operation");
456 0 : THROW<InternalException>("[P2PTransportLiteImpl]BatchTransfer not support WRITE");
457 0 : } else if (transferOp[i].transType == TransferType::READ) {
458 0 : if (transferOp[i].reduceIn.reduceOp == ReduceOp::INVALID) {
459 0 : BuildP2PRead(stream, loc[i], rmt[i]);
460 : } else {
461 0 : BuildP2PReadReduce(stream, loc[i], rmt[i], transferOp[i].reduceIn);
462 : }
463 : }
464 : }
465 : }
466 : } // namespace Hccl
|
