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