simple-scheduler.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_SIMPLE_SCHEDULER_H
#define STATESTORE_SIMPLE_SCHEDULER_H

#include <vector>
#include <string>
#include <list>
#include <boost/unordered_map.hpp>
#include <boost/thread/mutex.hpp>

#include "common/status.h"
#include "statestore/scheduler.h"
#include "statestore/statestore-subscriber.h"
#include "statestore/statestore.h"
#include "util/metrics.h"
#include "scheduling/admission-controller.h"
#include "gen-cpp/Types_types.h"  // for TNetworkAddress
#include "gen-cpp/ResourceBrokerService_types.h"
#include "rapidjson/rapidjson.h"

namespace impala {

class ResourceBroker;
class Coordinator;

//
class SimpleScheduler : public Scheduler {
 public:
  static const std::string IMPALA_MEMBERSHIP_TOPIC;

  SimpleScheduler(StatestoreSubscriber* subscriber, const std::string& backend_id,
      const TNetworkAddress& backend_address, MetricGroup* metrics, Webserver* webserver,
      ResourceBroker* resource_broker, RequestPoolService* request_pool_service);

  SimpleScheduler(const std::vector<TNetworkAddress>& backends, MetricGroup* metrics,
      Webserver* webserver, ResourceBroker* resource_broker,
      RequestPoolService* request_pool_service);

  virtual Status GetBackends(const std::vector<TNetworkAddress>& data_locations,
      BackendList* backends);

  virtual impala::Status GetBackend(const TNetworkAddress& data_location,
      TBackendDescriptor* backend);

  virtual void GetAllKnownBackends(BackendList* backends);

  virtual bool HasLocalBackend(const TNetworkAddress& data_location) {
    boost::lock_guard<boost::mutex> l(backend_map_lock_);
    BackendMap::iterator entry = backend_map_.find(data_location.hostname);
    return (entry != backend_map_.end() && entry->second.size() > 0);
  }

  virtual impala::Status Init();

  virtual Status Schedule(Coordinator* coord, QuerySchedule* schedule);
  virtual Status Release(QuerySchedule* schedule);
  virtual void HandlePreemptedReservation(const TUniqueId& reservation_id);
  virtual void HandlePreemptedResource(const TUniqueId& client_resource_id);
  virtual void HandleLostResource(const TUniqueId& client_resource_id);

 private:
  boost::mutex backend_map_lock_;

  typedef boost::unordered_map<std::string, std::list<TBackendDescriptor> > BackendMap;
  BackendMap backend_map_;

  typedef boost::unordered_map<std::string, std::string> BackendIpAddressMap;
  BackendIpAddressMap backend_ip_map_;

  typedef boost::unordered_map<std::string, TBackendDescriptor> BackendIdMap;
  BackendIdMap current_membership_;

  MetricGroup* metrics_;

  Webserver* webserver_;

  BackendMap::iterator next_nonlocal_backend_entry_;

  StatestoreSubscriber* statestore_subscriber_;

  const std::string backend_id_;

  TBackendDescriptor backend_descriptor_;

  ThriftSerializer thrift_serializer_;

  IntCounter* total_assignments_;
  IntCounter* total_local_assignments_;

  BooleanProperty* initialised_;
  IntGauge* num_backends_metric_;

  uint32_t update_count_;

  boost::mutex active_resources_lock_;

  typedef boost::unordered_map<TUniqueId, Coordinator*> ActiveReservationsMap;
  ActiveReservationsMap active_reservations_;

  typedef boost::unordered_map<TUniqueId, Coordinator*> ActiveClientResourcesMap;
  ActiveClientResourcesMap active_client_resources_;

  ResourceBroker* resource_broker_;

  RequestPoolService* request_pool_service_;

  boost::scoped_ptr<AdmissionController> admission_controller_;

  void AddToActiveResourceMaps(
      const TResourceBrokerReservationResponse& reservation, Coordinator* coord);

  void RemoveFromActiveResourceMaps(
      const TResourceBrokerReservationResponse& reservation);

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

  void BackendsUrlCallback(const Webserver::ArgumentMap& args,
      rapidjson::Document* document);

  Status GetRequestPool(const std::string& user,
      const TQueryOptions& query_options, std::string* pool) const;

  Status ComputeScanRangeAssignment(const TQueryExecRequest& exec_request,
      QuerySchedule* schedule);

  Status ComputeScanRangeAssignment(PlanNodeId node_id,
      const std::vector<TScanRangeLocations>& locations,
      const std::vector<TNetworkAddress>& host_list, bool exec_at_coord,
      const TQueryOptions& query_options, FragmentScanRangeAssignment* assignment);

  void ComputeFragmentExecParams(const TQueryExecRequest& exec_request,
      QuerySchedule* schedule);

  void ComputeFragmentHosts(const TQueryExecRequest& exec_request,
      QuerySchedule* schedule);

  PlanNodeId FindLeftmostNode(
      const TPlan& plan, const std::vector<TPlanNodeType::type>& types);

  int FindLeftmostInputFragment(
      int fragment_idx, const TQueryExecRequest& exec_request);

  void GetScanHosts(TPlanNodeId scan_id, const TQueryExecRequest& exec_request,
      const FragmentExecParams& params, std::vector<TNetworkAddress>* scan_hosts);

  bool ContainsNode(const TPlan& plan, TPlanNodeType::type type);

  void FindNodes(const TPlan& plan,
      const std::vector<TPlanNodeType::type>& types, std::vector<TPlanNodeId>* results);

  int FindSenderFragment(TPlanNodeId exch_id, int fragment_idx,
      const TQueryExecRequest& exec_request);
};

}

#endif