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