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 "aiv_urma_channel.h"
12 : #include "endpoint.h"
13 : #include "orion_adpt_utils.h"
14 : #include "acl_device_slab_guard.h"
15 : #include "shared_jetty_channel_helper.h"
16 :
17 : #include "hcomm_c_adpt.h"
18 : #include "hcomm_res_mgr.h"
19 :
20 : // Orion
21 : #include "topo_common_types.h"
22 : #include "virtual_topo.h"
23 : #include "makebufs_helper.h"
24 : #include "orion_adapter_hccp.h"
25 : #include "adapter_rts.h"
26 : #include "acl/acl_rt.h"
27 :
28 : #include <algorithm>
29 : #include <cstdint>
30 : #include <cstdlib>
31 : #include <cstring>
32 :
33 : namespace hcomm {
34 : constexpr uint16_t DEFAULT_LISTENING_PORT = 60001;
35 :
36 : namespace {
37 : constexpr size_t AIV_URMA_ENTITY_ALIGN_SIZE = 64;
38 : constexpr size_t QUEUE_INDEX_MEM_UNIT_SIZE = sizeof(void*);
39 :
40 : struct DeviceEntitySection {
41 : size_t offset{0};
42 : size_t size{0};
43 : };
44 :
45 : struct DeviceChannelEntityLayout {
46 : DeviceEntitySection entitySection{0, sizeof(ChannelEntity)};
47 : DeviceEntitySection localNotifySection;
48 : DeviceEntitySection remoteNotifySection;
49 : DeviceEntitySection localBufferSection;
50 : DeviceEntitySection remoteBufferSection;
51 : DeviceEntitySection sqContextSection;
52 : DeviceEntitySection cqContextSection;
53 : DeviceEntitySection sqPiSection;
54 : DeviceEntitySection sqCiSection;
55 : DeviceEntitySection cqPiSection;
56 : DeviceEntitySection cqCiSection;
57 : size_t slabSize{0};
58 : };
59 :
60 6 : HcclResult SecureMemset(void* dest, size_t destMax, int value, size_t count, const char* fieldName)
61 : {
62 6 : if (dest == nullptr) {
63 0 : HCCL_ERROR("[SecureMemset] dest is nullptr, field[%s]", fieldName);
64 0 : return HCCL_E_PTR;
65 : }
66 6 : if (count > destMax) {
67 0 : HCCL_ERROR("[SecureMemset] invalid size, field[%s], count[%zu], destMax[%zu]", fieldName, count, destMax);
68 0 : return HCCL_E_PARA;
69 : }
70 :
71 6 : errno_t ret = memset_s(dest, destMax, value, count);
72 6 : if (ret != EOK) {
73 0 : HCCL_ERROR(
74 : "[SecureMemset] memset_s failed, field[%s], ret[%d], count[%zu], destMax[%zu]", fieldName, ret, count,
75 : destMax);
76 0 : return HCCL_E_MEMORY;
77 : }
78 6 : return HCCL_SUCCESS;
79 : }
80 :
81 3 : HcclResult GetAllMemHandles(EndpointHandle endpointHandle, void** memHandles, uint32_t* memHandleNum)
82 : {
83 3 : return static_cast<HcclResult>(HcommMemGetAllMemHandles(endpointHandle, memHandles, memHandleNum));
84 : }
85 :
86 84 : size_t AlignUp(size_t value, size_t alignment) { return (value + alignment - 1) / alignment * alignment; }
87 :
88 70 : HcclResult AddDeviceEntitySection(
89 : size_t elemSize, uint32_t elemNum, size_t& offset, DeviceEntitySection& section, const char* sectionName)
90 : {
91 70 : section.offset = AlignUp(offset, AIV_URMA_ENTITY_ALIGN_SIZE);
92 70 : if (elemNum == 0) {
93 28 : section.size = 0;
94 28 : offset = section.offset;
95 28 : return HCCL_SUCCESS;
96 : }
97 42 : CHK_PRT_RET(
98 : elemSize != 0 && elemNum > (SIZE_MAX / elemSize),
99 : HCCL_ERROR(
100 : "[AivUrmaChannel::AddDeviceEntitySection] %s size overflow, elemSize[%zu], elemNum[%u]", sectionName,
101 : elemSize, elemNum),
102 : HCCL_E_PARA);
103 42 : section.size = elemSize * static_cast<size_t>(elemNum);
104 42 : CHK_PRT_RET(
105 : section.offset > (SIZE_MAX - section.size),
106 : HCCL_ERROR(
107 : "[AivUrmaChannel::AddDeviceEntitySection] %s offset overflow, offset[%zu], size[%zu]", sectionName,
108 : section.offset, section.size),
109 : HCCL_E_PARA);
110 42 : offset = section.offset + section.size;
111 42 : return HCCL_SUCCESS;
112 : }
113 :
114 52 : void* GetSlabPtr(void* base, const DeviceEntitySection& section)
115 : {
116 52 : if (section.size == 0) {
117 0 : return nullptr;
118 : }
119 52 : return reinterpret_cast<void*>(reinterpret_cast<uintptr_t>(base) + section.offset);
120 : }
121 :
122 : template <typename T>
123 18 : HcclResult CopyArrayToSlab(
124 : void* slabBase, const T* hostArray, uint32_t arrayNum, const DeviceEntitySection& section, T** deviceArrayPtr,
125 : const char* arrayName)
126 : {
127 18 : CHK_PTR_NULL(deviceArrayPtr);
128 18 : if (arrayNum == 0 || hostArray == nullptr) {
129 12 : CHK_PRT_RET(
130 : arrayNum != 0,
131 : HCCL_ERROR("[AivUrmaChannel::CopyArrayToSlab] %s hostArray is nullptr, num[%u]", arrayName, arrayNum),
132 : HCCL_E_PTR);
133 12 : *deviceArrayPtr = nullptr;
134 12 : return HCCL_SUCCESS;
135 : }
136 6 : CHK_PRT_RET(
137 : section.size != static_cast<size_t>(arrayNum) * sizeof(T),
138 : HCCL_ERROR(
139 : "[AivUrmaChannel::CopyArrayToSlab] %s size mismatch, sectionSize[%zu], expect[%zu]", arrayName,
140 : section.size, static_cast<size_t>(arrayNum) * sizeof(T)),
141 : HCCL_E_PARA);
142 6 : void* sectionPtr = GetSlabPtr(slabBase, section);
143 6 : CHK_PTR_NULL(sectionPtr);
144 6 : Hccl::HrtMemcpy(
145 6 : sectionPtr, section.size, hostArray, section.size, Hccl::tagRtMemcpyKind::RT_MEMCPY_HOST_TO_DEVICE);
146 6 : *deviceArrayPtr = reinterpret_cast<T*>(sectionPtr);
147 6 : HCCL_INFO(
148 : "[AivUrmaChannel::CopyArrayToSlab] %s: host[%p] -> dev[%p], num[%u], size[%zu]", arrayName, hostArray,
149 : sectionPtr, arrayNum, section.size);
150 6 : return HCCL_SUCCESS;
151 : }
152 :
153 7 : HcclResult BuildDeviceChannelEntityLayout(const ChannelEntity& hostChannel, DeviceChannelEntityLayout& layout)
154 : {
155 7 : layout.slabSize = AlignUp(sizeof(ChannelEntity), AIV_URMA_ENTITY_ALIGN_SIZE);
156 7 : CHK_RET(AddDeviceEntitySection(
157 : sizeof(RegedNotifyEntity), hostChannel.localNotifyNum, layout.slabSize, layout.localNotifySection,
158 : "localNotifyAddr"));
159 7 : CHK_RET(AddDeviceEntitySection(
160 : sizeof(RegedNotifyEntity), hostChannel.remoteNotifyNum, layout.slabSize, layout.remoteNotifySection,
161 : "remoteNotifyAddr"));
162 7 : CHK_RET(AddDeviceEntitySection(
163 : sizeof(RegedBufferEntity), hostChannel.localBufferNum, layout.slabSize, layout.localBufferSection,
164 : "localBufferAddr"));
165 7 : CHK_RET(AddDeviceEntitySection(
166 : sizeof(RegedBufferEntity), hostChannel.remoteBufferNum, layout.slabSize, layout.remoteBufferSection,
167 : "remoteBufferAddr"));
168 7 : CHK_RET(AddDeviceEntitySection(
169 : sizeof(SqContext), hostChannel.sqNum, layout.slabSize, layout.sqContextSection, "sqContextAddr"));
170 7 : CHK_RET(AddDeviceEntitySection(
171 : sizeof(CqContext), hostChannel.cqNum, layout.slabSize, layout.cqContextSection, "cqContextAddr"));
172 7 : CHK_RET(AddDeviceEntitySection(
173 : QUEUE_INDEX_MEM_UNIT_SIZE, hostChannel.sqNum, layout.slabSize, layout.sqPiSection, "sqPiAddr"));
174 7 : CHK_RET(AddDeviceEntitySection(
175 : QUEUE_INDEX_MEM_UNIT_SIZE, hostChannel.sqNum, layout.slabSize, layout.sqCiSection, "sqCiAddr"));
176 7 : CHK_RET(AddDeviceEntitySection(
177 : QUEUE_INDEX_MEM_UNIT_SIZE, hostChannel.cqNum, layout.slabSize, layout.cqPiSection, "cqPiAddr"));
178 7 : CHK_RET(AddDeviceEntitySection(
179 : QUEUE_INDEX_MEM_UNIT_SIZE, hostChannel.cqNum, layout.slabSize, layout.cqCiSection, "cqCiAddr"));
180 7 : layout.slabSize = AlignUp(layout.slabSize, AIV_URMA_ENTITY_ALIGN_SIZE);
181 7 : return HCCL_SUCCESS;
182 : }
183 :
184 6 : HcclResult AllocDeviceEntitySlab(size_t slabSize, AclDeviceSlabGuard& slabGuard, void*& slabPtr)
185 : {
186 6 : HcclResult ret = hrtMalloc(&slabPtr, slabSize);
187 6 : CHK_PRT_RET(
188 : ret != HCCL_SUCCESS || slabPtr == nullptr,
189 : HCCL_ERROR("[AivUrmaChannel::%s] hrtMalloc slab failed, ret[%d], size[%zu]", __func__, ret, slabSize),
190 : HCCL_E_MEMORY);
191 5 : slabGuard.Reset(slabPtr, slabSize);
192 5 : return HCCL_SUCCESS;
193 : }
194 :
195 20 : HcclResult ZeroQueueIndexSection(
196 : void* slabPtr, const std::vector<uint8_t>& zeroQueueIndexMem, const DeviceEntitySection& section)
197 : {
198 20 : if (section.size == 0) {
199 0 : return HCCL_SUCCESS;
200 : }
201 20 : void* sectionPtr = GetSlabPtr(slabPtr, section);
202 20 : CHK_PTR_NULL(sectionPtr);
203 20 : Hccl::HrtMemcpy(
204 20 : sectionPtr, section.size, zeroQueueIndexMem.data(), section.size,
205 : Hccl::tagRtMemcpyKind::RT_MEMCPY_HOST_TO_DEVICE);
206 20 : return HCCL_SUCCESS;
207 : }
208 :
209 5 : HcclResult InitQueueIndexSections(void* slabPtr, const DeviceChannelEntityLayout& layout, uint32_t queueNum)
210 : {
211 5 : std::vector<uint8_t> zeroQueueIndexMem(QUEUE_INDEX_MEM_UNIT_SIZE * queueNum, 0);
212 5 : CHK_RET(ZeroQueueIndexSection(slabPtr, zeroQueueIndexMem, layout.sqPiSection));
213 5 : CHK_RET(ZeroQueueIndexSection(slabPtr, zeroQueueIndexMem, layout.sqCiSection));
214 5 : CHK_RET(ZeroQueueIndexSection(slabPtr, zeroQueueIndexMem, layout.cqPiSection));
215 5 : CHK_RET(ZeroQueueIndexSection(slabPtr, zeroQueueIndexMem, layout.cqCiSection));
216 5 : return HCCL_SUCCESS;
217 5 : }
218 :
219 5 : void SetQueueIndexDeviceMem(
220 : Hccl::AivUrmaTransport& transport, void* slabPtr, const DeviceChannelEntityLayout& layout, uint32_t queueNum)
221 : {
222 5 : transport.SetQueueIndexDeviceMem(
223 5 : GetSlabPtr(slabPtr, layout.sqPiSection), GetSlabPtr(slabPtr, layout.sqCiSection),
224 5 : GetSlabPtr(slabPtr, layout.cqPiSection), GetSlabPtr(slabPtr, layout.cqCiSection),
225 5 : queueNum * QUEUE_INDEX_MEM_UNIT_SIZE);
226 5 : }
227 :
228 3 : HcclResult CopyChannelEntityToSlab(
229 : void* slabPtr, const ChannelEntity& hostChannel, const DeviceChannelEntityLayout& layout,
230 : ChannelEntity& devChannel)
231 : {
232 3 : devChannel = hostChannel;
233 3 : CHK_RET(CopyArrayToSlab(
234 : slabPtr, hostChannel.localNotifyAddr, hostChannel.localNotifyNum, layout.localNotifySection,
235 : &devChannel.localNotifyAddr, "localNotifyAddr"));
236 3 : CHK_RET(CopyArrayToSlab(
237 : slabPtr, hostChannel.remoteNotifyAddr, hostChannel.remoteNotifyNum, layout.remoteNotifySection,
238 : &devChannel.remoteNotifyAddr, "remoteNotifyAddr"));
239 3 : CHK_RET(CopyArrayToSlab(
240 : slabPtr, hostChannel.localBufferAddr, hostChannel.localBufferNum, layout.localBufferSection,
241 : &devChannel.localBufferAddr, "localBufferAddr"));
242 3 : CHK_RET(CopyArrayToSlab(
243 : slabPtr, hostChannel.remoteBufferAddr, hostChannel.remoteBufferNum, layout.remoteBufferSection,
244 : &devChannel.remoteBufferAddr, "remoteBufferAddr"));
245 3 : CHK_RET(CopyArrayToSlab(
246 : slabPtr, hostChannel.sqContextAddr, hostChannel.sqNum, layout.sqContextSection, &devChannel.sqContextAddr,
247 : "sqContextAddr"));
248 3 : CHK_RET(CopyArrayToSlab(
249 : slabPtr, hostChannel.cqContextAddr, hostChannel.cqNum, layout.cqContextSection, &devChannel.cqContextAddr,
250 : "cqContextAddr"));
251 3 : return HCCL_SUCCESS;
252 : }
253 :
254 3 : HcclResult CopyChannelEntityHeaderToSlab(
255 : void* slabPtr, const DeviceChannelEntityLayout& layout, const ChannelEntity& devChannel, void*& entityDevPtr)
256 : {
257 3 : entityDevPtr = GetSlabPtr(slabPtr, layout.entitySection);
258 3 : CHK_PTR_NULL(entityDevPtr);
259 3 : Hccl::HrtMemcpy(
260 : entityDevPtr, sizeof(ChannelEntity), &devChannel, sizeof(ChannelEntity),
261 : Hccl::tagRtMemcpyKind::RT_MEMCPY_HOST_TO_DEVICE);
262 3 : return HCCL_SUCCESS;
263 : }
264 : } // namespace
265 :
266 35 : AivUrmaChannel::AivUrmaChannel(EndpointHandle endpointHandle, const HcommChannelDesc& channelDesc)
267 35 : : endpointHandle_(endpointHandle),
268 35 : channelDesc_(channelDesc)
269 : {
270 35 : channelKind_ = HcommChannelKind::AIV_URMA;
271 35 : }
272 :
273 36 : AivUrmaChannel::~AivUrmaChannel()
274 : {
275 35 : PutSocketIfNeeded();
276 35 : ReleaseDeviceChannelEntity();
277 36 : }
278 :
279 37 : void AivUrmaChannel::PutSocketIfNeeded()
280 : {
281 37 : if (socket_ == nullptr) {
282 35 : return;
283 : }
284 2 : if (socketConfig_ == nullptr) {
285 1 : socket_ = nullptr;
286 1 : return;
287 : }
288 1 : if (channelDesc_.socket == nullptr && socket_ != nullptr) {
289 1 : (void)SocketMgr::GetInstance(devicePhyId_).PutSocket(socketConfig_, socket_);
290 : }
291 1 : socket_ = nullptr;
292 : }
293 :
294 40 : void AivUrmaChannel::ReleaseDeviceChannelEntity()
295 : {
296 40 : if (devChannelEntitySlab_ != nullptr) {
297 5 : HcclResult ret = hrtFree(devChannelEntitySlab_);
298 5 : if (ret != HCCL_SUCCESS) {
299 0 : HCCL_WARNING(
300 : "[AivUrmaChannel::%s] hrtFree devChannelEntitySlab failed, ptr[%p], size[%zu], ret[%d]", __func__,
301 : devChannelEntitySlab_, devChannelEntitySlabSize_, ret);
302 : }
303 5 : devChannelEntitySlab_ = nullptr;
304 5 : devChannelEntitySlabSize_ = 0;
305 : }
306 40 : deviceMemories_.clear();
307 40 : devChannelEntity_ = nullptr;
308 40 : }
309 :
310 6 : HcclResult AivUrmaChannel::ParseInputParam()
311 : {
312 : // 1. 从 endpointHandle_，获得 localEp_ 和 rdmaHandle_
313 6 : Endpoint* localEpPtr = reinterpret_cast<Endpoint*>(endpointHandle_);
314 6 : CHK_PTR_NULL(localEpPtr);
315 5 : localEp_ = localEpPtr->GetEndpointDesc();
316 5 : rdmaHandle_ = localEpPtr->GetRdmaHandle();
317 5 : devicePhyId_ = localEp_.loc.device.devPhyId;
318 :
319 5 : socket_ = reinterpret_cast<Hccl::Socket*>(channelDesc_.socket);
320 5 : remoteEp_ = channelDesc_.remoteEndpoint;
321 5 : notifyNum_ = channelDesc_.notifyNum;
322 5 : commonRes_.bufferVec.clear();
323 5 : if (channelDesc_.exchangeAllMems) {
324 3 : HCCL_INFO("[AivUrmaChannel][%s] exchangeAllMems == true. Get memHandles from endpoint.", __func__);
325 3 : std::shared_ptr<Hccl::LocalUbRmaBuffer>* memHandles = nullptr;
326 3 : uint32_t memHandleNum = 0;
327 3 : CHK_RET(GetAllMemHandles(endpointHandle_, reinterpret_cast<void**>(&memHandles), &memHandleNum));
328 2 : HCCL_INFO("[AivUrmaChannel][%s] Got memHandleNum[%u].", __func__, memHandleNum);
329 3 : for (uint32_t i = 0; i < memHandleNum; ++i) {
330 1 : std::shared_ptr<Hccl::LocalUbRmaBuffer>& localUbRmaBuffer = memHandles[i];
331 1 : CHK_SMART_PTR_NULL(localUbRmaBuffer);
332 1 : Hccl::Buffer* buf = localUbRmaBuffer->GetBuf();
333 1 : CHK_PTR_NULL(buf);
334 1 : HCCL_INFO(
335 : "[AivUrmaChannel][%s] Got memHandle No.%u: addr[0x%llx], size[0x%llx], memInfo[%s].", __func__, i,
336 : localUbRmaBuffer->GetAddr(), localUbRmaBuffer->GetSize(), buf->GetMemInfo().c_str());
337 1 : commonRes_.bufferVec.push_back(localUbRmaBuffer.get());
338 : }
339 : } else {
340 2 : HCCL_INFO("[AivUrmaChannel][%s] exchangeAllMems == false. Get memHandles from channelDesc.", __func__);
341 2 : CHK_RET(MakeRmaBufferVecFromMemHandles(
342 : channelDesc_.memHandles, channelDesc_.memHandleNum, commonRes_.bufferVec, "AivUrmaChannel"));
343 : }
344 :
345 4 : return HCCL_SUCCESS;
346 : }
347 :
348 4 : HcclResult AivUrmaChannel::BuildSocket()
349 : {
350 4 : if (socket_ != nullptr) {
351 2 : return HCCL_SUCCESS;
352 : }
353 2 : HCCL_INFO("[AivUrmaChannel][%s] socket ptr is NULL, rebuildSocket", __func__);
354 :
355 2 : Hccl::LinkData linkData = BuildDefaultLinkData();
356 2 : CHK_RET(EndpointDescPairToLinkData(localEp_, remoteEp_, linkData));
357 2 : HCCL_INFO("[AivUrmaChannel][%s] built linkData: %s", __func__, linkData.Describe().c_str());
358 2 : uint16_t port = channelDesc_.port;
359 2 : if (port == 0) {
360 0 : port = DEFAULT_LISTENING_PORT;
361 0 : HCCL_INFO("[AivUrmaChannel::%s] channelDesc port is 0, use default port [%u]", __func__, port);
362 : }
363 : std::string socketTag
364 6 : = (channelDesc_.channelName != nullptr) ? std::string(channelDesc_.channelName) : "AUTOMATIC_SOCKET_TAG";
365 2 : if (channelDesc_.role == HCOMM_SOCKET_ROLE_RESERVED) {
366 1 : EXCEPTION_CATCH(
367 : socketConfigHolder_ = std::make_unique<Hccl::SocketConfig>(linkData, port, socketTag), return HCCL_E_PTR);
368 1 : socketConfigHolder_->noRankId = true;
369 : } else {
370 1 : bool isServer = (channelDesc_.role == HCOMM_SOCKET_ROLE_SERVER);
371 1 : EXCEPTION_CATCH(
372 : socketConfigHolder_ = std::make_unique<Hccl::SocketConfig>(linkData, port, socketTag, isServer),
373 : return HCCL_E_PTR);
374 : }
375 2 : socketConfig_ = socketConfigHolder_.get();
376 2 : CHK_RET(SocketMgr::GetInstance(devicePhyId_).GetSocket(*socketConfigHolder_, socket_));
377 2 : HCCL_INFO("[AivUrmaChannel::%s] SUCCESS. port[%u].", __func__, port);
378 2 : return HCCL_SUCCESS;
379 2 : }
380 :
381 5 : HcclResult AivUrmaChannel::StartListen()
382 : {
383 5 : if (channelDesc_.role != HCOMM_SOCKET_ROLE_SERVER) {
384 3 : return HCCL_SUCCESS;
385 : }
386 :
387 2 : uint16_t port = channelDesc_.port;
388 2 : HCCL_INFO("[AivUrmaChannel::%s] Start. EndpointHandle[%p], port[%u]", __func__, endpointHandle_, port);
389 2 : if (port == 0) {
390 1 : port = DEFAULT_LISTENING_PORT;
391 1 : HCCL_INFO("[AivUrmaChannel::%s] channelDesc port is 0, use default port [%u]", __func__, port);
392 : }
393 2 : CHK_RET(static_cast<HcclResult>(HcommEndpointStartListen(endpointHandle_, port, nullptr)));
394 2 : HCCL_INFO("[AivUrmaChannel::%s] SUCCESS. port[%u].", __func__, port);
395 2 : return HCCL_SUCCESS;
396 : }
397 :
398 2 : HcclResult AivUrmaChannel::BuildAttr()
399 : {
400 2 : attr_.devicePhyId = localEp_.loc.device.devPhyId;
401 2 : attr_.opMode = Hccl::OpMode::OPBASE;
402 2 : attr_.opAcceState = Hccl::AcceleratorState::AIV;
403 2 : return HCCL_SUCCESS;
404 : }
405 :
406 4 : HcclResult AivUrmaChannel::CreateUbConnectionByProtocol(
407 : const UbConnBuildContext& ctx, std::unique_ptr<Hccl::DevUbConnection>& ubConn,
408 : Hccl::DevUbConnection::JettyMode jettyMode)
409 : {
410 4 : Hccl::OpMode opMode = Hccl::OpMode::OPBASE;
411 4 : bool devUsed = true;
412 4 : Hccl::HrtUbJfcMode jfcMode = Hccl::HrtUbJfcMode::USER_CTL;
413 : // UB_CTP → HCOMM_TA_CTP_UB_TIMEOUT，UB_TP → HCOMM_TA_RTP_UB_TIMEOUT
414 4 : u8 taTimeOut = 0;
415 4 : uint32_t taTimeOutValue = 0;
416 4 : if (ctx.protocol == Hccl::LinkProtocol::UB_CTP) {
417 1 : CHK_RET(hcomm::HcommResMgr::GetInstance().GetConfigMgr().GetRdmaConfig().GetTaCtpUbTimeOut(taTimeOutValue));
418 : } else {
419 3 : CHK_RET(hcomm::HcommResMgr::GetInstance().GetConfigMgr().GetRdmaConfig().GetTaRtpUbTimeOut(taTimeOutValue));
420 : }
421 4 : taTimeOut = static_cast<u8>(taTimeOutValue);
422 4 : switch (ctx.protocol) {
423 1 : case Hccl::LinkProtocol::UB_TP:
424 1 : EXCEPTION_CATCH(
425 : ubConn = std::make_unique<Hccl::DevUbTpConnection>(
426 : rdmaHandle_, ctx.locAddr, ctx.rmtAddr, opMode, devUsed, jfcMode, Hccl::IpAddress(),
427 : Hccl::IpAddress(), ctx.qosPre, taTimeOut, COMM_ENGINE_AIV, ctx.sqDepth, jettyMode),
428 : return HCCL_E_PTR);
429 1 : break;
430 1 : case Hccl::LinkProtocol::UB_CTP:
431 1 : EXCEPTION_CATCH(
432 : ubConn = std::make_unique<Hccl::DevUbCtpConnection>(
433 : rdmaHandle_, ctx.locAddr, ctx.rmtAddr, opMode, devUsed, jfcMode, Hccl::IpAddress(),
434 : Hccl::IpAddress(), ctx.qosPre, taTimeOut, COMM_ENGINE_AIV, ctx.sqDepth, jettyMode),
435 : return HCCL_E_PTR);
436 1 : break;
437 2 : case Hccl::LinkProtocol::UB_RTP:
438 2 : EXCEPTION_CATCH(
439 : ubConn = std::make_unique<Hccl::DevUbRtpConnection>(
440 : rdmaHandle_, ctx.locAddr, ctx.rmtAddr, opMode, devUsed, jfcMode, ctx.locAddr, ctx.rmtAddr,
441 : ctx.qosPre, taTimeOut, COMM_ENGINE_AIV, ctx.sqDepth, jettyMode),
442 : return HCCL_E_PTR);
443 2 : break;
444 0 : default:
445 0 : HCCL_ERROR("%s No LinkProtocol to match", __func__);
446 0 : break;
447 : }
448 4 : return HCCL_SUCCESS;
449 : }
450 :
451 : HcclResult
452 0 : AivUrmaChannel::AcquireSharedJettyInBuildConnection(const UbConnBuildContext& ctx, Hccl::DevUbConnection* connection)
453 : {
454 : // 共享 jetty 模式：复用同 Endpoint 下已创建的 jetty
455 0 : Endpoint* endpoint = reinterpret_cast<Endpoint*>(endpointHandle_);
456 : // UB_CTP → HCOMM_TA_CTP_UB_TIMEOUT，UB_TP → HCOMM_TA_RTP_UB_TIMEOUT
457 0 : u8 taTimeOut = 0;
458 0 : uint32_t taTimeOutValue = 0;
459 0 : if (ctx.protocol == Hccl::LinkProtocol::UB_CTP) {
460 0 : CHK_RET(hcomm::HcommResMgr::GetInstance().GetConfigMgr().GetRdmaConfig().GetTaCtpUbTimeOut(taTimeOutValue));
461 : } else {
462 0 : CHK_RET(hcomm::HcommResMgr::GetInstance().GetConfigMgr().GetRdmaConfig().GetTaRtpUbTimeOut(taTimeOutValue));
463 : }
464 0 : taTimeOut = static_cast<u8>(taTimeOutValue);
465 0 : auto tempFactory = [rdmaHandle = rdmaHandle_, &ctxLoc = ctx.locAddr, &ctxRmt = ctx.rmtAddr, qosPre = ctx.qosPre,
466 0 : protocol = ctx.protocol, sqDepth = ctx.sqDepth,
467 : taTimeOut = taTimeOut]() -> std::unique_ptr<Hccl::DevUbConnection> {
468 0 : return CreateSharedJettyConnection(rdmaHandle, ctxLoc, ctxRmt, qosPre, protocol, taTimeOut, sqDepth);
469 0 : };
470 0 : Endpoint::SharedJettyCtx sharedCtx{};
471 0 : CHK_RET(hcomm::AcquireSharedJettyForChannel(endpoint, connection, tempFactory, sharedCtx));
472 : // 保存共享 PI/CI 指针，供 BuildChannelEntityToDevice 绑给 transport
473 0 : sharedSqPiPtr_ = sharedCtx.sqPiPtr;
474 0 : sharedSqCiPtr_ = sharedCtx.sqCiPtr;
475 0 : sharedCqPiPtr_ = sharedCtx.cqPiPtr;
476 0 : sharedCqCiPtr_ = sharedCtx.cqCiPtr;
477 0 : return HCCL_SUCCESS;
478 : }
479 :
480 0 : std::unique_ptr<Hccl::DevUbConnection> AivUrmaChannel::CreateSharedJettyConnection(
481 : Hccl::RdmaHandle rdmaHandle, const Hccl::IpAddress& locAddr, const Hccl::IpAddress& rmtAddr, u8 qosPre,
482 : Hccl::LinkProtocol protocol, u8 taTimeOut, u32 sqDepth)
483 : {
484 0 : switch (protocol) {
485 0 : case Hccl::LinkProtocol::UB_TP: {
486 0 : std::unique_ptr<Hccl::DevUbConnection> conn;
487 0 : EXCEPTION_CATCH(
488 : conn = std::make_unique<Hccl::DevUbTpConnection>(
489 : rdmaHandle, locAddr, rmtAddr, Hccl::OpMode::OPBASE, true, Hccl::HrtUbJfcMode::USER_CTL,
490 : Hccl::IpAddress(), Hccl::IpAddress(), qosPre, taTimeOut, COMM_ENGINE_AIV, sqDepth),
491 : return nullptr);
492 0 : return conn;
493 0 : }
494 0 : case Hccl::LinkProtocol::UB_CTP: {
495 0 : std::unique_ptr<Hccl::DevUbConnection> conn;
496 0 : EXCEPTION_CATCH(
497 : conn = std::make_unique<Hccl::DevUbCtpConnection>(
498 : rdmaHandle, locAddr, rmtAddr, Hccl::OpMode::OPBASE, true, Hccl::HrtUbJfcMode::USER_CTL,
499 : Hccl::IpAddress(), Hccl::IpAddress(), qosPre, taTimeOut, COMM_ENGINE_AIV, sqDepth),
500 : return nullptr);
501 0 : return conn;
502 0 : }
503 0 : case Hccl::LinkProtocol::UB_RTP: {
504 0 : std::unique_ptr<Hccl::DevUbConnection> conn;
505 0 : EXCEPTION_CATCH(
506 : conn = std::make_unique<Hccl::DevUbRtpConnection>(
507 : rdmaHandle, locAddr, rmtAddr, Hccl::OpMode::OPBASE, true, Hccl::HrtUbJfcMode::USER_CTL, locAddr,
508 : rmtAddr, qosPre, taTimeOut, COMM_ENGINE_AIV, sqDepth),
509 : return nullptr);
510 0 : return conn;
511 0 : }
512 0 : default:
513 0 : HCCL_ERROR(
514 : "[AivUrmaChannel][CreateSharedJettyConnection] unsupported protocol[%s], return nullptr.",
515 : protocol.Describe().c_str());
516 0 : return nullptr;
517 : }
518 : }
519 :
520 4 : HcclResult AivUrmaChannel::BuildConnection()
521 : {
522 4 : UbConnBuildContext ctx;
523 4 : CHK_RET(PrepareUbConnBuildContext(localEp_, remoteEp_, channelDesc_, ctx));
524 4 : CHK_RET(CheckUbSqDepth(ctx, devBaseAttr_));
525 :
526 : // 共享 jetty 模式：主 connection 构造时传 EXTERNAL_INJECT，跳过建 JFC/jetty，等 SetSharedJettyFields 填充
527 4 : auto jettyMode = IsSharedJetty() ? Hccl::DevUbConnection::JettyMode::EXTERNAL_INJECT :
528 4 : Hccl::DevUbConnection::JettyMode::SELF_CREATE;
529 4 : std::unique_ptr<Hccl::DevUbConnection> ubConn = nullptr;
530 4 : CHK_RET(CreateUbConnectionByProtocol(ctx, ubConn, jettyMode));
531 4 : CHK_SMART_PTR_NULL(ubConn);
532 :
533 : // 共享 jetty 模式：复用同 Endpoint 下已创建的 jetty。
534 : // 必须在 push_back(move(ubConn)) 之前调用：AcquireSharedJetty 失败时 ubConn 仍为局部变量，
535 : // 函数返回时自动析构，不会在 connections_/connVec 中残留不完整 connection。
536 4 : if (IsSharedJetty()) {
537 0 : CHK_RET(AcquireSharedJettyInBuildConnection(ctx, ubConn.get()));
538 : }
539 :
540 4 : commonRes_.connVec.clear();
541 4 : connections_.clear();
542 4 : commonRes_.connVec.emplace_back(ubConn.get());
543 4 : connections_.push_back(std::move(ubConn));
544 :
545 4 : return HCCL_SUCCESS;
546 4 : }
547 :
548 1 : HcclResult AivUrmaChannel::BuildAivUrmaTransport()
549 : {
550 1 : const Hccl::Socket& socket = *socket_;
551 :
552 1 : Hccl::LinkData linkData = BuildDefaultLinkData();
553 1 : CHK_RET(EndpointDescPairToLinkData(localEp_, remoteEp_, linkData));
554 :
555 : // make_unique / make_shared / release 包一层抛异常的宏
556 1 : EXCEPTION_CATCH(
557 : transport_ = std::make_unique<Hccl::AivUrmaTransport>(
558 : commonRes_, attr_, linkData, socket, rdmaHandle_), // 这里区分是否是优先recv
559 : return HCCL_E_PTR);
560 1 : return HCCL_SUCCESS;
561 : }
562 :
563 5 : HcclResult AivUrmaChannel::BuildChannelEntityToDevice(void** devChannelPtr)
564 : {
565 5 : if (devChannelPtr == nullptr) {
566 1 : HCCL_ERROR("[AivUrmaChannel] BuildChannelEntityToDevice devChannelPtr is nullptr");
567 1 : return HCCL_E_PTR;
568 : }
569 :
570 4 : CHK_PTR_NULL(transport_.get());
571 :
572 : ChannelEntity hostChannel;
573 3 : CHK_RET(SecureMemset(&hostChannel, sizeof(ChannelEntity), 0, sizeof(ChannelEntity), "hostChannel"));
574 :
575 3 : transport_->PrepareHostChannelEntity(&hostChannel);
576 :
577 3 : DeviceChannelEntityLayout layout;
578 3 : CHK_RET(BuildDeviceChannelEntityLayout(hostChannel, layout));
579 3 : void* slabPtr = nullptr;
580 3 : AclDeviceSlabGuard slabGuard;
581 3 : CHK_RET(AllocDeviceEntitySlab(layout.slabSize, slabGuard, slabPtr));
582 2 : uint32_t queueNum = std::max(hostChannel.sqNum, hostChannel.cqNum);
583 2 : if (IsSharedJetty() && sharedSqPiPtr_ != nullptr) {
584 : // 共享 jetty：PI/CI 用同 endpoint 下多 channel 共享的 device 内存，slab 内 PI/CI 段闲置不用。
585 : // 共享内存在首次 AcquireSharedJettyForChannel 时已分配并清零，此处直接绑给 transport。
586 0 : transport_->SetQueueIndexDeviceMem(
587 0 : sharedSqPiPtr_, sharedSqCiPtr_, sharedCqPiPtr_, sharedCqCiPtr_, queueNum * QUEUE_INDEX_MEM_UNIT_SIZE);
588 : } else {
589 2 : CHK_RET(InitQueueIndexSections(slabPtr, layout, queueNum));
590 2 : SetQueueIndexDeviceMem(*transport_, slabPtr, layout, queueNum);
591 : }
592 :
593 2 : CHK_RET(SecureMemset(&hostChannel, sizeof(ChannelEntity), 0, sizeof(ChannelEntity), "hostChannel"));
594 2 : transport_->GetHostChannelEntity(&hostChannel);
595 2 : hostChannel.abiHeader = channelDesc_.header;
596 2 : hostChannel.engine = COMM_ENGINE_AIV;
597 2 : hostChannel.protocol = channelDesc_.remoteEndpoint.protocol;
598 :
599 : ChannelEntity devChannel;
600 2 : CHK_RET(CopyChannelEntityToSlab(slabPtr, hostChannel, layout, devChannel));
601 2 : void* entityDevPtr = nullptr;
602 2 : CHK_RET(CopyChannelEntityHeaderToSlab(slabPtr, layout, devChannel, entityDevPtr));
603 2 : ReleaseDeviceChannelEntity();
604 2 : devChannelEntitySlab_ = slabGuard.Release();
605 2 : devChannelEntitySlabSize_ = layout.slabSize;
606 2 : devChannelEntity_ = entityDevPtr;
607 2 : *devChannelPtr = devChannelEntity_;
608 2 : HCCL_INFO(
609 : "[AivUrmaChannel] Build channel entity to device success, devPtr[%p], slabPtr[%p], slabSize[%zu]",
610 : devChannelEntity_, devChannelEntitySlab_, devChannelEntitySlabSize_);
611 2 : return HCCL_SUCCESS;
612 3 : }
613 :
614 6 : HcclResult AivUrmaChannel::PreAllocChannelEntityToDevice(void** devChannelPtr)
615 : {
616 6 : if (devChannelPtr == nullptr) {
617 1 : HCCL_ERROR("[AivUrmaChannel::%s] devChannelPtr is nullptr", __func__);
618 1 : return HCCL_E_PTR;
619 : }
620 5 : CHK_PTR_NULL(transport_.get());
621 :
622 4 : if (devChannelEntitySlab_ != nullptr) {
623 1 : *devChannelPtr = devChannelEntity_;
624 1 : HCCL_INFO("[AivUrmaChannel::%s] already built, return cached devPtr[%p]", __func__, devChannelEntity_);
625 1 : return HCCL_SUCCESS;
626 : }
627 :
628 3 : uint32_t bufNum = 0;
629 3 : uint32_t connNum = 0;
630 3 : transport_->GetEntityCountsForLayout(bufNum, connNum);
631 :
632 3 : ChannelEntity tmp{};
633 3 : tmp.localBufferNum = bufNum;
634 3 : tmp.remoteBufferNum = bufNum;
635 3 : tmp.sqNum = connNum;
636 3 : tmp.cqNum = connNum;
637 :
638 3 : DeviceChannelEntityLayout layout;
639 3 : CHK_RET(BuildDeviceChannelEntityLayout(tmp, layout));
640 :
641 3 : void* slabPtr = nullptr;
642 3 : AclDeviceSlabGuard slabGuard;
643 3 : CHK_RET(AllocDeviceEntitySlab(layout.slabSize, slabGuard, slabPtr));
644 :
645 3 : uint32_t queueNum = std::max(tmp.sqNum, tmp.cqNum);
646 3 : if (IsSharedJetty() && sharedSqPiPtr_ != nullptr) {
647 : // 共享 jetty：PI/CI 用共享 device 内存，slab 内 PI/CI 段闲置不用（已在首次分配时清零）
648 0 : transport_->SetQueueIndexDeviceMem(
649 0 : sharedSqPiPtr_, sharedSqCiPtr_, sharedCqPiPtr_, sharedCqCiPtr_, queueNum * QUEUE_INDEX_MEM_UNIT_SIZE);
650 : } else {
651 3 : CHK_RET(InitQueueIndexSections(slabPtr, layout, queueNum));
652 : }
653 :
654 3 : devChannelEntitySlab_ = slabGuard.Release();
655 3 : devChannelEntitySlabSize_ = layout.slabSize;
656 3 : devChannelEntity_ = GetSlabPtr(devChannelEntitySlab_, layout.entitySection);
657 3 : if (!IsSharedJetty() || sharedSqPiPtr_ == nullptr) {
658 3 : SetQueueIndexDeviceMem(*transport_, devChannelEntitySlab_, layout, queueNum);
659 : }
660 3 : *devChannelPtr = devChannelEntity_;
661 :
662 3 : HCCL_INFO(
663 : "[AivUrmaChannel::%s] pre-alloc success, devPtr[%p], slabPtr[%p], slabSize[%zu]", __func__, devChannelEntity_,
664 : devChannelEntitySlab_, devChannelEntitySlabSize_);
665 3 : return HCCL_SUCCESS;
666 3 : }
667 :
668 2 : HcclResult AivUrmaChannel::FillChannelEntityToDevice()
669 : {
670 2 : if (devChannelEntitySlab_ == nullptr) {
671 1 : HCCL_ERROR("[AivUrmaChannel::%s] devChannelEntitySlab_ is nullptr, not pre-allocated.", __func__);
672 1 : return HCCL_E_INTERNAL;
673 : }
674 1 : CHK_PTR_NULL(transport_.get());
675 :
676 : ChannelEntity hostChannel;
677 1 : CHK_RET(SecureMemset(&hostChannel, sizeof(ChannelEntity), 0, sizeof(ChannelEntity), "hostChannel"));
678 1 : transport_->GetHostChannelEntity(&hostChannel);
679 1 : hostChannel.abiHeader = channelDesc_.header;
680 1 : hostChannel.engine = COMM_ENGINE_AIV;
681 1 : hostChannel.protocol = channelDesc_.remoteEndpoint.protocol;
682 :
683 1 : DeviceChannelEntityLayout layout;
684 1 : CHK_RET(BuildDeviceChannelEntityLayout(hostChannel, layout));
685 1 : if (layout.slabSize > devChannelEntitySlabSize_) {
686 0 : HCCL_ERROR(
687 : "[AivUrmaChannel::%s] slabSize[%zu] > preAllocSize[%zu]", __func__, layout.slabSize,
688 : devChannelEntitySlabSize_);
689 0 : return HCCL_E_INTERNAL;
690 : }
691 :
692 : ChannelEntity devChannel;
693 1 : CHK_RET(CopyChannelEntityToSlab(devChannelEntitySlab_, hostChannel, layout, devChannel));
694 1 : void* entityDevPtr = nullptr;
695 1 : CHK_RET(CopyChannelEntityHeaderToSlab(devChannelEntitySlab_, layout, devChannel, entityDevPtr));
696 :
697 1 : devChannelEntity_ = entityDevPtr;
698 1 : HCCL_INFO("[AivUrmaChannel::%s] fill success, devPtr[%p]", __func__, devChannelEntity_);
699 1 : return HCCL_SUCCESS;
700 : }
701 :
702 1 : HcclResult AivUrmaChannel::GetNotifyNum([[maybe_unused]] uint32_t* notifyNum) const
703 : {
704 1 : HCCL_INFO("AivUrmaChannel GetNotifyNum is not supported.");
705 1 : return HCCL_SUCCESS;
706 : }
707 :
708 0 : HcclResult AivUrmaChannel::GetRemoteMems(uint32_t* memNum, CommMem** remoteMem, char*** memInfos)
709 : {
710 0 : return transport_->GetRemoteMems(memNum, remoteMem, memInfos);
711 : }
712 :
713 3 : HcclResult AivUrmaChannel::Clean()
714 : {
715 3 : ReleaseDeviceChannelEntity();
716 3 : ReleasePtrArrayDevMems();
717 3 : transport_.reset();
718 3 : return HCCL_SUCCESS;
719 : }
720 :
721 1 : HcclResult AivUrmaChannel::Resume()
722 : {
723 1 : BuildConnection();
724 1 : BuildAivUrmaTransport();
725 1 : return HCCL_SUCCESS;
726 : }
727 :
728 1 : HcclResult AivUrmaChannel::NotifyRecord([[maybe_unused]] const uint32_t remoteNotifyIdx)
729 : {
730 1 : HCCL_INFO("[AivUrmaChannel::%s] not supported yet.", __func__);
731 1 : return HCCL_E_NOT_SUPPORT;
732 : }
733 :
734 : HcclResult
735 1 : AivUrmaChannel::NotifyWait([[maybe_unused]] const uint32_t localNotifyIdx, [[maybe_unused]] const uint32_t timeout)
736 : {
737 1 : HCCL_INFO("[AivUrmaChannel::%s] not supported yet.", __func__);
738 1 : return HCCL_E_NOT_SUPPORT;
739 : }
740 :
741 1 : HcclResult AivUrmaChannel::WriteWithNotify(
742 : [[maybe_unused]] void* dst, [[maybe_unused]] const void* src, [[maybe_unused]] const uint64_t len,
743 : [[maybe_unused]] uint32_t remoteNotifyIdx)
744 : {
745 1 : HCCL_INFO("[AivUrmaChannel::%s] not supported yet.", __func__);
746 1 : return HCCL_E_NOT_SUPPORT;
747 : }
748 :
749 : HcclResult
750 1 : AivUrmaChannel::Write([[maybe_unused]] void* dst, [[maybe_unused]] const void* src, [[maybe_unused]] uint64_t len)
751 : {
752 1 : HCCL_INFO("[AivUrmaChannel::%s] not supported yet.", __func__);
753 1 : return HCCL_E_NOT_SUPPORT;
754 : }
755 :
756 : HcclResult
757 1 : AivUrmaChannel::Read([[maybe_unused]] void* dst, [[maybe_unused]] const void* src, [[maybe_unused]] uint64_t len)
758 : {
759 1 : HCCL_INFO("[AivUrmaChannel::%s] not supported yet.", __func__);
760 1 : return HCCL_E_NOT_SUPPORT;
761 : }
762 :
763 1 : HcclResult AivUrmaChannel::ChannelFence()
764 : {
765 1 : HCCL_INFO("[AivUrmaChannel::%s] not supported yet.", __func__);
766 1 : return HCCL_E_NOT_SUPPORT;
767 : }
768 :
769 3 : HcclResult AivUrmaChannel::Init()
770 : {
771 : /*
772 : Argue result: make_unique 配合一场捕获的宏 EXCEPTION CATCH
773 : Attention: const 和引用
774 : */
775 3 : CHK_RET(ParseInputParam());
776 2 : CHK_RET(StartListen());
777 2 : CHK_RET(BuildSocket());
778 2 : CHK_RET(BuildAttr());
779 :
780 2 : CHK_RET(HccpRaGetDevBaseAttr(rdmaHandle_, &devBaseAttr_));
781 2 : CHK_RET(BuildConnection());
782 2 : CHK_RET(BuildAivUrmaTransport());
783 2 : return HCCL_SUCCESS;
784 : }
785 :
786 1 : ChannelStatus AivUrmaChannel::GetStatus()
787 : {
788 1 : Hccl::TransportStatus transportStatus = transport_->GetStatus();
789 1 : ChannelStatus out = ChannelStatus::INIT;
790 1 : switch (transportStatus) {
791 0 : case Hccl::TransportStatus::INIT:
792 0 : out = ChannelStatus::INIT;
793 0 : break;
794 0 : case Hccl::TransportStatus::SOCKET_OK:
795 0 : out = ChannelStatus::SOCKET_OK;
796 0 : break;
797 1 : case Hccl::TransportStatus::SOCKET_TIMEOUT:
798 1 : out = ChannelStatus::SOCKET_TIMEOUT;
799 1 : break;
800 0 : case Hccl::TransportStatus::READY:
801 0 : out = ChannelStatus::READY;
802 0 : break;
803 0 : default:
804 0 : HCCL_ERROR("[AivUrmaChannel][%s] Invalid TransportStatus[%d]", __func__, transportStatus);
805 0 : out = ChannelStatus::INVALID;
806 0 : break;
807 : }
808 1 : return out;
809 : }
810 :
811 : } // namespace hcomm
|
