| 134 | } |
| 135 | |
| 136 | void ExecutorGroup::RemoveExecutor(const BackendDescriptorPB& be_desc) { |
| 137 | auto be_descs_it = executor_map_.find(be_desc.ip_address()); |
| 138 | if (be_descs_it == executor_map_.end()) { |
| 139 | LOG(DFATAL) << "Tried to remove a backend from non-existing host: " |
| 140 | << be_desc.krpc_address(); |
| 141 | return; |
| 142 | } |
| 143 | auto eq = [&be_desc](const BackendDescriptorPB& existing) { |
| 144 | // The IP addresses must already match, so it is sufficient to check the port. |
| 145 | DCHECK_EQ(existing.ip_address(), be_desc.ip_address()); |
| 146 | return existing.address().port() == be_desc.address().port(); |
| 147 | }; |
| 148 | |
| 149 | Executors& be_descs = be_descs_it->second; |
| 150 | auto remove_it = find_if(be_descs.begin(), be_descs.end(), eq); |
| 151 | if (remove_it == be_descs.end()) { |
| 152 | LOG(DFATAL) << "Tried to remove non-existing backend from per-host list: " |
| 153 | << be_desc.krpc_address(); |
| 154 | return; |
| 155 | } |
| 156 | |
| 157 | // Copy the data necessary to update the internal state variables since the erase call |
| 158 | // has the potential to destroy the object referenced by be_desc. |
| 159 | const std::string be_hostname = be_desc.address().hostname(); |
| 160 | const std::string be_ip_address = be_desc.ip_address(); |
| 161 | const int64_t be_admin_mem_limit = be_desc.admit_mem_limit(); |
| 162 | |
| 163 | be_descs.erase(remove_it); |
| 164 | |
| 165 | if (per_executor_admit_mem_limit_ == be_admin_mem_limit) { |
| 166 | CalculatePerExecutorMemLimitForAdmission(); |
| 167 | } |
| 168 | if (be_descs.empty()) { |
| 169 | executor_map_.erase(be_descs_it); |
| 170 | executor_ip_map_.erase(be_hostname); |
| 171 | executor_ip_hash_ring_.RemoveNode(be_ip_address); |
| 172 | } |
| 173 | } |
| 174 | |
| 175 | bool ExecutorGroup::LookUpExecutorIp(const Hostname& hostname, IpAddr* ip) const { |
| 176 | // Check if hostname is already a valid IP address. |