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 "dev_aicpu_ts_roce_channel.h"
12 : #include <securec.h>
13 : #include <chrono>
14 : #include <cstdint>
15 : #include <cstring>
16 : #include <vector>
17 : #include "channel_param.h"
18 : #include "adapter_hal_pub.h"
19 : #include "adapter_rts_common.h"
20 : #include "dispatcher_ctx.h"
21 : #include "hccl_dispatcher_ctx.h"
22 : #include "log.h"
23 : #include "transport_pub.h"
24 :
25 : using namespace hccl;
26 :
27 : u64 DevAicpuTsRoceChannel::commSeq_{0};
28 :
29 2 : DevAicpuTsRoceChannel::~DevAicpuTsRoceChannel()
30 : {
31 1 : for (auto &pair : slots_) {
32 0 : if (pair.second.link != nullptr) {
33 0 : (void)pair.second.link->DeInit();
34 0 : pair.second.link.reset();
35 : }
36 0 : if (pair.second.ctx != nullptr) {
37 0 : (void)DestroyDispatcherCtx(pair.second.ctx, pair.second.commId);
38 : }
39 : }
40 1 : slots_.clear();
41 2 : }
42 :
43 : namespace {
44 :
45 : constexpr u32 RDMA_QP_MAX_NUM = 32U;
46 : constexpr u32 HCCL_MULTI_QP_THRESHOLD_DEFAULT = 0U;
47 :
48 : constexpr u32 DEFAULT_TIMEOUT_MS = 10000;
49 :
50 1 : HcclResult FillIbverbsDataFromRes(const HcommRoceChannelRes *res, TransportDeviceIbverbsData &ibd)
51 : {
52 1 : std::vector<RoceMemDetails> localMd;
53 1 : auto localBase = static_cast<const RoceMemDetails *>(res->localMem);
54 1 : if (localBase != nullptr) {
55 0 : for (u32 i = 0; i < res->localMemCount; ++i) {
56 0 : localMd.push_back(localBase[i]);
57 : }
58 : }
59 1 : std::vector<RoceMemDetails> remoteMd;
60 1 : auto remoteBase = static_cast<const RoceMemDetails *>(res->remoteMem);
61 1 : if (remoteBase != nullptr) {
62 0 : for (u32 i = 0; i < res->remoteMemCount; ++i) {
63 0 : remoteMd.push_back(remoteBase[i]);
64 : }
65 : }
66 1 : HCCL_INFO("[DevAicpuTsRoceChannel][Create] Roce mem from channel res: localMemCount[%u] remoteMemCount[%u] "
67 : "parsed local[%zu] remote[%zu]",
68 : res->localMemCount, res->remoteMemCount, localMd.size(), remoteMd.size());
69 :
70 1 : const u32 qpInfoSize = res->qpsPerConnection + static_cast<u32>(res->qpsPerConnection != 1U);
71 1 : if (qpInfoSize < 1U || qpInfoSize > RDMA_QP_MAX_NUM) {
72 0 : HCCL_ERROR("[DevAicpuTsRoceChannel][Create] bad qp layout qpsPerConn[%u]", res->qpsPerConnection);
73 0 : return HCCL_E_PARA;
74 : }
75 1 : std::vector<HcclQpInfoV2> qpVec(qpInfoSize);
76 2 : for (u32 i = 0; i < qpInfoSize; ++i) {
77 1 : qpVec[i] = res->QpInfo[i];
78 : }
79 :
80 1 : ibd.qpInfo = std::move(qpVec);
81 1 : ibd.qpsPerConnection = res->qpsPerConnection;
82 1 : ibd.multiQpThreshold = HCCL_MULTI_QP_THRESHOLD_DEFAULT;
83 1 : ibd.localRoceMemDetailsList = std::move(localMd);
84 1 : ibd.remoteRoceMemDetailsList = std::move(remoteMd);
85 1 : ibd.useMemDetailsMgr = true;
86 1 : ibd.remoteNotifyValueAddr = reinterpret_cast<uint64_t>(res->remoteNotifyAddr);
87 1 : ibd.remoteNotifyValueKey = res->remoteNotifyKey;
88 1 : ibd.localDataNotifyAddr = reinterpret_cast<uint64_t>(res->localDataNotifyAddr);
89 1 : ibd.localDataNotifyKey = res->localDataNotifyKey;
90 1 : ibd.notifySize = res->notifySize;
91 1 : HCCL_DEBUG("[%s]remoteNotifyAddr[%llu], remoteNotifyKey[%u], localDataNotifyAddr[%llu], localDataNotifyKey[%u]," \
92 : "notifySize[%u]", __func__, ibd.remoteNotifyValueAddr, ibd.remoteNotifyValueKey, ibd.localDataNotifyAddr,
93 : ibd.localDataNotifyKey, ibd.notifySize);
94 :
95 1 : EXCEPTION_CATCH((ibd.dataNotify = std::make_shared<LocalNotify>()), return HCCL_E_PTR);
96 1 : CHK_SMART_PTR_NULL(ibd.dataNotify);
97 1 : CHK_RET(ibd.dataNotify->Init(res->localDataSignal, NotifyLoadType::DEVICE_NOTIFY));
98 1 : return HCCL_SUCCESS;
99 1 : }
100 :
101 1 : HcclResult OpenDispatcherForTsRoce(const HcommDeviceInfo &deviceInfo, char *commId, size_t commIdLen,
102 : u32 &outDevId, DispatcherCtxPtr &outDctx, HcclDispatcher &outDispatcher)
103 : {
104 1 : outDevId = INVALID_UINT;
105 1 : CHK_RET(hrtDrvGetLocalDevIDByHostDevID(deviceInfo.devicePhyId, &outDevId));
106 1 : CHK_PRT_RET(outDevId == INVALID_UINT,
107 : HCCL_ERROR("[DevAicpuTsRoceChannel][Create] invalid devId for logicId[%d]", deviceInfo.deviceLogicId),
108 : HCCL_E_PARA);
109 :
110 1 : DispatcherCtxPtr dctxPtr = nullptr;
111 1 : CHK_RET(CreateDispatcherCtx(&dctxPtr, outDevId, commId));
112 1 : CHK_PTR_NULL(dctxPtr);
113 1 : auto *dctx = static_cast<DispatcherCtx *>(dctxPtr);
114 1 : const HcclDispatcher dispatcher = dctx->GetDispatcher();
115 1 : if (dispatcher == nullptr) {
116 0 : (void)DestroyDispatcherCtx(dctxPtr, commId);
117 0 : HCCL_ERROR("[DevAicpuTsRoceChannel][Create] null dispatcher");
118 0 : return HCCL_E_PTR;
119 : }
120 1 : outDctx = dctxPtr;
121 1 : outDispatcher = dispatcher;
122 1 : return HCCL_SUCCESS;
123 : }
124 :
125 1 : HcclResult CreateAndInitTsRoceTransport(const HcommDeviceInfo &deviceInfo, DispatcherCtxPtr dctxPtr,
126 : const char *commId, HcclDispatcher dispatcher, TransportDeviceIbverbsData &&ibd,
127 : std::shared_ptr<Transport> &outLink)
128 : {
129 1 : MachinePara machinePara{};
130 1 : machinePara.deviceLogicId = deviceInfo.deviceLogicId;
131 1 : machinePara.localDeviceId = deviceInfo.devicePhyId;
132 1 : DevType devType = DevType::DEV_TYPE_COUNT;
133 1 : CHK_RET(hrtGetDeviceType(devType));
134 1 : machinePara.deviceType = devType;
135 1 : machinePara.isAicpuModeEn = true;
136 1 : machinePara.isIndOp = true;
137 1 : machinePara.notifyNum = 0;
138 1 : machinePara.nicDeploy = NICDeployment::NIC_DEPLOYMENT_DEVICE;
139 1 : machinePara.tag = "hcomm_aicpu_ts_roce" + std::string(commId);
140 1 : machinePara.userMemEnable = false;
141 1 : machinePara.drainEnable = true;
142 1 : machinePara.dctxPtr = dctxPtr;
143 :
144 1 : TransportPara transportPara{};
145 1 : transportPara.timeout = std::chrono::milliseconds(DEFAULT_TIMEOUT_MS);
146 1 : transportPara.nicDeploy = NICDeployment::NIC_DEPLOYMENT_DEVICE;
147 :
148 1 : static const std::unique_ptr<NotifyPool> kEmptyNotifyPool;
149 1 : std::shared_ptr<Transport> link;
150 1 : link.reset(new (std::nothrow) Transport(TransportType::TRANS_TYPE_DEVICE_IBVERBS, transportPara, dispatcher,
151 2 : kEmptyNotifyPool, machinePara, TransportDeviceP2pData(), ibd));
152 1 : if (link == nullptr) {
153 0 : (void)DestroyDispatcherCtx(dctxPtr, commId);
154 0 : HCCL_ERROR("[DevAicpuTsRoceChannel][Create] Transport alloc failed");
155 0 : return HCCL_E_PTR;
156 : }
157 1 : HcclResult tr = link->Init();
158 1 : if (tr != HCCL_SUCCESS) {
159 0 : link.reset();
160 0 : (void)DestroyDispatcherCtx(dctxPtr, commId);
161 0 : return tr;
162 : }
163 1 : tr = link->InitDrainNotifyInfo();
164 1 : if (tr != HCCL_SUCCESS) {
165 0 : link.reset();
166 0 : (void)DestroyDispatcherCtx(dctxPtr, commId);
167 0 : return tr;
168 : }
169 1 : outLink = std::move(link);
170 1 : return HCCL_SUCCESS;
171 1 : }
172 :
173 : } // namespace
174 :
175 1 : HcclResult DevAicpuTsRoceChannel::Create(const void *blob, u64 blobBytes,
176 : const HcommDeviceInfo &deviceInfo, ChannelHandle &outHandle)
177 : {
178 1 : CHK_PTR_NULL(blob);
179 1 : if (blobBytes < sizeof(HcommRoceChannelRes)) {
180 0 : HCCL_ERROR("[DevAicpuTsRoceChannel][Create] blob too small[%llu]",
181 : static_cast<unsigned long long>(blobBytes));
182 0 : return HCCL_E_PARA;
183 : }
184 1 : const auto *res = static_cast<const HcommRoceChannelRes *>(blob);
185 :
186 1 : TransportDeviceIbverbsData ibd{};
187 1 : CHK_RET(FillIbverbsDataFromRes(res, ibd));
188 1 : const u32 qpInfoSize = res->qpsPerConnection + static_cast<u32>(res->qpsPerConnection != 1U);
189 :
190 : char commId[sizeof(RoceSlot::commId)];
191 1 : u32 devId = INVALID_UINT;
192 1 : DispatcherCtxPtr dctxPtr = nullptr;
193 1 : HcclDispatcher dispatcher = nullptr;
194 :
195 : {
196 1 : std::lock_guard<std::mutex> lock(mutex_);
197 1 : ++commSeq_;
198 2 : int nc = snprintf_s(commId, sizeof(commId), sizeof(commId) - 1U, "hcomm_ts_roce_%d_%llu",
199 1 : deviceInfo.deviceLogicId, static_cast<unsigned long long>(commSeq_));
200 1 : CHK_PRT_RET(nc < 0, HCCL_ERROR("[DevAicpuTsRoceChannel][Create] snprintf_s failed"), HCCL_E_INTERNAL);
201 1 : }
202 :
203 1 : CHK_RET(OpenDispatcherForTsRoce(deviceInfo, commId, sizeof(commId), devId, dctxPtr, dispatcher));
204 :
205 1 : std::shared_ptr<Transport> link;
206 1 : CHK_RET(CreateAndInitTsRoceTransport(deviceInfo, dctxPtr, commId, dispatcher, std::move(ibd), link));
207 :
208 1 : outHandle = reinterpret_cast<ChannelHandle>(link.get());
209 1 : RoceSlot slot;
210 1 : slot.ctx = dctxPtr;
211 1 : slot.link = std::move(link);
212 1 : CHK_SAFETY_FUNC_RET(memcpy_s(slot.commId, sizeof(slot.commId), commId, sizeof(commId)));
213 :
214 : {
215 1 : std::lock_guard<std::mutex> lock(mutex_);
216 1 : slots_.emplace(outHandle, std::move(slot));
217 1 : }
218 :
219 1 : HCCL_INFO("[DevAicpuTsRoceChannel][Create] success logicId[%d] phyId[%u] devId[%u] blobBytes[%llu] "
220 : "localMem[%u] remoteMem[%u] qpsPerConn[%u] qpNum[%u] chipId[%lld] commId[%s] handle[0x%llx]",
221 : deviceInfo.deviceLogicId, deviceInfo.devicePhyId, devId,
222 : static_cast<unsigned long long>(blobBytes), res->localMemCount, res->remoteMemCount, res->qpsPerConnection,
223 : qpInfoSize, static_cast<long long>(res->chipId), commId,
224 : static_cast<unsigned long long>(reinterpret_cast<uintptr_t>(outHandle)));
225 1 : return HCCL_SUCCESS;
226 1 : }
227 :
228 1 : bool DevAicpuTsRoceChannel::Destroy(ChannelHandle handle)
229 : {
230 1 : RoceSlot slot;
231 : {
232 1 : std::lock_guard<std::mutex> lock(mutex_);
233 1 : auto it = slots_.find(handle);
234 1 : if (it == slots_.end()) {
235 0 : return false;
236 : }
237 1 : slot = std::move(it->second);
238 1 : slots_.erase(it);
239 1 : }
240 1 : if (slot.link != nullptr) {
241 1 : (void)slot.link->DeInit();
242 1 : slot.link.reset();
243 : }
244 1 : if (slot.ctx != nullptr) {
245 1 : (void)DestroyDispatcherCtx(slot.ctx, slot.commId);
246 : }
247 1 : HCCL_DEBUG("[DevAicpuTsRoceChannel][Destroy] destroyed handle[0x%llx]", handle);
248 1 : return true;
249 1 : }
|
