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_roce_mem.h"
12 : #include "log.h"
13 : #include "adapter_hal.h"
14 : #include "adapter_hccp.h"
15 : #include "adapter_rts.h"
16 : #include "network_manager_pub.h"
17 : #include "dispatcher_pub.h"
18 : #include "hccl_network.h"
19 : #include "device_capacity.h"
20 : #include "externalinput.h"
21 :
22 : namespace hccl {
23 : using namespace std;
24 : using LocalRdmaRmaBufferMgr = NetDevContext::LocalRdmaRmaBufferMgr;
25 :
26 : constexpr s32 REG_VALID = 1;
27 : constexpr u32 WAIT_LINK_BUILD_DELAY_TIME_US = 10;
28 : constexpr s32 QP_FLAG_RC = 0; // flag: 0 = RC, 1= UD，其它预留
29 : constexpr s32 OPBASE_QP_MODE_EXT = 4; // 单算子模式(910B/910_93)的QP
30 : constexpr u32 WR_NUM = 1; // 当前只支持一个WR
31 : std::atomic<uint64_t> TransportRoceMem::sendWrHandle{0};
32 0 : TransportRoceMem::TransportRoceMem(
33 : const std::unique_ptr<NotifyPool>& notifyPool, const HcclNetDevCtx& netDevCtx, const HcclDispatcher& dispatcher,
34 0 : AttrInfo& attrInfo, bool aicpuUnfoldMode)
35 : : TransportMem(notifyPool, netDevCtx, dispatcher, attrInfo, aicpuUnfoldMode),
36 0 : trafficClass_(attrInfo.trafficClass),
37 0 : serviceLevel_(attrInfo.serviceLevel)
38 0 : {}
39 :
40 0 : TransportRoceMem::~TransportRoceMem()
41 : {
42 : // de rdmaSignal and Mr
43 0 : if (rdmaSignalMrHandle_ != nullptr) {
44 0 : HcclResult ret = HCCL_SUCCESS;
45 0 : ret = hrtRaDeRegGlobalMr(nicRdmaHandle_, rdmaSignalMrHandle_);
46 0 : rdmaSignalMrHandle_ = nullptr;
47 0 : if (ret != 0) {
48 0 : HCCL_ERROR("deReg rdmaSignal GlobalMr failed, ret[%d]", ret);
49 : }
50 : }
51 : // destroy notify mem and notifyMem Mr
52 0 : if (notifyValueMemMrHandle_ != nullptr) {
53 0 : HcclResult ret = HCCL_SUCCESS;
54 0 : ret = hrtRaDeRegGlobalMr(nicRdmaHandle_, notifyValueMemMrHandle_);
55 0 : notifyValueMemMrHandle_ = nullptr;
56 0 : if (ret != 0) {
57 0 : HCCL_ERROR("deReg notify Mem Mr failed, ret[%d]", ret);
58 : }
59 : }
60 0 : if (notifyMem_.ptr() != nullptr) {
61 0 : notifyMem_.free();
62 : }
63 : // destroy QP
64 0 : DestroyCqAndQp();
65 0 : }
66 :
67 0 : HcclResult TransportRoceMem::CheckRaSendNormalWrlistSupport()
68 : {
69 0 : if ((LIKELY(isSupportRaSendNormalWrlist_ == SupportStatus::SUPPORT))) {
70 : // 已判断支持，直接返回成功，避免重复判断
71 0 : return HCCL_SUCCESS;
72 0 : } else if (isSupportRaSendNormalWrlist_ == SupportStatus::NOT_SUPPORT) {
73 0 : HCCL_ERROR("[TransportRoceMem]RDMALite and RaSendNormalWrlist are not supported");
74 0 : return HCCL_E_NOT_SUPPORT;
75 : } else {
76 : // 判断是否支持
77 0 : bool isSupportRDMALite = IsSupportRDMALite(deviceLogicId_);
78 0 : if (isSupportRDMALite) {
79 : // 支持RDMALite场景可直接支持
80 0 : isSupportRaSendNormalWrlist_ = SupportStatus::SUPPORT;
81 : } else {
82 : // 不支持RDMALite场景，需要根据opcode检查是否支持RaSendNormalWrlist接口
83 : bool isSupportTmp;
84 0 : CHK_RET(IsSupportRaSendNormalWrlist(isSupportTmp));
85 0 : isSupportRaSendNormalWrlist_ = isSupportTmp ? SupportStatus::SUPPORT : SupportStatus::NOT_SUPPORT;
86 : }
87 0 : if (isSupportRaSendNormalWrlist_ == SupportStatus::SUPPORT) {
88 0 : HCCL_RUN_INFO("[TransportRoceMem]RaSendNormalWrlist is supported");
89 : } else {
90 0 : HCCL_ERROR("[TransportRoceMem]RDMALite and RaSendNormalWrlist are not supported");
91 0 : return HCCL_E_NOT_SUPPORT;
92 : }
93 : }
94 0 : return HCCL_SUCCESS;
95 : }
96 :
97 : HcclResult
98 0 : TransportRoceMem::ExchangeMemDesc(const RmaMemDescs& localMemDescs, RmaMemDescs& remoteMemDescs, u32& actualNumOfRemote)
99 : {
100 0 : return DoExchangeMemDesc(localMemDescs, remoteMemDescs, actualNumOfRemote);
101 : }
102 :
103 0 : HcclResult TransportRoceMem::EnableMemAccess(const RmaMemDesc& remoteMemDesc, RmaMem& remoteMem)
104 : {
105 0 : std::string tempDesc = RmaMemDescCopyToStr(remoteMemDesc);
106 0 : std::shared_ptr<RemoteRdmaRmaBuffer> tempRemoteBufferPtr = make_shared<RemoteRdmaRmaBuffer>();
107 0 : HcclResult ret = tempRemoteBufferPtr->Deserialize(tempDesc);
108 0 : CHK_PRT_RET(
109 : (ret != HCCL_SUCCESS), HCCL_ERROR("[TransportRoceMem][EnableMemAccess]RemoteBuffer Deserialize failed."), ret);
110 :
111 : BufferKey<uintptr_t, u64> tempKey(
112 0 : reinterpret_cast<uintptr_t>(tempRemoteBufferPtr->GetAddr()), tempRemoteBufferPtr->GetSize());
113 0 : auto resultPair = remoteRdmaRmaBufferMgr_.Add(tempKey, tempRemoteBufferPtr);
114 0 : if (resultPair.first == remoteRdmaRmaBufferMgr_.End()) {
115 : // 输入key是表中某一个最相近key的交集、子集。返回空迭代器
116 0 : HCCL_ERROR("[TransportRoceMem][EnableMemAccess]The memory that is expected to enable"
117 : " overlaps with the memory that has been enabled, please check params");
118 0 : return HCCL_E_INTERNAL;
119 : }
120 :
121 : // 已使能：输入key是表中某一最相近key的全集。 返回添加该key的迭代器，及false
122 : // 未使能：输入key是表中某一最相近key的空集。 返回添加成功的迭代器，及true
123 0 : std::string logInfo = resultPair.second ? "Enable memory access success!" :
124 0 : "Memory is already enabled, just increase the reference count.";
125 0 : HCCL_INFO("[TransportRoceMem][EnableMemAccess]:%s", logInfo.c_str());
126 : // 填充出参TransportRmaMem信息
127 0 : remoteMem.addr = tempRemoteBufferPtr->GetAddr();
128 0 : remoteMem.size = tempRemoteBufferPtr->GetSize();
129 0 : remoteMem.type = tempRemoteBufferPtr->GetMemType();
130 0 : return HCCL_SUCCESS;
131 0 : }
132 :
133 0 : HcclResult TransportRoceMem::DisableMemAccess(const RmaMemDesc& remoteMemDesc)
134 : {
135 : // 内存去使能管理
136 0 : std::string tempDesc = RmaMemDescCopyToStr(remoteMemDesc);
137 0 : RemoteRdmaRmaBuffer tempRemoteBuffer;
138 0 : HcclResult ret = tempRemoteBuffer.Deserialize(tempDesc);
139 0 : CHK_PRT_RET(
140 : (ret != HCCL_SUCCESS), HCCL_ERROR("[TransportRoceMem][DisableMemAccess]RemoteBuffer Deserialize failed."), ret);
141 :
142 : BufferKey<uintptr_t, u64> tempKey(
143 0 : reinterpret_cast<uintptr_t>(tempRemoteBuffer.GetAddr()), tempRemoteBuffer.GetSize());
144 : try {
145 0 : if (remoteRdmaRmaBufferMgr_.Del(tempKey)) {
146 : // 删除成功：输入key是表中某一最相近key的全集，计数-1后为0，返回true
147 0 : HCCL_INFO("[TransportRoceMem][DisableMemAccess]Memory reference count is 0, disable memory access.");
148 : } else {
149 : // 删除失败：输入key是表中某一最相近key的全集，计数不为0（存在其他remoteRank使用），返回false
150 0 : HCCL_INFO("[TransportRoceMem][DisableMemAccess]Memory reference count is larger than 0"
151 : "(used by other RemoteRank), do not disable memory.");
152 : }
153 0 : return HCCL_SUCCESS;
154 0 : } catch (std::out_of_range& e) {
155 0 : HCCL_ERROR("[TransportRoceMem][DisableMemAccess] catch RmaBufferMgr Del exception: %s", e.what());
156 0 : return HCCL_E_NOT_FOUND;
157 0 : }
158 0 : }
159 :
160 0 : HcclResult TransportRoceMem::FillRmaBufferSlice(
161 : const HcclBuf& localMem, const HcclBuf& remoteMem, RmaBufferSlice& localRmaBufferSlice,
162 : RmaBufferSlice& remoteRmaBufferSlice)
163 : {
164 0 : void* remoteAddr = remoteMem.addr;
165 0 : void* localAddr = localMem.addr;
166 0 : u64 byteSize = std::min(remoteMem.len, localMem.len);
167 0 : auto localKey = BufferKey<uintptr_t, u64>(reinterpret_cast<uintptr_t>(localAddr), byteSize);
168 :
169 0 : NetDevContext* netDevCtx = static_cast<NetDevContext*>(netDevCtx_);
170 0 : std::shared_ptr<LocalRdmaRmaBufferMgr> localRmaBufferMgr = netDevCtx->GetlocalRdmaRmaBufferMgr();
171 0 : if (!localRmaBufferMgr) {
172 0 : HCCL_ERROR("[TransportRoceMem] can't get LocalRdmaRmaBufferMgr");
173 0 : return HCCL_E_INTERNAL;
174 : }
175 0 : auto localBuffer = localRmaBufferMgr->Find(localKey);
176 0 : CHK_PRT_RET(
177 : !localBuffer.first,
178 : HCCL_ERROR(
179 : "[TransportRoceMem][FillRmaBufferSlice] Can't find localBuffer by key {%p, %llu}", localAddr, byteSize),
180 : HCCL_E_INTERNAL);
181 0 : CHK_PRT_RET(
182 : !localBuffer.second->GetAddr(),
183 : HCCL_ERROR("[TransportRoceMem][FillRmaBufferSlice] The addr of local Buffer or remote buffer is nullptr."),
184 : HCCL_E_NOT_FOUND);
185 0 : CHK_PRT_RET(
186 : !localBuffer.second->GetDevAddr(),
187 : HCCL_ERROR("[TransportRoceMem][FillRmaBufferSlice]The dev addr of local Buffer is nullptr."), HCCL_E_NOT_FOUND);
188 0 : CHK_RET(CheckHcclBuffer(localAddr, localBuffer.second.get()));
189 :
190 0 : RmaBuffer* remoteBuffer = static_cast<RmaBuffer*>(remoteMem.handle);
191 0 : CHK_PRT_RET(
192 : !remoteBuffer->GetDevAddr(),
193 : HCCL_ERROR("[TransportRoceMem][FillRmaBufferSlice] The dev addr of remote buffer is nullptr."),
194 : HCCL_E_NOT_FOUND);
195 0 : CHK_PRT_RET(
196 : !remoteBuffer->GetAddr(),
197 : HCCL_ERROR("[TransportRoceMem][FillRmaBufferSlice] The addr of remote buffer is nullptr."), HCCL_E_NOT_FOUND);
198 0 : CHK_RET(CheckHcclBuffer(remoteAddr, remoteBuffer));
199 0 : u64 localDataOffSet = static_cast<u8*>(localAddr) - static_cast<u8*>(localBuffer.second->GetAddr());
200 0 : u64 remoteDataOffSet = static_cast<u8*>(remoteAddr) - static_cast<u8*>(remoteBuffer->GetAddr());
201 0 : localRmaBufferSlice.addr = static_cast<void*>(static_cast<u8*>(localBuffer.second->GetDevAddr()) + localDataOffSet);
202 0 : localRmaBufferSlice.len = byteSize;
203 0 : localRmaBufferSlice.rmaBuffer = localBuffer.second;
204 0 : localRmaBufferSlice.memType = localBuffer.second->GetMemType();
205 :
206 0 : remoteRmaBufferSlice.addr = static_cast<void*>(static_cast<u8*>(remoteBuffer->GetDevAddr()) + remoteDataOffSet);
207 0 : remoteRmaBufferSlice.len = byteSize;
208 : std::shared_ptr<RmaBuffer> temp(
209 0 : remoteBuffer, []([[maybe_unused]] RmaBuffer* p) {}); // 在外部进行删除操作，内部不能用智能指针进行生命周期管理
210 0 : remoteRmaBufferSlice.rmaBuffer = temp;
211 0 : remoteRmaBufferSlice.memType = remoteBuffer->GetMemType();
212 0 : HCCL_INFO(
213 : "[TransportRoceMem][FillRmaBufferSlice] Local address before mapping is [%p], after mapping is [%p]."
214 : "Remote address before mapping is [%p], after mapping is [%p]. Datasize is [%llu].",
215 : localAddr, localRmaBufferSlice.addr, remoteAddr, remoteRmaBufferSlice.addr, byteSize);
216 0 : return HCCL_SUCCESS;
217 0 : }
218 :
219 0 : HcclResult TransportRoceMem::FillRmaBufferSlice(
220 : const RmaOpMem& localMem, const RmaOpMem& remoteMem, RmaBufferSlice& localRmaBufferSlice,
221 : RmaBufferSlice& remoteRmaBufferSlice)
222 : {
223 0 : void* remoteAddr = remoteMem.addr;
224 0 : void* localAddr = localMem.addr;
225 0 : u64 byteSize = std::min(remoteMem.size, localMem.size);
226 0 : auto localKey = BufferKey<uintptr_t, u64>(reinterpret_cast<uintptr_t>(localAddr), byteSize);
227 0 : auto remoteKey = BufferKey<uintptr_t, u64>(reinterpret_cast<uintptr_t>(remoteAddr), byteSize);
228 :
229 0 : NetDevContext* netDevCtx = static_cast<NetDevContext*>(netDevCtx_);
230 0 : std::shared_ptr<LocalRdmaRmaBufferMgr> localRmaBufferMgr = netDevCtx->GetlocalRdmaRmaBufferMgr();
231 0 : if (!localRmaBufferMgr) {
232 0 : HCCL_ERROR("[TransportRoceMem] can't get LocalRdmaRmaBufferMgr");
233 0 : return HCCL_E_INTERNAL;
234 : }
235 0 : auto localBuffer = localRmaBufferMgr->Find(localKey);
236 0 : CHK_PRT_RET(
237 : !localBuffer.first,
238 : HCCL_ERROR(
239 : "[TransportRoceMem][FillRmaBufferSlice] Can't find localBuffer by key {%p, %llu}", localAddr, byteSize),
240 : HCCL_E_INTERNAL);
241 0 : CHK_PRT_RET(
242 : !localBuffer.second->GetAddr(),
243 : HCCL_ERROR("[TransportRoceMem][FillRmaBufferSlice] The addr of local Buffer or remote buffer is nullptr."),
244 : HCCL_E_NOT_FOUND);
245 0 : CHK_PRT_RET(
246 : !localBuffer.second->GetDevAddr(),
247 : HCCL_ERROR("[TransportRoceMem][FillRmaBufferSlice]The dev addr of local Buffer is nullptr."), HCCL_E_NOT_FOUND);
248 0 : CHK_RET(CheckHcclBuffer(localAddr, localBuffer.second.get()));
249 :
250 0 : auto remoteBuffer = remoteRdmaRmaBufferMgr_.Find(remoteKey);
251 0 : CHK_PRT_RET(
252 : !remoteBuffer.first,
253 : HCCL_ERROR(
254 : "[TransportRoceMem][FillRmaBufferSlice]Can't find remoteBuffer by key {%p, %llu}", remoteAddr, byteSize),
255 : HCCL_E_INTERNAL);
256 0 : CHK_PRT_RET(
257 : !remoteBuffer.second->GetDevAddr(),
258 : HCCL_ERROR("[TransportRoceMem][FillRmaBufferSlice]The dev addr of remote buffer is nullptr."),
259 : HCCL_E_NOT_FOUND);
260 0 : CHK_PRT_RET(
261 : !remoteBuffer.second->GetAddr(),
262 : HCCL_ERROR("[TransportRoceMem][FillRmaBufferSlice]The addr of remote buffer is nullptr."), HCCL_E_NOT_FOUND);
263 0 : CHK_RET(CheckHcclBuffer(remoteAddr, remoteBuffer.second.get()));
264 :
265 0 : u64 localDataOffSet = static_cast<u8*>(localAddr) - static_cast<u8*>(localBuffer.second->GetAddr());
266 0 : u64 remoteDataOffSet = static_cast<u8*>(remoteAddr) - static_cast<u8*>(remoteBuffer.second->GetAddr());
267 0 : localRmaBufferSlice.addr = static_cast<void*>(static_cast<u8*>(localBuffer.second->GetDevAddr()) + localDataOffSet);
268 0 : localRmaBufferSlice.len = byteSize;
269 0 : localRmaBufferSlice.rmaBuffer = localBuffer.second;
270 0 : localRmaBufferSlice.memType = localBuffer.second->GetMemType();
271 :
272 : remoteRmaBufferSlice.addr
273 0 : = static_cast<void*>(static_cast<u8*>(remoteBuffer.second->GetDevAddr()) + remoteDataOffSet);
274 0 : remoteRmaBufferSlice.len = byteSize;
275 0 : remoteRmaBufferSlice.rmaBuffer = remoteBuffer.second;
276 0 : remoteRmaBufferSlice.memType = remoteBuffer.second->GetMemType();
277 :
278 0 : HCCL_INFO(
279 : "[TransportRoceMem][FillRmaBufferSlice] Local address before mapping is [%p], after mapping is [%p]."
280 : "Remote address before mapping is [%p], after mapping is [%p]. Datasize is [%llu].",
281 : localAddr, localRmaBufferSlice.addr, remoteAddr, remoteRmaBufferSlice.addr, byteSize);
282 0 : return HCCL_SUCCESS;
283 0 : }
284 :
285 0 : HcclResult TransportRoceMem::SetSocket(const std::shared_ptr<HcclSocket>& socket)
286 : {
287 0 : CHK_SMART_PTR_NULL(socket);
288 0 : if (socket->GetStatus() != HcclSocketStatus::SOCKET_OK) {
289 0 : HCCL_ERROR("sockets does not connected");
290 0 : return HCCL_E_PARA;
291 : }
292 0 : socket_ = socket;
293 0 : return HCCL_SUCCESS;
294 : }
295 :
296 0 : HcclResult TransportRoceMem::GetRdmaHandle()
297 : {
298 0 : RaResourceInfo raResourceInfo;
299 0 : CHK_RET(NetworkManager::GetInstance(deviceLogicId_).GetRaResourceInfo(raResourceInfo));
300 0 : auto it = raResourceInfo.nicSocketMap.find(socket_->GetLocalIp());
301 0 : if (it == raResourceInfo.nicSocketMap.end()) {
302 0 : HCCL_ERROR("[TransportRoceMem][GetRdmaHandle]nic socket handle did not found");
303 0 : return HCCL_E_PARA;
304 : }
305 0 : nicRdmaHandle_ = it->second.nicRdmaHandle;
306 0 : CHK_PTR_NULL(nicRdmaHandle_);
307 0 : HCCL_INFO(
308 : "TransportRoceMem GetNetworkResource deviceLogicId_[%d] nicRdmaHandle_[%p]", deviceLogicId_, nicRdmaHandle_);
309 0 : return HCCL_SUCCESS;
310 0 : }
311 :
312 0 : HcclResult TransportRoceMem::CheckRdmaVal(void)
313 : {
314 : DevType devType;
315 0 : const u32 HCCL_RDMA_TC_MAX = 255;
316 0 : const u32 HCCL_RDMA_SL_MAX = 7;
317 0 : CHK_RET(hrtGetDeviceType(devType));
318 0 : if (devType == DevType::DEV_TYPE_910B || devType == DevType::DEV_TYPE_910_93) {
319 0 : if ((trafficClass_ != HCCL_COMM_TRAFFIC_CLASS_CONFIG_NOT_SET) && (trafficClass_ > HCCL_RDMA_TC_MAX)) {
320 0 : HCCL_ERROR("[TransportRoceMem][CheckRdmaVal]trafficClass is invalid, trafficClass:%u", trafficClass_);
321 0 : return HCCL_E_PARA;
322 : }
323 :
324 0 : if ((serviceLevel_ != HCCL_COMM_SERVICE_LEVEL_CONFIG_NOT_SET) && (serviceLevel_ > HCCL_RDMA_SL_MAX)) {
325 0 : HCCL_ERROR("[TransportRoceMem][CheckRdmaVal]serviceLevel is invalid, serviceLevel:%u", serviceLevel_);
326 0 : return HCCL_E_PARA;
327 : }
328 : }
329 0 : return HCCL_SUCCESS;
330 : }
331 :
332 0 : HcclResult TransportRoceMem::ConnectImpl(s32 timeoutSec)
333 : {
334 : // 增加1s的超时时间防止剩余超时时间不足
335 0 : s32 redundantTimeout = timeoutSec == INT_MAX ? timeoutSec : timeoutSec + 1;
336 0 : CHK_RET(GetRdmaHandle());
337 0 : CHK_RET(CreateCqAndQp());
338 0 : CHK_RET(CreatSignalMesg());
339 0 : CHK_RET(CreateNotifyValueBuffer());
340 0 : CHK_RET(ExchangeNotifyValueBuffer(redundantTimeout));
341 0 : CHK_RET(QpConnect(redundantTimeout));
342 0 : CHK_RET(WaitQPLinkComplete(redundantTimeout));
343 0 : return HCCL_SUCCESS;
344 : }
345 :
346 0 : HcclResult TransportRoceMem::Connect(s32 timeoutSec)
347 : {
348 0 : devicePhyId_ = (static_cast<NetDevContext*>(netDevCtx_))->GetPhyId();
349 0 : CHK_PRT_RET(devicePhyId_ == HOST_DEVICE_ID, HCCL_ERROR("[Connect] devicePhyId is invalid"), HCCL_E_INTERNAL);
350 0 : deviceLogicId_ = (static_cast<NetDevContext*>(netDevCtx_))->GetLogicId();
351 0 : CHK_PRT_RET(
352 : deviceLogicId_ == HOST_DEVICE_ID, HCCL_ERROR("deviceLogicId is same as host device id"), HCCL_E_INTERNAL);
353 0 : CHK_RET(CheckRdmaVal());
354 0 : CHK_PTR_NULL(dispatcher_);
355 0 : CHK_SMART_PTR_NULL(notifyPool_);
356 0 : CHK_RET(notifyPool_->RegisterOp(socket_->GetTag()));
357 0 : auto ret = ConnectImpl(timeoutSec);
358 : // 解注册之后再返回ret
359 0 : CHK_RET(notifyPool_->UnregisterOp(socket_->GetTag()));
360 0 : return ret;
361 : }
362 :
363 0 : HcclResult TransportRoceMem::TransportRdmaWithType(
364 : const RmaBufferSlice& localRmaBufferSlice, const RmaBufferSlice& remoteRmaBufferSlice, const rtStream_t& stream,
365 : const RdmaOp& rdmaOp)
366 : {
367 0 : CHK_PTR_NULL(localRmaBufferSlice.addr);
368 0 : CHK_PTR_NULL(remoteRmaBufferSlice.addr);
369 0 : u64 processedOffset = 0;
370 0 : u64 remainingBytes = remoteRmaBufferSlice.len;
371 0 : u64 byteSizeChunk = 0;
372 0 : uint64_t localStartAddr = 0;
373 0 : uint64_t remoteStartAddr = 0;
374 0 : while (remainingBytes > 0) {
375 0 : localStartAddr = reinterpret_cast<uint64_t>(static_cast<u8*>(localRmaBufferSlice.addr) + processedOffset);
376 0 : remoteStartAddr = reinterpret_cast<uint64_t>(static_cast<u8*>(remoteRmaBufferSlice.addr) + processedOffset);
377 0 : byteSizeChunk = remainingBytes > MAX_RDMA_WQE_SIZE ? MAX_RDMA_WQE_SIZE : remainingBytes;
378 : std::shared_ptr<RemoteRdmaRmaBuffer> remoteRdmaRmaBuffer
379 0 : = dynamic_pointer_cast<RemoteRdmaRmaBuffer>(remoteRmaBufferSlice.rmaBuffer);
380 : std::shared_ptr<LocalRdmaRmaBuffer> localRdmaRmaBuffer
381 0 : = dynamic_pointer_cast<LocalRdmaRmaBuffer>(localRmaBufferSlice.rmaBuffer);
382 : struct WrInfo wr[WR_NUM];
383 0 : wr[0].wrId = sendWrHandle.fetch_add(1, std::memory_order_relaxed);
384 0 : wr[0].memList.addr = localStartAddr;
385 0 : wr[0].memList.len = byteSizeChunk;
386 0 : wr[0].memList.lkey = localRdmaRmaBuffer->GetKey();
387 0 : wr[0].dstAddr = remoteStartAddr;
388 0 : wr[0].rkey = remoteRdmaRmaBuffer->GetKey();
389 0 : wr[0].op = static_cast<u32>(rdmaOp);
390 0 : wr[0].sendFlags = remainingBytes > MAX_RDMA_WQE_SIZE ? 0 : RA_SEND_SIGNALED;
391 :
392 : struct SendWrRsp opRsp[WR_NUM];
393 :
394 0 : HCCL_DEBUG(
395 : "Op type[%d], wr.wrId[%llu], src addr[%p], dest addr[%p], len[%u]", rdmaOp, wr[0].wrId,
396 : localRmaBufferSlice.addr, remoteRmaBufferSlice.addr, wr[0].memList.len);
397 0 : u32 completeNum = 0;
398 0 : CHK_RET(HrtRaSendNormalWrlist(dataQpInfo_.qpHandle, wr, opRsp, WR_NUM, &completeNum));
399 0 : CHK_RET(DoorBellSend(dataQpInfo_.qpMode, wr[0], opRsp[0], stream));
400 0 : remainingBytes -= byteSizeChunk;
401 0 : processedOffset += byteSizeChunk;
402 0 : }
403 0 : return HCCL_SUCCESS;
404 : }
405 :
406 0 : HcclResult TransportRoceMem::TransportIpc(
407 : const RmaBufferSlice& dstRmaBufferSlice, const RmaBufferSlice& srcRmaBufferSlice, const rtStream_t& stream)
408 : {
409 0 : Stream hcclStream(stream);
410 0 : DispatcherPub* dispatcher = reinterpret_cast<DispatcherPub*>(dispatcher_);
411 0 : CHK_RET(dispatcher->MemcpyAsync(
412 : dstRmaBufferSlice.addr, dstRmaBufferSlice.len, srcRmaBufferSlice.addr, srcRmaBufferSlice.len,
413 : HcclRtMemcpyKind::HCCL_RT_MEMCPY_KIND_DEVICE_TO_DEVICE, hcclStream, remoteRankId_, hccl::LinkType::LINK_HCCS));
414 0 : return HCCL_SUCCESS;
415 0 : }
416 :
417 0 : HcclResult TransportRoceMem::Write(const HcclBuf& remoteMem, const HcclBuf& localMem, const rtStream_t& stream)
418 : {
419 0 : CHK_RET(CheckRaSendNormalWrlistSupport());
420 0 : CHK_PRT_RET(
421 : (localMem.addr == nullptr) || (remoteMem.addr == nullptr),
422 : HCCL_ERROR("[TransportRoceMem]localMem addr or remoteMem addr is invalid"), HCCL_E_PARA);
423 0 : CHK_PRT_RET(
424 : (localMem.len == 0U) || (remoteMem.len == 0U),
425 : HCCL_ERROR(
426 : "[TransportRoceMem]localMem size[%llu] or remoteMem size[%llu]is invalid", localMem.len, remoteMem.len),
427 : HCCL_E_PARA);
428 0 : CHK_PRT_RET(stream == nullptr, HCCL_ERROR("[TransportRoceMem]stream is invalid"), HCCL_E_PARA);
429 :
430 0 : RmaBufferSlice localRmaBufferSlice{};
431 0 : RmaBufferSlice remoteRmaBufferSlice{};
432 0 : CHK_RET(FillRmaBufferSlice(localMem, remoteMem, localRmaBufferSlice, remoteRmaBufferSlice));
433 0 : CHK_RET(TransportRdmaWithType(localRmaBufferSlice, remoteRmaBufferSlice, stream, RdmaOp::OP_WRITE));
434 0 : return HCCL_SUCCESS;
435 0 : }
436 :
437 0 : HcclResult TransportRoceMem::Write(const RmaOpMem& remoteMem, const RmaOpMem& localMem, const rtStream_t& stream)
438 : {
439 0 : CHK_RET(CheckRaSendNormalWrlistSupport());
440 0 : CHK_PRT_RET(
441 : (localMem.addr == nullptr) || (remoteMem.addr == nullptr),
442 : HCCL_ERROR("[TransportRoceMem]localMem addr or remoteMem addr is invalid"), HCCL_E_PARA);
443 0 : CHK_PRT_RET(
444 : (localMem.size == 0U) || (remoteMem.size == 0U),
445 : HCCL_ERROR(
446 : "[TransportRoceMem]localMem size[%llu] or remoteMem size[%llu]is invalid", localMem.size, remoteMem.size),
447 : HCCL_E_PARA);
448 0 : CHK_PRT_RET(stream == nullptr, HCCL_ERROR("[TransportRoceMem]stream is invalid"), HCCL_E_PARA);
449 :
450 0 : RmaBufferSlice localRmaBufferSlice{};
451 0 : RmaBufferSlice remoteRmaBufferSlice{};
452 0 : CHK_RET(FillRmaBufferSlice(localMem, remoteMem, localRmaBufferSlice, remoteRmaBufferSlice));
453 0 : CHK_RET(TransportRdmaWithType(localRmaBufferSlice, remoteRmaBufferSlice, stream, RdmaOp::OP_WRITE));
454 0 : return HCCL_SUCCESS;
455 0 : }
456 :
457 0 : HcclResult TransportRoceMem::Read(const HcclBuf& localMem, const HcclBuf& remoteMem, const rtStream_t& stream)
458 : {
459 0 : CHK_RET(CheckRaSendNormalWrlistSupport());
460 0 : CHK_PRT_RET(
461 : (localMem.addr == nullptr) || (remoteMem.addr == nullptr),
462 : HCCL_ERROR("[TransportRoceMem]localMem addr or remoteMem addr is invalid"), HCCL_E_PARA);
463 0 : CHK_PRT_RET(
464 : (localMem.len == 0U) || (remoteMem.len == 0U),
465 : HCCL_ERROR(
466 : "[TransportRoceMem]localMem size[%llu] or remoteMem size[%llu]is invalid", localMem.len, remoteMem.len),
467 : HCCL_E_PARA);
468 0 : CHK_PRT_RET(stream == nullptr, HCCL_ERROR("[TransportRoceMem]stream is invalid"), HCCL_E_PARA);
469 :
470 0 : RmaBufferSlice localRmaBufferSlice{};
471 0 : RmaBufferSlice remoteRmaBufferSlice{};
472 0 : CHK_RET(FillRmaBufferSlice(localMem, remoteMem, localRmaBufferSlice, remoteRmaBufferSlice));
473 0 : CHK_RET(TransportRdmaWithType(localRmaBufferSlice, remoteRmaBufferSlice, stream, RdmaOp::OP_READ));
474 0 : return HCCL_SUCCESS;
475 0 : }
476 :
477 0 : HcclResult TransportRoceMem::Read(const RmaOpMem& localMem, const RmaOpMem& remoteMem, const rtStream_t& stream)
478 : {
479 0 : CHK_RET(CheckRaSendNormalWrlistSupport());
480 0 : CHK_PRT_RET(
481 : (localMem.addr == nullptr) || (remoteMem.addr == nullptr),
482 : HCCL_ERROR("[TransportRoceMem]localMem addr or remoteMem addr is invalid"), HCCL_E_PARA);
483 0 : CHK_PRT_RET(
484 : (localMem.size == 0U) || (remoteMem.size == 0U),
485 : HCCL_ERROR(
486 : "[TransportRoceMem]localMem size[%llu] or remoteMem size[%llu]is invalid", localMem.size, remoteMem.size),
487 : HCCL_E_PARA);
488 0 : CHK_PRT_RET(stream == nullptr, HCCL_ERROR("[TransportRoceMem]stream is invalid"), HCCL_E_PARA);
489 :
490 0 : RmaBufferSlice localRmaBufferSlice{};
491 0 : RmaBufferSlice remoteRmaBufferSlice{};
492 0 : CHK_RET(FillRmaBufferSlice(localMem, remoteMem, localRmaBufferSlice, remoteRmaBufferSlice));
493 0 : CHK_RET(TransportRdmaWithType(localRmaBufferSlice, remoteRmaBufferSlice, stream, RdmaOp::OP_READ));
494 0 : return HCCL_SUCCESS;
495 0 : }
496 :
497 0 : HcclResult TransportRoceMem::AddOpFence(const rtStream_t& stream)
498 : {
499 0 : CHK_RET(CheckRaSendNormalWrlistSupport());
500 0 : auto opType = static_cast<u32>(MemType::SEND_NOTIFY_MEM);
501 : struct WrInfo wr[WR_NUM];
502 0 : wr[0].wrId = sendWrHandle.fetch_add(1, std::memory_order_relaxed);
503 0 : wr[0].memList.addr = reinterpret_cast<uint64_t>(rdmaSignal_[0].addr);
504 0 : wr[0].memList.len = notifyMemMsg_[opType].len;
505 0 : wr[0].memList.lkey = rdmaSignal_[0].lkey;
506 0 : wr[0].dstAddr = static_cast<u64>(reinterpret_cast<uintptr_t>(notifyMemMsg_[opType].addr));
507 0 : wr[0].rkey = notifyMemMsg_[opType].rkey;
508 0 : wr[0].op = static_cast<u32>(RdmaOp::OP_READ);
509 0 : wr[0].sendFlags = RA_SEND_SIGNALED | RA_SEND_FENCE;
510 0 : u32 completeNum = 0;
511 : struct SendWrRsp opRsp[WR_NUM];
512 0 : CHK_RET(HrtRaSendNormalWrlist(dataQpInfo_.qpHandle, wr, opRsp, WR_NUM, &completeNum));
513 0 : CHK_RET(DoorBellSend(dataQpInfo_.qpMode, wr[0], opRsp[0], stream));
514 0 : CHK_RET(WaitOpFence(stream));
515 0 : HCCL_DEBUG(
516 : "[AddOpFence] wr.wrId[%llu], local addr[%p], remote addr[%p], len[%u], lkey[%u], rkey[%u]", wr[0].wrId,
517 : rdmaSignal_[0].addr, notifyMemMsg_[opType].addr, wr[0].memList.len, wr[0].memList.lkey, wr[0].rkey);
518 0 : return HCCL_SUCCESS;
519 : }
520 :
521 0 : HcclResult TransportRoceMem::GetQpInfo(HcclQpInfoV2& qpInfo)
522 : {
523 0 : qpInfo.qpPtr = aiQpInfo_.aiQpAddr; // reinterpret_cast<u64>(dataQpInfo_.qp)
524 0 : qpInfo.sqIndex = aiQpInfo_.sqIndex;
525 0 : qpInfo.dbIndex = aiQpInfo_.dbIndex;
526 0 : qpInfo.retryCnt = static_cast<u16>(GetExternalInputRdmaRetryCnt());
527 0 : qpInfo.retryTime = static_cast<u16>(GetExternalInputRdmaTimeOut());
528 0 : struct ibv_qp* qp = reinterpret_cast<struct ibv_qp*>(qpInfo.qpPtr);
529 0 : HCCL_DEBUG("[%s] qp=%p", __func__, qp);
530 0 : return HCCL_SUCCESS;
531 : }
532 :
533 0 : HcclResult TransportRoceMem::GetMemInfo(u32& lkey, u32& rkey, HcclBuf& localMem, HcclBuf& remoteMem)
534 : {
535 0 : CHK_PRT_RET(
536 : (localMem.addr == nullptr) || (remoteMem.addr == nullptr),
537 : HCCL_ERROR(
538 : "[TransportRoceMem] localMem addr[%p] or remoteMem addr[%p] is invalid", localMem.addr, remoteMem.addr),
539 : HCCL_E_PARA);
540 0 : CHK_PRT_RET(
541 : (localMem.len == 0U) || (remoteMem.len == 0U),
542 : HCCL_ERROR(
543 : "[TransportRoceMem] localMem size[%llu] or remoteMem size[%llu]is invalid", localMem.len, remoteMem.len),
544 : HCCL_E_PARA);
545 :
546 0 : RmaBufferSlice localRmaBufferSlice{};
547 0 : RmaBufferSlice remoteRmaBufferSlice{};
548 0 : CHK_RET(FillRmaBufferSlice(localMem, remoteMem, localRmaBufferSlice, remoteRmaBufferSlice));
549 :
550 0 : auto localRdmaRmaBuffer = dynamic_pointer_cast<LocalRdmaRmaBuffer>(localRmaBufferSlice.rmaBuffer);
551 0 : lkey = localRdmaRmaBuffer->GetKey();
552 0 : localMem.addr = localRmaBufferSlice.addr;
553 0 : localMem.len = localRmaBufferSlice.len;
554 :
555 0 : auto remoteRdmaRmaBuffer = dynamic_pointer_cast<RemoteRdmaRmaBuffer>(remoteRmaBufferSlice.rmaBuffer);
556 0 : rkey = remoteRdmaRmaBuffer->GetKey();
557 0 : remoteMem.addr = remoteRmaBufferSlice.addr;
558 0 : remoteMem.len = remoteRmaBufferSlice.len;
559 :
560 0 : return HCCL_SUCCESS;
561 0 : }
562 :
563 0 : HcclResult TransportRoceMem::GetOpFence(u32& lkey, u32& rkey, HcclBuf& localMem, HcclBuf& remoteMem)
564 : {
565 0 : auto opType = static_cast<u32>(MemType::SEND_NOTIFY_MEM);
566 0 : lkey = rdmaSignal_[0].lkey;
567 0 : localMem.addr = rdmaSignal_[0].addr;
568 0 : localMem.len = rdmaSignal_[0].len;
569 0 : rkey = notifyMemMsg_[opType].rkey;
570 0 : remoteMem.addr = notifyMemMsg_[opType].addr;
571 0 : HCCL_DEBUG(
572 : "[GetOpFence] local addr[%p], remote addr[%p], len[%u], lkey[%u], rkey[%u]", localMem.addr, remoteMem.addr,
573 : localMem.len, lkey, rkey);
574 0 : return HCCL_SUCCESS;
575 : }
576 :
577 0 : HcclResult TransportRoceMem::GetTransInfo(
578 : HcclQpInfoV2& qpInfo, u32* lkey, u32* rkey, HcclBuf* localMem, HcclBuf* remoteMem, u32 num)
579 : {
580 0 : CHK_PTR_NULL(lkey);
581 0 : CHK_PTR_NULL(rkey);
582 0 : CHK_PTR_NULL(localMem);
583 0 : CHK_PTR_NULL(remoteMem);
584 0 : CHK_PRT_RET(
585 : num == 0, HCCL_ERROR("[GetTransInfo] mem num should not be zero, at least one for OpFence"), HCCL_E_PARA);
586 0 : CHK_RET(GetQpInfo(qpInfo));
587 0 : for (u32 i = 0; i < num - 1; ++i) { // last element is signal
588 0 : HcclResult ret = GetMemInfo(lkey[i], rkey[i], localMem[i], remoteMem[i]);
589 0 : CHK_PRT_RET(
590 : ret != HCCL_SUCCESS,
591 : HCCL_ERROR(
592 : "[GetTransInfo] failed at index[%u], localAddr[%p/%llu], "
593 : "remoteAddr[%p/%llu]",
594 : i, localMem[i].addr, localMem[i].len, remoteMem[i].addr, remoteMem[i].len),
595 : ret);
596 : }
597 0 : CHK_RET(GetOpFence(lkey[num - 1], rkey[num - 1], localMem[num - 1], remoteMem[num - 1]));
598 0 : return HCCL_SUCCESS;
599 : }
600 :
601 0 : HcclResult TransportRoceMem::WaitOpFence(const rtStream_t& stream)
602 : {
603 0 : auto opType = static_cast<u32>(MemType::SEND_NOTIFY_MEM);
604 0 : hccl::Stream hcclStream(stream);
605 0 : DispatcherPub* dispatcher = reinterpret_cast<DispatcherPub*>(dispatcher_);
606 0 : const u32 timeOut = (GetExternalInputHcclExecTimeoutSet() != HcclExecTimeoutSet::HCCL_EXEC_TIMEOUT_NOT_SET)
607 0 : || dispatcher->GetExecTimeOutSet() ?
608 0 : dispatcher->GetExecTimeOut() :
609 0 : NOTIFY_DEFAULT_WAIT_TIME;
610 0 : HcclResult ret = LocalIpcNotify::Wait(
611 0 : hcclStream, dispatcher, remoteIsendDoneSignal_, INVALID_VALUE_STAGE, timeOut, localRankId_, remoteRankId_);
612 0 : CHK_PRT_RET(
613 : ret != HCCL_SUCCESS,
614 : HCCL_ERROR(
615 : "[WaitOpFence] timeout[%u s], local addr[%p], remote addr[%p], remoteRankId[%u], streamId[%u]", timeOut,
616 : rdmaSignal_[0].addr, notifyMemMsg_[opType].addr, remoteRankId_, hcclStream.id()),
617 : ret);
618 0 : HCCL_DEBUG(
619 : "[WaitOpFence] local addr[%p], remote addr[%p], remoteRankId[%u], streamId[%u]", rdmaSignal_[0].addr,
620 : notifyMemMsg_[opType].addr, remoteRankId_, hcclStream.id());
621 0 : return HCCL_SUCCESS;
622 0 : }
623 :
624 0 : HcclResult TransportRoceMem::BatchWrite(
625 : [[maybe_unused]] const std::vector<MemDetails>& remoteMems,
626 : [[maybe_unused]] const std::vector<MemDetails>& localMems, [[maybe_unused]] Stream& stream)
627 : {
628 0 : HCCL_ERROR("TransportRoceMem doesn't support BatchWrite");
629 0 : return HCCL_E_NOT_SUPPORT;
630 : }
631 :
632 0 : HcclResult TransportRoceMem::BatchRead(
633 : [[maybe_unused]] const std::vector<MemDetails>& localMems,
634 : [[maybe_unused]] const std::vector<MemDetails>& remoteMems, [[maybe_unused]] Stream& stream)
635 : {
636 0 : HCCL_ERROR("TransportRoceMem doesn't support BatchRead");
637 0 : return HCCL_E_NOT_SUPPORT;
638 : }
639 :
640 0 : HcclResult TransportRoceMem::AddOpFence(
641 : [[maybe_unused]] const MemDetails& localFenceMem, [[maybe_unused]] const MemDetails& remoteFenceMem,
642 : [[maybe_unused]] Stream& stream)
643 : {
644 0 : HCCL_ERROR("TransportRoceMem doesn't support AICPU AddOpFence");
645 0 : return HCCL_E_NOT_SUPPORT;
646 : }
647 :
648 0 : HcclResult TransportRoceMem::CreateCqAndQp()
649 : {
650 0 : dataQpInfo_.flag = QP_FLAG_RC;
651 0 : dataQpInfo_.qpMode = OPBASE_QP_MODE_EXT;
652 0 : dataQpInfo_.trafficClass = trafficClass_;
653 0 : dataQpInfo_.serviceLevel = serviceLevel_;
654 0 : if (aicpuUnfoldMode_) {
655 0 : CHK_RET(CreateAiQp(nicRdmaHandle_, aiQpInfo_, dataQpInfo_, devicePhyId_));
656 : } else {
657 0 : CHK_RET(CreateQpWithCq(nicRdmaHandle_, -1, -1, nullptr, nullptr, dataQpInfo_, true, true));
658 : }
659 0 : return HCCL_SUCCESS;
660 : }
661 :
662 0 : HcclResult TransportRoceMem::QpConnect(s32 timeoutSec)
663 : {
664 0 : CHK_RET(HrtRaQpConnectAsync(
665 : dataQpInfo_.qpHandle, socket_->GetFdHandle(),
666 : [this]() -> bool {
667 : return this->socket_->GetStopFlag();
668 : },
669 : timeoutSec));
670 :
671 0 : return HCCL_SUCCESS;
672 : }
673 :
674 0 : HcclResult TransportRoceMem::RecoverNotifyMsg(MemMsg* remoteRdmaSignal, u64 signalNum)
675 : {
676 0 : if (signalNum <= 0) {
677 0 : return HCCL_E_NOT_FOUND;
678 : }
679 0 : MemType tmpMemType = MemType::MEM_TYPE_RESERVED;
680 0 : for (u64 i = 0; i < signalNum; i++) {
681 0 : HCCL_DEBUG(
682 : "recv mrRegFlag:[%d] notifyAddr:[%p] len:[%lu] memType:[%d], rkey:[%u] ",
683 : (remoteRdmaSignal + i)->mrRegFlag, (remoteRdmaSignal + i)->addr, (remoteRdmaSignal + i)->len,
684 : static_cast<int>((remoteRdmaSignal + i)->memType), (remoteRdmaSignal + i)->lkey);
685 0 : tmpMemType = (remoteRdmaSignal + i)->memType;
686 0 : if ((remoteRdmaSignal + i)->memType == MemType::NOTIFY_SRC_MEM) {
687 0 : tmpMemType = MemType::SEND_NOTIFY_MEM;
688 0 : notifyMemMsg_[tmpMemType].mrRegFlag = (remoteRdmaSignal + i)->mrRegFlag;
689 0 : notifyMemMsg_[tmpMemType].addr = (remoteRdmaSignal + i)->addr;
690 0 : notifyMemMsg_[tmpMemType].len = (remoteRdmaSignal + i)->len;
691 0 : notifyMemMsg_[tmpMemType].memType = MemType::SEND_NOTIFY_MEM;
692 0 : notifyMemMsg_[tmpMemType].rkey = (remoteRdmaSignal + i)->lkey;
693 : }
694 : }
695 :
696 0 : return HCCL_SUCCESS;
697 : }
698 :
699 0 : HcclResult TransportRoceMem::CreatSignalMesg()
700 : {
701 0 : CHK_RET(GetNotifySize());
702 0 : CHK_RET(CreateRdmaSignal(remoteIsendDoneSignal_, rdmaSignal_[0], MemType::RECV_NOTIFY_MEM));
703 0 : return HCCL_SUCCESS;
704 : }
705 :
706 0 : HcclResult TransportRoceMem::GetNotifySize()
707 : {
708 : DevType devType;
709 0 : CHK_RET(hrtHalGetDeviceType(deviceLogicId_, devType));
710 0 : if ((devType == DevType::DEV_TYPE_910B) || (devType == DevType::DEV_TYPE_910_93)) {
711 0 : notifySize_ = 4; // 910B/910_93 每个notify占4个字节
712 : } else {
713 0 : notifySize_ = 8; // 其余芯片类型每个notify占8个字节
714 : }
715 0 : HCCL_INFO("devType[%d] notifySize[%d]", devType, notifySize_);
716 0 : return HCCL_SUCCESS;
717 : }
718 :
719 0 : HcclResult TransportRoceMem::ExchangeNotifyValueBuffer(s32 timeoutSec)
720 : {
721 0 : CHK_RET(socket_->Send(
722 : ¬ifyMemMsg_[static_cast<u32>(MemType::NOTIFY_SRC_MEM)], sizeof(MemMsg) * REMOTE_RDMA_SIGNAL_SIZE));
723 0 : HCCL_DEBUG(
724 : "send mrRegFlag:[%d] notifyAddr:[%p] len:[%lu] memType:[%d], rkey:[%u] ",
725 : notifyMemMsg_[static_cast<u32>(MemType::NOTIFY_SRC_MEM)].mrRegFlag,
726 : notifyMemMsg_[static_cast<u32>(MemType::NOTIFY_SRC_MEM)].addr,
727 : notifyMemMsg_[static_cast<u32>(MemType::NOTIFY_SRC_MEM)].len,
728 : notifyMemMsg_[static_cast<u32>(MemType::NOTIFY_SRC_MEM)].memType,
729 : notifyMemMsg_[static_cast<u32>(MemType::NOTIFY_SRC_MEM)].lkey);
730 0 : MemMsg remoteNotifyValue[REMOTE_RDMA_SIGNAL_SIZE];
731 0 : CHK_RET(socket_->Recv(remoteNotifyValue, sizeof(MemMsg) * REMOTE_RDMA_SIGNAL_SIZE, timeoutSec));
732 0 : CHK_RET(RecoverNotifyMsg(remoteNotifyValue, REMOTE_RDMA_SIGNAL_SIZE));
733 0 : return HCCL_SUCCESS;
734 : }
735 :
736 0 : HcclResult TransportRoceMem::CreateRdmaSignal(
737 : std::shared_ptr<LocalIpcNotify>& localNotify, MemMsg& rdmaSignalInfo, MemType notifyType)
738 : {
739 0 : u64 notifyOffset = 0;
740 0 : u64 notifyBaseVa = 0; // notify寄存器虚拟地址
741 0 : u64 notifyTotalSize = 0;
742 :
743 0 : RemoteRankInfo info(devicePhyId_, remoteRankId_);
744 0 : CHK_RET(SalGetBareTgid(&info.remotePid)); // 当前进程id
745 0 : CHK_RET(notifyPool_->Alloc(socket_->GetTag(), info, localNotify));
746 : // 设置remote id
747 0 : s64 recvId = 0xFFFFFFFF00000000 | (static_cast<s64>(info.remotePid) & 0xFFFFFFFF);
748 0 : CHK_RET(localNotify->Grant(recvId));
749 :
750 0 : CHK_RET(HrtRaGetNotifyBaseAddr(nicRdmaHandle_, ¬ifyBaseVa, ¬ifyTotalSize));
751 0 : CHK_RET(localNotify->GetNotifyOffset(notifyOffset));
752 0 : u64 notifyVa = notifyBaseVa + notifyOffset;
753 0 : rdmaSignalInfo.mrRegFlag = 0;
754 0 : rdmaSignalInfo.addr = reinterpret_cast<void*>(static_cast<uintptr_t>(notifyVa));
755 0 : rdmaSignalInfo.len = notifySize_;
756 0 : rdmaSignalInfo.memType = notifyType;
757 :
758 0 : HCCL_INFO(
759 : "notifyBaseVa=0x%llx, notifyTotalSize=0x%x, notifyOffset=0x%llx, notifyVa=0x%llx", notifyBaseVa,
760 : notifyTotalSize, notifyOffset, notifyVa);
761 :
762 0 : struct MrInfoT mrInfo = {};
763 0 : CHK_RET(HrtRaGetNotifyMrInfo(devicePhyId_, nicRdmaHandle_, &mrInfo));
764 0 : rdmaSignalInfo.lkey = mrInfo.lkey;
765 :
766 0 : HcclSignalInfo notifyInfo{};
767 0 : notifyInfo.resId = INVALID_U64;
768 0 : CHK_RET(localNotify->GetNotifyData(notifyInfo));
769 0 : HCCL_INFO("CreateRdmaSignal localNotify id[%llu]", notifyInfo.resId);
770 0 : return HCCL_SUCCESS;
771 : }
772 :
773 0 : HcclResult TransportRoceMem::CreateNotifyValueBuffer()
774 : {
775 0 : if (notifyMem_.ptr() == nullptr) {
776 0 : u64 notifyVaule = 1; // notify值写1表示record
777 0 : CHK_RET(DeviceMem::alloc(notifyMem_, notifyValueSize_));
778 :
779 0 : CHK_RET(hrtMemSyncCopy(
780 : notifyMem_.ptr(), notifyMem_.size(), ¬ifyVaule, notifySize_,
781 : HcclRtMemcpyKind::HCCL_RT_MEMCPY_KIND_HOST_TO_DEVICE));
782 : }
783 :
784 0 : struct MrInfoT mrInfo = {};
785 0 : mrInfo.addr = notifyMem_.ptr();
786 0 : mrInfo.size = notifySize_;
787 0 : mrInfo.access = access_;
788 0 : CHK_RET(hrtRaRegGlobalMr(nicRdmaHandle_, mrInfo, notifyValueMemMrHandle_));
789 0 : notifyMemMsg_[static_cast<u32>(MemType::NOTIFY_SRC_MEM)].mrRegFlag = REG_VALID;
790 0 : notifyMemMsg_[static_cast<u32>(MemType::NOTIFY_SRC_MEM)].addr = notifyMem_.ptr();
791 0 : notifyMemMsg_[static_cast<u32>(MemType::NOTIFY_SRC_MEM)].len = notifySize_;
792 0 : notifyMemMsg_[static_cast<u32>(MemType::NOTIFY_SRC_MEM)].memType = MemType::NOTIFY_SRC_MEM;
793 0 : notifyMemMsg_[static_cast<u32>(MemType::NOTIFY_SRC_MEM)].lkey = mrInfo.lkey;
794 0 : HCCL_DEBUG("notifyValueMem_=%p", notifyMem_.ptr());
795 0 : return HCCL_SUCCESS;
796 : }
797 :
798 0 : HcclResult TransportRoceMem::DoorBellSend(
799 : [[maybe_unused]] const s32 qpMode, WrInfo& sendWrInfo, const SendWrRsp& opRsp, rtStream_t stream)
800 : {
801 0 : struct SendWr sendwr = {};
802 0 : sendwr.bufList = &sendWrInfo.memList;
803 0 : sendwr.bufNum = 1; /* 此处list只有一个，设置为1 */
804 0 : sendwr.dstAddr = sendWrInfo.dstAddr;
805 0 : sendwr.rkey = sendWrInfo.rkey;
806 0 : sendwr.op = sendWrInfo.op;
807 0 : sendwr.sendFlag = sendWrInfo.sendFlags;
808 0 : u32 dbIndex = static_cast<u32>(opRsp.db.dbIndex);
809 0 : u64 dbInfo = static_cast<u64>(opRsp.db.dbInfo);
810 0 : CHK_RET(RdmaDbSend(dbIndex, dbInfo, sendwr, stream));
811 0 : return HCCL_SUCCESS;
812 : }
813 :
814 0 : HcclResult TransportRoceMem::RdmaDbSend(u32 dbindex, u64 dbinfo, const struct SendWr& sendWr, rtStream_t stream)
815 : {
816 0 : hccl::Stream hcclStream(stream);
817 0 : DispatcherPub* dispatcher = reinterpret_cast<DispatcherPub*>(dispatcher_);
818 0 : s32 ret = dispatcher->RdmaSend(dbindex, dbinfo, sendWr, hcclStream, remoteRankId_);
819 0 : CHK_PRT_RET(
820 : ret != HCCL_SUCCESS,
821 : HCCL_ERROR(
822 : "[RdmaDbSend]errNo[0x%016llx] rdma db send fail, "
823 : "return[%d]. para: dbindex[%u]dbinfo[%llu].",
824 : HCCL_ERROR_CODE(HCCL_E_INTERNAL), ret, dbindex, dbinfo),
825 : HCCL_E_INTERNAL);
826 0 : return HCCL_SUCCESS;
827 0 : }
828 :
829 0 : HcclResult TransportRoceMem::WaitQPLinkComplete(s32 timeoutSec)
830 : {
831 0 : auto startTime = chrono::steady_clock::now();
832 0 : maxTimeOut_ = std::chrono::seconds(timeoutSec);
833 0 : while ((chrono::steady_clock::now() - startTime) < maxTimeOut_) {
834 0 : HcclResult ret = GetQpStatus();
835 0 : if (ret == HCCL_E_AGAIN) {
836 0 : SaluSleep(WAIT_LINK_BUILD_DELAY_TIME_US);
837 0 : continue;
838 : }
839 0 : if (ret == HCCL_SUCCESS) {
840 0 : HCCL_INFO("TransportRoceMem QP connect success");
841 : } else {
842 0 : HCCL_ERROR("TransportRoceMem QP connect failed, ret[%d]!", ret);
843 : }
844 0 : return ret;
845 : }
846 0 : HCCL_RUN_INFO(
847 : "WaitBuildLinkComplete timeOut[%d] s, localRank[%u], remoteRank[%u]", timeoutSec, localRankId_, remoteRankId_);
848 0 : return HCCL_E_TIMEOUT;
849 : }
850 :
851 0 : HcclResult TransportRoceMem::GetQpStatus()
852 : {
853 0 : int qpStatus = 0;
854 0 : s32 ret = 0;
855 :
856 0 : ret = hrtGetRaQpStatus(dataQpInfo_.qpHandle, &qpStatus);
857 0 : if (ret != 0) {
858 0 : return HCCL_E_INTERNAL;
859 0 : } else if (ret == 0 && qpStatus != 1) { // 为1时，qp 建链成功
860 0 : return HCCL_E_AGAIN;
861 : }
862 0 : return HCCL_SUCCESS;
863 : }
864 :
865 0 : HcclResult TransportRoceMem::DestroyCqAndQp()
866 : {
867 0 : HCCL_INFO("TransportRoceMem DestroyCqAndQp");
868 0 : if (aicpuUnfoldMode_) {
869 0 : CHK_RET(DestroyAiQp(dataQpInfo_));
870 : } else {
871 0 : CHK_RET(DestroyQpWithCq(dataQpInfo_, true));
872 : }
873 0 : return HCCL_SUCCESS;
874 : }
875 :
876 : } // namespace hccl
|
