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