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

// Date: 2015/09/15 15:14:32

#include "bvar/detail/percentile.h"
#include "butil/logging.h"

namespace bvar {
namespace detail {
#if !WITH_BABYLON_COUNTER
inline uint32_t ones32(uint32_t x) {
    /* 32-bit recursive reduction using SWAR...
     * but first step is mapping 2-bit values
     * into sum of 2 1-bit values in sneaky way
     */
    x -= ((x >> 1) & 0x55555555);
    x = (((x >> 2) & 0x33333333) + (x & 0x33333333));
    x = (((x >> 4) + x) & 0x0f0f0f0f);
    x += (x >> 8);
    x += (x >> 16);
    return (x & 0x0000003f);
}

inline uint32_t log2(uint32_t x) {
    int y = (x & (x - 1));
    y |= -y;
    y >>= 31;
    x |= (x >> 1);
    x |= (x >> 2);
    x |= (x >> 4);
    x |= (x >> 8);
    x |= (x >> 16);
    return(ones32(x) - 1 - y);
}

inline size_t get_interval_index(int64_t &x) {
    if (x <= 2) {
        return 0;
    } else if (x > std::numeric_limits<uint32_t>::max()) {
        x = std::numeric_limits<uint32_t>::max();
        return 31;
    } else {
        return log2(x) - 1;
    }
}

class AddLatency {
public:
    AddLatency(int64_t latency) : _latency(latency) {}
    
    void operator()(GlobalValue<Percentile::combiner_type>& global_value,
                    ThreadLocalPercentileSamples& local_value) const {
        // Copy to latency since get_interval_index may change input.
        int64_t latency = _latency;
        const size_t index = get_interval_index(latency);
        PercentileInterval<ThreadLocalPercentileSamples::SAMPLE_SIZE>&
            interval = local_value.get_interval_at(index);
        if (interval.full()) {
            GlobalPercentileSamples* g = global_value.lock();
            g->get_interval_at(index).merge(interval);
            g->_num_added += interval.added_count();
            global_value.unlock();
            local_value._num_added -= interval.added_count();
            interval.clear();
        }
        interval.add64(latency);
        ++local_value._num_added;
    }
private:
    int64_t _latency;
};

Percentile::Percentile()
    : _combiner(std::make_shared<combiner_type>()), _sampler(NULL) {}

Percentile::~Percentile() {
    // Have to destroy sampler first to avoid the race between destruction and
    // sampler
    if (_sampler != NULL) {
        _sampler->destroy();
        _sampler = NULL;
    }
}

Percentile::value_type Percentile::reset() {
    return _combiner->reset_all_agents();
}

Percentile::value_type Percentile::get_value() const {
    return _combiner->combine_agents();
}

Percentile &Percentile::operator<<(int64_t latency) {
    agent_type* agent = _combiner->get_or_create_tls_agent();
    if (BAIDU_UNLIKELY(!agent)) {
        return *this;
    }
    if (latency < 0) {
        // we don't check overflow(of uint32) in percentile because the
        // overflowed value which is included in last range does not affect
        // overall distribution of other values too much.
        if (!_debug_name.empty()) {
            LOG(WARNING) << "Input=" << latency << " to `" << _debug_name
                       << "' is negative, drop";
        } else {
            LOG(WARNING) << "Input=" << latency << " to Percentile("
                       << (void*)this << ") is negative, drop";
        }
        return *this;
    }
    agent->merge_global(AddLatency(latency), _combiner);
    return *this;
}
#else
Percentile::value_type Percentile::reset() {
    constexpr static size_t SAMPLE_SIZE = value_type::SAMPLE_SIZE;
    value_type result;
    _concurrent_sampler.for_each([&](
        size_t index, const babylon::ConcurrentSampler::SampleBucket& bucket) {
        result.merge(bucket, index);
        auto capacity = _concurrent_sampler.bucket_capacity(index);
        auto num_added = bucket.record_num.load(::std::memory_order_relaxed);
        if (capacity < SAMPLE_SIZE && num_added > capacity) {
            capacity = std::min<size_t>(SAMPLE_SIZE, num_added * 1.5);
            _concurrent_sampler.set_bucket_capacity(index, capacity);
        }
    });
    _concurrent_sampler.reset();
    return result;
}

Percentile& Percentile::operator<<(int64_t value) {
    if (BAIDU_UNLIKELY(value < 0)) {
        // we don't check overflow(of uint32) in percentile because the
        // overflowed value which is included in last range does not affect
        // overall distribution of other values too much.
        if (!_debug_name.empty()) {
            LOG_EVERY_SECOND(WARNING) << "Input=" << value << " to `" << _debug_name
                                      << "' is negative, drop";
        } else {
            LOG_EVERY_SECOND(WARNING) << "Input=" << value << " to Percentile("
                                      << (void*)this << ") is negative, drop";
        }
    } else {
        _concurrent_sampler << value;
    }
    return *this;
}
#endif // WITH_BABYLON_COUNTER
}  // namespace detail
}  // namespace bvar