// Licensed to the Apache Software Foundation (ASF) under one // or more contributor license agreements. See the NOTICE file // distributed with this work for additional information // regarding copyright ownership. The ASF licenses this file // to you 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. #include "util/histogram.h" #include #include #include #include #include #include namespace doris { HistogramBucketMapper::HistogramBucketMapper() { // If you change this, you also need to change // size of array buckets_ in HistogramStat _bucket_values = {1, 2}; _value_index_map = {{1, 0}, {2, 1}}; double bucket_val = static_cast(_bucket_values.back()); while ((bucket_val = 1.5 * bucket_val) <= static_cast(std::numeric_limits::max())) { _bucket_values.push_back(static_cast(bucket_val)); // Extracts two most significant digits to make histogram buckets more // human-readable. E.g., 172 becomes 170. uint64_t pow_of_ten = 1; while (_bucket_values.back() / 10 > 10) { _bucket_values.back() /= 10; pow_of_ten *= 10; } _bucket_values.back() *= pow_of_ten; _value_index_map[_bucket_values.back()] = _bucket_values.size() - 1; } _max_bucket_value = _bucket_values.back(); _min_bucket_value = _bucket_values.front(); } size_t HistogramBucketMapper::index_for_value(const uint64_t& value) const { if (value >= _max_bucket_value) { return _bucket_values.size() - 1; } else if (value >= _min_bucket_value) { std::map::const_iterator lowerBound = _value_index_map.lower_bound(value); if (lowerBound != _value_index_map.end()) { return static_cast(lowerBound->second); } else { return 0; } } else { return 0; } } namespace { const HistogramBucketMapper bucket_mapper; } HistogramStat::HistogramStat() : _num_buckets(bucket_mapper.bucket_count()) { DCHECK(_num_buckets == sizeof(_buckets) / sizeof(*_buckets)); clear(); } void HistogramStat::clear() { _min.store(bucket_mapper.last_value(), std::memory_order_relaxed); _max.store(0, std::memory_order_relaxed); _num.store(0, std::memory_order_relaxed); _sum.store(0, std::memory_order_relaxed); _sum_squares.store(0, std::memory_order_relaxed); for (unsigned int b = 0; b < _num_buckets; b++) { _buckets[b].store(0, std::memory_order_relaxed); } }; bool HistogramStat::is_empty() const { return num() == 0; } void HistogramStat::add(const uint64_t& value) { // This function is designed to be lock free, as it's in the critical path // of any operation. Each individual value is atomic and the order of updates // by concurrent threads is tolerable. const size_t index = bucket_mapper.index_for_value(value); DCHECK(index < _num_buckets); _buckets[index].store(_buckets[index].load(std::memory_order_relaxed) + 1, std::memory_order_relaxed); uint64_t old_min = min(); if (value < old_min) { _min.store(value, std::memory_order_relaxed); } uint64_t old_max = max(); if (value > old_max) { _max.store(value, std::memory_order_relaxed); } _num.store(_num.load(std::memory_order_relaxed) + 1, std::memory_order_relaxed); _sum.store(_sum.load(std::memory_order_relaxed) + value, std::memory_order_relaxed); _sum_squares.store(_sum_squares.load(std::memory_order_relaxed) + value * value, std::memory_order_relaxed); } void HistogramStat::merge(const HistogramStat& other) { // This function needs to be performed with the outer lock acquired // However, atomic operation on every member is still need, since Add() // requires no lock and value update can still happen concurrently uint64_t old_min = min(); uint64_t other_min = other.min(); while (other_min < old_min && !_min.compare_exchange_weak(old_min, other_min)) { } uint64_t old_max = max(); uint64_t other_max = other.max(); while (other_max > old_max && !_max.compare_exchange_weak(old_max, other_max)) { } _num.fetch_add(other.num(), std::memory_order_relaxed); _sum.fetch_add(other.sum(), std::memory_order_relaxed); _sum_squares.fetch_add(other.sum_squares(), std::memory_order_relaxed); for (unsigned int b = 0; b < _num_buckets; b++) { _buckets[b].fetch_add(other.bucket_at(b), std::memory_order_relaxed); } } double HistogramStat::median() const { return percentile(50.0); } double HistogramStat::percentile(double p) const { double threshold = num() * (p / 100.0); uint64_t cumulative_sum = 0; for (unsigned int b = 0; b < _num_buckets; b++) { uint64_t bucket_value = bucket_at(b); cumulative_sum += bucket_value; if (cumulative_sum >= threshold) { // Scale linearly within this bucket uint64_t left_point = (b == 0) ? 0 : bucket_mapper.bucket_limit(b - 1); uint64_t right_point = bucket_mapper.bucket_limit(b); uint64_t left_sum = cumulative_sum - bucket_value; uint64_t right_sum = cumulative_sum; double pos = 0; uint64_t right_left_diff = right_sum - left_sum; if (right_left_diff != 0) { pos = (threshold - left_sum) / right_left_diff; } double r = left_point + (right_point - left_point) * pos; uint64_t cur_min = min(); uint64_t cur_max = max(); if (r < cur_min) r = static_cast(cur_min); if (r > cur_max) r = static_cast(cur_max); return r; } } return static_cast(max()); } double HistogramStat::average() const { uint64_t cur_num = num(); uint64_t cur_sum = sum(); if (cur_num == 0) return 0; return static_cast(cur_sum) / static_cast(cur_num); } double HistogramStat::standard_deviation() const { uint64_t cur_num = num(); uint64_t cur_sum = sum(); uint64_t cur_sum_squares = sum_squares(); if (cur_num == 0) return 0; double variance = static_cast(cur_sum_squares * cur_num - cur_sum * cur_sum) / static_cast(cur_num * cur_num); return std::sqrt(variance); } std::string HistogramStat::to_string() const { uint64_t cur_num = num(); std::string r; char buf[1650]; snprintf(buf, sizeof(buf), "Count: %" PRIu64 " Average: %.4f StdDev: %.2f\n", cur_num, average(), standard_deviation()); r.append(buf); snprintf(buf, sizeof(buf), "Min: %" PRIu64 " Median: %.4f Max: %" PRIu64 "\n", (cur_num == 0 ? 0 : min()), median(), (cur_num == 0 ? 0 : max())); r.append(buf); snprintf(buf, sizeof(buf), "Percentiles: " "P50: %.2f P75: %.2f P99: %.2f P99.9: %.2f P99.99: %.2f\n", percentile(50), percentile(75), percentile(99), percentile(99.9), percentile(99.99)); r.append(buf); r.append("------------------------------------------------------\n"); if (cur_num == 0) return r; // all buckets are empty const double mult = 100.0 / cur_num; uint64_t cumulative_sum = 0; for (unsigned int b = 0; b < _num_buckets; b++) { uint64_t bucket_value = bucket_at(b); if (bucket_value <= 0.0) continue; cumulative_sum += bucket_value; snprintf(buf, sizeof(buf), "%c %7" PRIu64 ", %7" PRIu64 " ] %8" PRIu64 " %7.3f%% %7.3f%% ", (b == 0) ? '[' : '(', (b == 0) ? 0 : bucket_mapper.bucket_limit(b - 1), // left bucket_mapper.bucket_limit(b), // right bucket_value, // count (mult * bucket_value), // percentage (mult * cumulative_sum)); // cumulative percentage r.append(buf); // Add hash marks based on percentage; 20 marks for 100%. size_t marks = static_cast(mult * bucket_value / 5 + 0.5); r.append(marks, '#'); r.push_back('\n'); } return r; } } // namespace doris