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