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