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