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 "p2p_transport.h"
12 :
13 : #include "orion_adapter_rts.h"
14 : #include "exchange_ipc_notify_dto.h"
15 : #include "exchange_ipc_buffer_dto.h"
16 : #include "user_remote_mem_getter.h"
17 :
18 : namespace Hccl {
19 :
20 19 : P2PTransport::P2PTransport(CommonLocRes &commonLocRes, Attribution &attr, const LinkData &linkData,
21 19 : const Socket &socket)
22 19 : : BaseMemTransport(commonLocRes, attr, linkData, socket, TransportType::P2P)
23 19 : {}
24 :
25 0 : P2PTransport::P2PTransport(CommonLocRes &commonLocRes, Attribution &attr, const LinkData &linkData,
26 0 : const Socket &socket, std::function<void(u32 streamId, u32 taskId, TaskParam taskParam)> callback)
27 0 : : BaseMemTransport(commonLocRes, attr, linkData, socket, TransportType::P2P, callback)
28 0 : {}
29 :
30 0 : std::string P2PTransport::Describe() const
31 : {
32 : string msg = StringFormat("P2PTransport:commonLocRes=%s, pidMsgSize=%u, myPid=%u, rmtPid=%u, rmtPidValid=%d,",
33 0 : commonLocRes.Describe().c_str(), pidMsgSize, myPid, rmtPid, rmtPidValid);
34 0 : msg += StringFormat("exchangeDataSize=%u, ", exchangeDataSize);
35 0 : u32 pos = 0;
36 0 : for (auto &it : rmtNotifyVec) {
37 0 : msg += StringFormat("rmtNotify[%u]=%s, ", pos, it->Describe().c_str());
38 0 : pos++;
39 : }
40 :
41 0 : pos = 0;
42 0 : for (auto &it : rmtNotifyVec) {
43 0 : if (it != nullptr) {
44 0 : msg += StringFormat("rmtBuffer[%u]=%s, ", pos, it->Describe().c_str());
45 : } else {
46 0 : msg += StringFormat("rmtBuffer[%u]=nullptr, ", pos);
47 : }
48 0 : pos++;
49 : }
50 0 : return msg;
51 0 : }
52 :
53 0 : static void SubmitTask(const TaskP2pMemcpy &p2pMemcpy, const Stream &stream)
54 : {
55 0 : HCCL_INFO("[P2PTransport::%s]not support, p2p dst addr[%llu], stream[%p]", __func__, p2pMemcpy.GetDstAddr(), stream.GetPtr());
56 0 : }
57 :
58 0 : static void SubmitTask(const TaskSdmaReduce &sdmaReduce, const Stream &stream)
59 : {
60 0 : HCCL_INFO("[P2PTransport::%s]not support, sdmaReduce dst addr[%llu], stream[%p]", __func__, sdmaReduce.GetDstAddr(), stream.GetPtr());
61 0 : }
62 :
63 2 : template <typename TaskType> std::function<void(const BaseTask &, const Stream &)> GetSubmitP2PTaskFunction()
64 : {
65 2 : return [](const BaseTask &task, const Stream &stream) {
66 0 : SubmitTask(static_cast<const TaskType &>(task), stream);
67 2 : };
68 : }
69 :
70 : std::map<TaskType, std::function<void(const BaseTask &, const Stream &)>> g_p2pTaskSubmitRuleMap
71 : = {{TaskType::P2P_MEMCPY, GetSubmitP2PTaskFunction<TaskP2pMemcpy>()},
72 : {TaskType::SDMA_REDUCE, GetSubmitP2PTaskFunction<TaskSdmaReduce>()}};
73 :
74 0 : static void SubmitP2PTask(unique_ptr<BaseTask> task, const Stream &stream)
75 : {
76 0 : if (task != nullptr) { // task为空的情况下，不需要提交task
77 0 : g_p2pTaskSubmitRuleMap.at(task->GetType())(*task.get(), stream);
78 : }
79 0 : }
80 :
81 0 : MemoryBuffer P2PTransport::GetLocMemBuffer(const RmaBufferSlice &locSlice) const
82 : {
83 0 : return MemoryBuffer(locSlice.addr, locSlice.size, 0);
84 : }
85 :
86 0 : MemoryBuffer P2PTransport::GetRmtMemBuffer(const RmtRmaBufferSlice &rmtSlice) const
87 : {
88 0 : return MemoryBuffer(rmtSlice.addr, rmtSlice.size, 0);
89 : }
90 :
91 0 : void P2PTransport::Post(u32 index, const Stream &stream)
92 : {
93 0 : rmtNotifyVec[index]->Post(stream);
94 0 : }
95 :
96 0 : void P2PTransport::Read(const RmaBufferSlice &locSlice, const RmtRmaBufferSlice &rmtSlice, const Stream &stream)
97 : {
98 0 : SqeConfig config;
99 0 : SubmitP2PTask(commonLocRes.connVec[0]->PrepareRead(GetRmtMemBuffer(rmtSlice), GetLocMemBuffer(locSlice), config),
100 : stream);
101 0 : }
102 :
103 0 : void P2PTransport::ReadReduce(const RmaBufferSlice &locSlice, const RmtRmaBufferSlice &rmtSlice,
104 : const ReduceIn &reduceIn, const Stream &stream)
105 : {
106 0 : SqeConfig config;
107 0 : SubmitP2PTask(commonLocRes.connVec[0]->PrepareReadReduce(GetRmtMemBuffer(rmtSlice), GetLocMemBuffer(locSlice),
108 : reduceIn.dataType, reduceIn.reduceOp, config),
109 : stream);
110 0 : }
111 :
112 0 : void P2PTransport::Write(const RmaBufferSlice &locSlice, const RmtRmaBufferSlice &rmtSlice, const Stream &stream)
113 : {
114 0 : SqeConfig config;
115 0 : SubmitP2PTask(commonLocRes.connVec[0]->PrepareWrite(GetRmtMemBuffer(rmtSlice), GetLocMemBuffer(locSlice), config),
116 : stream);
117 0 : }
118 :
119 0 : void P2PTransport::WriteReduce(const RmaBufferSlice &locSlice, const RmtRmaBufferSlice &rmtSlice,
120 : const ReduceIn &reduceIn, const Stream &stream)
121 : {
122 0 : SqeConfig config;
123 0 : SubmitP2PTask(commonLocRes.connVec[0]->PrepareWriteReduce(GetRmtMemBuffer(rmtSlice), GetLocMemBuffer(locSlice),
124 : reduceIn.dataType, reduceIn.reduceOp, config),
125 : stream);
126 0 : }
127 :
128 0 : TransportStatus P2PTransport::GetStatus()
129 : {
130 0 : if (baseStatus == TransportStatus::READY) {
131 0 : return baseStatus;
132 0 : } else if (baseStatus == TransportStatus::INIT) {
133 0 : p2pStatus = P2PStatus::INIT;
134 : }
135 :
136 0 : if (!IsSocketReady()) {
137 0 : return baseStatus;
138 : }
139 :
140 0 : switch (p2pStatus) {
141 0 : case P2PStatus::INIT:
142 0 : p2pStatus = P2PStatus::SOCKET_OK;
143 0 : baseStatus = TransportStatus::SOCKET_OK;
144 0 : break;
145 0 : case P2PStatus::SOCKET_OK:
146 0 : PrepareSendData();
147 0 : p2pStatus = P2PStatus::SEND_DATA_SIZE;
148 0 : break;
149 0 : case P2PStatus::SEND_DATA_SIZE:
150 0 : RecvDataSize();
151 0 : p2pStatus = P2PStatus::RECV_DATA_SIZE;
152 0 : break;
153 0 : case P2PStatus::RECV_DATA_SIZE:
154 0 : SendExchangeData();
155 0 : p2pStatus = P2PStatus::SEND_DATA;
156 0 : break;
157 0 : case P2PStatus::SEND_DATA:
158 0 : RecvExchangeData();
159 0 : p2pStatus = P2PStatus::RECV_DATA;
160 0 : break;
161 0 : case P2PStatus::RECV_DATA:
162 0 : ProcessRecvData();
163 0 : SetBaseStatusReady();
164 0 : break;
165 0 : default:
166 0 : break;
167 : }
168 0 : HCCL_INFO("%s, baseStatus=%s, p2pStatus = %s", GetLinkDescInfo().c_str(), baseStatus.Describe().c_str(),
169 : p2pStatus.Describe().c_str());
170 0 : return baseStatus;
171 : }
172 :
173 0 : bool P2PTransport::IsRmtPidValid() const
174 : {
175 : // 优化方向：基于remoteAddr保存PID的单例，这样可以减少 PID交换
176 0 : return rmtPidValid;
177 : }
178 :
179 0 : void P2PTransport::SendPid()
180 : {
181 0 : myPid = HrtDeviceGetBareTgid();
182 0 : HCCL_INFO("P2PTransport: send pid %u", myPid);
183 :
184 0 : BinaryStream binaryStream;
185 0 : binaryStream << myPid;
186 :
187 0 : std::vector<char> data;
188 0 : binaryStream.Dump(data);
189 0 : pidMsgSize = data.size();
190 0 : socket->SendAsync(&data[0], data.size());
191 :
192 0 : HCCL_INFO("send pid %s, size=%llu, data=0x%s", GetLinkDescInfo().c_str(), data.size(),
193 : Bytes2hex(data.data(), data.size()).c_str());
194 0 : }
195 :
196 0 : void P2PTransport::RecvPid()
197 : {
198 0 : std::vector<char> data(pidMsgSize);
199 0 : socket->RecvAsync(reinterpret_cast<u8 *>(&data[0]), data.size());
200 0 : HCCL_INFO("recv pid %s, size=%llu, data=%s", GetLinkDescInfo().c_str(), data.size(),
201 : Bytes2hex(data.data(), data.size()).c_str());
202 :
203 0 : BinaryStream binaryStream(data);
204 0 : binaryStream >> rmtPid;
205 0 : HCCL_INFO("P2PTransport: recv pid %u", rmtPid);
206 0 : }
207 :
208 0 : void P2PTransport::Grant()
209 : {
210 : // 暂时不做Grant处理
211 0 : return;
212 : }
213 :
214 0 : void P2PTransport::PrepareSendData()
215 : {
216 0 : notifyNum = commonLocRes.notifyVec.size(); // 需要交换的notify数量
217 0 : bufferNum = commonLocRes.bufferVec.size(); // 需要交换的buffer数量
218 :
219 0 : HCCL_INFO("%s commLocResExchange %s, notifyNum=%u, bufferNum=%u", GetLinkDescInfo().c_str(),
220 : commonLocRes.Describe().c_str(), notifyNum, bufferNum);
221 :
222 0 : BinaryStream binaryStream;
223 :
224 0 : HandshakeMsgPack(binaryStream);
225 0 : NotifyVecPack(binaryStream);
226 0 : BufferVecPack(binaryStream);
227 :
228 0 : sendData.clear();
229 0 : binaryStream.Dump(sendData);
230 0 : exchangeDataSize = sendData.size();
231 0 : socket->SendAsync(&exchangeDataSize, sizeof(exchangeDataSize));
232 :
233 0 : HCCL_INFO("send datasize %s, size=%u", GetLinkDescInfo().c_str(), exchangeDataSize);
234 0 : }
235 :
236 0 : void P2PTransport::RecvDataSize()
237 : {
238 0 : socket->RecvAsync(reinterpret_cast<u8 *>(&exchangeDataSize), sizeof(exchangeDataSize));
239 0 : HCCL_INFO("P2PTransport: recv datasize %u", exchangeDataSize);
240 0 : }
241 :
242 0 : void P2PTransport::SendExchangeData()
243 : {
244 0 : socket->SendAsync(&sendData[0], sendData.size());
245 0 : HCCL_INFO("send data %s, size=%llu, data=0x%s", GetLinkDescInfo().c_str(), sendData.size(),
246 : Bytes2hex(sendData.data(), sendData.size()).c_str());
247 0 : }
248 :
249 0 : void P2PTransport::RecvExchangeData()
250 : {
251 0 : recvData.resize(exchangeDataSize);
252 0 : socket->RecvAsync(reinterpret_cast<u8 *>(&recvData[0]), recvData.size());
253 0 : HCCL_INFO("RecvExchangeData recv data %s, size=%llu, data=%s", GetLinkDescInfo().c_str(), recvData.size(),
254 : Bytes2hex(recvData.data(), recvData.size()).c_str());
255 0 : }
256 :
257 0 : void P2PTransport::ProcessRecvData()
258 : {
259 0 : HCCL_INFO("ProcessRecvData recv data %s, size=%llu, data=%s", GetLinkDescInfo().c_str(), recvData.size(),
260 : Bytes2hex(recvData.data(), recvData.size()).c_str());
261 :
262 0 : BinaryStream binaryStream(recvData);
263 0 : HandshakeMsgUnpack(binaryStream);
264 0 : RmtNotifyVecUnpackProc(binaryStream);
265 0 : RmtBufferVecUnpackProc(binaryStream);
266 :
267 0 : HCCL_INFO("%s unpack success", GetLinkDescInfo().c_str());
268 0 : }
269 :
270 0 : void P2PTransport::RmtNotifyVecUnpackProc(BinaryStream &binaryStream)
271 : {
272 : u32 rmtNotifyNum;
273 0 : binaryStream >> rmtNotifyNum;
274 0 : HCCL_INFO("unpack notify %s locNum=%u, rmtNum=%u", GetLinkDescInfo().c_str(), notifyNum, rmtNotifyNum);
275 0 : if (rmtNotifyNum != notifyNum) {
276 0 : MACRO_THROW(InvalidParamsException,
277 : StringFormat("notifyNum=%u is not equal to rmtNotifyNum=%u", notifyNum, rmtNotifyNum));
278 : }
279 :
280 0 : rmtNotifyVec.clear(); // 清空remote资源
281 0 : for (u32 i = 0; i < rmtNotifyNum; i++) {
282 : u32 pos;
283 0 : binaryStream >> pos;
284 0 : ExchangeIpcNotifyDto dto;
285 0 : dto.Deserialize(binaryStream);
286 0 : HCCL_INFO("unpack notify pos=%u dto %s", pos, dto.Describe().c_str());
287 0 : rmtNotifyVec.push_back(make_unique<IpcRemoteNotify>(dto));
288 0 : HCCL_INFO("unpack notify pos=%u, rmtNotify=%s", pos, rmtNotifyVec[i]->Describe().c_str());
289 0 : }
290 0 : }
291 :
292 0 : void P2PTransport::BufferVecPack(BinaryStream &binaryStream)
293 : {
294 0 : binaryStream << bufferNum;
295 0 : HCCL_INFO("start pack %s bufferVec", transportType.Describe().c_str());
296 0 : u32 pos = 0;
297 0 : for (auto &it : commonLocRes.bufferVec) {
298 0 : binaryStream << pos;
299 0 : if (it != nullptr) { // 非空的buffer，从buffer中获取 dto
300 0 : std::unique_ptr<Serializable> dto = it->GetExchangeDto();
301 0 : dto->Serialize(binaryStream);
302 0 : HCCL_INFO("pack buffer pos=%u dto %s", pos, dto->Describe().c_str());
303 0 : } else { // 空的buffer，dto所有字段为0(size=0)
304 0 : ExchangeIpcBufferDto exchangeDto;
305 0 : exchangeDto.Serialize(binaryStream);
306 0 : HCCL_INFO("pack buffer pos=%u, dto is null %s", pos, exchangeDto.Describe().c_str());
307 0 : }
308 0 : pos++;
309 : }
310 0 : }
311 :
312 0 : void P2PTransport::RmtBufferVecUnpackProc(BinaryStream &binaryStream)
313 : {
314 : u32 rmtBufferNum;
315 0 : binaryStream >> rmtBufferNum;
316 0 : HCCL_INFO("unpack buffer %s locNum=%u rmtNum=%u", GetLinkDescInfo().c_str(), bufferNum, rmtBufferNum);
317 0 : if (rmtBufferNum != bufferNum) {
318 0 : MACRO_THROW(InvalidParamsException,
319 : StringFormat("bufferNum=%u is not equal to rmtBufferNum=%u", bufferNum, rmtBufferNum));
320 : }
321 :
322 0 : rmtBufferVec.clear();
323 0 : rmtRmaBufferVec.clear();
324 0 : for (u32 i = 0; i < rmtBufferNum; i++) {
325 : u32 pos;
326 0 : binaryStream >> pos;
327 0 : ExchangeIpcBufferDto dto;
328 0 : dto.Deserialize(binaryStream);
329 0 : HCCL_INFO("unpack buffer pos=%u, dto %s", pos, dto.Describe().c_str());
330 :
331 0 : if (dto.size == 0) { // size为0，则为 remote 空buffer
332 0 : HCCL_INFO("unpack nullptr, pos=%u", pos);
333 0 : rmtBufferVec.push_back(nullptr);
334 0 : rmtRmaBufferVec.push_back((nullptr));
335 : } else { // size非0，则构造一个remote buffer
336 0 : rmtBufferVec.push_back(make_unique<RemoteIpcRmaBuffer>(dto));
337 0 : rmtRmaBufferVec.push_back(rmtBufferVec.back().get());
338 0 : HCCL_INFO("unpack buffer pos=%u, rmtRmaBuffer=%s", pos, rmtBufferVec.back()->Describe().c_str());
339 : }
340 0 : }
341 0 : }
342 :
343 0 : std::vector<char> P2PTransport::GetUniqueId()
344 : {
345 0 : if (baseStatus != TransportStatus::READY) {
346 0 : MACRO_THROW(InternalException, StringFormat("transport status is not ready, please check"));
347 : }
348 0 : u32 type = static_cast<u32>(transportType);
349 0 : BinaryStream binaryStream;
350 0 : binaryStream << type;
351 0 : binaryStream << notifyNum;
352 0 : binaryStream << bufferNum;
353 :
354 : // [header...][notifyUniqueId...][rmtNotifyUniqueId...][rmtBufferUniqueIds...]
355 0 : auto notifyUniqueIds = GetNotifyUniqueIds();
356 0 : binaryStream << notifyUniqueIds;
357 :
358 0 : auto rmtNotifyUniqueIds = GetRmtNotifyUniqueIds();
359 0 : binaryStream << rmtNotifyUniqueIds;
360 :
361 0 : auto rmtBufferUniqueIds = GetRmtBufferUniqueIds();
362 0 : binaryStream << rmtBufferUniqueIds;
363 :
364 0 : std::vector<char> result;
365 0 : binaryStream.Dump(result);
366 0 : return result;
367 0 : }
368 :
369 0 : std::vector<char> P2PTransport::GetUniqueIdV2()
370 : {
371 0 : if (baseStatus != TransportStatus::READY) {
372 0 : MACRO_THROW(InternalException, StringFormat("transport status is not ready, please check"));
373 : }
374 0 : u32 type = static_cast<u32>(transportType);
375 0 : BinaryStream binaryStream;
376 0 : binaryStream << type;
377 0 : binaryStream << notifyNum;
378 0 : binaryStream << bufferNum;
379 :
380 : // [header...][notifyUniqueId...][rmtNotifyUniqueId...][rmtBufferUniqueIds...]
381 0 : auto notifyUniqueIds = GetNotifyUniqueIds();
382 0 : binaryStream << notifyUniqueIds;
383 :
384 0 : auto rmtNotifyUniqueIds = GetRmtNotifyUniqueIds();
385 0 : binaryStream << rmtNotifyUniqueIds;
386 :
387 0 : auto locBufferUniqueIds = GetLocBufferUniqueIds();
388 0 : binaryStream << locBufferUniqueIds;
389 :
390 0 : auto rmtBufferUniqueIds = GetRmtBufferUniqueIds();
391 0 : binaryStream << rmtBufferUniqueIds;
392 :
393 0 : std::vector<char> result;
394 0 : binaryStream.Dump(result);
395 0 : return result;
396 0 : }
397 :
398 0 : std::vector<char> P2PTransport::GetNotifyUniqueIds()
399 : {
400 0 : HCCL_INFO("start packing all notify uniqueIds");
401 0 : std::vector<char> result(0);
402 0 : for (auto &it : commonLocRes.notifyVec) {
403 0 : HCCL_INFO("p2pMemTransport Notify %s", it->Describe().c_str());
404 0 : auto uniqueId = it->GetUniqueId();
405 0 : result.insert(result.end(), uniqueId.begin(), uniqueId.end());
406 0 : }
407 0 : return result;
408 0 : }
409 :
410 0 : std::vector<char> P2PTransport::GetSingleRmtNotifyUniqueId(u64 addr, u64 size, u32 notifyId) const
411 : {
412 0 : BinaryStream binaryStream;
413 0 : binaryStream << addr;
414 0 : binaryStream << size;
415 0 : binaryStream << notifyId;
416 0 : HCCL_INFO("P2PTransport RmtNotifyAddr[addr=0x%llx, size=0x%llx, notifyId=%u]", addr, size, notifyId);
417 0 : std::vector<char> result;
418 0 : binaryStream.Dump(result);
419 0 : return result;
420 0 : }
421 :
422 0 : std::vector<char> P2PTransport::GetRmtNotifyUniqueIds() const
423 : {
424 0 : HCCL_INFO("start packing all remote notify uniqueIds");
425 0 : std::vector<char> result(0);
426 0 : for (auto &it : rmtNotifyVec) {
427 0 : std::vector<char> uniqueId;
428 0 : if (it != nullptr) {
429 0 : uniqueId = GetSingleRmtNotifyUniqueId(it->GetAddr(), it->GetSize(), it->GetId());
430 0 : HCCL_INFO("P2PTransport::GetRmtNotifyUniqueIds, %s", it->Describe().c_str());
431 : } else {
432 0 : uniqueId = GetSingleRmtNotifyUniqueId(0, 0, 0); // 填充一个空的buffer
433 0 : HCCL_INFO("P2PTransport::GetRmtNotifyUniqueIds, null buffer");
434 : }
435 0 : result.insert(result.end(), uniqueId.begin(), uniqueId.end());
436 0 : }
437 0 : return result;
438 0 : }
439 :
440 0 : std::vector<char> P2PTransport::GetSingleBufferUniqueId(u64 addr, u64 size) const
441 : {
442 0 : BinaryStream binaryStream;
443 0 : binaryStream << addr;
444 0 : binaryStream << size;
445 0 : HCCL_INFO("P2PTransport BufferAddr[addr=0x%llx, size=0x%llx]", addr, size);
446 0 : std::vector<char> result;
447 0 : binaryStream.Dump(result);
448 0 : return result;
449 0 : }
450 :
451 0 : std::vector<char> P2PTransport::GetLocBufferUniqueIds() const
452 : {
453 0 : HCCL_INFO("start packing all local buffer uniqueIds");
454 0 : std::vector<char> result(0);
455 0 : for (auto &it : commonLocRes.bufferVec) {
456 0 : std::vector<char> uniqueId;
457 0 : if (it != nullptr) {
458 0 : uniqueId = GetSingleBufferUniqueId(it->GetAddr(), it->GetSize());
459 0 : HCCL_INFO("P2PTransport::GetLocBufferUniqueIds, %s", it->Describe().c_str());
460 : } else {
461 0 : uniqueId = GetSingleBufferUniqueId(0, 0); // 填充一个空的buffer
462 0 : HCCL_INFO("P2PTransport::GetLocBufferUniqueIds, null buffer");
463 : }
464 0 : result.insert(result.end(), uniqueId.begin(), uniqueId.end());
465 0 : }
466 0 : return result;
467 0 : }
468 :
469 0 : std::vector<char> P2PTransport::GetRmtBufferUniqueIds() const
470 : {
471 0 : HCCL_INFO("start packing all remote buffer uniqueIds");
472 0 : std::vector<char> result(0);
473 0 : for (auto &it : rmtBufferVec) {
474 0 : std::vector<char> uniqueId;
475 0 : if (it != nullptr) {
476 0 : uniqueId = GetSingleBufferUniqueId(it->GetAddr(), it->GetSize());
477 0 : HCCL_INFO("P2PTransport::GetRmtBufferUniqueIds, %s", it->Describe().c_str());
478 : } else {
479 0 : uniqueId = GetSingleBufferUniqueId(0, 0); // 填充一个空的buffer
480 0 : HCCL_INFO("P2PTransport::GetRmtBufferUniqueIds, null buffer");
481 : }
482 0 : result.insert(result.end(), uniqueId.begin(), uniqueId.end());
483 0 : }
484 0 : return result;
485 0 : }
486 :
487 0 : HcclResult P2PTransport::GetRemoteMems(uint32_t *memNum, CommMem **remoteMem, char ***memInfos)
488 : {
489 0 : std::lock_guard<std::mutex> lock(remoteMemsMutex_);
490 0 : Hccl::RemoteMemCtx<std::unique_ptr<RemoteIpcRmaBuffer>> remoteMemCtx{cacheValid_, rmtBufferVec,
491 0 : remoteUserMems_, memInfoCopies_, memInfoPointers_, remoteMem, memInfos, memNum};
492 0 : CHK_RET(GetRemoteUserMems(remoteMemCtx));
493 0 : return HCCL_SUCCESS;
494 0 : }
495 : } // namespace Hccl
|
