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