Statestore subscriber callback for IMPALA_REQUEST_QUEUE_TOPIC.
| 2144 | |
| 2145 | // Statestore subscriber callback for IMPALA_REQUEST_QUEUE_TOPIC. |
| 2146 | void AdmissionController::UpdatePoolStats( |
| 2147 | const StatestoreSubscriber::TopicDeltaMap& incoming_topic_deltas, |
| 2148 | vector<TTopicDelta>* subscriber_topic_updates) { |
| 2149 | { |
| 2150 | lock_guard<mutex> lock(admission_ctrl_lock_); |
| 2151 | AddPoolAndPerHostStatsUpdates(subscriber_topic_updates); |
| 2152 | |
| 2153 | StatestoreSubscriber::TopicDeltaMap::const_iterator topic = |
| 2154 | incoming_topic_deltas.find(request_queue_topic_name_); |
| 2155 | set<string> pool_stats_removed_hosts; |
| 2156 | if (topic != incoming_topic_deltas.end()) { |
| 2157 | const TTopicDelta& delta = topic->second; |
| 2158 | // Delta and non-delta updates are handled the same way, except for a full update |
| 2159 | // we first clear the backend TPoolStats. We then update the global map |
| 2160 | // and then re-compute the pool stats for any pools that changed. |
| 2161 | if (!delta.is_delta) { |
| 2162 | VLOG_ROW << "Full impala-request-queue stats update"; |
| 2163 | for (auto& entry : pool_stats_) entry.second.ClearRemoteStats(); |
| 2164 | } |
| 2165 | HandleTopicUpdates(delta.topic_entries, pool_stats_removed_hosts); |
| 2166 | } |
| 2167 | UpdateClusterAggregates(pool_stats_removed_hosts); |
| 2168 | last_topic_update_time_ms_ = MonotonicMillis(); |
| 2169 | pending_dequeue_ = true; |
| 2170 | } |
| 2171 | dequeue_cv_.NotifyOne(); // Dequeue and admit queries on the dequeue thread |
| 2172 | } |
| 2173 | |
| 2174 | void AdmissionController::PoolStats::UpdateRemoteStats( |
| 2175 | const string& host_id, TPoolStats* host_stats) { |