runtime-profile.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 IMPALA_UTIL_RUNTIME_PROFILE_H
#define IMPALA_UTIL_RUNTIME_PROFILE_H

#include <boost/function.hpp>
#include <boost/scoped_ptr.hpp>
#include <boost/thread/mutex.hpp>
#include <boost/unordered_map.hpp>
#include <iostream>
#include <sys/time.h>
#include <sys/resource.h>

#include "common/atomic.h"
#include "common/logging.h"
#include "common/object-pool.h"
#include "util/thread.h"
#include "util/stopwatch.h"
#include "util/streaming-sampler.h"
#include "gen-cpp/RuntimeProfile_types.h"

namespace impala {

#define ENABLE_COUNTERS 1

#define CONCAT_IMPL(x, y) x##y
#define MACRO_CONCAT(x, y) CONCAT_IMPL(x, y)

#if ENABLE_COUNTERS
  #define ADD_COUNTER(profile, name, unit) (profile)->AddCounter(name, unit)
  #define ADD_TIME_SERIES_COUNTER(profile, name, src_counter) \
      (profile)->AddTimeSeriesCounter(name, src_counter)
  #define ADD_TIMER(profile, name) (profile)->AddCounter(name, TUnit::TIME_NS)
  #define ADD_CHILD_TIMER(profile, name, parent) \
      (profile)->AddCounter(name, TUnit::TIME_NS, parent)
  #define SCOPED_TIMER(c) \
      ScopedTimer<MonotonicStopWatch> MACRO_CONCAT(SCOPED_TIMER, __COUNTER__)(c)
  #define COUNTER_ADD(c, v) (c)->Add(v)
  #define COUNTER_SET(c, v) (c)->Set(v)
  #define ADD_THREAD_COUNTERS(profile, prefix) (profile)->AddThreadCounters(prefix)
  #define SCOPED_THREAD_COUNTER_MEASUREMENT(c) \
    ThreadCounterMeasurement \
      MACRO_CONCAT(SCOPED_THREAD_COUNTER_MEASUREMENT, __COUNTER__)(c)
#else
  #define ADD_COUNTER(profile, name, unit) NULL
  #define ADD_TIME_SERIES_COUNTER(profile, name, src_counter) NULL
  #define ADD_TIMER(profile, name) NULL
  #define ADD_CHILD_TIMER(profile, name, parent) NULL
  #define SCOPED_TIMER(c)
  #define COUNTER_ADD(c, v)
  #define COUNTER_SET(c, v)
  #define ADD_THREAD_COUNTERS(profile, prefix) NULL
  #define SCOPED_THREAD_COUNTER_MEASUREMENT(c)
#endif

class ObjectPool;

class RuntimeProfile {
 public:
  class Counter {
   public:
    Counter(TUnit::type unit, int64_t value = 0) :
      value_(value),
      unit_(unit) {
    }
    virtual ~Counter(){}

    virtual void Add(int64_t delta) {
      value_ += delta;
    }

    void BitOr(int64_t delta) {
      value_ |= delta;
    }

    virtual void Set(int64_t value) { value_ = value; }

    virtual void Set(double value) {
      value_ = *reinterpret_cast<int64_t*>(&value);
    }

    virtual int64_t value() const { return value_; }

    virtual double double_value() const {
      return *reinterpret_cast<const double*>(&value_);
    }

    TUnit::type unit() const { return unit_; }

   protected:
    friend class RuntimeProfile;

    AtomicInt<int64_t> value_;
    TUnit::type unit_;
  };

  class HighWaterMarkCounter : public Counter {
   public:
    HighWaterMarkCounter(TUnit::type unit) : Counter(unit) {}

    virtual void Add(int64_t delta) {
      int64_t new_val = current_value_.UpdateAndFetch(delta);
      value_.UpdateMax(new_val);
    }

    bool TryAdd(int64_t delta, int64_t max) {
      while (true) {
        int64_t old_val = current_value_;
        int64_t new_val = old_val + delta;
        if (new_val > max) return false;
        if (LIKELY(current_value_.CompareAndSwap(old_val, new_val))) {
          value_.UpdateMax(new_val);
          return true;
        }
      }
    }

    virtual void Set(int64_t v) {
      current_value_ = v;
      value_.UpdateMax(v);
    }

    int64_t current_value() const { return current_value_; }

   private:
    AtomicInt<int64_t> current_value_;
  };

  typedef boost::function<int64_t ()> DerivedCounterFunction;

  class DerivedCounter : public Counter {
   public:
    DerivedCounter(TUnit::type unit, const DerivedCounterFunction& counter_fn)
      : Counter(unit),
        counter_fn_(counter_fn) {}

