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