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