    virtual int64_t value() const {
      return counter_fn_();
    }

   private:
    DerivedCounterFunction counter_fn_;
  };

  class AveragedCounter : public Counter {
   public:
    AveragedCounter(TUnit::type unit)
     : Counter(unit),
       current_double_sum_(0.0),
       current_int_sum_(0) {
    }

    void UpdateCounter(Counter* new_counter) {
      DCHECK_EQ(new_counter->unit_, unit_);
      boost::unordered_map<Counter*, int64_t>::iterator it =
          counter_value_map_.find(new_counter);
      int64_t old_val = 0;
      if (it != counter_value_map_.end()) {
        old_val = it->second;
        it->second = new_counter->value_;
      } else {
        counter_value_map_[new_counter] = new_counter->value();
      }

      if (unit_ == TUnit::DOUBLE_VALUE) {
        double old_double_val = *reinterpret_cast<double*>(&old_val);
        current_double_sum_ += (new_counter->double_value() - old_double_val);
        double result_val = current_double_sum_ / (double) counter_value_map_.size();
        value_ = *reinterpret_cast<int64_t*>(&result_val);
      } else {
        current_int_sum_ += (new_counter->value_ - old_val);
        value_ = current_int_sum_ / counter_value_map_.size();
      }
    }

    virtual void Set(double value) {
      DCHECK(false);
    }

    virtual void Set(int64_t value) {
      DCHECK(false);
    }

    virtual void Add(int64_t delta) {
      DCHECK(false);
    }

   private:
    boost::unordered_map<Counter*, int64_t> counter_value_map_;

    double current_double_sum_;
    int64_t current_int_sum_;
  };

  class ThreadCounters {
   private:
    friend class ThreadCounterMeasurement;
    friend class RuntimeProfile;

    Counter* total_time_; // total wall clock time
    Counter* user_time_;  // user CPU time
    Counter* sys_time_;   // system CPU time

    Counter* voluntary_context_switches_;

    Counter* involuntary_context_switches_;
  };

  class EventSequence {
   public:
    EventSequence() { }

    EventSequence(const std::vector<int64_t>& timestamps,
                  const std::vector<std::string>& labels) {
      DCHECK(timestamps.size() == labels.size());
      for (int i = 0; i < timestamps.size(); ++i) {
        events_.push_back(make_pair(labels[i], timestamps[i]));
      }
    }

    void Start() { sw_.Start(); }

    void MarkEvent(const std::string& label) {
      boost::lock_guard<boost::mutex> event_lock(lock_);
      events_.push_back(make_pair(label, sw_.ElapsedTime()));
    }

    int64_t ElapsedTime() { return sw_.ElapsedTime(); }

    typedef std::pair<std::string, int64_t> Event;

    typedef std::vector<Event> EventList;

    void GetEvents(std::vector<Event>* events) {
      events->clear();
      boost::lock_guard<boost::mutex> event_lock(lock_);
      events->insert(events->end(), events_.begin(), events_.end());
    }

    void ToThrift(TEventSequence* seq) const;

   private:
    boost::mutex lock_;

    EventList events_;

    MonotonicStopWatch sw_;
  };

  typedef StreamingSampler<int64_t, 64> StreamingCounterSampler;
  class TimeSeriesCounter {
   public:
    std::string DebugString() const;

    void AddSample(int ms_elapsed) {
      int64_t sample = sample_fn_();
      samples_.AddSample(sample, ms_elapsed);
    }

   private:
    friend class RuntimeProfile;

    TimeSeriesCounter(const std::string& name, TUnit::type unit,
        DerivedCounterFunction fn)
      : name_(name), unit_(unit), sample_fn_(fn) {
    }

    TimeSeriesCounter(const std::string& name, TUnit::type unit, int period,
        const std::vector<int64_t>& values)
      : name_(name), unit_(unit), sample_fn_(NULL), samples_(period, values) {
    }

    void ToThrift(TTimeSeriesCounter* counter);

    std::string name_;
    TUnit::type unit_;
    DerivedCounterFunction sample_fn_;
    StreamingCounterSampler samples_;
  };

  RuntimeProfile(ObjectPool* pool, const std::string& name,
      bool is_averaged_profile = false);

  ~RuntimeProfile();

  static RuntimeProfile* CreateFromThrift(ObjectPool* pool,
      const TRuntimeProfileTree& profiles);

  void AddChild(RuntimeProfile* child,
      bool indent = true, RuntimeProfile* location = NULL);

  template <class Compare>
  void SortChildren(const Compare& cmp) {
    boost::lock_guard<boost::mutex> l(children_lock_);
    std::sort(children_.begin(), children_.end(), cmp);
  }

