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 "transport_p2p.h"
12 : #include <securec.h>
13 : #include <sys/socket.h>
14 : #include <sys/types.h>
15 : #include <arpa/inet.h>
16 : #include <unistd.h>
17 :
18 : #include "mem_name_repository_pub.h"
19 : #include "adapter_rts.h"
20 : #include "mem_host_pub.h"
21 :
22 : namespace hccl {
23 : std::array<DeviceMem, MAX_MODULE_DEVICE_NUM> TransportP2p::notifyValueMem_;
24 : std::array<std::mutex, MAX_MODULE_DEVICE_NUM> TransportP2p::notifyValueMutex_;
25 : std::array<Referenced, MAX_MODULE_DEVICE_NUM> TransportP2p::instanceRef_;
26 4 : TransportP2p::TransportP2p(
27 : DispatcherPub* dispatcher, const std::unique_ptr<NotifyPool>& notifyPool, MachinePara& machinePara,
28 4 : std::chrono::milliseconds timeout)
29 : : TransportBase(dispatcher, notifyPool, machinePara, timeout),
30 4 : remoteInputPtr_(nullptr),
31 4 : remoteOutputPtr_(nullptr),
32 4 : remoteOutputOffsetValue_(0),
33 4 : remoteInputOffsetValue_(0),
34 4 : remoteOutputMemName_(),
35 8 : remoteInputMemName_()
36 : {
37 4 : if (machinePara_.deviceLogicId >= 0 && (static_cast<u32>(machinePara_.deviceLogicId) < MAX_MODULE_DEVICE_NUM)) {
38 4 : instanceRef_[machinePara_.deviceLogicId].Ref();
39 : }
40 4 : userLocalNotify_.resize(notifyNum_);
41 4 : userRemoteNotify_.resize(notifyNum_);
42 4 : userRemoteNotifyAddr_.resize(notifyNum_);
43 4 : userRemoteNotifyOffset_.resize(notifyNum_);
44 4 : remoteIpcMemPtrVector_.resize(machinePara.mem.size());
45 4 : remoteIpcMemOffsetValueVector_.resize(machinePara.mem.size());
46 4 : remoteIpcMemSizeVector_.resize(machinePara.mem.size());
47 4 : remoteIpcMemNameVector_.resize(machinePara.mem.size());
48 4 : }
49 :
50 6 : TransportP2p::~TransportP2p()
51 : {
52 4 : HCCL_DEBUG("~TransportP2p Enter!");
53 :
54 : // 关闭rtIpcOpenMemory打开的对端共享内存和内存名称映射
55 4 : if (!isMemInclude_) {
56 : MemNameRepository::GetInstance(machinePara_.deviceLogicId)
57 4 : ->CloseIpcMem(static_cast<const u8*>(remoteOutputMemName_.ipcName));
58 4 : HCCL_DEBUG("remoteOutputMemName_.ipcName[%d]", remoteOutputMemName_.ipcName);
59 : MemNameRepository::GetInstance(machinePara_.deviceLogicId)
60 4 : ->CloseIpcMem(static_cast<const u8*>(remoteInputMemName_.ipcName));
61 4 : HCCL_DEBUG("remoteInputMemName_.ipcName[%d]", remoteInputMemName_.ipcName);
62 : }
63 4 : for (u32 i = 0; i < machinePara_.mem.size(); i++) {
64 : MemNameRepository::GetInstance(machinePara_.deviceLogicId)
65 0 : ->CloseIpcMem(static_cast<const u8*>(remoteIpcMemNameVector_[i].ipcName));
66 0 : HCCL_DEBUG("remoteIpcMemNameVector_[%u].ipcName[%s]", i, remoteIpcMemNameVector_[i].ipcName);
67 : }
68 :
69 : // 关闭rtIpcSetMemoryName 设置的内存名
70 4 : if (!isMemInclude_) {
71 : MemNameRepository::GetInstance(machinePara_.deviceLogicId)
72 4 : ->DestroyIpcMem(machinePara_.outputMem.ptr(), machinePara_.outputMem.size(), isSioToHccs_);
73 4 : HCCL_DEBUG(
74 : "machinePara_.outputMem addr:[%p], size:[%llu]", machinePara_.outputMem.ptr(),
75 : machinePara_.outputMem.size());
76 : MemNameRepository::GetInstance(machinePara_.deviceLogicId)
77 4 : ->DestroyIpcMem(machinePara_.inputMem.ptr(), machinePara_.inputMem.size(), isSioToHccs_);
78 4 : HCCL_DEBUG(
79 : "machinePara_.inputMem addr:[%p], size:[%llu]", machinePara_.inputMem.ptr(), machinePara_.inputMem.size());
80 : }
81 4 : for (u32 i = 0; i < machinePara_.mem.size(); i++) {
82 : MemNameRepository::GetInstance(machinePara_.deviceLogicId)
83 0 : ->DestroyIpcMem(machinePara_.mem[i].ptr(), machinePara_.mem[i].size(), isSioToHccs_);
84 0 : HCCL_DEBUG(
85 : "machinePara_.mem[%u] addr:[%p], size:[%llu]", machinePara_.mem[i].ptr(), machinePara_.mem[i].size());
86 : }
87 :
88 4 : SignalDestroy();
89 :
90 4 : if (machinePara_.deviceLogicId >= 0 && (static_cast<u32>(machinePara_.deviceLogicId) < MAX_MODULE_DEVICE_NUM)) {
91 4 : if (instanceRef_[machinePara_.deviceLogicId].Unref() == 0) {
92 4 : std::unique_lock<std::mutex> lock(notifyValueMutex_[machinePara_.deviceLogicId]);
93 4 : notifyValueMem_[machinePara_.deviceLogicId].free();
94 4 : }
95 : }
96 4 : HCCL_DEBUG("~TransportP2p Success!");
97 6 : }
98 :
99 2 : HcclResult TransportP2p::Init()
100 : {
101 2 : HCCL_INFO(
102 : "machineType=[%d], serverId=[%s], localDeviceId=[%d], remoteDeviceId=[%d], "
103 : "localRank=[%u], localUserRank=[%u], remoteRank=[%u], remoteUserRank=[%u], "
104 : "deviceType=[%d], input_ptr=[%p], output_ptr=[%p], linkAttribute=[0x%x], linkMode=[%d], "
105 : "notifyNum[%u], isIndOp[%d], custom exchange data size [%llu], specifyLink[%d].",
106 : machinePara_.machineType, machinePara_.serverId.c_str(), machinePara_.localDeviceId,
107 : machinePara_.remoteDeviceId, machinePara_.localUserrank, machinePara_.localWorldRank,
108 : machinePara_.remoteUserrank, machinePara_.remoteWorldRank, machinePara_.deviceType, machinePara_.inputMem.ptr(),
109 : machinePara_.outputMem.ptr(), machinePara_.linkAttribute, machinePara_.linkMode, machinePara_.notifyNum,
110 : machinePara_.isIndOp, machinePara_.exchangeInfo.size(), machinePara_.specifyLink);
111 2 : HcclUs startut = TIME_NOW();
112 :
113 : /* make input memory shared interprocess and assigned a name */
114 2 : if (!machinePara_.isNewOneSide) {
115 0 : CHK_SMART_PTR_NULL(machinePara_.inputMem);
116 0 : CHK_SMART_PTR_NULL(machinePara_.outputMem);
117 : }
118 :
119 2 : CHK_PTR_NULL(dispatcher_);
120 2 : CHK_SMART_PTR_NULL(notifyPool_);
121 2 : CHK_RET(CheckDeviceId());
122 2 : CHK_RET(CheckExchangeData());
123 2 : SetMemIncludeFlag();
124 : // 上层初始化时保证 machinePara_.sockets 非空
125 2 : if (machinePara_.sockets.size() == 0) {
126 0 : HCCL_ERROR("machinePara sockets is empty.");
127 0 : return HCCL_E_INTERNAL;
128 : }
129 2 : defaultSocket_ = machinePara_.sockets[0];
130 2 : CHK_PTR_NULL(defaultSocket_);
131 :
132 2 : CHK_RET(CheckLinkMode());
133 :
134 : /* 本端与远端交换tgid 信息 */
135 2 : CHK_RET(ExchangeTgidMesg()); // tgid 无法合并交换，因为依赖对端的tgid判定是同一个进程还是跨进程
136 :
137 2 : CHK_RET(SetLinkType()); // 需要在交换sdid之后调用，确定是否超节点内节点间HCCS场景
138 :
139 2 : CHK_RET(FillExchangeDataTotalSize());
140 :
141 2 : CHK_RET(ConstructExchangeForSend());
142 :
143 2 : HcclResult ret = defaultSocket_->Send(exchangeDataForSend_.data(), exchangeDataTotalSize_);
144 2 : CHK_PRT_RET(
145 : ret != HCCL_SUCCESS,
146 : HCCL_ERROR(
147 : "[TransportP2p][Init] failed to send exchangeData exchangeDataTotalSize[%llu], custom exchange data "
148 : "size [%llu].",
149 : exchangeDataTotalSize_, machinePara_.exchangeInfo.size()),
150 : ret);
151 :
152 2 : exchangeDataForRecv_.resize(exchangeDataTotalSize_);
153 2 : ret = defaultSocket_->Recv(exchangeDataForRecv_.data(), exchangeDataTotalSize_);
154 2 : CHK_PRT_RET(
155 : ret != HCCL_SUCCESS,
156 : HCCL_ERROR(
157 : "[TransportP2p][Init] failed to recv exchangeData exchangeDataTotalSize[%llu], custom exchange data "
158 : "size [%llu].",
159 : exchangeDataTotalSize_, machinePara_.exchangeInfo.size()),
160 : ret);
161 :
162 2 : HCCL_DEBUG("[TransportP2p][Init] Socket Data Received");
163 :
164 2 : CHK_RET(ParseReceivedExchangeData());
165 :
166 2 : SetTransportRelationship();
167 2 : SetUseSdmaToSignalRecord();
168 2 : CHK_RET(CreateNotifyValueBuffer());
169 :
170 2 : HcclUs endut = TIME_NOW();
171 2 : HCCL_INFO("Time:%lld us", DURATION_US(endut - startut));
172 :
173 2 : HCCL_USER_CRITICAL_LOG(
174 : "create hccl transport:communicator[%s], local rank[%u], remote rank[%u], "
175 : "transporttype[%s]",
176 : machinePara_.tag.c_str(), machinePara_.localUserrank, machinePara_.remoteUserrank,
177 : GetLinkTypeEnumStr(GetLinkType()).c_str());
178 :
179 2 : return HCCL_SUCCESS;
180 : }
181 :
182 4 : void TransportP2p::SetUseSdmaToSignalRecord()
183 : {
184 : // AICPU展开时，在节点间使用SDMA进行notify record操作，STARS可检出节点间链路异常，触发HCCL重执行
185 : useSdmaToSignalRecord_
186 8 : = ((transportAttr_.relationship & HCCL_TRANSPORT_RELATIONSHIP_SAME_SERVER) == 0)
187 4 : && ((transportAttr_.linkType == LinkType::LINK_HCCS_SW) || (transportAttr_.linkType == LinkType::LINK_HCCS));
188 4 : }
189 :
190 2 : HcclResult TransportP2p::ParseSpecifyLink(LinkTypeInServer& linkType)
191 : {
192 2 : if (machinePara_.specifyLink == LinkTypeInServer::RESERVED_LINK_TYPE || machinePara_.specifyLink == linkType) {
193 2 : return HCCL_SUCCESS; // 未指定切换链路，保持默认
194 0 : } else if (machinePara_.specifyLink == LinkTypeInServer::HCCS_SW_TYPE && linkType == LinkTypeInServer::SIO_TYPE) {
195 : // 切换链路基于ipc实现, 多线程场景暂不支持
196 0 : s32 sendPid = 0;
197 0 : CHK_RET(SalGetBareTgid(&sendPid));
198 0 : CHK_PRT_RET(
199 : sendPid == recvPid_, HCCL_WARNING("%s specifyLink is not supported in multi-thread", __func__),
200 : HCCL_SUCCESS);
201 :
202 : // A3 DIE间通信场景, 将链路从SIO切换到HCCS
203 0 : linkType = LinkTypeInServer::HCCS_SW_TYPE;
204 0 : isSioToHccs_ = true;
205 0 : HCCL_INFO("%s specifyLink change to HCCS_SW_TYPE", __func__);
206 0 : } else {
207 0 : HCCL_ERROR(
208 : "%s fail, linkType:%d, specifyLink:%d is not supported", __func__, linkType, machinePara_.specifyLink);
209 0 : return HCCL_E_NOT_SUPPORT;
210 : }
211 0 : return HCCL_SUCCESS;
212 : }
213 :
214 2 : HcclResult TransportP2p::SetLinkType()
215 : {
216 : // 计算linkType
217 2 : LinkTypeInServer linkType = LinkTypeInServer::HCCS_TYPE;
218 2 : if (recvSdid_ != INVALID_INT) { // 超节点内节点间走p2p通信时，链路类型为LINK_HCCS_SW
219 0 : linkType = LinkTypeInServer::HCCS_SW_TYPE;
220 : } else {
221 2 : CHK_RET(hrtGetPairDeviceLinkType(
222 : static_cast<u32>(machinePara_.localDeviceId), static_cast<u32>(machinePara_.remoteDeviceId), linkType));
223 : }
224 :
225 2 : CHK_RET(ParseSpecifyLink(linkType));
226 :
227 2 : switch (linkType) {
228 2 : case LinkTypeInServer::HCCS_TYPE:
229 2 : transportAttr_.linkType = hccl::LinkType::LINK_HCCS;
230 2 : break;
231 0 : case LinkTypeInServer::HCCS_SW_TYPE:
232 0 : transportAttr_.linkType = hccl::LinkType::LINK_HCCS_SW;
233 0 : break;
234 0 : case LinkTypeInServer::SIO_TYPE:
235 0 : transportAttr_.linkType = hccl::LinkType::LINK_SIO;
236 0 : break;
237 0 : default:
238 0 : transportAttr_.linkType = hccl::LinkType::LINK_PCIE;
239 0 : break;
240 : }
241 :
242 2 : HCCL_DEBUG("[TransportP2p] transportattr linktype: 0x%x", transportAttr_.linkType);
243 2 : return HCCL_SUCCESS;
244 : }
245 :
246 2 : HcclResult TransportP2p::CreateNotifyValueBuffer()
247 : {
248 2 : if (!useSdmaToSignalRecord_) {
249 2 : return HCCL_SUCCESS;
250 : }
251 :
252 0 : u32 notifySize = 0;
253 0 : CHK_RET(hrtGetNotifySize(notifySize));
254 0 : std::unique_lock<std::mutex> lock(notifyValueMutex_[machinePara_.deviceLogicId]);
255 0 : if (notifyValueMem_[machinePara_.deviceLogicId].ptr() == nullptr) {
256 0 : u64 notifyVaule = 1; // notify值写1表示record
257 0 : CHK_RET(DeviceMem::alloc(notifyValueMem_[machinePara_.deviceLogicId], notifyValueSize_));
258 0 : HCCL_DEBUG(
259 : "create notify value buffer[%p], size[%u]", notifyValueMem_[machinePara_.deviceLogicId].ptr(), notifySize);
260 :
261 0 : CHK_RET(hrtMemSyncCopy(
262 : notifyValueMem_[machinePara_.deviceLogicId].ptr(), notifyValueMem_[machinePara_.deviceLogicId].size(),
263 : ¬ifyVaule, notifySize, HcclRtMemcpyKind::HCCL_RT_MEMCPY_KIND_HOST_TO_DEVICE));
264 : }
265 0 : transportAttr_.signalRecordBuff.address = reinterpret_cast<u64>(notifyValueMem_[machinePara_.deviceLogicId].ptr());
266 0 : transportAttr_.signalRecordBuff.length = notifySize;
267 :
268 0 : HCCL_DEBUG(
269 : "[TransportP2p] transportattr signalRecordBuff.address[%p], signalRecordBuff.length[%llu]",
270 : transportAttr_.signalRecordBuff.address, transportAttr_.signalRecordBuff.length);
271 0 : return HCCL_SUCCESS;
272 0 : }
273 :
274 2 : void TransportP2p::SetTransportRelationship()
275 : {
276 2 : if (transportAttr_.linkType == hccl::LinkType::LINK_SIO) {
277 : // 芯片内
278 0 : transportAttr_.relationship |= HCCL_TRANSPORT_RELATIONSHIP_SAME_CHIP;
279 0 : transportAttr_.relationship |= HCCL_TRANSPORT_RELATIONSHIP_SAME_SERVER;
280 0 : transportAttr_.relationship |= HCCL_TRANSPORT_RELATIONSHIP_SAME_SUPERPOD;
281 2 : } else if (recvSdid_ == INVALID_INT) {
282 : // 节点内
283 2 : transportAttr_.relationship |= HCCL_TRANSPORT_RELATIONSHIP_SAME_SERVER;
284 2 : transportAttr_.relationship |= HCCL_TRANSPORT_RELATIONSHIP_SAME_SUPERPOD;
285 : } else {
286 : // 节点间
287 0 : transportAttr_.relationship |= HCCL_TRANSPORT_RELATIONSHIP_SAME_SUPERPOD;
288 : }
289 :
290 2 : HCCL_DEBUG("[TransportP2p] transportattr relationship: 0x%x", transportAttr_.relationship);
291 2 : return;
292 : }
293 :
294 2 : HcclResult TransportP2p::FillExchangeDataTotalSize()
295 : {
296 2 : exchangeDataTotalSize_ = 0;
297 2 : s32 sendPid = 0;
298 2 : CHK_RET(SalGetBareTgid(&sendPid));
299 2 : u64 ipcMemDataSize = 0;
300 2 : if (sendPid != recvPid_ || recvSdid_ != INVALID_INT) {
301 : // 输入输出内存
302 0 : HCCL_DEBUG("[TransportP2p][FillExchangeDataTotalSize] Inter Proc");
303 0 : ipcMemDataSize = HCCL_IPC_MEM_NAME_LEN + sizeof(u64) + sizeof(u64); // size + offset
304 0 : if (!isMemInclude_) {
305 0 : exchangeInfoSize_.ipcMenSize = ipcMemDataSize * (2 + machinePara_.mem.size());
306 : } else {
307 : // in和out包含在整块CCLbuf的时候，不需要传ipcName,但是size和offset不能少
308 0 : exchangeInfoSize_.ipcMenSize = ipcMemDataSize * machinePara_.mem.size() + 2 * (sizeof(u64) + sizeof(u64));
309 : }
310 : } else {
311 2 : HCCL_DEBUG("[TransportP2p][FillExchangeDataTotalSize] intra Proc");
312 2 : ipcMemDataSize = sizeof(u64) + sizeof(u64); // addr + length
313 : exchangeInfoSize_.ipcMenSize
314 2 : = ipcMemDataSize * (2 + machinePara_.mem.size()); // 2: input & output + mem.size()
315 : }
316 :
317 2 : if (!machinePara_.isNewOneSide) {
318 : // notify 信息
319 0 : if (machinePara_.linkMode != LinkMode::LINK_SIMPLEX_MODE
320 0 : || machinePara_.machineType == MachineType::MACHINE_CLIENT_TYPE) {
321 0 : exchangeInfoSize_.notifySize = NOTIFY_INFO_LENGTH;
322 : }
323 0 : if (machinePara_.linkMode != LinkMode::LINK_SIMPLEX_MODE
324 0 : || machinePara_.machineType == MachineType::MACHINE_SERVER_TYPE) {
325 0 : exchangeInfoSize_.notifySize += NOTIFY_INFO_LENGTH;
326 : }
327 : // 3.新增notify资源
328 0 : exchangeInfoSize_.notifySize += NOTIFY_INFO_LENGTH * notifyNum_;
329 : }
330 :
331 : // 自定义信息
332 2 : exchangeInfoSize_.exDataSize = machinePara_.exchangeInfo.size();
333 :
334 : // 独立算子内存
335 2 : if (machinePara_.isIndOp) {
336 : // userDeviceMem数量\userDeviceMem\userHostMem数量\userHostMem
337 0 : const int kMemCountItems = 2;
338 : exchangeInfoSize_.indOpMemSize
339 0 : = ipcMemDataSize * (machinePara_.userDeviceMem.size() + machinePara_.userHostMem.size());
340 0 : exchangeInfoSize_.indOpMemSize += sizeof(u64) * kMemCountItems;
341 : }
342 :
343 2 : exchangeDataTotalSize_ = exchangeInfoSize_.ipcMenSize + exchangeInfoSize_.notifySize + exchangeInfoSize_.exDataSize
344 2 : + exchangeInfoSize_.indOpMemSize + sizeof(ExchangeInfoSize);
345 2 : HCCL_INFO(
346 : "[TransportP2p][FillExchangeDataTotalSize] exchangeDataTotalSize[%llu] memSize[%d]", exchangeDataTotalSize_,
347 : machinePara_.mem.size());
348 2 : return HCCL_SUCCESS;
349 : }
350 :
351 2 : HcclResult TransportP2p::ConstructExchangeForSend()
352 : {
353 2 : exchangeDataForSend_.resize(exchangeDataTotalSize_);
354 2 : u8* exchangeDataPtr = exchangeDataForSend_.data();
355 2 : u64 exchangeDataBlankSize = exchangeDataTotalSize_;
356 2 : CHK_RET(ConstructDataLenForSend(exchangeDataPtr, exchangeDataBlankSize));
357 2 : u64 blankSizeRecord = exchangeDataBlankSize;
358 :
359 2 : s32 sendPid = 0;
360 2 : CHK_RET(SalGetBareTgid(&sendPid));
361 2 : HCCL_DEBUG("%s sendPid %d, recvPid %d, recvSdid %d", __func__, sendPid, recvPid_, recvSdid_);
362 2 : if (sendPid != recvPid_ || recvSdid_ != INVALID_INT) { // 跨进程方式交换
363 : // 构造IPC内存地址交换数据结构
364 0 : for (auto ipcMem : machinePara_.mem) {
365 0 : CHK_RET(ConstructIpcMemInfoForSend(ipcMem.ptr(), ipcMem.size(), exchangeDataPtr, exchangeDataBlankSize));
366 0 : }
367 0 : if (!isMemInclude_) {
368 0 : CHK_RET(ConstructIpcMemInfoForSend(
369 : machinePara_.outputMem.ptr(), machinePara_.outputMem.size(), exchangeDataPtr, exchangeDataBlankSize));
370 0 : CHK_RET(ConstructIpcMemInfoForSend(
371 : machinePara_.inputMem.ptr(), machinePara_.inputMem.size(), exchangeDataPtr, exchangeDataBlankSize));
372 : } else {
373 0 : CHK_RET(ConstructMemIncludeInfoForSend(exchangeDataPtr, exchangeDataBlankSize));
374 : }
375 0 : } else {
376 : // 构造进程内内存地址交换数据结构
377 2 : CHK_RET(ConstructIntraProcMemInfoForSend(
378 : machinePara_.outputMem.ptr(), machinePara_.outputMem.size(), exchangeDataPtr, exchangeDataBlankSize));
379 2 : CHK_RET(ConstructIntraProcMemInfoForSend(
380 : machinePara_.inputMem.ptr(), machinePara_.inputMem.size(), exchangeDataPtr, exchangeDataBlankSize));
381 2 : for (auto ipcMem : machinePara_.mem) {
382 0 : CHK_RET(
383 : ConstructIntraProcMemInfoForSend(ipcMem.ptr(), ipcMem.size(), exchangeDataPtr, exchangeDataBlankSize));
384 0 : }
385 : }
386 2 : CHK_RET(SumCheckSizeAndConsisten(
387 : ExInfoType::EX_IPCMEN_SIZE, exchangeInfoSize_.ipcMenSize, blankSizeRecord, exchangeDataBlankSize));
388 :
389 2 : CHK_RET(ConstructNotifyInfoForSend(exchangeDataPtr, exchangeDataBlankSize));
390 2 : CHK_RET(ConstructNotifyVectorInfoForSend(exchangeDataPtr, exchangeDataBlankSize)); // 新增notify资源的创建
391 2 : CHK_RET(SumCheckSizeAndConsisten(
392 : ExInfoType::EX_NOTIFY_SIZE, exchangeInfoSize_.notifySize, blankSizeRecord, exchangeDataBlankSize));
393 :
394 2 : CHK_RET(ConstructExchangeDataForSend(exchangeDataPtr, exchangeDataBlankSize));
395 2 : CHK_RET(SumCheckSizeAndConsisten(
396 : ExInfoType::EX_EXDATA_SIZE, exchangeInfoSize_.exDataSize, blankSizeRecord, exchangeDataBlankSize));
397 :
398 : // 独立算子内存资源，无需检查大小
399 2 : if (machinePara_.isIndOp) {
400 0 : if (sendPid != recvPid_ || recvSdid_ != INVALID_INT) { // 跨进程方式交换
401 0 : CHK_RET(ConstructNumInfoForSend(machinePara_.userDeviceMem.size(), exchangeDataPtr, exchangeDataBlankSize));
402 0 : for (auto ipcMem : machinePara_.userDeviceMem) {
403 0 : CHK_RET(
404 : ConstructIpcMemInfoForSend(ipcMem.ptr(), ipcMem.size(), exchangeDataPtr, exchangeDataBlankSize));
405 0 : }
406 0 : CHK_RET(ConstructNumInfoForSend(machinePara_.userHostMem.size(), exchangeDataPtr, exchangeDataBlankSize));
407 0 : for (auto ipcMem : machinePara_.userHostMem) {
408 0 : CHK_RET(
409 : ConstructIpcMemInfoForSend(ipcMem.ptr(), ipcMem.size(), exchangeDataPtr, exchangeDataBlankSize));
410 0 : }
411 0 : } else {
412 0 : CHK_RET(ConstructNumInfoForSend(machinePara_.userDeviceMem.size(), exchangeDataPtr, exchangeDataBlankSize));
413 0 : for (auto ipcMem : machinePara_.userDeviceMem) {
414 0 : CHK_RET(ConstructIntraProcMemInfoForSend(
415 : ipcMem.ptr(), ipcMem.size(), exchangeDataPtr, exchangeDataBlankSize));
416 0 : }
417 0 : CHK_RET(ConstructNumInfoForSend(machinePara_.userHostMem.size(), exchangeDataPtr, exchangeDataBlankSize));
418 0 : for (auto ipcMem : machinePara_.userHostMem) {
419 0 : CHK_RET(ConstructIntraProcMemInfoForSend(
420 : ipcMem.ptr(), ipcMem.size(), exchangeDataPtr, exchangeDataBlankSize));
421 0 : }
422 : }
423 : }
424 2 : if (exchangeDataBlankSize != 0) {
425 0 : HCCL_ERROR(
426 : "[TransportP2p][ConstructExchangeForSend] failed to construct exchange Data "
427 : "exchangeDataBlankSize[%llu]",
428 : exchangeDataBlankSize);
429 0 : return HCCL_E_INTERNAL;
430 : }
431 2 : return HCCL_SUCCESS; // this function should not be called in normal process
432 : }
433 :
434 : // exchangeDataPtr对指针进行了引用，因为需要改变exchangeDataPtr的值
435 : HcclResult
436 0 : TransportP2p::ConstructIpcMemInfoForSend(void* ptr, u64 size, u8*& exchangeDataPtr, u64& exchangeDataBlankSize)
437 : {
438 : HcclResult ret;
439 : u64 memOffset;
440 0 : SecIpcName_t memName;
441 :
442 0 : if (!machinePara_.isNewOneSide) {
443 : ret = MemNameRepository::GetInstance(machinePara_.deviceLogicId)
444 0 : ->SetIpcMem(
445 0 : ptr, size, memName.ipcName, HCCL_IPC_MEM_NAME_LEN, memOffset, recvPid_, recvSdid_, isSioToHccs_);
446 0 : CHK_PRT_RET(
447 : ret != HCCL_SUCCESS,
448 : HCCL_ERROR(
449 : "[Send][IpcMemMesg]errNo[0x%016llx], In send ipc mesg, get para mem name failed. "
450 : "mem addr[%p] local rank[%u]",
451 : HCCL_ERROR_CODE(ret), machinePara_.outputMem.ptr(), machinePara_.localUserrank),
452 : ret);
453 : }
454 :
455 : // 设置ipc mem属性，指定通信链路从sio切换至hccs
456 0 : if (isSioToHccs_) {
457 0 : u32 ipcAttr = 1; // 0: SIO(默认), 1: HCCS
458 0 : CHK_RET(hrtIpcSetMemoryAttr(memName.ipcName, ACL_RT_IPC_MEM_ATTR_ACCESS_LINK, ipcAttr));
459 : }
460 :
461 0 : CHK_SAFETY_FUNC_RET(memcpy_s(exchangeDataPtr, exchangeDataBlankSize, memName.ipcName, HCCL_IPC_MEM_NAME_LEN));
462 0 : exchangeDataPtr += HCCL_IPC_MEM_NAME_LEN;
463 0 : exchangeDataBlankSize -= HCCL_IPC_MEM_NAME_LEN;
464 0 : CHK_SAFETY_FUNC_RET(memcpy_s(exchangeDataPtr, exchangeDataBlankSize, &size, sizeof(u64)));
465 0 : exchangeDataPtr += sizeof(u64);
466 0 : exchangeDataBlankSize -= sizeof(u64);
467 0 : CHK_SAFETY_FUNC_RET(memcpy_s(exchangeDataPtr, exchangeDataBlankSize, &memOffset, sizeof(u64)));
468 0 : exchangeDataPtr += sizeof(u64);
469 0 : exchangeDataBlankSize -= sizeof(u64);
470 :
471 0 : return HCCL_SUCCESS;
472 0 : }
473 :
474 : HcclResult
475 4 : TransportP2p::ConstructIntraProcMemInfoForSend(void* ptr, u64 size, u8*& exchangeDataPtr, u64& exchangeDataBlankSize)
476 : {
477 4 : if (!machinePara_.isNewOneSide) {
478 0 : CHK_SAFETY_FUNC_RET(memcpy_s(exchangeDataPtr, exchangeDataBlankSize, &ptr, sizeof(u64)));
479 : }
480 4 : exchangeDataPtr += sizeof(u64);
481 4 : exchangeDataBlankSize -= sizeof(u64);
482 4 : CHK_SAFETY_FUNC_RET(memcpy_s(exchangeDataPtr, exchangeDataBlankSize, &size, sizeof(u64)));
483 4 : exchangeDataPtr += sizeof(u64);
484 4 : exchangeDataBlankSize -= sizeof(u64);
485 :
486 4 : return HCCL_SUCCESS;
487 : }
488 :
489 0 : HcclResult TransportP2p::ConstructNumInfoForSend(u64 num, u8*& exchangeDataPtr, u64& exchangeDataBlankSize)
490 : {
491 0 : CHK_SAFETY_FUNC_RET(memcpy_s(exchangeDataPtr, exchangeDataBlankSize, &num, sizeof(u64)));
492 0 : exchangeDataPtr += sizeof(u64);
493 0 : exchangeDataBlankSize -= sizeof(u64);
494 0 : return HCCL_SUCCESS;
495 : }
496 :
497 0 : HcclResult TransportP2p::ParseMemNumInfo(u64& memNum, u8*& exchangeDataPtr, u64& exchangeDataBlankSize)
498 : {
499 0 : CHK_SAFETY_FUNC_RET(memcpy_s(&memNum, sizeof(u64), exchangeDataPtr, sizeof(u64)));
500 0 : exchangeDataPtr += sizeof(u64);
501 0 : exchangeDataBlankSize -= sizeof(u64);
502 0 : return HCCL_SUCCESS;
503 : }
504 :
505 0 : HcclResult TransportP2p::ParseIpcMemInfo(
506 : void** memPtr, u64& size, u8* memName, u64& offset, u8*& exchangeDataPtr, u64& exchangeDataBlankSize)
507 : {
508 0 : CHK_SAFETY_FUNC_RET(memcpy_s(memName, HCCL_IPC_MEM_NAME_LEN, exchangeDataPtr, HCCL_IPC_MEM_NAME_LEN));
509 0 : exchangeDataPtr += HCCL_IPC_MEM_NAME_LEN;
510 0 : exchangeDataBlankSize -= HCCL_IPC_MEM_NAME_LEN;
511 :
512 0 : CHK_SAFETY_FUNC_RET(memcpy_s(&size, sizeof(u64), exchangeDataPtr, sizeof(u64)));
513 0 : exchangeDataPtr += sizeof(u64);
514 0 : exchangeDataBlankSize -= sizeof(u64);
515 :
516 0 : CHK_SAFETY_FUNC_RET(memcpy_s(&offset, sizeof(u64), exchangeDataPtr, sizeof(u64)));
517 0 : exchangeDataPtr += sizeof(u64);
518 0 : exchangeDataBlankSize -= sizeof(u64);
519 :
520 0 : if (!machinePara_.isNewOneSide) {
521 : /* 根据名字，获取对端IPC 内存 */
522 0 : HcclResult ret = WaitPeerMemConfig(memPtr, const_cast<u8*>(memName), size, offset);
523 0 : CHK_PRT_RET(
524 : ret != HCCL_SUCCESS,
525 : HCCL_ERROR(
526 : "[Recv][IpcMemMesg]errNo[0x%016llx] In recv ipc mem mesg, wait peer mem config "
527 : "failed. local rank[%u]",
528 : HCCL_ERROR_CODE(ret), machinePara_.localUserrank),
529 : ret);
530 :
531 0 : CHK_PTR_NULL(*memPtr);
532 : }
533 :
534 0 : HCCL_DEBUG(
535 : "localUserrank[%u] receive from remoteUserrank[%u]", machinePara_.localUserrank, machinePara_.remoteUserrank);
536 :
537 0 : return HCCL_SUCCESS;
538 : }
539 :
540 4 : HcclResult TransportP2p::ParseIntraProcMemInfo(u64* addr, u64* size, u8*& exchangeDataPtr, u64& exchangeDataBlankSize)
541 : {
542 4 : if (!machinePara_.isNewOneSide) {
543 0 : CHK_SAFETY_FUNC_RET(memcpy_s(addr, sizeof(u64), exchangeDataPtr, sizeof(u64)));
544 0 : CHK_PTR_NULL(reinterpret_cast<void*>(*addr));
545 : }
546 :
547 4 : exchangeDataPtr += sizeof(u64);
548 4 : exchangeDataBlankSize -= sizeof(u64);
549 4 : CHK_SAFETY_FUNC_RET(memcpy_s(size, sizeof(u64), exchangeDataPtr, sizeof(u64)));
550 4 : exchangeDataPtr += sizeof(u64);
551 4 : exchangeDataBlankSize -= sizeof(u64);
552 4 : return HCCL_SUCCESS;
553 : }
554 :
555 0 : HcclResult TransportP2p::ParseNotifyInfo(u8*& exchangeDataPtr, u64& exchangeDataBlankSize)
556 : {
557 0 : s32 sendPid = 0;
558 0 : CHK_RET(SalGetBareTgid(&sendPid)); // 当前进程id
559 0 : HCCL_INFO("LinkRecvNotifyMesg, sendPid[%d], recvPid[%d]", sendPid, recvPid_);
560 :
561 0 : if (machinePara_.isAicpuModeEn) {
562 0 : if ((machinePara_.linkMode != LinkMode::LINK_SIMPLEX_MODE
563 0 : || machinePara_.machineType == MachineType::MACHINE_SERVER_TYPE)
564 0 : && machinePara_.isAicpuModeEn == true) {
565 0 : std::vector<u8> data(NOTIFY_INFO_LENGTH, 0);
566 0 : CHK_SAFETY_FUNC_RET(memcpy_s(&data[0], data.size(), exchangeDataPtr, NOTIFY_INFO_LENGTH));
567 0 : exchangeDataPtr += NOTIFY_INFO_LENGTH;
568 0 : exchangeDataBlankSize -= NOTIFY_INFO_LENGTH;
569 0 : CHK_RET(OpenRemoteNotify(data, remoteSendReadyDeviceNotify_));
570 0 : }
571 :
572 0 : if ((machinePara_.linkMode != LinkMode::LINK_SIMPLEX_MODE
573 0 : || machinePara_.machineType == MachineType::MACHINE_CLIENT_TYPE)
574 0 : && machinePara_.isAicpuModeEn == true) {
575 0 : std::vector<u8> data(NOTIFY_INFO_LENGTH, 0);
576 0 : CHK_SAFETY_FUNC_RET(memcpy_s(&data[0], data.size(), exchangeDataPtr, NOTIFY_INFO_LENGTH));
577 0 : exchangeDataPtr += NOTIFY_INFO_LENGTH;
578 0 : exchangeDataBlankSize -= NOTIFY_INFO_LENGTH;
579 0 : CHK_RET(OpenRemoteNotify(data, remoteSendDoneDeviceNotify_));
580 0 : }
581 : } else {
582 0 : if (machinePara_.linkMode != LinkMode::LINK_SIMPLEX_MODE
583 0 : || machinePara_.machineType == MachineType::MACHINE_SERVER_TYPE) {
584 0 : std::vector<u8> data(NOTIFY_INFO_LENGTH, 0);
585 0 : CHK_SAFETY_FUNC_RET(memcpy_s(&data[0], data.size(), exchangeDataPtr, NOTIFY_INFO_LENGTH));
586 0 : exchangeDataPtr += NOTIFY_INFO_LENGTH;
587 0 : exchangeDataBlankSize -= NOTIFY_INFO_LENGTH;
588 0 : CHK_RET(OpenRemoteNotify(data, remoteSendReadyNotify_));
589 :
590 : HcclSignalInfo notifyInfo;
591 0 : CHK_RET(remoteSendReadyNotify_->GetNotifyData(notifyInfo));
592 0 : CHK_RET(remoteSendReadyNotify_->GetNotifyOffset(remoteSendReadyOffset_));
593 :
594 0 : remoteSendReadyAddress_ = notifyInfo.addr;
595 0 : }
596 :
597 0 : if (machinePara_.linkMode != LinkMode::LINK_SIMPLEX_MODE
598 0 : || machinePara_.machineType == MachineType::MACHINE_CLIENT_TYPE) {
599 0 : std::vector<u8> data(NOTIFY_INFO_LENGTH, 0);
600 0 : CHK_SAFETY_FUNC_RET(memcpy_s(&data[0], data.size(), exchangeDataPtr, NOTIFY_INFO_LENGTH));
601 0 : exchangeDataPtr += NOTIFY_INFO_LENGTH;
602 0 : exchangeDataBlankSize -= NOTIFY_INFO_LENGTH;
603 0 : CHK_RET(OpenRemoteNotify(data, remoteSendDoneNotify_));
604 : HcclSignalInfo notifyInfo;
605 0 : CHK_RET(remoteSendDoneNotify_->GetNotifyData(notifyInfo));
606 0 : CHK_RET(remoteSendDoneNotify_->GetNotifyOffset(remoteSendDoneOffset_));
607 :
608 0 : remoteSendDoneAddress_ = notifyInfo.addr;
609 0 : }
610 : }
611 0 : return HCCL_SUCCESS;
612 : }
613 :
614 2 : HcclResult TransportP2p::ParseNotifyInfoEx(u8*& exchangeDataPtr, u64& exchangeDataBlankSize)
615 : {
616 2 : if (machinePara_.isNewOneSide) {
617 2 : return HCCL_SUCCESS;
618 : }
619 0 : return ParseNotifyInfo(exchangeDataPtr, exchangeDataBlankSize);
620 : }
621 :
622 2 : HcclResult TransportP2p::ParseNotifyVectorInfo(u8*& exchangeDataPtr, u64& exchangeDataBlankSize)
623 : {
624 2 : if (machinePara_.isNewOneSide) {
625 2 : return HCCL_SUCCESS;
626 : }
627 0 : for (u32 i = 0; i < notifyNum_; i++) {
628 0 : std::vector<u8> data(NOTIFY_INFO_LENGTH, 0);
629 0 : CHK_SAFETY_FUNC_RET(memcpy_s(&data[0], data.size(), exchangeDataPtr, NOTIFY_INFO_LENGTH));
630 0 : exchangeDataPtr += NOTIFY_INFO_LENGTH;
631 0 : exchangeDataBlankSize -= NOTIFY_INFO_LENGTH;
632 0 : CHK_RET(OpenRemoteNotify(data, userRemoteNotify_[i]));
633 :
634 0 : if (!machinePara_.isAicpuModeEn) {
635 : HcclSignalInfo notifyInfo;
636 0 : CHK_RET(userRemoteNotify_[i]->GetNotifyData(notifyInfo));
637 0 : CHK_RET(userRemoteNotify_[i]->GetNotifyOffset(userRemoteNotifyOffset_[i]));
638 0 : userRemoteNotifyAddr_[i] = notifyInfo.addr;
639 : }
640 0 : }
641 0 : return HCCL_SUCCESS;
642 : }
643 :
644 : HcclResult
645 2 : TransportP2p::ParseCheckDataLen(ExchangeInfoSize& remoteInfoSize, u8*& exchangeDataPtr, u64& exchangeDataBlankSize)
646 : {
647 2 : CHK_SAFETY_FUNC_RET(memcpy_s(&remoteInfoSize, sizeof(remoteInfoSize), exchangeDataPtr, sizeof(ExchangeInfoSize)));
648 2 : exchangeDataPtr += sizeof(ExchangeInfoSize);
649 2 : exchangeDataBlankSize -= sizeof(ExchangeInfoSize);
650 2 : return HCCL_SUCCESS;
651 : }
652 :
653 2 : HcclResult TransportP2p::ConstructNotifyInfoForSend(u8*& exchangeDataPtr, u64& exchangeDataBlankSize)
654 : {
655 2 : if (machinePara_.isNewOneSide) {
656 2 : return HCCL_SUCCESS;
657 : }
658 0 : if (machinePara_.isAicpuModeEn) {
659 0 : if ((machinePara_.linkMode != LinkMode::LINK_SIMPLEX_MODE
660 0 : || machinePara_.machineType == MachineType::MACHINE_CLIENT_TYPE)) {
661 0 : RemoteRankInfo info(machinePara_.remoteDeviceId, machinePara_.remoteWorldRank, recvPid_, recvSdid_);
662 0 : CHK_RET(
663 : notifyPool_->Alloc(machinePara_.tag, info, localSendReadyDeviceNotify_, NotifyLoadType::DEVICE_NOTIFY));
664 0 : std::vector<u8> data(NOTIFY_INFO_LENGTH, 0);
665 0 : CHK_RET(localSendReadyDeviceNotify_->Serialize(data));
666 0 : CHK_SAFETY_FUNC_RET(memcpy_s(exchangeDataPtr, exchangeDataBlankSize, &data[0], data.size()));
667 0 : exchangeDataPtr += data.size();
668 0 : exchangeDataBlankSize -= data.size();
669 0 : }
670 :
671 0 : if ((machinePara_.linkMode != LinkMode::LINK_SIMPLEX_MODE
672 0 : || machinePara_.machineType == MachineType::MACHINE_SERVER_TYPE)) {
673 0 : RemoteRankInfo info(machinePara_.remoteDeviceId, machinePara_.remoteWorldRank, recvPid_, recvSdid_);
674 0 : CHK_RET(
675 : notifyPool_->Alloc(machinePara_.tag, info, localSendDoneDeviceNotify_, NotifyLoadType::DEVICE_NOTIFY));
676 0 : std::vector<u8> data(NOTIFY_INFO_LENGTH, 0);
677 0 : CHK_RET(localSendDoneDeviceNotify_->Serialize(data));
678 0 : CHK_SAFETY_FUNC_RET(memcpy_s(exchangeDataPtr, exchangeDataBlankSize, &data[0], data.size()));
679 0 : exchangeDataPtr += data.size();
680 0 : exchangeDataBlankSize -= data.size();
681 0 : }
682 : } else {
683 0 : if (machinePara_.linkMode != LinkMode::LINK_SIMPLEX_MODE
684 0 : || machinePara_.machineType == MachineType::MACHINE_CLIENT_TYPE) {
685 0 : RemoteRankInfo info(machinePara_.remoteDeviceId, machinePara_.remoteWorldRank, recvPid_, recvSdid_);
686 0 : CHK_RET(notifyPool_->Alloc(machinePara_.tag, info, localSendReadyNotify_));
687 0 : std::vector<u8> data(NOTIFY_INFO_LENGTH, 0);
688 0 : CHK_RET(localSendReadyNotify_->Serialize(data));
689 0 : CHK_SAFETY_FUNC_RET(memcpy_s(exchangeDataPtr, exchangeDataBlankSize, &data[0], data.size()));
690 0 : exchangeDataPtr += data.size();
691 0 : exchangeDataBlankSize -= data.size();
692 0 : }
693 :
694 0 : if (machinePara_.linkMode != LinkMode::LINK_SIMPLEX_MODE
695 0 : || machinePara_.machineType == MachineType::MACHINE_SERVER_TYPE) {
696 0 : RemoteRankInfo info(machinePara_.remoteDeviceId, machinePara_.remoteWorldRank, recvPid_, recvSdid_);
697 0 : CHK_RET(notifyPool_->Alloc(machinePara_.tag, info, localSendDoneNotify_));
698 0 : std::vector<u8> data(NOTIFY_INFO_LENGTH, 0);
699 0 : CHK_RET(localSendDoneNotify_->Serialize(data));
700 0 : CHK_SAFETY_FUNC_RET(memcpy_s(exchangeDataPtr, exchangeDataBlankSize, &data[0], data.size()));
701 0 : exchangeDataPtr += data.size();
702 0 : exchangeDataBlankSize -= data.size();
703 0 : }
704 : }
705 0 : return HCCL_SUCCESS;
706 : }
707 :
708 2 : HcclResult TransportP2p::ConstructNotifyVectorInfoForSend(u8*& exchangeDataPtr, u64& exchangeDataBlankSize)
709 : {
710 2 : if (machinePara_.isNewOneSide) {
711 2 : return HCCL_SUCCESS;
712 : }
713 0 : NotifyLoadType notifyLoadType
714 0 : = machinePara_.isAicpuModeEn ? NotifyLoadType::DEVICE_NOTIFY : NotifyLoadType::HOST_NOTIFY;
715 0 : for (u32 i = 0; i < notifyNum_; i++) {
716 0 : RemoteRankInfo info(machinePara_.remoteDeviceId, machinePara_.remoteWorldRank, recvPid_, recvSdid_);
717 0 : CHK_RET(notifyPool_->Alloc(machinePara_.tag, info, userLocalNotify_[i], notifyLoadType));
718 0 : std::vector<u8> data(NOTIFY_INFO_LENGTH, 0);
719 0 : CHK_RET(userLocalNotify_[i]->Serialize(data));
720 0 : CHK_SAFETY_FUNC_RET(memcpy_s(exchangeDataPtr, exchangeDataBlankSize, &data[0], data.size()));
721 0 : exchangeDataPtr += data.size();
722 0 : exchangeDataBlankSize -= data.size();
723 0 : }
724 0 : return HCCL_SUCCESS;
725 : }
726 :
727 2 : HcclResult TransportP2p::ConstructDataLenForSend(u8*& exchangeDataPtr, u64& exchangeDataBlankSize)
728 : {
729 2 : CHK_SAFETY_FUNC_RET(
730 : memcpy_s(exchangeDataPtr, exchangeDataBlankSize, &exchangeInfoSize_, sizeof(exchangeInfoSize_)));
731 2 : exchangeDataPtr += sizeof(exchangeInfoSize_);
732 2 : exchangeDataBlankSize -= sizeof(exchangeInfoSize_);
733 2 : return HCCL_SUCCESS;
734 : }
735 :
736 2 : HcclResult TransportP2p::ParseReceivedExchangeData()
737 : {
738 2 : s32 sendPid = 0;
739 2 : CHK_RET(SalGetBareTgid(&sendPid)); // 当前进程id
740 2 : HCCL_INFO("ParseReceivedExchangeData, sendPid[%d], recvPid[%d]", sendPid, recvPid_);
741 2 : u8* exchangeDataPtr = exchangeDataForRecv_.data();
742 2 : u64 exchangeDataBlankSize = exchangeDataTotalSize_;
743 : ExchangeInfoSize remoteInfoSize;
744 2 : CHK_RET(ParseCheckDataLen(remoteInfoSize, exchangeDataPtr, exchangeDataBlankSize));
745 2 : if (!exchangeInfoSize_.compare(remoteInfoSize)) {
746 0 : HCCL_ERROR(
747 : "remoteExchangeDataSize check fail, localIpcMemSize[%u] localNotifySize[%u] localExDataSize[%u] "
748 : "remoteIpcMemSize[%u] remoteNotifySize[%u] remoteExDataSize[%u]",
749 : exchangeInfoSize_.ipcMenSize, exchangeInfoSize_.notifySize, exchangeInfoSize_.exDataSize,
750 : remoteInfoSize.ipcMenSize, remoteInfoSize.notifySize, remoteInfoSize.exDataSize);
751 0 : return HCCL_E_INTERNAL;
752 : }
753 :
754 2 : if (sendPid != recvPid_ || recvSdid_ != INVALID_INT) {
755 0 : for (u32 i = 0; i < remoteIpcMemPtrVector_.size(); ++i) {
756 0 : CHK_RET(ParseIpcMemInfo(
757 : &remoteIpcMemPtrVector_[i], remoteIpcMemSizeVector_[i], remoteIpcMemNameVector_[i].ipcName,
758 : remoteIpcMemOffsetValueVector_[i], exchangeDataPtr, exchangeDataBlankSize));
759 0 : HCCL_INFO(
760 : "[TransportP2p][ParseReceivedExchangeData]index[%d]: remoteIpcMemPtr:[%p], "
761 : "remoteIpcMemSize:[%llu]",
762 : i, remoteIpcMemPtrVector_[i], remoteIpcMemSizeVector_[i]);
763 : }
764 0 : if (!isMemInclude_) {
765 0 : CHK_RET(ParseIpcMemInfo(
766 : &remoteOutputPtr_, remoteOutputSize_, remoteOutputMemName_.ipcName, remoteOutputOffsetValue_,
767 : exchangeDataPtr, exchangeDataBlankSize));
768 0 : CHK_RET(ParseIpcMemInfo(
769 : &remoteInputPtr_, remoteInputSize_, remoteInputMemName_.ipcName, remoteInputOffsetValue_,
770 : exchangeDataPtr, exchangeDataBlankSize));
771 : } else {
772 0 : CHK_RET(ParseMemIncludeInfo(&remoteOutputPtr_, remoteOutputSize_, exchangeDataPtr, exchangeDataBlankSize));
773 0 : CHK_RET(ParseMemIncludeInfo(&remoteInputPtr_, remoteInputSize_, exchangeDataPtr, exchangeDataBlankSize));
774 : }
775 0 : } else {
776 : u64 memAddr;
777 2 : CHK_RET(ParseIntraProcMemInfo(&memAddr, &remoteOutputSize_, exchangeDataPtr, exchangeDataBlankSize));
778 2 : remoteOutputPtr_ = reinterpret_cast<void*>(memAddr);
779 2 : CHK_RET(ParseIntraProcMemInfo(&memAddr, &remoteInputSize_, exchangeDataPtr, exchangeDataBlankSize));
780 2 : remoteInputPtr_ = reinterpret_cast<void*>(memAddr);
781 2 : for (u32 i = 0; i < remoteIpcMemPtrVector_.size(); ++i) {
782 0 : CHK_RET(
783 : ParseIntraProcMemInfo(&memAddr, &remoteIpcMemSizeVector_[i], exchangeDataPtr, exchangeDataBlankSize));
784 0 : remoteIpcMemPtrVector_[i] = reinterpret_cast<void*>(memAddr);
785 0 : HCCL_INFO(
786 : "[TransportP2p][ParseReceivedExchangeData]index[%d]: remoteIpcMemPtr:[%p], "
787 : "remoteIpcMemSize:[%llu]",
788 : i, remoteIpcMemPtrVector_[i], remoteIpcMemSizeVector_[i]);
789 : }
790 : }
791 : // 将本端和远端的Mem都打印。
792 2 : HCCL_INFO(
793 : "[TransportP2p][ParseReceivedExchangeData]remoteOutputPtr_[%p], remoteOutputSize_[%llu], "
794 : "remoteInputPtr_[%p], remoteInputSize_[%llu]",
795 : remoteOutputPtr_, remoteOutputSize_, remoteInputPtr_, remoteInputSize_);
796 :
797 2 : CHK_RET(ParseNotifyInfoEx(exchangeDataPtr, exchangeDataBlankSize));
798 2 : CHK_RET(ParseNotifyVectorInfo(exchangeDataPtr, exchangeDataBlankSize));
799 2 : CHK_RET(ParseExchangeData(exchangeDataPtr, exchangeDataBlankSize));
800 :
801 2 : if (machinePara_.isIndOp) {
802 0 : if (sendPid != recvPid_ || recvSdid_ != INVALID_INT) {
803 : u64 deviceMemNum;
804 0 : CHK_RET(ParseMemNumInfo(deviceMemNum, exchangeDataPtr, exchangeDataBlankSize));
805 0 : remoteIndOpDeviceMemPtrVector_.resize(deviceMemNum);
806 0 : remoteIndOpDeviceMemSizeVector_.resize(deviceMemNum);
807 0 : remoteIndOpDeviceMemOffsetValueVector_.resize(deviceMemNum);
808 0 : remoteIndOpDeviceMemNameVector_.resize(deviceMemNum);
809 0 : for (u64 i = 0; i < deviceMemNum; ++i) {
810 0 : CHK_RET(ParseIpcMemInfo(
811 : &remoteIndOpDeviceMemPtrVector_[i], remoteIndOpDeviceMemSizeVector_[i],
812 : remoteIndOpDeviceMemNameVector_[i].ipcName, remoteIndOpDeviceMemOffsetValueVector_[i],
813 : exchangeDataPtr, exchangeDataBlankSize));
814 0 : HCCL_INFO(
815 : "[TransportP2p][ParseReceivedExchangeData]independent operator device mem index[%d]: "
816 : "remoteIndOpDeviceMemPtr:[%p], remoteIndOpDeviceMemSize:[%llu]",
817 : i, remoteIndOpDeviceMemPtrVector_[i], remoteIndOpDeviceMemSizeVector_[i]);
818 : }
819 : u64 hostMemNum;
820 0 : CHK_RET(ParseMemNumInfo(hostMemNum, exchangeDataPtr, exchangeDataBlankSize));
821 0 : remoteIndOpHostMemPtrVector_.resize(hostMemNum);
822 0 : remoteIndOpHostMemSizeVector_.resize(hostMemNum);
823 0 : remoteIndOpHostMemOffsetValueVector_.resize(hostMemNum);
824 0 : remoteIndOpHostMemNameVector_.resize(hostMemNum);
825 0 : for (u64 i = 0; i < hostMemNum; ++i) {
826 0 : CHK_RET(ParseIpcMemInfo(
827 : &remoteIndOpHostMemPtrVector_[i], remoteIndOpHostMemSizeVector_[i],
828 : remoteIndOpHostMemNameVector_[i].ipcName, remoteIndOpHostMemOffsetValueVector_[i], exchangeDataPtr,
829 : exchangeDataBlankSize));
830 0 : HCCL_INFO(
831 : "[TransportP2p][ParseReceivedExchangeData]independent operator host mem index[%d]: "
832 : "remoteIndOpHostMemPtr:[%p], remoteIndOpHostMemSize:[%llu]",
833 : i, remoteIndOpHostMemPtrVector_[i], remoteIndOpHostMemSizeVector_[i]);
834 : }
835 0 : } else {
836 : u64 deviceMemNum;
837 : u64 memAddr;
838 0 : CHK_RET(ParseMemNumInfo(deviceMemNum, exchangeDataPtr, exchangeDataBlankSize));
839 0 : remoteIndOpDeviceMemPtrVector_.resize(deviceMemNum);
840 0 : remoteIndOpDeviceMemSizeVector_.resize(deviceMemNum);
841 0 : for (u32 i = 0; i < deviceMemNum; ++i) {
842 0 : CHK_RET(ParseIntraProcMemInfo(
843 : &memAddr, &remoteIndOpDeviceMemSizeVector_[i], exchangeDataPtr, exchangeDataBlankSize));
844 0 : remoteIndOpDeviceMemPtrVector_[i] = reinterpret_cast<void*>(memAddr);
845 0 : HCCL_INFO(
846 : "[TransportP2p][ParseReceivedExchangeData]independent operator device mem index[%d]: "
847 : "remoteIndOpDeviceMemPtr:[%p], remoteIndOpDeviceMemSize:[%llu]",
848 : i, remoteIndOpDeviceMemPtrVector_[i], remoteIndOpDeviceMemSizeVector_[i]);
849 : }
850 : u64 hostMemNum;
851 0 : CHK_RET(ParseMemNumInfo(hostMemNum, exchangeDataPtr, exchangeDataBlankSize));
852 0 : remoteIndOpHostMemPtrVector_.resize(hostMemNum);
853 0 : remoteIndOpHostMemSizeVector_.resize(hostMemNum);
854 0 : for (u32 i = 0; i < hostMemNum; ++i) {
855 0 : CHK_RET(ParseIntraProcMemInfo(
856 : &memAddr, &remoteIndOpHostMemSizeVector_[i], exchangeDataPtr, exchangeDataBlankSize));
857 0 : remoteIndOpHostMemPtrVector_[i] = reinterpret_cast<void*>(memAddr);
858 0 : HCCL_INFO(
859 : "[TransportP2p][ParseReceivedExchangeData]independent operator host mem index[%d]: "
860 : "remoteIndOpHostMemPtr:[%p], remoteIndOpHostMemSize:[%llu]",
861 : i, remoteIndOpHostMemPtrVector_[i], remoteIndOpHostMemSizeVector_[i]);
862 : }
863 : }
864 : }
865 :
866 2 : if (exchangeDataBlankSize != 0) {
867 0 : HCCL_ERROR(
868 : "[TransportP2p][ParseReceivedExchangeData] failed to Parse exchange Data "
869 : "exchangeDataBlankSize[%llu]",
870 : exchangeDataBlankSize);
871 0 : return HCCL_E_INTERNAL;
872 : }
873 2 : return HCCL_SUCCESS; // this function should not be called in normal process
874 : }
875 :
876 0 : HcclResult TransportP2p::SignalRecord(
877 : std::shared_ptr<RemoteNotify>& remoteSignal, u64 remoteSignalAddr, u64 remoteSignalOffset, Stream& stream)
878 : {
879 0 : return dispatcher_->SignalRecord(
880 : remoteSignal->ptr(), stream, machinePara_.remoteWorldRank, remoteSignalOffset, INVALID_VALUE_STAGE, false,
881 0 : remoteSignalAddr);
882 : }
883 :
884 0 : HcclResult TransportP2p::TxDataSignal(Stream& stream)
885 : {
886 : HcclResult ret;
887 : /* 发起send_ready_event事件 */
888 0 : ret = SignalRecord(remoteSendReadyNotify_, remoteSendReadyAddress_, remoteSendReadyOffset_, stream);
889 0 : CHK_PRT_RET(
890 : ret != HCCL_SUCCESS,
891 : HCCL_ERROR(
892 : "[TransportP2p][TxDataSignal]errNo[0x%016llx] In tx data signal, signal record failed.",
893 : HCCL_ERROR_CODE(ret)),
894 : ret);
895 0 : return HCCL_SUCCESS;
896 : }
897 :
898 0 : HcclResult TransportP2p::RxDataSignal(Stream& stream)
899 : {
900 : /* 等待send_ready_event事件 */
901 0 : CHK_RET(dispatcher_->SignalWait(
902 : localSendReadyNotify_->ptr(), stream, machinePara_.localUserrank, machinePara_.remoteWorldRank,
903 : INVALID_VALUE_STAGE, false, localSendReadyNotify_->notifyId_));
904 0 : return HCCL_SUCCESS;
905 : }
906 :
907 0 : HcclResult TransportP2p::TxAck(Stream& stream)
908 : {
909 : /* 发起send_done_signal事件 */
910 0 : CHK_RET(SignalRecord(remoteSendDoneNotify_, remoteSendDoneAddress_, remoteSendDoneOffset_, stream));
911 0 : return HCCL_SUCCESS;
912 : }
913 :
914 0 : HcclResult TransportP2p::RxAck(Stream& stream)
915 : {
916 : /* 等待send_done_signal事件 */
917 0 : CHK_RET(dispatcher_->SignalWait(
918 : localSendDoneNotify_->ptr(), stream, machinePara_.localUserrank, machinePara_.remoteWorldRank,
919 : INVALID_VALUE_STAGE, false, localSendDoneNotify_->notifyId_));
920 0 : return HCCL_SUCCESS;
921 : }
922 :
923 0 : HcclResult TransportP2p::TxPrepare(Stream& stream)
924 : {
925 0 : CHK_RET(TxAck(stream));
926 :
927 0 : return HCCL_SUCCESS;
928 : }
929 :
930 0 : HcclResult TransportP2p::RxPrepare(Stream& stream)
931 : {
932 0 : CHK_RET(RxAck(stream));
933 :
934 0 : return HCCL_SUCCESS;
935 : }
936 :
937 0 : HcclResult TransportP2p::TxDone(Stream& stream)
938 : {
939 0 : HcclResult ret = RxDataSignal(stream);
940 0 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[TransportP2p][TxDone]RxDataSignal failed"), ret);
941 0 : return HCCL_SUCCESS;
942 : }
943 :
944 0 : HcclResult TransportP2p::RxDone(Stream& stream)
945 : {
946 0 : HcclResult ret = TxDataSignal(stream);
947 0 : CHK_PRT_RET(ret != HCCL_SUCCESS, HCCL_ERROR("[TransportP2p][RxDone]TxDataSignal failed"), ret);
948 0 : return HCCL_SUCCESS;
949 : }
950 :
951 0 : HcclResult TransportP2p::Post(u32 notifyIdx, Stream& stream)
952 : {
953 : // 校验notifyIdx有效性
954 0 : bool bRet = (notifyIdx >= notifyNum_);
955 0 : CHK_PRT_RET(
956 : bRet,
957 : HCCL_ERROR(
958 : "[TransportP2p][Post]notifyNum[%u], notifyIdx[%u] out of range[0, %u]", notifyNum_, notifyIdx,
959 : notifyNum_ - 1),
960 : HCCL_E_INTERNAL);
961 :
962 : // 发起send_done_signal事件
963 0 : CHK_RET(SignalRecord(
964 : userRemoteNotify_[notifyIdx], userRemoteNotifyAddr_[notifyIdx], userRemoteNotifyOffset_[notifyIdx], stream));
965 0 : return HCCL_SUCCESS;
966 : }
967 :
968 0 : HcclResult TransportP2p::Wait(u32 notifyIdx, Stream& stream, const u32 timeOut)
969 : {
970 : // 校验notifyIdx有效性
971 0 : bool bRet = (notifyIdx >= notifyNum_);
972 0 : CHK_PRT_RET(
973 : bRet,
974 : HCCL_ERROR(
975 : "[TransportP2p][Wait]notifyNum[%u], notifyIdx[%u] out of range[0, %u]", notifyNum_, notifyIdx,
976 : notifyNum_ - 1),
977 : HCCL_E_INTERNAL);
978 :
979 : // 等待send_done_signal事件
980 0 : CHK_RET(dispatcher_->SignalWait(
981 : userLocalNotify_[notifyIdx]->ptr(), stream, machinePara_.localUserrank, machinePara_.remoteWorldRank,
982 : INVALID_VALUE_STAGE, false, userLocalNotify_[notifyIdx]->notifyId_, timeOut));
983 0 : return HCCL_SUCCESS;
984 : }
985 :
986 0 : HcclResult TransportP2p::ExchangeMemAndNotifyWithoutIpc()
987 : {
988 : HcclResult ret;
989 : /* 发送 output 内存 */
990 0 : ret = SendMemMesgWithoutIpc(machinePara_.outputMem.ptr(), machinePara_.outputMem.size());
991 0 : CHK_PRT_RET(
992 : ret != HCCL_SUCCESS,
993 : HCCL_ERROR(
994 : "[Exchange][IpcMesg] In exchange ipc mesg, send ipc mem output mesg fail. ret[%d], "
995 : "ptr[%p], size[%llu]",
996 : ret, machinePara_.outputMem.ptr(), machinePara_.outputMem.size()),
997 : ret);
998 :
999 : /* 发送 input 内存 */
1000 0 : ret = SendMemMesgWithoutIpc(machinePara_.inputMem.ptr(), machinePara_.inputMem.size());
1001 0 : CHK_PRT_RET(
1002 : ret != HCCL_SUCCESS,
1003 : HCCL_ERROR(
1004 : "[Exchange][IpcMesg] In exchange ipc mesg, send ipc mem input mesg fail. ret[%d], "
1005 : "ptr[%p], size[%llu]",
1006 : ret, machinePara_.inputMem.ptr(), machinePara_.inputMem.size()),
1007 : ret);
1008 :
1009 : /* 发送 notify 信息 */
1010 0 : CHK_RET(LinkSendNotifyMesg());
1011 :
1012 : /* 接收 output 内存 */
1013 : u64 memAddr;
1014 0 : ret = RecvMemMesgWithoutIpc(memAddr, remoteOutputMemName_.ipcName, remoteOutputOffsetValue_);
1015 0 : remoteOutputPtr_ = reinterpret_cast<void*>(memAddr);
1016 0 : CHK_PRT_RET(
1017 : ret != HCCL_SUCCESS,
1018 : HCCL_ERROR(
1019 : "[Exchange][IpcMesg]In exchange ipc mesg, receive ipc output mem mesg fail. ret[%d], "
1020 : "ptr[%p], memptr[%p], offset[%llu]",
1021 : ret, remoteOutputPtr_, remoteOutputMemName_.ipcName, remoteOutputOffsetValue_),
1022 : ret);
1023 :
1024 : /* 接收 input 内存 */
1025 0 : ret = RecvMemMesgWithoutIpc(memAddr, remoteInputMemName_.ipcName, remoteInputOffsetValue_);
1026 0 : remoteInputPtr_ = reinterpret_cast<void*>(memAddr);
1027 0 : CHK_PRT_RET(
1028 : ret != HCCL_SUCCESS,
1029 : HCCL_ERROR(
1030 : "[Exchange][IpcMesg]In exchange ipc mesg, receive ipc input mem mesg fail. ret[%d], "
1031 : "ptr[%p], memptr[%p], offset[%llu]",
1032 : ret, remoteInputPtr_, remoteInputMemName_.ipcName, remoteInputOffsetValue_),
1033 : ret);
1034 :
1035 : /* 接收 notify 信息 */
1036 0 : CHK_RET(LinkRecvNotifyMesg());
1037 0 : return HCCL_SUCCESS;
1038 : }
1039 :
1040 0 : HcclResult TransportP2p::ExchangeMemAndNotifyWithIpc()
1041 : {
1042 : HcclResult ret;
1043 :
1044 : /* 发送IPC output 内存 */
1045 0 : ret = SendIpcMemMesg(machinePara_.outputMem.ptr(), machinePara_.outputMem.size());
1046 0 : CHK_PRT_RET(
1047 : ret != HCCL_SUCCESS,
1048 : HCCL_ERROR(
1049 : "[Exchange][IpcMesg] In exchange ipc mesg, send ipc mem output mesg fail. ret[%d], "
1050 : "ptr[%p], size[%llu]",
1051 : ret, machinePara_.outputMem.ptr(), machinePara_.outputMem.size()),
1052 : ret);
1053 :
1054 : /* 发送IPC input 内存 */
1055 0 : ret = SendIpcMemMesg(machinePara_.inputMem.ptr(), machinePara_.inputMem.size());
1056 0 : CHK_PRT_RET(
1057 : ret != HCCL_SUCCESS,
1058 : HCCL_ERROR(
1059 : "[Exchange][IpcMesg] In exchange ipc mesg, send ipc mem input mesg fail. ret[%d], "
1060 : "ptr[%p], size[%llu]",
1061 : ret, machinePara_.inputMem.ptr(), machinePara_.inputMem.size()),
1062 : ret);
1063 :
1064 : /* 发送IPC notify 信息 */
1065 0 : CHK_RET(LinkSendNotifyMesg());
1066 :
1067 : /* 接收IPC output 内存 */
1068 0 : ret = RecvIpcMemMesg(&remoteOutputPtr_, remoteOutputMemName_.ipcName, remoteOutputOffsetValue_);
1069 0 : CHK_PRT_RET(
1070 : ret != HCCL_SUCCESS,
1071 : HCCL_ERROR(
1072 : "[Exchange][IpcMesg]In exchange ipc mesg, receive ipc output mem mesg fail. ret[%d], "
1073 : "ptr[%p], memptr[%p], offset[%llu]",
1074 : ret, remoteOutputPtr_, remoteOutputMemName_.ipcName, remoteOutputOffsetValue_),
1075 : ret);
1076 :
1077 : /* 接收IPC input 内存 */
1078 0 : ret = RecvIpcMemMesg(&remoteInputPtr_, remoteInputMemName_.ipcName, remoteInputOffsetValue_);
1079 0 : CHK_PRT_RET(
1080 : ret != HCCL_SUCCESS,
1081 : HCCL_ERROR(
1082 : "[Exchange][IpcMesg]In exchange ipc mesg, receive ipc input mem mesg fail. ret[%d], "
1083 : "ptr[%p], memptr[%p], offset[%llu]",
1084 : ret, remoteInputPtr_, remoteInputMemName_.ipcName, remoteInputOffsetValue_),
1085 : ret);
1086 :
1087 : /* 发送IPC notify 信息 */
1088 0 : CHK_RET(LinkRecvNotifyMesg());
1089 0 : return HCCL_SUCCESS;
1090 : }
1091 :
1092 0 : HcclResult TransportP2p::ExchangeMemAndNotifyMesg()
1093 : {
1094 0 : s32 sendPid = 0;
1095 0 : CHK_RET(SalGetBareTgid(&sendPid)); // 当前进程id
1096 0 : HCCL_INFO("ExchangeMemAndNotifyMesg, sendPid[%d], recvPid[%d]", sendPid, recvPid_);
1097 0 : if (sendPid != recvPid_) {
1098 0 : CHK_RET(ExchangeMemAndNotifyWithIpc()); // 跨进程时处于安全考虑，交换的是IPC Memory Name
1099 : } else {
1100 0 : CHK_RET(ExchangeMemAndNotifyWithoutIpc()); // 不跨进程时，仍然使用vnic来交换，直接交换VA，不需要转成Name
1101 : }
1102 0 : return HCCL_SUCCESS;
1103 : }
1104 :
1105 0 : HcclResult TransportP2p::SendMemMesgWithoutIpc(void* ptr, u64 size) const
1106 : {
1107 : HcclResult ret;
1108 : /* send memaddr to remote rank */
1109 0 : std::stringstream ss;
1110 0 : ss << ptr;
1111 0 : std::string memAddr = ss.str();
1112 0 : ret = defaultSocket_->Send(memAddr);
1113 0 : CHK_PRT_RET(
1114 : ret != HCCL_SUCCESS,
1115 : HCCL_ERROR(
1116 : "[Send]errNo[0x%016llx], In send ipc mesg, send name failed.remote "
1117 : "userrank[%u] local rank[%u]",
1118 : HCCL_ERROR_CODE(ret), machinePara_.remoteUserrank, machinePara_.localUserrank),
1119 : ret);
1120 :
1121 : /* send memsize to remote rank */
1122 0 : std::string memSize = std::to_string(size);
1123 0 : ret = defaultSocket_->Send(memSize);
1124 0 : CHK_PRT_RET(
1125 : ret != HCCL_SUCCESS,
1126 : HCCL_ERROR(
1127 : "[Send]errNo[0x%016llx] In send ipc mesg, send size failed. remote rank[%u] "
1128 : "size[%s] local rank[%u]",
1129 : HCCL_ERROR_CODE(ret), machinePara_.remoteUserrank, memSize.c_str(), machinePara_.localUserrank),
1130 : ret);
1131 0 : return HCCL_SUCCESS;
1132 0 : }
1133 :
1134 0 : HcclResult TransportP2p::RecvMemMesgWithoutIpc(u64& addr, [[maybe_unused]] u8* memName, u64& offset)
1135 : {
1136 : HcclResult ret;
1137 0 : std::string memAddr;
1138 :
1139 : /* 获取对端地址 */
1140 0 : ret = defaultSocket_->Recv(memAddr);
1141 0 : CHK_PRT_RET(
1142 : ret != HCCL_SUCCESS,
1143 : HCCL_ERROR(
1144 : "[Recv]errNo[0x%016llx] In recv ipc mem mesg, receive mem name failed."
1145 : "remote userrank[%u] local rank[%u]",
1146 : HCCL_ERROR_CODE(ret), machinePara_.remoteUserrank, machinePara_.localUserrank),
1147 : ret);
1148 :
1149 0 : CHK_RET(SalStrToULonglong(memAddr, HCCL_BASE_HEX, addr));
1150 : /* 获取对端内存的大小 */
1151 0 : std::string remoteMemSize;
1152 0 : u64 size = 0;
1153 0 : ret = defaultSocket_->Recv(remoteMemSize);
1154 0 : CHK_PRT_RET(
1155 : ret != HCCL_SUCCESS,
1156 : HCCL_ERROR(
1157 : "[Recv]errNo[0x%016llx] In recv ipc mem mesg, receive offset name failed."
1158 : "remote userrank[%u] local rank[%u], remoteMemSize[%s]",
1159 : HCCL_ERROR_CODE(ret), machinePara_.remoteUserrank, machinePara_.localUserrank, remoteMemSize.c_str()),
1160 : ret);
1161 :
1162 0 : CHK_RET(SalStrToULonglong(remoteMemSize, HCCL_BASE_DECIMAL, size));
1163 : /* 获取对端内存的偏移值 */
1164 0 : offset = 0;
1165 0 : return ret;
1166 0 : }
1167 :
1168 0 : HcclResult TransportP2p::SendIpcMemMesg(void* ptr, u64 size) const
1169 : {
1170 : HcclResult ret;
1171 : /* make memory shared interprocess and assigned a name */
1172 : u64 offset;
1173 0 : SecIpcName_t memName;
1174 0 : ret = MemNameRepository::GetInstance(machinePara_.deviceLogicId)
1175 0 : ->SetIpcMem(ptr, size, memName.ipcName, HCCL_IPC_MEM_NAME_LEN, offset, recvPid_, recvSdid_, isSioToHccs_);
1176 0 : CHK_PRT_RET(
1177 : ret != HCCL_SUCCESS,
1178 : HCCL_ERROR(
1179 : "[Send][IpcMemMesg]errNo[0x%016llx], In send ipc mesg, get para mem name failed. "
1180 : "mem addr[%p] local rank[%u]",
1181 : HCCL_ERROR_CODE(ret), ptr, machinePara_.localUserrank),
1182 : ret);
1183 :
1184 0 : std::string memOffset = std::to_string(offset);
1185 : /* send memName to remote rank */
1186 0 : ret = defaultSocket_->Send(memName.ipcName, HCCL_IPC_MEM_NAME_LEN);
1187 0 : CHK_PRT_RET(
1188 : ret != HCCL_SUCCESS,
1189 : HCCL_ERROR(
1190 : "[Send][IpcMemMesg]errNo[0x%016llx], In send ipc mesg, send name failed.remote "
1191 : "userrank[%u] local rank[%u]",
1192 : HCCL_ERROR_CODE(ret), machinePara_.remoteUserrank, machinePara_.localUserrank),
1193 : ret);
1194 0 : HCCL_INFO(
1195 : "localUserrank=%u, ptr=%p, remoteUserrank=%u, mem_offset=%s", machinePara_.localUserrank, ptr,
1196 : machinePara_.remoteUserrank, memOffset.c_str());
1197 :
1198 : /* send memsize to remote rank */
1199 0 : std::string memSize = std::to_string(size);
1200 0 : ret = defaultSocket_->Send(memSize);
1201 0 : CHK_PRT_RET(
1202 : ret != HCCL_SUCCESS,
1203 : HCCL_ERROR(
1204 : "[Send][IpcMemMesg]errNo[0x%016llx] In send ipc mesg, send size failed. remote rank[%u] "
1205 : "size[%s] local rank[%u]",
1206 : HCCL_ERROR_CODE(ret), machinePara_.remoteUserrank, memSize.c_str(), machinePara_.localUserrank),
1207 : ret);
1208 :
1209 : /* send memOffset to remote rank */
1210 0 : ret = defaultSocket_->Send(memOffset);
1211 0 : CHK_PRT_RET(
1212 : ret != HCCL_SUCCESS,
1213 : HCCL_ERROR(
1214 : "[Send][IpcMemMesg]errNo[0x%016llx] In send ipc mesg, send offset failed. remote rank[%u] "
1215 : "offset[%s] local rank[%u]",
1216 : HCCL_ERROR_CODE(ret), machinePara_.remoteUserrank, memOffset.c_str(), machinePara_.localUserrank),
1217 : ret);
1218 :
1219 0 : HCCL_DEBUG(
1220 : "localUserrank=%u, ptr=%p, remoteUserrank=%u, offset=%s", machinePara_.localUserrank, ptr,
1221 : machinePara_.remoteUserrank, memOffset.c_str());
1222 0 : return HCCL_SUCCESS;
1223 0 : }
1224 :
1225 0 : HcclResult TransportP2p::RecvIpcMemMesg(void** memPtr, u8* memName, u64& offset)
1226 : {
1227 : HcclResult ret;
1228 : /* 获取对端内存名字 */
1229 0 : ret = defaultSocket_->Recv(memName, HCCL_IPC_MEM_NAME_LEN);
1230 0 : CHK_PRT_RET(
1231 : ret != HCCL_SUCCESS,
1232 : HCCL_ERROR(
1233 : "[Recv][IpcMemMesg]errNo[0x%016llx] In recv ipc mem mesg, receive mem name failed."
1234 : "remote userrank[%u] local rank[%u]",
1235 : HCCL_ERROR_CODE(ret), machinePara_.remoteUserrank, machinePara_.localUserrank),
1236 : ret);
1237 : /* 获取对端内存的大小 */
1238 0 : std::string remoteMemSize;
1239 0 : u64 size = 0;
1240 0 : ret = defaultSocket_->Recv(remoteMemSize);
1241 0 : CHK_PRT_RET(
1242 : ret != HCCL_SUCCESS,
1243 : HCCL_ERROR(
1244 : "[Recv][IpcMemMesg]errNo[0x%016llx] In recv ipc mem mesg, receive offset name failed."
1245 : "remote userrank[%u] local rank[%u], remoteMemSize[%s]",
1246 : HCCL_ERROR_CODE(ret), machinePara_.remoteUserrank, machinePara_.localUserrank, remoteMemSize.c_str()),
1247 : ret);
1248 :
1249 0 : CHK_RET(SalStrToULonglong(remoteMemSize, HCCL_BASE_DECIMAL, size));
1250 :
1251 : /* 获取对端内存的偏移值 */
1252 0 : std::string remoteOffsetName;
1253 0 : ret = defaultSocket_->Recv(remoteOffsetName);
1254 0 : CHK_PRT_RET(
1255 : ret != HCCL_SUCCESS,
1256 : HCCL_ERROR(
1257 : "[Recv][IpcMemMesg]errNo[0x%016llx] In recv ipc mem mesg, receive offset name failed."
1258 : "remote userrank[%u] local rank[%u], remoteOffsetName[%s]",
1259 : HCCL_ERROR_CODE(ret), machinePara_.remoteUserrank, machinePara_.localUserrank, remoteOffsetName.c_str()),
1260 : ret);
1261 :
1262 0 : CHK_RET(SalStrToULonglong(remoteOffsetName, HCCL_BASE_DECIMAL, offset));
1263 :
1264 : /* 根据名字，获取对端IPC 内存 */
1265 0 : ret = WaitPeerMemConfig(memPtr, const_cast<u8*>(memName), size, offset);
1266 0 : CHK_PRT_RET(
1267 : ret != HCCL_SUCCESS,
1268 : HCCL_ERROR(
1269 : "[Recv][IpcMemMesg]errNo[0x%016llx] In recv ipc mem mesg, wait peer mem config "
1270 : "failed. local rank[%u]",
1271 : HCCL_ERROR_CODE(ret), machinePara_.localUserrank),
1272 : ret);
1273 :
1274 0 : HCCL_DEBUG(
1275 : "localUserrank[%u] receive from remoteUserrank[%u]", machinePara_.localUserrank, machinePara_.remoteUserrank);
1276 :
1277 0 : return HCCL_SUCCESS;
1278 0 : }
1279 :
1280 0 : HcclResult TransportP2p::TxAsync(UserMemType dstMemType, u64 dstOffset, const void* src, u64 len, Stream& stream)
1281 : {
1282 : HcclResult ret;
1283 : /* 源端发起数据传输 */
1284 0 : if (((machinePara_.linkAttribute & 0x2) == 0) && (src != nullptr)) { // 不支持目的端发起
1285 0 : void* dstMemPtr = nullptr;
1286 0 : CHK_RET(GetRemoteMem(dstMemType, &dstMemPtr));
1287 :
1288 0 : DeviceMem dstDevMem(static_cast<s8*>(dstMemPtr) + dstOffset, len);
1289 0 : DeviceMem srcDevMem(const_cast<void*>(src), len);
1290 : /* 增加hccl 数据传输时数据地址和size记录 */
1291 0 : HCCL_INFO(
1292 : "HCCL_KEY_INFO: srcAddr=[%p],srcSize=[%llu],dstAddr=[%p],dstSize=[%llu]", srcDevMem.ptr(), srcDevMem.size(),
1293 : dstDevMem.ptr(), dstDevMem.size());
1294 0 : CHK_RET(HcclD2DMemcpyAsync(
1295 : dispatcher_, dstDevMem, srcDevMem, stream, machinePara_.remoteWorldRank, transportAttr_.linkType));
1296 0 : }
1297 :
1298 : /* 发起send_ready_signal事件 */
1299 0 : ret = SignalRecord(remoteSendReadyNotify_, remoteSendReadyAddress_, remoteSendReadyOffset_, stream);
1300 0 : CHK_PRT_RET(
1301 : ret != HCCL_SUCCESS,
1302 : HCCL_ERROR("[TransportP2p][TxAsync]errNo[0x%016llx] In tx async, signal record failed.", HCCL_ERROR_CODE(ret)),
1303 : ret);
1304 :
1305 0 : return HCCL_SUCCESS;
1306 : }
1307 :
1308 0 : HcclResult TransportP2p::TxData(UserMemType dstMemType, u64 dstOffset, const void* src, u64 len, Stream& stream)
1309 : {
1310 : /* 源端发起数据传输 */
1311 0 : if (((machinePara_.linkAttribute & 0x2) == 0) && (src != nullptr)) { // 不支持目的端发起
1312 0 : void* dstMemPtr = nullptr;
1313 0 : CHK_RET(GetRemoteMem(dstMemType, &dstMemPtr));
1314 :
1315 0 : DeviceMem dstDevMem(static_cast<s8*>(dstMemPtr) + dstOffset, len);
1316 0 : DeviceMem srcDevMem(const_cast<void*>(src), len);
1317 : /* 增加hccl 数据传输时数据地址和size记录 */
1318 0 : HCCL_INFO(
1319 : "HCCL_KEY_INFO: srcAddr=[%p],srcSize=[%llu],dstAddr=[%p],dstSize=[%llu]", srcDevMem.ptr(), srcDevMem.size(),
1320 : dstDevMem.ptr(), dstDevMem.size());
1321 0 : CHK_RET(HcclD2DMemcpyAsync(
1322 : dispatcher_, dstDevMem, srcDevMem, stream, machinePara_.remoteWorldRank, transportAttr_.linkType));
1323 0 : }
1324 :
1325 0 : return HCCL_SUCCESS;
1326 : }
1327 :
1328 0 : HcclResult TransportP2p::RxData(UserMemType srcMemType, u64 srcOffset, void* dst, u64 len, Stream& stream)
1329 : {
1330 : /* 目的端发起数据传输 */
1331 0 : if ((machinePara_.linkAttribute & 0x2) && (dst != nullptr)) { // 支持目的端发起
1332 0 : void* srcMemPtr = nullptr;
1333 0 : CHK_RET(GetRemoteMem(srcMemType, &srcMemPtr));
1334 :
1335 0 : DeviceMem srcDevMem(static_cast<s8*>(srcMemPtr) + srcOffset, len);
1336 0 : DeviceMem dstDevMem(static_cast<s8*>(dst), len);
1337 0 : CHK_RET(HcclD2DMemcpyAsync(
1338 : dispatcher_, dstDevMem, srcDevMem, stream, machinePara_.remoteWorldRank, transportAttr_.linkType));
1339 0 : }
1340 :
1341 0 : return HCCL_SUCCESS;
1342 : }
1343 :
1344 0 : HcclResult TransportP2p::TxAsync(std::vector<TxMemoryInfo>& txMems, Stream& stream)
1345 : {
1346 : HcclResult ret;
1347 : /* 源端发起数据传输 */
1348 0 : if ((machinePara_.linkAttribute & 0x2) == 0) { // 不支持目的端发起
1349 0 : for (auto& mem : txMems) {
1350 0 : CHK_PTR_NULL(mem.src);
1351 0 : void* dstMemPtr = nullptr;
1352 0 : CHK_RET(GetRemoteMem(mem.dstMemType, &dstMemPtr));
1353 :
1354 0 : DeviceMem dstDevMem(static_cast<s8*>(dstMemPtr) + mem.dstOffset, mem.len);
1355 0 : DeviceMem srcDevMem(const_cast<void*>(mem.src), mem.len);
1356 : /* 增加hccl 数据传输时数据地址和size记录 */
1357 0 : HCCL_INFO(
1358 : "HCCL_KEY_INFO: srcAddr=[%p],srcSize=[%llu],dstAddr=[%p],dstSize=[%llu]", srcDevMem.ptr(),
1359 : srcDevMem.size(), dstDevMem.ptr(), dstDevMem.size());
1360 0 : CHK_RET(HcclD2DMemcpyAsync(
1361 : dispatcher_, dstDevMem, srcDevMem, stream, machinePara_.remoteWorldRank, transportAttr_.linkType));
1362 0 : }
1363 : }
1364 :
1365 : /* 发起send_ready_signal事件 */
1366 0 : ret = SignalRecord(remoteSendReadyNotify_, remoteSendReadyAddress_, remoteSendReadyOffset_, stream);
1367 0 : CHK_PRT_RET(
1368 : ret != HCCL_SUCCESS,
1369 : HCCL_ERROR("[TransportP2p][TxAsync]errNo[0x%016llx] In tx async, signal record failed.", HCCL_ERROR_CODE(ret)),
1370 : ret);
1371 :
1372 0 : return HCCL_SUCCESS;
1373 : }
1374 :
1375 0 : HcclResult TransportP2p::RxAsync(UserMemType srcMemType, u64 srcOffset, void* dst, u64 len, Stream& stream)
1376 : {
1377 : /* 等待send_ready_signal事件 */
1378 0 : CHK_RET(dispatcher_->SignalWait(
1379 : localSendReadyNotify_->ptr(), stream, machinePara_.localUserrank, machinePara_.remoteWorldRank,
1380 : INVALID_VALUE_STAGE, false, localSendReadyNotify_->notifyId_));
1381 :
1382 : /* 目的端发起数据传输 */
1383 0 : if ((machinePara_.linkAttribute & 0x2) && (dst != nullptr)) { // 支持目的端发起
1384 0 : void* srcMemPtr = nullptr;
1385 0 : CHK_RET(GetRemoteMem(srcMemType, &srcMemPtr));
1386 :
1387 0 : DeviceMem srcDevMem(static_cast<s8*>(srcMemPtr) + srcOffset, len);
1388 0 : DeviceMem dstDevMem(static_cast<s8*>(dst), len);
1389 0 : CHK_RET(HcclD2DMemcpyAsync(
1390 : dispatcher_, dstDevMem, srcDevMem, stream, machinePara_.remoteWorldRank, transportAttr_.linkType));
1391 0 : }
1392 :
1393 0 : return HCCL_SUCCESS;
1394 : }
1395 :
1396 0 : HcclResult TransportP2p::RxAsync(std::vector<RxMemoryInfo>& rxMems, Stream& stream)
1397 : {
1398 : /* 等待send_ready_signal事件 */
1399 0 : CHK_RET(dispatcher_->SignalWait(
1400 : localSendReadyNotify_->ptr(), stream, machinePara_.localUserrank, machinePara_.remoteWorldRank,
1401 : INVALID_VALUE_STAGE, false, localSendReadyNotify_->notifyId_));
1402 :
1403 : /* 目的端发起数据传输 */
1404 0 : if ((machinePara_.linkAttribute & 0x2) != 0) { // 支持目的端发起
1405 0 : for (auto& mem : rxMems) {
1406 0 : CHK_PTR_NULL(mem.dst);
1407 0 : void* srcMemPtr = nullptr;
1408 0 : CHK_RET(GetRemoteMem(mem.srcMemType, &srcMemPtr));
1409 :
1410 0 : DeviceMem srcDevMem(static_cast<s8*>(srcMemPtr) + mem.srcOffset, mem.len);
1411 0 : DeviceMem dstDevMem(static_cast<s8*>(mem.dst), mem.len);
1412 0 : CHK_RET(HcclD2DMemcpyAsync(
1413 : dispatcher_, dstDevMem, srcDevMem, stream, machinePara_.remoteWorldRank, transportAttr_.linkType));
1414 0 : }
1415 : }
1416 :
1417 0 : return HCCL_SUCCESS;
1418 : }
1419 :
1420 0 : HcclResult TransportP2p::DataReceivedAck(Stream& stream)
1421 : {
1422 0 : CHK_RET(TxAck(stream));
1423 0 : CHK_RET(RxAck(stream));
1424 0 : CHK_RET(TxDataSignal(stream));
1425 0 : CHK_RET(RxDataSignal(stream));
1426 :
1427 0 : return HCCL_SUCCESS;
1428 : }
1429 :
1430 0 : HcclResult TransportP2p::GetLocalNotify(std::vector<HcclSignalInfo>& localNotify)
1431 : {
1432 0 : if (machinePara_.isNewOneSide) {
1433 0 : return HCCL_SUCCESS;
1434 : }
1435 : HcclSignalInfo notifyInfo;
1436 :
1437 0 : if ((machinePara_.linkMode != LinkMode::LINK_SIMPLEX_MODE
1438 0 : || machinePara_.machineType == MachineType::MACHINE_CLIENT_TYPE)) {
1439 0 : if (machinePara_.isAicpuModeEn) {
1440 0 : CHK_SMART_PTR_NULL(localSendReadyDeviceNotify_);
1441 0 : CHK_RET(localSendReadyDeviceNotify_->GetNotifyData(notifyInfo));
1442 0 : localNotify.push_back(notifyInfo);
1443 : } else {
1444 0 : CHK_SMART_PTR_NULL(localSendReadyNotify_);
1445 0 : CHK_RET(localSendReadyNotify_->GetNotifyData(notifyInfo));
1446 0 : localNotify.push_back(notifyInfo);
1447 : }
1448 : }
1449 :
1450 0 : if (machinePara_.linkMode != LinkMode::LINK_SIMPLEX_MODE
1451 0 : || machinePara_.machineType == MachineType::MACHINE_SERVER_TYPE) {
1452 0 : if (machinePara_.isAicpuModeEn) {
1453 0 : CHK_SMART_PTR_NULL(localSendDoneDeviceNotify_);
1454 0 : CHK_RET(localSendDoneDeviceNotify_->GetNotifyData(notifyInfo));
1455 0 : localNotify.push_back(notifyInfo);
1456 : } else {
1457 0 : CHK_SMART_PTR_NULL(localSendDoneNotify_);
1458 0 : CHK_RET(localSendDoneNotify_->GetNotifyData(notifyInfo));
1459 0 : localNotify.push_back(notifyInfo);
1460 : }
1461 : }
1462 :
1463 0 : bool bRet = !(notifyNum_ == userLocalNotify_.size());
1464 0 : CHK_PRT_RET(
1465 : bRet,
1466 : HCCL_ERROR(
1467 : "[TransportP2p][GetLocalNotify]size of userLocalNotify_ doesn't equal to notifyNum_[%u]", notifyNum_),
1468 : HCCL_E_INTERNAL);
1469 :
1470 : // 提取新增的notify资源
1471 0 : for (u32 i = 0; i < notifyNum_; i++) {
1472 0 : CHK_SMART_PTR_NULL(userLocalNotify_[i]);
1473 0 : CHK_RET(userLocalNotify_[i]->GetNotifyData(notifyInfo));
1474 0 : localNotify.push_back(notifyInfo);
1475 : }
1476 0 : return HCCL_SUCCESS;
1477 : }
1478 :
1479 0 : HcclResult TransportP2p::GetRemoteNotify(std::vector<HcclSignalInfo>& localNotify)
1480 : {
1481 0 : if (machinePara_.isNewOneSide) {
1482 0 : return HCCL_SUCCESS;
1483 : }
1484 : HcclSignalInfo notifyInfo;
1485 0 : if ((machinePara_.linkMode != LinkMode::LINK_SIMPLEX_MODE
1486 0 : || machinePara_.machineType == MachineType::MACHINE_SERVER_TYPE)) {
1487 0 : if (machinePara_.isAicpuModeEn) {
1488 0 : CHK_SMART_PTR_NULL(remoteSendReadyDeviceNotify_);
1489 0 : CHK_RET(remoteSendReadyDeviceNotify_->GetNotifyData(notifyInfo));
1490 0 : localNotify.push_back(notifyInfo);
1491 : } else {
1492 0 : CHK_SMART_PTR_NULL(remoteSendReadyNotify_);
1493 0 : CHK_RET(remoteSendReadyNotify_->GetNotifyData(notifyInfo));
1494 0 : localNotify.push_back(notifyInfo);
1495 : }
1496 : }
1497 :
1498 0 : if (machinePara_.linkMode != LinkMode::LINK_SIMPLEX_MODE
1499 0 : || machinePara_.machineType == MachineType::MACHINE_CLIENT_TYPE) {
1500 0 : if (machinePara_.isAicpuModeEn) {
1501 0 : CHK_SMART_PTR_NULL(remoteSendDoneDeviceNotify_);
1502 0 : CHK_RET(remoteSendDoneDeviceNotify_->GetNotifyData(notifyInfo));
1503 0 : localNotify.push_back(notifyInfo);
1504 : } else {
1505 0 : CHK_SMART_PTR_NULL(remoteSendDoneNotify_);
1506 0 : CHK_RET(remoteSendDoneNotify_->GetNotifyData(notifyInfo));
1507 0 : localNotify.push_back(notifyInfo);
1508 : }
1509 : }
1510 :
1511 0 : bool bRet = !(notifyNum_ == userRemoteNotify_.size());
1512 0 : CHK_PRT_RET(
1513 : bRet,
1514 : HCCL_ERROR(
1515 : "[TransportP2p][GetRemoteNotify]size of userRemoteNotify_ doesn't equal to notifyNum_[%u]", notifyNum_),
1516 : HCCL_E_INTERNAL);
1517 :
1518 : // 新增notify的提取
1519 0 : for (u32 i = 0; i < notifyNum_; i++) {
1520 0 : CHK_SMART_PTR_NULL(userRemoteNotify_[i]);
1521 0 : CHK_RET(userRemoteNotify_[i]->GetNotifyData(notifyInfo));
1522 0 : localNotify.push_back(notifyInfo);
1523 : }
1524 0 : return HCCL_SUCCESS;
1525 : }
1526 :
1527 0 : HcclResult TransportP2p::GetIndOpRemoteMem(HcclMem** remoteMem, uint32_t* memNum)
1528 : {
1529 0 : CHK_PRT_RET(remoteMem == nullptr, HCCL_ERROR("[%s] remoteMem is nullptr", __func__), HCCL_E_PARA);
1530 0 : CHK_PRT_RET(memNum == nullptr, HCCL_ERROR("[%s] memNum is nullptr", __func__), HCCL_E_PARA);
1531 :
1532 0 : *remoteMem = nullptr;
1533 0 : *memNum = 0;
1534 0 : uint32_t totalCount = remoteIndOpHostMemPtrVector_.size() + remoteIndOpDeviceMemPtrVector_.size();
1535 0 : if (totalCount == 0) {
1536 0 : HCCL_DEBUG("[%s] No remote memory regions available", __func__);
1537 0 : return HCCL_SUCCESS;
1538 : }
1539 : // 检查向量大小是否匹配
1540 0 : if (remoteIndOpHostMemPtrVector_.size() != remoteIndOpHostMemSizeVector_.size()
1541 0 : || remoteIndOpDeviceMemPtrVector_.size() != remoteIndOpDeviceMemSizeVector_.size()) {
1542 0 : HCCL_ERROR("[%s] Memory pointer and size vectors size mismatch", __func__);
1543 0 : return HCCL_E_INTERNAL;
1544 : }
1545 : // 外部需要手动释放内存
1546 0 : HcclMem* resultArray = static_cast<HcclMem*>(malloc(totalCount * sizeof(HcclMem)));
1547 0 : CHK_PTR_NULL(resultArray);
1548 0 : uint32_t index = 0;
1549 0 : for (size_t i = 0; i < remoteIndOpDeviceMemPtrVector_.size(); ++i) {
1550 0 : resultArray[index].type = HcclMemType::HCCL_MEM_TYPE_DEVICE;
1551 0 : resultArray[index].addr = remoteIndOpDeviceMemPtrVector_[i];
1552 0 : resultArray[index].size = remoteIndOpDeviceMemSizeVector_[i];
1553 0 : index++;
1554 : }
1555 0 : for (size_t i = 0; i < remoteIndOpHostMemPtrVector_.size(); ++i) {
1556 0 : resultArray[index].type = HcclMemType::HCCL_MEM_TYPE_HOST;
1557 0 : resultArray[index].addr = remoteIndOpHostMemPtrVector_[i];
1558 0 : resultArray[index].size = remoteIndOpHostMemSizeVector_[i];
1559 0 : index++;
1560 : }
1561 0 : *remoteMem = resultArray;
1562 0 : *memNum = index;
1563 :
1564 0 : HCCL_DEBUG("[%s] Successfully returned %u remote memory regions", __func__, index);
1565 :
1566 0 : return HCCL_SUCCESS;
1567 : }
1568 :
1569 0 : HcclResult TransportP2p::GetRemoteMem(UserMemType memType, void** remotePtr)
1570 : {
1571 0 : switch (memType) {
1572 0 : case UserMemType::INPUT_MEM: {
1573 0 : *remotePtr = remoteInputPtr_;
1574 0 : break;
1575 : }
1576 :
1577 0 : case UserMemType::OUTPUT_MEM: {
1578 0 : *remotePtr = remoteOutputPtr_;
1579 0 : break;
1580 : }
1581 :
1582 0 : default: {
1583 0 : HCCL_ERROR("[Get][RemoteMem]not support dst_mem_type=%d", memType);
1584 0 : return HCCL_E_NOT_SUPPORT;
1585 : }
1586 : }
1587 :
1588 0 : return HCCL_SUCCESS;
1589 : }
1590 :
1591 0 : HcclResult TransportP2p::GetRemoteMem(std::vector<void*>* remotePtr)
1592 : {
1593 0 : *remotePtr = remoteIpcMemPtrVector_;
1594 0 : return HCCL_SUCCESS;
1595 : }
1596 :
1597 0 : HcclResult TransportP2p::GetRemoteMemSize(UserMemType memType, u64& size)
1598 : {
1599 0 : switch (memType) {
1600 0 : case UserMemType::INPUT_MEM: {
1601 0 : size = remoteInputSize_;
1602 0 : break;
1603 : }
1604 :
1605 0 : case UserMemType::OUTPUT_MEM: {
1606 0 : size = remoteOutputSize_;
1607 0 : break;
1608 : }
1609 :
1610 0 : default: {
1611 0 : HCCL_ERROR("[Get][RemoteMem]not support dst_mem_type=%d", memType);
1612 0 : return HCCL_E_NOT_SUPPORT;
1613 : }
1614 : }
1615 :
1616 0 : return HCCL_SUCCESS;
1617 : }
1618 :
1619 0 : HcclResult TransportP2p::WaitPeerMemConfig(void** memPtr, const u8* memName, uint64_t size, u64 offset)
1620 : {
1621 0 : CHK_PTR_NULL(memPtr);
1622 0 : CHK_PTR_NULL(memName);
1623 :
1624 0 : bool firstOpened = false;
1625 : // 支持进程间、进程内都可以通过name获取对端内存
1626 : HcclResult ret = MemNameRepository::GetInstance(machinePara_.deviceLogicId)
1627 0 : ->OpenIpcMem(memPtr, size, memName, HCCL_IPC_MEM_NAME_LEN, offset, firstOpened, isSioToHccs_);
1628 0 : CHK_PRT_RET(
1629 : ret != HCCL_SUCCESS,
1630 : HCCL_ERROR(
1631 : "[Wait][WaitPeerMemConfig]errNo[0x%016llx] In link pcie, open mem failed. "
1632 : "offset[%llu], size[%llu Byte], linkType[%d]",
1633 : HCCL_ERROR_CODE(ret), offset, size, transportAttr_.linkType),
1634 : ret);
1635 0 : return HCCL_SUCCESS;
1636 : }
1637 :
1638 0 : HcclResult TransportP2p::PostReady(Stream& stream)
1639 : {
1640 0 : CHK_RET(SignalRecord(remoteSendReadyNotify_, remoteSendReadyAddress_, remoteSendReadyOffset_, stream));
1641 0 : return HCCL_SUCCESS;
1642 : }
1643 :
1644 0 : HcclResult TransportP2p::WaitReady(Stream& stream)
1645 : {
1646 0 : CHK_RET(dispatcher_->SignalWait(
1647 : localSendReadyNotify_->ptr(), stream, machinePara_.localUserrank, machinePara_.remoteWorldRank,
1648 : INVALID_VALUE_STAGE, false, localSendReadyNotify_->notifyId_));
1649 0 : return HCCL_SUCCESS;
1650 : }
1651 :
1652 0 : HcclResult TransportP2p::PostFin(Stream& stream)
1653 : {
1654 0 : CHK_RET(SignalRecord(remoteSendDoneNotify_, remoteSendDoneAddress_, remoteSendDoneOffset_, stream));
1655 0 : return HCCL_SUCCESS;
1656 : }
1657 :
1658 0 : HcclResult TransportP2p::WaitFin(Stream& stream)
1659 : {
1660 0 : CHK_RET(dispatcher_->SignalWait(
1661 : localSendDoneNotify_->ptr(), stream, machinePara_.localUserrank, machinePara_.remoteWorldRank,
1662 : INVALID_VALUE_STAGE, false, localSendDoneNotify_->notifyId_));
1663 0 : return HCCL_SUCCESS;
1664 : }
1665 :
1666 : HcclResult
1667 0 : TransportP2p::WriteSync(struct Transport::Buffer& remoteBuf, struct Transport::Buffer& localBuf, Stream& stream)
1668 : {
1669 0 : DeviceMem remoteDevMem(const_cast<void*>(remoteBuf.addr), remoteBuf.size);
1670 0 : DeviceMem localDevMem(const_cast<void*>(localBuf.addr), localBuf.size);
1671 0 : HCCL_INFO(
1672 : "HCCL_KEY_INFO: localAddr=[%p],localSize=[%llu],remoteAddr=[%p],remoteSize=[%llu]", localDevMem.ptr(),
1673 : localDevMem.size(), remoteDevMem.ptr(), remoteDevMem.size());
1674 0 : CHK_RET(HcclD2DMemcpyAsync(
1675 : dispatcher_, remoteDevMem, localDevMem, stream, machinePara_.remoteWorldRank, transportAttr_.linkType));
1676 0 : return HCCL_SUCCESS;
1677 0 : }
1678 :
1679 : HcclResult
1680 0 : TransportP2p::WriteAsyncEx(struct Transport::Buffer& remoteBuf, struct Transport::Buffer& localBuf, Stream& stream)
1681 : {
1682 0 : bool isLocalHostAddr = false;
1683 0 : bool isRemoteHostAddr = false;
1684 0 : struct Transport::Buffer newLocalBuf {};
1685 0 : struct Transport::Buffer newRemoteBuf {};
1686 0 : CHK_RET(ReplaceMemAddr(localBuf, remoteBuf, newLocalBuf, newRemoteBuf, isLocalHostAddr, isRemoteHostAddr));
1687 0 : DeviceMem dstDevMem(const_cast<void*>(newRemoteBuf.addr), newRemoteBuf.size);
1688 0 : DeviceMem srcDevMem(const_cast<void*>(newLocalBuf.addr), newLocalBuf.size);
1689 0 : CHK_RET(reinterpret_cast<DispatcherPub*>(dispatcher_)
1690 : ->MemcpyAsync(dstDevMem, srcDevMem, stream, machinePara_.remoteWorldRank, transportAttr_.linkType));
1691 0 : return HCCL_SUCCESS;
1692 0 : }
1693 :
1694 : HcclResult
1695 0 : TransportP2p::WriteAsync(struct Transport::Buffer& remoteBuf, struct Transport::Buffer& localBuf, Stream& stream)
1696 : {
1697 0 : if (machinePara_.isNewOneSide) {
1698 0 : return WriteAsyncEx(remoteBuf, localBuf, stream);
1699 : }
1700 :
1701 0 : DeviceMem remoteDevMem(const_cast<void*>(remoteBuf.addr), remoteBuf.size);
1702 0 : DeviceMem localDevMem(const_cast<void*>(localBuf.addr), localBuf.size);
1703 0 : HCCL_INFO(
1704 : "HCCL_KEY_INFO: localAddr=[%p],localSize=[%llu],remoteAddr=[%p],remoteSize=[%llu]", localDevMem.ptr(),
1705 : localDevMem.size(), remoteDevMem.ptr(), remoteDevMem.size());
1706 0 : CHK_RET(HcclD2DMemcpyAsync(
1707 : dispatcher_, remoteDevMem, localDevMem, stream, machinePara_.remoteWorldRank, transportAttr_.linkType));
1708 0 : return HCCL_SUCCESS;
1709 0 : }
1710 :
1711 0 : HcclResult TransportP2p::WriteReduceAsync(
1712 : struct Transport::Buffer& remoteBuf, struct Transport::Buffer& localBuf, const HcclDataType datatype,
1713 : HcclReduceOp redOp, Stream& stream)
1714 : {
1715 0 : HCCL_INFO(
1716 : "HCCL_KEY_INFO: localAddr=[%p],localSize=[%llu],remoteAddr=[%p],remoteSize=[%llu]", localBuf.addr,
1717 : localBuf.size, remoteBuf.addr, remoteBuf.size);
1718 :
1719 0 : u64 reduceAttr = 0;
1720 0 : if (IsSpInlineReduce()) {
1721 0 : reduceAttr = INLINE_REDUCE_BIT;
1722 : }
1723 0 : CHK_RET(HcclReduceAsync(
1724 : dispatcher_, const_cast<void*>(localBuf.addr), remoteBuf.size / SIZE_TABLE[datatype], datatype, redOp, stream,
1725 : const_cast<void*>(remoteBuf.addr), GetRemoteRank(), GetLinkType(), reduceAttr));
1726 0 : return HCCL_SUCCESS;
1727 : }
1728 :
1729 : HcclResult
1730 0 : TransportP2p::ReadSync(struct Transport::Buffer& localBuf, struct Transport::Buffer& remoteBuf, Stream& stream)
1731 : {
1732 0 : DeviceMem remoteDevMem(const_cast<void*>(remoteBuf.addr), remoteBuf.size);
1733 0 : DeviceMem localDevMem(const_cast<void*>(localBuf.addr), localBuf.size);
1734 0 : HCCL_INFO(
1735 : "HCCL_KEY_INFO: localAddr=[%p],localSize=[%llu],remoteAddr=[%p],remoteSize=[%llu]", localDevMem.ptr(),
1736 : localDevMem.size(), remoteDevMem.ptr(), remoteDevMem.size());
1737 0 : CHK_RET(HcclD2DMemcpyAsync(
1738 : dispatcher_, localDevMem, remoteDevMem, stream, machinePara_.remoteWorldRank, transportAttr_.linkType));
1739 0 : return HCCL_SUCCESS;
1740 0 : }
1741 :
1742 0 : HcclResult TransportP2p::ReadReduceSync(
1743 : struct Transport::Buffer& localBuf, struct Transport::Buffer& remoteBuf, const HcclDataType datatype,
1744 : HcclReduceOp redOp, Stream& stream)
1745 : {
1746 0 : HCCL_INFO(
1747 : "HCCL_KEY_INFO: localAddr=[%p],localSize=[%llu],remoteAddr=[%p],remoteSize=[%llu]", localBuf.addr,
1748 : localBuf.size, remoteBuf.addr, remoteBuf.size);
1749 :
1750 0 : u64 reduceAttr = 0;
1751 0 : if (IsSpInlineReduce()) {
1752 0 : reduceAttr = INLINE_REDUCE_BIT;
1753 : }
1754 0 : CHK_RET(HcclReduceAsync(
1755 : dispatcher_, const_cast<void*>(remoteBuf.addr), remoteBuf.size / SIZE_TABLE[datatype], datatype, redOp, stream,
1756 : const_cast<void*>(localBuf.addr), GetRemoteRank(), GetLinkType(), reduceAttr));
1757 0 : return HCCL_SUCCESS;
1758 : }
1759 :
1760 : HcclResult
1761 0 : TransportP2p::ReadAsyncEx(struct Transport::Buffer& localBuf, struct Transport::Buffer& remoteBuf, Stream& stream)
1762 : {
1763 0 : bool isLocalHostAddr = false;
1764 0 : bool isRemoteHostAddr = false;
1765 0 : struct Transport::Buffer newLocalBuf {};
1766 0 : struct Transport::Buffer newRemoteBuf {};
1767 0 : CHK_RET(ReplaceMemAddr(localBuf, remoteBuf, newLocalBuf, newRemoteBuf, isLocalHostAddr, isRemoteHostAddr));
1768 0 : DeviceMem dstDevMem(const_cast<void*>(newLocalBuf.addr), newLocalBuf.size);
1769 0 : DeviceMem srcDevMem(const_cast<void*>(newRemoteBuf.addr), newRemoteBuf.size);
1770 0 : CHK_RET(reinterpret_cast<DispatcherPub*>(dispatcher_)
1771 : ->MemcpyAsync(dstDevMem, srcDevMem, stream, machinePara_.remoteWorldRank, transportAttr_.linkType));
1772 0 : return HCCL_SUCCESS;
1773 0 : }
1774 :
1775 : HcclResult
1776 0 : TransportP2p::ReadAsync(struct Transport::Buffer& localBuf, struct Transport::Buffer& remoteBuf, Stream& stream)
1777 : {
1778 0 : if (machinePara_.isNewOneSide) {
1779 0 : return ReadAsyncEx(localBuf, remoteBuf, stream);
1780 : }
1781 0 : DeviceMem dstDevMem(const_cast<void*>(localBuf.addr), localBuf.size);
1782 0 : DeviceMem srcDevMem(const_cast<void*>(remoteBuf.addr), remoteBuf.size);
1783 0 : return HcclD2DMemcpyAsync(
1784 0 : dispatcher_, dstDevMem, srcDevMem, stream, machinePara_.remoteWorldRank, transportAttr_.linkType);
1785 0 : }
1786 :
1787 6 : HcclResult TransportP2p::SumCheckSizeAndConsisten(
1788 : ExInfoType exInfoType, u32 rightInfoSize, u64& blankSizeRecord, u64 exchangeDataBlankSize)
1789 : {
1790 6 : u32 checkInfoSize = blankSizeRecord - exchangeDataBlankSize;
1791 6 : if (checkInfoSize != rightInfoSize) {
1792 0 : HCCL_ERROR(
1793 : "[SumCheckSizeAndConsisten] ExInfoType[%d] check size failed, checkInfoSize[%u] rightInfoSize[%u]",
1794 : exInfoType, checkInfoSize, rightInfoSize);
1795 0 : return HCCL_E_INTERNAL;
1796 : }
1797 6 : blankSizeRecord = exchangeDataBlankSize;
1798 6 : return HCCL_SUCCESS;
1799 : }
1800 :
1801 0 : HcclResult TransportP2p::ConstructMemIncludeInfoForSend(u8*& exchangeDataPtr, u64& exchangeDataBlankSize)
1802 : {
1803 0 : u64 outputSize = machinePara_.outputMem.size();
1804 : u64 outputOffset
1805 0 : = reinterpret_cast<u64>(machinePara_.outputMem.ptr()) - reinterpret_cast<u64>(machinePara_.mem[0].ptr());
1806 0 : CHK_SAFETY_FUNC_RET(memcpy_s(exchangeDataPtr, exchangeDataBlankSize, &outputSize, sizeof(u64)));
1807 0 : exchangeDataPtr += sizeof(u64);
1808 0 : exchangeDataBlankSize -= sizeof(u64);
1809 0 : CHK_SAFETY_FUNC_RET(memcpy_s(exchangeDataPtr, exchangeDataBlankSize, &outputOffset, sizeof(u64)));
1810 0 : exchangeDataPtr += sizeof(u64);
1811 0 : exchangeDataBlankSize -= sizeof(u64);
1812 :
1813 0 : u64 inputSize = machinePara_.inputMem.size();
1814 : u64 inputOffset
1815 0 : = reinterpret_cast<u64>(machinePara_.inputMem.ptr()) - reinterpret_cast<u64>(machinePara_.mem[0].ptr());
1816 0 : CHK_SAFETY_FUNC_RET(memcpy_s(exchangeDataPtr, exchangeDataBlankSize, &inputSize, sizeof(u64)));
1817 0 : exchangeDataPtr += sizeof(u64);
1818 0 : exchangeDataBlankSize -= sizeof(u64);
1819 0 : CHK_SAFETY_FUNC_RET(memcpy_s(exchangeDataPtr, exchangeDataBlankSize, &inputOffset, sizeof(u64)));
1820 0 : exchangeDataPtr += sizeof(u64);
1821 0 : exchangeDataBlankSize -= sizeof(u64);
1822 :
1823 0 : return HCCL_SUCCESS;
1824 : }
1825 :
1826 0 : HcclResult TransportP2p::ParseMemIncludeInfo(void** memPtr, u64& size, u8*& exchangeDataPtr, u64& exchangeDataBlankSize)
1827 : {
1828 0 : u64 memOffset = 0;
1829 0 : CHK_SAFETY_FUNC_RET(memcpy_s(&size, sizeof(u64), exchangeDataPtr, sizeof(u64)));
1830 0 : exchangeDataPtr += sizeof(u64);
1831 0 : exchangeDataBlankSize -= sizeof(u64);
1832 0 : CHK_SAFETY_FUNC_RET(memcpy_s(&memOffset, sizeof(u64), exchangeDataPtr, sizeof(u64)));
1833 0 : exchangeDataPtr += sizeof(u64);
1834 0 : exchangeDataBlankSize -= sizeof(u64);
1835 0 : if (!machinePara_.isNewOneSide) {
1836 0 : *memPtr = reinterpret_cast<void*>(reinterpret_cast<u64>(remoteIpcMemPtrVector_[0]) + memOffset);
1837 : }
1838 0 : return HCCL_SUCCESS;
1839 : }
1840 :
1841 2 : void TransportP2p::SetMemIncludeFlag()
1842 : {
1843 2 : if (machinePara_.mem.empty()) {
1844 2 : return;
1845 : }
1846 : // 当前只取mem[0] ->expMem
1847 0 : u64 memPtr = reinterpret_cast<u64>(machinePara_.mem[0].ptr());
1848 0 : u64 memEndPtr = memPtr + machinePara_.mem[0].size();
1849 0 : u64 inputMemPtr = reinterpret_cast<u64>(machinePara_.inputMem.ptr());
1850 0 : u64 inputMemEndPtr = inputMemPtr + machinePara_.inputMem.size();
1851 0 : u64 outputMemPtr = reinterpret_cast<u64>(machinePara_.outputMem.ptr());
1852 0 : u64 outputMemEndPtr = outputMemPtr + machinePara_.outputMem.size();
1853 0 : HCCL_DEBUG(
1854 : "[SetMemIncludeFlag] memPtr[%u] memEndPtr[%u], inputMemPtr[%u] inputMemEndPtr[%u], "
1855 : "outputMemPtr[%u] outputMemEndPtr[%u]",
1856 : memPtr, memEndPtr, inputMemPtr, inputMemEndPtr, outputMemPtr, outputMemEndPtr);
1857 0 : if ((memPtr <= inputMemPtr && inputMemEndPtr <= memEndPtr)
1858 0 : && (memPtr <= outputMemPtr && outputMemEndPtr <= memEndPtr)) {
1859 0 : isMemInclude_ = true;
1860 : }
1861 0 : return;
1862 : }
1863 :
1864 0 : HcclResult TransportP2p::ReplaceMemAddr(
1865 : Transport::Buffer& localMem, Transport::Buffer& remoteMem, Transport::Buffer& newLocalMem,
1866 : Transport::Buffer& newRemoteMem, bool& isLocalHostAddr, bool& isRemoteHostAddr)
1867 : {
1868 0 : HCCL_DEBUG(
1869 : "[TransportP2p][ReplaceMemAddr]old localAddr=[%p],localSize=[%llu],remoteAddr=[%p],remoteSize=[%llu]",
1870 : localMem.addr, localMem.size, remoteMem.addr, remoteMem.size);
1871 :
1872 0 : isLocalHostAddr = false;
1873 0 : isRemoteHostAddr = false;
1874 0 : void* localAddr = const_cast<void*>(localMem.addr);
1875 0 : u64 localSize = localMem.size;
1876 0 : auto localKey = BufferKey<uintptr_t, u64>(reinterpret_cast<uintptr_t>(localAddr), localSize);
1877 0 : auto localBufferPair = localHcclMemExMgr_.Find(localKey);
1878 0 : if (localBufferPair.first) {
1879 0 : std::shared_ptr<HcclMemEx>& localBufMemPtr = localBufferPair.second;
1880 0 : u64 localDataOffSet = static_cast<u8*>(localAddr) - static_cast<u8*>(localBufMemPtr->addr);
1881 0 : newLocalMem.addr = static_cast<void*>(static_cast<u8*>(localBufMemPtr->devAddr) + localDataOffSet);
1882 0 : newLocalMem.size = localMem.size;
1883 0 : if (localBufMemPtr->type == HcclMemType::HCCL_MEM_TYPE_HOST) {
1884 0 : isLocalHostAddr = true;
1885 : }
1886 : } else {
1887 0 : HCCL_DEBUG("[TransportP2p][ReplaceMemAddr] Can't find localBufferPair by key {%p, %llu}", localAddr, localSize);
1888 0 : newLocalMem.addr = localAddr;
1889 0 : newLocalMem.size = localSize;
1890 : }
1891 :
1892 0 : void* remoteAddr = const_cast<void*>(remoteMem.addr);
1893 0 : auto remoteKey = BufferKey<uintptr_t, u64>(reinterpret_cast<uintptr_t>(remoteAddr), remoteMem.size);
1894 0 : auto remoteBufferPair = remoteHcclMemExMgr_.Find(remoteKey);
1895 0 : if (remoteBufferPair.first) {
1896 0 : std::shared_ptr<HcclMemEx>& remoteBufMemPtr = remoteBufferPair.second;
1897 0 : u64 remoteDataOffSet = static_cast<u8*>(remoteAddr) - static_cast<u8*>(remoteBufMemPtr->addr);
1898 0 : newRemoteMem.addr = static_cast<void*>(static_cast<u8*>(remoteBufMemPtr->devAddr) + remoteDataOffSet);
1899 0 : if (remoteBufMemPtr->type == HcclMemType::HCCL_MEM_TYPE_HOST) {
1900 0 : isRemoteHostAddr = true;
1901 : }
1902 : } else {
1903 0 : HCCL_DEBUG(
1904 : "[TransportP2p][ReplaceMemAddr] Can't find remoteBuffer by key {%p, %llu}", remoteAddr, remoteMem.size);
1905 0 : newRemoteMem.addr = remoteMem.addr;
1906 : }
1907 0 : newRemoteMem.size = remoteMem.size;
1908 :
1909 0 : HCCL_DEBUG(
1910 : "[TransportP2p][ReplaceMemAddr]old localAddr=[%p],localSize=[%llu],remoteAddr=[%p],remoteSize=[%llu], "
1911 : "isLocalHostAddr[%u] isRemoteHostAddr[%u]",
1912 : newLocalMem.addr, newLocalMem.size, newRemoteMem.addr, newRemoteMem.size,
1913 : static_cast<uint32_t>(isLocalHostAddr), static_cast<uint32_t>(isRemoteHostAddr));
1914 0 : return HCCL_SUCCESS;
1915 0 : }
1916 :
1917 0 : HcclResult TransportP2p::InitHcclMemExMgrWithMem(HcclMemEx* bufMem, u32 bufSize, HcclMemExMgr& hcommMemExMgr)
1918 : {
1919 0 : for (u32 i = 0; i < bufSize; i++) {
1920 0 : HcclMemEx& bufMemTmp = bufMem[i];
1921 :
1922 0 : std::shared_ptr<HcclMemEx> hcclMemEx = nullptr;
1923 0 : hcclMemEx = std::make_shared<HcclMemEx>();
1924 0 : CHK_PTR_NULL(hcclMemEx);
1925 :
1926 0 : HcclMemEx* hcclMemExPtr = reinterpret_cast<HcclMemEx*>(hcclMemEx.get());
1927 0 : *hcclMemExPtr = bufMemTmp;
1928 :
1929 0 : hccl::BufferKey<uintptr_t, u64> tempKey(reinterpret_cast<uintptr_t>(bufMemTmp.addr), bufMemTmp.size);
1930 0 : auto resultPair = hcommMemExMgr.Add(tempKey, hcclMemEx);
1931 0 : if (!resultPair.second) {
1932 0 : HCCL_ERROR(
1933 : "[TransportP2p][InitHcclMemExMgrWithMem]add addr:%p, size[%lu], type[%u], devAddr[%p] fail",
1934 : bufMemTmp.addr, bufMemTmp.size, static_cast<u32>(bufMemTmp.type), bufMemTmp.devAddr);
1935 0 : return HCCL_E_INTERNAL;
1936 : } else {
1937 0 : HCCL_INFO(
1938 : "[TransportP2p][InitHcclMemExMgrWithMem]add addr:%p, size[%lu], type[%u], devAddr[%p] done",
1939 : bufMemTmp.addr, bufMemTmp.size, static_cast<u32>(bufMemTmp.type), bufMemTmp.devAddr);
1940 : }
1941 0 : }
1942 :
1943 0 : HCCL_INFO("[TransportP2p][InitHcclMemExMgrWithMem] done");
1944 0 : return HCCL_SUCCESS;
1945 : }
1946 :
1947 0 : HcclResult TransportP2p::InitHcclMemExMgr(MachinePara& machinePara)
1948 : {
1949 0 : HCCL_INFO("[TransportP2p][InitHcclMemExMgr] start");
1950 0 : CHK_RET(InitHcclMemExMgrWithMem(machinePara.localBufMem, machinePara.localBufSize, localHcclMemExMgr_));
1951 0 : machinePara.localBufMem = nullptr;
1952 0 : machinePara.localBufSize = 0;
1953 0 : HCCL_INFO("[TransportP2p][InitHcclMemExMgr] local done");
1954 0 : CHK_RET(InitHcclMemExMgrWithMem(machinePara.remoteBufMem, machinePara.remoteBufSize, remoteHcclMemExMgr_));
1955 0 : machinePara.remoteBufMem = nullptr;
1956 0 : machinePara.remoteBufSize = 0;
1957 0 : HCCL_INFO("[TransportP2p][InitHcclMemExMgr] remote done");
1958 0 : return HCCL_SUCCESS;
1959 : }
1960 : } // namespace hccl
|
