| 223 | } |
| 224 | |
| 225 | comm::RetCode SchedulerMgr::LoadBalance(const uint64_t now) { |
| 226 | comm::RetCode ret{comm::RetCode::RET_ERR_UNINIT}; |
| 227 | |
| 228 | ret = ReloadSchedConfig(); |
| 229 | if (comm::RetCode::RET_OK != ret) { |
| 230 | QLErr("ReloadSchedConfig err %d", ret); |
| 231 | |
| 232 | return ret; |
| 233 | } |
| 234 | |
| 235 | ret = ReloadTopicConfig(); |
| 236 | if (comm::RetCode::RET_OK != ret) { |
| 237 | QLErr("ReloadTopicConfig err %d", ret); |
| 238 | |
| 239 | return ret; |
| 240 | } |
| 241 | |
| 242 | comm::utils::RWLock rwlock_read(impl_->rwlock, comm::utils::RWLock::LockMode::READ); |
| 243 | |
| 244 | for (auto &&topic_id2data_kv : impl_->topic_id2data_map) { |
| 245 | const int topic_id{topic_id2data_kv.first}; |
| 246 | QLInfo("topic_id %d", topic_id); |
| 247 | |
| 248 | comm::SchedulerLoadBalanceBP::GetThreadInstance()->OnLoadBalance(topic_id); |
| 249 | |
| 250 | auto &topic_data(topic_id2data_kv.second); |
| 251 | auto topic_id2lock_it(impl_->topic_id2lock_map.find(topic_id)); |
| 252 | if (impl_->topic_id2lock_map.end() == topic_id2lock_it) { |
| 253 | QLErr("topic %d lock lost", topic_id); |
| 254 | |
| 255 | continue; |
| 256 | } |
| 257 | auto &topic_lock(topic_id2lock_it->second); |
| 258 | |
| 259 | // update topic data |
| 260 | bool consumer_conf_modified{false}; |
| 261 | comm::RetCode topic_ret{ReloadConsumerConfig(topic_id, topic_data, topic_lock, |
| 262 | now, consumer_conf_modified)}; |
| 263 | if (comm::RetCode::RET_OK != topic_ret) { |
| 264 | QLErr("topic %d ReloadConsumerConfig %d", topic_id, topic_ret); |
| 265 | comm::SchedulerLoadBalanceBP::GetThreadInstance()->OnReloadConsumerConfigFail(topic_id); |
| 266 | |
| 267 | continue; |
| 268 | } |
| 269 | comm::SchedulerLoadBalanceBP::GetThreadInstance()->OnReloadConsumerConfigSucc(topic_id, consumer_conf_modified); |
| 270 | |
| 271 | Mode mode{GetMode(topic_id, topic_data, now)}; |
| 272 | if (Mode::Dynamic == mode) { |
| 273 | comm::SchedulerLoadBalanceBP::GetThreadInstance()->OnDynamicMode(topic_id); |
| 274 | |
| 275 | // 1. live modified |
| 276 | bool live_modified{false}; |
| 277 | topic_ret = UpdateLive(topic_id, topic_data, topic_lock, now, live_modified); |
| 278 | if (comm::RetCode::RET_OK != topic_ret) { |
| 279 | QLErr("topic %d UpdateLive err %d", topic_id, topic_ret); |
| 280 | comm::SchedulerLoadBalanceBP::GetThreadInstance()->OnUpdateLiveFail(topic_id); |
| 281 | |
| 282 | continue; |
no test coverage detected