admission-controller.h

Go to the documentation of this file.

// Copyright 2014 Cloudera Inc.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.


#ifndef SCHEDULING_ADMISSION_CONTROLLER_H
#define SCHEDULING_ADMISSION_CONTROLLER_H

#include <vector>
#include <string>
#include <list>

#include <boost/unordered_map.hpp>
#include <boost/unordered_set.hpp>
#include <boost/thread/mutex.hpp>

#include "common/status.h"
#include "scheduling/request-pool-service.h"
#include "statestore/statestore-subscriber.h"
#include "statestore/query-schedule.h"
#include "util/internal-queue.h"
#include "util/thread.h"

namespace impala {

class QuerySchedule;
class ExecEnv;

//
//
class AdmissionController {
 public:
  AdmissionController(RequestPoolService* request_pool_service, MetricGroup* metrics,
      const std::string& backend_id);
  ~AdmissionController();

  Status AdmitQuery(QuerySchedule* schedule);

  Status ReleaseQuery(QuerySchedule* schedule);

  Status Init(StatestoreSubscriber* subscriber);

 private:
  static const std::string IMPALA_REQUEST_QUEUE_TOPIC;

  struct QueueNode : public InternalQueue<QueueNode>::Node {
    QueueNode(const QuerySchedule& query_schedule) : schedule(query_schedule) { }

    Promise<bool> is_admitted;

    const QuerySchedule& schedule;
  };

  struct PoolMetrics {
    IntCounter* local_admitted;
    IntCounter* local_queued;
    IntCounter* local_dequeued;
    IntCounter* local_rejected;
    IntCounter* local_timed_out;
    IntCounter* local_completed;
    IntCounter* local_time_in_queue_ms;

    IntGauge* cluster_num_running;
    IntGauge* cluster_in_queue;
    IntGauge* cluster_mem_usage;
    IntGauge* cluster_mem_estimate;
    IntGauge* local_num_running;
    IntGauge* local_in_queue;
    IntGauge* local_mem_usage;
    IntGauge* local_mem_estimate;
  };

  RequestPoolService* request_pool_service_;

  MetricGroup* metrics_;

  boost::scoped_ptr<Thread> dequeue_thread_;

  const std::string backend_id_;

  ThriftSerializer thrift_serializer_;

  boost::mutex admission_ctrl_lock_;

  typedef boost::unordered_map<std::string, TPoolStats> PoolStatsMap;

  PoolStatsMap local_pool_stats_;

  typedef boost::unordered_set<std::string> PoolSet;

  PoolSet pools_for_updates_;

  typedef boost::unordered_map<std::string, PoolStatsMap> PerBackendPoolStatsMap;
  PerBackendPoolStatsMap per_backend_pool_stats_map_;

  PoolStatsMap cluster_pool_stats_;

  typedef InternalQueue<QueueNode> RequestQueue;

  typedef boost::unordered_map<std::string, RequestQueue> RequestQueueMap;
  RequestQueueMap request_queue_map_;

  typedef boost::unordered_map<std::string, PoolMetrics> PoolMetricsMap;
  PoolMetricsMap pool_metrics_map_;

  typedef boost::unordered_map<std::string, TPoolConfigResult> PoolConfigMap;
  PoolConfigMap pool_config_cache_;

  boost::condition_variable dequeue_cv_;

  bool done_;

  void UpdatePoolStats(
      const StatestoreSubscriber::TopicDeltaMap& incoming_topic_deltas,
      std::vector<TTopicDelta>* subscriber_topic_updates);

  void HandleTopicUpdates(const std::vector<TTopicItem>& topic_updates);

  void HandleTopicDeletions(const std::vector<std::string>& topic_deletions);

  void UpdateClusterAggregates(const std::string& pool_name);

  void UpdateLocalMemUsage(const std::string& pool_name);

  void AddPoolUpdates(std::vector<TTopicDelta>* subscriber_topic_updates);

  void DequeueLoop();

  Status CanAdmitRequest(const std::string& pool, const int64_t max_requests,
      const int64_t mem_limit, const QuerySchedule& schedule, bool admit_from_queue);

  Status RejectRequest(const std::string& pool, const int64_t max_requests,
      const int64_t mem_limit, const int64_t max_queued, const QuerySchedule& schedule);

  PoolMetrics* GetPoolMetrics(const std::string& pool_name);
};

}

#endif // SCHEDULING_ADMISSION_CONTROLLER_H