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