query-resource-mgr.h

Go to the documentation of this file.

// Copyright 2012 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 STATESTORE_QUERY_RESOURCE_MGR_H
#define STATESTORE_QUERY_RESOURCE_MGR_H

#include "common/atomic.h"
#include "common/status.h"
#include "gen-cpp/Types_types.h"
#include "gen-cpp/ResourceBrokerService.h"
#include "gen-cpp/ImpalaInternalService.h"
#include "gen-cpp/Frontend_types.h"
#include "util/promise.h"
#include "util/thread.h"

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

namespace impala {

class ResourceBroker;

class ResourceResolver {
 public:
  ResourceResolver(const boost::unordered_set<TNetworkAddress>& unique_hosts);

  void GetResourceHostport(const TNetworkAddress& src, TNetworkAddress* dst);

 private:
  boost::unordered_map<TNetworkAddress, TNetworkAddress> impalad_to_dn_;
  boost::unordered_map<TNetworkAddress, TNetworkAddress> dn_to_impalad_;

  void CreateLocalLlamaNodeMapping(
      const boost::unordered_set<TNetworkAddress>& unique_hosts);
};

//
//
//
//
class QueryResourceMgr {
 public:
  QueryResourceMgr(const TUniqueId& reservation_id,
      const TNetworkAddress& local_resource_location, const TUniqueId& query_id);

  void InitVcoreAcquisition(int32_t init_vcores);

  //
  inline bool IsVcoreOverSubscribed() {
    boost::lock_guard<boost::mutex> l(threads_running_lock_);
    return threads_running_ > vcores_ * max_vcore_oversubscription_ratio_;
  }

  void NotifyThreadUsageChange(int delta);

  typedef boost::function<void ()> VcoreAvailableCb;
  int32_t AddVcoreAvailableCb(const VcoreAvailableCb& callback);

  void RemoveVcoreAvailableCb(int32_t callback_id);

  Status CreateExpansionRequest(int64_t memory_mb, int64_t vcores,
      TResourceBrokerExpansionRequest* request);

  void Shutdown();

  ~QueryResourceMgr();

 private:
  TUniqueId reservation_id_;

  TUniqueId query_id_;

  TNetworkAddress local_resource_location_;

  boost::mutex exit_lock_;
  bool exit_;

  boost::mutex callbacks_lock_;

  typedef boost::unordered_map<int32_t, VcoreAvailableCb> CallbackMap;
  CallbackMap callbacks_;

  CallbackMap::iterator callbacks_it_;

  int32_t callback_count_;

  boost::mutex threads_running_lock_;

  boost::condition_variable threads_changed_cv_;

  int64_t threads_running_;

  int64_t vcores_;

  float max_vcore_oversubscription_ratio_;

  boost::scoped_ptr<Thread> acquire_vcore_thread_;

  boost::shared_ptr<AtomicInt<int16_t> > early_exit_;

  boost::shared_ptr<AtomicInt<int16_t> > thread_in_expand_;

  void AcquireVcoreResources(boost::shared_ptr<AtomicInt<int16_t> > thread_in_expand,
      boost::shared_ptr<AtomicInt<int16_t> > early_exit);

  bool AboveVcoreSubscriptionThreshold();

  bool ShouldExit();
};

}

#endif