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_heterog.h"
12 : #include "log.h"
13 : #include "network_manager_pub.h"
14 : #include "hccl_socket.h"
15 : #include "externalinput_pub.h"
16 :
17 : using namespace std;
18 : namespace hccl {
19 : constexpr u32 SINGLE_WHITE_LIST_NUM = 1;
20 : constexpr u32 WAIT_LINK_BUILD_DELAY_TIME_US = 10;
21 : constexpr s32 MAX_LINK_NUM = 10;
22 :
23 : std::atomic<u32> TransportHeterog::rankTableCrc_ = {0};
24 0 : TransportHeterog::TransportHeterog(const string &tag, HcclIpAddress &selfIp, HcclIpAddress &peerIp, u32 peerPort,
25 0 : u32 selfPort, const TransportResourceInfo &transportResourceInfo)
26 0 : : transTag_(tag),
27 0 : nicSocketHandle_(nullptr),
28 0 : selfIp_(selfIp),
29 0 : peerIp_(peerIp),
30 0 : peerPort_(peerPort),
31 0 : selfPort_(selfPort),
32 0 : pMsgInfosMem_(transportResourceInfo.pMsgInfosMem),
33 0 : pReqInfosMem_(transportResourceInfo.pReqInfosMem),
34 0 : recvEnvelopNum_(0)
35 0 : {}
36 0 : TransportHeterog::TransportHeterog(const TransportResourceInfo &transportResourceInfo)
37 0 : : nicSocketHandle_(nullptr),
38 0 : peerPort_(0),
39 0 : selfPort_(0),
40 0 : pMsgInfosMem_(transportResourceInfo.pMsgInfosMem),
41 0 : pReqInfosMem_(transportResourceInfo.pReqInfosMem),
42 0 : recvEnvelopNum_(0)
43 0 : {}
44 :
45 0 : TransportHeterog::~TransportHeterog() {}
46 :
47 0 : HcclResult TransportHeterog::Init(u32 localUserRank, u32 remoteUserRank)
48 : {
49 0 : return HCCL_SUCCESS;
50 : }
51 :
52 0 : HcclResult TransportHeterog::Init(SocketInfoT &socketInfo, RdmaHandle rdmaHandle, MrHandle mrHandle)
53 : {
54 0 : return HCCL_SUCCESS;
55 : }
56 :
57 0 : HcclResult TransportHeterog::Improbe(const TransportEndPointParam &epParam, s32 &matched, HcclMessageInfo *&msg,
58 : HcclStatus &status, bool &flag)
59 : {
60 0 : return HCCL_SUCCESS;
61 : }
62 :
63 0 : HcclResult TransportHeterog::Imrecv(const TransData &recvData, HcclMessageInfo &msg, HcclRequestInfo *&request,
64 : bool flag, bool needRecordFlag)
65 : {
66 0 : return HCCL_SUCCESS;
67 : }
68 :
69 0 : HcclResult TransportHeterog::ImrecvScatter(void *buf[], int count[], int bufCount, HcclDataType datatype,
70 : HcclMessageInfo &msg, HcclRequestInfo *&request)
71 : {
72 0 : return HCCL_SUCCESS;
73 : }
74 :
75 0 : HcclResult TransportHeterog::CheckRecvMsgAndRequestBuffer()
76 : {
77 0 : CHK_SMART_PTR_NULL(pMsgInfosMem_);
78 0 : CHK_SMART_PTR_NULL(pReqInfosMem_);
79 0 : return HCCL_SUCCESS;
80 : }
81 :
82 0 : HcclResult TransportHeterog::GenerateSendRequest(const TransData &sendData, const TransportEndPointParam &epParam,
83 : HcclRequestInfo *&request)
84 : {
85 0 : request = pReqInfosMem_->Alloc();
86 0 : CHK_PTR_NULL(request);
87 0 : request->transportHandle = this;
88 0 : request->transportRequest.transData = sendData;
89 0 : request->transportRequest.epParam = epParam;
90 0 : request->transportRequest.requestType = HcclRequestType::HCCL_REQUEST_SEND;
91 0 : request->transportRequest.protocol = 0;
92 0 : request->transportRequest.msn = reinterpret_cast<u64>(request);
93 0 : request->transportRequest.status = -1;
94 0 : request->transportRequest.envoffset = 0;
95 0 : request->transportRequest.tranoffset = 0;
96 :
97 0 : return HCCL_SUCCESS;
98 : }
99 :
100 0 : HcclResult TransportHeterog::GenerateRecvRequest(const TransData &recvData, const HcclMessageInfo &msg,
101 : HcclRequestInfo *&request)
102 : {
103 0 : request = pReqInfosMem_->Alloc();
104 0 : CHK_PTR_NULL(request);
105 0 : request->transportHandle = this;
106 0 : request->transportRequest.transData = recvData;
107 0 : request->transportRequest.transData.srcBuf = msg.envelope.envelope.transData.srcBuf;
108 0 : request->transportRequest.epParam = msg.envelope.envelope.epParam;
109 0 : request->transportRequest.requestType = HcclRequestType::HCCL_REQUEST_RECV;
110 0 : request->transportRequest.protocol = msg.envelope.envelope.protocol;
111 0 : request->transportRequest.msn = msg.envelope.envelope.msn;
112 0 : request->transportRequest.status = -1;
113 0 : request->transportRequest.envoffset = 0;
114 0 : request->transportRequest.tranoffset = 0;
115 :
116 0 : return HCCL_SUCCESS;
117 : }
118 :
119 0 : HcclResult TransportHeterog::GenerateRecvScatterRequest(const HcclMessageInfo &msg, HcclRequestInfo *&request)
120 : {
121 0 : request = pReqInfosMem_->Alloc();
122 0 : CHK_PTR_NULL(request);
123 0 : request->transportHandle = this;
124 0 : request->transportRequest.transData.srcBuf = msg.envelope.envelope.transData.srcBuf;
125 0 : request->transportRequest.transData.count = 0;
126 0 : request->transportRequest.epParam = msg.envelope.envelope.epParam;
127 0 : request->transportRequest.requestType = HcclRequestType::HCCL_REQUEST_RECV;
128 0 : request->transportRequest.protocol = msg.envelope.envelope.protocol;
129 0 : request->transportRequest.msn = msg.envelope.envelope.msn;
130 0 : request->transportRequest.status = -1;
131 0 : request->transportRequest.envoffset = 0;
132 0 : request->transportRequest.tranoffset = 0;
133 :
134 0 : return HCCL_SUCCESS;
135 : }
136 :
137 0 : HcclResult TransportHeterog::FreeRequest(HcclRequestInfo &request) const
138 : {
139 0 : request.commHandle = nullptr;
140 0 : request.transportHandle = nullptr;
141 0 : CHK_RET(pReqInfosMem_->Free(&request));
142 0 : return HCCL_SUCCESS;
143 : }
144 :
145 0 : HcclResult TransportHeterog::CheckTransportEndPointInfo(const TransportEndPointInfo &epInfo,
146 : const TransportEndPointInfo &epInfoCheck) const
147 : {
148 0 : if (epInfo.commId != epInfoCheck.commId) {
149 0 : HCCL_ERROR("[Check][Tag]errNo[0x%016llx] commId[%u] is invalid, expect:%u", HCCL_ERROR_CODE(HCCL_E_PARA),
150 : epInfo.commId, epInfoCheck.commId);
151 0 : return HCCL_E_PARA;
152 : }
153 :
154 0 : if (epInfo.tag != epInfoCheck.tag) {
155 0 : HCCL_ERROR("[Check][Tag]errNo[0x%016llx] tag[%u] is invalid, expect:%u", HCCL_ERROR_CODE(HCCL_E_PARA),
156 : epInfo.tag, epInfoCheck.tag);
157 0 : return HCCL_E_PARA;
158 : }
159 0 : return HCCL_SUCCESS;
160 : }
161 :
162 0 : HcclResult TransportHeterog::CheckRecvEnvelope(const TransData &recvDataCheck, const HcclEnvelopeSummary &envelope)
163 : {
164 0 : if (envelope.status != 0) {
165 0 : HCCL_ERROR("[Check][EnvelopeStatus] envelope status:[%u] is invalid", envelope.status);
166 0 : return HCCL_E_PARA;
167 : }
168 :
169 0 : CHK_RET(CheckTransportEndPointInfo(envelope.envelope.epParam.src, envelope.envelope.epParam.dst));
170 :
171 0 : if (recvDataCheck.count < envelope.envelope.transData.count) {
172 0 : HCCL_ERROR("[Check][RecvEnvelope]Imrecv input count[%llu] should be not less than Isend count[%llu]",
173 : recvDataCheck.count, envelope.envelope.transData.count);
174 0 : return HCCL_E_PARA;
175 : }
176 :
177 0 : if ((recvDataCheck.dstBuf == 0) && ((recvDataCheck.count != 0) || (envelope.envelope.transData.count != 0))) {
178 0 : HCCL_ERROR("[Check][RecvEnvelope]Imrecv buffer[%p] or count[%llu] is invalid", recvDataCheck.dstBuf,
179 : envelope.envelope.transData.count);
180 0 : return HCCL_E_PARA;
181 : }
182 :
183 0 : if (recvDataCheck.dataType != envelope.envelope.transData.dataType) {
184 0 : HCCL_ERROR("[Check][RecvEnvelope]Imrecv input dataType[%s] should be Isend dataType[%s]",
185 : GetDataTypeEnumStr(recvDataCheck.dataType).c_str(),
186 : GetDataTypeEnumStr(envelope.envelope.transData.dataType).c_str());
187 0 : return HCCL_E_PARA;
188 : }
189 :
190 0 : return HCCL_SUCCESS;
191 : }
192 :
193 0 : HcclResult TransportHeterog::CheckRecvScatterEnvelope(void *buf[], int count[], int bufCount, HcclDataType datatype,
194 : const HcclEnvelopeSummary &envelope)
195 : {
196 0 : if (envelope.status != 0) {
197 0 : HCCL_ERROR("[Check][EnvelopeStatus] envelope status:[%u] is invalid", envelope.status);
198 0 : return HCCL_E_PARA;
199 : }
200 :
201 0 : CHK_RET(CheckTransportEndPointInfo(envelope.envelope.epParam.src, envelope.envelope.epParam.dst));
202 :
203 0 : u32 recvSize = 0;
204 0 : u32 envelopSize = envelope.envelope.transData.count * SIZE_TABLE[envelope.envelope.transData.dataType];
205 0 : for (s32 i = 0; i < bufCount; i++) {
206 0 : if ((reinterpret_cast<u64>(buf[i]) == 0) && ((bufCount != 0) || (envelope.envelope.transData.count != 0))) {
207 0 : HCCL_ERROR("[Check][RecvEnvelope]Imrecv buffer[%p] or count[%llu] is invalid",
208 : reinterpret_cast<u64>(buf[i]), envelope.envelope.transData.count);
209 0 : return HCCL_E_PARA;
210 : }
211 0 : recvSize += count[i] * SIZE_TABLE[datatype];
212 : }
213 :
214 0 : if (recvSize < envelopSize) {
215 0 : HCCL_ERROR("[Check][RecvEnvelope] recvSize[%u Byte] is less than envelop total Size[%u Byte]", recvSize, envelopSize);
216 0 : return HCCL_E_PARA;
217 : }
218 0 : if (datatype != envelope.envelope.transData.dataType) {
219 0 : HCCL_ERROR("[Check][RecvEnvelope]Imrecv input dataType[%s] should be Isend dataType[%s]",
220 : GetDataTypeEnumStr(datatype).c_str(),
221 : GetDataTypeEnumStr(envelope.envelope.transData.dataType).c_str());
222 0 : return HCCL_E_PARA;
223 : }
224 :
225 0 : return HCCL_SUCCESS;
226 : }
227 :
228 0 : HcclResult TransportHeterog::GenerateRecvMessage(HcclEnvelopeSummary &recvEnvelope, HcclMessageInfo *&msg,
229 : HcclStatus &status)
230 : {
231 0 : msg = pMsgInfosMem_->Alloc();
232 0 : CHK_PTR_NULL(msg);
233 :
234 0 : msg->transportHandle = this;
235 0 : msg->envelope = recvEnvelope;
236 0 : status.srcRank = recvEnvelope.envelope.epParam.src.rank;
237 0 : status.tag = recvEnvelope.envelope.epParam.src.tag;
238 0 : status.count = recvEnvelope.envelope.transData.count;
239 0 : status.error = recvEnvelope.status;
240 0 : return HCCL_SUCCESS;
241 : }
242 :
243 0 : HcclResult TransportHeterog::FreeRecvMessage(HcclMessageInfo &msg) const
244 : {
245 0 : msg.commHandle = nullptr;
246 0 : msg.transportHandle = nullptr;
247 0 : CHK_RET(pMsgInfosMem_->Free(&msg));
248 0 : return HCCL_SUCCESS;
249 : }
250 :
251 0 : HcclResult TransportHeterog::ProbeNothing(s32 &flag, HcclMessageInfo *&msg, HcclStatus &status) const
252 : {
253 0 : flag = HCCL_IMPROBE_INCOMPLETED;
254 0 : msg = nullptr;
255 0 : status.srcRank = -1;
256 0 : status.tag = -1;
257 0 : status.error = -1;
258 0 : status.count = -1;
259 0 : return HCCL_SUCCESS;
260 : }
261 :
262 0 : HcclResult TransportHeterog::AddSocketWhiteList(string& tag)
263 : {
264 0 : std::vector<SocketWlistInfoT> whiteList(1);
265 0 : constexpr u32 connLimit = 4096;
266 0 : whiteList[0].remoteIp.addr = peerIp_.GetBinaryAddress().addr;
267 0 : whiteList[0].remoteIp.addr6 = peerIp_.GetBinaryAddress().addr6;
268 0 : whiteList[0].connLimit = connLimit;
269 0 : CHK_SAFETY_FUNC_RET(memcpy_s(&whiteList[0].tag, sizeof(whiteList[0].tag), tag.c_str(), tag.size() + 1));
270 :
271 0 : CHK_RET(hrtRaSocketWhiteListAdd(nicSocketHandle_, whiteList.data(), SINGLE_WHITE_LIST_NUM));
272 0 : HCCL_INFO("TransportHeterogRoce::AddSocketWhiteList ip[%s], tag[%s]",
273 : peerIp_.GetReadableAddress(), whiteList[0].tag);
274 0 : return HCCL_SUCCESS;
275 0 : }
276 :
277 0 : HcclResult TransportHeterog::PrepareSocketInfo(s32 type, s32 linkNum, const string &clientTag, const string &serverTag)
278 : {
279 0 : initSM_.locInitInfo.signal = SYNC_SIGNAL;
280 0 : initSM_.locInitInfo.protocolType = type;
281 0 : HcclInAddr peerAddr = peerIp_.GetBinaryAddress();
282 0 : for (int i = initSM_.locInitInfo.socketInfo.size(); i < linkNum; i++) {
283 0 : string tag = transTag_ + "_" + to_string(i) + "_";
284 0 : if (initSM_.locInitInfo.role == CLIENT_ROLE_SOCKET) {
285 0 : tag += clientTag;
286 : SocketConnectInfoT tmpConnInfo;
287 0 : tmpConnInfo.socketHandle = nicSocketHandle_;
288 0 : tmpConnInfo.remoteIp.addr = peerAddr.addr;
289 0 : tmpConnInfo.remoteIp.addr6 = peerAddr.addr6;
290 0 : tmpConnInfo.port = peerPort_;
291 0 : CHK_SAFETY_FUNC_RET(strncpy_s(tmpConnInfo.tag, SOCK_CONN_TAG_SIZE, tag.c_str(), tag.length() + 1));
292 0 : initSM_.locInitInfo.socketConnInfo.emplace_back(tmpConnInfo);
293 : } else {
294 0 : tag += serverTag;
295 : }
296 :
297 0 : HCCL_INFO("link[%d] tag[%s]", i, tag.c_str());
298 0 : SocketInfoT tmpInfo = {};
299 0 : tmpInfo.socketHandle = nicSocketHandle_;
300 0 : tmpInfo.fdHandle = nullptr;
301 0 : tmpInfo.remoteIp.addr = peerAddr.addr;
302 0 : tmpInfo.remoteIp.addr6 = peerAddr.addr6;
303 0 : tmpInfo.status = CONNECT_FAIL;
304 0 : CHK_SAFETY_FUNC_RET(strncpy_s(tmpInfo.tag, SOCK_CONN_TAG_SIZE, tag.c_str(), tag.length() + 1));
305 0 : initSM_.locInitInfo.socketInfo.emplace_back(tmpInfo);
306 0 : if (isHdcMode_ || remoteIsHdc_) {
307 : // hdc模式下hccp默认开启白名单校验,因此要配置tag进入白名单
308 0 : CHK_RET(AddSocketWhiteList(tag));
309 : }
310 0 : }
311 :
312 0 : return HCCL_SUCCESS;
313 : }
314 :
315 0 : HcclResult TransportHeterog::InitTransportConnect(s32 type, s32 linkNum)
316 : {
317 0 : if (selfIp_ == peerIp_) {
318 0 : initSM_.locInitInfo.role = (selfPort_ < peerPort_) ? SERVER_ROLE_SOCKET : CLIENT_ROLE_SOCKET;
319 : } else {
320 0 : initSM_.locInitInfo.role = (selfIp_ < peerIp_) ? SERVER_ROLE_SOCKET : CLIENT_ROLE_SOCKET;
321 : }
322 :
323 0 : string clientTag = string(peerIp_.GetReadableIP()) + to_string(peerPort_) +
324 0 : string(selfIp_.GetReadableIP()) + to_string(selfPort_);
325 0 : string serverTag = string(selfIp_.GetReadableIP()) + to_string(selfPort_) +
326 0 : string(peerIp_.GetReadableIP()) + to_string(peerPort_);
327 0 : CHK_RET(PrepareSocketInfo(type, linkNum, clientTag, serverTag));
328 :
329 0 : if (initSM_.locInitInfo.role == CLIENT_ROLE_SOCKET) {
330 0 : CHK_RET(TryTransition(HCCL_SUCCESS, true, ConnState::CONN_STATE_CONNECT_CHECK_SOCKET));
331 : } else {
332 0 : CHK_RET(TryTransition(HCCL_SUCCESS, true, ConnState::CONN_STATE_GET_CHECK_SOCKET));
333 : }
334 :
335 0 : u32 rankTableCrc = TransportHeterog::rankTableCrc_.load();
336 : //序列化信息
337 0 : std::ostringstream oss;
338 0 : oss.write(reinterpret_cast<const char_t *>(&rankTableCrc),
339 : sizeof(rankTableCrc));
340 0 : oss.write(reinterpret_cast<const char_t *>(&initSM_.locInitInfo.protocolType),
341 : sizeof(initSM_.locInitInfo.protocolType));
342 :
343 0 : CHK_SAFETY_FUNC_RET(memcpy_s(&(initSM_.locInitInfo.checkFrame[0]), HETEROG_MAX_FRAME_LEN - 1,
344 : oss.str().c_str(), oss.str().size()));
345 :
346 0 : return HCCL_SUCCESS;
347 0 : }
348 :
349 0 : HcclResult TransportHeterog::InitTransportConnect(s32 type, u32 role, s32 linkNum, u32 tag)
350 : {
351 0 : HCCL_DEBUG("TransportHeterog InitTransportConnect start type[%d] role[%u] linkNum[%d]", type, role, linkNum);
352 0 : initSM_.locInitInfo.role = role;
353 :
354 0 : string clientTag = string(selfIp_.GetReadableIP()) + "_" + to_string(0) + "_" +
355 0 : string(peerIp_.GetReadableIP()) + "_" + to_string(peerPort_) + "_" + to_string(tag);
356 0 : string serverTag = string(peerIp_.GetReadableIP()) + "_" + to_string(0) + "_" +
357 0 : string(selfIp_.GetReadableIP()) + "_" + to_string(selfPort_) + "_" + to_string(tag);
358 0 : CHK_RET(PrepareSocketInfo(type, linkNum, clientTag, serverTag));
359 :
360 0 : if (initSM_.locInitInfo.role == CLIENT_ROLE_SOCKET) {
361 0 : CHK_RET(TryTransition(HCCL_SUCCESS, true, ConnState::CONN_STATE_CONNECT_ALL_SOCKET));
362 : } else {
363 0 : CHK_RET(TryTransition(HCCL_SUCCESS, true, ConnState::CONN_STATE_GET_ALL_SOCKET));
364 : }
365 :
366 0 : return HCCL_SUCCESS;
367 0 : }
368 :
369 0 : HcclResult TransportHeterog::ConnectSocket(SocketConnectInfoT conn[], u32 num, bool &completed)
370 : {
371 0 : HcclResult ret = hrtRaSocketNonBlockBatchConnect(conn, num);
372 0 : completed = (ret == HCCL_SUCCESS);
373 0 : return ret;
374 : }
375 :
376 0 : HcclResult TransportHeterog::GetSocket(u32 role, struct SocketInfoT info[], u32 num, u32 &connectedNum,
377 : bool &completed)
378 : {
379 0 : HcclResult ret = HCCL_SUCCESS;
380 0 : for (u32 i = 0; i < num; i++) {
381 0 : if (info[i].status == CONNECT_FAIL) {
382 0 : SocketInfoT tmpInfo = info[i];
383 0 : u32 tmpNum = 0;
384 0 : ret = hrtRaNonBlockGetSockets(role, &tmpInfo, 1, &tmpNum);
385 0 : if (ret == HCCL_SUCCESS && tmpNum == 1 && tmpInfo.status == CONNECT_OK && tmpInfo.fdHandle != nullptr) {
386 0 : info[i].status = CONNECT_OK;
387 0 : info[i].fdHandle = tmpInfo.fdHandle;
388 0 : connectedNum += 1;
389 0 : } else if (ret == HCCL_E_AGAIN) {
390 0 : continue;
391 : } else {
392 0 : HCCL_WARNING("hrtRaNonBlockGetSockets ret[%d]", ret);
393 0 : return ret;
394 : }
395 : }
396 : }
397 :
398 0 : completed = connectedNum == num ? true : false;
399 0 : return ret;
400 : }
401 :
402 0 : HcclResult TransportHeterog::SocketSend(const FdHandle fdHandle, void *data, u64 size, u64 &sentSize, bool &completed)
403 : {
404 0 : HCCL_DEBUG("TransportHeterog::SocketSend start fdHandle[%p]", fdHandle);
405 0 : u64 tmpSize = 0;
406 : HcclResult ret =
407 0 : hrtRaSocketNonBlockSendHeterog(fdHandle, reinterpret_cast<char *>(data) + sentSize, size - sentSize, &tmpSize);
408 0 : if (ret == HCCL_SUCCESS) {
409 0 : sentSize += tmpSize;
410 0 : if (size == sentSize) {
411 0 : completed = true;
412 0 : } else if (sentSize > size) {
413 0 : HCCL_ERROR("SocketSend sentSize[%llu Byte] bigger than size[%llu Byte] completed[%u Byte] tmpSize[%llu Byte]",
414 : sentSize, size, completed, tmpSize);
415 0 : return HCCL_E_NETWORK;
416 : }
417 : } else {
418 0 : completed = false;
419 : }
420 0 : if (ret != HCCL_SUCCESS && ret != HCCL_E_AGAIN) {
421 0 : HCCL_ERROR("TransportHeterog::SocketSend size[%llu Byte] recvSize[%llu Byte] completed[%u Byte] tmpSize[%llu Byte]",
422 : size, sentSize, completed, tmpSize);
423 : }
424 0 : return ret;
425 : }
426 :
427 0 : HcclResult TransportHeterog::SocketRecv(const FdHandle fdHandle, void *data, u64 size, u64 &recvSize, bool &completed)
428 : {
429 0 : HCCL_DEBUG("TransportHeterog::SocketRecv start fdHandle[%p]", fdHandle);
430 0 : u64 tmpSize = 0;
431 : HcclResult ret =
432 0 : hrtRaSocketNonBlockRecvHeterog(fdHandle, reinterpret_cast<char *>(data) + recvSize, size - recvSize, &tmpSize);
433 0 : if (ret == HCCL_SUCCESS) {
434 0 : recvSize += tmpSize;
435 0 : if (size == recvSize) {
436 0 : completed = true;
437 0 : } else if (recvSize > size) {
438 0 : HCCL_ERROR("SocketRecv recvSize[%llu Byte] bigger than size[%llu Byte] completed[%u Byte] tmpSize[%llu Byte]",
439 : recvSize, size, completed, tmpSize);
440 0 : return HCCL_E_NETWORK;
441 : }
442 : } else {
443 0 : completed = false;
444 : }
445 0 : if (ret != HCCL_SUCCESS && ret != HCCL_E_AGAIN) {
446 0 : HCCL_ERROR("TransportHeterog::SocketRecv size[%llu Byte] recvSize[%llu Byte] completed[%u Byte] tmpSize[%llu Byte]",
447 : size, recvSize, completed, tmpSize);
448 : }
449 0 : return ret;
450 : }
451 :
452 0 : HcclResult TransportHeterog::SocketClose()
453 : {
454 0 : u32 closeConnCount = 0;
455 0 : SocketCloseInfoT conns[MAX_LINK_NUM]{};
456 0 : CHK_PRT_RET(initSM_.locInitInfo.socketInfo.size() > MAX_LINK_NUM,
457 : HCCL_ERROR("locInitInfo.socketInfo size can't exceed MAX_LINK_NUM, size[%d]",
458 : initSM_.locInitInfo.socketInfo.size()), HCCL_E_PARA);
459 0 : for (size_t i = 0; i < initSM_.locInitInfo.socketInfo.size(); i++) {
460 0 : if (initSM_.locInitInfo.socketInfo[i].fdHandle != nullptr) {
461 0 : conns[i].socketHandle = nicSocketHandle_;
462 0 : conns[i].fdHandle = initSM_.locInitInfo.socketInfo[i].fdHandle;
463 0 : conns[i].disuseLinger = static_cast<s32>(forceClose_);
464 0 : closeConnCount++;
465 : }
466 : }
467 :
468 0 : if (closeConnCount > 0) {
469 0 : if (hrtRaSocketBatchClose(conns, closeConnCount) != HCCL_SUCCESS) {
470 0 : HCCL_ERROR("[Destroy][TransportHeterog]ra socket batch close failed");
471 : }
472 : }
473 0 : return HCCL_SUCCESS;
474 : }
475 :
476 0 : HcclResult TransportHeterog::CheckConsistentFrame()
477 : {
478 0 : std::string msg;
479 0 : msg.resize(sizeof(initSM_.locInitInfo.checkFrame));
480 0 : CHK_SAFETY_FUNC_RET(memcpy_s(reinterpret_cast<void*>(const_cast<char_t*>(msg.data())), HETEROG_MAX_FRAME_LEN - 1,
481 : &(initSM_.remInitInfo.checkFrame[0]), HETEROG_MAX_FRAME_LEN - 1));
482 :
483 0 : std::istringstream iss(msg);
484 0 : u32 localRankTableCrc = TransportHeterog::rankTableCrc_.load();
485 0 : u32 remoteRankTableCrc = 0;
486 0 : iss.read(reinterpret_cast<char_t *>(&remoteRankTableCrc), sizeof(remoteRankTableCrc));
487 0 : iss.read(reinterpret_cast<char_t *>(&initSM_.remInitInfo.protocolType), sizeof(initSM_.remInitInfo.protocolType));
488 :
489 0 : bool bIsDiff = false;
490 0 : if (remoteRankTableCrc != localRankTableCrc) {
491 0 : RPT_INPUT_ERR(true,
492 : "EI0005",
493 : std::vector<std::string>({"ccl_op", "group", "para_name", "local_para", "remote_para"}),
494 : std::vector<std::string>(
495 : {"HcomBatchGet", transTag_, "ranktable CRC", std::to_string(localRankTableCrc), std::to_string(remoteRankTableCrc)}));
496 0 : HCCL_ERROR("[%s][%s]errNo[0x%016llx] ranktable CRC check failed, crcValue[%u], receive crcvalue[%u].",
497 : LOG_KEYWORDS_INIT_CHANNEL.c_str(),
498 : LOG_KEYWORDS_PARAMETER_CONFLICT.c_str(),
499 : HCCL_ERROR_CODE(HCCL_E_INTERNAL),
500 : localRankTableCrc,
501 : remoteRankTableCrc);
502 0 : bIsDiff = true;
503 : }
504 :
505 0 : if (initSM_.remInitInfo.protocolType != initSM_.locInitInfo.protocolType) {
506 0 : HCCL_ERROR("[CheckConsistentFrame][CompareFrame]errNo[0x%016llx] ProtocolType check fail",
507 : HCCL_ERROR_CODE(HCCL_E_INTERNAL));
508 0 : bIsDiff = true;
509 : }
510 0 : if (bIsDiff) {
511 0 : return HCCL_E_INTERNAL;
512 : }
513 0 : return HCCL_SUCCESS;
514 0 : }
515 :
516 0 : ConnState TransportHeterog::GetState()
517 : {
518 0 : return connState_.load();
519 : }
520 :
521 0 : HcclResult TransportHeterog::TryTransition(HcclResult ret, bool completed, ConnState nextState)
522 : {
523 0 : if (ret == HCCL_SUCCESS && completed) {
524 0 : HCCL_INFO("link[%s]: state[%d] transfer to state[%d]", initSM_.locInitInfo.socketInfo[0].tag, GetState(),
525 : nextState);
526 0 : connState_.store(nextState);
527 0 : CHK_RET(EnterStateProcess(nextState));
528 0 : } else if ((ret == HCCL_SUCCESS && !completed) || ret == HCCL_E_AGAIN) {
529 0 : HCCL_DEBUG("link[%s]: state[%d] not complete, hold", initSM_.locInitInfo.socketInfo[0].tag, GetState());
530 : } else {
531 0 : HCCL_ERROR("link[%s]: State[%d] execute failed errno[%d][%s]", initSM_.locInitInfo.socketInfo[0].tag,
532 : nextState, errno, strerror(errno));
533 0 : return HCCL_E_INTERNAL;
534 : }
535 :
536 0 : return HCCL_SUCCESS;
537 : }
538 :
539 0 : HcclResult TransportHeterog::ConnectAsync()
540 : {
541 0 : if (initSM_.locInitInfo.socketInfo.size() == 0) {
542 0 : HCCL_ERROR("[ConnectAsync]initSM_.locInitInfo.socketInfo is invalid!");
543 0 : return HCCL_E_PARA;
544 : }
545 0 : HCCL_DEBUG("link[%s]: Call ConnectAsync", initSM_.locInitInfo.socketInfo[0].tag);
546 0 : CHK_RET(LoopStateProcess());
547 0 : return HCCL_SUCCESS;
548 : }
549 :
550 0 : HcclResult TransportHeterog::SetDeviceIndex(s32 index)
551 : {
552 0 : index_ = index;
553 0 : return HCCL_SUCCESS;
554 : }
555 :
556 0 : void TransportHeterog::AddRecvEnvelopNum()
557 : {
558 0 : recvEnvelopNum_++;
559 0 : return;
560 : }
561 :
562 0 : void TransportHeterog::SubRecvEnvelopNum()
563 : {
564 0 : recvEnvelopNum_--;
565 0 : return;
566 : }
567 :
568 0 : u32 TransportHeterog::GetRecvEnvelopNum()
569 : {
570 0 : return recvEnvelopNum_;
571 : }
572 :
573 0 : HcclResult TransportHeterog::BlockSend(const TransData &sendData, const TransportEndPointParam &epParam,
574 : HcclRequestInfo *&request, s32 waitTimeOut)
575 : {
576 0 : return HCCL_SUCCESS;
577 : }
578 :
579 0 : HcclResult TransportHeterog::BlockRecv(const TransData &recvData, bool matched,
580 : TransportHeterog *&transport, s32 waitTimeOut, s32 waitPayloadTimeOut)
581 : {
582 0 : return HCCL_SUCCESS;
583 : }
584 :
585 0 : HcclResult TransportHeterog::CheckAndPushBuildLink()
586 : {
587 : // 建链未完成时，继续推进建链流程；
588 0 : if (GetState() != ConnState::CONN_STATE_COMPLETE) {
589 0 : CHK_RET(ConnectAsync());
590 : }
591 :
592 0 : return (GetState() == ConnState::CONN_STATE_COMPLETE) ? HCCL_SUCCESS : HCCL_E_AGAIN;
593 : }
594 :
595 0 : HcclResult TransportHeterog::WaitBuildLinkComplete()
596 : {
597 0 : HCCL_INFO("linkTag[%s] WaitBuildLinkComplete Begin! State[%d]", initSM_.locInitInfo.socketInfo[0].tag, GetState());
598 0 : auto startTime = chrono::steady_clock::now();
599 0 : auto timeout = chrono::seconds(GetExternalInputHcclLinkTimeOut());
600 :
601 0 : while ((chrono::steady_clock::now() - startTime) < timeout) {
602 0 : HcclResult ret = CheckAndPushBuildLink();
603 0 : if (ret == HCCL_E_AGAIN) {
604 0 : SaluSleep(WAIT_LINK_BUILD_DELAY_TIME_US);
605 0 : continue;
606 : }
607 :
608 0 : if (ret == HCCL_SUCCESS) {
609 0 : HCCL_INFO("Transport heterog connect success, localRank[%u], localIp[%s], remoteRank[%u], "
610 : "remoteIp[%s], linkTag[%s]!", localRank_, selfIp_.GetReadableAddress(),
611 : remoteRank_, peerIp_.GetReadableAddress(), initSM_.locInitInfo.socketInfo[0].tag);
612 : } else {
613 0 : HCCL_ERROR("Transport heterog connect failed, ret[%d]!", ret);
614 : }
615 :
616 0 : return ret;
617 : }
618 :
619 0 : HCCL_ERROR("WaitBuildLinkComplete timeOut[%d] s, localIp[%s], "
620 : "remoteIp[%s], linkTag[%s], State[%d]", GetExternalInputHcclLinkTimeOut(),
621 : selfIp_.GetReadableAddress(), peerIp_.GetReadableAddress(), initSM_.locInitInfo.socketInfo[0].tag,
622 : GetState());
623 :
624 0 : return HCCL_E_TIMEOUT;
625 : }
626 :
627 0 : HcclResult TransportHeterog::Iwrite(const TransData &sendData, const HcclEnvelope &envelope, HcclRequestInfo *&request)
628 : {
629 0 : HCCL_WARNING("Empty TransportHeterog::Iwrite is called.");
630 0 : return HCCL_SUCCESS;
631 : }
632 :
633 0 : HcclResult TransportHeterog::GetRemoteIsendDoneSignal(std::shared_ptr<LocalIpcNotify> &signal)
634 : {
635 0 : HCCL_WARNING("Empty TransportHeterog::GetRemoteIsendDoneSignal is called.");
636 0 : return HCCL_SUCCESS;
637 : }
638 :
639 0 : HcclResult TransportHeterog::GetRemoteImrecvDoneSignal(std::shared_ptr<LocalIpcNotify> &signal)
640 : {
641 0 : HCCL_WARNING("Empty TransportHeterog::GetRemoteImrecvDoneSignal is called.");
642 0 : return HCCL_SUCCESS;
643 : }
644 :
645 0 : void TransportHeterog::GetLinkTag(std::string &tag)
646 : {
647 0 : HCCL_WARNING("Empty TransportHeterog::GetLinkTag is called.");
648 0 : return;
649 : }
650 :
651 0 : void TransportHeterog::SetForceClose()
652 : {
653 0 : forceClose_ = true;
654 0 : }
655 :
656 0 : void TransportHeterog::RecordRankTableCrc(const u32 crcValue)
657 : {
658 : rankTableCrc_.store(crcValue);
659 0 : return;
660 : }
661 :
662 : } // namespace hccl
|
