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