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