Method AddPoolAndPerHostStatsUpdates

be/src/scheduling/admission-controller.cc:2624–2670

}

void AdmissionController::AddPoolAndPerHostStatsUpdates(
    vector<TTopicDelta>* topic_updates) {
  // local_stats_ are updated eagerly except for backend_mem_reserved (which isn't used
  // for local admission control decisions). Update that now before sending local_stats_.
  for (auto& entry : pool_stats_) {
    entry.second.UpdateMemTrackerStats();
  }
  if (pools_for_updates_.empty()) {
    // No pool updates means no changes to host stats as well, so just return.
    return;
  }
  topic_updates->push_back(TTopicDelta());
  TTopicDelta& topic_delta = topic_updates->back();
  topic_delta.topic_name = request_queue_topic_name_;
  for (const string& pool_name: pools_for_updates_) {
    DCHECK(pool_stats_.find(pool_name) != pool_stats_.end());
    PoolStats* stats = GetPoolStats(pool_name);
    VLOG_ROW << "Sending topic update " << stats->DebugString();
    topic_delta.topic_entries.push_back(TTopicItem());
    TTopicItem& topic_item = topic_delta.topic_entries.back();
    topic_item.key = MakePoolTopicKey(pool_name, host_id_);
    Status status =
        thrift_serializer_.SerializeToString(&stats->local_stats(), &topic_item.value);
    if (!status.ok()) {
      LOG(WARNING) << "Failed to serialize query pool stats: " << status.GetDetail();
      topic_delta.topic_entries.pop_back();
    }
  }
  pools_for_updates_.clear();

  // Now add the host stats
  topic_delta.topic_entries.push_back(TTopicItem());
  TTopicItem& topic_item = topic_delta.topic_entries.back();
  topic_item.key = Substitute("$0$1", TOPIC_KEY_STAT_PREFIX, host_id_);
  TPerHostStatsUpdate update;
  for (const auto& elem : host_stats_) {
    update.per_host_stats.emplace_back();
    TPerHostStatsUpdateElement& inserted_elem = update.per_host_stats.back();
    inserted_elem.__set_host_addr(elem.first);
    inserted_elem.__set_stats(elem.second);
  }
  Status status = thrift_serializer_.SerializeToString(&update, &topic_item.value);
  if (!status.ok()) {
    LOG(WARNING) << "Failed to serialize host stats: " << status.GetDetail();
    topic_delta.topic_entries.pop_back();
  }
}

void AdmissionController::DequeueLoop() {
  unique_lock<mutex> lock(admission_ctrl_lock_);