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