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_remote_access.h"
12 : #include "externalinput_pub.h"
13 :
14 : namespace hccl {
15 : using namespace std;
16 : std::array<DeviceMem, MAX_MODULE_DEVICE_NUM> TransportRemoteAccess::notifyValueMem_;
17 : std::array<std::mutex, MAX_MODULE_DEVICE_NUM> TransportRemoteAccess::notifyValueMutex_;
18 : std::array<Referenced, MAX_MODULE_DEVICE_NUM> TransportRemoteAccess::instanceRef_; // 实例计数，用于释放静态资源
19 0 : TransportRemoteAccess::TransportRemoteAccess(
20 : const std::string tag, const HcclDispatcher dispatcher, const std::unique_ptr<NotifyPool>& notifyPool,
21 0 : const RemoteAccessPara& remoteAccessPara, const std::vector<MemRegisterAddr>& memRegistInfos, s32 deviceLogicId)
22 0 : : dispatcher_(dispatcher),
23 0 : notifyPool_(notifyPool),
24 0 : MemRegistInfos_(memRegistInfos),
25 0 : RemoteAccessPara_(remoteAccessPara),
26 0 : handle_(nullptr),
27 0 : ackNotify_(nullptr),
28 0 : access_(RA_ACCESS_LOCAL_WRITE | RA_ACCESS_REMOTE_WRITE | RA_ACCESS_REMOTE_READ),
29 0 : notifySize_(NOTIFY_BUFFER_SIZE),
30 0 : tag_(tag),
31 0 : timeout_(HCCL_LINK_TIME_OUT_S),
32 0 : deviceLogicId_(deviceLogicId)
33 : {
34 0 : instanceRef_[deviceLogicId_].Ref();
35 0 : }
36 :
37 0 : TransportRemoteAccess::~TransportRemoteAccess()
38 : {
39 0 : HCCL_DEBUG("~TransportRemoteAccess Enter!");
40 : HcclResult ret;
41 0 : struct MrInfoT mrInfo = {};
42 : /* 销毁本端mr */
43 0 : for (u32 idx = 0; idx < localRegMem_.size(); idx++) {
44 0 : mrInfo.addr = localRegMem_[idx];
45 0 : ret = HrtRaMrDereg(handle_, &mrInfo);
46 0 : if (ret != HCCL_SUCCESS) {
47 0 : HCCL_WARNING(
48 : "errNo[0x%016llx] in TransportRemoteAccess deconstruct, mr dereg failed. ", HCCL_ERROR_CODE(ret));
49 : }
50 : }
51 :
52 0 : ackNotify_ = nullptr;
53 0 : if (handle_ != nullptr) {
54 0 : ret = HrtRaQpDestroy(handle_);
55 0 : if (ret != HCCL_SUCCESS) {
56 0 : HCCL_WARNING(
57 : "errNo[0x%016llx] in TransportRemoteAccess deconstruct, qp destroy failed. ", HCCL_ERROR_CODE(ret));
58 : }
59 : }
60 0 : if (instanceRef_[deviceLogicId_].Unref() == 0) {
61 0 : std::unique_lock<std::mutex> lock(notifyValueMutex_[deviceLogicId_]);
62 0 : notifyValueMem_[deviceLogicId_].free();
63 0 : }
64 0 : HCCL_DEBUG("~TransportRemoteAccess Success!");
65 0 : }
66 0 : HcclResult TransportRemoteAccess::Init()
67 : {
68 0 : HCCL_DEBUG("TransportRemoteAccess Init start");
69 : // 创建QP操作句柄
70 0 : CHK_RET(CreateQp());
71 : // 本端host/device内存地址注册
72 0 : CHK_RET(MrRegister());
73 : // notify注册，用于rdma消息同步
74 0 : CHK_RET(NotifyRegister());
75 0 : CHK_RET(ConnectQp());
76 :
77 0 : HCCL_DEBUG("TransportRemoteAccess Init end");
78 0 : return HCCL_SUCCESS;
79 : }
80 :
81 0 : HcclResult TransportRemoteAccess::CreateQp()
82 : {
83 : // 创建qp handle, mode:普通qp
84 0 : HcclResult ret = HrtRaQpCreate(RemoteAccessPara_.nicRdmaHandle, QP_FLAG_RC, NORMAL_QP_MODE, handle_);
85 0 : CHK_PRT_RET(
86 : ret != HCCL_SUCCESS,
87 : HCCL_ERROR(
88 : "[Create][Qp]create qp mode failed, handle is null, "
89 : "localRank[%u], qpMode[%d]",
90 : RemoteAccessPara_.localRank, NORMAL_QP_MODE),
91 : HCCL_E_ROCE_CONNECT);
92 0 : CHK_RET(SetQpAttrQos(handle_));
93 : // 配置RDMA Timeout时间
94 0 : CHK_RET(SetQpAttrTimeOut(handle_));
95 : // 配置RDMA Retry Cnt重传次数
96 0 : CHK_RET(SetQpAttrRetryCnt(handle_));
97 0 : return HCCL_SUCCESS;
98 : }
99 :
100 0 : HcclResult TransportRemoteAccess::NotifyRegister()
101 : {
102 : // notify内存注册
103 0 : CHK_RET(SetLocalNotify());
104 : // 获取对端notify
105 0 : CHK_RET(GetRemoteNotifyInfo());
106 0 : return HCCL_SUCCESS;
107 : }
108 :
109 0 : HcclResult TransportRemoteAccess::MrRegister()
110 : {
111 0 : void* memPtr = nullptr;
112 0 : if (MemRegistInfos_.size() == 0) {
113 0 : HCCL_ERROR("[Register][Mr]local mem info to register is empty!");
114 0 : return HCCL_E_PARA;
115 : }
116 0 : struct MrInfoT mrInfo = {};
117 0 : mrInfo.access = access_;
118 0 : for (size_t idx = 0; idx < MemRegistInfos_.size(); idx++) {
119 0 : memPtr = reinterpret_cast<void*>(static_cast<uintptr_t>(MemRegistInfos_[idx].addr));
120 0 : mrInfo.addr = memPtr;
121 0 : mrInfo.size = MemRegistInfos_[idx].length;
122 :
123 0 : CHK_RET(HrtRaMrReg(handle_, &mrInfo));
124 0 : localRegMem_.push_back(memPtr);
125 : }
126 0 : return HCCL_SUCCESS;
127 : }
128 :
129 0 : HcclResult TransportRemoteAccess::SetLocalNotify()
130 : {
131 : // 申请ack notify，并发送至对端
132 0 : CHK_RET(CreateNotify());
133 : // 注册notify 内存信息
134 0 : CHK_RET(CreateNotifyValueBuffer());
135 0 : return HCCL_SUCCESS;
136 : }
137 :
138 0 : HcclResult TransportRemoteAccess::CreateNotify()
139 : {
140 0 : u64 offset = 0;
141 0 : u64 notifyBaseVa = 0; // notify寄存器虚拟地址
142 0 : u64 notifyTotalSize = 0;
143 :
144 : /* 申请Notify Group ID */
145 0 : RemoteRankInfo info(deviceLogicId_, RemoteAccessPara_.remoteRank);
146 0 : CHK_RET(SalGetBareTgid(&info.remotePid)); // 当前进程id
147 0 : CHK_RET(notifyPool_->Alloc(tag_, info, ackNotify_));
148 : // 设置remote id
149 0 : s64 recvId = 0xFFFFFFFF00000000 | (static_cast<s64>(info.remotePid) & 0xFFFFFFFF);
150 0 : CHK_RET(ackNotify_->Grant(recvId));
151 :
152 : /* 获取notify寄存器虚拟基地址、大小, 物理地址回传值为空 */
153 0 : CHK_RET(HrtRaGetNotifyBaseAddr(RemoteAccessPara_.nicRdmaHandle, ¬ifyBaseVa, ¬ifyTotalSize));
154 :
155 : /* 获取notify虚拟地址 */
156 0 : CHK_RET(ackNotify_->GetNotifyOffset(offset));
157 :
158 : // notify寄存器的虚拟地址与物理地址偏移相同，所以虚拟地址为虚拟基地址加偏移
159 0 : u64 notifyVa = notifyBaseVa + offset;
160 :
161 0 : HCCL_INFO(
162 : "notifyBaseVa=0x%llx, notifyTotalSize=0x%x, offset=0x%llx, notifyVa=0x%llx ", notifyBaseVa, notifyTotalSize,
163 : offset, notifyVa);
164 :
165 : /* notify地址注册为mr, 在roce驱动中注册 */
166 0 : ackNotifyMsg_.mrRegFlag = 0; // mem注册给网卡标志位
167 0 : ackNotifyMsg_.addr = reinterpret_cast<void*>(static_cast<uintptr_t>(notifyVa)); // 本端notify地址交换给对端
168 0 : ackNotifyMsg_.len = notifySize_;
169 0 : ackNotifyMsg_.offset = offset;
170 :
171 0 : return HCCL_SUCCESS;
172 : }
173 :
174 0 : HcclResult TransportRemoteAccess::GetRemoteNotifyInfo()
175 : {
176 0 : NotifyMsg mrMsg;
177 0 : s32 sRet = memset_s(&mrMsg, sizeof(NotifyMsg), 0, sizeof(NotifyMsg));
178 0 : CHK_PRT_RET(
179 : sRet != EOK,
180 : HCCL_ERROR(
181 : "[Get][NotifyInfo]errNo[0x%016llx]get remote addr, memory set 0 failed. "
182 : "params: destMaxSize[%zu], count[%zu]",
183 : HCCL_ERROR_CODE(HCCL_E_MEMORY), sizeof(NotifyMsg), sizeof(NotifyMsg)),
184 : HCCL_E_MEMORY);
185 :
186 0 : CHK_RET(hrtRaSocketBlockRecv(RemoteAccessPara_.socketFdhandle, &mrMsg, sizeof(NotifyMsg)));
187 0 : sRet = memcpy_s(&remoteNotifyDataMsg_, sizeof(NotifyMsg), &mrMsg, sizeof(NotifyMsg));
188 0 : CHK_PRT_RET(
189 : sRet != EOK,
190 : HCCL_ERROR(
191 : "[Get][NotifyInfo]errNo[0x%016llx] In TransportRemoteAccess get remote addr, memcpy failed. "
192 : "errorno[%d], params:destMaxSize[%zu],count[%zu]",
193 : HCCL_ERROR_CODE(HCCL_E_MEMORY), sRet, sizeof(NotifyMsg), sizeof(NotifyMsg)),
194 : HCCL_E_MEMORY);
195 0 : HCCL_INFO("recv success:len=%llu", mrMsg.len);
196 :
197 0 : return HCCL_SUCCESS;
198 : }
199 :
200 0 : HcclResult TransportRemoteAccess::CreateNotifyValueBuffer()
201 : {
202 0 : std::unique_lock<std::mutex> lock(notifyValueMutex_[deviceLogicId_]);
203 0 : if (notifyValueMem_[deviceLogicId_].ptr() == nullptr) {
204 0 : u64 notifyVaule = 1; // notify值写1表示record
205 0 : CHK_RET(DeviceMem::alloc(notifyValueMem_[deviceLogicId_], notifyValueSize_));
206 0 : HCCL_DEBUG("create notify value size[%u]", notifySize_);
207 :
208 0 : CHK_RET(hrtMemSyncCopy(
209 : notifyValueMem_[deviceLogicId_].ptr(), notifyValueMem_[deviceLogicId_].size(), ¬ifyVaule, notifySize_,
210 : HcclRtMemcpyKind::HCCL_RT_MEMCPY_KIND_HOST_TO_DEVICE));
211 : }
212 0 : lock.unlock();
213 0 : struct MrInfoT mrInfo = {};
214 0 : mrInfo.addr = notifyValueMem_[deviceLogicId_].ptr();
215 0 : mrInfo.size = notifySize_;
216 0 : mrInfo.access = access_;
217 0 : CHK_RET(HrtRaMrReg(handle_, &mrInfo));
218 : // 将notify buffer地址保存
219 0 : localRegMem_.push_back(notifyValueMem_[deviceLogicId_].ptr());
220 0 : NotifyMsg msg = {};
221 0 : msg.mrRegFlag = REG_VALID;
222 0 : msg.addr = notifyValueMem_[deviceLogicId_].ptr();
223 0 : msg.len = notifySize_;
224 :
225 : /* 发送mem消息给对端 */
226 0 : HcclResult ret = hrtRaSocketBlockSend(RemoteAccessPara_.socketFdhandle, &msg, sizeof(NotifyMsg));
227 0 : if (ret != HCCL_SUCCESS) { // 发送成功字节数与发送字节数不等，发送失败
228 0 : HCCL_ERROR("[Create][NotifyValueBuffer]send=%zu", sizeof(NotifyMsg));
229 0 : return HCCL_E_INTERNAL;
230 : }
231 :
232 0 : return HCCL_SUCCESS;
233 0 : }
234 0 : HcclResult TransportRemoteAccess::RemoteRead(const std::vector<HcomRemoteAccessAddrInfo>& addrInfos, Stream& stream)
235 : {
236 0 : CHK_SMART_PTR_NULL(stream);
237 0 : HCCL_INFO("TransportRemoteAccess RemoteRead begin");
238 0 : CHK_RET(RdmaDataTransport(addrInfos, RDMA_OP_READ));
239 :
240 : // 读取远端notify buffer，用于notify同步
241 0 : CHK_RET(ReadRemoteNotifyBuffer());
242 :
243 : // 等待TS把任务处理完成
244 0 : CHK_RET(LocalIpcNotify::Wait(
245 : stream, const_cast<HcclDispatcher>(dispatcher_), ackNotify_, INVALID_VALUE_STAGE, NOTIFY_INVALID_WAIT_TIME,
246 : RemoteAccessPara_.localRank, RemoteAccessPara_.remoteRank));
247 :
248 0 : HCCL_INFO("TransportRemoteAccess RemoteRead end");
249 0 : return HCCL_SUCCESS;
250 : }
251 :
252 0 : HcclResult TransportRemoteAccess::RemoteWrite(const std::vector<HcomRemoteAccessAddrInfo>& addrInfos, Stream& stream)
253 : {
254 0 : CHK_SMART_PTR_NULL(stream);
255 0 : HCCL_RUN_INFO("TransportRemoteAccess RemoteWrite begin, addressNum[%u]", addrInfos.size());
256 0 : CHK_RET(RdmaDataTransport(addrInfos, RDMA_OP_WRITE));
257 : // 读取远端notify buffer，用于notify同步
258 0 : CHK_RET(ReadRemoteNotifyBuffer());
259 : // 等待TS把任务处理完成
260 0 : CHK_RET(LocalIpcNotify::Wait(
261 : stream, const_cast<HcclDispatcher>(dispatcher_), ackNotify_, INVALID_VALUE_STAGE, NOTIFY_INVALID_WAIT_TIME,
262 : RemoteAccessPara_.localRank, RemoteAccessPara_.remoteRank));
263 :
264 0 : return HCCL_SUCCESS;
265 : }
266 :
267 0 : HcclResult TransportRemoteAccess::RdmaDataTransport(const std::vector<HcomRemoteAccessAddrInfo>& addrInfos, s32 rdmaOp)
268 : {
269 0 : if ((rdmaOp != RDMA_OP_WRITE) && (rdmaOp != RDMA_OP_READ)) {
270 0 : HCCL_ERROR("[Transport][RdmaData]invalid rdma op type, op:[%d]", rdmaOp);
271 0 : return HCCL_E_PARA;
272 : }
273 0 : CHK_PRT_RET(addrInfos.empty(), HCCL_ERROR("[Transport][RdmaData]addrInfos is empty!"), HCCL_E_PARA);
274 0 : u32 addressNum = addrInfos.size();
275 : // 构造wr信息
276 0 : std::vector<struct SendWrlistDataExt> wrVec(addressNum);
277 0 : std::vector<struct SendWrRsp> opRspVec(addressNum);
278 0 : struct SendWrlistDataExt* wr = wrVec.data();
279 0 : struct SendWrRsp* opRsp = opRspVec.data();
280 0 : struct SgList list = {};
281 0 : u64 length = addrInfos[0].length;
282 :
283 0 : HCCL_RUN_INFO("RdmaDataTransport begin, addressNum[%u], length[%u]", addressNum, length);
284 0 : for (size_t idx = 0; idx < addrInfos.size(); idx++) {
285 0 : list.addr = static_cast<u64>(static_cast<uintptr_t>(addrInfos[idx].localAddr));
286 0 : list.len = addrInfos[idx].length;
287 :
288 0 : wr[idx].memList = list;
289 0 : wr[idx].dstAddr = static_cast<u64>(static_cast<uintptr_t>(addrInfos[idx].remoteAddr));
290 0 : wr[idx].op = rdmaOp; /* RDMA_WRITE: 0 RDMA_READ: 4 */
291 0 : wr[idx].sendFlags = RA_SEND_SIGNALED;
292 : }
293 0 : u32 nowIdex = 0;
294 0 : u32 singleCompleteNum = 0;
295 0 : u32 sendNum = 0;
296 : HcclResult ret;
297 0 : u32 tryCount = 0;
298 0 : while (nowIdex < addressNum) {
299 0 : sendNum = addressNum - nowIdex;
300 0 : ret = HrtRaSendWrlistExt(handle_, wr, opRsp, sendNum, &singleCompleteNum);
301 0 : if (ret == HCCL_SUCCESS) {
302 0 : HCCL_INFO("dlRaSendWrlist success singleCompleteNum[%u], addressNum[%u]", singleCompleteNum, addressNum);
303 0 : return HCCL_SUCCESS;
304 0 : } else if (ret == ENOENT) { // 未完成发送，需重试
305 0 : nowIdex += singleCompleteNum;
306 0 : wr += singleCompleteNum;
307 0 : opRsp += singleCompleteNum;
308 0 : tryCount++;
309 0 : CHK_PRT_RET(
310 : tryCount > SEND_WRLIST_MAX_COUNT,
311 : HCCL_ERROR(
312 : "[Transport][RdmaData]dlRaSendWrlist count beyond maxnum[%u], completenum[%u]",
313 : SEND_WRLIST_MAX_COUNT, nowIdex),
314 : HCCL_E_NETWORK);
315 0 : continue;
316 : } else {
317 0 : HCCL_ERROR(
318 : "[Transport][RdmaData]In RdmaDataTransport, hrtRaSendWrlist failed. op[%d], ret[%d]", rdmaOp, ret);
319 0 : return HCCL_E_NETWORK;
320 : }
321 : }
322 :
323 0 : return HCCL_SUCCESS;
324 0 : }
325 :
326 0 : HcclResult TransportRemoteAccess::ReadRemoteNotifyBuffer()
327 : {
328 0 : HCCL_INFO("In TransportRemoteAccess ReadRemoteNotifyBuffer begin");
329 0 : struct SgList list = {};
330 0 : struct SendWr wr = {};
331 :
332 : // 构造wr信息
333 0 : list.addr = static_cast<u64>(reinterpret_cast<uintptr_t>(ackNotifyMsg_.addr));
334 0 : list.len = remoteNotifyDataMsg_.len;
335 :
336 0 : wr.bufList = &list;
337 0 : wr.bufNum = 1; /* 此处list只有一个，设置为1 */
338 0 : wr.dstAddr = static_cast<u64>(reinterpret_cast<uintptr_t>(remoteNotifyDataMsg_.addr));
339 0 : wr.op = RDMA_OP_READ; /* RDMA_WRITE: 0 */
340 0 : wr.sendFlag = RA_SEND_SIGNALED;
341 :
342 0 : struct SendWrRsp opRsp = {};
343 0 : CHK_RET(HrtRaSendWr(handle_, &wr, &opRsp));
344 0 : HCCL_INFO("In TransportRemoteAccess ReadRemoteNotifyBuffer end");
345 0 : return HCCL_SUCCESS;
346 : }
347 :
348 0 : HcclResult TransportRemoteAccess::ConnectQp()
349 : {
350 : // QP建链
351 0 : CHK_RET(HrtRaQpConnectAsync(handle_, RemoteAccessPara_.socketFdhandle));
352 :
353 0 : HCCL_INFO(
354 : "TransportRemoteAccess ConnectQp LocalRank[%u] "
355 : "RemoteRank[%u] LocalIp[%s]",
356 : RemoteAccessPara_.localRank, RemoteAccessPara_.remoteRank, RemoteAccessPara_.localIp.GetReadableAddress());
357 :
358 : // 查询QP建链是否成功
359 0 : s32 qpStatus = 0;
360 0 : auto startTime = std::chrono::steady_clock::now();
361 0 : HCCL_INFO("In link ibv, waiting for qp status ready...");
362 : while (true) {
363 0 : if ((std::chrono::steady_clock::now() - startTime) >= timeout_) {
364 0 : HCCL_ERROR("[Connect][Qp]get qp status timeout_=%lld s, qp_status=%d", timeout_, qpStatus);
365 0 : return HCCL_E_TIMEOUT;
366 : }
367 :
368 0 : s32 raRet = hrtGetRaQpStatus(handle_, &qpStatus);
369 0 : if ((!raRet) && (qpStatus == 1)) { // 为1时，qp 建链成功
370 0 : HCCL_INFO("GetRaQpStatus Success!");
371 0 : break;
372 : } else {
373 : // qp建链需要时间，获取qp状态直至超时
374 0 : SaluSleep(WAIT_US_COUNT);
375 : }
376 0 : }
377 0 : return HCCL_SUCCESS;
378 : }
379 : } // namespace hccl
|
