[Feature] Implementation of histogram metric

LICENSE.txt

@@ -647,3 +647,15 @@ LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
 OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
 THE SOFTWARE.
 
+--------------------------------------------------------------------------------
+
+be/src/util/histogram* : GPLv2, Apache 2.0 License
+
+Copyright (c) 2011-present, 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.

be/src/util/CMakeLists.txt

@@ -37,6 +37,7 @@ set(UTIL_FILES
   disk_info.cpp
   errno.cpp
   hash_util.hpp
+  histogram.cpp
   json_util.cpp
   doris_metrics.cpp
   mem_info.cpp

be/src/util/histogram.cpp

@@ -0,0 +1,230 @@
+// 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 <stdio.h>
+#include <limits>
+#include <cinttypes>
+#include <cmath>
+
+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<double>(_bucket_values.back());
+    while ((bucket_val = 1.5 * bucket_val) <= static_cast<double>(std::numeric_limits<uint64_t>::max())) {
+        _bucket_values.push_back(static_cast<uint64_t>(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<uint64_t, uint64_t>::const_iterator lowerBound =
+        _value_index_map.lower_bound(value);
+        if (lowerBound != _value_index_map.end()) {
+            return static_cast<size_t>(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 performned 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<double>(cur_min);
+            if (r > cur_max) r = static_cast<double>(cur_max);
+            return r;
+        }
+    }
+    return static_cast<double>(max());
+}
+
+double HistogramStat::average() const {
+    uint64_t cur_num = num();
+    uint64_t cur_sum = sum();
+    if (cur_num == 0) return 0;
+    return static_cast<double>(cur_sum) / static_cast<double>(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<double>(cur_sum_squares * cur_num - cur_sum * cur_sum) /
+        static_cast<double>(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<size_t>(mult * bucket_value / 5 + 0.5);
+        r.append(marks, '#');
+        r.push_back('\n');
+    }
+    return r;
+}
+
+}  // namespace doris

be/src/util/histogram.h

@@ -0,0 +1,113 @@
+// 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.
+
+#pragma once
+
+#include <string>
+#include <vector>
+#include <map>
+#include <atomic>
+
+#include "common/logging.h"
+
+namespace doris {
+
+// Histogram data structure implementation:
+//
+// After construction, the 'value_index_map' will be set to:
+// 
+// BucketValue: |   1   |   2   | 2*1.5 |2*1.5^2|2*1.5^3|  ...  |2*1.5^n|  ...  |UINT64MAX|
+// Index:       |   0   |   1   |   2   |   3   |   4   |  ...  |  n-1  |  ...  |   108   |
+//
+// The width of bucket is growing by 1.5 times and its initial values are 1 and 2.
+// The input value will be add to the bucket which lower bound is just greater than
+// input value. For example, input value A > 2*1.5^n and A <= 2*1.5^(n+1), A will be added
+// to the latter bucket.
+class HistogramBucketMapper {
+public:
+    HistogramBucketMapper();
+
+    // converts a value to the bucket index.
+    size_t index_for_value(const uint64_t& value) const;
+    // number of buckets required.
+
+    size_t bucket_count() const {
+        return _bucket_values.size();
+    }
+
+    uint64_t last_value() const {
+        return _max_bucket_value;
+    }
+
+    uint64_t first_value() const {
+        return _min_bucket_value;
+    }
+
+    uint64_t bucket_limit(const size_t bucket_number) const {
+        DCHECK(bucket_number < bucket_count());
+        return _bucket_values[bucket_number];
+    }
+
+private:
+    std::vector<uint64_t> _bucket_values;
+    uint64_t _max_bucket_value;
+    uint64_t _min_bucket_value;
+    std::map<uint64_t, uint64_t> _value_index_map;
+};
+
+struct HistogramStat {
+    HistogramStat();
+    ~HistogramStat() {}
+
+    HistogramStat(const HistogramStat&) = delete;
+    HistogramStat& operator=(const HistogramStat&) = delete;
+
+    void clear();
+    bool is_empty() const;
+    void add(const uint64_t& value);
+    void merge(const HistogramStat& other);
+
+    inline uint64_t min() const { return _min.load(std::memory_order_relaxed); }
+    inline uint64_t max() const { return _max.load(std::memory_order_relaxed); }
+    inline uint64_t num() const { return _num.load(std::memory_order_relaxed); }
+    inline uint64_t sum() const { return _sum.load(std::memory_order_relaxed); }
+    inline uint64_t sum_squares() const {
+        return _sum_squares.load(std::memory_order_relaxed);
+    }
+    inline uint64_t bucket_at(size_t b) const {
+        return _buckets[b].load(std::memory_order_relaxed);
+    }
+
+    double median() const;
+    double percentile(double p) const;
+    double average() const;
+    double standard_deviation() const;
+    std::string to_string() const;
+
+    // To be able to use HistogramStat as thread local variable, it
+    // cannot have dynamic allocated member. That's why we're
+    // using manually values from BucketMapper
+    std::atomic<uint64_t> _min;
+    std::atomic<uint64_t> _max;
+    std::atomic<uint64_t> _num;
+    std::atomic<uint64_t> _sum;
+    std::atomic<uint64_t> _sum_squares;
+    std::atomic<uint64_t> _buckets[109]; // 109==BucketMapper::bucket_count()
+    const uint64_t _num_buckets;
+};
+
+}  // namespace doris