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