| 116 | StoreSelectorDefault::~StoreSelectorDefault() {} |
| 117 | |
| 118 | comm::RetCode StoreSelectorDefault::GetStoreID(int &store_id) { |
| 119 | shared_ptr<const config::StoreConfig> store_config; |
| 120 | |
| 121 | comm::RetCode ret; |
| 122 | if (comm::RetCode::RET_OK != (ret = config::GlobalConfig::GetThreadInstance()->GetStoreConfig(impl_->topic_id, store_config))) { |
| 123 | QLErr("GetStoreConfig ret %d", as_integer(ret)); |
| 124 | return ret; |
| 125 | } |
| 126 | |
| 127 | vector<shared_ptr<const config::proto::Store> > stores; |
| 128 | if (comm::RetCode::RET_OK != (ret = store_config->GetAllStore(stores))) { |
| 129 | QLErr("GetAllStore ret %d", as_integer(ret)); |
| 130 | return ret; |
| 131 | } |
| 132 | |
| 133 | set<int> store_ids; |
| 134 | for (auto &&store : stores) { |
| 135 | set<int> pub_ids; |
| 136 | if (comm::RetCode::RET_OK != (ret = config::utils::GetPubIDsByStoreID(impl_->topic_id, store->store_id(), pub_ids))) { |
| 137 | QLErr("GetPubIDsByStoreID ret %d topic_id %d store_id %d", as_integer(ret), impl_->topic_id, store->store_id()); |
| 138 | continue; |
| 139 | } |
| 140 | if (pub_ids.end() != pub_ids.find(impl_->pub_id)) { |
| 141 | store_ids.insert(store->store_id()); |
| 142 | } |
| 143 | } |
| 144 | |
| 145 | if (store_ids.empty()) { |
| 146 | QLErr("store_ids size 0"); |
| 147 | return comm::RetCode::RET_ERR_RANGE_STORE; |
| 148 | } |
| 149 | |
| 150 | auto store_ids_size = store_ids.size(); |
| 151 | |
| 152 | // rebuild consisten hash |
| 153 | static __thread map<int, uint64_t> topic_id2last_rebuild_time; |
| 154 | static __thread map<int, map<uint64_t, int>> topic_id2hash_ring; |
| 155 | static __thread map<int, map<int, set<uint64_t>>> topic_id2store_id2hash_list; |
| 156 | |
| 157 | auto &hash_ring = topic_id2hash_ring[impl_->topic_id]; |
| 158 | auto &store_id2hash_list = topic_id2store_id2hash_list[impl_->topic_id]; |
| 159 | do { |
| 160 | auto &last_rebuild_time = topic_id2last_rebuild_time[impl_->topic_id]; |
| 161 | |
| 162 | auto last_mod_time = store_config->GetLastModTime(); |
| 163 | if (last_mod_time == last_rebuild_time) break; |
| 164 | last_rebuild_time = last_mod_time; |
| 165 | |
| 166 | hash_ring.clear(); |
| 167 | store_id2hash_list.clear(); |
| 168 | |
| 169 | for (auto &&tmp_store_id : store_ids) { |
| 170 | shared_ptr<const config::proto::Store> store; |
| 171 | if (comm::RetCode::RET_OK != (ret = store_config->GetStoreByStoreID(tmp_store_id, store))) { |
| 172 | QLErr("GetStoreByStoreID ret %d topic_id %d store_id %d", as_integer(ret), impl_->topic_id, tmp_store_id); |
| 173 | continue; |
| 174 | } |
| 175 | |