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