// Copyright (c) 2013, Facebook, Inc. All rights reserved. // This source code is licensed under both the GPLv2 (found in the // COPYING file in the root directory) and Apache 2.0 License // (found in the LICENSE.Apache file in the root directory). // // Copyright (c) 2011 The LevelDB Authors. All rights reserved. // Use of this source code is governed by a BSD-style license that can be // found in the LICENSE file. See the AUTHORS file for names of contributors. #include "monitoring/histogram_windowing.h" #include #include "monitoring/histogram.h" #include "rocksdb/system_clock.h" #include "util/cast_util.h" namespace ROCKSDB_NAMESPACE { HistogramWindowingImpl::HistogramWindowingImpl() { clock_ = SystemClock::Default(); window_stats_.reset(new HistogramStat[static_cast(num_windows_)]); Clear(); } HistogramWindowingImpl::HistogramWindowingImpl(uint64_t num_windows, uint64_t micros_per_window, uint64_t min_num_per_window) : num_windows_(num_windows), micros_per_window_(micros_per_window), min_num_per_window_(min_num_per_window) { clock_ = SystemClock::Default(); window_stats_.reset(new HistogramStat[static_cast(num_windows_)]); Clear(); } HistogramWindowingImpl::~HistogramWindowingImpl() = default; void HistogramWindowingImpl::Clear() { std::lock_guard lock(mutex_); stats_.Clear(); for (size_t i = 0; i < num_windows_; i++) { window_stats_[i].Clear(); } current_window_.store(0, std::memory_order_relaxed); last_swap_time_.store(clock_->NowMicros(), std::memory_order_relaxed); } bool HistogramWindowingImpl::Empty() const { return stats_.Empty(); } // This function is designed to be lock free, as it's in the critical path // of any operation. // Each individual value is atomic, it is just that some samples can go // in the older bucket which is tolerable. void HistogramWindowingImpl::Add(uint64_t value) { TimerTick(); // Parent (global) member update stats_.Add(value); // Current window update window_stats_[static_cast(current_window())].Add(value); } void HistogramWindowingImpl::Merge(const Histogram& other) { if (strcmp(Name(), other.Name()) == 0) { Merge(*static_cast_with_check(&other)); } } void HistogramWindowingImpl::Merge(const HistogramWindowingImpl& other) { std::lock_guard lock(mutex_); stats_.Merge(other.stats_); if (stats_.num_buckets_ != other.stats_.num_buckets_ || micros_per_window_ != other.micros_per_window_) { return; } uint64_t cur_window = current_window(); uint64_t other_cur_window = other.current_window(); // going backwards for alignment for (unsigned int i = 0; i < std::min(num_windows_, other.num_windows_); i++) { uint64_t window_index = (cur_window + num_windows_ - i) % num_windows_; uint64_t other_window_index = (other_cur_window + other.num_windows_ - i) % other.num_windows_; size_t windex = static_cast(window_index); size_t other_windex = static_cast(other_window_index); window_stats_[windex].Merge(other.window_stats_[other_windex]); } } std::string HistogramWindowingImpl::ToString() const { return stats_.ToString(); } double HistogramWindowingImpl::Median() const { return Percentile(50.0); } double HistogramWindowingImpl::Percentile(double p) const { // Retry 3 times in total for (int retry = 0; retry < 3; retry++) { uint64_t start_num = stats_.num(); double result = stats_.Percentile(p); // Detect if swap buckets or Clear() was called during calculation if (stats_.num() >= start_num) { return result; } } return 0.0; } double HistogramWindowingImpl::Average() const { return stats_.Average(); } double HistogramWindowingImpl::StandardDeviation() const { return stats_.StandardDeviation(); } void HistogramWindowingImpl::Data(HistogramData* const data) const { stats_.Data(data); } void HistogramWindowingImpl::TimerTick() { uint64_t curr_time = clock_->NowMicros(); size_t curr_window_ = static_cast(current_window()); if (curr_time - last_swap_time() > micros_per_window_ && window_stats_[curr_window_].num() >= min_num_per_window_) { SwapHistoryBucket(); } } void HistogramWindowingImpl::SwapHistoryBucket() { // Threads executing Add() would be competing for this mutex, the first one // who got the metex would take care of the bucket swap, other threads // can skip this. // If mutex is held by Merge() or Clear(), next Add() will take care of the // swap, if needed. if (mutex_.try_lock()) { last_swap_time_.store(clock_->NowMicros(), std::memory_order_relaxed); uint64_t curr_window = current_window(); uint64_t next_window = (curr_window == num_windows_ - 1) ? 0 : curr_window + 1; // subtract next buckets from totals and swap to next buckets HistogramStat& stats_to_drop = window_stats_[static_cast(next_window)]; if (!stats_to_drop.Empty()) { for (size_t b = 0; b < stats_.num_buckets_; b++) { stats_.buckets_[b].fetch_sub(stats_to_drop.bucket_at(b), std::memory_order_relaxed); } if (stats_.min() == stats_to_drop.min()) { uint64_t new_min = std::numeric_limits::max(); for (unsigned int i = 0; i < num_windows_; i++) { if (i != next_window) { uint64_t m = window_stats_[i].min(); if (m < new_min) { new_min = m; } } } stats_.min_.store(new_min, std::memory_order_relaxed); } if (stats_.max() == stats_to_drop.max()) { uint64_t new_max = 0; for (unsigned int i = 0; i < num_windows_; i++) { if (i != next_window) { uint64_t m = window_stats_[i].max(); if (m > new_max) { new_max = m; } } } stats_.max_.store(new_max, std::memory_order_relaxed); } stats_.num_.fetch_sub(stats_to_drop.num(), std::memory_order_relaxed); stats_.sum_.fetch_sub(stats_to_drop.sum(), std::memory_order_relaxed); stats_.sum_squares_.fetch_sub(stats_to_drop.sum_squares(), std::memory_order_relaxed); stats_to_drop.Clear(); } // advance to next window bucket current_window_.store(next_window, std::memory_order_relaxed); mutex_.unlock(); } } } // namespace ROCKSDB_NAMESPACE