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 "preempt_port_manager.h"
12 : #include <sstream>
13 : #include "adapter_error_manager_pub.h"
14 : #include "hccl_common.h"
15 :
16 : namespace hccl {
17 :
18 : bool PreemptPortManager::initialized = false;
19 :
20 196 : PreemptPortManager::PreemptPortManager()
21 : {
22 : // 根据host or device nic区分
23 196 : IpPortRef hostPortRef;
24 196 : preemptSockets_.emplace(NICDeployment::NIC_DEPLOYMENT_HOST, hostPortRef);
25 196 : IpPortRef devPortRef;
26 196 : preemptSockets_.emplace(NICDeployment::NIC_DEPLOYMENT_DEVICE, devPortRef);
27 196 : initialized = true;
28 196 : }
29 :
30 196 : PreemptPortManager::~PreemptPortManager()
31 : {
32 196 : preemptSockets_.clear();
33 196 : initialized = false;
34 196 : }
35 :
36 32 : PreemptPortManager& PreemptPortManager::GetInstance(s32 deviceLogicId)
37 : {
38 227 : static PreemptPortManager instance[MAX_MODULE_DEVICE_NUM];
39 32 : if (deviceLogicId == HOST_DEVICE_ID) {
40 5 : HCCL_INFO("[GetInstance] deviceLogicId[-1] is HOST_DEVICE_ID");
41 5 : return instance[0];
42 : }
43 27 : CHK_PRT_RET((static_cast<u32>(deviceLogicId) >= MAX_MODULE_DEVICE_NUM || deviceLogicId < 0),
44 : HCCL_RUN_WARNING("[PreemptPortManager][GetInstance]deviceLogicId[%d] is invalid",
45 : deviceLogicId), instance[0]);
46 :
47 26 : return instance[deviceLogicId];
48 : }
49 :
50 17 : HcclResult PreemptPortManager::ListenPreempt(const std::shared_ptr<HcclSocket> &listenSocket,
51 : const std::vector<HcclSocketPortRange> &portRange, u32 &usePort)
52 : {
53 17 : CHK_PRT_RET(!initialized,
54 : HCCL_ERROR("[PreemptPortManager][ListenPreempt] preempt port manager has already been release."),
55 : HCCL_E_INTERNAL);
56 :
57 17 : CHK_SMART_PTR_NULL(listenSocket);
58 17 : NicType socketType = listenSocket->GetSocketType();
59 17 : NICDeployment nicDeploy = socketType == NicType::HOST_NIC_TYPE ?
60 : NICDeployment::NIC_DEPLOYMENT_HOST : NICDeployment::NIC_DEPLOYMENT_DEVICE;
61 17 : std::lock_guard<std::mutex> lock(preemptMutex_);
62 17 : HcclResult ret = PreemptPortInRange(preemptSockets_[nicDeploy], listenSocket, nicDeploy, portRange, usePort);
63 17 : CHK_PRT_RET(ret != HCCL_SUCCESS,
64 : HCCL_ERROR("[PreemptPortManager][ListenPreempt] listen preempt fail, socketType[%u]", socketType), ret);
65 17 : HCCL_INFO("[PreemptPortManager][ListenPreempt] listening on port[%u] for socketType[%u] success.",
66 : usePort, socketType);
67 17 : return HCCL_SUCCESS;
68 17 : }
69 :
70 3 : HcclResult PreemptPortManager::Release(const std::shared_ptr<HcclSocket> &listenSocket)
71 : {
72 3 : CHK_PRT_RET(!initialized,
73 : HCCL_RUN_WARNING("[PreemptPortManager][Release] preempt port manager has already been release."),
74 : HCCL_SUCCESS);
75 :
76 3 : CHK_SMART_PTR_NULL(listenSocket);
77 3 : NicType socketType = listenSocket->GetSocketType();
78 3 : NICDeployment nicDeploy = socketType == NicType::HOST_NIC_TYPE ?
79 : NICDeployment::NIC_DEPLOYMENT_HOST : NICDeployment::NIC_DEPLOYMENT_DEVICE;
80 :
81 3 : std::lock_guard<std::mutex> lock(preemptMutex_);
82 3 : HcclResult ret = ReleasePreempt(preemptSockets_[nicDeploy], listenSocket, nicDeploy);
83 3 : CHK_PRT_RET(ret != HCCL_SUCCESS,
84 : HCCL_ERROR("[PreemptPortManager][Release] release fail, socketType[%u]", socketType), ret);
85 3 : HCCL_INFO("[PreemptPortManager][Release] release socket of type[%u] success.", socketType);
86 3 : return HCCL_SUCCESS;
87 3 : }
88 :
89 20 : HcclResult PreemptPortManager::PreemptPortInRange(IpPortRef& portRef, const std::shared_ptr<HcclSocket> &listenSocket,
90 : NICDeployment nicDeploy, const std::vector<HcclSocketPortRange> &portRange, u32 &usePort)
91 : {
92 20 : std::string ipAddr(listenSocket->GetLocalIp().GetReadableAddress());
93 20 : if (portRef.find(ipAddr) != portRef.end()) {
94 : // 如果在这个IP上已经有已经抢占的port，则复用这个port
95 1 : usePort = portRef[ipAddr].first;
96 1 : CHK_RET(listenSocket->Listen(usePort));
97 1 : portRef[ipAddr].second.Ref();
98 1 : HCCL_INFO("[PreemptPortManager][PreemptPortInRange] for ip[%s], port[%u] has already been listened, "
99 : "ref count[%u].", ipAddr.c_str(), usePort, portRef[ipAddr].second.Count());
100 1 : return HCCL_SUCCESS;
101 : }
102 :
103 : //如果不设置portRange，不走端口抢占逻辑
104 19 : bool isPreempt = true;
105 19 : if (!GetExternalInputHostPortSwitch()) {
106 18 : isPreempt = false;
107 : }
108 :
109 : // 如果这个IP上没有抢占过的port，则轮询输入的端口范围，找到一个可用的端口
110 20 : for (auto &range: portRange) {
111 20 : for (u32 port = range.min; port <= range.max; ++port) {
112 19 : HcclResult ret = listenSocket->Listen(port);
113 19 : if (ret == HCCL_SUCCESS) {
114 17 : usePort = listenSocket->GetLocalPort();
115 17 : if (isPreempt) {
116 : // 抢占端口成功，将端口记录到计数器中，并作为出参返回
117 1 : portRef[ipAddr].first = usePort;
118 1 : portRef[ipAddr].second.Ref();
119 : }
120 17 : HCCL_INFO("[PreemptPortManager][PreemptPortInRange] listen on ip[%s] and port[%u] success.",
121 : ipAddr.c_str(), usePort);
122 18 : return HCCL_SUCCESS;
123 : }
124 : // 非已占用的错误，直接报错退出
125 2 : CHK_PRT_RET(ret != HCCL_E_UNAVAIL,
126 : HCCL_ERROR("[PreemptPortManager][PreemptPortInRange] attempt to listen on port[%u] for ip[%s] fail."
127 : " some unexpected error occurs, errNo[0x%016llx]. attemptation is stopped.",
128 : port, ipAddr.c_str(), ret), ret);
129 : // 当前端口已被占用，尝试抢占下一个端口
130 1 : HCCL_INFO("[PreemptPortManager][PreemptPortInRange] could not listen on ip[%s], port[%u].",
131 : ipAddr.c_str(), port);
132 : }
133 : }
134 : // 所有端口范围内的端口都已经被占用，没有可用的端口，抢占监听失败
135 1 : std::string errormessage = "The IP address " + ipAddr + " and port " + std::to_string(usePort) + " have already been bound.";
136 1 : NicType socketType = listenSocket->GetSocketType();
137 1 : if (socketType == NicType::HOST_NIC_TYPE) {
138 0 : RPT_INPUT_ERR(true, "EI0019", std::vector<std::string>({"reason"}), std::vector<std::string>({errormessage}));
139 : } else {
140 7 : RPT_INPUT_ERR(true, "EI0020", std::vector<std::string>({"reason"}), std::vector<std::string>({errormessage}));
141 : }
142 1 : std::string portRangeStr = GetRangeStr(portRange);
143 1 : HCCL_ERROR("[PreemptPortManager][PreemptPortInRange] Complete polling of socket port range:%s",
144 : portRangeStr.c_str());
145 1 : HCCL_ERROR("[PreemptPortManager][PreemptPortInRange] All ports in socket port range are bound already. "
146 : "no available port to listen. Please check the ports status, or change the port range to listen on.");
147 1 : HCCL_ERROR("NOTICE: Users need to make sure ports in HCCL_HOST_SOCKET_PORT_RANGE and HCCL_NPU_SOCKET_PORT_RANGE "
148 : "are available for HCCL. Please double check whether the port are used by others unexpected process. "
149 : "The port ranges size should also be enough when running multi-process HCCL.");
150 1 : HCCL_ERROR("NOTICE: The host/npu port range size is not suggested to be smaller than the process number"
151 : " on current rank.");
152 1 : return HCCL_E_UNAVAIL;
153 21 : }
154 :
155 3 : HcclResult PreemptPortManager::ReleasePreempt(IpPortRef& portRef, const std::shared_ptr<HcclSocket> &listenSocket,
156 : NICDeployment nicDeploy)
157 : {
158 3 : std::string ipAddr(listenSocket->GetLocalIp().GetReadableAddress());
159 3 : u32 port = listenSocket->GetLocalPort();
160 3 : HCCL_INFO("[PreemptPortManager][ReleasePreempt] releasing socket, ip[%s], port[%u].", ipAddr.c_str(), port);
161 :
162 3 : bool isListening = IsAlreadyListening(portRef, ipAddr, port);
163 : // 释放的端口并非正在抢占的端口
164 3 : CHK_PRT_RET(!isListening,
165 : HCCL_RUN_WARNING("[PreemptPortManager][ReleasePreempt] "
166 : "socket ip[%u], port[%u] is not preempted or has already been released.",
167 : ipAddr.c_str(), port), HCCL_SUCCESS);
168 :
169 : // 释放的端口计数异常
170 1 : Referenced &ref = portRef[ipAddr].second;
171 1 : CHK_PRT_RET(ref.Count() <= 0,
172 : HCCL_ERROR("[PreemptPortManager][ReleasePreempt] ref[%u], ip[%s] port[%u] has already been released. "
173 : "Please do not dulplicate release.", ref.Count(), ipAddr.c_str(), port), HCCL_E_INTERNAL);
174 :
175 : // 释放绑定端口的Socket
176 1 : CHK_RET(listenSocket->DeInit());
177 1 : int count = ref.Unref();
178 1 : CHK_PRT_RET(count > 0,
179 : HCCL_INFO("[PreemptPortManager][ReleasePreempt] release a socket on ip[%s], port[%u], ref[%u].",
180 : ipAddr.c_str(), port, count), HCCL_SUCCESS);
181 :
182 : // 如果端口的计数归零，则不再抢占该端口
183 1 : portRef.erase(ipAddr);
184 1 : HCCL_INFO("[PreemptPortManager][ReleasePreempt] release preemption of socket on ip[%s], port[%u].",
185 : ipAddr.c_str(), port);
186 1 : return HCCL_SUCCESS;
187 3 : }
188 :
189 3 : bool PreemptPortManager::IsAlreadyListening(const IpPortRef& ipPortRef, const std::string &ipAddr, const u32 port)
190 : {
191 3 : auto iterPortRef = ipPortRef.find(ipAddr);
192 6 : return iterPortRef != ipPortRef.end()
193 1 : && iterPortRef->second.first == port
194 7 : && iterPortRef->second.second.Count() > 0;
195 : }
196 :
197 2 : std::string PreemptPortManager::GetRangeStr(const std::vector<HcclSocketPortRange> &portRangeVec)
198 : {
199 2 : std::ostringstream portRangeOss;
200 4 : for (auto range : portRangeVec) {
201 2 : portRangeOss << " [" << std::to_string(range.min) << ", " << std::to_string(range.max) << "]";
202 : }
203 4 : return portRangeOss.str();
204 2 : }
205 : }
|