  void UpdateAverage(RuntimeProfile* src);

  void Update(const TRuntimeProfileTree& thrift_profile);

  Counter* AddCounter(const std::string& name, TUnit::type unit,
      const std::string& parent_counter_name = "");

  HighWaterMarkCounter* AddHighWaterMarkCounter(const std::string& name,
      TUnit::type unit, const std::string& parent_counter_name = "");

  DerivedCounter* AddDerivedCounter(const std::string& name, TUnit::type unit,
      const DerivedCounterFunction& counter_fn,
      const std::string& parent_counter_name = "");

  ThreadCounters* AddThreadCounters(const std::string& prefix);

  Counter* GetCounter(const std::string& name);

  void GetCounters(const std::string& name, std::vector<Counter*>* counters);

  void AddInfoString(const std::string& key, const std::string& value);

  EventSequence* AddEventSequence(const std::string& key);
  EventSequence* AddEventSequence(const std::string& key, const TEventSequence& from);

  EventSequence* GetEventSequence(const std::string& name) const;

  const std::string* GetInfoString(const std::string& key) const;

  Counter* total_time_counter() { return counter_map_[TOTAL_TIME_COUNTER_NAME]; }
  Counter* inactive_timer() { return counter_map_[INACTIVE_TIME_COUNTER_NAME]; }
  Counter* total_async_timer() { return counter_map_[ASYNC_TIME_COUNTER_NAME]; }

  int64_t local_time() { return local_time_ns_; }

  void PrettyPrint(std::ostream* s, const std::string& prefix="") const;

  void ToThrift(TRuntimeProfileTree* tree) const;
  void ToThrift(std::vector<TRuntimeProfileNode>* nodes) const;

  std::string SerializeToArchiveString() const;
  void SerializeToArchiveString(std::stringstream* out) const;

  void Divide(int n);

  void GetChildren(std::vector<RuntimeProfile*>* children);

  void GetAllChildren(std::vector<RuntimeProfile*>* children);

  int num_counters() const { return counter_map_.size(); }

  const std::string& name() const { return name_; }

  void set_name(const std::string& name) { name_ = name; }

  int64_t metadata() const { return metadata_; }
  void set_metadata(int64_t md) { metadata_ = md; }

  static int64_t UnitsPerSecond(
      const Counter* total_counter, const Counter* timer);

  static int64_t CounterSum(const std::vector<Counter*>* counters);

  Counter* AddRateCounter(const std::string& name, Counter* src_counter);

  Counter* AddRateCounter(const std::string& name, DerivedCounterFunction fn,
      TUnit::type unit);

  Counter* AddSamplingCounter(const std::string& name, Counter* src_counter);

  Counter* AddSamplingCounter(const std::string& name, DerivedCounterFunction fn);

  void RegisterBucketingCounters(Counter* src_counter, std::vector<Counter*>* buckets);

  TimeSeriesCounter* AddTimeSeriesCounter(const std::string& name,
      TUnit::type unit, DerivedCounterFunction sample_fn);

  TimeSeriesCounter* AddTimeSeriesCounter(const std::string& name, Counter* src_counter);

  void ComputeTimeInProfile();

 private:
  ObjectPool* pool_;

  bool own_pool_;

  std::string name_;

  int64_t metadata_;

  bool is_averaged_profile_;

  typedef std::map<std::string, Counter*> CounterMap;
  CounterMap counter_map_;

  typedef std::map<std::string, std::set<std::string> > ChildCounterMap;
  ChildCounterMap child_counter_map_;

  std::set<std::vector<Counter*>* > bucketing_counters_;

  mutable boost::mutex counter_map_lock_;

  typedef std::map<std::string, RuntimeProfile*> ChildMap;
  ChildMap child_map_;
  typedef std::vector<std::pair<RuntimeProfile*, bool> > ChildVector;
  ChildVector children_;
  mutable boost::mutex children_lock_;  // protects child_map_ and children_

  typedef std::map<std::string, std::string> InfoStrings;
  InfoStrings info_strings_;

  typedef std::vector<std::string> InfoStringsDisplayOrder;
  InfoStringsDisplayOrder info_strings_display_order_;

  mutable boost::mutex info_strings_lock_;

  typedef std::map<std::string, EventSequence*> EventSequenceMap;
  EventSequenceMap event_sequence_map_;
  mutable boost::mutex event_sequence_lock_;

  typedef std::map<std::string, TimeSeriesCounter*> TimeSeriesCounterMap;
  TimeSeriesCounterMap time_series_counter_map_;
  mutable boost::mutex time_series_counter_map_lock_;

  Counter counter_total_time_;

  Counter total_async_timer_;

