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