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_raw_roce.h"
12 : #include "adapter_hccp.h"
13 : #include "adapter_verbs.h"
14 : #include "externalinput.h"
15 : #include "network/hccp_common.h"
16 :
17 : using namespace std;
18 : namespace hccl {
19 :
20 : constexpr s32 TAG_QP_APPEND = 1;
21 : constexpr s32 DATA_QP_APPEND = 2;
22 : constexpr u32 RECV_WQE_BATCH_NUM = MEM_BLOCK_RECV_WQE_BATCH_NUM;
23 : constexpr u32 RECV_WQE_NUM_THRESHOLD = 96;
24 : constexpr u32 RECV_WQE_BATCH_SUPPLEMENT = 96;
25 :
26 0 : TransportHeterogRawRoce::TransportHeterogRawRoce(const std::string &transTag, HcclIpAddress &selfIp,
27 0 : HcclIpAddress &peerIp, u32 peerPort, u32 selfPort, const TransportResourceInfo &transportResourceInfo)
28 0 : : TransportHeterogRoce(transTag, selfIp, peerIp, peerPort, selfPort, transportResourceInfo)
29 : {
30 0 : }
31 :
32 0 : TransportHeterogRawRoce::~TransportHeterogRawRoce()
33 : {
34 0 : }
35 :
36 0 : HcclResult TransportHeterogRawRoce::Init()
37 : {
38 0 : return HCCL_SUCCESS;
39 : }
40 :
41 0 : HcclResult TransportHeterogRawRoce::Init(SocketInfoT &socketInfo, RdmaHandle rdmaHandle, MrHandle mrHandle)
42 : {
43 0 : CHK_PTR_NULL(socketInfo.socketHandle);
44 0 : CHK_PTR_NULL(socketInfo.fdHandle);
45 0 : CHK_PTR_NULL(rdmaHandle);
46 0 : CHK_PTR_NULL(mrHandle);
47 0 : HCCL_INFO("TransportHeterogRawRoce Init start socketHandle[%p] fdHandle[%p] rdmaHandle[%p] mrHandle[%p]",
48 : socketInfo.socketHandle, socketInfo.fdHandle, rdmaHandle, mrHandle);
49 :
50 0 : mrManager_ = static_cast<MrManager *>(mrHandle);
51 0 : nicRdmaHandle_ = rdmaHandle;
52 0 : nicSocketHandle_ = socketInfo.socketHandle;
53 :
54 0 : SocketConnectInfoT tmpConnInfo{};
55 0 : tmpConnInfo.port = HETEROG_CCL_PORT;
56 0 : initSM_.locInitInfo.socketConnInfo.emplace_back(tmpConnInfo);
57 0 : initSM_.locInitInfo.socketInfo.emplace_back(socketInfo);
58 :
59 0 : CHK_RET(CheckRecvMsgAndRequestBuffer());
60 :
61 0 : CHK_RET(PreQpConnect());
62 :
63 0 : CHK_RET(TryTransition(HCCL_SUCCESS, true, ConnState::CONN_STATE_GET_TAG_QP_ATTR));
64 :
65 0 : CHK_RET(ConnectAsync());
66 :
67 0 : HCCL_INFO("TransportHeterogRoce Init success");
68 0 : return HCCL_SUCCESS;
69 : }
70 :
71 0 : HcclResult TransportHeterogRawRoce::CreateCqAndQp()
72 : {
73 0 : CHK_RET(CreateQpWithSharedCq(nicRdmaHandle_, selfIp_, peerIp_, -1, -1,
74 : tagQpInfo_, TAG_QP_APPEND, MAX_SCATTER_BUF_NUM));
75 0 : CHK_RET(CreateQpWithSharedCq(nicRdmaHandle_, selfIp_, peerIp_, -1, -1,
76 : dataQpInfo_, DATA_QP_APPEND, MAX_SCATTER_BUF_NUM));
77 0 : return HCCL_SUCCESS;
78 : }
79 :
80 0 : HcclResult TransportHeterogRawRoce::DestroyCqAndQp()
81 : {
82 0 : HCCL_INFO("TransportHeterogRawRoce DestroyCqAndQp");
83 0 : CHK_RET(DestroyQpWithSharedCq(tagQpInfo_, TAG_QP_APPEND));
84 0 : tagQpInfo_ = QpInfo();
85 0 : CHK_RET(DestroyQpWithSharedCq(dataQpInfo_, DATA_QP_APPEND));
86 0 : dataQpInfo_ = QpInfo();
87 0 : return HCCL_SUCCESS;
88 : }
89 :
90 0 : HcclResult TransportHeterogRawRoce::PreQpConnect()
91 : {
92 0 : CHK_RET(TransportHeterogRoce::PreQpConnect());
93 :
94 0 : CHK_SAFETY_FUNC_RET(memset_s(&dataReadWrScatter_, sizeof(struct ibv_send_wr), 0, sizeof(struct ibv_send_wr)));
95 0 : dataReadWrScatter_.next = &dataAckWrScatter_;
96 0 : dataReadWrScatter_.opcode = IBV_WR_RDMA_READ;
97 0 : dataReadWrScatter_.send_flags = IBV_SEND_SIGNALED | IBV_SEND_FENCE;
98 :
99 0 : CHK_SAFETY_FUNC_RET(memset_s(&dataAckWrScatter_, sizeof(struct ibv_send_wr), 0, sizeof(struct ibv_send_wr)));
100 0 : dataAckWrScatter_.sg_list = &dataAckSge_;
101 0 : dataAckWrScatter_.next = nullptr;
102 0 : dataAckWrScatter_.num_sge = 1;
103 0 : dataAckWrScatter_.opcode = IBV_WR_SEND_WITH_IMM;
104 0 : dataAckWrScatter_.send_flags = IBV_SEND_FENCE | IBV_SEND_INLINE;
105 :
106 0 : return HCCL_SUCCESS;
107 : }
108 :
109 0 : HcclResult TransportHeterogRawRoce::EnterStateProcess(ConnState nextState)
110 : {
111 0 : switch (nextState) {
112 0 : case ConnState::CONN_STATE_GET_TAG_QP_ATTR:
113 0 : break;
114 0 : case ConnState::CONN_STATE_SEND_TAG_QP_INFO:
115 0 : CHK_RET(PrepareModifyInfo(localTagQpAttr_, localTagModifyInfo_));
116 0 : initSM_.size = sizeof(localTagModifyInfo_);
117 0 : initSM_.completeSize = 0;
118 0 : break;
119 0 : case ConnState::CONN_STATE_RECV_TAG_QP_INFO:
120 0 : initSM_.size = sizeof(remoteTagModifyInfo_);
121 0 : initSM_.completeSize = 0;
122 0 : break;
123 0 : case ConnState::CONN_STATE_MODIFY_TAG_QP:
124 0 : break;
125 0 : case ConnState::CONN_STATE_GET_DATA_QP_ATTR:
126 0 : break;
127 0 : case ConnState::CONN_STATE_SEND_DATA_QP_INFO:
128 0 : CHK_RET(PrepareModifyInfo(localDataQpAttr_, localDataModifyInfo_));
129 0 : initSM_.size = sizeof(localDataModifyInfo_);
130 0 : initSM_.completeSize = 0;
131 0 : break;
132 0 : case ConnState::CONN_STATE_RECV_DATA_QP_INFO:
133 0 : initSM_.size = sizeof(remoteDataModifyInfo_);
134 0 : initSM_.completeSize = 0;
135 0 : break;
136 0 : case ConnState::CONN_STATE_MODIFY_DATA_QP:
137 0 : break;
138 0 : case ConnState::CONN_STATE_SEND_STATUS:
139 0 : initSM_.locInitInfo.signal = SYNC_SIGNAL;
140 0 : initSM_.size = sizeof(initSM_.locInitInfo.signal);
141 0 : initSM_.completeSize = 0;
142 0 : break;
143 0 : case ConnState::CONN_STATE_RECV_STATUS:
144 0 : initSM_.size = sizeof(initSM_.remInitInfo.signal);
145 0 : initSM_.completeSize = 0;
146 0 : break;
147 0 : case ConnState::CONN_STATE_COMPLETE:
148 0 : HCCL_INFO("link[%s]: connect complete", initSM_.locInitInfo.socketInfo[0].tag);
149 0 : break;
150 0 : default:
151 0 : HCCL_INFO("link[%s]: state[%u] no need to do anything", initSM_.locInitInfo.socketInfo[0].tag, nextState);
152 : }
153 :
154 0 : return HCCL_SUCCESS;
155 : }
156 :
157 0 : HcclResult TransportHeterogRawRoce::LoopStateProcess()
158 : {
159 0 : HcclResult testRet = HCCL_SUCCESS;
160 0 : bool completed = false;
161 0 : switch (GetState()) {
162 0 : case ConnState::CONN_STATE_GET_TAG_QP_ATTR:
163 0 : testRet = GetQpAttr(tagQpInfo_.qpHandle, &localTagQpAttr_, completed);
164 0 : CHK_RET(TryTransition(testRet, completed, ConnState::CONN_STATE_SEND_TAG_QP_INFO));
165 0 : break;
166 0 : case ConnState::CONN_STATE_SEND_TAG_QP_INFO:
167 0 : testRet = SocketSend(initSM_.locInitInfo.socketInfo[0].fdHandle, &localTagModifyInfo_,
168 0 : initSM_.size, initSM_.completeSize, completed);
169 0 : CHK_RET(TryTransition(testRet, completed, ConnState::CONN_STATE_RECV_TAG_QP_INFO));
170 0 : break;
171 0 : case ConnState::CONN_STATE_RECV_TAG_QP_INFO:
172 0 : testRet = SocketRecv(initSM_.locInitInfo.socketInfo[0].fdHandle, &remoteTagModifyInfo_,
173 0 : initSM_.size, initSM_.completeSize, completed);
174 0 : CHK_RET(TryTransition(testRet, completed, ConnState::CONN_STATE_MODIFY_TAG_QP));
175 0 : break;
176 0 : case ConnState::CONN_STATE_MODIFY_TAG_QP:
177 0 : testRet = TypicalQpModify(tagQpInfo_.qpHandle, &localTagModifyInfo_, &remoteTagModifyInfo_, completed);
178 0 : CHK_RET(TryTransition(testRet, completed, ConnState::CONN_STATE_GET_DATA_QP_ATTR));
179 0 : break;
180 0 : case ConnState::CONN_STATE_GET_DATA_QP_ATTR:
181 0 : testRet = GetQpAttr(dataQpInfo_.qpHandle, &localDataQpAttr_, completed);
182 0 : CHK_RET(TryTransition(testRet, completed, ConnState::CONN_STATE_SEND_DATA_QP_INFO));
183 0 : break;
184 0 : case ConnState::CONN_STATE_SEND_DATA_QP_INFO:
185 0 : testRet = SocketSend(initSM_.locInitInfo.socketInfo[0].fdHandle, &localDataModifyInfo_,
186 0 : initSM_.size, initSM_.completeSize, completed);
187 0 : CHK_RET(TryTransition(testRet, completed, ConnState::CONN_STATE_RECV_DATA_QP_INFO));
188 0 : break;
189 0 : case ConnState::CONN_STATE_RECV_DATA_QP_INFO:
190 0 : testRet = SocketRecv(initSM_.locInitInfo.socketInfo[0].fdHandle, &remoteDataModifyInfo_,
191 0 : initSM_.size, initSM_.completeSize, completed);
192 0 : CHK_RET(TryTransition(testRet, completed, ConnState::CONN_STATE_MODIFY_DATA_QP));
193 0 : break;
194 0 : case ConnState::CONN_STATE_MODIFY_DATA_QP:
195 0 : testRet = TypicalQpModify(dataQpInfo_.qpHandle, &localDataModifyInfo_, &remoteDataModifyInfo_, completed);
196 0 : CHK_RET(TryTransition(testRet, completed, ConnState::CONN_STATE_SEND_STATUS));
197 0 : break;
198 0 : case ConnState::CONN_STATE_SEND_STATUS:
199 0 : testRet = SocketSend(initSM_.locInitInfo.socketInfo[0].fdHandle,
200 0 : &(initSM_.locInitInfo.signal), initSM_.size, initSM_.completeSize, completed);
201 0 : CHK_RET(TryTransition(testRet, completed, ConnState::CONN_STATE_RECV_STATUS));
202 0 : break;
203 0 : case ConnState::CONN_STATE_RECV_STATUS:
204 0 : testRet = SocketRecv(initSM_.locInitInfo.socketInfo[0].fdHandle,
205 0 : &(initSM_.remInitInfo.signal), initSM_.size, initSM_.completeSize, completed);
206 0 : CHK_RET(TryTransition(testRet, completed, ConnState::CONN_STATE_COMPLETE));
207 0 : break;
208 0 : default:
209 0 : HCCL_ERROR("Establish communication connection failed[%s]: state[%u]",
210 : initSM_.locInitInfo.socketInfo[0].tag, GetState());
211 0 : return HCCL_E_INTERNAL;
212 : }
213 0 : return HCCL_SUCCESS;
214 : }
215 :
216 0 : HcclResult TransportHeterogRawRoce::PrepareModifyInfo(struct QpAttr &qpAttr, struct TypicalQp &typicalQpInfo)
217 : {
218 0 : typicalQpInfo.qpn = qpAttr.qpn;
219 0 : typicalQpInfo.psn = qpAttr.psn;
220 0 : typicalQpInfo.gidIdx = qpAttr.gidIdx;
221 0 : typicalQpInfo.tc = GetExternalInputRdmaTrafficClass();
222 0 : typicalQpInfo.sl = GetExternalInputRdmaServerLevel();
223 0 : typicalQpInfo.retryCnt = GetExternalInputRdmaRetryCnt();
224 0 : typicalQpInfo.retryTime = GetExternalInputRdmaTimeOut();
225 0 : CHK_SAFETY_FUNC_RET(memcpy_s(typicalQpInfo.gid, HCCP_GID_RAW_LEN , qpAttr.gid, HCCP_GID_RAW_LEN ));
226 0 : HCCL_INFO("TransportHeterogRawRoce ModifyInfo qpn[%u] psn[%u] gid_idxp[%u] gid[%p] tc[%u] sl[%u] retryCnt[%u]"
227 : "retryTime[%u]", typicalQpInfo.qpn, typicalQpInfo.psn, typicalQpInfo.gidIdx, typicalQpInfo.gid,
228 : typicalQpInfo.tc, typicalQpInfo.sl, typicalQpInfo.retryCnt, typicalQpInfo.retryTime);
229 :
230 0 : return HCCL_SUCCESS;
231 : }
232 :
233 0 : HcclResult TransportHeterogRawRoce::ImrecvScatter(void *buf[], int count[], int bufCount, HcclDataType datatype,
234 : HcclMessageInfo &msg, HcclRequestInfo *&request)
235 : {
236 0 : CHK_RET(CheckRecvScatterEnvelope(buf, count, bufCount, datatype, msg.envelope));
237 0 : CHK_RET(GenerateRecvScatterRequest(msg, request));
238 :
239 0 : HcclEnvelope &envelope = msg.envelope.envelope;
240 0 : u32 dataSize = SIZE_TABLE[datatype];
241 0 : for (s32 i = 0; i < bufCount; i++) {
242 0 : u32 lkey = 0;
243 0 : u64 dstBuf = reinterpret_cast<u64>(buf[i]);
244 0 : CHK_RET(RegMr(reinterpret_cast<void *>(dstBuf), static_cast<u64>(count[i] * dataSize), lkey));
245 0 : dataReadSgeArry_[i].lkey = lkey;
246 : }
247 :
248 0 : for (s32 i = 0; i < bufCount; i++) {
249 0 : dataReadSgeArry_[i].addr = reinterpret_cast<uint64_t>(buf[i]);
250 : }
251 :
252 0 : for (s32 i = 0; i < bufCount; i++) {
253 0 : dataReadSgeArry_[i].length = count[i] * dataSize;
254 : }
255 :
256 0 : dataReadWrScatter_.sg_list = dataReadSgeArry_;
257 0 : dataReadWrScatter_.num_sge = bufCount;
258 0 : dataReadWrScatter_.wr_id = reinterpret_cast<uint64_t>(request);
259 0 : dataReadWrScatter_.wr.rdma.remote_addr = static_cast<uint64_t>(envelope.transData.srcBuf);
260 0 : dataReadWrScatter_.wr.rdma.rkey = envelope.key;
261 :
262 0 : dataAckSge_.addr = reinterpret_cast<uint64_t>(&envelope.msn);
263 0 : dataAckSge_.length = sizeof(uint64_t);
264 0 : dataAckSge_.lkey = 0;
265 0 : dataAckWrScatter_.wr_id = 0;
266 :
267 0 : struct ibv_send_wr *badWr = nullptr;
268 0 : HCCL_INFO("rdma read: remote addr[%llu] count[%d] datatype[%s] wrId[%llu] num_sge[%d] qpHandle[%p]",
269 : reinterpret_cast<u64>(envelope.transData.srcBuf), envelope.transData.count,
270 : GetDataTypeEnumStr(envelope.transData.dataType).c_str(), dataReadWr_.wr_id, bufCount,
271 : dataQpInfo_.qp);
272 0 : HcclResult ret = hrtIbvPostSend(dataQpInfo_.qp, &dataReadWrScatter_, &badWr);
273 0 : if (ret != HCCL_SUCCESS) {
274 0 : if (ret == HCCL_E_AGAIN) {
275 0 : HCCL_WARNING("rdma read post send wqe overflow.[%d]", ret);
276 : } else {
277 0 : HCCL_ERROR("rdma read fail: remote addr[%llx] count[%d] datatype[%s] wrId[%llu] bufCount[%d]",
278 : reinterpret_cast<u64>(envelope.transData.srcBuf), envelope.transData.count,
279 : GetDataTypeEnumStr(envelope.transData.dataType).c_str(), dataReadWr_.wr_id, bufCount);
280 : }
281 0 : return ret;
282 : }
283 0 : HCCL_INFO("rdma send ack: msn:0x%016llx request:%p", envelope.msn, &request);
284 :
285 0 : CHK_RET(FreeRecvMessage(msg));
286 0 : return HCCL_SUCCESS;
287 : }
288 :
289 0 : HcclResult TransportHeterogRawRoce::GetQpAttr(QpHandle &qpHandle, struct QpAttr *attr, bool &completed)
290 : {
291 0 : HcclResult ret = hrtRaGetQpAttr(qpHandle, attr);
292 0 : if (ret == HCCL_SUCCESS) {
293 0 : completed = true;
294 : } else {
295 0 : HCCL_ERROR("GetQpAttr fail qpHandle[%p] qpn[%u] udpSport[%u] psn[%u] gidIdx[%u] gid[%p] completed[%u]",
296 : qpHandle, attr->qpn, attr->udpSport, attr->psn, attr->gidIdx, attr->gid, completed);
297 : }
298 0 : return ret;
299 : }
300 :
301 0 : HcclResult TransportHeterogRawRoce::TypicalQpModify(QpHandle &qpHandle, struct TypicalQp* localQpInfo,
302 : struct TypicalQp* remoteQpInfo, bool &completed)
303 : {
304 0 : HcclResult ret = hrtRaTypicalQpModify(qpHandle, localQpInfo, remoteQpInfo);
305 0 : if (ret == HCCL_SUCCESS) {
306 0 : completed = true;
307 0 : } else if (ret != HCCL_E_AGAIN) {
308 0 : HCCL_ERROR("hrtRaTypicalQpModify fail qpHandle[%p] completed[%u]"
309 : "local: qpn[%u] psn[%u] gidIdx[%u] gid[%p] tc[%u] sl[%u] retryCnt[%u] retryTime[%u]"
310 : "remote: qpn[%u] psn[%u] gidIdx[%u] gid[%p] tc[%u] sl[%u] retryCnt[%u] retryTime[%u] ",
311 : qpHandle, completed, localQpInfo->qpn, localQpInfo->psn, localQpInfo->gidIdx, localQpInfo->gid,
312 : localQpInfo->tc, localQpInfo->sl, localQpInfo->retryCnt, localQpInfo->retryTime,
313 : remoteQpInfo->qpn, remoteQpInfo->psn, remoteQpInfo->gidIdx, remoteQpInfo->gid, remoteQpInfo->tc,
314 : remoteQpInfo->sl, remoteQpInfo->retryCnt, remoteQpInfo->retryTime);
315 : }
316 0 : return ret;
317 : }
318 : } // namespace hccl
|