  Counter inactive_timer_;

  double local_time_percent_;

  int64_t local_time_ns_;

  void Update(const std::vector<TRuntimeProfileNode>& nodes, int* idx);

  void ComputeTimeInProfile(int64_t total_time);

  static const std::string TOTAL_TIME_COUNTER_NAME;
  static const std::string INACTIVE_TIME_COUNTER_NAME;
  static const std::string ASYNC_TIME_COUNTER_NAME;

  static RuntimeProfile* CreateFromThrift(ObjectPool* pool,
      const std::vector<TRuntimeProfileNode>& nodes, int* node_idx);

  static void PrintChildCounters(const std::string& prefix,
      const std::string& counter_name, const CounterMap& counter_map,
      const ChildCounterMap& child_counter_map, std::ostream* s);
};

class ScopedEvent {
 public:
  ScopedEvent(RuntimeProfile::EventSequence* event_sequence, const std::string& label)
    : label_(label),
      event_sequence_(event_sequence) {
  }

  ~ScopedEvent() {
    event_sequence_->MarkEvent(label_);
  }

 private:
  ScopedEvent(const ScopedEvent& event);
  ScopedEvent& operator=(const ScopedEvent& event);

  const std::string label_;
  RuntimeProfile::EventSequence* event_sequence_;
};

class ScopedCounter {
 public:
  ScopedCounter(RuntimeProfile::Counter* counter, int64_t val) :
    val_(val),
    counter_(counter) {
    if (counter == NULL) return;
    counter_->Add(-1L * val_);
  }

  ~ScopedCounter() {
    if (counter_ != NULL) counter_->Add(val_);
  }

 private:
  ScopedCounter(const ScopedCounter& counter);
  ScopedCounter& operator=(const ScopedCounter& counter);

  int64_t val_;
  RuntimeProfile::Counter* counter_;
};

template<class T>
class ScopedTimer {
 public:
  ScopedTimer(RuntimeProfile::Counter* counter) :
    counter_(counter) {
    if (counter == NULL) return;
    DCHECK(counter->unit() == TUnit::TIME_NS);
    sw_.Start();
  }

  void Stop() { sw_.Stop(); }
  void Start() { sw_.Start(); }

  void UpdateCounter() {
    if (counter_ != NULL) {
      counter_->Add(sw_.ElapsedTime());
    }
  }

  void ReleaseCounter() {
    UpdateCounter();
    counter_ = NULL;
  }

  ~ScopedTimer() {
    sw_.Stop();
    UpdateCounter();
  }

 private:
  ScopedTimer(const ScopedTimer& timer);
  ScopedTimer& operator=(const ScopedTimer& timer);

  T sw_;
  RuntimeProfile::Counter* counter_;
};

class ThreadCounterMeasurement {
 public:
  ThreadCounterMeasurement(RuntimeProfile::ThreadCounters* counters) :
    stop_(false), counters_(counters) {
    DCHECK(counters != NULL);
    sw_.Start();
    int ret = getrusage(RUSAGE_THREAD, &usage_base_);
    DCHECK_EQ(ret, 0);
  }

  void Stop() {
    if (stop_) return;
    stop_ = true;
    sw_.Stop();
    rusage usage;
    int ret = getrusage(RUSAGE_THREAD, &usage);
    DCHECK_EQ(ret, 0);
    int64_t utime_diff =
        (usage.ru_utime.tv_sec - usage_base_.ru_utime.tv_sec) * 1000L * 1000L * 1000L +
        (usage.ru_utime.tv_usec - usage_base_.ru_utime.tv_usec) * 1000L;
    int64_t stime_diff =
        (usage.ru_stime.tv_sec - usage_base_.ru_stime.tv_sec) * 1000L * 1000L * 1000L +
        (usage.ru_stime.tv_usec - usage_base_.ru_stime.tv_usec) * 1000L;
    counters_->total_time_->Add(sw_.ElapsedTime());
    counters_->user_time_->Add(utime_diff);
    counters_->sys_time_->Add(stime_diff);
    counters_->voluntary_context_switches_->Add(usage.ru_nvcsw - usage_base_.ru_nvcsw);
    counters_->involuntary_context_switches_->Add(
        usage.ru_nivcsw - usage_base_.ru_nivcsw);
  }

  ~ThreadCounterMeasurement() {
    Stop();
  }

 private:
  ThreadCounterMeasurement(const ThreadCounterMeasurement& timer);
  ThreadCounterMeasurement& operator=(const ThreadCounterMeasurement& timer);

  bool stop_;
  rusage usage_base_;
  MonotonicStopWatch sw_;
  RuntimeProfile::ThreadCounters* counters_;
};

}

#endif