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