be/src/util/histogram.cpp

// 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 <algorithm>
#include <cinttypes>
#include <cmath>
#include <limits>
#include <utility>

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 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<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