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