statestore.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_STATESTORE_H
#define STATESTORE_STATESTORE_H

#include <stdint.h>
#include <string>
#include <vector>
#include <map>
#include <boost/unordered_map.hpp>
#include <boost/scoped_ptr.hpp>
#include <boost/thread/condition_variable.hpp>
#include <boost/uuid/uuid_generators.hpp>

#include "gen-cpp/Types_types.h"
#include "gen-cpp/StatestoreSubscriber.h"
#include "gen-cpp/StatestoreService.h"
#include "util/metrics.h"
#include "util/collection-metrics.h"
#include "rpc/thrift-client.h"
#include "util/thread-pool.h"
#include "util/webserver.h"
#include "runtime/client-cache.h"
#include "runtime/timestamp-value.h"
#include "statestore/failure-detector.h"

namespace impala {

using namespace impala;

class Status;

//
//
//
//
//
//
class Statestore {
 public:
  typedef std::string SubscriberId;

  typedef std::string TopicId;

  typedef std::string TopicEntryKey;

  Statestore(MetricGroup* metrics);

  //
  Status RegisterSubscriber(const SubscriberId& subscriber_id,
      const TNetworkAddress& location,
      const std::vector<TTopicRegistration>& topic_registrations,
      TUniqueId* registration_id);

  void RegisterWebpages(Webserver* webserver);

  //
  Status MainLoop();

  const boost::shared_ptr<StatestoreServiceIf>& thrift_iface() const {
    return thrift_iface_;
  }

  void SetExitFlag();

 private:
  class TopicEntry {
   public:
    typedef std::string Value;

    typedef uint64_t Version;

    static const Version TOPIC_ENTRY_INITIAL_VERSION = 1L;

    static const Value NULL_VALUE;

    void SetValue(const Value& bytes, Version version);

    TopicEntry() : value_(NULL_VALUE), version_(TOPIC_ENTRY_INITIAL_VERSION) { }

    const Value& value() const { return value_; }
    uint64_t version() const { return version_; }
    uint32_t length() const { return value_.size(); }

   private:
    Value value_;

    Version version_;
  };

  typedef boost::unordered_map<TopicEntryKey, TopicEntry> TopicEntryMap;

  typedef std::map<TopicEntry::Version, TopicEntryKey> TopicUpdateLog;

  //
  class Topic {
   public:
    Topic(const TopicId& topic_id, IntGauge* key_size_metric,
        IntGauge* value_size_metric, IntGauge* topic_size_metric)
        : topic_id_(topic_id), last_version_(0L), total_key_size_bytes_(0L),
          total_value_size_bytes_(0L), key_size_metric_(key_size_metric),
          value_size_metric_(value_size_metric), topic_size_metric_(topic_size_metric) { }

    //
    TopicEntry::Version Put(const TopicEntryKey& key, const TopicEntry::Value& bytes);

    //
    //
    void DeleteIfVersionsMatch(TopicEntry::Version version, const TopicEntryKey& key);

    const TopicId& id() const { return topic_id_; }
    const TopicEntryMap& entries() const { return entries_; }
    TopicEntry::Version last_version() const { return last_version_; }
    const TopicUpdateLog& topic_update_log() const { return topic_update_log_; }
    int64_t total_key_size_bytes() const { return total_key_size_bytes_; }
    int64_t total_value_size_bytes() const { return total_value_size_bytes_; }

   private:
    TopicEntryMap entries_;

    const TopicId topic_id_;

    TopicEntry::Version last_version_;

    //
    TopicUpdateLog topic_update_log_;

    int64_t total_key_size_bytes_;

    int64_t total_value_size_bytes_;

    IntGauge* key_size_metric_;
    IntGauge* value_size_metric_;
    IntGauge* topic_size_metric_;
  };


  boost::mutex exit_flag_lock_;
  bool exit_flag_;

  boost::mutex topic_lock_;

  typedef boost::unordered_map<TopicId, Topic> TopicMap;
  TopicMap topics_;

  class Subscriber {
   public:
    Subscriber(const SubscriberId& subscriber_id, const TUniqueId& registration_id,
        const TNetworkAddress& network_address,
        const std::vector<TTopicRegistration>& subscribed_topics);

    struct TopicState {
      bool is_transient;
      TopicEntry::Version last_version;
    };

    typedef boost::unordered_map<TopicId, TopicState> Topics;

    static const TopicEntry::Version TOPIC_INITIAL_VERSION;

    const Topics& subscribed_topics() const { return subscribed_topics_; }
    const TNetworkAddress& network_address() const { return network_address_; }
    const SubscriberId& id() const { return subscriber_id_; }
    const TUniqueId& registration_id() const { return registration_id_; }

    void AddTransientUpdate(const TopicId& topic_id, const TopicEntryKey& topic_key,
        TopicEntry::Version version);

    typedef boost::unordered_map<std::pair<TopicId, TopicEntryKey>, TopicEntry::Version>
        TransientEntryMap;

    const TransientEntryMap& transient_entries() const { return transient_entries_; }

    const TopicEntry::Version LastTopicVersionProcessed(const TopicId& topic_id) const;

    void SetLastTopicVersionProcessed(const TopicId& topic_id,
        TopicEntry::Version version);

   private:
    const SubscriberId subscriber_id_;

    const TUniqueId registration_id_;

    const TNetworkAddress network_address_;

    Topics subscribed_topics_;

    TransientEntryMap transient_entries_;
  };

  boost::mutex subscribers_lock_;

  //
  typedef boost::unordered_map<SubscriberId, boost::shared_ptr<Subscriber> >
    SubscriberMap;
  SubscriberMap subscribers_;

  boost::uuids::random_generator subscriber_uuid_generator_;

  typedef std::pair<int64_t, SubscriberId> ScheduledSubscriberUpdate;

  //
  //
  //
  //
  ThreadPool<ScheduledSubscriberUpdate> subscriber_topic_update_threadpool_;

  ThreadPool<ScheduledSubscriberUpdate> subscriber_heartbeat_threadpool_;

  boost::scoped_ptr<ClientCache<StatestoreSubscriberClient> > update_state_client_cache_;

  boost::scoped_ptr<ClientCache<StatestoreSubscriberClient> > heartbeat_client_cache_;

  boost::shared_ptr<StatestoreServiceIf> thrift_iface_;

  boost::scoped_ptr<MissedHeartbeatFailureDetector> failure_detector_;

  IntGauge* num_subscribers_metric_;
  SetMetric<std::string>* subscriber_set_metric_;

  IntGauge* key_size_metric_;
  IntGauge* value_size_metric_;
  IntGauge* topic_size_metric_;

  StatsMetric<double>* topic_update_duration_metric_;

  StatsMetric<double>* heartbeat_duration_metric_;

  Status OfferUpdate(const ScheduledSubscriberUpdate& update,
      ThreadPool<ScheduledSubscriberUpdate>* thread_pool);

  void DoSubscriberUpdate(bool is_heartbeat, int thread_id,
      const ScheduledSubscriberUpdate& update);

  //
  Status SendTopicUpdate(Subscriber* subscriber, bool* update_skipped);

  Status SendHeartbeat(Subscriber* subscriber);

  void UnregisterSubscriber(Subscriber* subscriber);

  void GatherTopicUpdates(const Subscriber& subscriber,
      TUpdateStateRequest* update_state_request);

  //
  //
  const TopicEntry::Version GetMinSubscriberTopicVersion(const TopicId& topic_id,
      SubscriberId* subscriber_id = NULL);

  bool ShouldExit();

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

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

};

}

#endif