ceph/src/rocksdb/monitoring/histogram_windowing.cc

   1 //  Copyright (c) 2013, Facebook, Inc.  All rights reserved.
   2 //  This source code is licensed under both the GPLv2 (found in the
   3 //  COPYING file in the root directory) and Apache 2.0 License
   4 //  (found in the LICENSE.Apache file in the root directory).
   5 //
   6 // Copyright (c) 2011 The LevelDB Authors. All rights reserved.
   7 // Use of this source code is governed by a BSD-style license that can be
   8 // found in the LICENSE file. See the AUTHORS file for names of contributors.
   9
  10 #include "monitoring/histogram_windowing.h"
  11 #include "monitoring/histogram.h"
  12 #include "util/cast_util.h"
  13
  14 #include <algorithm>
  15
  16 namespace ROCKSDB_NAMESPACE {
  17
  18 HistogramWindowingImpl::HistogramWindowingImpl() {
  19   env_ = Env::Default();
  20   window_stats_.reset(new HistogramStat[static_cast<size_t>(num_windows_)]);
  21   Clear();
  22 }
  23
  24 HistogramWindowingImpl::HistogramWindowingImpl(
  25     uint64_t num_windows,
  26     uint64_t micros_per_window,
  27     uint64_t min_num_per_window) :
  28       num_windows_(num_windows),
  29       micros_per_window_(micros_per_window),
  30       min_num_per_window_(min_num_per_window) {
  31   env_ = Env::Default();
  32   window_stats_.reset(new HistogramStat[static_cast<size_t>(num_windows_)]);
  33   Clear();
  34 }
  35
  36 HistogramWindowingImpl::~HistogramWindowingImpl() {
  37 }
  38
  39 void HistogramWindowingImpl::Clear() {
  40   std::lock_guard<std::mutex> lock(mutex_);
  41
  42   stats_.Clear();
  43   for (size_t i = 0; i < num_windows_; i++) {
  44     window_stats_[i].Clear();
  45   }
  46   current_window_.store(0, std::memory_order_relaxed);
  47   last_swap_time_.store(env_->NowMicros(), std::memory_order_relaxed);
  48 }
  49
  50 bool HistogramWindowingImpl::Empty() const { return stats_.Empty(); }
  51
  52 // This function is designed to be lock free, as it's in the critical path
  53 // of any operation.
  54 // Each individual value is atomic, it is just that some samples can go
  55 // in the older bucket which is tolerable.
  56 void HistogramWindowingImpl::Add(uint64_t value){
  57   TimerTick();
  58
  59   // Parent (global) member update
  60   stats_.Add(value);
  61
  62   // Current window update
  63   window_stats_[static_cast<size_t>(current_window())].Add(value);
  64 }
  65
  66 void HistogramWindowingImpl::Merge(const Histogram& other) {
  67   if (strcmp(Name(), other.Name()) == 0) {
  68     Merge(*static_cast_with_check<const HistogramWindowingImpl>(&other));
  69   }
  70 }
  71
  72 void HistogramWindowingImpl::Merge(const HistogramWindowingImpl& other) {
  73   std::lock_guard<std::mutex> lock(mutex_);
  74   stats_.Merge(other.stats_);
  75
  76   if (stats_.num_buckets_ != other.stats_.num_buckets_ ||
  77       micros_per_window_ != other.micros_per_window_) {
  78     return;
  79   }
  80
  81   uint64_t cur_window = current_window();
  82   uint64_t other_cur_window = other.current_window();
  83   // going backwards for alignment
  84   for (unsigned int i = 0;
  85                     i < std::min(num_windows_, other.num_windows_); i++) {
  86     uint64_t window_index =
  87         (cur_window + num_windows_ - i) % num_windows_;
  88     uint64_t other_window_index =
  89         (other_cur_window + other.num_windows_ - i) % other.num_windows_;
  90     size_t windex = static_cast<size_t>(window_index);
  91     size_t other_windex = static_cast<size_t>(other_window_index);
  92
  93     window_stats_[windex].Merge(
  94       other.window_stats_[other_windex]);
  95   }
  96 }
  97
  98 std::string HistogramWindowingImpl::ToString() const {
  99   return stats_.ToString();
 100 }
 101
 102 double HistogramWindowingImpl::Median() const {
 103   return Percentile(50.0);
 104 }
 105
 106 double HistogramWindowingImpl::Percentile(double p) const {
 107   // Retry 3 times in total
 108   for (int retry = 0; retry < 3; retry++) {
 109     uint64_t start_num = stats_.num();
 110     double result = stats_.Percentile(p);
 111     // Detect if swap buckets or Clear() was called during calculation
 112     if (stats_.num() >= start_num) {
 113       return result;
 114     }
 115   }
 116   return 0.0;
 117 }
 118
 119 double HistogramWindowingImpl::Average() const {
 120   return stats_.Average();
 121 }
 122
 123 double HistogramWindowingImpl::StandardDeviation() const {
 124   return stats_.StandardDeviation();
 125 }
 126
 127 void HistogramWindowingImpl::Data(HistogramData * const data) const {
 128   stats_.Data(data);
 129 }
 130
 131 void HistogramWindowingImpl::TimerTick() {
 132   uint64_t curr_time = env_->NowMicros();
 133   size_t curr_window_ = static_cast<size_t>(current_window());
 134   if (curr_time - last_swap_time() > micros_per_window_ &&
 135       window_stats_[curr_window_].num() >= min_num_per_window_) {
 136     SwapHistoryBucket();
 137   }
 138 }
 139
 140 void HistogramWindowingImpl::SwapHistoryBucket() {
 141   // Threads executing Add() would be competing for this mutex, the first one
 142   // who got the metex would take care of the bucket swap, other threads
 143   // can skip this.
 144   // If mutex is held by Merge() or Clear(), next Add() will take care of the
 145   // swap, if needed.
 146   if (mutex_.try_lock()) {
 147     last_swap_time_.store(env_->NowMicros(), std::memory_order_relaxed);
 148
 149     uint64_t curr_window = current_window();
 150     uint64_t next_window = (curr_window == num_windows_ - 1) ?
 151                                                     0 : curr_window + 1;
 152
 153     // subtract next buckets from totals and swap to next buckets
 154     HistogramStat& stats_to_drop =
 155       window_stats_[static_cast<size_t>(next_window)];
 156
 157     if (!stats_to_drop.Empty()) {
 158       for (size_t b = 0; b < stats_.num_buckets_; b++){
 159         stats_.buckets_[b].fetch_sub(
 160             stats_to_drop.bucket_at(b), std::memory_order_relaxed);
 161       }
 162
 163       if (stats_.min() == stats_to_drop.min()) {
 164         uint64_t new_min = std::numeric_limits<uint64_t>::max();
 165         for (unsigned int i = 0; i < num_windows_; i++) {
 166           if (i != next_window) {
 167             uint64_t m = window_stats_[i].min();
 168             if (m < new_min) new_min = m;
 169           }
 170         }
 171         stats_.min_.store(new_min, std::memory_order_relaxed);
 172       }
 173
 174       if (stats_.max() == stats_to_drop.max()) {
 175         uint64_t new_max = 0;
 176         for (unsigned int i = 0; i < num_windows_; i++) {
 177           if (i != next_window) {
 178             uint64_t m = window_stats_[i].max();
 179             if (m > new_max) new_max = m;
 180           }
 181         }
 182         stats_.max_.store(new_max, std::memory_order_relaxed);
 183       }
 184
 185       stats_.num_.fetch_sub(stats_to_drop.num(), std::memory_order_relaxed);
 186       stats_.sum_.fetch_sub(stats_to_drop.sum(), std::memory_order_relaxed);
 187       stats_.sum_squares_.fetch_sub(
 188                   stats_to_drop.sum_squares(), std::memory_order_relaxed);
 189
 190       stats_to_drop.Clear();
 191     }
 192
 193     // advance to next window bucket
 194     current_window_.store(next_window, std::memory_order_relaxed);
 195
 196     mutex_.unlock();
 197   }
 198 }
 199
 200 }  // namespace ROCKSDB_NAMESPACE