Line data Source code
1 : /**
2 : * Copyright (c) 2026 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 "endpoint.h"
12 : #include "aicpu_res_package_helper.h"
13 : #include "hcomm_c_adpt.h"
14 : #include "exception_handler.h"
15 : #include "mem_transport_common.h"
16 : #include "acl_device_slab_guard.h"
17 :
18 : #include "acl/acl_rt.h"
19 :
20 : // Orion
21 : #include "exchange_rdma_buffer_dto.h"
22 : #include "dev_capability.h"
23 : #include "orion_adapter_rts.h"
24 : #include "aicpu_ts_roce_channel_v2.h"
25 : #include "../../../../common/orion_adpt_utils.h"
26 : #include "../../sockets/socket_mgr.h"
27 : #include "user_remote_mem_getter.h"
28 : #include "adapter_rts.h"
29 :
30 : namespace hcomm {
31 :
32 : constexpr uint16_t DEFAULT_LISTENING_PORT = 60001;
33 : constexpr uint32_t TC_TEMP = 132;
34 : constexpr uint32_t SL_TEMP = 4;
35 : constexpr uint32_t RETRY_CNT_TEMP = 7;
36 : constexpr uint32_t RETRY_TIME_TEMP = 20;
37 :
38 : namespace {
39 : constexpr size_t AICPU_TS_ROCE_ENTITY_ALIGN_SIZE = 64;
40 :
41 : struct DeviceEntitySection {
42 : size_t offset{0};
43 : size_t size{0};
44 : };
45 :
46 : struct DeviceChannelEntityLayout {
47 : DeviceEntitySection entitySection{0, sizeof(ChannelEntity)};
48 : DeviceEntitySection localNotifySection;
49 : DeviceEntitySection remoteNotifySection;
50 : DeviceEntitySection localBufferSection;
51 : DeviceEntitySection remoteBufferSection;
52 : DeviceEntitySection sqContextSection;
53 : DeviceEntitySection cqContextSection;
54 : size_t slabSize{0};
55 : };
56 :
57 80 : size_t AlignUp(size_t value, size_t alignment) { return (value + alignment - 1) / alignment * alignment; }
58 :
59 60 : HcclResult AddDeviceEntitySection(
60 : size_t elemSize, uint32_t elemNum, size_t& offset, DeviceEntitySection& section, const char* sectionName)
61 : {
62 60 : section.offset = AlignUp(offset, AICPU_TS_ROCE_ENTITY_ALIGN_SIZE);
63 60 : if (elemNum == 0) {
64 20 : section.size = 0;
65 20 : offset = section.offset;
66 20 : return HCCL_SUCCESS;
67 : }
68 40 : CHK_PRT_RET(
69 : elemSize != 0 && elemNum > (SIZE_MAX / elemSize),
70 : HCCL_ERROR(
71 : "[AicpuTsRoceChannelV2::AddDeviceEntitySection] %s size overflow, elemSize[%zu], elemNum[%u]",
72 : sectionName, elemSize, elemNum),
73 : HCCL_E_PARA);
74 40 : section.size = elemSize * static_cast<size_t>(elemNum);
75 40 : CHK_PRT_RET(
76 : section.offset > (SIZE_MAX - section.size),
77 : HCCL_ERROR(
78 : "[AicpuTsRoceChannelV2::AddDeviceEntitySection] %s offset overflow, offset[%zu], size[%zu]",
79 : sectionName, section.offset, section.size),
80 : HCCL_E_PARA);
81 40 : offset = section.offset + section.size;
82 40 : return HCCL_SUCCESS;
83 : }
84 :
85 30 : void* GetSlabPtr(void* base, const DeviceEntitySection& section)
86 : {
87 30 : if (section.size == 0) {
88 0 : return nullptr;
89 : }
90 30 : return reinterpret_cast<void*>(reinterpret_cast<uintptr_t>(base) + section.offset);
91 : }
92 :
93 : template <typename T>
94 36 : HcclResult CopyArrayToSlab(
95 : void* slabBase, const T* hostArray, uint32_t arrayNum, const DeviceEntitySection& section, T** deviceArrayPtr,
96 : const char* arrayName)
97 : {
98 36 : CHK_PTR_NULL(deviceArrayPtr);
99 36 : if (arrayNum == 0 || hostArray == nullptr) {
100 12 : CHK_PRT_RET(
101 : arrayNum != 0,
102 : HCCL_ERROR(
103 : "[AicpuTsRoceChannelV2::CopyArrayToSlab] %s hostArray is nullptr, num[%u]", arrayName, arrayNum),
104 : HCCL_E_PTR);
105 12 : *deviceArrayPtr = nullptr;
106 12 : return HCCL_SUCCESS;
107 : }
108 24 : CHK_PRT_RET(
109 : section.size != static_cast<size_t>(arrayNum) * sizeof(T),
110 : HCCL_ERROR(
111 : "[AicpuTsRoceChannelV2::CopyArrayToSlab] %s size mismatch, sectionSize[%zu], expect[%zu]", arrayName,
112 : section.size, static_cast<size_t>(arrayNum) * sizeof(T)),
113 : HCCL_E_PARA);
114 24 : void* sectionPtr = GetSlabPtr(slabBase, section);
115 24 : CHK_PTR_NULL(sectionPtr);
116 24 : Hccl::HrtMemcpy(
117 24 : sectionPtr, section.size, hostArray, section.size, Hccl::tagRtMemcpyKind::RT_MEMCPY_HOST_TO_DEVICE);
118 24 : *deviceArrayPtr = reinterpret_cast<T*>(sectionPtr);
119 24 : HCCL_INFO(
120 : "[AicpuTsRoceChannelV2::CopyArrayToSlab] %s: host[%p] -> dev[%p], num[%u], size[%zu]", arrayName, hostArray,
121 : sectionPtr, arrayNum, section.size);
122 24 : return HCCL_SUCCESS;
123 : }
124 :
125 10 : HcclResult BuildDeviceChannelEntityLayout(const ChannelEntity& hostChannel, DeviceChannelEntityLayout& layout)
126 : {
127 10 : layout.slabSize = AlignUp(sizeof(ChannelEntity), AICPU_TS_ROCE_ENTITY_ALIGN_SIZE);
128 10 : CHK_RET(AddDeviceEntitySection(
129 : sizeof(RegedNotifyEntity), hostChannel.localNotifyNum, layout.slabSize, layout.localNotifySection,
130 : "localNotifyAddr"));
131 10 : CHK_RET(AddDeviceEntitySection(
132 : sizeof(RegedNotifyEntity), hostChannel.remoteNotifyNum, layout.slabSize, layout.remoteNotifySection,
133 : "remoteNotifyAddr"));
134 10 : CHK_RET(AddDeviceEntitySection(
135 : sizeof(RegedBufferEntity), hostChannel.localBufferNum, layout.slabSize, layout.localBufferSection,
136 : "localBufferAddr"));
137 10 : CHK_RET(AddDeviceEntitySection(
138 : sizeof(RegedBufferEntity), hostChannel.remoteBufferNum, layout.slabSize, layout.remoteBufferSection,
139 : "remoteBufferAddr"));
140 10 : CHK_RET(AddDeviceEntitySection(
141 : sizeof(SqContext), hostChannel.sqNum, layout.slabSize, layout.sqContextSection, "sqContextAddr"));
142 10 : CHK_RET(AddDeviceEntitySection(
143 : sizeof(CqContext), hostChannel.cqNum, layout.slabSize, layout.cqContextSection, "cqContextAddr"));
144 10 : layout.slabSize = AlignUp(layout.slabSize, AICPU_TS_ROCE_ENTITY_ALIGN_SIZE);
145 10 : return HCCL_SUCCESS;
146 : }
147 :
148 9 : HcclResult AllocDeviceEntitySlab(size_t slabSize, AclDeviceSlabGuard& slabGuard, void*& slabPtr)
149 : {
150 9 : HcclResult ret = hrtMalloc(&slabPtr, slabSize);
151 9 : CHK_PRT_RET(
152 : ret != HCCL_SUCCESS || slabPtr == nullptr,
153 : HCCL_ERROR("[AicpuTsRoceChannelV2::%s] hrtMalloc slab failed, ret[%d], size[%zu]", __func__, ret, slabSize),
154 : HCCL_E_MEMORY);
155 8 : slabGuard.Reset(slabPtr, slabSize);
156 8 : return HCCL_SUCCESS;
157 : }
158 :
159 6 : HcclResult CopyChannelEntityArrayToSlab(
160 : void* slabPtr, const ChannelEntity& hostChannel, const DeviceChannelEntityLayout& layout,
161 : ChannelEntity& devChannel)
162 : {
163 6 : devChannel = hostChannel;
164 6 : CHK_RET(CopyArrayToSlab(
165 : slabPtr, hostChannel.localNotifyAddr, hostChannel.localNotifyNum, layout.localNotifySection,
166 : &devChannel.localNotifyAddr, "localNotifyAddr"));
167 6 : CHK_RET(CopyArrayToSlab(
168 : slabPtr, hostChannel.remoteNotifyAddr, hostChannel.remoteNotifyNum, layout.remoteNotifySection,
169 : &devChannel.remoteNotifyAddr, "remoteNotifyAddr"));
170 6 : CHK_RET(CopyArrayToSlab(
171 : slabPtr, hostChannel.localBufferAddr, hostChannel.localBufferNum, layout.localBufferSection,
172 : &devChannel.localBufferAddr, "localBufferAddr"));
173 6 : CHK_RET(CopyArrayToSlab(
174 : slabPtr, hostChannel.remoteBufferAddr, hostChannel.remoteBufferNum, layout.remoteBufferSection,
175 : &devChannel.remoteBufferAddr, "remoteBufferAddr"));
176 6 : CHK_RET(CopyArrayToSlab(
177 : slabPtr, hostChannel.sqContextAddr, hostChannel.sqNum, layout.sqContextSection, &devChannel.sqContextAddr,
178 : "sqContextAddr"));
179 6 : CHK_RET(CopyArrayToSlab(
180 : slabPtr, hostChannel.cqContextAddr, hostChannel.cqNum, layout.cqContextSection, &devChannel.cqContextAddr,
181 : "cqContextAddr"));
182 6 : return HCCL_SUCCESS;
183 : }
184 :
185 6 : HcclResult CopyChannelEntityToSlab(
186 : void* slabPtr, const DeviceChannelEntityLayout& layout, const ChannelEntity& devChannel, void*& entityDevPtr)
187 : {
188 6 : entityDevPtr = GetSlabPtr(slabPtr, layout.entitySection);
189 6 : CHK_PTR_NULL(entityDevPtr);
190 6 : Hccl::HrtMemcpy(
191 : entityDevPtr, sizeof(ChannelEntity), &devChannel, sizeof(ChannelEntity),
192 : Hccl::tagRtMemcpyKind::RT_MEMCPY_HOST_TO_DEVICE);
193 6 : return HCCL_SUCCESS;
194 : }
195 : } // namespace
196 :
197 41 : AicpuTsRoceChannelV2::AicpuTsRoceChannelV2(
198 41 : EndpointHandle endpointHandle, HcommChannelDesc channelDesc, CommEngine engine)
199 41 : : endpointHandle_(endpointHandle),
200 41 : channelDesc_(channelDesc),
201 41 : engine_(engine)
202 41 : {}
203 :
204 82 : AicpuTsRoceChannelV2::~AicpuTsRoceChannelV2()
205 : {
206 41 : FreeDeviceMemories();
207 41 : if (channelDesc_.socket == nullptr && socket_ != nullptr) {
208 1 : SocketMgr::GetInstance(devicePhyId_).PutSocket(socketConfig_, socket_);
209 1 : socket_ = nullptr;
210 : }
211 82 : }
212 :
213 35 : HcclResult AicpuTsRoceChannelV2::ParseInputParam()
214 : {
215 : // 1. 从 endpointHandle_，获得 localEp_ 和 rdmaHandle_
216 35 : CHK_PTR_NULL(endpointHandle_);
217 35 : HCCL_INFO(
218 : "[AicpuTsRoceChannelV2][%s] Start. endpointHandle[0x%llx]", __func__,
219 : reinterpret_cast<uint64_t>(endpointHandle_));
220 35 : Endpoint* localEpPtr = reinterpret_cast<Endpoint*>(endpointHandle_);
221 35 : localEp_ = localEpPtr->GetEndpointDesc();
222 35 : rdmaHandle_ = localEpPtr->GetRdmaHandle();
223 35 : CHK_PTR_NULL(rdmaHandle_);
224 :
225 : // 2. 从 channelDesc_，获得 remoteEp_, socket_ 和 notifyNum_
226 35 : remoteEp_ = channelDesc_.remoteEndpoint;
227 35 : socket_ = reinterpret_cast<Hccl::Socket*>(channelDesc_.socket);
228 35 : notifyNum_ = channelDesc_.notifyNum;
229 :
230 35 : return HCCL_SUCCESS;
231 : }
232 :
233 0 : HcclResult AicpuTsRoceChannelV2::StartListen()
234 : {
235 0 : uint16_t port = channelDesc_.port;
236 0 : HCCL_INFO(
237 : "[AicpuTsRoceChannelV2::%s] Start. EndpointHandle[0x%llx], port[%u]", __func__,
238 : reinterpret_cast<uint64_t>(endpointHandle_), port);
239 0 : if (port == 0) {
240 0 : port = DEFAULT_LISTENING_PORT;
241 0 : HCCL_INFO("[AicpuTsRoceChannelV2::%s] channelDesc port is 0, use default port [%u]", __func__, port);
242 : }
243 0 : CHK_RET(static_cast<HcclResult>(HcommEndpointStartListen(endpointHandle_, port, nullptr)));
244 0 : HCCL_INFO("[AicpuTsRoceChannelV2::%s] SUCCESS. port[%u].", __func__, port);
245 0 : return HCCL_SUCCESS;
246 : }
247 :
248 35 : HcclResult AicpuTsRoceChannelV2::BuildSocket()
249 : {
250 35 : if (socket_ != nullptr) {
251 34 : return HCCL_SUCCESS;
252 : }
253 1 : HCCL_INFO("[AicpuTsRoceChannelV2::%s] socket ptr is NULL, rebuild Socket", __func__);
254 :
255 1 : Hccl::LinkData linkData = BuildDefaultLinkData();
256 1 : CHK_RET(EndpointDescPairToLinkData(localEp_, remoteEp_, linkData));
257 1 : HCCL_INFO("[AicpuTsRoceChannelV2::%s] built linkData: %s", __func__, linkData.Describe().c_str());
258 1 : uint16_t port = channelDesc_.port;
259 1 : if (port == 0) {
260 1 : port = DEFAULT_LISTENING_PORT;
261 1 : HCCL_INFO("[AicpuTsRoceChannelV2::%s] channelDesc port is 0, use default port [%u]", __func__, port);
262 : }
263 : std::string socketTag
264 3 : = (channelDesc_.channelName != nullptr) ? std::string(channelDesc_.channelName) : "AUTOMATIC_SOCKET_TAG";
265 : Hccl::SocketConfig socketConfig
266 1 : = (channelDesc_.role != HCOMM_SOCKET_ROLE_RESERVED) ?
267 1 : Hccl::SocketConfig(linkData, port, socketTag, channelDesc_.role == HCOMM_SOCKET_ROLE_SERVER) :
268 1 : Hccl::SocketConfig(linkData, port, socketTag);
269 1 : CHK_RET(SocketMgr::GetInstance(devicePhyId_).GetSocket(socketConfig, socket_));
270 1 : HCCL_INFO("[AicpuTsRoceChannelV2::%s] SUCCESS. port[%u].", __func__, port);
271 1 : return HCCL_SUCCESS;
272 1 : }
273 :
274 35 : HcclResult AicpuTsRoceChannelV2::BuildConnection()
275 : {
276 35 : std::unique_ptr<DevRdmaConnectionV2> conn;
277 35 : EXCEPTION_CATCH(
278 : conn = std::make_unique<DevRdmaConnectionV2>(socket_, rdmaHandle_, channelDesc_.roceAttr.cqAttrFlags),
279 : return HCCL_E_INTERNAL);
280 35 : CHK_PTR_NULL(conn);
281 35 : CHK_RET(conn->Init());
282 35 : Hccl::QpInfo& qpInfo = conn->GetQpInfo();
283 35 : qpInfo.serviceLevel = channelDesc_.roceAttr.sl == 0xFF ? SL_TEMP : channelDesc_.roceAttr.sl;
284 35 : qpInfo.trafficClass = channelDesc_.roceAttr.tc == 0xFF ? TC_TEMP : channelDesc_.roceAttr.tc;
285 35 : qpInfo.retryCnt = channelDesc_.roceAttr.retryCnt == 0xFFFFFFFF ? RETRY_CNT_TEMP : channelDesc_.roceAttr.retryCnt;
286 : qpInfo.retryInterval
287 35 : = channelDesc_.roceAttr.retryInterval == 0xFFFFFFFF ? RETRY_TIME_TEMP : channelDesc_.roceAttr.retryInterval;
288 35 : HCCL_INFO(
289 : "[AicpuTsRoceChannelV2::BuildConnection] QpInfo: serviceLevel[%u], trafficClass[%u], retryCnt[%u], "
290 : "retryInterval[%u].",
291 : qpInfo.serviceLevel, qpInfo.trafficClass, qpInfo.retryCnt, qpInfo.retryInterval);
292 35 : connections_.emplace_back(std::move(conn));
293 35 : connNum_ = connections_.size();
294 35 : HCCL_INFO("[AicpuTsRoceChannelV2::%s] connection num [%u]", __func__, connNum_);
295 35 : return HCCL_SUCCESS;
296 35 : }
297 :
298 35 : HcclResult AicpuTsRoceChannelV2::BuildNotify()
299 : {
300 35 : if (engine_ == COMM_ENGINE_AIV) {
301 0 : return HCCL_SUCCESS;
302 : }
303 :
304 35 : CHK_PRT_RET(
305 : notifyNum_ != RDMA_NOTIFY_NUM,
306 : HCCL_ERROR(
307 : "[AicpuTsRoceChannelV2::%s] rdma notify num false, actual num [%u], expected num [%u]", __func__,
308 : notifyNum_, RDMA_NOTIFY_NUM),
309 : HCCL_E_PARA);
310 :
311 35 : localNotifies_.clear();
312 35 : bool devUsed = true;
313 140 : for (uint32_t i = 0; i < notifyNum_; ++i) {
314 105 : std::unique_ptr<Hccl::RdmaLocalNotify> notifyPtr = nullptr;
315 105 : EXCEPTION_CATCH(notifyPtr = std::make_unique<Hccl::RdmaLocalNotify>(rdmaHandle_, devUsed), return HCCL_E_PTR);
316 105 : localNotifies_.emplace_back(std::move(notifyPtr));
317 105 : }
318 35 : HCCL_INFO("[AicpuTsRoceChannelV2::%s] notify num [%u]", __func__, notifyNum_);
319 35 : return HCCL_SUCCESS;
320 : }
321 :
322 35 : HcclResult AicpuTsRoceChannelV2::BuildBuffer()
323 : {
324 35 : if (channelDesc_.exchangeAllMems) {
325 : // Get memHandles from endpoint
326 0 : HCCL_INFO("[AicpuTsRoceChannelV2][%s] exchangeAllMems == True. Get memHandles from endpoint.", __func__);
327 0 : std::shared_ptr<Hccl::LocalRdmaRmaBuffer>* memHandles = nullptr;
328 0 : uint32_t memHandleNum = 0;
329 0 : CHK_RET(static_cast<HcclResult>(
330 : HcommMemGetAllMemHandles(endpointHandle_, reinterpret_cast<void**>(&memHandles), &memHandleNum)));
331 0 : HCCL_INFO("[AicpuTsRoceChannelV2][%s] Got memHandleNum[%u].", __func__, memHandleNum);
332 0 : for (uint32_t i = 0; i < memHandleNum; ++i) {
333 0 : std::shared_ptr<Hccl::LocalRdmaRmaBuffer>& localRdmaBuffer = memHandles[i];
334 0 : CHK_SMART_PTR_NULL(localRdmaBuffer);
335 0 : Hccl::Buffer* buf = localRdmaBuffer->GetBuf();
336 0 : CHK_PTR_NULL(buf);
337 0 : HCCL_INFO(
338 : "[AicpuTsRoceChannelV2][%s] Got memHandle No.%u: addr[0x%llx], size[0x%llx], memType[%d], memInfo[%s].",
339 : __func__, i, static_cast<unsigned long long>(localRdmaBuffer->GetAddr()),
340 : static_cast<unsigned long long>(localRdmaBuffer->GetSize()), static_cast<int>(buf->GetMemType()),
341 : buf->GetMemInfo().c_str());
342 0 : localRmaBuffers_.emplace_back(localRdmaBuffer.get());
343 : }
344 : } else {
345 : // 从 channelDesc 的 memHandle，获得 localRmaBuffers_
346 35 : HCCL_INFO("[AicpuTsRoceChannelV2][%s] exchangeAllMems == false. Get memHandles from channelDesc.", __func__);
347 35 : CHK_PTR_NULL(channelDesc_.memHandles);
348 70 : for (uint32_t i = 0; i < channelDesc_.memHandleNum; ++i) {
349 35 : CHK_PTR_NULL(channelDesc_.memHandles[i]);
350 35 : auto* localRdmaBuffer = reinterpret_cast<Hccl::LocalRdmaRmaBuffer*>(channelDesc_.memHandles[i]);
351 35 : HCCL_INFO(
352 : "[AicpuTsRoceChannelV2][%s] Got memHandle No.%u: addr[0x%llx], size[0x%llx], memType[%d], memInfo[%s].",
353 : __func__, i, static_cast<unsigned long long>(localRdmaBuffer->GetAddr()),
354 : static_cast<unsigned long long>(localRdmaBuffer->GetSize()),
355 : static_cast<int>(localRdmaBuffer->GetBuf()->GetMemType()),
356 : localRdmaBuffer->GetBuf()->GetMemInfo().c_str());
357 35 : localRmaBuffers_.emplace_back(localRdmaBuffer);
358 : }
359 : }
360 35 : bufferNum_ = localRmaBuffers_.size();
361 35 : HCCL_INFO("[AicpuTsRoceChannelV2::%s] buffer num [%u]", __func__, bufferNum_);
362 35 : return HCCL_SUCCESS;
363 : }
364 :
365 0 : HcclResult AicpuTsRoceChannelV2::BuildNotifyValueBuffer()
366 : {
367 0 : if (engine_ == COMM_ENGINE_AIV) {
368 0 : return HCCL_SUCCESS;
369 : }
370 :
371 0 : Hccl::DevCapability::GetInstance().Init(Hccl::HrtGetDeviceType());
372 0 : u32 notifysize = Hccl::DevCapability::GetInstance().GetNotifySize();
373 0 : EXCEPTION_CATCH((notifyValueMem_ = std::make_shared<Hccl::DevBuffer>(notifysize)), return HCCL_E_PTR);
374 0 : HCCL_DEBUG(
375 : "[AicpuTsRoceChannelV2::%s] create notify value buffer[%p], size[%u]", __func__, notifyValueMem_.get(),
376 : notifyValueMem_->GetSize());
377 0 : u64 notifyValue = 1; // notify值写1表示record
378 0 : Hccl::HrtMemcpy(
379 0 : reinterpret_cast<void*>(notifyValueMem_->GetAddr()), notifyValueMem_->GetSize(), ¬ifyValue, notifysize,
380 : Hccl::tagRtMemcpyKind::RT_MEMCPY_HOST_TO_DEVICE);
381 0 : EXCEPTION_CATCH(
382 : (notifyValueBuffer_ = std::make_unique<Hccl::LocalRdmaRmaBuffer>(notifyValueMem_, rdmaHandle_)),
383 : return HCCL_E_PTR);
384 0 : HCCL_INFO("[AicpuTsRoceChannelV2::%s] build notify value buffer success.", __func__);
385 0 : return HCCL_SUCCESS;
386 : }
387 :
388 35 : HcclResult AicpuTsRoceChannelV2::Init()
389 : {
390 35 : s32 devLogicId = Hccl::HrtGetDevice();
391 35 : devicePhyId_ = Hccl::HrtGetDevicePhyIdByIndex(static_cast<u32>(devLogicId));
392 :
393 35 : CHK_RET(ParseInputParam());
394 35 : if (channelDesc_.exchangeAllMems && channelDesc_.role != HCOMM_SOCKET_ROLE_CLIENT) {
395 0 : CHK_RET(StartListen());
396 : }
397 35 : CHK_RET(BuildSocket());
398 35 : CHK_RET(BuildConnection());
399 35 : CHK_RET(BuildNotify());
400 35 : CHK_RET(BuildBuffer());
401 35 : CHK_RET(BuildNotifyValueBuffer());
402 35 : return HCCL_SUCCESS;
403 : }
404 :
405 : // 当前AICPU和框架没有改为返回错误码形式，所有暂时使用该方法转换
406 4 : ChannelStatus AicpuTsRoceChannelV2::GetStatus()
407 : {
408 4 : ChannelStatus status;
409 4 : HcclResult ret = GetStatus(status);
410 4 : if (ret != HCCL_SUCCESS && ret != HCCL_E_AGAIN) {
411 0 : HCCL_ERROR("[AicpuTsRoceChannelV2::GetStatus] get status exception occurred, HcclResult=[%d]", ret);
412 0 : return ChannelStatus::FAILED;
413 : }
414 4 : return status;
415 : }
416 :
417 4 : HcclResult AicpuTsRoceChannelV2::ProcessStatus()
418 : {
419 4 : switch (channelStatus_) {
420 1 : case ChannelStatus::READY:
421 1 : return HCCL_SUCCESS;
422 0 : case ChannelStatus::SOCKET_TIMEOUT:
423 0 : HCCL_ERROR("[AicpuTsRoceChannelV2::ProcessStatus] get socket timeout");
424 0 : return HCCL_E_ROCE_CONNECT;
425 3 : default:
426 3 : return HCCL_E_AGAIN;
427 : }
428 : }
429 :
430 4 : HcclResult AicpuTsRoceChannelV2::GetStatus(ChannelStatus& status)
431 : {
432 4 : switch (rdmaStatus_) {
433 1 : case RdmaStatus::INIT:
434 : // 检查socket状态
435 1 : CHK_RET(CheckSocketStatus());
436 1 : break;
437 1 : case RdmaStatus::SOCKET_OK:
438 : // 准备资源
439 1 : CHK_RET(CreateQp());
440 1 : rdmaStatus_ = RdmaStatus::QP_CREATED;
441 1 : break;
442 1 : case RdmaStatus::QP_CREATED:
443 : // 发送交换数据
444 1 : CHK_RET(ExchangeData());
445 1 : rdmaStatus_ = RdmaStatus::DATA_EXCHANGE;
446 1 : break;
447 1 : case RdmaStatus::DATA_EXCHANGE:
448 1 : CHK_RET(ModifyQp());
449 1 : rdmaStatus_ = RdmaStatus::QP_MODIFIED;
450 : [[fallthrough]];
451 1 : case RdmaStatus::QP_MODIFIED:
452 : default:
453 1 : rdmaStatus_ = RdmaStatus::CONN_OK;
454 1 : channelStatus_ = ChannelStatus::READY;
455 : }
456 :
457 4 : status = channelStatus_;
458 4 : return ProcessStatus();
459 : }
460 :
461 1 : HcclResult AicpuTsRoceChannelV2::CheckSocketStatus()
462 : {
463 1 : CHK_PTR_NULL(socket_);
464 1 : Hccl::SocketStatus socketStatus = socket_->GetStatus(); // socket状态机
465 1 : HCCL_DEBUG("[AicpuTsRoceChannelV2::CheckSocketStatus] socket status = %s", socketStatus.Describe().c_str());
466 1 : if (socketStatus == Hccl::SocketStatus::OK) {
467 1 : rdmaStatus_ = RdmaStatus::SOCKET_OK;
468 1 : channelStatus_ = ChannelStatus::SOCKET_OK;
469 0 : } else if (socketStatus == Hccl::SocketStatus::TIMEOUT) {
470 0 : channelStatus_ = ChannelStatus::SOCKET_TIMEOUT;
471 : }
472 1 : return HCCL_SUCCESS;
473 : }
474 :
475 : // 准备资源（创建QP）
476 1 : HcclResult AicpuTsRoceChannelV2::CreateQp()
477 : {
478 2 : for (auto& conn : connections_) {
479 1 : Hccl::CHECK_NULLPTR(
480 2 : conn, Hccl::StringFormat("[AicpuTsRoceChannelV2::%s] failed, connection pointer is nullptr", __func__));
481 1 : HcclResult ret = conn->CreateQp();
482 1 : if (ret == HCCL_E_AGAIN) {
483 0 : return HCCL_SUCCESS;
484 : }
485 1 : if (ret != HCCL_SUCCESS) {
486 0 : return ret;
487 : }
488 : }
489 1 : HCCL_INFO("[AicpuTsRoceChannelV2::%s] all connections resources connected.", __func__);
490 1 : return HCCL_SUCCESS;
491 : }
492 :
493 : // 交换数据
494 1 : HcclResult AicpuTsRoceChannelV2::ExchangeData()
495 : {
496 1 : HCCL_INFO(
497 : "[AicpuTsRoceChannelV2::%s] Start to SendExchangeData, notifyNum=%u, bufferNum=%u, connNum=%u", __func__,
498 : notifyNum_, bufferNum_, connNum_);
499 :
500 : // 同步数据打包
501 1 : Hccl::BinaryStream binaryStream;
502 1 : NotifyVecPack(binaryStream);
503 1 : CHK_RET(BufferVecPack(binaryStream));
504 1 : CHK_RET(ConnVecPack(binaryStream));
505 :
506 1 : std::vector<char> sendData{};
507 1 : binaryStream.Dump(sendData);
508 1 : uint64_t sendSize = sendData.size();
509 1 : std::vector<char> recvData{};
510 1 : uint64_t recvSize = 0;
511 :
512 : EXCEPTION_HANDLE_BEGIN
513 : // 同步发送数据包尺寸
514 1 : CHK_PRT_RET(
515 : !socket_->Send(reinterpret_cast<void*>(&sendSize), sizeof(sendSize)),
516 : HCCL_ERROR("[AicpuTsRoceChannelV2::%s] Send sendSize failed", __func__), HCCL_E_NETWORK);
517 1 : HCCL_INFO(
518 : "[AicpuTsRoceChannelV2::%s] Send size[%llu] of data success. [%llu] bytes sent.", __func__, sendSize,
519 : sizeof(sendSize));
520 :
521 : // 同步接收数据包尺寸
522 1 : CHK_PRT_RET(
523 : !socket_->Recv(reinterpret_cast<void*>(&recvSize), sizeof(recvSize)),
524 : HCCL_ERROR("[AicpuTsRoceChannelV2::%s] Recv recvSize failed", __func__), HCCL_E_NETWORK);
525 1 : HCCL_INFO(
526 : "[AicpuTsRoceChannelV2::%s] Receive size[%llu] of data success. [%llu] bytes received.", __func__, recvSize,
527 : sizeof(recvSize));
528 :
529 : // 同步发送数据
530 1 : CHK_PRT_RET(
531 : !socket_->Send(reinterpret_cast<void*>(sendData.data()), sendSize),
532 : HCCL_ERROR("[AicpuTsRoceChannelV2::%s] Send exchange data failed", __func__), HCCL_E_NETWORK);
533 1 : HCCL_INFO("[AicpuTsRoceChannelV2::%s] Send Exchange Data success. [%llu] bytes sent.", __func__, sendSize);
534 :
535 : // 同步接收数据
536 1 : HCCL_INFO("[AicpuTsRoceChannelV2::%s] Start to Receive Exchange Data", __func__);
537 1 : recvData.resize(recvSize);
538 1 : CHK_PRT_RET(
539 : !socket_->Recv(reinterpret_cast<void*>(recvData.data()), recvSize),
540 : HCCL_ERROR("[AicpuTsRoceChannelV2::%s] Recv exchange data failed", __func__), HCCL_E_NETWORK);
541 1 : HCCL_INFO("[AicpuTsRoceChannelV2::%s] Receive Exchange Data success. [%llu] bytes received.", __func__, recvSize);
542 0 : EXCEPTION_HANDLE_END
543 :
544 : // 同步数据解包
545 1 : Hccl::BinaryStream recvBinStream(recvData);
546 1 : CHK_RET(NotifyVecUnpack(recvBinStream));
547 1 : CHK_RET(RmtBufferVecUnpackProc(recvBinStream));
548 1 : CHK_RET(ConnVecUnpackProc(recvBinStream));
549 :
550 1 : HCCL_INFO("[AicpuTsRoceChannelV2::%s] Unpack exchange Data success. ", __func__);
551 1 : return HCCL_SUCCESS;
552 1 : }
553 :
554 0 : void AicpuTsRoceChannelV2::NotifyVecPack(Hccl::BinaryStream& binaryStream)
555 : {
556 0 : if (engine_ == COMM_ENGINE_AIV) {
557 0 : return;
558 : }
559 :
560 0 : binaryStream << notifyNum_;
561 0 : HCCL_INFO("start pack notifyVec");
562 0 : u32 pos = 0;
563 0 : for (auto& it : localNotifies_) {
564 0 : binaryStream << pos;
565 0 : std::unique_ptr<Hccl::Serializable> dto = it->GetExchangeDto();
566 0 : dto->Serialize(binaryStream);
567 0 : HCCL_INFO("pack notify pos=%u, dto %s", pos, dto->Describe().c_str());
568 0 : pos++;
569 0 : }
570 : }
571 :
572 0 : HcclResult AicpuTsRoceChannelV2::BufferVecPack(Hccl::BinaryStream& binaryStream)
573 : {
574 0 : binaryStream << bufferNum_;
575 0 : HCCL_INFO("[AicpuTsRoceChannelV2::%s] start to pack RmaBuffers", __func__);
576 0 : u32 pos = 0;
577 0 : for (auto& it : localRmaBuffers_) {
578 0 : binaryStream << pos;
579 0 : if (it != nullptr) { // 非空的buffer，从buffer中获取 dto
580 0 : std::unique_ptr<Hccl::Serializable> dto = it->GetExchangeDto();
581 0 : dto->Serialize(binaryStream);
582 0 : HCCL_INFO("pack buffer pos=%u dto %s", pos, dto->Describe().c_str());
583 0 : } else { // 空的buffer，dto所有字段为0(size=0)
584 0 : Hccl::ExchangeRdmaBufferDto exchangeDto;
585 0 : exchangeDto.Serialize(binaryStream);
586 0 : HCCL_INFO("pack buffer pos=%u, dto is null %s", pos, exchangeDto.Describe().c_str());
587 0 : }
588 0 : pos++;
589 : }
590 0 : HCCL_INFO("[AicpuTsRoceChannelV2::%s] pack RmaBuffers finish", __func__);
591 0 : return HCCL_SUCCESS;
592 : }
593 :
594 0 : HcclResult AicpuTsRoceChannelV2::ConnVecPack(Hccl::BinaryStream& binaryStream)
595 : {
596 0 : binaryStream << connNum_;
597 0 : HCCL_INFO("[AicpuTsRoceChannelV2::%s] start to pack connections", __func__);
598 0 : u32 pos = 0;
599 0 : for (auto& it : connections_) {
600 0 : binaryStream << pos;
601 0 : std::unique_ptr<Hccl::Serializable> dto = nullptr;
602 0 : CHK_RET(it->GetExchangeDto(dto));
603 0 : dto->Serialize(binaryStream);
604 0 : HCCL_INFO("pack connection pos=%u, dto %s", pos, dto->Describe().c_str());
605 0 : pos++;
606 0 : }
607 0 : HCCL_INFO("[AicpuTsRoceChannelV2::%s] pack connections finish", __func__);
608 0 : return HCCL_SUCCESS;
609 : }
610 :
611 0 : HcclResult AicpuTsRoceChannelV2::RmtBufferVecUnpackProc(Hccl::BinaryStream& binaryStream)
612 : {
613 : u32 rmtNum;
614 0 : binaryStream >> rmtNum;
615 :
616 0 : HCCL_INFO("[AicpuTsRoceChannelV2::%s] bufferNum_=%u, rmtNum=%u", __func__, bufferNum_, rmtNum);
617 :
618 0 : rmtRmaBuffers_.resize(rmtNum);
619 0 : for (u32 i = 0; i < rmtNum; i++) {
620 : u32 pos;
621 0 : binaryStream >> pos;
622 0 : if (pos >= rmtNum) {
623 0 : HCCL_ERROR("[AicpuTsRoceChannelV2::%s] pos=%u out of range (rmtNum=%u)", __func__, pos, rmtNum);
624 0 : return HCCL_E_INTERNAL;
625 : }
626 0 : Hccl::ExchangeRdmaBufferDto dto;
627 0 : dto.Deserialize(binaryStream);
628 :
629 0 : HCCL_INFO("[AicpuTsRoceChannelV2::%s] pos=%u, dto %s", __func__, pos, dto.Describe().c_str());
630 0 : EXCEPTION_CATCH(
631 : rmtRmaBuffers_[pos] = std::make_unique<Hccl::RemoteRdmaRmaBuffer>(rdmaHandle_, dto),
632 : HCCL_ERROR(
633 : "[AicpuTsRoceChannelV2::%s] make_unique<Hccl::RemoteRdmaRmaBuffer> throws an exception!", __func__);
634 : return HCCL_E_INTERNAL);
635 0 : HCCL_INFO(
636 : "[AicpuTsRoceChannelV2::%s] pos=%u, rmtRmaBuffer=%s", __func__, pos,
637 : rmtRmaBuffers_[pos]->Describe().c_str());
638 0 : }
639 :
640 0 : return HCCL_SUCCESS;
641 : }
642 :
643 0 : HcclResult AicpuTsRoceChannelV2::NotifyVecUnpack(Hccl::BinaryStream& binaryStream)
644 : {
645 0 : if (engine_ == COMM_ENGINE_AIV) {
646 0 : return HCCL_SUCCESS;
647 : }
648 :
649 0 : uint32_t notifySize = 0;
650 0 : binaryStream >> notifySize;
651 0 : if (notifySize != notifyNum_) {
652 0 : HCCL_ERROR(
653 : "[AicpuTsRoceChannelV2::NotifyVecUnpack] rmtNum=%u is not equal to localNum=%u", notifySize, notifyNum_);
654 0 : return HCCL_E_ROCE_CONNECT;
655 : }
656 0 : remoteNotifies_.clear();
657 0 : u32 pos = 0;
658 0 : for (pos = 0; pos < notifySize; pos++) {
659 0 : binaryStream >> pos;
660 0 : Hccl::ExchangeRdmaBufferDto dto;
661 0 : dto.Deserialize(binaryStream);
662 0 : HCCL_INFO("unpack pos=%u, dto %s", pos, dto.Describe().c_str());
663 0 : remoteNotifies_.push_back(std::make_unique<Hccl::RemoteRdmaRmaBuffer>(rdmaHandle_, dto));
664 0 : HCCL_INFO("unpack notify pos=%u, rmtRmaBuffer=%s", pos, remoteNotifies_.back()->Describe().c_str());
665 0 : }
666 0 : return HCCL_SUCCESS;
667 : }
668 :
669 0 : HcclResult AicpuTsRoceChannelV2::ConnVecUnpackProc(Hccl::BinaryStream& binaryStream)
670 : {
671 : u32 rmtConnNum;
672 0 : binaryStream >> rmtConnNum;
673 0 : HCCL_INFO("start unpack conn, connNum=%u, rmtConnNum=%u", connNum_, rmtConnNum);
674 0 : if (connNum_ != rmtConnNum) {
675 0 : HCCL_ERROR("connNum=%u is not equal to rmtConnNum=%u", connNum_, rmtConnNum);
676 0 : return HCCL_E_ROCE_CONNECT;
677 : }
678 :
679 0 : for (u32 i = 0; i < rmtConnNum; i++) {
680 : u32 pos;
681 0 : binaryStream >> pos;
682 0 : rmtConnDto_.Deserialize(binaryStream);
683 : }
684 0 : return HCCL_SUCCESS;
685 : }
686 :
687 2 : HcclResult AicpuTsRoceChannelV2::ModifyQp()
688 : {
689 4 : for (auto& conn : connections_) {
690 2 : Hccl::CHECK_NULLPTR(
691 4 : conn, Hccl::StringFormat("[AicpuTsRoceChannelV2::%s] failed, connection pointer is nullptr", __func__));
692 2 : CHK_RET(conn->ParseRmtExchangeDto(rmtConnDto_));
693 2 : HcclResult ret = conn->ModifyQp();
694 2 : if (ret == HCCL_E_AGAIN) {
695 0 : return HCCL_SUCCESS;
696 : }
697 2 : if (ret != HCCL_SUCCESS) {
698 0 : return ret;
699 : }
700 : }
701 2 : HCCL_INFO("[AicpuTsRoceChannelV2::%s] all connections resources modify success.", __func__);
702 2 : return HCCL_SUCCESS;
703 : }
704 :
705 8 : HcclResult AicpuTsRoceChannelV2::BuildAndGetLocNotifyInfo([[maybe_unused]] RegedNotifyEntity** notify)
706 : {
707 8 : return HCCL_SUCCESS;
708 : }
709 :
710 8 : HcclResult AicpuTsRoceChannelV2::BuildAndGetRmtNotifyInfo([[maybe_unused]] RegedNotifyEntity** notify)
711 : {
712 : // 目前仅用于aiv模式，无notify
713 8 : return HCCL_SUCCESS;
714 : }
715 :
716 8 : HcclResult AicpuTsRoceChannelV2::BuildAndGetRmtBufInfo(
717 : std::vector<RegedBufferEntity>& bufList, RegedBufferEntity** bufferEntityPtr)
718 : {
719 8 : if (channelStatus_ != ChannelStatus::READY) {
720 0 : HCCL_ERROR(
721 : "[AicpuTsRoceChannelV2::%s] channel status[%d] is not ready[%d], please check.", __func__, channelStatus_,
722 : ChannelStatus::READY);
723 0 : return HCCL_E_INTERNAL;
724 : }
725 :
726 8 : if (bufferNum_ == 0) {
727 0 : HCCL_INFO("[AicpuTsRoceChannelV2::%s] No Remote memory regions available", __func__);
728 0 : return HCCL_SUCCESS;
729 : }
730 :
731 8 : if (bufferEntityPtr == nullptr) {
732 0 : HCCL_ERROR("[AicpuTsRoceChannelV2::%s] input param is null", __func__);
733 0 : return HCCL_E_PARA;
734 : }
735 :
736 16 : for (uint32_t i = 0; i < bufferNum_; i++) {
737 8 : auto& rmtRmaBuffer = rmtRmaBuffers_[i];
738 8 : bufList[i].type = REGED_BUFFER_RMA;
739 8 : bufList[i].bufferInfo.rma.addr = static_cast<uint64_t>(rmtRmaBuffer->GetAddr());
740 8 : bufList[i].bufferInfo.rma.size = rmtRmaBuffer->GetSize();
741 8 : bufList[i].bufferInfo.rma.protectionInfo.type = PROTECTION_TYPE_ROCE;
742 8 : bufList[i].bufferInfo.rma.protectionInfo.memInfo.roce.rkey = rmtRmaBuffer->GetRkey();
743 8 : HCCL_INFO(
744 : "[AicpuTsRoceChannelV2::%s] rmtBuf[addr[%p], size[%lu]]", __func__, bufList[i].bufferInfo.rma.addr,
745 : bufList[i].bufferInfo.rma.size);
746 : }
747 8 : *bufferEntityPtr = bufList.data();
748 8 : return HCCL_SUCCESS;
749 : }
750 :
751 8 : HcclResult AicpuTsRoceChannelV2::BuildAndGetLocBufInfo(
752 : std::vector<RegedBufferEntity>& bufList, RegedBufferEntity** bufferEntityPtr)
753 : {
754 8 : if (channelStatus_ != ChannelStatus::READY) {
755 0 : HCCL_ERROR(
756 : "[AicpuTsRoceChannelV2::%s] channel status[%d] is not ready[%d], please check.", __func__, channelStatus_,
757 : ChannelStatus::READY);
758 0 : return HCCL_E_INTERNAL;
759 : }
760 :
761 8 : if (bufferNum_ == 0) {
762 0 : HCCL_INFO("[AicpuTsRoceChannelV2::%s] No local memory regions available", __func__);
763 0 : return HCCL_SUCCESS;
764 : }
765 :
766 8 : if (bufferEntityPtr == nullptr) {
767 0 : HCCL_ERROR("[AicpuTsRoceChannelV2::%s] input param is null", __func__);
768 0 : return HCCL_E_PARA;
769 : }
770 :
771 16 : for (uint32_t i = 0; i < bufferNum_; i++) {
772 8 : auto& locRmaBuffer = localRmaBuffers_[i];
773 8 : bufList[i].type = REGED_BUFFER_RMA;
774 8 : bufList[i].bufferInfo.rma.addr = static_cast<uint64_t>(locRmaBuffer->GetAddr());
775 8 : bufList[i].bufferInfo.rma.size = locRmaBuffer->GetSize();
776 8 : bufList[i].bufferInfo.rma.protectionInfo.type = PROTECTION_TYPE_ROCE;
777 8 : bufList[i].bufferInfo.rma.protectionInfo.memInfo.roce.lkey = locRmaBuffer->GetLkey();
778 8 : HCCL_INFO(
779 : "[AicpuTsRoceChannelV2::%s] locBuf[addr[%p], size[%lu]]", __func__, bufList[i].bufferInfo.rma.addr,
780 : bufList[i].bufferInfo.rma.size);
781 : }
782 8 : *bufferEntityPtr = bufList.data();
783 8 : return HCCL_SUCCESS;
784 : }
785 :
786 8 : HcclResult AicpuTsRoceChannelV2::BuildAndGetSqContext(std::vector<SqContext>& sqList, SqContext** sqContextPtr)
787 : {
788 8 : if (channelStatus_ != ChannelStatus::READY) {
789 0 : HCCL_ERROR(
790 : "[AicpuTsRoceChannelV2::%s] channel status[%d] is not ready[%d], please check.", __func__, channelStatus_,
791 : ChannelStatus::READY);
792 0 : return HCCL_E_INTERNAL;
793 : }
794 :
795 8 : if (connNum_ == 0) {
796 0 : HCCL_INFO("[AicpuTsRoceChannelV2::%s] No conn available", __func__);
797 0 : return HCCL_SUCCESS;
798 : }
799 :
800 8 : if (sqContextPtr == nullptr) {
801 0 : HCCL_ERROR("[AicpuTsRoceChannelV2::%s] input param is null", __func__);
802 0 : return HCCL_E_PARA;
803 : }
804 :
805 16 : for (uint32_t i = 0; i < connNum_; i++) {
806 8 : auto& conn = connections_[i];
807 8 : Hccl::CHECK_NULLPTR(
808 16 : conn, Hccl::StringFormat("[AicpuTsRoceChannelV2::%s] failed, connection pointer is nullptr", __func__));
809 : SqContext sqContext;
810 8 : CHK_RET(conn->BuildSqContext(&sqContext));
811 8 : sqList[i] = sqContext;
812 : }
813 8 : *sqContextPtr = sqList.data();
814 8 : return HCCL_SUCCESS;
815 : }
816 :
817 8 : HcclResult AicpuTsRoceChannelV2::BuildAndGetCqContext(std::vector<CqContext>& cqList, CqContext** cqContextPtr)
818 : {
819 8 : if (channelStatus_ != ChannelStatus::READY) {
820 0 : HCCL_ERROR(
821 : "[AicpuTsRoceChannelV2::%s] channel status[%d] is not ready[%d], please check.", __func__, channelStatus_,
822 : ChannelStatus::READY);
823 0 : return HCCL_E_INTERNAL;
824 : }
825 :
826 8 : if (connNum_ == 0) {
827 0 : HCCL_INFO("[AicpuTsRoceChannelV2::%s] No conn available", __func__);
828 0 : return HCCL_SUCCESS;
829 : }
830 :
831 8 : if (cqContextPtr == nullptr) {
832 0 : HCCL_ERROR("[AicpuTsRoceChannelV2::%s] input param is null", __func__);
833 0 : return HCCL_E_PARA;
834 : }
835 :
836 16 : for (uint32_t i = 0; i < connNum_; i++) {
837 8 : auto& conn = connections_[i];
838 8 : Hccl::CHECK_NULLPTR(
839 16 : conn, Hccl::StringFormat("[AicpuTsRoceChannelV2::%s] failed, connection pointer is nullptr", __func__));
840 : CqContext cqContext;
841 8 : CHK_RET(conn->BuildCqContext(&cqContext));
842 8 : cqList[i] = cqContext;
843 : }
844 8 : *cqContextPtr = cqList.data();
845 8 : return HCCL_SUCCESS;
846 : }
847 :
848 7 : HcclResult AicpuTsRoceChannelV2::BuildHostEntity(
849 : ChannelEntity& hostEntity, std::vector<RegedBufferEntity>& locBufList, std::vector<RegedBufferEntity>& rmtBufList,
850 : std::vector<SqContext>& sqList, std::vector<CqContext>& cqList)
851 : {
852 7 : hostEntity.abiHeader.version = HCCL_CHANNEL_VERSION;
853 7 : hostEntity.abiHeader.magicWord = HCCL_CHANNEL_MAGIC_WORD;
854 7 : hostEntity.abiHeader.size = sizeof(ChannelEntity);
855 7 : hostEntity.abiHeader.reserved = 0;
856 7 : hostEntity.engine = GetCommEngine();
857 7 : hostEntity.protocol = GetCommProtocol();
858 :
859 7 : hostEntity.localNotifyNum = 0;
860 7 : CHK_RET(BuildAndGetLocNotifyInfo(&hostEntity.localNotifyAddr));
861 7 : hostEntity.remoteNotifyNum = 0;
862 7 : CHK_RET(BuildAndGetRmtNotifyInfo(&hostEntity.remoteNotifyAddr));
863 :
864 7 : locBufList.resize(bufferNum_);
865 7 : hostEntity.localBufferNum = bufferNum_;
866 7 : CHK_RET(BuildAndGetLocBufInfo(locBufList, &hostEntity.localBufferAddr));
867 :
868 7 : rmtBufList.resize(bufferNum_);
869 7 : hostEntity.remoteBufferNum = bufferNum_;
870 7 : CHK_RET(BuildAndGetRmtBufInfo(rmtBufList, &hostEntity.remoteBufferAddr));
871 :
872 7 : sqList.resize(connNum_);
873 7 : hostEntity.sqNum = connNum_;
874 7 : CHK_RET(BuildAndGetSqContext(sqList, &hostEntity.sqContextAddr));
875 :
876 7 : cqList.resize(connNum_);
877 7 : hostEntity.cqNum = connNum_;
878 7 : CHK_RET(BuildAndGetCqContext(cqList, &hostEntity.cqContextAddr));
879 :
880 7 : return HCCL_SUCCESS;
881 : }
882 :
883 7 : HcclResult AicpuTsRoceChannelV2::BuildAndGetDevChannelEntity(uint64_t* devChannelEntityPtr)
884 : {
885 7 : CHK_PTR_NULL(devChannelEntityPtr);
886 :
887 7 : if (devChannelEntitySlab_ != nullptr) {
888 1 : *devChannelEntityPtr = reinterpret_cast<uint64_t>(devChannelEntitySlab_);
889 1 : HCCL_INFO(
890 : "[AicpuTsRoceChannelV2::%s] already built, return cached devPtr=0x%lx", __func__, *devChannelEntityPtr);
891 1 : return HCCL_SUCCESS;
892 : }
893 :
894 6 : ChannelEntity hostEntity{};
895 6 : std::vector<RegedBufferEntity> locBufList;
896 6 : std::vector<RegedBufferEntity> rmtBufList;
897 6 : std::vector<SqContext> sqList;
898 6 : std::vector<CqContext> cqList;
899 6 : CHK_RET(BuildHostEntity(hostEntity, locBufList, rmtBufList, sqList, cqList));
900 :
901 6 : DeviceChannelEntityLayout layout;
902 6 : CHK_RET(BuildDeviceChannelEntityLayout(hostEntity, layout));
903 6 : void* slabPtr = nullptr;
904 6 : AclDeviceSlabGuard slabGuard;
905 6 : CHK_RET(AllocDeviceEntitySlab(layout.slabSize, slabGuard, slabPtr));
906 :
907 : ChannelEntity devEntity;
908 5 : CHK_RET(CopyChannelEntityArrayToSlab(slabPtr, hostEntity, layout, devEntity));
909 5 : void* entityDevPtr = nullptr;
910 5 : CHK_RET(CopyChannelEntityToSlab(slabPtr, layout, devEntity, entityDevPtr));
911 :
912 5 : ReleaseDeviceEntitySlab();
913 5 : devChannelEntitySlab_ = slabGuard.Release();
914 5 : devChannelEntitySlabSize_ = layout.slabSize;
915 :
916 5 : *devChannelEntityPtr = reinterpret_cast<uint64_t>(entityDevPtr);
917 5 : HCCL_INFO(
918 : "[AicpuTsRoceChannelV2::%s] Success, devPtr=0x%lx, slabPtr=%p, slabSize=%zu", __func__, *devChannelEntityPtr,
919 : devChannelEntitySlab_, devChannelEntitySlabSize_);
920 5 : return HCCL_SUCCESS;
921 6 : }
922 :
923 5 : HcclResult AicpuTsRoceChannelV2::PreAllocDevChannelEntity(uint64_t* devChannelEntityPtr)
924 : {
925 5 : CHK_PTR_NULL(devChannelEntityPtr);
926 :
927 4 : if (devChannelEntitySlab_ != nullptr) {
928 1 : *devChannelEntityPtr = reinterpret_cast<uint64_t>(devChannelEntitySlab_);
929 1 : HCCL_INFO(
930 : "[AicpuTsRoceChannelV2::%s] already built, return cached devPtr=0x%lx", __func__, *devChannelEntityPtr);
931 1 : return HCCL_SUCCESS;
932 : }
933 :
934 3 : ChannelEntity tmp{};
935 3 : tmp.localNotifyNum = 0;
936 3 : tmp.remoteNotifyNum = 0;
937 3 : tmp.localBufferNum = bufferNum_;
938 3 : tmp.remoteBufferNum = bufferNum_;
939 3 : tmp.sqNum = connNum_;
940 3 : tmp.cqNum = connNum_;
941 :
942 3 : DeviceChannelEntityLayout layout;
943 3 : CHK_RET(BuildDeviceChannelEntityLayout(tmp, layout));
944 :
945 3 : void* slabPtr = nullptr;
946 3 : AclDeviceSlabGuard slabGuard;
947 3 : CHK_RET(AllocDeviceEntitySlab(layout.slabSize, slabGuard, slabPtr));
948 :
949 3 : devChannelEntitySlab_ = slabGuard.Release();
950 3 : devChannelEntitySlabSize_ = layout.slabSize;
951 3 : *devChannelEntityPtr = reinterpret_cast<uint64_t>(devChannelEntitySlab_);
952 :
953 3 : HCCL_INFO(
954 : "[AicpuTsRoceChannelV2::%s] pre-alloc success, slabPtr=%p, slabSize=%zu", __func__, devChannelEntitySlab_,
955 : devChannelEntitySlabSize_);
956 3 : return HCCL_SUCCESS;
957 3 : }
958 :
959 3 : HcclResult AicpuTsRoceChannelV2::FillDevChannelEntity()
960 : {
961 3 : if (devChannelEntitySlab_ == nullptr) {
962 1 : HCCL_ERROR("[AicpuTsRoceChannelV2::%s] devChannelEntitySlab_ is nullptr, not pre-allocated.", __func__);
963 1 : return HCCL_E_INTERNAL;
964 : }
965 2 : if (channelStatus_ != ChannelStatus::READY) {
966 1 : HCCL_ERROR(
967 : "[AicpuTsRoceChannelV2::%s] channel status[%d] is not ready[%d], please check.", __func__, channelStatus_,
968 : ChannelStatus::READY);
969 1 : return HCCL_E_INTERNAL;
970 : }
971 :
972 1 : ChannelEntity hostEntity{};
973 1 : std::vector<RegedBufferEntity> locBufList;
974 1 : std::vector<RegedBufferEntity> rmtBufList;
975 1 : std::vector<SqContext> sqList;
976 1 : std::vector<CqContext> cqList;
977 1 : CHK_RET(BuildHostEntity(hostEntity, locBufList, rmtBufList, sqList, cqList));
978 :
979 1 : DeviceChannelEntityLayout layout;
980 1 : CHK_RET(BuildDeviceChannelEntityLayout(hostEntity, layout));
981 1 : if (layout.slabSize > devChannelEntitySlabSize_) {
982 0 : HCCL_ERROR(
983 : "[AicpuTsRoceChannelV2::%s] slabSize[%zu] > preAllocSize[%zu]", __func__, layout.slabSize,
984 : devChannelEntitySlabSize_);
985 0 : return HCCL_E_INTERNAL;
986 : }
987 :
988 : ChannelEntity devEntity;
989 1 : CHK_RET(CopyChannelEntityArrayToSlab(devChannelEntitySlab_, hostEntity, layout, devEntity));
990 1 : void* entityDevPtr = nullptr;
991 1 : CHK_RET(CopyChannelEntityToSlab(devChannelEntitySlab_, layout, devEntity, entityDevPtr));
992 :
993 1 : HCCL_INFO("[AicpuTsRoceChannelV2::%s] fill success, devPtr=%p", __func__, entityDevPtr);
994 1 : return HCCL_SUCCESS;
995 1 : }
996 :
997 55 : void AicpuTsRoceChannelV2::ReleaseDeviceEntitySlab()
998 : {
999 55 : if (devChannelEntitySlab_ != nullptr) {
1000 8 : HcclResult ret = hrtFree(devChannelEntitySlab_);
1001 8 : if (ret != HCCL_SUCCESS) {
1002 0 : HCCL_WARNING(
1003 : "[AicpuTsRoceChannelV2::%s] hrtFree devChannelEntitySlab failed, ptr[%p], size[%zu], ret[%d]", __func__,
1004 : devChannelEntitySlab_, devChannelEntitySlabSize_, ret);
1005 : }
1006 8 : devChannelEntitySlab_ = nullptr;
1007 8 : devChannelEntitySlabSize_ = 0;
1008 : }
1009 55 : }
1010 :
1011 48 : void AicpuTsRoceChannelV2::FreeDeviceMemories() { ReleaseDeviceEntitySlab(); }
1012 :
1013 1 : std::string AicpuTsRoceChannelV2::Describe() const
1014 : {
1015 1 : std::string msg = "AicpuTsRoceChannelV2{";
1016 1 : msg += Hccl::StringFormat("notifyNum:%u, localNotifies:[", notifyNum_);
1017 4 : for (auto& notify : localNotifies_) {
1018 3 : msg += notify->Describe();
1019 3 : msg += ", ";
1020 : }
1021 1 : msg += "]";
1022 1 : msg += Hccl::StringFormat(", bufferNum:%u, localRmaBuffers:[", bufferNum_);
1023 2 : for (auto& buf : localRmaBuffers_) {
1024 1 : msg += buf->Describe();
1025 1 : msg += ", ";
1026 : }
1027 1 : msg += "]";
1028 1 : msg += Hccl::StringFormat(", connNum:%u, connections:[", connNum_);
1029 2 : for (auto& conn : connections_) {
1030 1 : msg += conn->Describe();
1031 1 : msg += ", ";
1032 : }
1033 1 : msg += "]";
1034 1 : msg += Hccl::StringFormat(", rdmaHandle:%p, %s, ", rdmaHandle_, channelStatus_.Describe().c_str());
1035 1 : if (socket_ != nullptr) {
1036 1 : msg += socket_->Describe();
1037 : }
1038 1 : msg += "}";
1039 1 : return msg;
1040 0 : }
1041 :
1042 3 : std::vector<char> AicpuTsRoceChannelV2::GetLocalNotifyUniqueIds() const
1043 : {
1044 3 : HCCL_DEBUG("start packing local notify uniqueIds");
1045 3 : std::vector<char> result(0);
1046 12 : for (auto& it : localNotifies_) {
1047 9 : HCCL_INFO("AicpuTsRoceChannelV2 local notify %s", it->Describe().c_str());
1048 9 : auto uniqueId = it->GetUniqueId();
1049 9 : result.insert(result.end(), uniqueId.begin(), uniqueId.end());
1050 9 : }
1051 3 : return result;
1052 0 : }
1053 :
1054 3 : std::vector<char> AicpuTsRoceChannelV2::GetRemoteNotifyUniqueIds() const
1055 : {
1056 3 : HCCL_DEBUG("start packing remote notify uniqueIds");
1057 3 : std::vector<char> result(0);
1058 3 : Hccl::BinaryStream binaryStream;
1059 4 : for (auto& it : remoteNotifies_) {
1060 1 : std::vector<char> uniqueId;
1061 1 : uniqueId = GetSingleRmaBufferUniqueId(static_cast<uint64_t>(it->GetAddr()), it->GetSize(), it->GetRkey());
1062 1 : HCCL_INFO("AicpuTsRoceChannelV2 remote notify %s", it->Describe().c_str());
1063 1 : result.insert(result.end(), uniqueId.begin(), uniqueId.end());
1064 1 : }
1065 3 : binaryStream.Dump(result);
1066 3 : return result;
1067 3 : }
1068 :
1069 3 : std::vector<char> AicpuTsRoceChannelV2::GetNotifyValueBufferUniqueIds() const
1070 : {
1071 3 : HCCL_DEBUG("start packing notify value buffer uniqueIds");
1072 3 : std::vector<char> uniqueId;
1073 6 : uniqueId = GetSingleRmaBufferUniqueId(
1074 3 : static_cast<uint64_t>(notifyValueBuffer_->GetAddr()), notifyValueBuffer_->GetSize(),
1075 3 : notifyValueBuffer_->GetLkey());
1076 3 : HCCL_INFO("AicpuTsRoceChannelV2 notify value buffer %s", notifyValueBuffer_->Describe().c_str());
1077 3 : return uniqueId;
1078 0 : }
1079 :
1080 13 : std::vector<char> AicpuTsRoceChannelV2::GetSingleRmaBufferUniqueId(u64 addr, u64 size, u32 key) const
1081 : {
1082 13 : Hccl::BinaryStream binaryStream;
1083 13 : binaryStream << addr;
1084 13 : binaryStream << size;
1085 13 : binaryStream << key;
1086 13 : std::vector<char> result;
1087 13 : binaryStream.Dump(result);
1088 13 : return result;
1089 13 : }
1090 :
1091 4 : std::vector<char> AicpuTsRoceChannelV2::GetRmtBufferUniqueIds() const
1092 : {
1093 4 : HCCL_DEBUG("start packing remote buffer uniqueIds");
1094 4 : std::vector<char> result(0);
1095 8 : for (auto& it : rmtRmaBuffers_) {
1096 4 : std::vector<char> uniqueId;
1097 4 : if (it != nullptr) {
1098 3 : uniqueId = GetSingleRmaBufferUniqueId(static_cast<uint64_t>(it->GetAddr()), it->GetSize(), it->GetRkey());
1099 3 : HCCL_INFO("AicpuTsRoceChannelV2::GetRmtBufferUniqueIds, %s", it->Describe().c_str());
1100 : } else {
1101 1 : uniqueId = GetSingleRmaBufferUniqueId(0, 0, 0); // 填充一个空的buffer
1102 1 : HCCL_INFO("AicpuTsRoceChannelV2::GetRmtBufferUniqueIds, null buffer");
1103 : }
1104 4 : result.insert(result.end(), uniqueId.begin(), uniqueId.end());
1105 4 : }
1106 4 : return result;
1107 0 : }
1108 :
1109 4 : std::vector<char> AicpuTsRoceChannelV2::GetLocBufferUniqueIds() const
1110 : {
1111 4 : HCCL_DEBUG("start packing local buffer uniqueIds");
1112 4 : std::vector<char> result(0);
1113 9 : for (auto& it : localRmaBuffers_) {
1114 5 : std::vector<char> uniqueId;
1115 5 : if (it != nullptr) {
1116 4 : uniqueId = GetSingleRmaBufferUniqueId(static_cast<uint64_t>(it->GetAddr()), it->GetSize(), it->GetLkey());
1117 4 : HCCL_INFO("AicpuTsRoceChannelV2::GetLocBufferUniqueIds, %s", it->Describe().c_str());
1118 : } else {
1119 1 : uniqueId = GetSingleRmaBufferUniqueId(0, 0, 0); // 填充一个空的buffer
1120 1 : HCCL_INFO("AicpuTsRoceChannelV2::GetLocBufferUniqueIds, null buffer");
1121 : }
1122 5 : result.insert(result.end(), uniqueId.begin(), uniqueId.end());
1123 5 : }
1124 4 : return result;
1125 0 : }
1126 :
1127 3 : std::vector<char> AicpuTsRoceChannelV2::GetConnUniqueIds() const
1128 : {
1129 3 : HCCL_DEBUG("start packing all conn uniqueIds");
1130 3 : std::vector<char> result(0);
1131 6 : for (auto& it : connections_) {
1132 3 : HCCL_INFO("AicpuTsRoceChannelV2 %s", it->Describe().c_str());
1133 3 : auto uniqueId = it->GetUniqueId();
1134 3 : result.insert(result.end(), uniqueId.begin(), uniqueId.end());
1135 3 : }
1136 3 : return result;
1137 0 : }
1138 :
1139 2 : std::vector<char> AicpuTsRoceChannelV2::GetUniqueId() const
1140 : {
1141 2 : if (channelStatus_ != ChannelStatus::READY) {
1142 0 : HCCL_ERROR(
1143 : "[AicpuTsRoceChannelV2::%s] channel status[%d] is not ready[%d], please check.", __func__, channelStatus_,
1144 : ChannelStatus::READY);
1145 : }
1146 2 : u32 type = static_cast<u32>(Hccl::TransportType::ROCE);
1147 2 : Hccl::BinaryStream binaryStream;
1148 2 : binaryStream << type;
1149 2 : binaryStream << notifyNum_;
1150 2 : binaryStream << bufferNum_;
1151 2 : binaryStream << connNum_;
1152 :
1153 2 : auto locNotifyUniqueIds = GetLocalNotifyUniqueIds();
1154 2 : binaryStream << locNotifyUniqueIds;
1155 :
1156 2 : auto rmtNotifyUniqueIds = GetRemoteNotifyUniqueIds();
1157 2 : binaryStream << rmtNotifyUniqueIds;
1158 :
1159 2 : auto notifyValueBufferUniqueIds = GetNotifyValueBufferUniqueIds();
1160 2 : binaryStream << notifyValueBufferUniqueIds;
1161 :
1162 2 : auto locBufferUniqueIds = GetLocBufferUniqueIds();
1163 2 : binaryStream << locBufferUniqueIds;
1164 :
1165 2 : auto rmtBufferUniqueIds = GetRmtBufferUniqueIds();
1166 2 : binaryStream << rmtBufferUniqueIds;
1167 :
1168 2 : auto connUniqueIds = GetConnUniqueIds();
1169 2 : binaryStream << connUniqueIds;
1170 :
1171 2 : std::vector<char> result;
1172 2 : binaryStream.Dump(result);
1173 2 : return result;
1174 2 : }
1175 :
1176 1 : static HcclResult SetModuleDataName(Hccl::ModuleData& module, const std::string& name)
1177 : {
1178 1 : int ret = strcpy_s(module.name, sizeof(module.name), name.c_str());
1179 1 : if (ret != 0) {
1180 0 : HCCL_ERROR("[SetModuleDataName] strcpy_s name %s failed", name.c_str());
1181 0 : return HCCL_E_INTERNAL;
1182 : }
1183 :
1184 1 : return HCCL_SUCCESS;
1185 : }
1186 :
1187 1 : HcclResult AicpuTsRoceChannelV2::PackOpData(std::vector<char>& data) const
1188 : {
1189 1 : std::vector<Hccl::ModuleData> dataVec;
1190 1 : dataVec.resize(Hccl::AicpuResMgrType::__COUNT__);
1191 :
1192 1 : Hccl::AicpuResMgrType resType = Hccl::AicpuResMgrType::STREAM;
1193 2 : CHK_RET(SetModuleDataName(dataVec[resType], "AicpuTsRoceChannelV2"));
1194 :
1195 1 : std::vector<char> result;
1196 1 : Hccl::BinaryStream binaryStream;
1197 1 : binaryStream << GetUniqueId();
1198 :
1199 1 : binaryStream.Dump(result);
1200 :
1201 1 : dataVec[resType].data = result;
1202 :
1203 : Hccl::AicpuResPackageHelper helper;
1204 1 : data = helper.GetPackedData(dataVec);
1205 :
1206 1 : return HCCL_SUCCESS;
1207 1 : }
1208 :
1209 1 : HcclResult AicpuTsRoceChannelV2::H2DResPack(std::vector<char>& buffer)
1210 : {
1211 1 : CHK_RET(PackOpData(buffer));
1212 1 : HCCL_INFO(
1213 : "[AicpuTsRoceChannelV2][%s] Pack Buffer data[%p], Pack Buffer size[%zu].", __func__, buffer.data(),
1214 : buffer.size());
1215 1 : return HCCL_SUCCESS;
1216 : }
1217 :
1218 1 : HcclResult AicpuTsRoceChannelV2::GetNotifyNum(uint32_t* notifyNum) const
1219 : {
1220 1 : CHK_PTR_NULL(notifyNum);
1221 1 : *notifyNum = (engine_ == COMM_ENGINE_AIV) ? 0 : notifyNum_;
1222 1 : return HCCL_SUCCESS;
1223 : }
1224 :
1225 1 : HcclResult AicpuTsRoceChannelV2::GetBufferNum(uint32_t* bufferNum) const
1226 : {
1227 1 : CHK_PTR_NULL(bufferNum);
1228 1 : *bufferNum = bufferNum_;
1229 1 : return HCCL_SUCCESS;
1230 : }
1231 :
1232 1 : HcclResult AicpuTsRoceChannelV2::GetQpNum(uint32_t* qpNum) const
1233 : {
1234 1 : CHK_PTR_NULL(qpNum);
1235 1 : *qpNum = connNum_;
1236 1 : return HCCL_SUCCESS;
1237 : }
1238 :
1239 7 : HcclResult AicpuTsRoceChannelV2::GetRemoteMems(uint32_t* memNum, CommMem** remoteMem, char*** memInfos)
1240 : {
1241 7 : std::lock_guard<std::mutex> lock(remoteMemsMutex_);
1242 : Hccl::RemoteMemCtx<std::unique_ptr<Hccl::RemoteRdmaRmaBuffer>> remoteMemCtx{
1243 7 : cacheValid_, rmtRmaBuffers_, remoteUserMems_, memInfoCopies_, memInfoPointers_, remoteMem, memInfos, memNum};
1244 7 : CHK_RET(Hccl::GetRemoteUserMems(remoteMemCtx));
1245 4 : return HCCL_SUCCESS;
1246 7 : }
1247 :
1248 2 : HcclResult AicpuTsRoceChannelV2::Clean()
1249 : {
1250 2 : ReleaseDeviceEntitySlab();
1251 2 : return HCCL_SUCCESS;
1252 : }
1253 :
1254 1 : HcclResult AicpuTsRoceChannelV2::Resume() { return HCCL_SUCCESS; }
1255 :
1256 0 : HcclResult AicpuTsRoceChannelV2::Serialize(std::shared_ptr<hccl::DeviceMem>& out)
1257 : {
1258 0 : out.reset();
1259 0 : CHK_PRT_RET(
1260 : channelStatus_ != ChannelStatus::READY,
1261 : HCCL_ERROR("[AicpuTsRoceChannelV2][%s] channel not ready, status[%d]", __func__, channelStatus_),
1262 : HCCL_E_INTERNAL);
1263 :
1264 0 : std::vector<char> hostBuffer;
1265 0 : CHK_RET(H2DResPack(hostBuffer));
1266 :
1267 0 : u64 totalBytes = static_cast<u64>(hostBuffer.size());
1268 0 : CHK_PRT_RET(
1269 : totalBytes == 0, HCCL_ERROR("[AicpuTsRoceChannelV2][%s] serialized buffer is empty", __func__),
1270 : HCCL_E_INTERNAL);
1271 :
1272 0 : hccl::DeviceMem devMem;
1273 0 : EXCEPTION_CATCH(devMem = hccl::DeviceMem::alloc(totalBytes), return HCCL_E_PTR);
1274 :
1275 0 : Hccl::HrtMemcpy(
1276 0 : devMem.ptr(), totalBytes, hostBuffer.data(), totalBytes, Hccl::tagRtMemcpyKind::RT_MEMCPY_HOST_TO_DEVICE);
1277 :
1278 0 : out = std::make_shared<hccl::DeviceMem>(std::move(devMem));
1279 :
1280 0 : HCCL_INFO("[AicpuTsRoceChannelV2][%s] serialize success, size[%llu]", __func__, totalBytes);
1281 0 : return HCCL_SUCCESS;
1282 0 : }
1283 :
1284 1 : HcommChannelKind AicpuTsRoceChannelV2::GetChannelKind() const { return HcommChannelKind::AICPU_TS_ROCE_V2; }
1285 : } // namespace hcomm
|
