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