Line data Source code
1 : /**
2 : * Copyright (c) 2025 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 "multi_qpInfo_manager.h"
12 : #include "adapter_hccp_common.h"
13 : #include <queue>
14 : #include <cstdlib>
15 : #include <fstream>
16 : #include "externalinput_pub.h"
17 : #include "adapter_error_manager_pub.h"
18 : #include "../../../nslbdp/hccl_nslbdp.h"
19 : #include "mmpa_api.h"
20 : namespace hccl {
21 : static std::vector<std::string> devCfgPortMode{"multi_qp", "nslb_dp", ""};
22 :
23 0 : static std::string MultiQpFromToString(MUL_QP_FROM value)
24 : {
25 : static const std::map<MUL_QP_FROM, std::string> map = {
26 0 : {MUL_QP_FROM::MUL_QP_FROM_DEV_CFG, "DEV_CFG"},
27 0 : {MUL_QP_FROM::MUL_QP_FROM_DEV_NSLB, "DEV_NSLB"},
28 0 : {MUL_QP_FROM::MUL_QP_FROM_ENV_PORT_CONFIG_PATH, "ENV_HCCL_RDMA_QP_PORT_CONFIG_PATH"},
29 0 : {MUL_QP_FROM::MUL_QP_FROM_ENV_PER_CONNECTION, "ENV_HCCL_RDMA_QPS_PER_CONNECTION"},
30 0 : {MUL_QP_FROM::MUL_QP_FROM_UNKNOWN, "MUL_QP_FROM_UNKNOWN"}};
31 :
32 0 : auto it = map.find(value);
33 0 : if (it != map.end()) {
34 0 : return it->second;
35 : }
36 0 : return "UNKNOWN";
37 0 : }
38 :
39 0 : HcclResult MulQpInfoCacheBase::Init()
40 : {
41 0 : return initStatus_;
42 : }
43 :
44 0 : void MulQpInfoCacheBase::SetMulQpInfoFrom(MUL_QP_FROM mulQpInfoFrom)
45 : {
46 0 : mulQpInfoFrom_ = mulQpInfoFrom;
47 0 : }
48 :
49 0 : MUL_QP_FROM MulQpInfoCacheBase::MulQpInfoFrom() const
50 : {
51 0 : return mulQpInfoFrom_;
52 : }
53 :
54 0 : HcclResult MulQpInfoCacheBase::GetPortsNumByIpPair(PortNum &portNum, const KeyPair &ipPair) const
55 : {
56 : (void)portNum;
57 : (void)ipPair;
58 0 : return HcclResult::HCCL_E_INTERNAL;
59 : }
60 :
61 0 : HcclResult MulQpInfoCacheBase::GetSpecialSourcePortsByIpPair(MulQpSourcePorts &sourcePorts, const KeyPair &ipPair) const
62 : {
63 : (void)sourcePorts;
64 : (void)ipPair;
65 0 : return HcclResult::HCCL_E_INTERNAL;
66 : }
67 :
68 0 : DevCfgMulQpInfoCache::DevCfgMulQpInfoCache(const NICDeployment nicDeployment, const std::int32_t phyId)
69 0 : : nicDeployment_(nicDeployment),
70 0 : phyId_(phyId)
71 : {
72 0 : SetMulQpInfoFrom(MUL_QP_FROM::MUL_QP_FROM_DEV_CFG);
73 0 : }
74 :
75 0 : HcclResult MulQpInfo::Init(const InitParams ¶ms)
76 : {
77 0 : std::lock_guard<std::mutex> initLock(initLock_);
78 0 : if (GetWorkflowMode() != HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE ||
79 0 : (params.GetDevType() != DevType::DEV_TYPE_910_93 && params.GetDevType() != DevType::DEV_TYPE_910B)) {
80 0 : return HcclResult::HCCL_SUCCESS;
81 : }
82 0 : if (initStatus_ != HcclResult::HCCL_E_RESERVED) {
83 0 : return initStatus_;
84 : }
85 : // 生成队列排序
86 0 : std::queue<std::unique_ptr<MulQpInfoCacheBase>> parseOrderQueue;
87 0 : std::unique_ptr<DevCfgMulQpInfoCache> devCfgMulQpInfoCache;
88 0 : std::unique_ptr<DevNslbMulQpInfoCache> devNslbMulQpInfoCache;
89 0 : std::unique_ptr<EnvConfigPathCache> envConfigPathCache;
90 0 : std::unique_ptr<EnvPerConnectionQpInfoCache> envPerConnectionQpInfoCache;
91 0 : EXCEPTION_CATCH(
92 : (devCfgMulQpInfoCache = std::make_unique<DevCfgMulQpInfoCache>(params.GetNicDeployment(), params.GetPhyId())),
93 : return HCCL_E_INTERNAL);
94 0 : EXCEPTION_CATCH(
95 : (devNslbMulQpInfoCache = std::make_unique<DevNslbMulQpInfoCache>(params.GetNicDeployment(), params.GetPhyId())),
96 : return HCCL_E_INTERNAL);
97 0 : EXCEPTION_CATCH((envConfigPathCache = std::make_unique<EnvConfigPathCache>()), return HCCL_E_INTERNAL);
98 0 : EXCEPTION_CATCH(
99 : (envPerConnectionQpInfoCache = std::make_unique<EnvPerConnectionQpInfoCache>()), return HCCL_E_INTERNAL);
100 0 : parseOrderQueue.emplace(std::move(devCfgMulQpInfoCache));
101 0 : parseOrderQueue.emplace(std::move(devNslbMulQpInfoCache));
102 0 : parseOrderQueue.emplace(std::move(envConfigPathCache));
103 0 : parseOrderQueue.emplace(std::move(envPerConnectionQpInfoCache));
104 0 : while (!parseOrderQueue.empty()) {
105 0 : auto &&item = parseOrderQueue.front();
106 0 : initStatus_ = item->Init();
107 0 : if (initStatus_ != HcclResult::HCCL_SUCCESS) {
108 0 : return initStatus_;
109 : }
110 0 : if (item->IsAvailable()) {
111 0 : config_ = std::move(item);
112 0 : break; // 解析成功
113 : }
114 0 : parseOrderQueue.pop();
115 : }
116 0 : if (config_ && config_->IsAvailable()) {
117 0 : HCCL_RUN_INFO("[MultiQp][MulQpInfo] Init Success,Device PhyId[%d] MultiQp config from type[%s]",
118 : params.GetPhyId(), MultiQpFromToString(config_->MulQpInfoFrom()).c_str());
119 : }
120 0 : initStatus_ = HcclResult::HCCL_SUCCESS; // 均未设置 或某种方式解析成功
121 0 : return initStatus_;
122 0 : }
123 :
124 0 : bool MulQpInfo::IsInitialized()
125 : {
126 0 : std::lock_guard<std::mutex> initLock(initLock_);
127 0 : return initStatus_ != HcclResult::HCCL_E_RESERVED;
128 0 : }
129 :
130 0 : HcclResult MulQpInfo::IsEnableMulQp(bool &isEnableMulQp)
131 : {
132 0 : std::lock_guard<std::mutex> initLock(initLock_);
133 0 : isEnableMulQp = initStatus_ == HcclResult::HCCL_SUCCESS && config_ && config_->IsAvailable();
134 0 : return HcclResult::HCCL_SUCCESS;
135 0 : }
136 :
137 0 : static HcclResult DevStrToUint16(const std::string &value, std::uint16_t &count)
138 : {
139 0 : HcclResult result = HcclResult::HCCL_SUCCESS;
140 0 : unsigned long tempCount = 0;
141 : try {
142 0 : if (!value.empty())
143 : {
144 0 : std::size_t parsePos = 0;
145 0 : tempCount = std::stoul(value, &parsePos, 0);
146 0 : if (parsePos != value.size()) {
147 0 : HCCL_ERROR(
148 : "[MulQpInfo][StrToUint16]The string is not a valid unsigned integer, str[%s] pos[%llu]"
149 : " str size[%llu], val[%lu]",
150 : value.c_str(), static_cast<std::uint64_t>(parsePos), static_cast<std::uint64_t>(value.size()),
151 : tempCount);
152 0 : result = HcclResult::HCCL_E_PARA;
153 0 : } else if (tempCount <= 0xFFFFU) {
154 0 : count = static_cast<std::uint16_t>(tempCount);
155 : } else {
156 0 : HCCL_ERROR(
157 : "[MulQpInfo][StrToUint16]result of stoul is greater than 0xFFFFU, str[%s] base[%d] val[%lu]",
158 : value.c_str(), 0, tempCount);
159 0 : result = HcclResult::HCCL_E_PARA;
160 : }
161 : }
162 0 : } catch (std::invalid_argument &e) {
163 0 : HCCL_ERROR("[MulQpInfo][StrToUint16]stoul invalid arg: %s, str[%s] base[%d] val[%lu]", e.what(),
164 : value.c_str(), 0, tempCount);
165 0 : result = HcclResult::HCCL_E_PARA;
166 0 : } catch (std::out_of_range &e) {
167 0 : HCCL_ERROR("[MulQpInfo][StrToUint16]stoul out of range: %s, str[%s] base[%d] val[%lu]", e.what(),
168 : value.c_str(), 0, tempCount);
169 0 : result = HcclResult::HCCL_E_PARA;
170 0 : } catch (...) {
171 0 : HCCL_ERROR("[MulQpInfo][StrToUint16]stoul catch error, str[%s] base[%d] val[%lu]", value.c_str(), 0,
172 : tempCount);
173 0 : result = HcclResult::HCCL_E_PARA;
174 0 : }
175 0 : return result;
176 : };
177 :
178 0 : static HcclResult DevMulPorts(const std::string &value, std::vector<std::uint16_t> &ports)
179 : {
180 0 : HcclResult ret = HcclResult::HCCL_SUCCESS;
181 0 : if (!value.empty()) {
182 0 : std::uint16_t tempCount = 0;
183 0 : constexpr char separator = ',';
184 0 : std::size_t pos{0U};
185 0 : std::size_t start{0U};
186 0 : while ((pos = value.find(separator, start)) != std::string::npos) {
187 0 : if (pos == start) {
188 0 : HCCL_ERROR("[MulQpInfo][DevMulPorts]format error, value[%s]", value.c_str());
189 0 : ret = HcclResult::HCCL_E_PARA;
190 0 : break;
191 : }
192 0 : ret = DevStrToUint16(value.substr(start, pos - start), tempCount);
193 0 : if (ret != HcclResult::HCCL_SUCCESS) {
194 0 : return ret;
195 : }
196 0 : ports.emplace_back(tempCount);
197 0 : start = pos + 1U;
198 : }
199 0 : if (start >= value.size()) {
200 0 : HCCL_ERROR("[MulQpInfo][DevMulPorts]format error, value[%s]", value.c_str());
201 0 : ret = HcclResult::HCCL_E_NETWORK;
202 : }
203 0 : (void)ports.emplace_back(std::stoul(value.substr(start), nullptr, 0));
204 : }
205 0 : return ret;
206 : }
207 :
208 0 : HcclResult DevCfgMulQpInfoCache::Init()
209 : {
210 0 : std::string modeValue;
211 0 : initStatus_ =
212 0 : HrtRaGetHccnCfg(static_cast<std::uint32_t>(nicDeployment_), phyId_, HccnCfgKeyT::HCCN_UDP_PORT_MODE, modeValue);
213 0 : if (initStatus_ != HcclResult::HCCL_SUCCESS) {
214 0 : return initStatus_;
215 : }
216 0 : std::string countValue;
217 0 : initStatus_ = HrtRaGetHccnCfg(static_cast<std::uint32_t>(nicDeployment_), phyId_, HccnCfgKeyT::HCCN_MULTI_QP_COUNT,
218 : countValue);
219 0 : if (initStatus_ != HcclResult::HCCL_SUCCESS) {
220 0 : return initStatus_;
221 : }
222 0 : std::uint16_t qpCount{0U};
223 0 : initStatus_ = DevStrToUint16(countValue, qpCount);
224 0 : if (initStatus_ != HcclResult::HCCL_SUCCESS) {
225 0 : return initStatus_;
226 : }
227 0 : std::string portValue;
228 0 : std::vector<std::uint16_t> qpPorts{};
229 0 : initStatus_ = HrtRaGetHccnCfg(static_cast<std::uint32_t>(nicDeployment_), phyId_,
230 : HccnCfgKeyT::HCCN_MULTI_QP_UDP_PORTS, portValue);
231 0 : if (initStatus_ != HcclResult::HCCL_SUCCESS) {
232 0 : return initStatus_;
233 : }
234 0 : initStatus_ = DevMulPorts(portValue, qpPorts);
235 0 : if (initStatus_ != HcclResult::HCCL_SUCCESS) {
236 0 : return initStatus_;
237 : }
238 0 : const bool isNotConfig = modeValue.empty() && countValue.empty() && portValue.empty();
239 0 : const bool isNumQpConfigSuccess = !modeValue.empty() && modeValue == "multi_qp" && qpCount >= 1 &&
240 0 : qpPorts.size() == qpCount;
241 0 : constexpr std::size_t afterPortStartIndex = 1;
242 0 : if (!(isNotConfig || isNumQpConfigSuccess)) { // 对于dev multiQp cfg非正常场景
243 : // 合法 portMode 不应在此解析
244 0 : if (devCfgPortMode.end() !=
245 0 : std::find(devCfgPortMode.begin() + afterPortStartIndex, devCfgPortMode.end(), modeValue)) {
246 0 : initStatus_ = HcclResult::HCCL_SUCCESS;
247 : } else {
248 0 : HCCL_ERROR("[MulQpInfo][DevCfgMulQpInfoCache][Init]mul qp config invalid, mode[%s] count[%s] ports "
249 : "[%s]",
250 : modeValue.c_str(), countValue.c_str(), portValue.c_str());
251 0 : initStatus_ = HcclResult::HCCL_E_INTERNAL;
252 : }
253 : }
254 0 : if (initStatus_ == HcclResult::HCCL_SUCCESS && isNumQpConfigSuccess) {
255 0 : cacheInfo_ = qpPorts;
256 : }
257 0 : return initStatus_;
258 0 : }
259 :
260 0 : bool DevCfgMulQpInfoCache::IsAvailable() const
261 : {
262 0 : return initStatus_ == HcclResult::HCCL_SUCCESS && !cacheInfo_.empty();
263 : }
264 :
265 0 : HcclResult DevCfgMulQpInfoCache::GetPortsNumByIpPair(PortNum &portNum, const KeyPair &ipPair) const
266 : {
267 : (void)ipPair;
268 0 : portNum = cacheInfo_.size();
269 0 : return HcclResult::HCCL_SUCCESS;
270 : }
271 :
272 0 : HcclResult DevCfgMulQpInfoCache::GetSpecialSourcePortsByIpPair(MulQpSourcePorts &sourcePorts,
273 : const KeyPair &ipPair) const
274 : {
275 : (void)ipPair;
276 0 : sourcePorts = cacheInfo_;
277 0 : return HcclResult::HCCL_SUCCESS;
278 : }
279 :
280 0 : DevNslbMulQpInfoCache::DevNslbMulQpInfoCache(const NICDeployment nicDeployment, const std::int32_t phyId)
281 0 : : nicDeployment_(nicDeployment),
282 0 : phyId_(phyId),
283 0 : isEnableNslb_(false)
284 : {
285 0 : SetMulQpInfoFrom(MUL_QP_FROM::MUL_QP_FROM_DEV_NSLB);
286 0 : }
287 :
288 0 : HcclResult DevNslbMulQpInfoCache::Init()
289 : {
290 0 : std::string modeValue;
291 0 : initStatus_ =
292 0 : HrtRaGetHccnCfg(static_cast<std::uint32_t>(nicDeployment_), phyId_, HccnCfgKeyT::HCCN_UDP_PORT_MODE, modeValue);
293 0 : if (initStatus_ != HcclResult::HCCL_SUCCESS) {
294 0 : return initStatus_;
295 : }
296 :
297 0 : const bool isNumQpConfigSuccess = !modeValue.empty() && modeValue == "nslb_dp";
298 0 : constexpr std::size_t afterPortStartIndex = 2;
299 0 : if (initStatus_ == HcclResult::HCCL_SUCCESS && isNumQpConfigSuccess) {
300 0 : isEnableNslb_ = true;
301 0 : } else if (initStatus_ != HcclResult::HCCL_SUCCESS ||
302 0 : devCfgPortMode.end() ==
303 0 : std::find(devCfgPortMode.begin() + afterPortStartIndex, devCfgPortMode.end(), modeValue)) {
304 0 : initStatus_ = HcclResult::HCCL_E_INTERNAL;
305 0 : isEnableNslb_ = false;
306 0 : HCCL_ERROR("[MulQpInfo][DevCfgMulQpInfoCache][Init]mul qp config invalid, mode[%s]", modeValue.c_str());
307 : }
308 0 : return initStatus_;
309 0 : }
310 :
311 0 : bool DevNslbMulQpInfoCache::IsAvailable() const
312 : {
313 0 : return initStatus_ == HcclResult::HCCL_SUCCESS && isEnableNslb_;
314 : }
315 :
316 0 : HcclResult DevNslbMulQpInfoCache::GetPortsNumByIpPair(PortNum &portNum, const KeyPair &ipPair) const
317 : {
318 : (void)ipPair;
319 0 : portNum = 1;
320 0 : return HcclResult::HCCL_SUCCESS;
321 : }
322 :
323 0 : HcclResult DevNslbMulQpInfoCache::GetSpecialSourcePortsByIpPair(MulQpSourcePorts &sourcePorts,
324 : const KeyPair &ipPair) const
325 : {
326 : (void)ipPair;
327 0 : sourcePorts = MulQpSourcePorts{static_cast<Port>(hcclNslbDp::GetInstance().Getl4SPortId())};
328 0 : return HcclResult::HCCL_SUCCESS;
329 : }
330 :
331 0 : static std::vector<std::string> Split(std::string &s, const std::string &delimiter)
332 : {
333 0 : size_t posStart = 0;
334 0 : size_t posEnd = s.find(delimiter, posStart);
335 0 : std::vector<std::string> res;
336 0 : while (posEnd != std::string::npos) {
337 0 : std::string token = s.substr(posStart, posEnd - posStart);
338 0 : res.push_back(token);
339 0 : posStart = posEnd + delimiter.length();
340 0 : posEnd = s.find(delimiter, posStart);
341 0 : }
342 0 : res.push_back(s.substr(posStart));
343 0 : return res;
344 0 : }
345 :
346 0 : static HcclResult GetSrcPortsFromString(std::string &s, std::vector<std::uint16_t> &srcPorts, std::uint32_t lineCnt,
347 : const std::string &lineAvator)
348 : {
349 0 : const std::vector<std::string> strPorts = Split(s, ",");
350 0 : srcPorts.resize(strPorts.size(), 0);
351 0 : CHK_PRT_RET(strPorts.size() > MULTI_QP_CONFIG_SRC_PORT_NUM_MAX || strPorts.empty(),
352 : HCCL_ERROR("[MulQpInfo][GetSrcPortsFromString][line: %u]config ports num[%u] more than the "
353 : "threshold[%u].[%s]",
354 : lineCnt, static_cast<unsigned>(strPorts.size()), MULTI_QP_CONFIG_SRC_PORT_NUM_MAX,
355 : lineAvator.c_str()),
356 : HcclResult::HCCL_E_PARA);
357 :
358 0 : for (std::uint32_t i = 0; i < strPorts.size(); i++) {
359 : // 检查端口号是否为全数字的字符串
360 0 : CHK_PRT_RET(strPorts[i].empty() || DevStrToUint16(strPorts[i], srcPorts[i]) != HcclResult::HCCL_SUCCESS,
361 : HCCL_ERROR("[MulQpInfo][GetSrcPortsFromString][line: %u]src port[%s]"
362 : "should be within the range of[1, %u] and configured as a valid integer.[%s]",
363 : lineCnt, strPorts[i].c_str(), MULTI_QP_CONFIG_SRC_PORT_ID_MAX, lineAvator.c_str()),
364 : HcclResult::HCCL_E_PARA);
365 : }
366 0 : return HcclResult::HCCL_SUCCESS;
367 0 : }
368 :
369 0 : EnvConfigPathCache::EnvConfigPathCache()
370 : {
371 0 : SetMulQpInfoFrom(MUL_QP_FROM::MUL_QP_FROM_ENV_PORT_CONFIG_PATH);
372 0 : }
373 :
374 0 : HcclResult EnvConfigPathCache::Init()
375 : {
376 0 : if (!GetExternalInputQpSrcPortConfigPath().empty()) {
377 0 : initStatus_ = LoadMultiQpSrcPortFromFile();
378 0 : if (initStatus_ != HcclResult::HCCL_SUCCESS) {
379 0 : return initStatus_;
380 : }
381 : } else {
382 0 : initStatus_ = HcclResult::HCCL_SUCCESS;
383 : }
384 0 : return initStatus_;
385 : }
386 :
387 0 : bool EnvConfigPathCache::IsAvailable() const
388 : {
389 0 : return initStatus_ == HcclResult::HCCL_SUCCESS && !cacheInfo_.empty();
390 : }
391 :
392 0 : HcclResult EnvConfigPathCache::LoadMultiQpSrcPortFromFile()
393 : {
394 : // 读取配置文件
395 0 : std::string fileStr = GetExternalInputQpSrcPortConfigPath() + "/MultiQpSrcPort.cfg";
396 0 : std::array<char, PATH_MAX> realFile{};
397 0 : if (realpath(fileStr.c_str(), realFile.data()) == nullptr) {
398 0 : RPT_INPUT_ERR(true,
399 : "EI0001",
400 : std::vector<std::string>({"value", "env", "expect"}),
401 : std::vector<std::string>({fileStr, "config file path", "valid absolute path"}));
402 0 : HCCL_ERROR("[%s][%s]file[%s] path invalid.",
403 : LOG_KEYWORDS_INIT_GROUP.c_str(),
404 : LOG_KEYWORDS_ENV_CONFIG.c_str(),
405 : fileStr.c_str());
406 0 : return HcclResult::HCCL_E_PARA;
407 : }
408 :
409 0 : std::ifstream inFile(fileStr.c_str(), std::ifstream::in);
410 0 : if (!inFile) {
411 0 : RPT_INPUT_ERR(true, "EI0001", std::vector<std::string>({"value", "env", "expect"}),
412 : std::vector<std::string>({fileStr, "config file", "file exists and readable"}));
413 0 : HCCL_ERROR("[%s][%s]open config file[%s] failed.",
414 : LOG_KEYWORDS_INIT_GROUP.c_str(), LOG_KEYWORDS_ENV_CONFIG.c_str(),fileStr.c_str());
415 0 : return HcclResult::HCCL_E_PARA;
416 : }
417 0 : HCCL_INFO("[%s][%s]open config file[%s] success.",
418 : LOG_KEYWORDS_INIT_GROUP.c_str(), LOG_KEYWORDS_ENV_CONFIG.c_str(), fileStr.c_str());
419 :
420 : // 逐行解析配置文件
421 0 : std::uint32_t lineCnt = 1;
422 0 : std::string line;
423 0 : while (std::getline(inFile, line)) {
424 0 : std::string lineAvator = line; // 每行内容的快照, 用于dfx
425 : // 去除空格和tab
426 0 : line.erase(std::remove(line.begin(), line.end(), ' '), line.end());
427 0 : line.erase(std::remove(line.begin(), line.end(), '\t'), line.end());
428 :
429 : // 去除注释
430 0 : std::string lineInfo = Split(line, "#")[0]; // 只保留#号前的内容
431 0 : if (lineInfo.empty()) {
432 0 : HCCL_DEBUG("[EnvConfigPathCache][LoadMultiQpSrcPortFromFile][line: %u]comment line, do not parse.[%s]",
433 : lineCnt, lineAvator.c_str());
434 0 : lineCnt++;
435 0 : continue;
436 : }
437 :
438 : // 切分字符串, 检查配置格式
439 0 : std::vector<std::string> strIpPort = Split(lineInfo, "=");
440 0 : if (strIpPort.size() != MULTI_QP_CONFIG_IP_NUM) {
441 : const std::string formattedExpect =
442 0 : "[line: " + std::to_string(lineCnt) + "] Expected format: 'srcIPN,dstIPN=srcPort0,srcPort1,...,srcPortN'";
443 0 : RPT_INPUT_ERR(true,
444 : "EI0001",
445 : std::vector<std::string>({"value", "env", "expect"}),
446 : std::vector<std::string>({lineInfo, "config line format", formattedExpect}));
447 0 : HCCL_ERROR("[%s][%s] %s Config content[%s]",
448 : LOG_KEYWORDS_INIT_GROUP.c_str(),
449 : LOG_KEYWORDS_ENV_CONFIG.c_str(),
450 : formattedExpect.c_str(),
451 : lineAvator.c_str());
452 0 : inFile.close();
453 0 : return HcclResult::HCCL_E_PARA;
454 0 : }
455 :
456 : // 解析ip对
457 0 : std::string ipPair;
458 0 : auto ret = GetIpPairFromString(strIpPort[0], ipPair, lineCnt, lineAvator);
459 0 : if (ret != HcclResult::HCCL_SUCCESS) {
460 0 : RPT_INPUT_ERR(true, "EI0001", std::vector<std::string>({"value", "env", "expect"}),
461 : std::vector<std::string>({strIpPort[0], "IP pair", "valid IPv4 or IPv6 address"}));
462 0 : HCCL_ERROR("[%s][%s] %s",
463 : LOG_KEYWORDS_INIT_GROUP.c_str(), LOG_KEYWORDS_ENV_CONFIG.c_str(), "IP format error");
464 0 : inFile.close();
465 0 : return ret;
466 : }
467 :
468 : // 解析源端口号
469 0 : std::vector<std::uint16_t> srcPorts;
470 0 : ret = GetSrcPortsFromString(strIpPort[1], srcPorts, lineCnt, lineAvator);
471 0 : if (ret != HcclResult::HCCL_SUCCESS) {
472 0 : RPT_INPUT_ERR(true, "EI0001", std::vector<std::string>({"value", "env", "expect"}),
473 : std::vector<std::string>({strIpPort[1], "Source Ports", "comma-separated list of valid ports"}));
474 0 : HCCL_ERROR("[%s][%s] %s",
475 : LOG_KEYWORDS_INIT_GROUP.c_str(), LOG_KEYWORDS_ENV_CONFIG.c_str(), "port format error");
476 0 : inFile.close();
477 0 : return ret;
478 : }
479 :
480 : // 配置源端口号
481 0 : if (cacheInfo_.find(ipPair) != cacheInfo_.end()) {
482 0 : const std::string DUPLICATE_IPPAIR_ERROR = "[line: " + std::to_string(lineCnt) + "] ip pair: " + ipPair + " has existed";
483 0 : RPT_INPUT_ERR(true, "EI0001", std::vector<std::string>({"value", "env", "expect"}),
484 : std::vector<std::string>({ipPair, "IP pair Key", "unique IP pair without duplicates"}));
485 0 : HCCL_ERROR("[%s][%s][line: %u]ip pair[%s] has existed.[%s]",
486 : LOG_KEYWORDS_INIT_GROUP.c_str(),
487 : LOG_KEYWORDS_ENV_CONFIG.c_str(),
488 : lineCnt,
489 : ipPair.c_str(),
490 : lineAvator.c_str());
491 0 : inFile.close();
492 0 : return HcclResult::HCCL_E_PARA;
493 0 : }
494 0 : cacheInfo_[ipPair] = srcPorts;
495 :
496 : // 判断文件行数是否超过上限
497 0 : if (lineCnt >= MULTI_QP_CONFIG_FILE_LINE_MAX) {
498 0 : HCCL_RUN_INFO("[EnvConfigPathCache][LoadMultiQpSrcPortFromFile]config file is too large.");
499 0 : break;
500 : }
501 0 : lineCnt++;
502 0 : }
503 0 : inFile.close();
504 0 : return HcclResult::HCCL_SUCCESS;
505 0 : }
506 :
507 0 : HcclResult EnvConfigPathCache::GetIpPairFromString(std::string &s, std::string &ipPair, const std::uint32_t lineCnt,
508 : const std::string &lineAvator)
509 : {
510 0 : std::vector<std::string> strIps = Split(s, ",");
511 0 : CHK_PRT_RET(strIps.size() != MULTI_QP_CONFIG_IP_NUM,
512 : HCCL_ERROR("[EnvConfigPathCache][GetIpPairFromString][line: %u]invalid Ip format.[%s]", lineCnt,
513 : lineAvator.c_str()),
514 : HcclResult::HCCL_E_PARA);
515 :
516 0 : HcclIpAddress srcIpAddr{};
517 : // 解析源ip
518 0 : auto ret = srcIpAddr.SetReadableAddress(strIps[0]);
519 0 : CHK_PRT_RET(ret != HcclResult::HCCL_SUCCESS,
520 : HCCL_ERROR("[EnvConfigPathCache][GetIpPairFromString][line: %u]srcIp is an invalid format.[%s]",
521 : lineCnt, lineAvator.c_str()),
522 : HcclResult::HCCL_E_PARA);
523 :
524 : // 解析目的ip
525 0 : HcclIpAddress dstIpAddr{};
526 0 : ret = dstIpAddr.SetReadableAddress(strIps[1]);
527 0 : CHK_PRT_RET(ret != HcclResult::HCCL_SUCCESS,
528 : HCCL_ERROR("[EnvConfigPathCache][GetIpPairFromString][line: %u]dstIp is an invalid format.[%s]",
529 : lineCnt, lineAvator.c_str()),
530 : HcclResult::HCCL_E_PARA);
531 :
532 : // 记录ip对
533 0 : ipPair = s;
534 0 : return HcclResult::HCCL_SUCCESS;
535 0 : }
536 :
537 0 : HcclResult EnvConfigPathCache::GetPortsNumByIpPair(PortNum &portNum, const KeyPair &ipPair) const
538 : {
539 0 : const std::string srcIp = std::string(ipPair.first.GetReadableIP());
540 0 : const std::string dstIp = std::string(ipPair.second.GetReadableIP());
541 : // 匹配sip和dip
542 0 : std::string pair = srcIp + std::string(",") + dstIp;
543 0 : auto iter = cacheInfo_.find(pair);
544 0 : CHK_PRT_RET(iter != cacheInfo_.end(), portNum = iter->second.size(), HcclResult::HCCL_SUCCESS);
545 : // 匹配dip
546 0 : if (ipPair.first.GetFamily() == AF_INET) {
547 0 : pair = std::string("0.0.0.0,") + dstIp;
548 : } else {
549 0 : pair = std::string("::/128,") + dstIp;
550 : }
551 0 : iter = cacheInfo_.find(pair);
552 0 : CHK_PRT_RET(iter != cacheInfo_.end(), portNum = iter->second.size(), HcclResult::HCCL_SUCCESS);
553 : // 匹配sip
554 0 : if (ipPair.first.GetFamily() == AF_INET) {
555 0 : pair = srcIp + std::string(",0.0.0.0");
556 : } else {
557 0 : pair = srcIp + std::string(",::/128");
558 : }
559 0 : iter = cacheInfo_.find(pair);
560 0 : CHK_PRT_RET(iter != cacheInfo_.end(), portNum = iter->second.size(), HcclResult::HCCL_SUCCESS);
561 : // 通配
562 0 : if (ipPair.first.GetFamily() == AF_INET) {
563 0 : pair = std::string("0.0.0.0,0.0.0.0");
564 : } else {
565 0 : pair = std::string("::/128,::/128");
566 : }
567 0 : iter = cacheInfo_.find(pair);
568 0 : CHK_PRT_RET(iter != cacheInfo_.end(), portNum = iter->second.size(), HcclResult::HCCL_SUCCESS);
569 0 : portNum = 0;
570 0 : return HcclResult::HCCL_SUCCESS;
571 0 : }
572 :
573 0 : HcclResult EnvConfigPathCache::GetSpecialSourcePortsByIpPair(MulQpSourcePorts &sourcePorts, const KeyPair &ipPair) const
574 : {
575 0 : const std::string srcIp = std::string(ipPair.first.GetReadableIP());
576 0 : const std::string dstIp = std::string(ipPair.second.GetReadableIP());
577 : // 匹配sip和dip
578 0 : std::string pair = srcIp + std::string(",") + dstIp;
579 0 : auto iter = cacheInfo_.find(pair);
580 0 : CHK_PRT_RET(iter != cacheInfo_.end(), sourcePorts = iter->second, HcclResult::HCCL_SUCCESS);
581 : // 匹配dip
582 0 : if (ipPair.first.GetFamily() == AF_INET) {
583 0 : pair = std::string("0.0.0.0,") + dstIp;
584 : } else {
585 0 : pair = std::string("::/128,") + dstIp;
586 : }
587 0 : iter = cacheInfo_.find(pair);
588 0 : CHK_PRT_RET(iter != cacheInfo_.end(), sourcePorts = iter->second, HcclResult::HCCL_SUCCESS);
589 : // 匹配sip
590 0 : if (ipPair.first.GetFamily() == AF_INET) {
591 0 : pair = srcIp + std::string(",0.0.0.0");
592 : } else {
593 0 : pair = srcIp + std::string(",::/128");
594 : }
595 0 : iter = cacheInfo_.find(pair);
596 0 : CHK_PRT_RET(iter != cacheInfo_.end(), sourcePorts = iter->second, HcclResult::HCCL_SUCCESS);
597 : // 通配
598 0 : if (ipPair.first.GetFamily() == AF_INET) {
599 0 : pair = std::string("0.0.0.0,0.0.0.0");
600 : } else {
601 0 : pair = std::string("::/128,::/128");
602 : }
603 0 : iter = cacheInfo_.find(pair);
604 0 : CHK_PRT_RET(iter != cacheInfo_.end(), sourcePorts = iter->second, HcclResult::HCCL_SUCCESS);
605 0 : sourcePorts.clear();
606 0 : return HcclResult::HCCL_SUCCESS;
607 0 : }
608 :
609 0 : EnvPerConnectionQpInfoCache::EnvPerConnectionQpInfoCache() : cacheInfo_(0), isSetEnvPerConnectionQp_(false)
610 : {
611 0 : SetMulQpInfoFrom(MUL_QP_FROM::MUL_QP_FROM_ENV_PER_CONNECTION);
612 0 : }
613 :
614 0 : HcclResult EnvPerConnectionQpInfoCache::Init()
615 : {
616 0 : char* mmSysGetEnvValue = nullptr;
617 0 : MM_SYS_GET_ENV(MM_ENV_HCCL_RDMA_QPS_PER_CONNECTION, mmSysGetEnvValue);
618 0 : std::string rdmaQpsPerConnectionEnv = (mmSysGetEnvValue != nullptr) ? mmSysGetEnvValue : "EmptyString";
619 0 : if (rdmaQpsPerConnectionEnv == "EmptyString") {
620 0 : return HCCL_SUCCESS;
621 : }
622 0 : cacheInfo_ = GetExternalInputQpsPerConnection();
623 0 : isSetEnvPerConnectionQp_ = true;
624 0 : initStatus_ = HcclResult::HCCL_SUCCESS;
625 0 : return initStatus_;
626 0 : }
627 :
628 0 : bool EnvPerConnectionQpInfoCache::IsAvailable() const
629 : {
630 0 : return isSetEnvPerConnectionQp_ && cacheInfo_ >= 1;
631 : }
632 :
633 0 : HcclResult EnvPerConnectionQpInfoCache::GetPortsNumByIpPair(PortNum &portNum, const KeyPair &ipPair) const
634 : {
635 : (void)ipPair;
636 0 : portNum = cacheInfo_;
637 0 : return HcclResult::HCCL_SUCCESS;
638 : }
639 :
640 0 : HcclResult EnvPerConnectionQpInfoCache::GetSpecialSourcePortsByIpPair(MulQpSourcePorts &sourcePorts,
641 : const KeyPair &ipPair) const
642 : {
643 : (void)ipPair;
644 0 : sourcePorts.resize(cacheInfo_, 0);
645 0 : return HcclResult::HCCL_SUCCESS;
646 : }
647 :
648 0 : MulQpInfo::~MulQpInfo()
649 : {
650 0 : if (config_) {
651 0 : config_.reset();
652 : }
653 0 : }
654 :
655 0 : HcclResult MulQpInfo::GetMulQpFromType(MUL_QP_FROM &type)
656 : {
657 0 : std::lock_guard<std::mutex> initLock(initLock_);
658 0 : if (GetWorkflowMode() != HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE) {
659 0 : return HcclResult::HCCL_E_NOT_SUPPORT;
660 : }
661 0 : CHK_SMART_PTR_NULL(config_);
662 0 : type = config_->MulQpInfoFrom();
663 0 : return HcclResult::HCCL_SUCCESS;
664 0 : }
665 :
666 0 : HcclResult MulQpInfo::GetPortsNumByIpPair(PortNum &portNum, const KeyPair &ipPair)
667 : {
668 0 : std::lock_guard<std::mutex> initLock(initLock_);
669 0 : if (GetWorkflowMode() != HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE) {
670 0 : return HcclResult::HCCL_E_NOT_SUPPORT;
671 : }
672 0 : CHK_SMART_PTR_NULL(config_);
673 0 : return config_->GetPortsNumByIpPair(portNum, ipPair);
674 0 : }
675 :
676 0 : HcclResult MulQpInfo::GetSpecialSourcePortsByIpPair(MulQpSourcePorts &sourcePorts, const KeyPair &ipPair)
677 : {
678 0 : std::lock_guard<std::mutex> initLock(initLock_);
679 0 : if (GetWorkflowMode() != HcclWorkflowMode::HCCL_WORKFLOW_MODE_OP_BASE) {
680 0 : return HcclResult::HCCL_E_NOT_SUPPORT;
681 : }
682 0 : CHK_SMART_PTR_NULL(config_);
683 0 : return config_->GetSpecialSourcePortsByIpPair(sourcePorts, ipPair);
684 0 : }
685 : } // namespace hccl
|
