| 107 | } |
| 108 | |
| 109 | comm::RetCode SchedulerMgr::GetAddrScale(const comm::proto::GetAddrScaleRequest &req, |
| 110 | comm::proto::GetAddrScaleResponse &resp) { |
| 111 | comm::RetCode ret; |
| 112 | comm::SchedulerMgrBP::GetThreadInstance()->OnGetAddrScale(req); |
| 113 | |
| 114 | // TODO: |
| 115 | //OssAttrInc(oss_attr_id(), 11, 1); |
| 116 | uint64_t now{comm::utils::Time::GetSteadyClockMS()}; |
| 117 | if (!IsMaster(now)) { |
| 118 | comm::SchedulerMgrBP::GetThreadInstance()->OnIMNotMaster(req); |
| 119 | return comm::RetCode::RET_ERR_NOT_MASTER; |
| 120 | } |
| 121 | comm::SchedulerMgrBP::GetThreadInstance()->OnIMMaster(req); |
| 122 | |
| 123 | // prepare |
| 124 | int topic_id{req.topic_id()}; |
| 125 | comm::utils::RWLock rwlock_read(impl_->rwlock, comm::utils::RWLock::LockMode::READ); |
| 126 | auto topic_id2data_it(impl_->topic_id2data_map.find(topic_id)); |
| 127 | if (impl_->topic_id2data_map.end() == topic_id2data_it) { |
| 128 | QLErr("topic %d not found", topic_id); |
| 129 | // TODO: |
| 130 | //OssAttrInc(oss_attr_id(), 14u, 1u); |
| 131 | |
| 132 | return comm::RetCode::RET_ERR_RANGE_TOPIC; |
| 133 | } |
| 134 | |
| 135 | shared_ptr<const config::TopicConfig> topic_config; |
| 136 | if (comm::RetCode::RET_OK != (ret = config::GlobalConfig::GetThreadInstance()->GetTopicConfigByTopicID(topic_id, topic_config))) { |
| 137 | QLErr("GetTopicConfigByTopicID ret %d", as_integer(ret)); |
| 138 | |
| 139 | return ret; |
| 140 | } |
| 141 | |
| 142 | auto load_use_proc_used_ratio = topic_config->GetProto().topic().scheduler_load_use_proc_used_ratio(); |
| 143 | |
| 144 | auto &topic_data(topic_id2data_it->second); |
| 145 | auto topic_id2lock_it(impl_->topic_id2lock_map.find(topic_id)); |
| 146 | if (impl_->topic_id2lock_map.end() == topic_id2lock_it) { |
| 147 | QLErr("topic %d lock lost", topic_id); |
| 148 | |
| 149 | return comm::RetCode::RET_ERR_LOGIC; |
| 150 | } |
| 151 | auto &topic_lock(topic_id2lock_it->second); |
| 152 | |
| 153 | uint64_t addr{comm::utils::EncodeAddr(req.addr())}; |
| 154 | comm::utils::RWLock topic_rwlock_read(topic_lock.rwlock, comm::utils::RWLock::LockMode::READ); |
| 155 | |
| 156 | TraceMap(topic_id, topic_data); |
| 157 | auto it(topic_data.consumer_addr2info_map.find(addr)); |
| 158 | if (topic_data.consumer_addr2info_map.end() == it) { |
| 159 | QLErr("topic %d consumer %s not found", topic_id, comm::utils::AddrToString(req.addr()).c_str()); |
| 160 | comm::SchedulerMgrBP::GetThreadInstance()->OnConsumerNotFound(req); |
| 161 | |
| 162 | return comm::RetCode::RET_ERR_RANGE_CONSUMER; |
| 163 | } |
| 164 | |
| 165 | // found |
| 166 | if (impl_->scheduler->NeedSkipUpdateLoad(req)) { |
nothing calls this directly
no test coverage detected