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