已合并
修复共享锁问题 #213
wenjinhust创建于 6月22日
修复共享锁问题 #213
已合并
共 1 个文件变更+114-175
| @@ -9,28 +9,24 @@ | |||
| 9 | * MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE. | 9 | * MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE. |
| 10 | * See the Mulan PSL v2 for more details. | 10 | * See the Mulan PSL v2 for more details. |
| 11 | */ | 11 | */ |
| 12 | - | ||
| 13 | - | ||
| 14 | - | ||
| 15 | - | ||
| 16 | 12 | ||
| 17 | 13 | ||
| 18 | -static constexpr size_t INS_NUM_MAX = 4096; // 集群实例上限4096个 | 14 | +#include <algorithm> |
| 15 | + | ||
| 16 | + | ||
| 17 | + | ||
| 18 | + | ||
| 19 | + | ||
| 20 | +static constexpr size_t INS_NUM_MAX = 4096; // 集群实例上限4096个 | ||
| 19 | namespace MINDIE::MS { | 21 | namespace MINDIE::MS { |
| 20 | 22 | ||
| 21 | -bool ClusterNodes::AddInstance( | 23 | +bool ClusterNodes::AddInstance(uint64_t id, const std::string& ip, const std::string& port, |
| 22 | - uint64_t id, | 24 | + MINDIE::MS::DIGSInstanceRole role, const std::string& modelName) { |
| 23 | - const std::string &ip, | ||
| 24 | - const std::string &port, | ||
| 25 | - MINDIE::MS::DIGSInstanceRole role, | ||
| 26 | - const std::string &modelName) | ||
| 27 | -{ | ||
| 28 | std::unique_lock<std::shared_mutex> lock(mtx); | 25 | std::unique_lock<std::shared_mutex> lock(mtx); |
| 29 | auto it = instanceInfos.find(id); | 26 | auto it = instanceInfos.find(id); |
| 30 | if (it != instanceInfos.end()) { | 27 | if (it != instanceInfos.end()) { |
| 31 | LOG_E("[%s] [ClusterNodes] Add instance failed. Duplicate instance id %lu.", | 28 | LOG_E("[%s] [ClusterNodes] Add instance failed. Duplicate instance id %lu.", |
| 32 | - GetErrorCode(ErrorType::OPERATION_REPEAT, CoordinatorFeature::CLUSTER_NODES).c_str(), | 29 | + GetErrorCode(ErrorType::OPERATION_REPEAT, CoordinatorFeature::CLUSTER_NODES).c_str(), id); |
| 33 | - id); | ||
| 34 | return false; | 30 | return false; |
| 35 | } | 31 | } |
| 36 | try { | 32 | try { |
| @@ -48,18 +44,17 @@ bool ClusterNodes::AddInstance( | |||
| 48 | return true; | 44 | return true; |
| 49 | } catch (const std::exception& e) { | 45 | } catch (const std::exception& e) { |
| 50 | LOG_E("[%s] [ClusterNodes] Add instance failed, error is %s.", | 46 | LOG_E("[%s] [ClusterNodes] Add instance failed, error is %s.", |
| 51 | - GetErrorCode(ErrorType::EXCEPTION, CoordinatorFeature::CLUSTER_NODES).c_str(), e.what()); | 47 | + GetErrorCode(ErrorType::EXCEPTION, CoordinatorFeature::CLUSTER_NODES).c_str(), e.what()); |
| 52 | return false; | 48 | return false; |
| 53 | } | 49 | } |
| 54 | } | 50 | } |
| 55 | 51 | ||
| 56 | -void ClusterNodes::RemoveInstance(uint64_t id) | 52 | +void ClusterNodes::RemoveInstance(uint64_t id) { |
| 57 | -{ | ||
| 58 | auto ip = GetIp(id); | 53 | auto ip = GetIp(id); |
| 59 | auto port = GetPort(id); | 54 | auto port = GetPort(id); |
| 60 | auto role = GetRole(id); | 55 | auto role = GetRole(id); |
| 61 | LOG_D("[ClusterNodes] RemoveInstance %lu info - IP: %s, Port: %s, Role: %d", id, ip.c_str(), port.c_str(), | 56 | LOG_D("[ClusterNodes] RemoveInstance %lu info - IP: %s, Port: %s, Role: %d", id, ip.c_str(), port.c_str(), |
| 62 | - static_cast<int>(role)); | 57 | + static_cast<int>(role)); |
| 63 | 58 | ||
| 64 | std::unique_lock<std::shared_mutex> lock(mtx); | 59 | std::unique_lock<std::shared_mutex> lock(mtx); |
| 65 | std::list<uint64_t>::const_iterator iter = std::find(ids.begin(), ids.end(), id); | 60 | std::list<uint64_t>::const_iterator iter = std::find(ids.begin(), ids.end(), id); |
| @@ -81,8 +76,7 @@ void ClusterNodes::RemoveInstance(uint64_t id) | |||
| 81 | } | 76 | } |
| 82 | } | 77 | } |
| 83 | 78 | ||
| 84 | -int64_t ClusterNodes::GetTask(uint64_t id) | 79 | +int64_t ClusterNodes::GetTask(uint64_t id) { |
| 85 | -{ | ||
| 86 | std::shared_lock<std::shared_mutex> lock(mtx); | 80 | std::shared_lock<std::shared_mutex> lock(mtx); |
| 87 | auto iter = instanceInfos.find(id); | 81 | auto iter = instanceInfos.find(id); |
| 88 | if (iter == instanceInfos.end()) { | 82 | if (iter == instanceInfos.end()) { |
| @@ -92,8 +86,7 @@ int64_t ClusterNodes::GetTask(uint64_t id) | |||
| 92 | } | 86 | } |
| 93 | } | 87 | } |
| 94 | 88 | ||
| 95 | -bool ClusterNodes::IsFaultyNode(uint64_t id) | 89 | +bool ClusterNodes::IsFaultyNode(uint64_t id) { |
| 96 | -{ | ||
| 97 | std::shared_lock<std::shared_mutex> lock(mtx); | 90 | std::shared_lock<std::shared_mutex> lock(mtx); |
| 98 | const auto& nodeInfo = instanceInfos.find(id); | 91 | const auto& nodeInfo = instanceInfos.find(id); |
| 99 | if (nodeInfo == instanceInfos.end()) { | 92 | if (nodeInfo == instanceInfos.end()) { |
| @@ -114,20 +107,17 @@ bool ClusterNodes::IsFaultyNode(uint64_t id) | |||
| 114 | return false; | 107 | return false; |
| 115 | } | 108 | } |
| 116 | 109 | ||
| 117 | -bool ClusterNodes::HasInstance(uint64_t id) | 110 | +bool ClusterNodes::HasInstance(uint64_t id) { |
| 118 | -{ | ||
| 119 | std::shared_lock<std::shared_mutex> lock(mtx); | 111 | std::shared_lock<std::shared_mutex> lock(mtx); |
| 120 | return instanceInfos.find(id) != instanceInfos.end(); | 112 | return instanceInfos.find(id) != instanceInfos.end(); |
| 121 | } | 113 | } |
| 122 | 114 | ||
| 123 | -void ClusterNodes::AddFaultNode(uint64_t id) | 115 | +void ClusterNodes::AddFaultNode(uint64_t id) { |
| 124 | -{ | 116 | + std::unique_lock<std::shared_mutex> lock(mtx); |
| 125 | - std::shared_lock<std::shared_mutex> lock(mtx); | ||
| 126 | const auto& nodeInfo = instanceInfos.find(id); | 117 | const auto& nodeInfo = instanceInfos.find(id); |
| 127 | if (nodeInfo == instanceInfos.end()) { | 118 | if (nodeInfo == instanceInfos.end()) { |
| 128 | LOG_W("[%s] [ClusterNodes] Add fault node failed: Cannot find instance id %lu.", | 119 | LOG_W("[%s] [ClusterNodes] Add fault node failed: Cannot find instance id %lu.", |
| 129 | - GetWarnCode(ErrorType::NOT_FOUND, CoordinatorFeature::CLUSTER_NODES).c_str(), | 120 | + GetWarnCode(ErrorType::NOT_FOUND, CoordinatorFeature::CLUSTER_NODES).c_str(), id); |
| 130 | - id); | ||
| 131 | return; | 121 | return; |
| 132 | } | 122 | } |
| 133 | 123 | ||
| @@ -139,14 +129,12 @@ void ClusterNodes::AddFaultNode(uint64_t id) | |||
| 139 | LOG_I("[ClusterNodes] Add fault node: instance id %lu, virtual id %lu.", id, virtualId); | 129 | LOG_I("[ClusterNodes] Add fault node: instance id %lu, virtual id %lu.", id, virtualId); |
| 140 | } | 130 | } |
| 141 | 131 | ||
| 142 | -void ClusterNodes::RemoveFaultNode(uint64_t id) | 132 | +void ClusterNodes::RemoveFaultNode(uint64_t id) { |
| 143 | -{ | 133 | + std::unique_lock<std::shared_mutex> lock(mtx); |
 | |||
| 144 | - std::shared_lock<std::shared_mutex> lock(mtx); | ||
| 145 | const auto& nodeInfo = instanceInfos.find(id); | 134 | const auto& nodeInfo = instanceInfos.find(id); |
| 146 | if (nodeInfo == instanceInfos.end()) { | 135 | if (nodeInfo == instanceInfos.end()) { |
| 147 | LOG_W("[%s] [ClusterNodes] Remove fault node failed: Cannot find instance id %lu.", | 136 | LOG_W("[%s] [ClusterNodes] Remove fault node failed: Cannot find instance id %lu.", |
| 148 | - GetWarnCode(ErrorType::NOT_FOUND, CoordinatorFeature::CLUSTER_NODES).c_str(), | 137 | + GetWarnCode(ErrorType::NOT_FOUND, CoordinatorFeature::CLUSTER_NODES).c_str(), id); |
| 149 | - id); | ||
| 150 | return; | 138 | return; |
| 151 | } | 139 | } |
| 152 | 140 | ||
| @@ -163,14 +151,12 @@ void ClusterNodes::RemoveFaultNode(uint64_t id) | |||
| 163 | LOG_I("[ClusterNodes] Remove fault node: instance id %lu, virtual id %lu.", id, virtualId); | 151 | LOG_I("[ClusterNodes] Remove fault node: instance id %lu, virtual id %lu.", id, virtualId); |
| 164 | } | 152 | } |
| 165 | 153 | ||
| 166 | -std::unordered_set<uint64_t> ClusterNodes::GetVirtualIdToIds(uint64_t id) | 154 | +std::unordered_set<uint64_t> ClusterNodes::GetVirtualIdToIds(uint64_t id) { |
| 167 | -{ | ||
| 168 | std::shared_lock<std::shared_mutex> lock(mtx); | 155 | std::shared_lock<std::shared_mutex> lock(mtx); |
| 169 | const auto& nodeInfo = instanceInfos.find(id); | 156 | const auto& nodeInfo = instanceInfos.find(id); |
| 170 | if (nodeInfo == instanceInfos.end()) { | 157 | if (nodeInfo == instanceInfos.end()) { |
| 171 | LOG_W("[%s] [ClusterNodes] Get virtual id to ids failed: Cannot find instance id %lu.", | 158 | LOG_W("[%s] [ClusterNodes] Get virtual id to ids failed: Cannot find instance id %lu.", |
| 172 | - GetWarnCode(ErrorType::NOT_FOUND, CoordinatorFeature::CLUSTER_NODES).c_str(), | 159 | + GetWarnCode(ErrorType::NOT_FOUND, CoordinatorFeature::CLUSTER_NODES).c_str(), id); |
| 173 | - id); | ||
| 174 | return {}; | 160 | return {}; |
| 175 | } | 161 | } |
| 176 | 162 | ||
| @@ -179,15 +165,13 @@ std::unordered_set<uint64_t> ClusterNodes::GetVirtualIdToIds(uint64_t id) | |||
| 179 | return it != virtualToIdsMap.end() ? it->second : std::unordered_set<uint64_t>{}; | 165 | return it != virtualToIdsMap.end() ? it->second : std::unordered_set<uint64_t>{}; |
| 180 | } | 166 | } |
| 181 | 167 | ||
| 182 | -system_clock::time_point ClusterNodes::GetDeleteTime(uint64_t id) | 168 | +system_clock::time_point ClusterNodes::GetDeleteTime(uint64_t id) { |
| 183 | -{ | ||
| 184 | std::shared_lock<std::shared_mutex> lock(mtx); | 169 | std::shared_lock<std::shared_mutex> lock(mtx); |
| 185 | const auto& nodeInfo = instanceInfos.find(id); | 170 | const auto& nodeInfo = instanceInfos.find(id); |
| 186 | if (nodeInfo == instanceInfos.end()) { | 171 | if (nodeInfo == instanceInfos.end()) { |
| 187 | if (idToDelTimeMap.find(id) == idToDelTimeMap.end()) { | 172 | if (idToDelTimeMap.find(id) == idToDelTimeMap.end()) { |
| 188 | LOG_W("[%s] [ClusterNodes] Get delete time failed: Cannot find instance id %lu.", | 173 | LOG_W("[%s] [ClusterNodes] Get delete time failed: Cannot find instance id %lu.", |
| 189 | - GetWarnCode(ErrorType::NOT_FOUND, CoordinatorFeature::CLUSTER_NODES).c_str(), | 174 | + GetWarnCode(ErrorType::NOT_FOUND, CoordinatorFeature::CLUSTER_NODES).c_str(), id); |
| 190 | - id); | ||
| 191 | return system_clock::now(); | 175 | return system_clock::now(); |
| 192 | } else { | 176 | } else { |
| 193 | return idToDelTimeMap[id]; | 177 | return idToDelTimeMap[id]; |
| @@ -199,15 +183,13 @@ system_clock::time_point ClusterNodes::GetDeleteTime(uint64_t id) | |||
| 199 | return it != virtualIdToDelTimeMap.end() ? it->second : system_clock::now(); | 183 | return it != virtualIdToDelTimeMap.end() ? it->second : system_clock::now(); |
| 200 | } | 184 | } |
| 201 | 185 | ||
| 202 | -void ClusterNodes::UpdateExtraInfo(uint64_t id, std::pair<const std::string&, const std::string &> httpParam, | 186 | +void ClusterNodes::UpdateExtraInfo(uint64_t id, std::pair<const std::string&, const std::string&> httpParam, |
| 203 | - size_t totalBlockNum, size_t totalSlotsNum, uint64_t virtualId) | 187 | + size_t totalBlockNum, size_t totalSlotsNum, uint64_t virtualId) { |
| 204 | -{ | 188 | + std::unique_lock<std::shared_mutex> lock(mtx); |
| 205 | - std::shared_lock<std::shared_mutex> lock(mtx); | ||
| 206 | auto iter = instanceInfos.find(id); | 189 | auto iter = instanceInfos.find(id); |
| 207 | if (iter == instanceInfos.end()) { | 190 | if (iter == instanceInfos.end()) { |
| 208 | LOG_E("[%s] [ClusterNodes] Update extra information for instance failed. Cannot find instance id %lu.", | 191 | LOG_E("[%s] [ClusterNodes] Update extra information for instance failed. Cannot find instance id %lu.", |
| 209 | - GetErrorCode(ErrorType::NOT_FOUND, CoordinatorFeature::CLUSTER_NODES).c_str(), | 192 | + GetErrorCode(ErrorType::NOT_FOUND, CoordinatorFeature::CLUSTER_NODES).c_str(), id); |
| 210 | - id); | ||
| 211 | return; | 193 | return; |
| 212 | } | 194 | } |
| 213 | iter->second->metricPort = httpParam.first; | 195 | iter->second->metricPort = httpParam.first; |
| @@ -216,15 +198,14 @@ void ClusterNodes::UpdateExtraInfo(uint64_t id, std::pair<const std::string&, co | |||
| 216 | iter->second->totalSlotsNum = totalSlotsNum; | 198 | iter->second->totalSlotsNum = totalSlotsNum; |
| 217 | iter->second->virtualId = virtualId; | 199 | iter->second->virtualId = virtualId; |
| 218 | if (virtualToIdsMap.find(virtualId) == virtualToIdsMap.end()) { | 200 | if (virtualToIdsMap.find(virtualId) == virtualToIdsMap.end()) { |
| 219 | - std::unordered_set<uint64_t> initSet { id }; | 201 | + std::unordered_set<uint64_t> initSet{id}; |
| 220 | virtualToIdsMap[virtualId] = initSet; | 202 | virtualToIdsMap[virtualId] = initSet; |
| 221 | } else { | 203 | } else { |
| 222 | virtualToIdsMap[virtualId].insert(id); | 204 | virtualToIdsMap[virtualId].insert(id); |
| 223 | } | 205 | } |
| 224 | } | 206 | } |
| 225 | 207 | ||
| 226 | -std::string ClusterNodes::GetIp(uint64_t id) | 208 | +std::string ClusterNodes::GetIp(uint64_t id) { |
| 227 | -{ | ||
| 228 | std::shared_lock<std::shared_mutex> lock(mtx); | 209 | std::shared_lock<std::shared_mutex> lock(mtx); |
| 229 | auto iter = instanceInfos.find(id); | 210 | auto iter = instanceInfos.find(id); |
| 230 | if (iter == instanceInfos.end()) { | 211 | if (iter == instanceInfos.end()) { |
| @@ -234,8 +215,7 @@ std::string ClusterNodes::GetIp(uint64_t id) | |||
| 234 | } | 215 | } |
| 235 | } | 216 | } |
| 236 | 217 | ||
| 237 | -std::string ClusterNodes::GetPort(uint64_t id) | 218 | +std::string ClusterNodes::GetPort(uint64_t id) { |
| 238 | -{ | ||
| 239 | std::shared_lock<std::shared_mutex> lock(mtx); | 219 | std::shared_lock<std::shared_mutex> lock(mtx); |
| 240 | auto iter = instanceInfos.find(id); | 220 | auto iter = instanceInfos.find(id); |
| 241 | if (iter == instanceInfos.end()) { | 221 | if (iter == instanceInfos.end()) { |
| @@ -245,8 +225,7 @@ std::string ClusterNodes::GetPort(uint64_t id) | |||
| 245 | } | 225 | } |
| 246 | } | 226 | } |
| 247 | 227 | ||
| 248 | -std::string ClusterNodes::GetInterCommPort(uint64_t id) | 228 | +std::string ClusterNodes::GetInterCommPort(uint64_t id) { |
| 249 | -{ | ||
| 250 | std::shared_lock<std::shared_mutex> lock(mtx); | 229 | std::shared_lock<std::shared_mutex> lock(mtx); |
| 251 | auto iter = instanceInfos.find(id); | 230 | auto iter = instanceInfos.find(id); |
| 252 | if (iter == instanceInfos.end()) { | 231 | if (iter == instanceInfos.end()) { |
| @@ -256,8 +235,7 @@ std::string ClusterNodes::GetInterCommPort(uint64_t id) | |||
| 256 | } | 235 | } |
| 257 | } | 236 | } |
| 258 | 237 | ||
| 259 | -MINDIE::MS::DIGSInstanceRole ClusterNodes::GetRole(uint64_t id) | 238 | +MINDIE::MS::DIGSInstanceRole ClusterNodes::GetRole(uint64_t id) { |
| 260 | -{ | ||
| 261 | std::shared_lock<std::shared_mutex> lock(mtx); | 239 | std::shared_lock<std::shared_mutex> lock(mtx); |
| 262 | auto iter = instanceInfos.find(id); | 240 | auto iter = instanceInfos.find(id); |
| 263 | if (iter == instanceInfos.end()) { | 241 | if (iter == instanceInfos.end()) { |
| @@ -267,31 +245,29 @@ MINDIE::MS::DIGSInstanceRole ClusterNodes::GetRole(uint64_t id) | |||
| 267 | } | 245 | } |
| 268 | } | 246 | } |
| 269 | 247 | ||
| 270 | -ClusterNodes::RollType ClusterNodes::Roll(const std::vector<uint64_t> &newIds) | 248 | +ClusterNodes::RollType ClusterNodes::Roll(const std::vector<uint64_t>& newIds) { |
| 271 | -{ | ||
| 272 | std::shared_lock<std::shared_mutex> lock(mtx); | 249 | std::shared_lock<std::shared_mutex> lock(mtx); |
| 273 | std::vector<uint64_t> addVec; | 250 | std::vector<uint64_t> addVec; |
| 274 | std::vector<uint64_t> updateVec; | 251 | std::vector<uint64_t> updateVec; |
| 275 | std::vector<uint64_t> removeVec; | 252 | std::vector<uint64_t> removeVec; |
| 276 | - for (auto &id : newIds) { | 253 | + for (auto& id : newIds) { |
| 277 | std::list<uint64_t>::const_iterator iter = std::find(ids.begin(), ids.end(), id); | 254 | std::list<uint64_t>::const_iterator iter = std::find(ids.begin(), ids.end(), id); |
| 278 | - if (iter == ids.end()) { // 新增实例列表 | 255 | + if (iter == ids.end()) { // 新增实例列表 |
| 279 | addVec.emplace_back(id); | 256 | addVec.emplace_back(id); |
| 280 | - } else { // 更新实例列表 | 257 | + } else { // 更新实例列表 |
| 281 | updateVec.emplace_back(id); | 258 | updateVec.emplace_back(id); |
| 282 | } | 259 | } |
| 283 | } | 260 | } |
| 284 | - for (auto &id : std::as_const(ids)) { | 261 | + for (auto& id : std::as_const(ids)) { |
| 285 | auto iter = std::find(newIds.begin(), newIds.end(), id); | 262 | auto iter = std::find(newIds.begin(), newIds.end(), id); |
| 286 | - if (iter == newIds.end()) { // 删除实例列表 | 263 | + if (iter == newIds.end()) { // 删除实例列表 |
| 287 | removeVec.emplace_back(id); | 264 | removeVec.emplace_back(id); |
| 288 | } | 265 | } |
| 289 | } | 266 | } |
| 290 | return std::make_tuple(addVec, updateVec, removeVec); | 267 | return std::make_tuple(addVec, updateVec, removeVec); |
| 291 | } | 268 | } |
| 292 | 269 | ||
| 293 | -std::map<uint64_t, InstanceInfo> ClusterNodes::GetInstanceInfos() | 270 | +std::map<uint64_t, InstanceInfo> ClusterNodes::GetInstanceInfos() { |
| 294 | -{ | ||
| 295 | std::shared_lock<std::shared_mutex> lock(mtx); | 271 | std::shared_lock<std::shared_mutex> lock(mtx); |
| 296 | std::map<uint64_t, InstanceInfo> result; | 272 | std::map<uint64_t, InstanceInfo> result; |
| 297 | for (const auto& pair : instanceInfos) { | 273 | for (const auto& pair : instanceInfos) { |
| @@ -302,10 +278,9 @@ std::map<uint64_t, InstanceInfo> ClusterNodes::GetInstanceInfos() | |||
| 302 | return result; | 278 | return result; |
| 303 | } | 279 | } |
| 304 | 280 | ||
| 305 | -bool ClusterNodes::IsAvailable() | 281 | +bool ClusterNodes::IsAvailable() { |
| 306 | -{ | ||
| 307 | std::shared_lock<std::shared_mutex> lock(mtx); | 282 | std::shared_lock<std::shared_mutex> lock(mtx); |
| 308 | - std::string deployMode = Configure::Singleton()->schedulerConfig["deploy_mode"]; // 部署模式 | 283 | + std::string deployMode = Configure::Singleton()->schedulerConfig["deploy_mode"]; // 部署模式 |
| 309 | if (deployMode == "pd_separate" || deployMode == "pd_disaggregation" || | 284 | if (deployMode == "pd_separate" || deployMode == "pd_disaggregation" || |
| 310 | deployMode == "pd_disaggregation_single_container") { | 285 | deployMode == "pd_disaggregation_single_container") { |
| 311 | bool hasP = false; | 286 | bool hasP = false; |
| @@ -328,14 +303,13 @@ bool ClusterNodes::IsAvailable() | |||
| 328 | return isAvailable; | 303 | return isAvailable; |
| 329 | } else { | 304 | } else { |
| 330 | bool isAvailable = !instanceInfos.empty(); | 305 | bool isAvailable = !instanceInfos.empty(); |
| 331 | - LOG_D("[ClusterNodes] Non-PD mode - Instance count: %lu, Available: %s", | 306 | + LOG_D("[ClusterNodes] Non-PD mode - Instance count: %lu, Available: %s", instanceInfos.size(), |
| 332 | - instanceInfos.size(), isAvailable ? "yes" : "no"); | 307 | + isAvailable ? "yes" : "no"); |
| 333 | return isAvailable; | 308 | return isAvailable; |
| 334 | } | 309 | } |
| 335 | } | 310 | } |
| 336 | 311 | ||
| 337 | -uint64_t ClusterNodes::GetId(const std::string &ip, const std::string &port) | 312 | +uint64_t ClusterNodes::GetId(const std::string& ip, const std::string& port) { |
| 338 | -{ | ||
| 339 | std::shared_lock<std::shared_mutex> lock(mtx); | 313 | std::shared_lock<std::shared_mutex> lock(mtx); |
| 340 | for (auto& it : std::as_const(instanceInfos)) { | 314 | for (auto& it : std::as_const(instanceInfos)) { |
| 341 | if (it.second->ip == ip && it.second->port == port) { | 315 | if (it.second->ip == ip && it.second->port == port) { |
| @@ -345,8 +319,7 @@ uint64_t ClusterNodes::GetId(const std::string &ip, const std::string &port) | |||
| 345 | return UINT64_MAX; | 319 | return UINT64_MAX; |
| 346 | } | 320 | } |
| 347 | 321 | ||
| 348 | -void ClusterNodes::AddTask(uint64_t id, const std::string &reqId) | 322 | +void ClusterNodes::AddTask(uint64_t id, const std::string& reqId) { |
| 349 | -{ | ||
| 350 | std::unique_lock<std::shared_mutex> lock(mtx); | 323 | std::unique_lock<std::shared_mutex> lock(mtx); |
| 351 | auto iter = instanceInfos.find(id); | 324 | auto iter = instanceInfos.find(id); |
| 352 | if (iter == instanceInfos.end()) { | 325 | if (iter == instanceInfos.end()) { |
| @@ -356,14 +329,13 @@ void ClusterNodes::AddTask(uint64_t id, const std::string &reqId) | |||
| 356 | } | 329 | } |
| 357 | } | 330 | } |
| 358 | 331 | ||
| 359 | -void ClusterNodes::DecreaseTask(uint64_t id, const std::string &reqId) | 332 | +void ClusterNodes::DecreaseTask(uint64_t id, const std::string& reqId) { |
| 360 | -{ | ||
| 361 | std::unique_lock<std::shared_mutex> lock(mtx); | 333 | std::unique_lock<std::shared_mutex> lock(mtx); |
| 362 | auto iter = instanceInfos.find(id); | 334 | auto iter = instanceInfos.find(id); |
| 363 | if (iter == instanceInfos.end()) { | 335 | if (iter == instanceInfos.end()) { |
| 364 | return; | 336 | return; |
| 365 | } else { | 337 | } else { |
| 366 | - auto &instanceInfoTasks = iter->second->tasks; | 338 | + auto& instanceInfoTasks = iter->second->tasks; |
| 367 | auto iter1 = instanceInfoTasks.find(reqId); | 339 | auto iter1 = instanceInfoTasks.find(reqId); |
| 368 | if (iter1 != instanceInfoTasks.end()) { | 340 | if (iter1 != instanceInfoTasks.end()) { |
| 369 | instanceInfoTasks.erase(iter1); | 341 | instanceInfoTasks.erase(iter1); |
| @@ -371,8 +343,7 @@ void ClusterNodes::DecreaseTask(uint64_t id, const std::string &reqId) | |||
| 371 | } | 343 | } |
| 372 | } | 344 | } |
| 373 | 345 | ||
| 374 | -void ClusterNodes::AddRetry(uint64_t id) | 346 | +void ClusterNodes::AddRetry(uint64_t id) { |
| 375 | -{ | ||
| 376 | std::unique_lock<std::shared_mutex> lock(mtx); | 347 | std::unique_lock<std::shared_mutex> lock(mtx); |
| 377 | auto iter = instanceInfos.find(id); | 348 | auto iter = instanceInfos.find(id); |
| 378 | if (iter == instanceInfos.end()) { | 349 | if (iter == instanceInfos.end()) { |
| @@ -382,8 +353,7 @@ void ClusterNodes::AddRetry(uint64_t id) | |||
| 382 | } | 353 | } |
| 383 | } | 354 | } |
| 384 | 355 | ||
| 385 | -size_t ClusterNodes::GetRetry(uint64_t id) | 356 | +size_t ClusterNodes::GetRetry(uint64_t id) { |
| 386 | -{ | ||
| 387 | std::shared_lock<std::shared_mutex> lock(mtx); | 357 | std::shared_lock<std::shared_mutex> lock(mtx); |
| 388 | auto iter = instanceInfos.find(id); | 358 | auto iter = instanceInfos.find(id); |
| 389 | if (iter == instanceInfos.end()) { | 359 | if (iter == instanceInfos.end()) { |
| @@ -393,8 +363,7 @@ size_t ClusterNodes::GetRetry(uint64_t id) | |||
| 393 | } | 363 | } |
| 394 | } | 364 | } |
| 395 | 365 | ||
| 396 | -std::string ClusterNodes::GetModelName(uint64_t id) | 366 | +std::string ClusterNodes::GetModelName(uint64_t id) { |
| 397 | -{ | ||
| 398 | std::shared_lock<std::shared_mutex> lock(mtx); | 367 | std::shared_lock<std::shared_mutex> lock(mtx); |
| 399 | auto iter = instanceInfos.find(id); | 368 | auto iter = instanceInfos.find(id); |
| 400 | if (iter == instanceInfos.end()) { | 369 | if (iter == instanceInfos.end()) { |
| @@ -404,19 +373,18 @@ std::string ClusterNodes::GetModelName(uint64_t id) | |||
| 404 | } | 373 | } |
| 405 | } | 374 | } |
| 406 | 375 | ||
| 407 | -uint64_t ClusterNodes::GetTokenizerIns() | 376 | +uint64_t ClusterNodes::GetTokenizerIns() { |
| 408 | -{ | ||
| 409 | uint64_t taskMin = UINT64_MAX; | 377 | uint64_t taskMin = UINT64_MAX; |
| 410 | std::shared_lock<std::shared_mutex> lock(mtx); | 378 | std::shared_lock<std::shared_mutex> lock(mtx); |
| 411 | - for (auto &it : std::as_const(instanceInfos)) { | 379 | + for (auto& it : std::as_const(instanceInfos)) { |
| 412 | - auto &insInfo = it.second; | 380 | + auto& insInfo = it.second; |
| 413 | uint64_t tasksSize = insInfo->tasks.size(); | 381 | uint64_t tasksSize = insInfo->tasks.size(); |
| 414 | if (tasksSize < taskMin) { | 382 | if (tasksSize < taskMin) { |
| 415 | taskMin = tasksSize; | 383 | taskMin = tasksSize; |
| 416 | } | 384 | } |
| 417 | } | 385 | } |
| 418 | - for (auto &it : std::as_const(instanceInfos)) { | 386 | + for (auto& it : std::as_const(instanceInfos)) { |
| 419 | - auto &insInfo = it.second; | 387 | + auto& insInfo = it.second; |
| 420 | if (insInfo->tasks.size() == taskMin) { | 388 | if (insInfo->tasks.size() == taskMin) { |
| 421 | return it.first; | 389 | return it.first; |
| 422 | } | 390 | } |
| @@ -424,10 +392,9 @@ uint64_t ClusterNodes::GetTokenizerIns() | |||
| 424 | return UINT64_MAX; | 392 | return UINT64_MAX; |
| 425 | } | 393 | } |
| 426 | 394 | ||
| 427 | -bool ClusterNodes::HasModelName(const std::string &modelName) | 395 | +bool ClusterNodes::HasModelName(const std::string& modelName) { |
| 428 | -{ | ||
| 429 | std::shared_lock<std::shared_mutex> lock(mtx); | 396 | std::shared_lock<std::shared_mutex> lock(mtx); |
| 430 | - for (auto &it : std::as_const(instanceInfos)) { | 397 | + for (auto& it : std::as_const(instanceInfos)) { |
| 431 | if (GetModelName(it.first) == modelName) { | 398 | if (GetModelName(it.first) == modelName) { |
| 432 | return true; | 399 | return true; |
| 433 | } else { | 400 | } else { |
| @@ -437,8 +404,7 @@ bool ClusterNodes::HasModelName(const std::string &modelName) | |||
| 437 | return false; | 404 | return false; |
| 438 | } | 405 | } |
| 439 | 406 | ||
| 440 | -std::unordered_set<std::string> ClusterNodes::GetTasksById(uint64_t id) | 407 | +std::unordered_set<std::string> ClusterNodes::GetTasksById(uint64_t id) { |
| 441 | -{ | ||
| 442 | std::shared_lock<std::shared_mutex> lock(mtx); | 408 | std::shared_lock<std::shared_mutex> lock(mtx); |
| 443 | auto iter = instanceInfos.find(id); | 409 | auto iter = instanceInfos.find(id); |
| 444 | if (iter == instanceInfos.end()) { | 410 | if (iter == instanceInfos.end()) { |
| @@ -448,8 +414,7 @@ std::unordered_set<std::string> ClusterNodes::GetTasksById(uint64_t id) | |||
| 448 | } | 414 | } |
| 449 | } | 415 | } |
| 450 | 416 | ||
| 451 | -size_t ClusterNodes::GetTotalBlockNum(uint64_t id) | 417 | +size_t ClusterNodes::GetTotalBlockNum(uint64_t id) { |
| 452 | -{ | ||
| 453 | std::shared_lock<std::shared_mutex> lock(mtx); | 418 | std::shared_lock<std::shared_mutex> lock(mtx); |
| 454 | auto iter = instanceInfos.find(id); | 419 | auto iter = instanceInfos.find(id); |
| 455 | if (iter == instanceInfos.end()) { | 420 | if (iter == instanceInfos.end()) { |
| @@ -459,8 +424,7 @@ size_t ClusterNodes::GetTotalBlockNum(uint64_t id) | |||
| 459 | } | 424 | } |
| 460 | } | 425 | } |
| 461 | 426 | ||
| 462 | -size_t ClusterNodes::GetTotalSlotsNum(uint64_t id) | 427 | +size_t ClusterNodes::GetTotalSlotsNum(uint64_t id) { |
| 463 | -{ | ||
| 464 | std::shared_lock<std::shared_mutex> lock(mtx); | 428 | std::shared_lock<std::shared_mutex> lock(mtx); |
| 465 | auto iter = instanceInfos.find(id); | 429 | auto iter = instanceInfos.find(id); |
| 466 | if (iter == instanceInfos.end()) { | 430 | if (iter == instanceInfos.end()) { |
| @@ -470,8 +434,8 @@ size_t ClusterNodes::GetTotalSlotsNum(uint64_t id) | |||
| 470 | } | 434 | } |
| 471 | } | 435 | } |
| 472 | 436 | ||
| 473 | -int32_t ClusterNodes::RemoveRedundantInsInFlexPeers(nlohmann::json& flexInstance, const std::vector<uint64_t> dumpInfo) | 437 | +int32_t ClusterNodes::RemoveRedundantInsInFlexPeers(nlohmann::json& flexInstance, |
| 474 | -{ | 438 | + const std::vector<uint64_t> dumpInfo) { |
| 475 | try { | 439 | try { |
| 476 | auto& flexPeerVec = flexInstance.at("dynamic_info").at("peers"); | 440 | auto& flexPeerVec = flexInstance.at("dynamic_info").at("peers"); |
| 477 | for (const uint64_t& iter : dumpInfo) { | 441 | for (const uint64_t& iter : dumpInfo) { |
| @@ -483,14 +447,13 @@ int32_t ClusterNodes::RemoveRedundantInsInFlexPeers(nlohmann::json& flexInstance | |||
| 483 | return 0; | 447 | return 0; |
| 484 | } catch (const std::exception& e) { | 448 | } catch (const std::exception& e) { |
| 485 | LOG_E("[%s] [ClusterNodes] RemoveRedundantInsInFlexPeers error: %s", | 449 | LOG_E("[%s] [ClusterNodes] RemoveRedundantInsInFlexPeers error: %s", |
| 486 | - GetErrorCode(ErrorType::EXCEPTION, CoordinatorFeature::CLUSTER_NODES).c_str(), e.what()); | 450 | + GetErrorCode(ErrorType::EXCEPTION, CoordinatorFeature::CLUSTER_NODES).c_str(), e.what()); |
| 487 | return -1; | 451 | return -1; |
| 488 | } | 452 | } |
| 489 | } | 453 | } |
| 490 | 454 | ||
| 491 | -int32_t ClusterNodes::InstancePreProcInConvertMToD(nlohmann::json::iterator &iter, uint64_t flexId, | 455 | +int32_t ClusterNodes::InstancePreProcInConvertMToD(nlohmann::json::iterator& iter, uint64_t flexId, |
| 492 | - uint64_t flexGroupId, std::vector<uint64_t> &dInsIdVec) | 456 | + uint64_t flexGroupId, std::vector<uint64_t>& dInsIdVec) { |
| 493 | -{ | ||
| 494 | try { | 457 | try { |
| 495 | if (iter->at("static_info").at("group_id").template get<uint64_t>() != flexGroupId) { | 458 | if (iter->at("static_info").at("group_id").template get<uint64_t>() != flexGroupId) { |
| 496 | return 0; | 459 | return 0; |
| @@ -515,13 +478,12 @@ int32_t ClusterNodes::InstancePreProcInConvertMToD(nlohmann::json::iterator &ite | |||
| 515 | return 0; | 478 | return 0; |
| 516 | } catch (const std::exception& e) { | 479 | } catch (const std::exception& e) { |
| 517 | LOG_E("[%s] [ClusterNodes] InstancePreProcInConvertMToD error: %s", | 480 | LOG_E("[%s] [ClusterNodes] InstancePreProcInConvertMToD error: %s", |
| 518 | - GetErrorCode(ErrorType::EXCEPTION, CoordinatorFeature::CLUSTER_NODES).c_str(), e.what()); | 481 | + GetErrorCode(ErrorType::EXCEPTION, CoordinatorFeature::CLUSTER_NODES).c_str(), e.what()); |
| 519 | return -1; | 482 | return -1; |
| 520 | } | 483 | } |
| 521 | } | 484 | } |
| 522 | int32_t ClusterNodes::ConvertMInstanceToD(std::vector<uint64_t>& mNodeIds, nlohmann::json& instance, | 485 | int32_t ClusterNodes::ConvertMInstanceToD(std::vector<uint64_t>& mNodeIds, nlohmann::json& instance, |
| 523 | - nlohmann::json& instances) | 486 | + nlohmann::json& instances) { |
| 524 | -{ | ||
| 525 | try { | 487 | try { |
| 526 | uint64_t flexId = instance.at("id").template get<uint64_t>(); | 488 | uint64_t flexId = instance.at("id").template get<uint64_t>(); |
| 527 | uint64_t flexGroupId = instance.at("static_info").at("group_id").template get<uint64_t>(); | 489 | uint64_t flexGroupId = instance.at("static_info").at("group_id").template get<uint64_t>(); |
| @@ -556,14 +518,12 @@ int32_t ClusterNodes::ConvertMInstanceToD(std::vector<uint64_t>& mNodeIds, nlohm | |||
| 556 | } | 518 | } |
| 557 | } catch (const std::exception& e) { | 519 | } catch (const std::exception& e) { |
| 558 | LOG_E("[%s] [ClusterNodes] ConvertMInstanceToD error: %s", | 520 | LOG_E("[%s] [ClusterNodes] ConvertMInstanceToD error: %s", |
| 559 | - GetErrorCode(ErrorType::EXCEPTION, CoordinatorFeature::CLUSTER_NODES).c_str(), e.what()); | 521 | + GetErrorCode(ErrorType::EXCEPTION, CoordinatorFeature::CLUSTER_NODES).c_str(), e.what()); |
| 560 | return -1; | 522 | return -1; |
| 561 | } | 523 | } |
| 562 | } | 524 | } |
| 563 | 525 | ||
| 564 | -int32_t ClusterNodes::ConvertMInstanceToP(nlohmann::json& instance, | 526 | +int32_t ClusterNodes::ConvertMInstanceToP(nlohmann::json& instance, nlohmann::json& instances) { |
| 565 | - nlohmann::json& instances) | ||
| 566 | -{ | ||
| 567 | try { | 527 | try { |
| 568 | uint64_t flexId = instance.at("id").template get<uint64_t>(); | 528 | uint64_t flexId = instance.at("id").template get<uint64_t>(); |
| 569 | uint64_t flexGroupId = instance.at("static_info").at("group_id").template get<uint64_t>(); | 529 | uint64_t flexGroupId = instance.at("static_info").at("group_id").template get<uint64_t>(); |
| @@ -571,15 +531,14 @@ int32_t ClusterNodes::ConvertMInstanceToP(nlohmann::json& instance, | |||
| 571 | for (auto iter = instances.begin(); iter != instances.end(); iter++) { | 531 | for (auto iter = instances.begin(); iter != instances.end(); iter++) { |
| 572 | if (iter->at("static_info").at("group_id").template get<uint64_t>() != flexGroupId || | 532 | if (iter->at("static_info").at("group_id").template get<uint64_t>() != flexGroupId || |
| 573 | iter->at("static_info").at("role").template get<MINDIE::MS::DIGSInstanceRole>() != | 533 | iter->at("static_info").at("role").template get<MINDIE::MS::DIGSInstanceRole>() != |
| 574 | - MINDIE::MS::DIGSInstanceRole::PREFILL_INSTANCE) { | 534 | + MINDIE::MS::DIGSInstanceRole::PREFILL_INSTANCE) { |
| 575 | continue; | 535 | continue; |
| 576 | } | 536 | } |
| 577 | // 当Flex完全转化为P实例时，与Flex同组的P实例的Peers字段内需要移除原来存在的Flex的ID | 537 | // 当Flex完全转化为P实例时，与Flex同组的P实例的Peers字段内需要移除原来存在的Flex的ID |
| 578 | pInsIdVec.emplace_back(iter->at("id").template get<uint64_t>()); | 538 | pInsIdVec.emplace_back(iter->at("id").template get<uint64_t>()); |
| 579 | auto& peerVec = iter->at("dynamic_info").at("peers"); | 539 | auto& peerVec = iter->at("dynamic_info").at("peers"); |
| 580 | - auto find = std::find_if(peerVec.begin(), peerVec.end(), [flexId](const uint64_t& peer) { | 540 | + auto find = |
| 581 | - return peer == flexId; | 541 | + std::find_if(peerVec.begin(), peerVec.end(), [flexId](const uint64_t& peer) { return peer == flexId; }); |
| 582 | - }); | ||
| 583 | if (find != peerVec.end()) { | 542 | if (find != peerVec.end()) { |
| 584 | peerVec.erase(find); | 543 | peerVec.erase(find); |
| 585 | } | 544 | } |
| @@ -597,18 +556,17 @@ int32_t ClusterNodes::ConvertMInstanceToP(nlohmann::json& instance, | |||
| 597 | } | 556 | } |
| 598 | } catch (const std::exception& e) { | 557 | } catch (const std::exception& e) { |
| 599 | LOG_E("[%s] [ClusterNodes] ConvertMInstanceToP error: %s", | 558 | LOG_E("[%s] [ClusterNodes] ConvertMInstanceToP error: %s", |
| 600 | - GetErrorCode(ErrorType::EXCEPTION, CoordinatorFeature::CLUSTER_NODES).c_str(), e.what()); | 559 | + GetErrorCode(ErrorType::EXCEPTION, CoordinatorFeature::CLUSTER_NODES).c_str(), e.what()); |
| 601 | return -1; | 560 | return -1; |
| 602 | } | 561 | } |
| 603 | } | 562 | } |
| 604 | 563 | ||
| 605 | -int32_t ClusterNodes::FillInstancesInfoSplitedByFlex(nlohmann::json &instance, nlohmann::json &splitDIns, | 564 | +int32_t ClusterNodes::FillInstancesInfoSplitedByFlex(nlohmann::json& instance, nlohmann::json& splitDIns, |
| 606 | uint64_t flexId, std::vector<uint64_t> pInsIdVec, | 565 | uint64_t flexId, std::vector<uint64_t> pInsIdVec, |
| 607 | - std::vector<uint64_t> dInsIdVec) | 566 | + std::vector<uint64_t> dInsIdVec) { |
| 608 | -{ | ||
| 609 | try { | 567 | try { |
| 610 | auto pPercentage = static_cast<double>(instance.at("static_info").at("p_percentage").template get<uint64_t>()) / | 568 | auto pPercentage = static_cast<double>(instance.at("static_info").at("p_percentage").template get<uint64_t>()) / |
| 611 | - MINDIE::MS::FLEX_INSTANCE_P_PERCENTAGE_MAX; | 569 | + MINDIE::MS::FLEX_INSTANCE_P_PERCENTAGE_MAX; |
| 612 | auto flexTotalSlotsNum = instance.at("static_info").at("total_slots_num").template get<uint64_t>(); | 570 | auto flexTotalSlotsNum = instance.at("static_info").at("total_slots_num").template get<uint64_t>(); |
| 613 | auto flexTotalBlockNum = instance.at("static_info").at("total_block_num").template get<uint64_t>(); | 571 | auto flexTotalBlockNum = instance.at("static_info").at("total_block_num").template get<uint64_t>(); |
| 614 | auto flexAvaiSlotsNum = instance.at("dynamic_info").at("avail_slots_num").template get<uint64_t>(); | 572 | auto flexAvaiSlotsNum = instance.at("dynamic_info").at("avail_slots_num").template get<uint64_t>(); |
| @@ -639,14 +597,13 @@ int32_t ClusterNodes::FillInstancesInfoSplitedByFlex(nlohmann::json &instance, n | |||
| 639 | return 0; | 597 | return 0; |
| 640 | } catch (const std::exception& e) { | 598 | } catch (const std::exception& e) { |
| 641 | LOG_E("[%s] [ClusterNodes] FillInstancesInfoSplitedByFlex error: %s", | 599 | LOG_E("[%s] [ClusterNodes] FillInstancesInfoSplitedByFlex error: %s", |
| 642 | - GetErrorCode(ErrorType::EXCEPTION, CoordinatorFeature::CLUSTER_NODES).c_str(), e.what()); | 600 | + GetErrorCode(ErrorType::EXCEPTION, CoordinatorFeature::CLUSTER_NODES).c_str(), e.what()); |
| 643 | return -1; | 601 | return -1; |
| 644 | } | 602 | } |
| 645 | } | 603 | } |
| 646 | 604 | ||
| 647 | -void ClusterNodes::HandlePInsPeersInSameGroupWithFlex(nlohmann::json::iterator &iter, uint64_t flexId, | 605 | +void ClusterNodes::HandlePInsPeersInSameGroupWithFlex(nlohmann::json::iterator& iter, uint64_t flexId, |
| 648 | - std::vector<uint64_t> &pInsIdVec) | 606 | + std::vector<uint64_t>& pInsIdVec) { |
| 649 | -{ | ||
| 650 | try { | 607 | try { |
| 651 | pInsIdVec.emplace_back(iter->at("id").template get<uint64_t>()); | 608 | pInsIdVec.emplace_back(iter->at("id").template get<uint64_t>()); |
| 652 | auto& peerVec = iter->at("dynamic_info").at("peers"); | 609 | auto& peerVec = iter->at("dynamic_info").at("peers"); |
| @@ -657,14 +614,13 @@ void ClusterNodes::HandlePInsPeersInSameGroupWithFlex(nlohmann::json::iterator & | |||
| 657 | return; | 614 | return; |
| 658 | } catch (const std::exception& e) { | 615 | } catch (const std::exception& e) { |
| 659 | LOG_E("[%s] [ClusterNodes] HandlePInsPeersInSameGroupWithFlex error: %s", | 616 | LOG_E("[%s] [ClusterNodes] HandlePInsPeersInSameGroupWithFlex error: %s", |
| 660 | - GetErrorCode(ErrorType::EXCEPTION, CoordinatorFeature::CLUSTER_NODES).c_str(), e.what()); | 617 | + GetErrorCode(ErrorType::EXCEPTION, CoordinatorFeature::CLUSTER_NODES).c_str(), e.what()); |
| 661 | return; | 618 | return; |
| 662 | } | 619 | } |
| 663 | } | 620 | } |
| 664 | 621 | ||
| 665 | int32_t ClusterNodes::SplitMInstanceToPAndD(std::vector<uint64_t>& mNodeIds, nlohmann::json& instance, | 622 | int32_t ClusterNodes::SplitMInstanceToPAndD(std::vector<uint64_t>& mNodeIds, nlohmann::json& instance, |
| 666 | - nlohmann::json& instances) | 623 | + nlohmann::json& instances) { |
| 667 | -{ | ||
| 668 | try { | 624 | try { |
| 669 | uint64_t flexId = instance.at("id").template get<uint64_t>(); | 625 | uint64_t flexId = instance.at("id").template get<uint64_t>(); |
| 670 | uint64_t flexGroupId = instance.at("static_info").at("group_id").template get<uint64_t>(); | 626 | uint64_t flexGroupId = instance.at("static_info").at("group_id").template get<uint64_t>(); |
| @@ -698,17 +654,16 @@ int32_t ClusterNodes::SplitMInstanceToPAndD(std::vector<uint64_t>& mNodeIds, nlo | |||
| 698 | return 0; | 654 | return 0; |
| 699 | } catch (const std::exception& e) { | 655 | } catch (const std::exception& e) { |
| 700 | LOG_E("[%s] [ClusterNodes] SplitMInstanceToPAndD error: %s", | 656 | LOG_E("[%s] [ClusterNodes] SplitMInstanceToPAndD error: %s", |
| 701 | - GetErrorCode(ErrorType::EXCEPTION, CoordinatorFeature::CLUSTER_NODES).c_str(), e.what()); | 657 | + GetErrorCode(ErrorType::EXCEPTION, CoordinatorFeature::CLUSTER_NODES).c_str(), e.what()); |
| 702 | return -1; | 658 | return -1; |
| 703 | } | 659 | } |
| 704 | } | 660 | } |
| 705 | 661 | ||
| 706 | -int32_t ClusterNodes::ProcessFlexInstance(std::vector<uint64_t>& nodeIds, nlohmann::json& instances) | 662 | +int32_t ClusterNodes::ProcessFlexInstance(std::vector<uint64_t>& nodeIds, nlohmann::json& instances) { |
| 707 | -{ | ||
| 708 | try { | 663 | try { |
| 709 | auto it = std::find_if(instances.begin(), instances.end(), [](const nlohmann::json& ins) { | 664 | auto it = std::find_if(instances.begin(), instances.end(), [](const nlohmann::json& ins) { |
| 710 | return ins.at("static_info").at("role").template get<MINDIE::MS::DIGSInstanceRole>() == | 665 | return ins.at("static_info").at("role").template get<MINDIE::MS::DIGSInstanceRole>() == |
| 711 | - MINDIE::MS::DIGSInstanceRole::FLEX_INSTANCE; | 666 | + MINDIE::MS::DIGSInstanceRole::FLEX_INSTANCE; |
| 712 | }); | 667 | }); |
| 713 | if (it == instances.end()) { | 668 | if (it == instances.end()) { |
| 714 | return 0; | 669 | return 0; |
| @@ -734,13 +689,12 @@ int32_t ClusterNodes::ProcessFlexInstance(std::vector<uint64_t>& nodeIds, nlohma | |||
| 734 | } | 689 | } |
| 735 | } catch (const std::exception& e) { | 690 | } catch (const std::exception& e) { |
| 736 | LOG_E("[%s] [ClusterNodes] AddInstance error: %s", | 691 | LOG_E("[%s] [ClusterNodes] AddInstance error: %s", |
| 737 | - GetErrorCode(ErrorType::EXCEPTION, CoordinatorFeature::CLUSTER_NODES).c_str(), e.what()); | 692 | + GetErrorCode(ErrorType::EXCEPTION, CoordinatorFeature::CLUSTER_NODES).c_str(), e.what()); |
| 738 | return -1; | 693 | return -1; |
| 739 | } | 694 | } |
| 740 | } | 695 | } |
| 741 | 696 | ||
| 742 | -void ClusterNodes::UpdateClusterFlexInstanceInfo(uint64_t oriFlexId, uint64_t pPercentage) | 697 | +void ClusterNodes::UpdateClusterFlexInstanceInfo(uint64_t oriFlexId, uint64_t pPercentage) { |
| 743 | -{ | ||
| 744 | std::unique_lock<std::shared_mutex> lock(mtx); | 698 | std::unique_lock<std::shared_mutex> lock(mtx); |
| 745 | clusterFlexInsInfo.clusterHasFlex = true; | 699 | clusterFlexInsInfo.clusterHasFlex = true; |
| 746 | clusterFlexInsInfo.pPercentage = pPercentage; | 700 | clusterFlexInsInfo.pPercentage = pPercentage; |
| @@ -748,41 +702,35 @@ void ClusterNodes::UpdateClusterFlexInstanceInfo(uint64_t oriFlexId, uint64_t pP | |||
| 748 | clusterFlexInsInfo.splitDInsId = DECODE_INS_ID_TRANSFER_BY_FLEX; | 702 | clusterFlexInsInfo.splitDInsId = DECODE_INS_ID_TRANSFER_BY_FLEX; |
| 749 | } | 703 | } |
| 750 | 704 | ||
| 751 | -void ClusterNodes::ClearClusterFlexInstanceInfo() | 705 | +void ClusterNodes::ClearClusterFlexInstanceInfo() { |
| 752 | -{ | ||
| 753 | std::unique_lock<std::shared_mutex> lock(mtx); | 706 | std::unique_lock<std::shared_mutex> lock(mtx); |
| 754 | clusterFlexInsInfo.clusterHasFlex = false; | 707 | clusterFlexInsInfo.clusterHasFlex = false; |
| 755 | clusterFlexInsInfo.pPercentage = 0; | 708 | clusterFlexInsInfo.pPercentage = 0; |
| 756 | clusterFlexInsInfo.originFlexInsId = 0; | 709 | clusterFlexInsInfo.originFlexInsId = 0; |
| 757 | } | 710 | } |
| 758 | 711 | ||
| 759 | -bool ClusterNodes::IsClusterHasFlex() | 712 | +bool ClusterNodes::IsClusterHasFlex() { |
| 760 | -{ | ||
| 761 | std::shared_lock<std::shared_mutex> lock(mtx); | 713 | std::shared_lock<std::shared_mutex> lock(mtx); |
| 762 | return clusterFlexInsInfo.clusterHasFlex; | 714 | return clusterFlexInsInfo.clusterHasFlex; |
| 763 | } | 715 | } |
| 764 | 716 | ||
| 765 | -uint64_t ClusterNodes::GetOriFlexInsId() | 717 | +uint64_t ClusterNodes::GetOriFlexInsId() { |
| 766 | -{ | ||
| 767 | std::shared_lock<std::shared_mutex> lock(mtx); | 718 | std::shared_lock<std::shared_mutex> lock(mtx); |
| 768 | return clusterFlexInsInfo.originFlexInsId; | 719 | return clusterFlexInsInfo.originFlexInsId; |
| 769 | } | 720 | } |
| 770 | 721 | ||
| 771 | -uint64_t ClusterNodes::GetSplitedDInsId() | 722 | +uint64_t ClusterNodes::GetSplitedDInsId() { |
| 772 | -{ | ||
| 773 | std::shared_lock<std::shared_mutex> lock(mtx); | 723 | std::shared_lock<std::shared_mutex> lock(mtx); |
| 774 | return clusterFlexInsInfo.splitDInsId; | 724 | return clusterFlexInsInfo.splitDInsId; |
| 775 | } | 725 | } |
| 776 | 726 | ||
| 777 | -bool ClusterNodes::IsFlexSplitedIntoTwoInstance() | 727 | +bool ClusterNodes::IsFlexSplitedIntoTwoInstance() { |
| 778 | -{ | ||
| 779 | std::shared_lock<std::shared_mutex> lock(mtx); | 728 | std::shared_lock<std::shared_mutex> lock(mtx); |
| 780 | return clusterFlexInsInfo.clusterHasFlex && clusterFlexInsInfo.pPercentage != 0 && | 729 | return clusterFlexInsInfo.clusterHasFlex && clusterFlexInsInfo.pPercentage != 0 && |
| 781 | clusterFlexInsInfo.pPercentage != MINDIE::MS::FLEX_INSTANCE_P_PERCENTAGE_MAX; | 730 | clusterFlexInsInfo.pPercentage != MINDIE::MS::FLEX_INSTANCE_P_PERCENTAGE_MAX; |
| 782 | } | 731 | } |
| 783 | 732 | ||
| 784 | -bool ClusterNodes::IsVecContainsFlex(const std::vector<uint64_t>& vec) | 733 | +bool ClusterNodes::IsVecContainsFlex(const std::vector<uint64_t>& vec) { |
| 785 | -{ | ||
| 786 | bool isPFound = false; | 734 | bool isPFound = false; |
| 787 | bool isDFound = false; | 735 | bool isDFound = false; |
| 788 | std::shared_lock<std::shared_mutex> lock(mtx); | 736 | std::shared_lock<std::shared_mutex> lock(mtx); |
| @@ -792,7 +740,7 @@ bool ClusterNodes::IsVecContainsFlex(const std::vector<uint64_t>& vec) | |||
| 792 | if (clusterFlexInsInfo.pPercentage == MINDIE::MS::FLEX_INSTANCE_P_PERCENTAGE_MAX) { | 740 | if (clusterFlexInsInfo.pPercentage == MINDIE::MS::FLEX_INSTANCE_P_PERCENTAGE_MAX) { |
| 793 | isDFound = true; | 741 | isDFound = true; |
| 794 | } | 742 | } |
| 795 | - for (auto iter:vec) { | 743 | + for (auto iter : vec) { |
| 796 | if (iter == clusterFlexInsInfo.originFlexInsId) { | 744 | if (iter == clusterFlexInsInfo.originFlexInsId) { |
| 797 | isPFound = true; | 745 | isPFound = true; |
| 798 | } | 746 | } |
| @@ -803,43 +751,38 @@ bool ClusterNodes::IsVecContainsFlex(const std::vector<uint64_t>& vec) | |||
| 803 | return isPFound && isDFound; | 751 | return isPFound && isDFound; |
| 804 | } | 752 | } |
| 805 | 753 | ||
| 806 | -bool ClusterNodes::IsBothPAndDFromFlex(uint64_t pId, uint64_t dId) | 754 | +bool ClusterNodes::IsBothPAndDFromFlex(uint64_t pId, uint64_t dId) { |
| 807 | -{ | ||
| 808 | std::shared_lock<std::shared_mutex> lock(mtx); | 755 | std::shared_lock<std::shared_mutex> lock(mtx); |
| 809 | return pId == clusterFlexInsInfo.originFlexInsId && dId == clusterFlexInsInfo.splitDInsId; | 756 | return pId == clusterFlexInsInfo.originFlexInsId && dId == clusterFlexInsInfo.splitDInsId; |
| 810 | } | 757 | } |
| 811 | 758 | ||
| 812 | -bool ClusterNodes::IsInstanceFromFlex(uint64_t id) | 759 | +bool ClusterNodes::IsInstanceFromFlex(uint64_t id) { |
| 813 | -{ | ||
| 814 | std::shared_lock<std::shared_mutex> lock(mtx); | 760 | std::shared_lock<std::shared_mutex> lock(mtx); |
| 815 | return clusterFlexInsInfo.clusterHasFlex && | 761 | return clusterFlexInsInfo.clusterHasFlex && |
| 816 | (id == clusterFlexInsInfo.splitDInsId || id == clusterFlexInsInfo.originFlexInsId); | 762 | (id == clusterFlexInsInfo.splitDInsId || id == clusterFlexInsInfo.originFlexInsId); |
| 817 | } | 763 | } |
| 818 | 764 | ||
| 819 | -size_t ClusterNodes::GetInsNumMax(void) | 765 | +size_t ClusterNodes::GetInsNumMax(void) { |
| 820 | -{ | ||
| 821 | if (IsFlexSplitedIntoTwoInstance()) { | 766 | if (IsFlexSplitedIntoTwoInstance()) { |
| 822 | return INS_NUM_MAX + 1; | 767 | return INS_NUM_MAX + 1; |
| 823 | } | 768 | } |
| 824 | return INS_NUM_MAX; | 769 | return INS_NUM_MAX; |
| 825 | } | 770 | } |
| 826 | 771 | ||
| 827 | -void ClusterNodes::ProcSchedulerInfoUnderFlexSituation(std::vector<MINDIE::MS::DIGSInstanceScheduleInfo>& schedulerInfo) | 772 | +void ClusterNodes::ProcSchedulerInfoUnderFlexSituation( |
| 828 | -{ | 773 | + std::vector<MINDIE::MS::DIGSInstanceScheduleInfo>& schedulerInfo) { |
| 829 | if (!IsClusterHasFlex()) { | 774 | if (!IsClusterHasFlex()) { |
| 830 | return; | 775 | return; |
| 831 | } | 776 | } |
| 832 | auto oriFlexId = GetOriFlexInsId(); | 777 | auto oriFlexId = GetOriFlexInsId(); |
| 833 | auto splitedDInsId = GetSplitedDInsId(); | 778 | auto splitedDInsId = GetSplitedDInsId(); |
| 834 | if (IsFlexSplitedIntoTwoInstance()) { | 779 | if (IsFlexSplitedIntoTwoInstance()) { |
| 835 | - auto flexIter = std::find_if(schedulerInfo.begin(), schedulerInfo.end(), | 780 | + auto flexIter = std::find_if( |
| 836 | - [oriFlexId](const MINDIE::MS::DIGSInstanceScheduleInfo& info) { | 781 | + schedulerInfo.begin(), schedulerInfo.end(), |
| 837 | - return info.id == oriFlexId; | 782 | + [oriFlexId](const MINDIE::MS::DIGSInstanceScheduleInfo& info) { return info.id == oriFlexId; }); |
| 838 | - }); | 783 | + auto splitDIter = std::find_if( |
| 839 | - auto splitDIter = std::find_if(schedulerInfo.begin(), schedulerInfo.end(), | 784 | + schedulerInfo.begin(), schedulerInfo.end(), |
| 840 | - [splitedDInsId](const MINDIE::MS::DIGSInstanceScheduleInfo& info) { | 785 | + [splitedDInsId](const MINDIE::MS::DIGSInstanceScheduleInfo& info) { return info.id == splitedDInsId; }); |
| 841 | - return info.id == splitedDInsId; | ||
| 842 | - }); | ||
| 843 | if (flexIter == schedulerInfo.end()) { | 786 | if (flexIter == schedulerInfo.end()) { |
| 844 | LOG_E("[ClusterNodes] ProcSchedulerInfoUnderFlexSituation: cannot find ins with id %lu.\n", oriFlexId); | 787 | LOG_E("[ClusterNodes] ProcSchedulerInfoUnderFlexSituation: cannot find ins with id %lu.\n", oriFlexId); |
| 845 | return; | 788 | return; |
| @@ -853,10 +796,9 @@ void ClusterNodes::ProcSchedulerInfoUnderFlexSituation(std::vector<MINDIE::MS::D | |||
| 853 | schedulerInfo.erase(splitDIter); | 796 | schedulerInfo.erase(splitDIter); |
| 854 | } else { | 797 | } else { |
| 855 | // 若Flex不是拆成P+D的情况，那么这里只需要处理化作D的情况，P不用额外处理 | 798 | // 若Flex不是拆成P+D的情况，那么这里只需要处理化作D的情况，P不用额外处理 |
| 856 | - auto splitDIter = std::find_if(schedulerInfo.begin(), schedulerInfo.end(), | 799 | + auto splitDIter = std::find_if( |
| 857 | - [splitedDInsId](const MINDIE::MS::DIGSInstanceScheduleInfo& info) { | 800 | + schedulerInfo.begin(), schedulerInfo.end(), |
| 858 | - return info.id == splitedDInsId; | 801 | + [splitedDInsId](const MINDIE::MS::DIGSInstanceScheduleInfo& info) { return info.id == splitedDInsId; }); |
| 859 | - }); | ||
| 860 | if (splitDIter != schedulerInfo.end()) { | 802 | if (splitDIter != schedulerInfo.end()) { |
| 861 | splitDIter->id = oriFlexId; | 803 | splitDIter->id = oriFlexId; |
| 862 | } | 804 | } |
| @@ -864,8 +806,7 @@ void ClusterNodes::ProcSchedulerInfoUnderFlexSituation(std::vector<MINDIE::MS::D | |||
| 864 | return; | 806 | return; |
| 865 | } | 807 | } |
| 866 | 808 | ||
| 867 | -void ClusterNodes::ProcInstanceIdsUnderFlexSituation(std::vector<uint64_t>& nodeIds) | 809 | +void ClusterNodes::ProcInstanceIdsUnderFlexSituation(std::vector<uint64_t>& nodeIds) { |
| 868 | -{ | ||
| 869 | if (!IsClusterHasFlex()) { | 810 | if (!IsClusterHasFlex()) { |
| 870 | return; | 811 | return; |
| 871 | } | 812 | } |
| @@ -882,8 +823,7 @@ void ClusterNodes::ProcInstanceIdsUnderFlexSituation(std::vector<uint64_t>& node | |||
| 882 | } | 823 | } |
| 883 | } | 824 | } |
| 884 | 825 | ||
| 885 | -int64_t ClusterNodes::GetInstanceTaskNumUnderFlexSituation(uint64_t id) | 826 | +int64_t ClusterNodes::GetInstanceTaskNumUnderFlexSituation(uint64_t id) { |
| 886 | -{ | ||
| 887 | if (!IsClusterHasFlex() || id != GetOriFlexInsId()) { | 827 | if (!IsClusterHasFlex() || id != GetOriFlexInsId()) { |
| 888 | return GetTask(id); | 828 | return GetTask(id); |
| 889 | } | 829 | } |
| @@ -899,10 +839,9 @@ int64_t ClusterNodes::GetInstanceTaskNumUnderFlexSituation(uint64_t id) | |||
| 899 | } | 839 | } |
| 900 | } | 840 | } |
| 901 | 841 | ||
| 902 | -void ClusterNodes::ProcTaskQuaryDInstanceIdUnderFlexSituation(uint64_t& dId) | 842 | +void ClusterNodes::ProcTaskQuaryDInstanceIdUnderFlexSituation(uint64_t& dId) { |
| 903 | -{ | ||
| 904 | if (IsClusterHasFlex() && dId == clusterFlexInsInfo.originFlexInsId) { | 843 | if (IsClusterHasFlex() && dId == clusterFlexInsInfo.originFlexInsId) { |
| 905 | dId = DECODE_INS_ID_TRANSFER_BY_FLEX; | 844 | dId = DECODE_INS_ID_TRANSFER_BY_FLEX; |
| 906 | } | 845 | } |
| 907 | } | 846 | } |
| 908 | -} | 847 | +} // namespace MINDIE::MS |
unique_lock：独占锁。同一时刻，只有一个线程可以持有该锁并访问资源。 shared_lock：共享锁。同一时刻，多个线程可以同时持有该锁。
可能出现以前可以正常取锁，修改后变为阻塞。造成的影响需评估后再修改。