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 "host_cpu_roce_channel.h"
12 : #include "endpoint.h"
13 : #include "dpu_notify/dpu_notify_manager.h"
14 : #include "hcomm_res.h"
15 : #include "hcomm_c_adpt.h"
16 : #include "exception_handler.h"
17 : #include "cpu_roce_endpoint.h"
18 : #include "adapter_error_manager_pub.h"
19 :
20 : // Orion
21 : #include "orion_adapter_hccp.h"
22 : #include "orion_adpt_utils.h"
23 : #include "exchange_rdma_buffer_dto.h"
24 : #include "rdma_handle_manager.h"
25 : #include "exchange_rdma_conn_dto.h"
26 : #include "sal.h"
27 : #include "adapter_hccp.h"
28 : #include "binary_stream.h"
29 : #include "../../../../../legacy/ascend910/platform/resource/notify/notify_pool_impl.h"
30 : #include "../../../../../base_comm/resources/hccp/inc/network/hccp_common.h"
31 : #include "dlprof_function.h"
32 : #include "user_remote_mem_getter.h"
33 : #include "env_config/env_config.h"
34 :
35 : namespace hcomm {
36 : constexpr u32 MEM_BLOCK_SIZE = 128;
37 : constexpr uint16_t DEFAULT_LISTENING_PORT = 60001;
38 : constexpr u32 SEND_RQE_COUNT = 16;
39 :
40 117 : HostCpuRoceChannel::HostCpuRoceChannel(EndpointHandle endpointHandle, HcommChannelDesc channelDesc)
41 117 : : endpointHandle_(endpointHandle), channelDesc_(channelDesc) {}
42 :
43 184 : HostCpuRoceChannel::~HostCpuRoceChannel() {
44 : HcclResult ret;
45 :
46 117 : if (isHybridMode_ && connections_.size() != 0) {
47 1 : auto qpInfo = connections_[0]->GetQpInfo();
48 1 : struct MrInfoT mrInfo = {nullptr};
49 :
50 17 : for (uint32_t i = 0; i < hccl::MEM_TYPE_RESERVED; i++) {
51 16 : if (localMemMsg_[i].addr == nullptr) {
52 16 : continue;
53 : }
54 :
55 0 : mrInfo.addr = localMemMsg_[i].addr;
56 0 : ret = HrtRaMrDereg(qpInfo.qpHandle, &mrInfo);
57 0 : if (ret != HCCL_SUCCESS) {
58 0 : HCCL_INFO("[~HostCpuRoceChannel] Dereg mem, ret=%d, type=%d, addr:%p, lkey:%d, size:%llu, access:%d",
59 : ret, i, mrInfo.addr, mrInfo.lkey, mrInfo.size, mrInfo.access);
60 : }
61 :
62 0 : if (localMemMsg_[i].notifyId != INVALID_DPU_NOTIFY_ID) {
63 0 : delete[] (int8_t *)localMemMsg_[i].addr;
64 : }
65 0 : localMemMsg_[i].addr = nullptr;
66 :
67 0 : HCCL_INFO("[~HostCpuRoceChannel] Dereg mem, type=%d, addr:%p, lkey:%d, size:%llu, access:%d",
68 : i, mrInfo.addr, mrInfo.lkey, mrInfo.size, mrInfo.access);
69 : }
70 : }
71 :
72 117 : ret = DpuNotifyManager::GetInstance().FreeNotifyIds(notifyNum_, localDpuNotifyIds_);
73 117 : if (ret != HCCL_SUCCESS) {
74 5 : HCCL_ERROR("[HostCpuRoceChannel::~HostCpuRoceChannel] exception occurred, HcclResult=[%d]", ret);
75 : }
76 :
77 117 : if (channelDesc_.socket == nullptr && socket_ != nullptr) {
78 1 : SocketMgr::GetInstance(devicePhyId_).PutSocket(socketConfig_, socket_);
79 1 : socket_ = nullptr;
80 : }
81 184 : }
82 :
83 64 : HcclResult HostCpuRoceChannel::ParseInputParam()
84 : {
85 : // 1. 从 endpointHandle_，获得 localEp_ 和 rdmaHandle_
86 64 : CHK_PTR_NULL(endpointHandle_);
87 64 : HCCL_INFO("[HostCpuRoceChannel][%s] Start. endpointHandle[0x%llx]", __func__, reinterpret_cast<uint64_t>(endpointHandle_));
88 64 : Endpoint* localEpPtr = reinterpret_cast<Endpoint*>(endpointHandle_);
89 64 : localEp_ = localEpPtr->GetEndpointDesc();
90 64 : rdmaHandle_ = localEpPtr->GetRdmaHandle();
91 64 : CHK_PTR_NULL(rdmaHandle_);
92 :
93 : // 2. 从 channelDesc_，获得 remoteEp_, socket_ 和 notifyNum
94 64 : remoteEp_ = channelDesc_.remoteEndpoint;
95 64 : socket_ = reinterpret_cast<Hccl::Socket*>(channelDesc_.socket);
96 : // If HIXL, socket is nullptr for now, will be built later.
97 64 : notifyNum_ = channelDesc_.notifyNum;
98 :
99 64 : if (channelDesc_.exchangeAllMems) { // true for HIXL, false for HCCL
100 : // 3. Get memHandles from endpoint
101 1 : HCCL_INFO("[HostCpuRoceChannel][%s] exchangeAllMems == True. Get memHandles from endpoint.", __func__);
102 1 : std::shared_ptr<Hccl::LocalRdmaRmaBuffer> *memHandles = nullptr;
103 1 : uint32_t memHandleNum = 0;
104 1 : CHK_RET(static_cast<HcclResult>(HcommMemGetAllMemHandles(
105 : endpointHandle_, reinterpret_cast<void**>(&memHandles), &memHandleNum)));
106 1 : HCCL_INFO("[HostCpuRoceChannel][%s] Got memHandleNum[%u].", __func__, memHandleNum);
107 1 : for (uint32_t i = 0; i < memHandleNum; ++i) {
108 0 : std::shared_ptr<Hccl::LocalRdmaRmaBuffer> &localRdmaBuffer = memHandles[i];
109 0 : HCCL_INFO("[HostCpuRoceChannel][%s] Got memHandle No.%u: addr[0x%llx], size[0x%llx], memType[%d], memInfo[%s].",
110 : __func__, i, static_cast<unsigned long long>(localRdmaBuffer->GetAddr()), static_cast<unsigned long long>(localRdmaBuffer->GetSize()),
111 : static_cast<int>(localRdmaBuffer->GetBuf()->GetMemType()), localRdmaBuffer->GetBuf()->GetMemInfo().c_str());
112 0 : localRmaBuffers_.emplace_back(localRdmaBuffer.get());
113 : }
114 : } else {
115 : // 3. 从 channelDesc 的 memHandle，获得 bufs_
116 63 : HCCL_INFO("[HostCpuRoceChannel][%s] exchangeAllMems == false. Get memHandles from channelDesc.", __func__);
117 63 : CHK_PTR_NULL(channelDesc_.memHandles);
118 124 : for (uint32_t i = 0; i < channelDesc_.memHandleNum; ++i) {
119 62 : CHK_PTR_NULL(channelDesc_.memHandles[i]);
120 62 : auto* localRdmaBuffer = static_cast<Hccl::LocalRdmaRmaBuffer *>(channelDesc_.memHandles[i]);
121 62 : localRmaBuffers_.emplace_back(localRdmaBuffer);
122 : }
123 : }
124 :
125 63 : auto* localCpuRoceEpPtr = dynamic_cast<CpuRoceEndpoint *>(localEpPtr);
126 63 : if (localCpuRoceEpPtr == nullptr) {
127 1 : HCCL_ERROR("[HostCpuRoceChannel][%s] endpointHandle_ is not CpuRoceEndpoint.", __func__);
128 1 : return HCCL_E_INTERNAL;
129 : }
130 62 : CpuRoceEndpoint::Capabilities caps{};
131 62 : CHK_RET(localCpuRoceEpPtr->GetCapabilities(caps));
132 62 : maxMsgSize_ = caps.maxMsgSize;
133 62 : lbMax_ = caps.lbMax;
134 62 : constexpr uint64_t TWO_GB = 0x80000000ULL; // 2GB
135 62 : if (maxMsgSize_ > TWO_GB) {
136 0 : HCCL_RUN_WARNING("[HostCpuRoceChannel][%s] maxMsgSize_[0x%llx] exceeds 2GB, value may be incorrect.",
137 : __func__, maxMsgSize_);
138 : }
139 62 : HCCL_INFO("[HostCpuRoceChannel][%s] maxMsgSize_[0x%llx].", __func__, maxMsgSize_);
140 :
141 62 : return HCCL_SUCCESS;
142 : }
143 :
144 0 : HcclResult HostCpuRoceChannel::StartListen()
145 : {
146 0 : uint16_t port = channelDesc_.port;
147 0 : HCCL_INFO("[HostCpuRoceChannel::%s] Start. EndpointHandle[0x%llx], port[%u]", __func__, reinterpret_cast<uint64_t>(endpointHandle_), port);
148 0 : if (port == 0) {
149 0 : port = DEFAULT_LISTENING_PORT;
150 0 : HCCL_INFO("[HostCpuRoceChannel::%s] channelDesc port is 0, use default port [%u]", __func__, port);
151 : }
152 0 : CHK_RET(static_cast<HcclResult>(HcommEndpointStartListen(endpointHandle_, port, nullptr)));
153 0 : HCCL_INFO("[HostCpuRoceChannel::%s] SUCCESS. port[%u].", __func__, port);
154 0 : return HCCL_SUCCESS;
155 : }
156 :
157 62 : HcclResult HostCpuRoceChannel::BuildSocket()
158 : {
159 62 : if (socket_ != nullptr) {
160 61 : return HCCL_SUCCESS;
161 : }
162 1 : HCCL_INFO("[HostCpuRoceChannel::%s] socket ptr is NULL, rebuild Socket", __func__);
163 :
164 1 : Hccl::LinkData linkData = BuildDefaultLinkData();
165 1 : CHK_RET(EndpointDescPairToLinkData(localEp_, remoteEp_, linkData));
166 1 : HCCL_INFO("[HostCpuRoceChannel::%s] built linkData: %s", __func__, linkData.Describe().c_str());
167 1 : uint16_t port = channelDesc_.port;
168 1 : if (port == 0) {
169 1 : port = DEFAULT_LISTENING_PORT;
170 1 : HCCL_INFO("[HostCpuRoceChannel::%s] channelDesc port is 0, use default port [%u]", __func__, port);
171 : }
172 1 : std::string socketTag = (channelDesc_.channelName != nullptr)
173 3 : ? std::string(channelDesc_.channelName) : "AUTOMATIC_SOCKET_TAG";
174 1 : Hccl::SocketConfig socketConfig = (channelDesc_.role != HCOMM_SOCKET_ROLE_RESERVED)
175 1 : ? Hccl::SocketConfig(linkData, port, socketTag, channelDesc_.role == HCOMM_SOCKET_ROLE_SERVER)
176 1 : : Hccl::SocketConfig(linkData, port, socketTag);
177 1 : CHK_RET(SocketMgr::GetInstance(devicePhyId_).GetSocket(socketConfig, socket_));
178 1 : HCCL_INFO("[HostCpuRoceChannel::%s] SUCCESS. port[%u].", __func__, port);
179 1 : return HCCL_SUCCESS;
180 1 : }
181 :
182 62 : HcclResult HostCpuRoceChannel::BuildConnection()
183 : {
184 62 : u32 loopTimes = 0;
185 62 : if (lbMax_ > 0) {
186 0 : if (channelDesc_.roceAttr.queueNum == 1) {
187 0 : loopTimes = lbMax_;
188 : } else {
189 0 : loopTimes = channelDesc_.roceAttr.queueNum;
190 : }
191 : } else {
192 62 : loopTimes = channelDesc_.roceAttr.queueNum;
193 : }
194 :
195 67 : for (u32 i = 0; i < loopTimes; i++) {
196 5 : std::unique_ptr<HostRdmaConnection> conn;
197 5 : EXCEPTION_CATCH(
198 : conn = std::make_unique<HostRdmaConnection>(socket_, rdmaHandle_),
199 : return HCCL_E_INTERNAL);
200 5 : CHK_PTR_NULL(conn);
201 5 : CHK_RET(conn->Init());
202 5 : Hccl::QpInfo& qpInfo = conn->GetQpInfo();
203 5 : if (lbMax_ > 0) {
204 0 : qpInfo.lbValue = i % lbMax_;
205 : }
206 5 : qpInfo.serviceLevel = channelDesc_.roceAttr.sl;
207 5 : qpInfo.trafficClass = channelDesc_.roceAttr.tc;
208 5 : qpInfo.retryCnt = channelDesc_.roceAttr.retryCnt;
209 5 : qpInfo.retryInterval = channelDesc_.roceAttr.retryInterval;
210 5 : HCCL_INFO("[HostCpuRoceChannel::BuildConnection] QpInfo[%u]: lbValue[%u], serviceLevel[%u], trafficClass[%u], retryCnt[%u], retryInterval[%u].",
211 : i, qpInfo.lbValue, qpInfo.serviceLevel, qpInfo.trafficClass, qpInfo.retryCnt, qpInfo.retryInterval);
212 5 : connections_.emplace_back(std::move(conn));
213 5 : }
214 62 : connNum_ = connections_.size();
215 62 : wqeNums_.resize(connNum_, 0);
216 62 : HCCL_INFO("[HostCpuRoceChannel::BuildConnection] Success, Qp count = %u", connNum_);
217 62 : return HCCL_SUCCESS;
218 : }
219 :
220 62 : HcclResult HostCpuRoceChannel::BuildNotify()
221 : {
222 62 : CHK_RET(DpuNotifyManager::GetInstance().AllocNotifyIds(notifyNum_, localDpuNotifyIds_));
223 61 : return HCCL_SUCCESS;
224 : }
225 :
226 61 : HcclResult HostCpuRoceChannel::BuildBuffer()
227 : {
228 61 : bufferNum_ = localRmaBuffers_.size();
229 61 : return HCCL_SUCCESS;
230 : }
231 :
232 64 : HcclResult HostCpuRoceChannel::Init()
233 : {
234 : s32 devLogicId;
235 64 : CHK_RET(hrtGetDevice(&devLogicId));
236 64 : CHK_RET(hrtGetDevicePhyIdByIndex(static_cast<u32>(devLogicId), devicePhyId_));
237 :
238 64 : CHK_RET(ParseInputParam());
239 : // true for HIXL, false for HCCL
240 62 : if (channelDesc_.exchangeAllMems && channelDesc_.role != HCOMM_SOCKET_ROLE_CLIENT) {
241 0 : CHK_RET(StartListen());
242 : }
243 62 : CHK_RET(BuildSocket());
244 62 : CHK_RET(BuildConnection());
245 62 : CHK_RET(BuildNotify());
246 61 : CHK_RET(BuildBuffer());
247 :
248 61 : return HCCL_SUCCESS;
249 : }
250 :
251 : // 当前AICPU和框架没有改为返回错误码形式，所有暂时使用该方法转换
252 268 : ChannelStatus HostCpuRoceChannel::GetStatus()
253 : {
254 268 : ChannelStatus status;
255 268 : HcclResult ret = GetStatus(status);
256 268 : if (ret != HCCL_SUCCESS && ret != HCCL_E_AGAIN) {
257 8 : HCCL_ERROR("[HostCpuRoceChannel::GetStatus] get status exception occurred, HcclResult=[%d]", ret);
258 8 : return ChannelStatus::FAILED;
259 : }
260 260 : return status;
261 : }
262 :
263 262 : HcclResult HostCpuRoceChannel::ProcessStatus()
264 : {
265 262 : switch (channelStatus_) {
266 48 : case ChannelStatus::READY:
267 48 : return HCCL_SUCCESS;
268 2 : case ChannelStatus::SOCKET_TIMEOUT:
269 2 : HCCL_ERROR("[HostCpuRoceChannel::ProcessStatus] get socket timeout");
270 2 : return HCCL_E_ROCE_CONNECT;
271 212 : default:
272 212 : return HCCL_E_AGAIN;
273 : }
274 : }
275 :
276 268 : HcclResult HostCpuRoceChannel::GetStatus(ChannelStatus &status) {
277 268 : switch (rdmaStatus_) {
278 56 : case RdmaStatus::INIT:
279 : // 检查socket状态
280 56 : CHK_RET(CheckSocketStatus());
281 56 : break;
282 55 : case RdmaStatus::SOCKET_OK:
283 55 : CHK_RET(ExchangeCapability());
284 54 : rdmaStatus_ = RdmaStatus::CAP_EXCHANGED;
285 54 : break;
286 55 : case RdmaStatus::CAP_EXCHANGED:
287 : // 准备资源
288 55 : CHK_RET(CreateQp());
289 53 : rdmaStatus_ = RdmaStatus::QP_CREATED;
290 53 : break;
291 52 : case RdmaStatus::QP_CREATED:
292 : // 发送交换数据
293 52 : if (isHybridMode_) {
294 1 : CHK_RET(ExchangeDataHybird());
295 : } else {
296 51 : CHK_RET(ExchangeData());
297 : }
298 51 : rdmaStatus_ = RdmaStatus::DATA_EXCHANGE;
299 51 : break;
300 50 : case RdmaStatus::DATA_EXCHANGE:
301 50 : if (isHybridMode_) {
302 1 : CHK_RET(ConnectSingleQpHybrid([]() -> bool {return 0;}));
303 : } else {
304 49 : CHK_RET(ModifyQp());
305 : }
306 48 : rdmaStatus_ = RdmaStatus::QP_MODIFIED;
307 : // modify完就不需要再轮询状态了，直接向下走准备Rqe的流程。
308 : [[fallthrough]];
309 48 : case RdmaStatus::QP_MODIFIED:
310 : // Prepare Rqes
311 48 : if (!isHybridMode_) {
312 816 : for (uint32_t i = 0; i < SEND_RQE_COUNT; ++i) {
313 768 : CHK_RET(IbvPostRecv());
314 : }
315 : }
316 : [[fallthrough]];
317 : default:
318 48 : rdmaStatus_ = RdmaStatus::CONN_OK;
319 48 : channelStatus_ = ChannelStatus::READY;
320 : }
321 :
322 262 : status = channelStatus_;
323 262 : return ProcessStatus();
324 : }
325 :
326 56 : HcclResult HostCpuRoceChannel::CheckSocketStatus() {
327 56 : CHK_PTR_NULL(socket_);
328 56 : HCCL_DEBUG("[HostCpuRoceChannel::CheckSocketStatus] socket GetStatus start");
329 56 : Hccl::SocketStatus socketStatus = socket_->GetStatus(); // socket状态机
330 56 : HCCL_DEBUG("[HostCpuRoceChannel::CheckSocketStatus] socket status = %s", socketStatus.Describe().c_str());
331 56 : if (socketStatus == Hccl::SocketStatus::OK) {
332 54 : rdmaStatus_ = RdmaStatus::SOCKET_OK;
333 54 : channelStatus_ = ChannelStatus::SOCKET_OK;
334 2 : } else if (socketStatus == Hccl::SocketStatus::TIMEOUT) {
335 2 : channelStatus_ = ChannelStatus::SOCKET_TIMEOUT;
336 : }
337 56 : return HCCL_SUCCESS;
338 : }
339 :
340 : // 准备资源（创建QP）
341 55 : HcclResult HostCpuRoceChannel::CreateQp() {
342 59 : for (auto &conn : connections_) {
343 6 : Hccl::CHECK_NULLPTR(conn,
344 12 : Hccl::StringFormat("[HostCpuRoceChannel::%s] failed, connection pointer is nullptr", __func__));
345 6 : HcclResult ret = conn->CreateQp();
346 6 : if (ret == HCCL_E_AGAIN) {
347 2 : return HCCL_SUCCESS;
348 : }
349 6 : if (ret != HCCL_SUCCESS) {
350 2 : return ret;
351 : }
352 : }
353 53 : HCCL_INFO("[HostCpuRoceChannel::IsResReady] all connections resources connected.");
354 53 : return HCCL_SUCCESS;
355 : }
356 :
357 : // 交换数据
358 2 : HcclResult HostCpuRoceChannel::ExchangeData()
359 : {
360 2 : HCCL_INFO("[HostCpuRoceChannel::%s] Start to SendExchangeData, notifyNum=%u, bufferNum=%u, connNum=%u",
361 : __func__, notifyNum_, bufferNum_, connNum_);
362 :
363 : // 同步数据打包
364 2 : Hccl::BinaryStream binaryStream;
365 2 : NotifyVecPack(binaryStream);
366 2 : CHK_RET(BufferVecPack(binaryStream));
367 2 : CHK_RET(ConnVecPack(binaryStream));
368 :
369 2 : std::vector<char> sendData{};
370 2 : binaryStream.Dump(sendData);
371 2 : uint64_t sendSize = sendData.size();
372 2 : std::vector<char> recvData{};
373 2 : uint64_t recvSize = 0;
374 :
375 : EXCEPTION_HANDLE_BEGIN
376 : // 同步发送数据包尺寸
377 2 : CHK_PRT_RET(!socket_->Send(reinterpret_cast<void *>(&sendSize), sizeof(sendSize)),
378 : HCCL_ERROR("[HostCpuRoceChannel::%s] Send sendSize failed", __func__), HCCL_E_NETWORK);
379 2 : HCCL_INFO("[HostCpuRoceChannel::%s] Send size[%llu] of data success. [%llu] bytes sent.",
380 : __func__, sendSize, sizeof(sendSize));
381 :
382 : // 同步接收数据包尺寸
383 2 : CHK_PRT_RET(!socket_->Recv(reinterpret_cast<void *>(&recvSize), sizeof(recvSize)),
384 : HCCL_ERROR("[HostCpuRoceChannel::%s] Recv recvSize failed", __func__), HCCL_E_NETWORK);
385 2 : HCCL_INFO("[HostCpuRoceChannel::%s] Receive size[%llu] of data success. [%llu] bytes received.",
386 : __func__, recvSize, sizeof(recvSize));
387 :
388 : // 同步发送数据
389 2 : CHK_PRT_RET(!socket_->Send(reinterpret_cast<void *>(sendData.data()), sendSize),
390 : HCCL_ERROR("[HostCpuRoceChannel::%s] Send exchange data failed", __func__), HCCL_E_NETWORK);
391 2 : HCCL_INFO("[HostCpuRoceChannel::%s] Send Exchange Data success. [%llu] bytes sent.",
392 : __func__, sendSize);
393 :
394 : // 同步接收数据
395 2 : HCCL_INFO("[HostCpuRoceChannel::%s] Start to Receive Exchange Data", __func__);
396 2 : recvData.resize(recvSize);
397 2 : CHK_PRT_RET(!socket_->Recv(reinterpret_cast<void *>(recvData.data()), recvSize),
398 : HCCL_ERROR("[HostCpuRoceChannel::%s] Recv exchange data failed", __func__), HCCL_E_NETWORK);
399 2 : HCCL_INFO("[HostCpuRoceChannel::%s] Receive Exchange Data success. [%llu] bytes received.",
400 : __func__, recvSize);
401 0 : EXCEPTION_HANDLE_END
402 :
403 : // 同步数据解包
404 2 : Hccl::BinaryStream recvBinStream(recvData);
405 : // CHK_RET(HandshakeMsgUnpack(recvBinStream));
406 2 : CHK_RET(NotifyVecUnpack(recvBinStream));
407 2 : CHK_RET(RmtBufferVecUnpackProc(recvBinStream));
408 2 : CHK_RET(ConnVecUnpackProc(recvBinStream));
409 :
410 1 : HCCL_INFO("[HostCpuRoceChannel::%s] Unpack exchange Data success.", __func__);
411 1 : return HCCL_SUCCESS;
412 2 : }
413 :
414 1 : void HostCpuRoceChannel::NotifyVecPack(Hccl::BinaryStream &binaryStream)
415 : {
416 1 : binaryStream << notifyNum_;
417 1 : HCCL_INFO("start pack DpuRoceChannel notifyVec");
418 1 : u32 pos = 0;
419 5 : for (auto &it : localDpuNotifyIds_) {
420 4 : binaryStream << it;
421 4 : HCCL_INFO("pack notify pos=%u, notifyId=%u", pos, it);
422 4 : pos++;
423 : }
424 1 : }
425 :
426 1 : HcclResult HostCpuRoceChannel::BufferVecPack(Hccl::BinaryStream &binaryStream)
427 : {
428 1 : binaryStream << bufferNum_;
429 1 : HCCL_INFO("[HostCpuRoceChannel::%s] start to pack RmaBuffers", __func__);
430 1 : u32 pos = 0;
431 1 : for (auto &it : localRmaBuffers_) {
432 1 : binaryStream << pos;
433 1 : if (it != nullptr) { // 非空的buffer，从buffer中获取 dto
434 1 : std::unique_ptr<Hccl::Serializable> dto = it->GetExchangeDto();
435 1 : if (dto == nullptr) {
436 1 : return HCCL_E_INTERNAL;
437 : }
438 0 : dto->Serialize(binaryStream);
439 0 : HCCL_INFO("pack buffer pos=%u dto %s", pos, dto->Describe().c_str());
440 1 : } else { // 空的buffer，dto所有字段为0(size=0)
441 0 : Hccl::ExchangeRdmaBufferDto exchangeDto;
442 0 : exchangeDto.Serialize(binaryStream);
443 0 : HCCL_INFO("pack buffer pos=%u, dto is null %s", pos, exchangeDto.Describe().c_str());
444 0 : }
445 0 : pos++;
446 : }
447 0 : HCCL_INFO("[HostCpuRoceChannel::%s] pack RmaBuffers finish", __func__);
448 0 : return HCCL_SUCCESS;
449 : }
450 :
451 1 : HcclResult HostCpuRoceChannel::ConnVecPack(Hccl::BinaryStream &binaryStream)
452 : {
453 1 : binaryStream << connNum_;
454 1 : HCCL_INFO("[HostCpuRoceChannel::%s] start to pack connections", __func__);
455 1 : u32 pos = 0;
456 1 : binaryStream << channelDesc_.roceAttr.queueNum;
457 2 : for (auto &it : connections_) {
458 1 : binaryStream << pos;
459 :
460 1 : binaryStream << channelDesc_.roceAttr.retryCnt;
461 1 : binaryStream << channelDesc_.roceAttr.retryInterval;
462 1 : binaryStream << channelDesc_.roceAttr.sl;
463 1 : binaryStream << channelDesc_.roceAttr.tc;
464 :
465 1 : std::unique_ptr<Hccl::Serializable> dto = nullptr;
466 1 : CHK_RET(it->GetExchangeDto(dto));
467 1 : dto->Serialize(binaryStream);
468 1 : HCCL_INFO("pack connection pos=%u, dto %s", pos, dto->Describe().c_str());
469 1 : pos++;
470 1 : }
471 1 : HCCL_INFO("[HostCpuRoceChannel::%s] pack connections finish", __func__);
472 1 : return HCCL_SUCCESS;
473 : }
474 :
475 0 : HcclResult HostCpuRoceChannel::RmtBufferVecUnpackProc(Hccl::BinaryStream &binaryStream)
476 : {
477 : u32 rmtNum;
478 0 : binaryStream >> rmtNum;
479 :
480 0 : HCCL_INFO("[HostCpuRoceChannel::%s] bufferNum_=%u, rmtNum=%u", __func__, bufferNum_, rmtNum);
481 :
482 0 : rmtRmaBuffers_.resize(rmtNum);
483 0 : for (u32 i = 0; i < rmtNum; i++) {
484 : u32 pos;
485 0 : binaryStream >> pos;
486 0 : if (pos >= rmtNum) {
487 0 : HCCL_ERROR("[HostCpuRoceChannel::%s] pos=%u out of range (rmtNum=%u)", __func__, pos, rmtNum);
488 0 : return HCCL_E_INTERNAL;
489 : }
490 0 : Hccl::ExchangeRdmaBufferDto dto;
491 0 : dto.Deserialize(binaryStream);
492 :
493 0 : HCCL_INFO("[HostCpuRoceChannel::%s] pos=%u, dto %s", __func__, pos, dto.Describe().c_str());
494 0 : EXCEPTION_CATCH(rmtRmaBuffers_[pos] = std::make_unique<Hccl::RemoteRdmaRmaBuffer>(rdmaHandle_, dto),
495 : HCCL_ERROR("[HostCpuRoceChannel::%s] make_unique<Hccl::RemoteRdmaRmaBuffer> throws an exception!", __func__);
496 : return HCCL_E_INTERNAL);
497 0 : HCCL_INFO("[HostCpuRoceChannel::%s] pos=%u, rmtRmaBuffer=%s", __func__, pos, rmtRmaBuffers_[pos]->Describe().c_str());
498 0 : }
499 :
500 0 : return HCCL_SUCCESS;
501 : }
502 :
503 1 : HcclResult HostCpuRoceChannel::NotifyVecUnpack(Hccl::BinaryStream &binaryStream)
504 : {
505 1 : uint32_t notifySize = 0;
506 1 : binaryStream >> notifySize;
507 1 : if (notifySize != notifyNum_) {
508 0 : HCCL_ERROR("[HostCpuRoceChannel::NotifyVecUnpack] rmtNum=%u is not equal to localNum=%u", notifySize, notifyNum_);
509 0 : return HCCL_E_ROCE_CONNECT;
510 : }
511 1 : remoteDpuNotifyIds_.clear();
512 1 : u32 pos = 0;
513 5 : for (pos = 0; pos < notifySize; pos++) {
514 : uint32_t notifyId;
515 4 : binaryStream >> notifyId;
516 4 : remoteDpuNotifyIds_.push_back(notifyId);
517 : }
518 1 : HCCL_INFO("[HostCpuRoceChannel::NotifyVecUnpack] unpack dpuNotify");
519 1 : return HCCL_SUCCESS;
520 : }
521 :
522 1 : HcclResult HostCpuRoceChannel::ConnVecUnpackProc(Hccl::BinaryStream &binaryStream)
523 : {
524 : u32 rmtConnNum;
525 1 : binaryStream >> rmtConnNum;
526 1 : HCCL_INFO("start unpack conn, connNum=%u, rmtConnNum=%u", connNum_, rmtConnNum);
527 1 : if (connNum_ != rmtConnNum) {
528 0 : HCCL_ERROR("connNum=%u is not equal to rmtConnNum=%u", connNum_, rmtConnNum);
529 0 : return HCCL_E_ROCE_CONNECT;
530 : }
531 :
532 1 : uint32_t localQpNum = channelDesc_.roceAttr.queueNum;
533 1 : binaryStream >> channelDesc_.roceAttr.queueNum;
534 1 : if (localQpNum != channelDesc_.roceAttr.queueNum) {
535 0 : HCCL_ERROR("localQpNum[%u] is not equal to remoteQpNum[%u]", localQpNum, channelDesc_.roceAttr.queueNum);
536 0 : return HCCL_E_ROCE_CONNECT;
537 : }
538 :
539 1 : rmtConnDtos_.resize(rmtConnNum);
540 2 : for (u32 i = 0; i < rmtConnNum; i++) {
541 : u32 pos;
542 1 : binaryStream >> pos;
543 1 : binaryStream >> channelDesc_.roceAttr.retryCnt;
544 1 : binaryStream >> channelDesc_.roceAttr.retryInterval;
545 1 : binaryStream >> channelDesc_.roceAttr.sl;
546 1 : binaryStream >> channelDesc_.roceAttr.tc;
547 1 : rmtConnDtos_[i].Deserialize(binaryStream);
548 : }
549 :
550 1 : return HCCL_SUCCESS;
551 : }
552 :
553 49 : HcclResult HostCpuRoceChannel::ModifyQp() {
554 51 : for (uint32_t i = 0; i < connections_.size(); i++) {
555 3 : auto &conn = connections_[i];
556 3 : Hccl::CHECK_NULLPTR(conn,
557 6 : Hccl::StringFormat("[HostCpuRoceChannel::%s] failed, connection pointer is nullptr", __func__));
558 3 : CHK_RET(conn->ParseRmtExchangeDto(rmtConnDtos_[i]));
559 3 : Hccl::QpInfo& qpInfo = conn->GetQpInfo();
560 3 : qpInfo.serviceLevel = channelDesc_.roceAttr.sl;
561 3 : qpInfo.trafficClass = channelDesc_.roceAttr.tc;
562 3 : qpInfo.retryCnt = channelDesc_.roceAttr.retryCnt;
563 3 : qpInfo.retryInterval = channelDesc_.roceAttr.retryInterval;
564 3 : HCCL_INFO("[HostCpuRoceChannel::ModifyQp] QpInfo: serviceLevel[%u], trafficClass[%u], retryCnt[%u], retryInterval[%u].",
565 : qpInfo.serviceLevel, qpInfo.trafficClass, qpInfo.retryCnt, qpInfo.retryInterval);
566 3 : HcclResult ret = conn->ModifyQp();
567 3 : if (ret == HCCL_E_AGAIN) {
568 0 : return HCCL_SUCCESS;
569 : }
570 3 : if (ret != HCCL_SUCCESS) {
571 1 : return ret;
572 : }
573 : }
574 48 : HCCL_INFO("[HostCpuRoceChannel::IsResReady] all connections resources connected.");
575 48 : return HCCL_SUCCESS;
576 : }
577 :
578 5 : HcclResult HostCpuRoceChannel::GetRemoteMems(uint32_t *memNum, CommMem **remoteMem, char ***memInfos)
579 : {
580 5 : std::lock_guard<std::mutex> lock(remoteMemsMutex_);
581 5 : Hccl::RemoteMemCtx<std::unique_ptr<Hccl::RemoteRdmaRmaBuffer>> remoteMemCtx{cacheValid_, rmtRmaBuffers_,
582 5 : userRemoteMems_, memInfoCopies_, memInfoPointers_, remoteMem, memInfos, memNum};
583 5 : CHK_RET(GetRemoteUserMems(remoteMemCtx));
584 4 : return HCCL_SUCCESS;
585 5 : }
586 :
587 0 : std::vector<Hccl::QpInfo> HostCpuRoceChannel::GetQpInfos() const
588 : {
589 0 : std::vector<Hccl::QpInfo> qpInfos;
590 0 : for (auto& rdmaConn : connections_) {
591 0 : qpInfos.emplace_back(rdmaConn->GetQpInfo());
592 : }
593 0 : return qpInfos;
594 0 : }
595 :
596 1 : std::string HostCpuRoceChannel::Describe() const
597 : {
598 1 : std::string msg = "HostCpuRoceChannel{";
599 1 : msg += Hccl::StringFormat("notifyNum:%u, dpuNotifyList:[-]", notifyNum_);
600 1 : msg += Hccl::StringFormat(", bufferNum:%u, localRmaBuffers: [", bufferNum_);
601 3 : for (auto& buf : localRmaBuffers_) {
602 2 : msg += buf->Describe();
603 2 : msg += ", ";
604 : }
605 1 : msg += Hccl::StringFormat("], connNum:%u, connections:[", connNum_);
606 1 : for (auto& conn : connections_) {
607 0 : msg += conn->Describe();
608 0 : msg += ", ";
609 : }
610 1 : msg += Hccl::StringFormat("], rdmaHandle: %p, %s, ", rdmaHandle_, channelStatus_.Describe().c_str());
611 :
612 1 : if (socket_ != nullptr) {
613 1 : msg += socket_->Describe();
614 : }
615 :
616 1 : msg += ", ";
617 : // msg += attr_.Describe();
618 1 : return msg;
619 0 : }
620 :
621 1 : HcclResult HostCpuRoceChannel::SetDfxCallback(std::function<HcclResult(const Hccl::TaskParam&, u64)> callback)
622 : {
623 1 : dfxCallback_ = callback;
624 1 : return HCCL_SUCCESS;
625 : }
626 :
627 0 : HcclResult HostCpuRoceChannel::IbvPostRecv() const {
628 0 : std::vector<Hccl::QpInfo> qpInfo = GetQpInfos();
629 0 : CHK_PRT_RET(qpInfo.empty(), HCCL_ERROR("[HostCpuRoceChannel::%s] qpInfos is Empty", __func__), HCCL_E_ROCE_CONNECT);
630 0 : CHK_PRT_RET(localRmaBuffers_.empty(), HCCL_ERROR("[HostCpuRoceChannel::%s] localRmaBuffer is Empty", __func__),
631 : HCCL_E_ROCE_CONNECT);
632 0 : CHK_PRT_RET(rmtRmaBuffers_.empty(), HCCL_ERROR("[HostCpuRoceChannel::%s] rmtRmaBuffers is Empty", __func__),
633 : HCCL_E_ROCE_CONNECT);
634 :
635 : // 准备wr
636 0 : HCCL_INFO("[HostCpuRoceChannel::%s] call ibv_post_recv", __func__);
637 0 : for (uint32_t i = 0; i < qpInfo.size(); i++) {
638 0 : ibv_recv_wr recvWr {};
639 0 : ibv_recv_wr *recvbadWr = nullptr;
640 0 : ibv_sge recvsgList {};
641 0 : recvsgList.addr = localRmaBuffers_[0]->GetBufferInfo().first + MEM_BLOCK_SIZE * i; // 本端起始地址，cclbuffer最小为1MB，足够使用
642 0 : recvsgList.length = MEM_BLOCK_SIZE;
643 0 : recvsgList.lkey = localRmaBuffers_[0]->GetLkey(); // 本端的访问秘钥
644 0 : recvWr.wr_id = i;
645 0 : recvWr.sg_list = &recvsgList;
646 0 : recvWr.next = nullptr;
647 0 : recvWr.num_sge = 1;
648 :
649 0 : HCCL_INFO("qp_state[%u] = [%u]", i, qpInfo[i].qp->state);
650 0 : int32_t ret = ibv_post_recv(qpInfo[i].qp, &recvWr, &recvbadWr);
651 0 : CHK_PRT_RET(ret == ENOMEM,
652 : HCCL_WARNING("[HostCpuRoceChannel][%s] post recv wqe overflow. ret:%d, "
653 : "badWr->wr_id[%llu], badWr->sg_list->addr[%llu]",
654 : __func__, ret, recvbadWr->wr_id, recvbadWr->sg_list->addr),
655 : HCCL_E_AGAIN);
656 :
657 0 : CHK_PRT_RET(ret != 0,
658 : HCCL_ERROR("[HostCpuRoceChannel][%s] ibv_post_recv failed. ret:%d, "
659 : "badWr->wr_id[%llu], badWr->sg_list->addr[%llu]",
660 : __func__, ret, recvbadWr->wr_id, recvbadWr->sg_list->addr),
661 : HCCL_E_NETWORK);
662 : }
663 :
664 0 : return HCCL_SUCCESS;
665 0 : }
666 :
667 3 : HcclResult HostCpuRoceChannel::PrepareNotifyWrResource(
668 : uint32_t qpIdx, const uint64_t len, const uint32_t remoteNotifyIdx, struct ibv_send_wr ¬ifyRecordWr, Hccl::TaskParam &taskParam) const
669 : {
670 3 : taskParam.beginTime = hccl::DlProfFunction::GetInstance().dlMsprofSysCycleTime();
671 3 : if (remoteNotifyIdx >= remoteDpuNotifyIds_.size()) {
672 2 : HCCL_ERROR("[HostCpuRoceChannel::%s] remoteNotifyIdx[%u] out of the range of remoteDpuNotifyIds_[%zu].",
673 : __func__, remoteNotifyIdx, remoteDpuNotifyIds_.size());
674 2 : return HCCL_E_PARA;
675 : }
676 1 : uint32_t dpuNotifyId = remoteDpuNotifyIds_[remoteNotifyIdx];
677 :
678 1 : CHK_PRT_RET(localRmaBuffers_.empty(), HCCL_ERROR("[HostCpuRoceChannel::%s] localRmaBuffer is Empty", __func__),
679 : HCCL_E_ROCE_CONNECT);
680 1 : CHK_PRT_RET(rmtRmaBuffers_.empty(), HCCL_ERROR("[HostCpuRoceChannel::%s] rmtRmaBuffers is Empty", __func__),
681 : HCCL_E_ROCE_CONNECT);
682 :
683 : // 构造send_WR
684 1 : notifyRecordWr.sg_list->addr = localRmaBuffers_[0]->GetBufferInfo().first; // 本端起始地址
685 1 : notifyRecordWr.sg_list->length = 0; // 取的本端长度
686 1 : notifyRecordWr.sg_list->lkey = localRmaBuffers_[0]->GetLkey(); // 本端的访问秘钥
687 1 : notifyRecordWr.opcode = IBV_WR_RDMA_WRITE_WITH_IMM;
688 1 : notifyRecordWr.send_flags = IBV_SEND_SIGNALED;
689 1 : notifyRecordWr.imm_data = dpuNotifyId;
690 1 : notifyRecordWr.next = nullptr;
691 1 : notifyRecordWr.num_sge = 1;
692 1 : notifyRecordWr.wr_id = qpIdx; // 用户定义工作请求id，建议：设为有意义的值
693 1 : notifyRecordWr.wr.rdma.rkey = rmtRmaBuffers_[0]->GetRkey(); // 远端秘钥
694 1 : notifyRecordWr.wr.rdma.remote_addr = static_cast<uint64_t>(rmtRmaBuffers_[0]->GetAddr()); // 远端地址
695 :
696 1 : taskParam.taskType = Hccl::TaskParamType::TASK_DPU_INLINE_WRITE;
697 1 : taskParam.taskPara.DMA.dst = reinterpret_cast<void *>(static_cast<uint64_t>(rmtRmaBuffers_[0]->GetAddr()));
698 1 : taskParam.taskPara.DMA.size = len;
699 1 : taskParam.taskPara.DMA.notifyID = dpuNotifyId;
700 1 : taskParam.taskPara.DMA.notifyValue = 1;
701 1 : taskParam.taskPara.DMA.linkType = Hccl::DfxLinkType::ROCE;
702 1 : taskParam.taskPara.DMA.dmaOp = Hccl::DmaOp::HCCL_DMA_WRITE;
703 1 : return HCCL_SUCCESS;
704 : }
705 :
706 9 : hccl::MemType HostCpuRoceChannel::NotifyIdToMemtypeHybird(uint32_t remoteNotifyIdx)
707 : {
708 9 : if (remoteNotifyIdx == 0) {
709 5 : return hccl::MemType::ACK_NOTIFY_MEM;
710 : } else {
711 4 : return hccl::MemType::DATA_NOTIFY_MEM;
712 : }
713 :
714 : return hccl::MemType::DATA_NOTIFY_MEM;
715 : }
716 :
717 1 : HcclResult HostCpuRoceChannel::BuildNotifyWrHybird(const uint32_t remoteNotifyIdx, struct ibv_send_wr ¬ifRecordWr)
718 : {
719 1 : hccl::MemType type = NotifyIdToMemtypeHybird(remoteNotifyIdx);
720 :
721 1 : notifRecordWr.sg_list->addr = reinterpret_cast<uint64_t>(localMemMsg_[hccl::NOTIFY_SRC_MEM].addr);
722 1 : notifRecordWr.sg_list->length = localMemMsg_[hccl::NOTIFY_SRC_MEM].len;
723 1 : notifRecordWr.sg_list->lkey = localMemMsg_[hccl::NOTIFY_SRC_MEM].lkey;
724 1 : notifRecordWr.opcode = IBV_WR_RDMA_WRITE;
725 1 : notifRecordWr.send_flags = IBV_SEND_SIGNALED;
726 1 : notifRecordWr.next = nullptr;
727 1 : notifRecordWr.num_sge = 1;
728 1 : notifRecordWr.wr_id = 0;
729 1 : notifRecordWr.wr.rdma.rkey = remoteMemMsg_[type].lkey;
730 1 : notifRecordWr.wr.rdma.remote_addr = reinterpret_cast<uint64_t>(remoteMemMsg_[type].addr);
731 :
732 1 : return HCCL_SUCCESS;
733 : }
734 :
735 2 : HcclResult HostCpuRoceChannel::NotifyRecord(const uint32_t remoteNotifyIdx)
736 : {
737 : // 1.构造send_WR
738 2 : struct ibv_send_wr notifyRecordWr {};
739 2 : struct ibv_send_wr *sendbadWr = nullptr;
740 2 : struct ibv_sge sgList {};
741 2 : notifyRecordWr.sg_list = &sgList;
742 2 : Hccl::TaskParam taskParam{};
743 :
744 2 : std::vector<Hccl::QpInfo> qpInfo = GetQpInfos();
745 2 : CHK_PRT_RET(qpInfo.empty(), HCCL_ERROR("[HostCpuRoceChannel::%s] qpInfos is Empty", __func__), HCCL_E_ROCE_CONNECT);
746 :
747 : // 3.调用ibv_post_send
748 2 : for (uint32_t i = 0; i < qpInfo.size(); i++) {
749 2 : HCCL_INFO("[HostCpuRoceChannel::%s] call ibv_post_send, qp_state[%u] = [%u]", __func__, i, qpInfo[i].qp->state);
750 2 : if (isHybridMode_) {
751 0 : BuildNotifyWrHybird(remoteNotifyIdx, notifyRecordWr);
752 : } else {
753 2 : CHK_RET(PrepareNotifyWrResource(i, MEM_BLOCK_SIZE, remoteNotifyIdx, notifyRecordWr, taskParam));
754 : }
755 0 : int32_t ret = ibv_post_send(qpInfo[i].qp, ¬ifyRecordWr, &sendbadWr);
756 0 : if (ret != 0 && sendbadWr == nullptr) {
757 0 : HCCL_ERROR("[HostCpuRoceChannel::%s] ibv_post_send failed while badWr is nullptr", __func__);
758 0 : return HCCL_E_INTERNAL;
759 : }
760 0 : CHK_PRT_RET(ret == ENOMEM,
761 : HCCL_WARNING("[HostCpuRoceChannel][%s] post send wqe overflow. ret:%d, badWr->wr_id[%llu], "
762 : "badWr->sg_list->addr[%llu], badWr->wr.rdma.remote_addr[%llu], badWr->wr.ud.remote_qpn[%u]",
763 : __func__, ret, sendbadWr->wr_id, sendbadWr->sg_list->addr, sendbadWr->wr.rdma.remote_addr, sendbadWr->wr.ud.remote_qpn),
764 : HCCL_E_AGAIN);
765 :
766 0 : CHK_PRT_RET(ret != 0,
767 : HCCL_ERROR("[HostCpuRoceChannel][%s] ibv_post_send failed. ret:%d, badWr->wr_id[%llu], "
768 : "badWr->sg_list->addr[%llu], badWr->wr.rdma.remote_addr[%llu], badWr->wr.ud.remote_qpn[%u]",
769 : __func__, ret, sendbadWr->wr_id, sendbadWr->sg_list->addr, sendbadWr->wr.rdma.remote_addr, sendbadWr->wr.ud.remote_qpn),
770 : HCCL_E_NETWORK);
771 0 : if (wqeNums_[i] == INT32_MAX) {
772 0 : HCCL_ERROR("[HostCpuRoceChannel::%s] wqeNums_[%u] has reached the maximum value of uint32_t.", __func__, i);
773 0 : return HCCL_E_INTERNAL;
774 : }
775 0 : wqeNums_[i]++;
776 0 : HCCL_INFO("[HostCpuRoceChannel::NotifyRecord] NotifyRecord end, wqeNums_[%u]=%d", i, wqeNums_[i]);
777 : }
778 :
779 0 : taskParam.endTime = hccl::DlProfFunction::GetInstance().dlMsprofSysCycleTime();
780 0 : if (dfxCallback_ != nullptr) {
781 0 : return dfxCallback_(taskParam, reinterpret_cast<u64>(this));
782 : }
783 0 : return HCCL_SUCCESS;
784 2 : }
785 :
786 3 : HcclResult HostCpuRoceChannel::NotifyWait(const uint32_t localNotifyIdx, const uint32_t timeout)
787 : {
788 3 : Hccl::TaskParam taskParam{};
789 3 : taskParam.beginTime = hccl::DlProfFunction::GetInstance().dlMsprofSysCycleTime();
790 :
791 3 : if (isHybridMode_) {
792 0 : return NotifyWaitHybrid(localNotifyIdx, timeout);
793 : }
794 :
795 3 : CHK_PRT_RET(localNotifyIdx >= localDpuNotifyIds_.size(), HCCL_ERROR("[HostCpuRoceChannel::%s] localNotifyIdx[%u] out of the range of localDpuNotifyIds_[%zu].",
796 : __func__, localNotifyIdx, localDpuNotifyIds_.size()), HCCL_E_PARA);
797 :
798 3 : uint32_t dpuNotifyId = localDpuNotifyIds_[localNotifyIdx];
799 :
800 : // 1. 准备WR
801 3 : struct ibv_wc wc{};
802 3 : std::lock_guard<std::mutex> lock(cq_mutex);
803 3 : std::vector<Hccl::QpInfo> qpInfo = GetQpInfos();
804 3 : CHK_PRT_RET(qpInfo.empty(), HCCL_ERROR("[HostCpuRoceChannel::%s] qpInfos is Empty", __func__), HCCL_E_ROCE_CONNECT);
805 3 : HCCL_INFO("[HostCpuRoceChannel::NotifyWait] poll recvCq = %p, localNotifyIdx = %u, notifyId = %u.",
806 : qpInfo[0].recvCq, localNotifyIdx, dpuNotifyId);
807 :
808 : // 2.轮询rq_cq
809 3 : auto startTime = std::chrono::steady_clock::now();
810 3 : auto waitTime = std::chrono::duration_cast<std::chrono::nanoseconds>(std::chrono::milliseconds(timeout));
811 3 : for (uint32_t i = 0; i < qpInfo.size(); i++) {
812 3 : CHK_PRT_RET(qpInfo[i].recvCq == nullptr, HCCL_ERROR("[HostCpuRoceChannel::%s] recvCq[%u] is null", __func__, i), HCCL_E_INTERNAL);
813 3 : CHK_PRT_RET(qpInfo[i].qp == nullptr, HCCL_ERROR("[HostCpuRoceChannel::%s] qp[%u] is null", __func__, i), HCCL_E_INTERNAL);
814 3 : CHK_PRT_RET(qpInfo[i].recvCq->context == nullptr, HCCL_ERROR("[HostCpuRoceChannel::%s] recvCq[%u]->context is null", __func__, i), HCCL_E_INTERNAL);
815 :
816 : while (true) {
817 0 : auto actualNum = ibv_poll_cq(qpInfo[i].recvCq, 1, &wc);
818 0 : CHK_PRT_RET(actualNum < 0, HCCL_ERROR("[HostCpuRoceChannel::%s] ibv_poll_cq err. actualNum=%d", __func__, actualNum),
819 : HCCL_E_NETWORK);
820 :
821 0 : if (actualNum > 0 && wc.imm_data == dpuNotifyId) {
822 0 : if (wc.status != IBV_WC_SUCCESS) {
823 0 : HCCL_ERROR("[HostCpuRoceChannel][%s] ibv_poll_cq return wc.status[%d], wc.opcode[%d], wc.vendorErr[%u], wc.byteLen[%u], wc.wcFlags[%u], wc.sl[%u], qpInfo[%u].qp->qp_num[%u]",
824 : __func__, wc.status, wc.opcode, wc.vendor_err, wc.byte_len, wc.wc_flags, wc.sl, i, qpInfo[i].qp->qp_num);
825 0 : return ReportWcStatusError(wc.status);
826 : }
827 0 : HCCL_INFO("[HostCpuRoceChannel::NotifyWait] poll cq success");
828 0 : break;
829 0 : } else if (actualNum > 0) {
830 0 : CHK_PRT_RET(true, HCCL_ERROR("[HostCpuRoceChannel::%s] polled cq unexpected. imm_data[%u] != dpuNotifyId[%u]",
831 : __func__, wc.imm_data, dpuNotifyId), HCCL_E_NETWORK);
832 : }
833 :
834 0 : if ((std::chrono::steady_clock::now() - startTime) >= waitTime) {
835 0 : CHK_PRT_RET(true, HCCL_ERROR("[HostCpuRoceChannel][%s] call ibv_poll_cq timeout. actualNum=%d", __func__, actualNum),
836 : HCCL_E_TIMEOUT);
837 : }
838 0 : }
839 : }
840 :
841 0 : CHK_RET(IbvPostRecv());
842 0 : taskParam.taskType = Hccl::TaskParamType::TASK_DPU_NOTIFY_WAIT;
843 0 : taskParam.taskPara.Notify.notifyID = dpuNotifyId;
844 0 : taskParam.taskPara.Notify.value = 1;
845 0 : taskParam.endTime = hccl::DlProfFunction::GetInstance().dlMsprofSysCycleTime();
846 0 : if (dfxCallback_ != nullptr) {
847 0 : return dfxCallback_(taskParam, reinterpret_cast<u64>(this));
848 : }
849 0 : return HCCL_SUCCESS;
850 3 : }
851 :
852 6 : HcclResult HostCpuRoceChannel::ReportWcStatusError(enum ibv_wc_status status)
853 : {
854 6 : Hccl::IpAddress localIp, remoteIp;
855 6 : (void)CommAddrToIpAddress(localEp_.commAddr, localIp);
856 6 : (void)CommAddrToIpAddress(remoteEp_.commAddr, remoteIp);
857 132 : RPT_INPUT_ERR(true, "EI0013", std::vector<std::string>({"localServerId", "localDeviceId", "localDeviceIp",
858 : "remoteServerId", "remoteDeviceId", "remoteDeviceIp"}),
859 : std::vector<std::string>({std::to_string(localEp_.loc.device.serverIdx), std::to_string(localEp_.loc.device.devPhyId),
860 : localIp.GetIpStr(), std::to_string(remoteEp_.loc.device.serverIdx),
861 : std::to_string(remoteEp_.loc.device.devPhyId), remoteIp.GetIpStr()}));
862 6 : return HCCL_E_NETWORK;
863 6 : }
864 :
865 2 : HcclResult HostCpuRoceChannel::PrepareWriteWrResource(const void *dst, const void *src, const uint64_t len,
866 : const uint32_t remoteNotifyIdx, struct ibv_send_wr &writeWithNotifyWr, Hccl::TaskParam &taskParam) const
867 : {
868 2 : taskParam.beginTime = hccl::DlProfFunction::GetInstance().dlMsprofSysCycleTime();
869 2 : if (remoteNotifyIdx >= remoteDpuNotifyIds_.size()) {
870 1 : HCCL_ERROR("[HostCpuRoceChannel::%s] remoteNotifyIdx[%u] out of the range of remoteDpuNotifyIds_[%zu].",
871 : __func__, remoteNotifyIdx, remoteDpuNotifyIds_.size());
872 1 : return HCCL_E_PARA;
873 : }
874 1 : uint32_t dpuNotifyId = remoteDpuNotifyIds_[remoteNotifyIdx];
875 :
876 1 : CHK_PRT_RET(localRmaBuffers_.empty(), HCCL_ERROR("[HostCpuRoceChannel::%s] localRmaBuffer is Empty", __func__),
877 : HCCL_E_ROCE_CONNECT);
878 1 : CHK_PRT_RET(rmtRmaBuffers_.empty(), HCCL_ERROR("[HostCpuRoceChannel::%s] rmtRmaBuffers is Empty", __func__),
879 : HCCL_E_ROCE_CONNECT);
880 :
881 : // 1. 构造WR
882 1 : CHK_PRT_RET(len > UINT32_MAX, HCCL_ERROR("[HostCpuRoceChannel][%s] the len[%llu] exceeds the size of u32.",
883 : __func__, len), HCCL_E_PARA);
884 :
885 1 : size_t localIdx = 0;
886 1 : CHK_RET(FindLocalBuffer(reinterpret_cast<uint64_t>(src), len, localIdx));
887 1 : size_t rmtIdx = 0;
888 1 : CHK_RET(FindRemoteBuffer(reinterpret_cast<uint64_t>(dst), len, rmtIdx));
889 :
890 1 : writeWithNotifyWr.sg_list->addr = reinterpret_cast<uint64_t>(src); // 本端起始地址
891 1 : writeWithNotifyWr.sg_list->length = static_cast<uint32_t>(len);
892 1 : writeWithNotifyWr.sg_list->lkey = localRmaBuffers_[localIdx]->GetLkey(); // 本端的访问秘钥
893 :
894 1 : writeWithNotifyWr.opcode = IBV_WR_RDMA_WRITE_WITH_IMM;
895 1 : writeWithNotifyWr.send_flags = IBV_SEND_SIGNALED;
896 1 : writeWithNotifyWr.next = nullptr;
897 1 : writeWithNotifyWr.num_sge = 1;
898 1 : writeWithNotifyWr.wr_id = 0;
899 1 : writeWithNotifyWr.imm_data = dpuNotifyId;
900 1 : writeWithNotifyWr.wr.rdma.rkey = rmtRmaBuffers_[rmtIdx]->GetRkey();
901 1 : writeWithNotifyWr.wr.rdma.remote_addr = reinterpret_cast<uint64_t>(dst);
902 :
903 1 : taskParam.taskType = Hccl::TaskParamType::TASK_DPU_WRITE_WITH_NOTIFY;
904 1 : taskParam.taskPara.DMA.src = src;
905 1 : taskParam.taskPara.DMA.dst = dst;
906 1 : taskParam.taskPara.DMA.size = len;
907 1 : taskParam.taskPara.DMA.notifyID = dpuNotifyId;
908 1 : taskParam.taskPara.DMA.notifyValue = 1;
909 1 : taskParam.taskPara.DMA.linkType = Hccl::DfxLinkType::ROCE;
910 1 : taskParam.taskPara.DMA.dmaOp = Hccl::DmaOp::HCCL_DMA_WRITE;
911 1 : return HCCL_SUCCESS;
912 : }
913 :
914 3 : HcclResult HostCpuRoceChannel::WriteWithNotify(
915 : void *dst, const void *src, const uint64_t len, const uint32_t remoteNotifyIdx)
916 : {
917 3 : CHK_PTR_NULL(src);
918 3 : CHK_PTR_NULL(dst);
919 3 : CHK_PRT_RET(maxMsgSize_ == 0,
920 : HCCL_ERROR("[HostCpuRoceChannel::%s] maxMsgSize_ is 0, channel not initialized", __func__),
921 : HCCL_E_INTERNAL);
922 2 : CHK_PRT_RET(GetQpInfos().empty(), HCCL_ERROR("[HostCpuRoceChannel::%s] qpInfos is Empty", __func__),
923 : HCCL_E_ROCE_CONNECT);
924 2 : HCCL_INFO("[HostCpuRoceChannel::%s] START. dst[%p], src[%p], len[0x%llx], remoteNotifyIdx[%u].",
925 : __func__, dst, src, len, remoteNotifyIdx);
926 2 : std::vector<int> wqeNumBefore = wqeNums_;
927 :
928 2 : if (isHybridMode_) {
929 0 : return WriteWithNotifyHybrid(dst, src, len, remoteNotifyIdx);
930 : }
931 :
932 : // 前 N-1 块: 普通 RDMA_WRITE
933 2 : uint64_t offset = 0;
934 4 : while (len - offset > maxMsgSize_) {
935 2 : CHK_RET(PostRdmaOp(__func__, IBV_WR_RDMA_WRITE,
936 : static_cast<char *>(const_cast<void *>(src)) + offset,
937 : static_cast<const char *>(dst) + offset,
938 : maxMsgSize_));
939 2 : offset += maxMsgSize_;
940 : }
941 :
942 : // 尾块: RDMA_WRITE_WITH_IMM，携带 notify
943 2 : void *tailDst = static_cast<char *>(dst) + offset;
944 2 : const void *tailSrc = static_cast<const char *>(src) + offset;
945 2 : uint64_t tailLen = len - offset;
946 :
947 : // 计算每个qp需要发送的数据量
948 2 : std::vector<Hccl::QpInfo> qpInfo = GetQpInfos();
949 2 : uint32_t useQpNum = qpInfo.size();
950 2 : uint32_t tileLen = tailLen / useQpNum;
951 2 : uint32_t tileLenTail = tailLen - (useQpNum - 1) * tileLen;
952 2 : if ((qpInfo.size() != 1) && (tileLen != 0) && (tileLen < channelDesc_.roceAttr.qpThreshold)) {
953 0 : useQpNum = (tailLen - 1) / channelDesc_.roceAttr.qpThreshold + 1; // 自适应选择QP数发送数据，保证每个qp分担的数据量满足最小阈值
954 0 : if (useQpNum > qpInfo.size()) {
955 0 : useQpNum = qpInfo.size();
956 : }
957 0 : tileLen = tailLen / useQpNum;
958 0 : tileLenTail = tailLen - (useQpNum - 1) * tileLen;
959 0 : HCCL_INFO("[HostCpuRoceChannel::%s] The data allocated to each Qp (%u) is below the Qp Threshold (%u). "
960 : "the count of Qp for data sending is adaptively adjusted to %u.",
961 : __func__, tileLen, channelDesc_.roceAttr.qpThreshold, useQpNum);
962 : }
963 :
964 : // 构造 WR
965 2 : Hccl::TaskParam taskParam{};
966 : uint32_t wrLen;
967 3 : for (uint32_t i = 0; i < qpInfo.size(); i++) {
968 2 : if (i < useQpNum - 1) {
969 0 : wrLen = tileLen;
970 2 : } else if (i == useQpNum - 1) {
971 2 : wrLen = tileLenTail;
972 : } else {
973 0 : wrLen = 0;
974 : }
975 2 : struct ibv_send_wr writeWithNotifyWr{};
976 2 : struct ibv_sge sgList{};
977 2 : writeWithNotifyWr.sg_list = &sgList;
978 2 : uint64_t offset = (wrLen == 0) ? 0 : (tileLen * i);
979 2 : CHK_RET(PrepareWriteWrResource(static_cast<char *>(tailDst) + offset,
980 : static_cast<const char *>(tailSrc) + offset, wrLen, remoteNotifyIdx,
981 : writeWithNotifyWr, taskParam));
982 1 : CHK_RET(PostAndCheckSend(qpInfo[i].qp, i, __func__, writeWithNotifyWr));
983 1 : HCCL_INFO("[HostCpuRoceChannel::%s] SUCCESS. qp[%u], wrlen[0x%llx], newWqe[%u], wqeNums_[%u].",
984 : __func__, i, wrLen, wqeNums_[i] - wqeNumBefore[i], wqeNums_[i]);
985 : }
986 1 : fenceFlag_ = false;
987 1 : taskParam.endTime = hccl::DlProfFunction::GetInstance().dlMsprofSysCycleTime();
988 1 : if (dfxCallback_ != nullptr) {
989 0 : return dfxCallback_(taskParam, reinterpret_cast<u64>(this));
990 : }
991 :
992 :
993 1 : return HCCL_SUCCESS;
994 2 : }
995 :
996 2 : void HostCpuRoceChannel::BuildRdmaWr(const char *caller, ibv_wr_opcode opcode, void *localAddr, const void *remoteAddr, uint64_t len,
997 : size_t localIdx, size_t rmtIdx, struct ibv_send_wr &wr, struct ibv_sge &sg) const
998 : {
999 2 : wr.sg_list = &sg;
1000 2 : wr.sg_list->addr = reinterpret_cast<uint64_t>(localAddr);
1001 2 : wr.sg_list->length = static_cast<uint32_t>(len);
1002 2 : wr.sg_list->lkey = localRmaBuffers_[localIdx]->GetLkey();
1003 :
1004 2 : wr.opcode = opcode;
1005 2 : wr.send_flags = (fenceFlag_ == true ? (IBV_SEND_SIGNALED | IBV_SEND_FENCE) : IBV_SEND_SIGNALED);
1006 2 : wr.next = nullptr;
1007 2 : wr.num_sge = 1;
1008 2 : wr.wr_id = 0;
1009 2 : wr.wr.rdma.rkey = rmtRmaBuffers_[rmtIdx]->GetRkey();
1010 2 : wr.wr.rdma.remote_addr = reinterpret_cast<uint64_t>(remoteAddr);
1011 2 : }
1012 :
1013 0 : HcclResult HostCpuRoceChannel::PostAndCheckSend(struct ibv_qp *qp, uint32_t qpIdx, const char *caller, struct ibv_send_wr &wr)
1014 : {
1015 0 : struct ibv_send_wr *badWr = nullptr;
1016 0 : s32 ret = ibv_post_send(qp, &wr, &badWr);
1017 0 : if (ret != 0 && badWr == nullptr) {
1018 0 : HCCL_ERROR("[HostCpuRoceChannel::%s] ibv_post_send failed while badWr is nullptr", caller);
1019 0 : return HCCL_E_INTERNAL;
1020 : }
1021 0 : CHK_PRT_RET(ret == ENOMEM,
1022 : HCCL_WARNING("[HostCpuRoceChannel::%s] post send wqe overflow. ret:%d, "
1023 : "badWr->wr_id[%llu], badWr->sg_list->addr[%llu], badWr->wr.rdma.remote_addr[%llu], badWr->wr.ud.remote_qpn[%u]",
1024 : caller, ret, badWr->wr_id, badWr->sg_list->addr, badWr->wr.rdma.remote_addr, badWr->wr.ud.remote_qpn),
1025 : HCCL_E_AGAIN);
1026 0 : CHK_PRT_RET(ret != 0,
1027 : HCCL_ERROR("[HostCpuRoceChannel::%s] ibv_post_send failed. ret:%d, "
1028 : "badWr->wr_id[%llu], badWr->sg_list->addr[%llu], badWr->wr.rdma.remote_addr[%llu], badWr->wr.ud.remote_qpn[%u]",
1029 : caller, ret, badWr->wr_id, badWr->sg_list->addr, badWr->wr.rdma.remote_addr, badWr->wr.ud.remote_qpn),
1030 : HCCL_E_NETWORK);
1031 0 : CHK_PRT_RET(wqeNums_[qpIdx] == INT32_MAX,
1032 : HCCL_ERROR("[HostCpuRoceChannel::%s] wqeNums_[%u] has reached the maximum value of uint32_t.", caller, qpIdx),
1033 : HCCL_E_INTERNAL);
1034 0 : wqeNums_[qpIdx]++;
1035 0 : return HCCL_SUCCESS;
1036 : }
1037 :
1038 2 : HcclResult HostCpuRoceChannel::PostRdmaOp(const char *caller, ibv_wr_opcode opcode, void *localAddr,
1039 : const void *remoteAddr, const uint64_t len)
1040 : {
1041 2 : HCCL_INFO("[HostCpuRoceChannel::%s] Slice START. localAddr[%p], remoteAddr[%p], len[0x%llx].", caller, localAddr,
1042 : remoteAddr, len);
1043 :
1044 2 : CHK_PRT_RET(GetQpInfos().empty(), HCCL_ERROR("[HostCpuRoceChannel::%s] qpInfos is Empty", caller), HCCL_E_ROCE_CONNECT);
1045 2 : CHK_PRT_RET(localRmaBuffers_.empty(), HCCL_ERROR("[HostCpuRoceChannel::%s] localRmaBuffer is Empty", caller),
1046 : HCCL_E_ROCE_CONNECT);
1047 2 : CHK_PRT_RET(rmtRmaBuffers_.empty(), HCCL_ERROR("[HostCpuRoceChannel::%s] rmtRmaBuffers is Empty", caller),
1048 : HCCL_E_ROCE_CONNECT);
1049 2 : if (len > maxMsgSize_) {
1050 0 : HCCL_WARNING(
1051 : "[HostCpuRoceChannel::%s] len[0x%llx] exceeds maxMsgSize_[0x%llx], caller should slice before posting.",
1052 : caller, len, maxMsgSize_);
1053 : }
1054 :
1055 : // 1. 查找 buffer 索引
1056 2 : auto startTime = std::chrono::steady_clock::now();
1057 2 : size_t localIdx = 0;
1058 2 : CHK_RET(FindLocalBuffer(reinterpret_cast<uint64_t>(localAddr), len, localIdx));
1059 2 : size_t rmtIdx = 0;
1060 2 : CHK_RET(FindRemoteBuffer(reinterpret_cast<uint64_t>(remoteAddr), len, rmtIdx));
1061 2 : auto endTime = std::chrono::steady_clock::now();
1062 2 : auto elapsed = std::chrono::duration_cast<std::chrono::microseconds>(endTime - startTime).count();
1063 2 : HCCL_INFO("[HostCpuRoceChannel::%s] check buffer takes time [%lld]us", caller, elapsed);
1064 :
1065 : // 2. 构造 WR 并发送
1066 2 : std::vector<Hccl::QpInfo> qpInfo = GetQpInfos();
1067 2 : uint32_t useQpNum = qpInfo.size();
1068 2 : uint32_t tileLen = len / useQpNum;
1069 2 : uint32_t tileLenTail = len - (useQpNum - 1) * tileLen;
1070 2 : if ((tileLen != 0) && (tileLen < channelDesc_.roceAttr.qpThreshold)) {
1071 0 : useQpNum = (len - 1) / channelDesc_.roceAttr.qpThreshold + 1; // 自适应选择QP数发送数据，保证每个qp分担的数据量满足最小阈值
1072 0 : if (useQpNum > qpInfo.size()) {
1073 0 : useQpNum = qpInfo.size();
1074 : }
1075 0 : tileLen = len / useQpNum;
1076 0 : tileLenTail = len - (useQpNum - 1) * tileLen;
1077 0 : HCCL_INFO("[HostCpuRoceChannel::%s] The data allocated to each Qp (%u) is below the Qp Threshold (%u). "
1078 : "the count of Qp for data sending is adaptively adjusted to %u.",
1079 : __func__, tileLen, channelDesc_.roceAttr.qpThreshold, useQpNum);
1080 : }
1081 :
1082 : uint32_t wrLen;
1083 4 : for (uint32_t i = 0; i < qpInfo.size(); i++) {
1084 2 : if (i < useQpNum - 1) {
1085 0 : wrLen = tileLen;
1086 2 : } else if (i == useQpNum - 1) {
1087 2 : wrLen = tileLenTail;
1088 : } else {
1089 0 : wrLen = 0;
1090 : }
1091 :
1092 2 : if (wrLen == 0) {
1093 0 : HCCL_DEBUG("[HostCpuRoceChannel::%s] wrLen is 0, qp[%u] qpn[%u] skip post send.", __func__, i, qpInfo[i].qp->qp_num);
1094 0 : continue;
1095 : }
1096 :
1097 2 : struct ibv_send_wr wr{};
1098 : struct ibv_sge sg;
1099 2 : BuildRdmaWr(caller, opcode, static_cast<char*>(localAddr) + tileLen * i, static_cast<const char*>(remoteAddr) + tileLen * i, wrLen, localIdx, rmtIdx, wr, sg);
1100 2 : CHK_RET(PostAndCheckSend(qpInfo[i].qp, i, caller, wr));
1101 2 : HCCL_INFO("[HostCpuRoceChannel::%s] Slice SUCCESS. qp[%u] wrLen[0x%llx], wqeNums_[%u]=%d", caller, i, wrLen, i, wqeNums_[i]);
1102 : }
1103 2 : fenceFlag_ = false;
1104 2 : return HCCL_SUCCESS;
1105 2 : }
1106 :
1107 5 : HcclResult HostCpuRoceChannel::Write(void *dst, const void *src, const uint64_t len)
1108 : {
1109 5 : CHK_PRT_RET(maxMsgSize_ == 0,
1110 : HCCL_ERROR("[HostCpuRoceChannel::%s] maxMsgSize_ is 0, channel not initialized", __func__),
1111 : HCCL_E_INTERNAL);
1112 4 : HCCL_INFO("[HostCpuRoceChannel::%s] START. dst[%p], src[%p], len[0x%llx].", __func__, dst, src, len);
1113 4 : std::vector<int> wqeNumBefore = wqeNums_;
1114 4 : uint64_t offset = 0;
1115 9 : while (offset < len) {
1116 6 : uint64_t chunkLen = std::min(len - offset, maxMsgSize_);
1117 6 : CHK_RET(PostRdmaOp(__func__, IBV_WR_RDMA_WRITE,
1118 : static_cast<char *>(const_cast<void *>(src)) + offset,
1119 : static_cast<const char *>(dst) + offset,
1120 : chunkLen));
1121 5 : offset += chunkLen;
1122 : }
1123 3 : HCCL_INFO("[HostCpuRoceChannel::%s] SUCCESS. len[0x%llx], newWqe[%d], wqeNums_[%d].",
1124 : __func__, len, wqeNums_[0] - wqeNumBefore[0], wqeNums_[0]);
1125 3 : return HCCL_SUCCESS;
1126 4 : }
1127 :
1128 3 : HcclResult HostCpuRoceChannel::Read(void *dst, const void *src, const uint64_t len)
1129 : {
1130 3 : CHK_PRT_RET(maxMsgSize_ == 0,
1131 : HCCL_ERROR("[HostCpuRoceChannel::%s] maxMsgSize_ is 0, channel not initialized", __func__),
1132 : HCCL_E_INTERNAL);
1133 2 : HCCL_INFO("[HostCpuRoceChannel::%s] START. dst[%p], src[%p], len[0x%llx].", __func__, dst, src, len);
1134 2 : std::vector<int> wqeNumBefore = wqeNums_;
1135 2 : uint64_t offset = 0;
1136 6 : while (offset < len) {
1137 4 : uint64_t chunkLen = std::min(len - offset, maxMsgSize_);
1138 4 : CHK_RET(PostRdmaOp(__func__, IBV_WR_RDMA_READ,
1139 : static_cast<char *>(dst) + offset,
1140 : static_cast<const char *>(src) + offset,
1141 : chunkLen));
1142 4 : offset += chunkLen;
1143 : }
1144 2 : HCCL_INFO("[HostCpuRoceChannel::%s] SUCCESS. len[0x%llx], newWqe[%d], wqeNums_[%d].",
1145 : __func__, len, wqeNums_[0] - wqeNumBefore[0], wqeNums_[0]);
1146 2 : return HCCL_SUCCESS;
1147 2 : }
1148 :
1149 2 : HcclResult HostCpuRoceChannel::FindLocalBuffer(const uint64_t addr, const uint64_t len, size_t &targetIdx) const
1150 : {
1151 2 : uint64_t endAddr = addr + len;
1152 2 : HCCL_INFO("[HostCpuRoceChannel::%s] START. Finding buffer addr[0x%llx], len[0x%llx], addr+len[0x%llx].", __func__, addr, len, endAddr);
1153 2 : for (size_t i = 0; i < localRmaBuffers_.size(); ++i) {
1154 2 : CHK_PTR_NULL(localRmaBuffers_[i]);
1155 2 : uint64_t bufAddr = localRmaBuffers_[i]->GetBufferInfo().first;
1156 2 : uint64_t bufSize = localRmaBuffers_[i]->GetBufferInfo().second;
1157 2 : uint64_t bufEndAddr = bufAddr + bufSize;
1158 2 : HCCL_INFO("[HostCpuRoceChannel::%s] Comparing with saved localRmaBuffer[%zu]: addr[0x%llx], len[0x%llx], addr+len[0x%llx].", __func__, i, bufAddr, bufSize, bufEndAddr);
1159 2 : if (addr >= bufAddr && endAddr <= bufEndAddr) {
1160 2 : targetIdx = i;
1161 2 : HCCL_INFO("[HostCpuRoceChannel::%s] SUCCESS. targetIdx[%zu]", __func__, targetIdx);
1162 2 : return HCCL_SUCCESS;
1163 : }
1164 : }
1165 0 : HCCL_ERROR("[HostCpuRoceChannel::%s] FAIL. Can not Found Target Buffer addr[0x%llx], len[0x%llx], addr+len[0x%llx].", __func__, addr, len, endAddr);
1166 0 : return HCCL_E_NOT_FOUND;
1167 : }
1168 :
1169 2 : HcclResult HostCpuRoceChannel::FindRemoteBuffer(const uint64_t addr, const uint64_t len, size_t &targetIdx) const
1170 : {
1171 2 : uint64_t endAddr = addr + len;
1172 2 : HCCL_INFO("[HostCpuRoceChannel::%s] START. Finding buffer addr[0x%llx], len[0x%llx], addr+len[0x%llx].", __func__, addr, len, endAddr);
1173 2 : for (size_t i = 0; i < rmtRmaBuffers_.size(); ++i) {
1174 2 : CHK_PTR_NULL(rmtRmaBuffers_[i]);
1175 2 : uint64_t bufAddr = static_cast<uint64_t>(rmtRmaBuffers_[i]->GetAddr());
1176 2 : uint64_t bufSize = rmtRmaBuffers_[i]->GetSize();
1177 2 : uint64_t bufEndAddr = bufAddr + bufSize;
1178 2 : HCCL_INFO("[HostCpuRoceChannel::%s] Comparing with saved rmtRmaBuffers[%zu]: addr[0x%llx], len[0x%llx], addr+len[0x%llx].", __func__, i, bufAddr, bufSize, bufEndAddr);
1179 2 : if (addr >= bufAddr && endAddr <= bufEndAddr) {
1180 2 : targetIdx = i;
1181 2 : HCCL_INFO("[HostCpuRoceChannel::%s] SUCCESS. targetIdx[%zu]", __func__, targetIdx);
1182 2 : return HCCL_SUCCESS;
1183 : }
1184 : }
1185 0 : HCCL_ERROR("[HostCpuRoceChannel::%s] FAIL. Can not Found Target Buffer addr[0x%llx], len[0x%llx], addr+len[0x%llx].", __func__, addr, len, endAddr);
1186 0 : return HCCL_E_NOT_FOUND;
1187 : }
1188 :
1189 3 : HcclResult HostCpuRoceChannel::WaitForFenceCompletion()
1190 : {
1191 3 : const std::vector<Hccl::QpInfo> qpInfo = GetQpInfos();
1192 3 : CHK_PRT_RET(qpInfo.empty(), HCCL_ERROR("[HostCpuRoceChannel::%s] qpInfos is Empty", __func__), HCCL_E_ROCE_CONNECT);
1193 3 : uint32_t fenceCount = 0;
1194 6 : for (uint32_t i = 0; i < qpInfo.size(); i++) {
1195 3 : if (wqeNums_[i] == 0) {
1196 1 : fenceCount += 1;
1197 : }
1198 : }
1199 3 : if (fenceCount == qpInfo.size()) {
1200 1 : fenceFlag_ = true;
1201 1 : HCCL_INFO("[HostCpuRoceChannel::%s] SUCCESS. elements in wqeNums_ are 0.", __func__);
1202 1 : return HCCL_SUCCESS;
1203 : }
1204 :
1205 : auto timeout = std::chrono::milliseconds(
1206 2 : static_cast<uint64_t>(Hccl::EnvConfig::GetInstance().GetRtsConfig().GetExecTimeOut()) * 1000ULL); // 乘1000转为毫秒
1207 2 : for (uint32_t i = 0; i < qpInfo.size(); i++) {
1208 2 : std::vector<struct ibv_wc> wc(wqeNums_[i]);
1209 2 : CHK_PRT_RET(qpInfo[i].sendCq == nullptr, HCCL_ERROR("[HostCpuRoceChannel::%s] qp[%u] sendCq is null", __func__, i), HCCL_E_INTERNAL);
1210 2 : CHK_PRT_RET(qpInfo[i].qp == nullptr, HCCL_ERROR("[HostCpuRoceChannel::%s] qp[%u] is null", __func__, i), HCCL_E_INTERNAL);
1211 2 : CHK_PRT_RET(qpInfo[i].sendCq->context == nullptr, HCCL_ERROR("[HostCpuRoceChannel::%s] qp[%u] sendCq->context is null", __func__, i), HCCL_E_INTERNAL);
1212 :
1213 2 : auto startTime = std::chrono::steady_clock::now();
1214 : while (true) {
1215 2 : int actualNum = IbvPollCq(qpInfo[i].sendCq, wqeNums_[i], wc.data());
1216 2 : if (actualNum < 0) {
1217 1 : HCCL_ERROR("[HostCpuRoceChannel::%s] qp[%u] ibv_poll_cq failed. actualNum: %d.", __func__, i, actualNum);
1218 1 : return HCCL_E_NETWORK;
1219 : }
1220 :
1221 1 : if (actualNum > wqeNums_[i]) {
1222 1 : HCCL_ERROR("[HostCpuRoceChannel::%s] qp[%u] ibv_poll_cq polled more completions (%d) than expected (%d).",
1223 : __func__, i, actualNum, wqeNums_[i]);
1224 1 : return HCCL_E_INTERNAL;
1225 0 : } else if (actualNum > 0) {
1226 0 : for (int j = 0; j < actualNum; j++) {
1227 0 : if (wc[j].status != IBV_WC_SUCCESS) {
1228 0 : HCCL_ERROR("[HostCpuRoceChannel::%s] ibv_poll_cq error. wc[%d] status[%d], opcode[%d], vendorErr[%u], byteLen[%u], wcFlags[%u], sl[%u]. qpInfo[%d].qp->qp_num[%u]",
1229 : __func__, j, wc[j].status, wc[j].opcode, wc[j].vendor_err, wc[j].byte_len, wc[j].wc_flags, wc[j].sl, i, qpInfo[i].qp->qp_num);
1230 0 : return HCCL_E_NETWORK;
1231 : }
1232 : }
1233 0 : wqeNums_[i] -= actualNum; // 减去已经完成的数量，继续等待剩余的完成
1234 0 : if (wqeNums_[i] == 0) {
1235 0 : break; // 所有的wqe都已经完成，退出循环
1236 : }
1237 0 : startTime = std::chrono::steady_clock::now(); // 有进展，重置超时计时
1238 : }
1239 :
1240 0 : if ((std::chrono::steady_clock::now() - startTime) >= timeout) {
1241 0 : HCCL_ERROR("[HostCpuRoceChannel][%s] qp[%u] call ibv_poll_cq timeout, remaining wqeNum[%u].", __func__, i, wqeNums_[i]);
1242 0 : return HCCL_E_TIMEOUT;
1243 : }
1244 0 : }
1245 0 : wqeNums_[i] = 0; // 所有的wqe都已经完成，重置计数器
1246 0 : HCCL_INFO("[HostCpuRoceChannel::%s] SUCCESS. wqeNums_[%u]=%d.", __func__, i, wqeNums_[i]);
1247 2 : }
1248 0 : fenceFlag_ = true;
1249 0 : return HCCL_SUCCESS;
1250 3 : }
1251 :
1252 3 : HcclResult HostCpuRoceChannel::ChannelFence()
1253 : {
1254 3 : std::lock_guard<std::mutex> lock(sendCq_mutex);
1255 3 : Hccl::TaskParam taskParam{};
1256 3 : taskParam.beginTime = hccl::DlProfFunction::GetInstance().dlMsprofSysCycleTime();
1257 3 : HCCL_INFO("[HostCpuRoceChannel::%s] ChannelFence start, wqeNums_[0]=%d", __func__, wqeNums_[0]);
1258 3 : HcclResult ret = WaitForFenceCompletion();
1259 3 : if (ret != HCCL_SUCCESS) {
1260 2 : return ret;
1261 : }
1262 :
1263 1 : taskParam.taskType = Hccl::TaskParamType::TASK_DPU_CHANNEL_FENCE;
1264 1 : taskParam.taskPara.Notify.notifyID = INVALID_U64;
1265 1 : taskParam.taskPara.Notify.value = 1;
1266 1 : taskParam.endTime = hccl::DlProfFunction::GetInstance().dlMsprofSysCycleTime();
1267 1 : if (dfxCallback_ != nullptr) {
1268 0 : return dfxCallback_(taskParam, reinterpret_cast<u64>(this));
1269 : }
1270 1 : return HCCL_SUCCESS;
1271 3 : }
1272 :
1273 1 : HcclResult HostCpuRoceChannel::GetNotifyNum(uint32_t *notifyNum) const
1274 : {
1275 1 : CHK_PTR_NULL(notifyNum);
1276 1 : *notifyNum = notifyNum_;
1277 1 : return HCCL_SUCCESS;
1278 : }
1279 :
1280 1 : HcclResult HostCpuRoceChannel::GetHcclBuffer(void*& addr, uint64_t& size)
1281 : {
1282 1 : if (rmtRmaBuffers_.empty()) {
1283 0 : HCCL_ERROR("[HostCpuRoceChannel::%s] remote buffer is empty, please check if channel complete exchange data",
1284 : __func__);
1285 0 : return HCCL_E_INTERNAL;
1286 : }
1287 1 : addr = reinterpret_cast<void*>(rmtRmaBuffers_[0]->GetAddr());
1288 1 : size = static_cast<uint64_t>(rmtRmaBuffers_[0]->GetSize());
1289 1 : return HCCL_SUCCESS;
1290 : }
1291 :
1292 0 : HcclResult HostCpuRoceChannel::Clean()
1293 : {
1294 0 : return HCCL_SUCCESS;
1295 : }
1296 :
1297 0 : HcclResult HostCpuRoceChannel::Resume()
1298 : {
1299 0 : return HCCL_SUCCESS;
1300 : }
1301 :
1302 : constexpr u32 DEFAULT_LOCAL_NOTIFY_ACCESS = 7;
1303 : constexpr u32 DEFAULT_LOCAL_NOTIFY_SIZE = 4;
1304 9 : HcclResult HostCpuRoceChannel::CreateNotifyHybird(hccl::MemType notifyType, uint32_t notifyId)
1305 : {
1306 9 : localNotifyAccess_ = DEFAULT_LOCAL_NOTIFY_ACCESS;
1307 9 : localNotifySize_ = DEFAULT_LOCAL_NOTIFY_SIZE;
1308 :
1309 9 : int8_t *ptr = new (std::nothrow) int8_t[localNotifySize_];
1310 9 : CHK_PTR_NULL(ptr);
1311 :
1312 9 : if (memset_s(ptr, localNotifySize_, 0, localNotifySize_) < 0) {
1313 0 : HCCL_ERROR("[HostCpuRoceChannel::CreateNotifyHybird] memset_s failed");
1314 0 : delete[] ptr;
1315 0 : return HCCL_E_MEMORY;
1316 : }
1317 :
1318 9 : struct MrInfoT mrInfo = {nullptr};
1319 9 : mrInfo.addr = ptr;
1320 9 : mrInfo.size = localNotifySize_;
1321 9 : mrInfo.access = localNotifyAccess_;
1322 9 : auto qpInfo = connections_[0]->GetQpInfo();
1323 9 : if (HrtRaMrReg(qpInfo.qpHandle, &mrInfo) != HCCL_SUCCESS) {
1324 1 : HCCL_ERROR("[HostCpuRoceChannel::CreateNotifyHybird] MrReg failed");
1325 1 : delete[] ptr;
1326 1 : return HCCL_E_MEMORY;
1327 : }
1328 :
1329 8 : localMemMsg_[notifyType].addr = ptr;
1330 8 : localMemMsg_[notifyType].lkey = mrInfo.lkey;
1331 8 : localMemMsg_[notifyType].memType = notifyType;
1332 8 : localMemMsg_[notifyType].len = localNotifySize_;
1333 8 : localMemMsg_[notifyType].notifyId = notifyId;
1334 :
1335 8 : return HCCL_SUCCESS;
1336 : }
1337 :
1338 2 : HcclResult HostCpuRoceChannel::CreateNotifyValueBufferHybird()
1339 : {
1340 2 : if (CreateNotifyHybird(hccl::MemType::NOTIFY_SRC_MEM, hccl::MemType::NOTIFY_SRC_MEM) != HCCL_SUCCESS) {
1341 0 : HCCL_ERROR("[HostCpuRoceChannel::CreateNotifyValueBufferHybird]Create host notify fail, type=%d",
1342 : hccl::MemType::NOTIFY_SRC_MEM);
1343 0 : return HCCL_E_MEMORY;
1344 : }
1345 2 : *reinterpret_cast<uint32_t *>(localMemMsg_[hccl::MemType::NOTIFY_SRC_MEM].addr) = 1;
1346 2 : return HCCL_SUCCESS;
1347 : }
1348 :
1349 4 : HcclResult HostCpuRoceChannel::CreateNotifyBufferHybird(hccl::MemType notifyType, uint32_t notifyId, u8 *&data, u64 &size)
1350 : {
1351 4 : if (CreateNotifyHybird(notifyType, notifyId) != HCCL_SUCCESS) {
1352 0 : HCCL_ERROR("[HostCpuRoceChannel::CreateNotifyBufferHybird]Create host notify buffer fail, type=%d", notifyType);
1353 0 : return HCCL_E_MEMORY;
1354 : }
1355 :
1356 4 : CHK_SAFETY_FUNC_RET(memcpy_s(data, size, reinterpret_cast<void *>(&localMemMsg_[notifyType]), sizeof(hccl::MemMsg)));
1357 :
1358 4 : data += sizeof(hccl::MemMsg);
1359 4 : size -= sizeof(hccl::MemMsg);
1360 :
1361 4 : return HCCL_SUCCESS;
1362 : }
1363 :
1364 2 : HcclResult HostCpuRoceChannel::ExchangeCapability()
1365 : {
1366 2 : HCCL_INFO("[Hybrid][HostCpuRoceChannel] Starting capability exchange");
1367 :
1368 : // 1. 构造本地能力信息（使用公共头文件中的默认值）
1369 : RoCECapability localCap;
1370 2 : localCap.InitDefaults();
1371 2 : localCap.nicDeploy = NICDeployment::NIC_DEPLOYMENT_HOST;
1372 2 : localCap.commStack = CommStackType::COMM_STACK_HOST_CPU_ROCE;
1373 :
1374 : // 2. 发送本地能力（合并为单次发送：totalLength已包含结构体大小）
1375 2 : CHK_PRT_RET(!socket_->Send(&localCap, sizeof(localCap)),
1376 : HCCL_ERROR("[HostCpuRoceChannel::%s] Send exchange localCap failed", __func__), HCCL_E_NETWORK);
1377 2 : HCCL_INFO("[Hybrid][HostCpuRoceChannel] Sent capability, version=%u", localCap.version);
1378 :
1379 : // 3. 接收对端能力（单次接收）
1380 : RoCECapability recvCap;
1381 2 : CHK_PRT_RET(!socket_->Recv(&recvCap, sizeof(recvCap)),
1382 : HCCL_ERROR("[HostCpuRoceChannel::%s] Recv recvCap failed", __func__), HCCL_E_NETWORK);
1383 2 : HCCL_INFO("[Hybrid][HostCpuRoceChannel] recvCap success");
1384 :
1385 : // 4. 先检查魔数，如果不对可能是旧版本，需要回退
1386 2 : if (!RoCECapability::CheckMagic(reinterpret_cast<uint8_t*>(&recvCap), sizeof(recvCap))) {
1387 1 : HCCL_WARNING("[Hybrid][HostCpuRoceChannel] Magic mismatch, peer may be old version. "
1388 : "Falling back to native mode.");
1389 : // 回退到原生模式
1390 1 : isHybridMode_ = false;
1391 : // 标记为"跳过混合模式协商"，后续流程继续使用原生模式
1392 1 : remoteCap_.magic = 0; // 标记为无效
1393 1 : return HCCL_SUCCESS;
1394 : }
1395 :
1396 : // 5. 魔数正确，解析对端能力
1397 1 : if (!remoteCap_.Deserialize(reinterpret_cast<uint8_t*>(&recvCap), sizeof(recvCap))) {
1398 0 : HCCL_ERROR("[Hybrid][HostCpuRoceChannel] Failed to deserialize capability");
1399 0 : return HCCL_E_PARA;
1400 : }
1401 :
1402 : // 6. 校验字段有效性
1403 1 : if (!remoteCap_.Validate()) {
1404 0 : HCCL_ERROR("[Hybrid][HostCpuRoceChannel] Capability validation failed");
1405 0 : return HCCL_E_INTERNAL;
1406 : }
1407 :
1408 : // 7. 版本兼容性处理（高版本兼容低版本）
1409 1 : if (remoteCap_.version > ROCE_CAPABILITY_VERSION) {
1410 : // 对端版本更高，使用本地版本的功能集（最小公分母）
1411 0 : HCCL_INFO("[Hybrid][HostCpuRoceChannel] Remote version %u > local %u, using local version features",
1412 : remoteCap_.version, ROCE_CAPABILITY_VERSION);
1413 1 : } else if (remoteCap_.version < ROCE_CAPABILITY_VERSION) {
1414 : // 对端版本更低，使用对端版本的功能集（向下兼容）
1415 0 : HCCL_INFO("[Hybrid][HostCpuRoceChannel] Remote version %u < local %u, using remote version features",
1416 : remoteCap_.version, ROCE_CAPABILITY_VERSION);
1417 : }
1418 :
1419 1 : isHybridMode_ = (remoteCap_.commStack == CommStackType::COMM_STACK_TRANSPORT_IBVERBS) ? true : false;
1420 :
1421 1 : HCCL_INFO("[Hybrid][HostCpuRoceChannel] Capability exchange success, "
1422 : "remote commStack=%u, version=%u, mode=%s",
1423 : static_cast<uint8_t>(remoteCap_.commStack), remoteCap_.version, isHybridMode_ ? "hybrid" : "normal");
1424 1 : return HCCL_SUCCESS;
1425 : }
1426 :
1427 : constexpr u32 DEFAULT_MRINFO_ACCESS = 7;
1428 1 : HcclResult HostCpuRoceChannel::RegisterUserMemHybird()
1429 : {
1430 1 : struct MrInfoT mrInfo = {nullptr};
1431 1 : mrInfo.addr = reinterpret_cast<void *>(localRmaBuffers_[0]->GetAddr());
1432 1 : mrInfo.size = localRmaBuffers_[0]->GetSize();
1433 1 : mrInfo.access = DEFAULT_MRINFO_ACCESS;
1434 1 : auto qpInfo = connections_[0]->GetQpInfo();
1435 1 : CHK_RET(HrtRaMrReg(qpInfo.qpHandle, &mrInfo));
1436 :
1437 1 : localMemMsg_[hccl::USER_OUTPUT_MEM].addr = reinterpret_cast<void *>(localRmaBuffers_[0]->GetAddr());
1438 1 : localMemMsg_[hccl::USER_OUTPUT_MEM].lkey = mrInfo.lkey;
1439 1 : localMemMsg_[hccl::USER_OUTPUT_MEM].memType = hccl::USER_OUTPUT_MEM;
1440 1 : localMemMsg_[hccl::USER_OUTPUT_MEM].len = localRmaBuffers_[0]->GetSize();
1441 1 : localMemMsg_[hccl::USER_OUTPUT_MEM].notifyId = INVALID_DPU_NOTIFY_ID;
1442 :
1443 1 : return HCCL_SUCCESS;
1444 : }
1445 :
1446 : constexpr u32 BUFFER_NUM = 2; // output、input buffer
1447 : constexpr u32 NOTIFY_NUM = 3; // 3个Notify
1448 3 : HcclResult HostCpuRoceChannel::BuildExchangeDataLengthHybird()
1449 : {
1450 3 : exchangeDataTotalSize_ = 0;
1451 3 : exchangeDataTotalSize_ += sizeof(u32); // qp数量
1452 3 : exchangeDataTotalSize_ += sizeof(hccl::MemMsg) * BUFFER_NUM; // output、input buffer
1453 3 : exchangeDataTotalSize_ += sizeof(hccl::MemMsg) * NOTIFY_NUM; // 3个Notify
1454 3 : exchangeDataTotalSize_ += sizeof(u8); // atomic value
1455 3 : HCCL_INFO("[BuildExchangeDataLengthHybird]ExchangeDataSize:%ld", exchangeDataTotalSize_);
1456 3 : return HCCL_SUCCESS;
1457 : }
1458 :
1459 : constexpr u32 DATA_NOTIFY_ID = 1; // DATA_NOTIFY_MEM类型使用的notifyId
1460 : constexpr u32 ACK_NOTIFY_ID = 0; // ACK_NOTIFY_MEM类型使用的notifyId
1461 : constexpr u32 DATA_ACK_NOTIFY_ID = 2; // DATA_ACK_NOTIFY_MEM类型使用的notifyId
1462 1 : HcclResult HostCpuRoceChannel::BuildExchangeDataHybird()
1463 : {
1464 1 : CHK_RET(BuildExchangeDataLengthHybird());
1465 :
1466 1 : exchangeDataForSend_.resize(exchangeDataTotalSize_);
1467 :
1468 1 : u8 *data = exchangeDataForSend_.data();
1469 1 : u64 size = exchangeDataTotalSize_;
1470 :
1471 1 : u32 qpNum = 1;
1472 1 : CHK_SAFETY_FUNC_RET(memcpy_s(data, size, reinterpret_cast<void *>(&qpNum), sizeof(u32)));
1473 1 : data += sizeof(u32);
1474 1 : size -= sizeof (u32);
1475 :
1476 1 : CHK_SAFETY_FUNC_RET(memcpy_s(data, size, reinterpret_cast<void *>(&localMemMsg_[hccl::USER_OUTPUT_MEM]), sizeof(hccl::MemMsg)));
1477 1 : data += sizeof(hccl::MemMsg);
1478 1 : size -= sizeof (hccl::MemMsg);
1479 1 : CHK_SAFETY_FUNC_RET(memcpy_s(data, size, reinterpret_cast<void *>(&localMemMsg_[hccl::USER_OUTPUT_MEM]), sizeof(hccl::MemMsg)));
1480 1 : data += sizeof(hccl::MemMsg);
1481 1 : size -= sizeof (hccl::MemMsg);
1482 :
1483 1 : CHK_RET(CreateNotifyValueBufferHybird());
1484 1 : CHK_RET(CreateNotifyBufferHybird(hccl::DATA_NOTIFY_MEM, DATA_NOTIFY_ID, data, size));
1485 1 : CHK_RET(CreateNotifyBufferHybird(hccl::ACK_NOTIFY_MEM, ACK_NOTIFY_ID, data, size));
1486 1 : CHK_RET(CreateNotifyBufferHybird(hccl::DATA_ACK_NOTIFY_MEM, DATA_ACK_NOTIFY_ID, data, size));
1487 :
1488 1 : u8 atomicWrite = 1;
1489 1 : CHK_SAFETY_FUNC_RET(memcpy_s(data, size, reinterpret_cast<void *>(&atomicWrite), sizeof(u8)));
1490 1 : data += sizeof(u8);
1491 1 : size -= sizeof (u8);
1492 :
1493 1 : if (size != 0) {
1494 0 : HCCL_ERROR("HostCpuRoceChannel::BuildExchangeDataHybird, failed to construct exchange data, size=%llu", size);
1495 0 : return HCCL_E_INTERNAL;
1496 : }
1497 :
1498 1 : return HCCL_SUCCESS;
1499 : }
1500 :
1501 6 : HcclResult HostCpuRoceChannel::GetRemoteAddrHybird(hccl::MemType memType, u8 *&data, u64 &size)
1502 : {
1503 6 : CHK_SAFETY_FUNC_RET(memcpy_s(&remoteMemMsg_[static_cast<u32>(memType)], sizeof(hccl::MemMsg), data, sizeof(hccl::MemMsg)));
1504 6 : data += sizeof(hccl::MemMsg);
1505 6 : size -= sizeof(hccl::MemMsg);
1506 6 : return HCCL_SUCCESS;
1507 : }
1508 :
1509 1 : HcclResult HostCpuRoceChannel::ParseRecvExchangeDataHybird()
1510 : {
1511 1 : u8 *data = exchangeDataForRecv_.data();
1512 1 : u64 size = exchangeDataTotalSize_;
1513 :
1514 1 : u32 remoteQpNum = 0;
1515 1 : CHK_SAFETY_FUNC_RET(memcpy_s(reinterpret_cast<void *>(&remoteQpNum), sizeof(u32), data, sizeof(u32)));
1516 1 : data += sizeof(u32);
1517 1 : size -= sizeof(u32);
1518 :
1519 1 : CHK_RET(GetRemoteAddrHybird(hccl::USER_OUTPUT_MEM, data, size));
1520 1 : CHK_RET(GetRemoteAddrHybird(hccl::USER_INPUT_MEM, data, size));
1521 1 : CHK_RET(GetRemoteAddrHybird(hccl::DATA_NOTIFY_MEM, data, size));
1522 1 : CHK_RET(GetRemoteAddrHybird(hccl::ACK_NOTIFY_MEM, data, size));
1523 1 : CHK_RET(GetRemoteAddrHybird(hccl::DATA_ACK_NOTIFY_MEM, data, size));
1524 :
1525 1 : data += sizeof(u8);
1526 1 : size -= sizeof(u8);
1527 :
1528 1 : if (size != 0) {
1529 0 : HCCL_ERROR("HostCpuRoceChannel::ParseRecvExchangeDataHybird: failed to parse exchange data, size=%lld", size);
1530 0 : return HCCL_E_INTERNAL;
1531 : }
1532 :
1533 2 : Hccl::ExchangeRdmaBufferDto dto((u64)remoteMemMsg_[static_cast<u32>(hccl::USER_OUTPUT_MEM)].addr,
1534 1 : remoteMemMsg_[static_cast<u32>(hccl::USER_OUTPUT_MEM)].len,
1535 1 : remoteMemMsg_[static_cast<u32>(hccl::USER_OUTPUT_MEM)].lkey, "HcclBuffer");
1536 1 : rmtRmaBuffers_.push_back(std::make_unique<Hccl::RemoteRdmaRmaBuffer>(rdmaHandle_, dto));
1537 :
1538 1 : return HCCL_SUCCESS;
1539 1 : }
1540 :
1541 : constexpr u32 GET_QP_SLEEP_TIME = 1000;
1542 1 : HcclResult HostCpuRoceChannel::ConnectSingleQpHybrid(std::function<bool()> needStop)
1543 : {
1544 1 : auto qpInfo = connections_[0]->GetQpInfo();
1545 1 : bool hasSocket = (socket_ != nullptr);
1546 1 : if (!hasSocket) {
1547 0 : CHK_RET(SocketMgr::GetInstance(devicePhyId_).GetSocket(*socketConfig_, socket_));
1548 : }
1549 1 : CHK_RET(HrtRaQpConnectAsync(qpInfo.qpHandle, socket_->GetFdHandle(), needStop));
1550 :
1551 : // 查询QP建链是否成功
1552 1 : s32 qpStatus = 0;
1553 1 : s32 raRet = 0;
1554 1 : constexpr uint32_t timeoutSec = 120;
1555 1 : constexpr auto timeout = std::chrono::seconds(timeoutSec);
1556 1 : auto startTime = std::chrono::steady_clock::now();
1557 1 : HCCL_INFO("HostCpuRoceChannel: waiting for qp status ready...");
1558 : while (true) {
1559 3 : CHK_PRT_RET(needStop(), HCCL_ERROR("Terminating operation due to external request"), HCCL_E_INTERNAL);
1560 :
1561 2 : if ((std::chrono::steady_clock::now() - startTime) >= timeout) {
1562 0 : HCCL_ERROR("[Connect][Qp]get qp status timeout_=%lld, qp_status=%d", timeout, qpStatus);
1563 0 : if (!hasSocket) {
1564 0 : SocketMgr::GetInstance(devicePhyId_).PutSocket(socketConfig_, socket_);
1565 : }
1566 0 : return HCCL_E_TIMEOUT;
1567 : }
1568 2 : raRet = hrtGetRaQpStatus(qpInfo.qpHandle, &qpStatus);
1569 2 : if ((!raRet) && (qpStatus == 1)) { // 为1时，qp 建链成功
1570 0 : HCCL_INFO("In link ibv, QP get status success.");
1571 0 : break;
1572 : } else {
1573 2 : SaluSleep(GET_QP_SLEEP_TIME);
1574 : }
1575 : }
1576 0 : if (!hasSocket) {
1577 0 : SocketMgr::GetInstance(devicePhyId_).PutSocket(socketConfig_, socket_);
1578 : }
1579 0 : return HCCL_SUCCESS;
1580 : }
1581 :
1582 5 : HcclResult HostCpuRoceChannel::ExchangeDataHybird()
1583 : {
1584 5 : HCCL_INFO("[Hybrid] Starting hybrid data exchange");
1585 :
1586 5 : CHK_RET(RegisterUserMemHybird());
1587 :
1588 4 : CHK_RET(BuildExchangeDataHybird());
1589 :
1590 3 : CHK_PRT_RET(!socket_->Send(exchangeDataForSend_.data(), exchangeDataTotalSize_),
1591 : HCCL_ERROR("[Hybrid] Send exchange data failed"), HCCL_E_NETWORK);
1592 :
1593 2 : exchangeDataForRecv_.resize(exchangeDataTotalSize_);
1594 2 : CHK_PRT_RET(!socket_->Recv(exchangeDataForRecv_.data(), exchangeDataTotalSize_),
1595 : HCCL_ERROR("[Hybrid] Recv exchange data failed"), HCCL_E_NETWORK);
1596 :
1597 1 : CHK_RET(ParseRecvExchangeDataHybird());
1598 :
1599 0 : return HCCL_SUCCESS;
1600 : }
1601 :
1602 : constexpr u32 WQE_NUM_STEP = 2; // 增加步长
1603 5 : HcclResult HostCpuRoceChannel::WriteWithNotifyHybrid(
1604 : void *dst, const void *src, uint64_t len, uint32_t remoteNotifyIdx)
1605 : {
1606 5 : CHK_PTR_NULL(src);
1607 4 : CHK_PTR_NULL(dst);
1608 3 : HCCL_INFO("[Hybrid] WriteWithNotifyHybrid start, len=%lu", len);
1609 :
1610 : // 参数校验
1611 3 : CHK_PRT_RET(localRmaBuffers_.empty(),
1612 : HCCL_ERROR("[Hybrid] localRmaBuffer is Empty"),
1613 : HCCL_E_ROCE_CONNECT);
1614 :
1615 3 : std::vector<Hccl::QpInfo> qpInfo = GetQpInfos();
1616 3 : CHK_PRT_RET(qpInfo.empty(),
1617 : HCCL_ERROR("[Hybrid] qpInfos is Empty"),
1618 : HCCL_E_ROCE_CONNECT);
1619 :
1620 : // 获取本地 buffer 信息
1621 2 : hccl::MemType type = NotifyIdToMemtypeHybird(remoteNotifyIdx);
1622 :
1623 : // 校验数据长度
1624 2 : CHK_PRT_RET(len > localRmaBuffers_[0]->GetSize(),
1625 : HCCL_ERROR("[Hybrid] Data length %lu exceeds buffer size %lu", len, localRmaBuffers_[0]->GetSize()),
1626 : HCCL_E_PARA);
1627 :
1628 : // 构造发送 WR 链：数据 WR + Notify WR
1629 0 : struct ibv_send_wr dataWr{};
1630 0 : struct ibv_send_wr notifyWr{};
1631 0 : struct ibv_send_wr *badWr = nullptr;
1632 0 : struct ibv_sge dataSge{};
1633 0 : struct ibv_sge notifySge{};
1634 :
1635 : // 1. 数据 WR（RDMA Write）
1636 0 : dataSge.addr = reinterpret_cast<uint64_t>(src);
1637 0 : dataSge.length = len;
1638 0 : dataSge.lkey = localRmaBuffers_[0]->GetLkey();
1639 :
1640 0 : dataWr.wr_id = 0;
1641 0 : dataWr.opcode = IBV_WR_RDMA_WRITE;
1642 0 : dataWr.send_flags = IBV_SEND_SIGNALED; // 需要 CQE 确认完成
1643 0 : dataWr.sg_list = &dataSge;
1644 0 : dataWr.num_sge = 1;
1645 0 : dataWr.wr.rdma.remote_addr = reinterpret_cast<uint64_t>(dst);
1646 0 : dataWr.wr.rdma.rkey = rmtRmaBuffers_[0]->GetRkey();
1647 :
1648 : // 2. Notify WR（写入对端 TransportIbverbs 的 Notify 内存）
1649 0 : notifySge.addr = reinterpret_cast<uint64_t>(localMemMsg_[hccl::NOTIFY_SRC_MEM].addr);
1650 0 : notifySge.length = localMemMsg_[hccl::NOTIFY_SRC_MEM].len;
1651 0 : notifySge.lkey = localMemMsg_[hccl::NOTIFY_SRC_MEM].lkey; // 使用本地 buffer 的 lkey
1652 :
1653 0 : notifyWr.wr_id = 1;
1654 0 : notifyWr.opcode = IBV_WR_RDMA_WRITE;
1655 0 : notifyWr.send_flags = IBV_SEND_SIGNALED;
1656 0 : notifyWr.sg_list = ¬ifySge;
1657 0 : notifyWr.num_sge = 1;
1658 : // Notify 写入对端 hostNotifyAddr 的偏移位置
1659 0 : notifyWr.wr.rdma.remote_addr = reinterpret_cast<uint64_t>(remoteMemMsg_[type].addr);
1660 0 : notifyWr.wr.rdma.rkey = remoteMemMsg_[type].lkey;
1661 :
1662 : // 链接 WR 链：dataWr -> notifyWr
1663 0 : dataWr.next = ¬ifyWr;
1664 0 : notifyWr.next = nullptr;
1665 :
1666 : // 3. 下发 WR 链
1667 0 : int32_t ret = ibv_post_send(qpInfo[0].qp, &dataWr, &badWr);
1668 0 : CHK_PRT_RET(ret != 0,
1669 : HCCL_ERROR("[Hybrid] ibv_post_send failed, ret=%d", ret),
1670 : HCCL_E_NETWORK);
1671 :
1672 0 : wqeNums_[0] += WQE_NUM_STEP;
1673 :
1674 0 : HCCL_INFO("[Hybrid] WriteWithNotifyHybrid success");
1675 0 : return HCCL_SUCCESS;
1676 3 : }
1677 :
1678 2 : HcclResult HostCpuRoceChannel::NotifyWaitHybrid(uint32_t localNotifyIdx, uint32_t timeout)
1679 : {
1680 2 : HCCL_INFO("[Hybrid] NotifyWaitHybrid start, idx=%u", localNotifyIdx);
1681 :
1682 2 : hccl::MemType type = NotifyIdToMemtypeHybird(localNotifyIdx);
1683 :
1684 : // 使用配置的超时时间和轮询间隔
1685 2 : uint32_t pollTimeout = (timeout == 0) ? 30000 : timeout;
1686 2 : uint32_t pollInterval = 1;
1687 :
1688 : // 使用原子操作读取 Notify 内存
1689 : std::atomic<uint32_t>* notifyAddr =
1690 2 : reinterpret_cast<std::atomic<uint32_t>*>(localMemMsg_[type].addr);
1691 2 : const uint64_t expectedValue = 1;
1692 :
1693 2 : auto startTime = std::chrono::steady_clock::now();
1694 2 : auto waitTime = std::chrono::milliseconds(pollTimeout);
1695 :
1696 : while (true) {
1697 : // 使用原子操作读取，确保内存可见性
1698 80 : if (notifyAddr->load(std::memory_order_acquire) == expectedValue) {
1699 : // 读取成功后清零，为下一次通知做准备
1700 0 : notifyAddr->store(0, std::memory_order_release);
1701 0 : HCCL_INFO("[Hybrid] NotifyWaitHybrid success");
1702 0 : return HCCL_SUCCESS;
1703 : }
1704 :
1705 : // 检查超时
1706 40 : if ((std::chrono::steady_clock::now() - startTime) >= waitTime) {
1707 2 : HCCL_ERROR("[Hybrid] NotifyWaitHybrid timeout, notify idx:%d", localNotifyIdx);
1708 2 : return HCCL_E_TIMEOUT;
1709 : }
1710 :
1711 38 : SaluSleep(pollInterval);
1712 : }
1713 : }
1714 :
1715 : } // namespace hcomm
|
