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 : #include "transport_heterog_event_roce.h"
11 :
12 : #include "log.h"
13 : #include "adapter_hal.h"
14 : #include "dlhal_function.h"
15 :
16 : using namespace std;
17 : namespace hccl {
18 : constexpr u32 MAX_CQECOUNT_ALLLINK = 64;
19 : constexpr s32 PROTOCOL_TYPE = 0;
20 :
21 : HcclReceivedEnvelope TransportHeterogEventRoce::gReceivedEnvelopes;
22 : std::mutex TransportHeterogEventRoce::gReceivedEnvelopesMutex;
23 :
24 : std::vector<std::atomic<int>> TransportHeterogEventRoce::gCqeCounterPerEvent(MAX_CQECOUNT_ALLLINK);
25 : std::vector<std::vector<void *>> TransportHeterogEventRoce::gAllLinkVec(MAX_CQECOUNT_ALLLINK);
26 : std::mutex TransportHeterogEventRoce::gAllLinkVecSendCompMutex;
27 : std::mutex TransportHeterogEventRoce::gAllLinkVecRecvReqMutex;
28 : std::mutex TransportHeterogEventRoce::gAllLinkVecRecvCompMutex;
29 :
30 : std::mutex TransportHeterogEventRoce::gPollTagRqLock;
31 : std::mutex TransportHeterogEventRoce::gPollDataRqLock;
32 : std::mutex TransportHeterogEventRoce::gPollDataSqLock;
33 :
34 : u32 TransportHeterogEventRoce::gEschedAckRef = 0;
35 : u32 TransportHeterogEventRoce::gAllLinkInitCount = 0;
36 : u32 TransportHeterogEventRoce::recvRequestEvent = 0;
37 : u32 TransportHeterogEventRoce::sendCompletionEvent = 0;
38 : u32 TransportHeterogEventRoce::recvCompletionEvent = 0;
39 :
40 : constexpr u32 RECV_WQE_BATCH_NUM = 8 * 1024;
41 : constexpr u32 RECV_WQE_NUM_THRESHOLD = 4 * 1024;
42 : constexpr u32 RECV_WQE_BATCH_SUPPLEMENT = 2 * 1024;
43 : constexpr u32 MAX_WR_NUM = 1023;
44 : constexpr s32 TAG_QP_APPEND = 1;
45 : constexpr s32 DATA_QP_APPEND = 2;
46 :
47 : atomic<u32> g_tagRecvWqeNum; // qp0上的recv wqe的数量,recv端消耗
48 : atomic<u32> g_dataRecvWqeNum; // qp1上的recv wqe的数量,send端消耗
49 : map<u32, TransportHeterogEventRoce *> TransportHeterogEventRoce::gQpnToTransportMap; // tag qpn和transport映射
50 : map<u32, atomic<u32>> TransportHeterogEventRoce::gQpnToSqMaxWrMap; // data qpn和sq max wr深度映射
51 : bool TransportHeterogEventRoce::gNeedRepoEvent = true;
52 :
53 0 : void TransportHeterogEventRoce::EschedAckCallbackRecvRequest(unsigned int devId, unsigned int subeventId, u8 *msg,
54 : unsigned int msgLen)
55 : {
56 : (void)subeventId;
57 : (void)msg;
58 : (void)msgLen;
59 0 : TransportHeterogEventRoce::EschedAckCallback(devId, HCCL_EVENT_RECV_REQUEST_MSG);
60 0 : }
61 :
62 0 : void TransportHeterogEventRoce::EschedAckCallbackSendCompletion(unsigned int devId, unsigned int subeventId, u8 *msg,
63 : unsigned int msgLen)
64 : {
65 : (void)subeventId;
66 : (void)msg;
67 : (void)msgLen;
68 0 : TransportHeterogEventRoce::EschedAckCallback(devId, HCCL_EVENT_SEND_COMPLETION_MSG);
69 0 : }
70 :
71 0 : void TransportHeterogEventRoce::EschedAckCallbackRecvCompletion(unsigned int devId, unsigned int subeventId, u8 *msg,
72 : unsigned int msgLen)
73 : {
74 : (void)subeventId;
75 : (void)msg;
76 : (void)msgLen;
77 0 : TransportHeterogEventRoce::EschedAckCallback(devId, HCCL_EVENT_RECV_COMPLETION_MSG);
78 0 : }
79 :
80 0 : TransportHeterogEventRoce::TransportHeterogEventRoce(const std::string &transTag, HcclIpAddress &selfIp,
81 0 : HcclIpAddress &peerIp, u32 peerPort, u32 selfPort, const TransportResourceInfo &transportResourceInfo)
82 0 : : TransportHeterogRoce(transTag, selfIp, peerIp, peerPort, selfPort, transportResourceInfo)
83 : {
84 0 : tagQpInfo_.srq = transportResourceInfo.tagSrqInfo.srq;
85 0 : tagQpInfo_.srqCq = transportResourceInfo.tagSrqInfo.srqCq;
86 0 : tagQpInfo_.srqContext = transportResourceInfo.tagSrqInfo.context;
87 0 : dataQpInfo_.srq = transportResourceInfo.dataSrqInfo.srq;
88 0 : dataQpInfo_.srqCq = transportResourceInfo.dataSrqInfo.srqCq;
89 0 : dataQpInfo_.srqContext = transportResourceInfo.dataSrqInfo.context;
90 0 : srqInit_ = ((tagQpInfo_.srq != nullptr) && (dataQpInfo_.srq != nullptr));
91 0 : }
92 0 : TransportHeterogEventRoce::TransportHeterogEventRoce(const TransportResourceInfo &transportResourceInfo)
93 0 : : TransportHeterogRoce(transportResourceInfo)
94 : {
95 0 : tagQpInfo_.srq = transportResourceInfo.tagSrqInfo.srq;
96 0 : tagQpInfo_.srqCq = transportResourceInfo.tagSrqInfo.srqCq;
97 0 : tagQpInfo_.srqContext = transportResourceInfo.tagSrqInfo.context;
98 0 : dataQpInfo_.srq = transportResourceInfo.dataSrqInfo.srq;
99 0 : dataQpInfo_.srqCq = transportResourceInfo.dataSrqInfo.srqCq;
100 0 : dataQpInfo_.srqContext = transportResourceInfo.dataSrqInfo.context;
101 0 : srqInit_ = ((tagQpInfo_.srq != nullptr) && (dataQpInfo_.srq != nullptr));
102 0 : }
103 :
104 0 : TransportHeterogEventRoce::~TransportHeterogEventRoce()
105 : {
106 0 : HcclResult ret = Deinit();
107 0 : if (ret != HCCL_SUCCESS) {
108 0 : HCCL_ERROR("TransportHeterogEventRoce:: destructor Deinit fail.");
109 : }
110 0 : }
111 :
112 0 : HcclResult TransportHeterogEventRoce::RegisterEschedAckCallback()
113 : {
114 0 : if (gEschedAckRef == 0) {
115 0 : CHK_RET(DlHalFunction::GetInstance().DlHalFunctionInit());
116 :
117 0 : recvRequestEvent = HCCL_EVENT_RECV_REQUEST_MSG;
118 0 : sendCompletionEvent = HCCL_EVENT_SEND_COMPLETION_MSG;
119 0 : recvCompletionEvent = HCCL_EVENT_RECV_COMPLETION_MSG;
120 :
121 0 : CHK_RET(hrtHalEschedRegisterAckFunc(HCCL_EVENT_RECV_REQUEST_MSG, EschedAckCallbackRecvRequest));
122 0 : CHK_RET(hrtHalEschedRegisterAckFunc(HCCL_EVENT_SEND_COMPLETION_MSG, EschedAckCallbackSendCompletion));
123 0 : CHK_RET(hrtHalEschedRegisterAckFunc(HCCL_EVENT_RECV_COMPLETION_MSG, EschedAckCallbackRecvCompletion));
124 :
125 0 : gCqeCounterPerEvent[HCCL_EVENT_RECV_REQUEST_MSG] = 0;
126 0 : gCqeCounterPerEvent[HCCL_EVENT_SEND_COMPLETION_MSG] = 0;
127 0 : gCqeCounterPerEvent[HCCL_EVENT_RECV_COMPLETION_MSG] = 0;
128 : }
129 0 : gEschedAckRef++;
130 :
131 0 : return HCCL_SUCCESS;
132 : }
133 :
134 0 : HcclResult TransportHeterogEventRoce::DeregisterEschedAckCallback()
135 : {
136 0 : if (gEschedAckRef > 0) {
137 0 : gEschedAckRef--;
138 0 : } else if (gEschedAckRef == 0) {
139 0 : HCCL_WARNING("TransportHeterogEventRoce:: EschedAckCallback has been deregistered.");
140 0 : return HCCL_SUCCESS;
141 : }
142 0 : return HCCL_SUCCESS;
143 : }
144 :
145 0 : HcclResult TransportHeterogEventRoce::InitAllLinkVec()
146 : {
147 0 : gAllLinkInitCount++;
148 0 : std::unique_lock<std::mutex> lockRecvReq(gAllLinkVecRecvReqMutex);
149 0 : gAllLinkVec[HCCL_EVENT_RECV_REQUEST_MSG].push_back(this);
150 0 : lockRecvReq.unlock();
151 :
152 0 : std::unique_lock<std::mutex> lockSendComp(gAllLinkVecSendCompMutex);
153 0 : gAllLinkVec[HCCL_EVENT_SEND_COMPLETION_MSG].push_back(this);
154 0 : lockSendComp.unlock();
155 :
156 0 : std::unique_lock<std::mutex> lockRecvComp(gAllLinkVecRecvCompMutex);
157 0 : gAllLinkVec[HCCL_EVENT_RECV_COMPLETION_MSG].push_back(this);
158 0 : lockRecvComp.unlock();
159 :
160 0 : return HCCL_SUCCESS;
161 0 : }
162 :
163 0 : HcclResult TransportHeterogEventRoce::DeinitAllLinkVec()
164 : {
165 0 : if (gAllLinkInitCount > 0) {
166 0 : std::unique_lock<std::mutex> lockRecvReq(gAllLinkVecRecvReqMutex);
167 0 : gAllLinkVec[HCCL_EVENT_RECV_REQUEST_MSG].clear();
168 0 : lockRecvReq.unlock();
169 :
170 0 : std::unique_lock<std::mutex> lockSendComp(gAllLinkVecSendCompMutex);
171 0 : gAllLinkVec[HCCL_EVENT_SEND_COMPLETION_MSG].clear();
172 0 : lockSendComp.unlock();
173 :
174 0 : std::unique_lock<std::mutex> lockRecvComp(gAllLinkVecRecvCompMutex);
175 0 : gAllLinkVec[HCCL_EVENT_RECV_COMPLETION_MSG].clear();
176 0 : lockRecvComp.unlock();
177 0 : }
178 :
179 0 : return HCCL_SUCCESS;
180 : }
181 :
182 0 : HcclResult TransportHeterogEventRoce::EraseTransportFromAllLinkVec(void *transportPtr)
183 : {
184 0 : std::unique_lock<std::mutex> lockRecvReq(gAllLinkVecRecvReqMutex);
185 0 : auto recvReqIter = find(gAllLinkVec[HCCL_EVENT_RECV_REQUEST_MSG].begin(),
186 0 : gAllLinkVec[HCCL_EVENT_RECV_REQUEST_MSG].end(), transportPtr);
187 0 : if (recvReqIter != gAllLinkVec[HCCL_EVENT_RECV_REQUEST_MSG].end()) {
188 0 : gAllLinkVec[HCCL_EVENT_RECV_REQUEST_MSG].erase(recvReqIter);
189 : }
190 0 : lockRecvReq.unlock();
191 :
192 0 : std::unique_lock<std::mutex> lockSendComp(gAllLinkVecSendCompMutex);
193 0 : auto sendCompIter = find(gAllLinkVec[HCCL_EVENT_SEND_COMPLETION_MSG].begin(),
194 0 : gAllLinkVec[HCCL_EVENT_SEND_COMPLETION_MSG].end(), transportPtr);
195 0 : if (sendCompIter != gAllLinkVec[HCCL_EVENT_SEND_COMPLETION_MSG].end()) {
196 0 : gAllLinkVec[HCCL_EVENT_SEND_COMPLETION_MSG].erase(sendCompIter);
197 : }
198 0 : lockSendComp.unlock();
199 :
200 0 : std::unique_lock<std::mutex> lockRecvComp(gAllLinkVecRecvCompMutex);
201 0 : auto recvCompIter = find(gAllLinkVec[HCCL_EVENT_RECV_COMPLETION_MSG].begin(),
202 0 : gAllLinkVec[HCCL_EVENT_RECV_COMPLETION_MSG].end(), transportPtr);
203 0 : if (recvCompIter != gAllLinkVec[HCCL_EVENT_RECV_COMPLETION_MSG].end()) {
204 0 : gAllLinkVec[HCCL_EVENT_RECV_COMPLETION_MSG].erase(recvCompIter);
205 : }
206 0 : lockRecvComp.unlock();
207 :
208 0 : return HCCL_SUCCESS;
209 0 : }
210 :
211 0 : HcclResult TransportHeterogEventRoce::Init()
212 : {
213 0 : CHK_RET(DlHalFunction::GetInstance().DlHalFunctionInit());
214 0 : CHK_RET(RegisterEschedAckCallback());
215 0 : CHK_RET(TransportHeterogRoce::Init());
216 :
217 0 : CHK_RET(InitAllLinkVec());
218 0 : isDeinited_ = false;
219 0 : return HCCL_SUCCESS;
220 : }
221 :
222 0 : HcclResult TransportHeterogEventRoce::Deinit()
223 : {
224 0 : if (isDeinited_) {
225 0 : return HCCL_SUCCESS;
226 : }
227 0 : CHK_RET(DeregisterEschedAckCallback());
228 0 : CHK_RET(DeinitAllLinkVec());
229 :
230 0 : CHK_RET(TransportHeterogRoce::Deinit());
231 0 : isDeinited_ = true;
232 0 : return HCCL_SUCCESS;
233 : }
234 :
235 0 : HcclResult TransportHeterogEventRoce::Isend(const TransData &sendData, const TransportEndPointParam &epParam,
236 : HcclRequestInfo *&request)
237 : {
238 0 : HcclResult ret = TransportHeterogRoce::Isend(sendData, epParam, request);
239 0 : if (ret != HCCL_SUCCESS && request != nullptr) {
240 0 : CHK_RET(FreeRequest(*request));
241 : }
242 0 : return ret;
243 : }
244 :
245 0 : HcclResult TransportHeterogEventRoce::Improbe(const TransportEndPointParam &epParam, s32 &matched,
246 : HcclMessageInfo *&msg, HcclStatus &status)
247 : {
248 0 : CHK_RET(TransportHeterogRoce::Improbe(epParam, matched, msg, status));
249 0 : if (matched == HCCL_IMPROBE_COMPLETED) {
250 0 : gCqeCounterPerEvent[recvRequestEvent].fetch_sub(1);
251 : }
252 0 : return HCCL_SUCCESS;
253 : }
254 :
255 0 : HcclResult TransportHeterogEventRoce::Imrecv(const TransData &recvData, HcclMessageInfo &msg, HcclRequestInfo *&request)
256 : {
257 0 : CHK_RET(TransportHeterogRoce::Imrecv(recvData, msg, request));
258 0 : return HCCL_SUCCESS;
259 : }
260 :
261 0 : HcclResult TransportHeterogEventRoce::Test(HcclRequestInfo &request, s32 &flag, HcclStatus &compState)
262 : {
263 : // 建链未完成时，继续推进建链流程；
264 0 : if (GetState() != ConnState::CONN_STATE_COMPLETE) {
265 0 : CHK_RET(ConnectAsync());
266 : }
267 :
268 0 : if (GetState() == ConnState::CONN_STATE_COMPLETE || GetState() == ConnState::CONN_STATE_FLUSH_QUEUE) {
269 0 : if (request.transportRequest.requestType == HcclRequestType::HCCL_REQUEST_SEND) {
270 0 : CHK_RET(PullSendStatus());
271 0 : } else if (request.transportRequest.requestType == HcclRequestType::HCCL_REQUEST_RECV) {
272 0 : CHK_RET(PullRecvStatus());
273 : } else {
274 0 : HCCL_ERROR("[HcclTest] requestType[%u] is invalid", request.transportRequest.requestType);
275 0 : return HCCL_E_PARA;
276 : }
277 : }
278 :
279 0 : return QueryRequestStatus(request, flag, compState);
280 : }
281 :
282 0 : HcclResult TransportHeterogEventRoce::PullRecvRequestStatus(bool allowNotify)
283 : {
284 0 : std::unique_lock<std::mutex> lock(gPollTagRqLock, std::defer_lock);
285 0 : if (lock.try_lock()) {
286 0 : CHK_RET(TransportHeterogRoce::PullRecvRequestStatus(allowNotify));
287 : } else {
288 0 : if ((allowNotify) && (gCqeCounterPerEvent[recvRequestEvent] <= 0)) {
289 0 : CHK_RET(hrtIbvReqNotifyCq(tagQpInfo_.recvCq, 0));
290 : }
291 : }
292 0 : return HCCL_SUCCESS;
293 0 : }
294 :
295 0 : HcclResult TransportHeterogEventRoce::PullSendStatus(bool allowNotify)
296 : {
297 0 : std::unique_lock<std::mutex> lock(gPollDataRqLock, std::defer_lock);
298 0 : if (lock.try_lock()) {
299 0 : CHK_RET(TransportHeterogRoce::PullSendStatus(allowNotify));
300 : } else {
301 0 : if ((allowNotify) && (gCqeCounterPerEvent[sendCompletionEvent] <= 0)) {
302 0 : CHK_RET(hrtIbvReqNotifyCq(dataQpInfo_.recvCq, 0));
303 : }
304 : }
305 :
306 0 : return HCCL_SUCCESS;
307 0 : }
308 :
309 0 : HcclResult TransportHeterogEventRoce::PullRecvStatus(bool allowNotify)
310 : {
311 0 : std::unique_lock<std::mutex> lock(gPollDataSqLock, std::defer_lock);
312 0 : if (lock.try_lock()) {
313 0 : CHK_RET(TransportHeterogRoce::PullRecvStatus(allowNotify));
314 : } else {
315 0 : if ((allowNotify) && (gCqeCounterPerEvent[recvCompletionEvent] <= 0)) {
316 0 : CHK_RET(hrtIbvReqNotifyCq(dataQpInfo_.sendCq, 0));
317 : }
318 : }
319 :
320 0 : return HCCL_SUCCESS;
321 0 : }
322 :
323 0 : HcclResult TransportHeterogEventRoce::ParseErrorTagSqe(const struct ibv_wc *wc, int index)
324 : {
325 0 : CHK_RET(TransportHeterogRoce::ParseErrorTagSqe(wc, index));
326 0 : gCqeCounterPerEvent[sendCompletionEvent].fetch_add(1);
327 0 : return HCCL_SUCCESS;
328 : }
329 :
330 0 : HcclResult TransportHeterogEventRoce::ParseTagRqes(const struct ibv_wc *wc, int num)
331 : {
332 0 : if (srqInit_) {
333 0 : CHK_RET(ParseTagSrqes(wc, num));
334 : } else {
335 0 : CHK_RET(TransportHeterogRoce::ParseTagRqes(wc, num));
336 : }
337 :
338 0 : gCqeCounterPerEvent[recvRequestEvent].fetch_add(num);
339 0 : return HCCL_SUCCESS;
340 : }
341 :
342 0 : HcclResult TransportHeterogEventRoce::ParseDataRqes(const struct ibv_wc *wc, int num)
343 : {
344 0 : if (srqInit_) {
345 0 : CHK_RET(ParseDataSrqes(wc, num));
346 : } else {
347 0 : CHK_RET(TransportHeterogRoce::ParseDataRqes(wc, num));
348 : }
349 0 : gCqeCounterPerEvent[sendCompletionEvent].fetch_add(num);
350 0 : return HCCL_SUCCESS;
351 : }
352 :
353 0 : HcclResult TransportHeterogEventRoce::ParseDataSqes(const struct ibv_wc *wc, int num)
354 : {
355 0 : CHK_RET(TransportHeterogRoce::ParseDataSqes(wc, num));
356 0 : gCqeCounterPerEvent[recvCompletionEvent].fetch_add(num);
357 0 : return HCCL_SUCCESS;
358 : }
359 :
360 0 : HcclResult TransportHeterogEventRoce::QueryRequestStatus(HcclRequestInfo &request, s32 &flag, HcclStatus &compState)
361 : {
362 0 : flag = HCCL_TEST_INCOMPLETED;
363 : u32 eventType;
364 0 : if (request.transportRequest.requestType == HcclRequestType::HCCL_REQUEST_SEND) {
365 0 : eventType = sendCompletionEvent;
366 0 : } else if (request.transportRequest.requestType == HcclRequestType::HCCL_REQUEST_RECV) {
367 0 : eventType = recvCompletionEvent;
368 : } else {
369 0 : HCCL_ERROR("[QueryRequestStatus]requestType is invalid! requestType[%u]", request.transportRequest.requestType);
370 0 : return HCCL_E_PARA;
371 : }
372 :
373 0 : if (gCqeCounterPerEvent[eventType] > 0) {
374 0 : CHK_RET(TransportHeterogRoce::QueryRequestStatus(request, flag, compState));
375 0 : if (flag == HCCL_TEST_COMPLETED) {
376 0 : gCqeCounterPerEvent[eventType].fetch_sub(1);
377 : }
378 : }
379 0 : return HCCL_SUCCESS;
380 : }
381 :
382 0 : HcclResult TransportHeterogEventRoce::PullRecvRequestStatus(void *transportHandle)
383 : {
384 0 : CHK_PTR_NULL(transportHandle);
385 0 : TransportHeterogEventRoce *transportPtr = reinterpret_cast<TransportHeterogEventRoce *>(transportHandle);
386 0 : if (transportPtr->GetState() == ConnState::CONN_STATE_COMPLETE) {
387 0 : CHK_RET(transportPtr->PullRecvRequestStatus(true));
388 : }
389 0 : return HCCL_SUCCESS;
390 : }
391 :
392 0 : HcclResult TransportHeterogEventRoce::PullSendStatus(void *transportHandle)
393 : {
394 0 : CHK_PTR_NULL(transportHandle);
395 0 : TransportHeterogEventRoce *transportPtr = reinterpret_cast<TransportHeterogEventRoce *>(transportHandle);
396 0 : if (transportPtr->GetState() == ConnState::CONN_STATE_COMPLETE ||
397 0 : transportPtr->GetState() == ConnState::CONN_STATE_FLUSH_QUEUE) {
398 0 : CHK_RET(transportPtr->PullSendStatus(true));
399 : }
400 0 : return HCCL_SUCCESS;
401 : }
402 :
403 0 : HcclResult TransportHeterogEventRoce::PullRecvStatus(void *transportHandle)
404 : {
405 0 : CHK_PTR_NULL(transportHandle);
406 0 : TransportHeterogEventRoce *transportPtr = reinterpret_cast<TransportHeterogEventRoce *>(transportHandle);
407 0 : if (transportPtr->GetState() == ConnState::CONN_STATE_COMPLETE) {
408 0 : CHK_RET(transportPtr->PullRecvStatus(true));
409 : }
410 0 : return HCCL_SUCCESS;
411 : }
412 :
413 0 : HcclResult TransportHeterogEventRoce::UpdateStatus(u32 eventId)
414 : {
415 0 : if (gCqeCounterPerEvent[eventId] > 0) {
416 0 : return HCCL_SUCCESS;
417 : }
418 : // 检查所有tag的对应cq中有没有cqe，如果没有则继续轮询下一个cq，如果检查到某一cq中存在cqe则退出循环。
419 0 : if (eventId == HCCL_EVENT_RECV_REQUEST_MSG) {
420 0 : std::unique_lock<std::mutex> lockRecvReq(gAllLinkVecRecvReqMutex);
421 0 : for (auto &iterLink : gAllLinkVec[HCCL_EVENT_RECV_REQUEST_MSG]) {
422 0 : CHK_RET(PullRecvRequestStatus(iterLink));
423 0 : if ((gCqeCounterPerEvent[HCCL_EVENT_RECV_REQUEST_MSG] > 0)) {
424 0 : return HCCL_SUCCESS;
425 : }
426 : }
427 0 : lockRecvReq.unlock();
428 0 : } else if (eventId == HCCL_EVENT_SEND_COMPLETION_MSG) {
429 0 : std::unique_lock<std::mutex> lockSendComp(gAllLinkVecSendCompMutex);
430 0 : for (auto &iterLink : gAllLinkVec[HCCL_EVENT_SEND_COMPLETION_MSG]) {
431 0 : CHK_RET(PullSendStatus(iterLink));
432 0 : if ((gCqeCounterPerEvent[HCCL_EVENT_SEND_COMPLETION_MSG] > 0)) {
433 0 : return HCCL_SUCCESS;
434 : }
435 : }
436 0 : lockSendComp.unlock();
437 0 : } else if (eventId == HCCL_EVENT_RECV_COMPLETION_MSG) {
438 0 : std::unique_lock<std::mutex> lockRecvComp(gAllLinkVecRecvCompMutex);
439 0 : for (auto &iterLink : gAllLinkVec[HCCL_EVENT_RECV_COMPLETION_MSG]) {
440 0 : CHK_RET(PullRecvStatus(iterLink));
441 0 : if ((gCqeCounterPerEvent[HCCL_EVENT_RECV_COMPLETION_MSG] > 0)) {
442 0 : return HCCL_SUCCESS;
443 : }
444 : }
445 0 : lockRecvComp.unlock();
446 0 : }
447 0 : return HCCL_SUCCESS;
448 : }
449 :
450 0 : void TransportHeterogEventRoce::EschedAckCallback(u32 devId, u32 eventId)
451 : {
452 0 : if (!gNeedRepoEvent) {
453 0 : HCCL_DEBUG("TransportHeterogEventRoce no need submit event.");
454 0 : return;
455 : }
456 :
457 0 : HCCL_DEBUG("EventCallback start. devId:%u, eventId:%u.", devId, eventId);
458 0 : HcclUs startut = TIME_NOW();
459 :
460 0 : if (UpdateStatus(eventId) != HCCL_SUCCESS) {
461 0 : HCCL_ERROR("poll all cqes failed. event id:%u", eventId);
462 0 : return;
463 : }
464 0 : if (gCqeCounterPerEvent[eventId] != 0) {
465 0 : hrtHalSubmitEvent(devId, eventId);
466 : }
467 :
468 0 : HcclUs endut = TIME_NOW();
469 0 : HCCL_INFO("EschedAckCallback cost time: %lld us, event id: %u, devId:%u, compCount:%d.",
470 : DURATION_US(endut - startut), eventId, devId, gCqeCounterPerEvent[eventId].load());
471 0 : return;
472 : }
473 :
474 0 : HcclResult TransportHeterogEventRoce::CreateCqAndQp()
475 : {
476 0 : HCCL_INFO("TransportHeterogEventRoce CreateCqAndQp. gNeedRepoEvent[%d]", gNeedRepoEvent);
477 0 : if (gNeedRepoEvent) {
478 0 : CHK_RET(CreateQpWithSharedCq(nicRdmaHandle_, selfIp_, peerIp_, -1, recvRequestEvent, tagQpInfo_));
479 0 : CHK_RET(CreateQpWithSharedCq(nicRdmaHandle_, selfIp_, peerIp_, recvCompletionEvent,
480 : sendCompletionEvent, dataQpInfo_));
481 : } else {
482 0 : tagQpAppend_ = TAG_QP_APPEND;
483 0 : dataQpAppend_ = DATA_QP_APPEND;
484 0 : CHK_RET(CreateQpWithSharedCq(nicRdmaHandle_, selfIp_, peerIp_, -1, -1,
485 : tagQpInfo_, tagQpAppend_, MAX_SCATTER_BUF_NUM));
486 0 : CHK_RET(CreateQpWithSharedCq(nicRdmaHandle_, selfIp_, peerIp_, -1, -1,
487 : dataQpInfo_, dataQpAppend_, MAX_SCATTER_BUF_NUM));
488 : }
489 :
490 0 : if (srqInit_) {
491 0 : gQpnToTransportMap[tagQpInfo_.qp->qp_num] = this;
492 0 : gQpnToSqMaxWrMap[dataQpInfo_.qp->qp_num] = MAX_WR_NUM;
493 : }
494 :
495 0 : return HCCL_SUCCESS;
496 : }
497 :
498 0 : HcclResult TransportHeterogEventRoce::DestroyCqAndQp()
499 : {
500 0 : HCCL_INFO("TransportHeterogEventRoce DestroyCqAndQp.");
501 0 : CHK_RET(DestroyQpWithSharedCq(tagQpInfo_, tagQpAppend_));
502 0 : tagQpInfo_ = QpInfo();
503 0 : CHK_RET(DestroyQpWithSharedCq(dataQpInfo_, dataQpAppend_));
504 0 : dataQpInfo_ = QpInfo();
505 0 : return HCCL_SUCCESS;
506 : }
507 :
508 0 : HcclResult TransportHeterogEventRoce::InitSrqRecvWqe()
509 : {
510 0 : CHK_RET(IssueRecvWqe(tagQpInfo_.srq, RECV_WQE_BATCH_NUM));
511 0 : g_tagRecvWqeNum = RECV_WQE_BATCH_NUM;
512 :
513 0 : CHK_RET(IssueRecvWqe(dataQpInfo_.srq, RECV_WQE_BATCH_NUM));
514 0 : g_dataRecvWqeNum = RECV_WQE_BATCH_NUM;
515 0 : HCCL_INFO("InitSrqRecvWqe success.");
516 0 : return HCCL_SUCCESS;
517 : }
518 :
519 0 : HcclResult TransportHeterogEventRoce::InitTagRecvWqe()
520 : {
521 0 : if (!srqInit_) {
522 0 : CHK_RET(TransportHeterogRoce::InitTagRecvWqe());
523 0 : return HCCL_SUCCESS;
524 : }
525 :
526 0 : CHK_RET(CheckTagRecvWqe());
527 0 : return HCCL_SUCCESS;
528 : }
529 :
530 0 : HcclResult TransportHeterogEventRoce::InitDataRecvWqe()
531 : {
532 0 : if (!srqInit_) {
533 0 : CHK_RET(TransportHeterogRoce::InitDataRecvWqe());
534 0 : return HCCL_SUCCESS;
535 : }
536 0 : CHK_RET(CheckDataRecvWqe());
537 0 : return HCCL_SUCCESS;
538 : }
539 :
540 0 : HcclResult TransportHeterogEventRoce::SendFlowControl()
541 : {
542 0 : if (!srqInit_) {
543 0 : CHK_RET(TransportHeterogRoce::SendFlowControl());
544 0 : return HCCL_SUCCESS;
545 : }
546 :
547 0 : CHK_RET(CheckDataRecvWqe());
548 0 : u32 sqMaxWrMap = gQpnToSqMaxWrMap[dataQpInfo_.qp->qp_num].load();
549 0 : if (sqMaxWrMap <= 0) {
550 0 : CHK_RET(PullSendStatus());
551 0 : HCCL_RUN_INFO("Flow control is activated, because sqMaxWrMap[%u] <= 0", sqMaxWrMap);
552 :
553 0 : return HCCL_E_AGAIN;
554 : }
555 0 : gQpnToSqMaxWrMap[dataQpInfo_.qp->qp_num]--;
556 0 : return HCCL_SUCCESS;
557 : }
558 :
559 0 : HcclResult TransportHeterogEventRoce::CheckTagRecvWqe()
560 : {
561 0 : if (g_tagRecvWqeNum <= RECV_WQE_NUM_THRESHOLD) {
562 0 : CHK_RET(IssueRecvWqe(tagQpInfo_.srq, RECV_WQE_BATCH_SUPPLEMENT));
563 0 : g_tagRecvWqeNum += RECV_WQE_BATCH_SUPPLEMENT;
564 : }
565 :
566 0 : return HCCL_SUCCESS;
567 : }
568 :
569 0 : HcclResult TransportHeterogEventRoce::SupplyTagRecvWqe()
570 : {
571 0 : if (!srqInit_) {
572 0 : CHK_RET(TransportHeterogRoce::SupplyTagRecvWqe());
573 0 : return HCCL_SUCCESS;
574 : }
575 :
576 0 : g_tagRecvWqeNum--;
577 0 : CHK_RET(CheckTagRecvWqe());
578 :
579 0 : return HCCL_SUCCESS;
580 : }
581 :
582 0 : HcclResult TransportHeterogEventRoce::CheckDataRecvWqe()
583 : {
584 0 : if (g_dataRecvWqeNum <= RECV_WQE_NUM_THRESHOLD) {
585 0 : CHK_RET(IssueRecvWqe(dataQpInfo_.srq, RECV_WQE_BATCH_SUPPLEMENT));
586 0 : g_dataRecvWqeNum += RECV_WQE_BATCH_SUPPLEMENT;
587 : }
588 0 : return HCCL_SUCCESS;
589 : }
590 :
591 0 : HcclResult TransportHeterogEventRoce::SupplyDataRecvWqe()
592 : {
593 0 : if (!srqInit_) {
594 0 : CHK_RET(TransportHeterogRoce::SupplyDataRecvWqe());
595 0 : return HCCL_SUCCESS;
596 : }
597 :
598 0 : g_dataRecvWqeNum--;
599 0 : CHK_RET(CheckDataRecvWqe());
600 :
601 0 : return HCCL_SUCCESS;
602 : }
603 :
604 0 : HcclResult TransportHeterogEventRoce::IssueRecvWqe(struct ibv_srq *srq, u32 num)
605 : {
606 0 : list<void *> blockList(num, nullptr);
607 0 : CHK_RET(AllocMemBlocks(blockList));
608 :
609 0 : auto iter = blockList.begin();
610 0 : struct ibv_recv_wr *nextRqWr = nullptr;
611 0 : struct ibv_recv_wr rqWr[num];
612 0 : struct ibv_sge sgeList[num];
613 0 : for (int i = num - 1; i >= 0; i--) {
614 0 : CHK_PTR_NULL(*iter);
615 0 : u64 wrId = 0;
616 0 : CHK_RET(GenerateRecvWrId(*iter, wrId));
617 :
618 0 : rqWr[i].wr_id = wrId;
619 0 : rqWr[i].next = nextRqWr;
620 0 : rqWr[i].sg_list = &sgeList[i];
621 0 : rqWr[i].num_sge = 1;
622 0 : sgeList[i].addr = reinterpret_cast<uint64_t>(*iter);
623 0 : sgeList[i].length = MEM_BLOCK_SIZE;
624 0 : sgeList[i].lkey = blockMemLkey_;
625 :
626 0 : nextRqWr = &rqWr[i];
627 0 : iter++;
628 : }
629 :
630 0 : struct ibv_recv_wr *badRqWr = nullptr;
631 0 : CHK_RET(hrtIbvPostSrqRecv(srq, &rqWr[0], &badRqWr));
632 0 : return HCCL_SUCCESS;
633 0 : }
634 :
635 0 : HcclResult TransportHeterogEventRoce::ParseTagSrqes(const struct ibv_wc *wc, int num)
636 : {
637 0 : for (int i = 0; i < num; i++) {
638 0 : HCCL_INFO("rq cqe info: wrId[%llu] status[%u] opcode[%u] qpn[%u].", wc[i].wr_id, wc[i].status,
639 : wc[i].opcode, wc[i].qp_num);
640 0 : CHK_PRT_RET(wc[i].status != 0, HCCL_ERROR("rdma send failed, cqe status[%u].", wc[i].status), HCCL_E_INTERNAL);
641 0 : RecvWrInfo *info = reinterpret_cast<RecvWrInfo *>(wc[i].wr_id);
642 0 : CHK_PTR_NULL(info);
643 :
644 0 : CHK_RET(SupplyTagRecvWqe());
645 :
646 0 : HcclEnvelope *envelope = reinterpret_cast<HcclEnvelope *>(info->buf);
647 0 : CHK_PTR_NULL(envelope);
648 :
649 0 : HCCL_INFO("recv request: tag:%d srcRank:%u dstRank:%u status:%u msn:0x%016llx count:%d.",
650 : envelope->epParam.src.tag, envelope->epParam.src.rank, envelope->epParam.dst.rank, wc[i].status,
651 : envelope->msn, envelope->transData.count);
652 :
653 0 : HcclEnvelopeSummary envelopSummary(*envelope, wc[i].status);
654 0 : if (gQpnToTransportMap.count(wc[i].qp_num) != 0) {
655 0 : gQpnToTransportMap[wc[i].qp_num]->SaveEnvelope(envelopSummary);
656 : } else {
657 0 : HCCL_ERROR("The transport is no exist, wrId[%llu] status[%u] opcode[%u] qpn[%u]", wc[i].wr_id,
658 : wc[i].status, wc[i].opcode, wc[i].qp_num);
659 0 : return HCCL_E_PTR;
660 : }
661 :
662 0 : CHK_RET(FreeMemBlock(info->buf));
663 0 : CHK_RET(FreeRecvWrId(wc[i].wr_id));
664 : }
665 0 : return HCCL_SUCCESS;
666 : }
667 :
668 0 : HcclResult TransportHeterogEventRoce::ParseDataSrqes(const struct ibv_wc *wc, int num)
669 : {
670 0 : for (int i = 0; i < num; i++) {
671 0 : HCCL_INFO("rq cqe info: wrId[%llu] status[%u] opcode[%u].", wc[i].wr_id, wc[i].status, wc[i].opcode);
672 0 : CHK_PRT_RET(wc[i].status != 0, HCCL_ERROR("rdma poll data rq failed, cqe status[%u].",
673 : wc[i].status), HCCL_E_INTERNAL);
674 0 : RecvWrInfo *info = reinterpret_cast<RecvWrInfo *>(wc[i].wr_id);
675 0 : CHK_PTR_NULL(info);
676 0 : HcclRequestInfo *wrPtr = reinterpret_cast<HcclRequestInfo *>(*reinterpret_cast<u64 *>(info->buf));
677 0 : CHK_PTR_NULL(wrPtr);
678 0 : CHK_RET(SupplyDataRecvWqe());
679 0 : wrPtr->transportRequest.status = wc[i].status;
680 0 : if (gQpnToSqMaxWrMap.count(wc[i].qp_num) != 0) {
681 0 : gQpnToSqMaxWrMap[wc[i].qp_num]++;
682 : } else {
683 0 : HCCL_ERROR("The qpn is no exist, wrId[%llu] status[%u] opcode[%u] qpn[%u]", wc[i].wr_id,
684 : wc[i].status, wc[i].opcode, wc[i].qp_num);
685 0 : return HCCL_E_PTR;
686 : }
687 :
688 0 : CHK_RET(DeregMr(reinterpret_cast<void *>(wrPtr->transportRequest.transData.srcBuf),
689 : static_cast<u64>(wrPtr->transportRequest.transData.count *
690 : SIZE_TABLE[wrPtr->transportRequest.transData.dataType])));
691 0 : CHK_RET(FreeMemBlock(info->buf));
692 0 : CHK_RET(FreeRecvWrId(wc[i].wr_id));
693 0 : HCCL_INFO("send completion: tag:%d peerRank:%u status:%d msn:0x%016llx request:%p.",
694 : wrPtr->transportRequest.epParam.src.tag, wrPtr->transportRequest.epParam.src.rank,
695 : wrPtr->transportRequest.status, wrPtr->transportRequest.msn, wrPtr);
696 : }
697 0 : return HCCL_SUCCESS;
698 : }
699 : } // namespace hccl
|
