statistics.hpp

Go to the documentation of this file.

 
#ifndef JOIN_CORE_STATISTICS_HPP
#define JOIN_CORE_STATISTICS_HPP
 
// libjoin.
#include <join/memory.hpp>
#include <join/clock.hpp>
 
// C++.
#include <algorithm>
#include <iomanip>
#include <ostream>
#include <sstream>
#include <limits>
#include <atomic>
 
// C.
#include <cstdint>
 
namespace join
{
    template <class ClockPolicy>
    class BasicStats
    {
    public:
        using Duration = typename ClockPolicy::Duration;
        using TimePoint = typename ClockPolicy::TimePoint;
 
        explicit BasicStats (const std::string& name = {})
        : _countsMem (_countsSize)
        , _counts (static_cast<std::atomic<uint64_t>*> (_countsMem.get ()))
        , _name (name)
        {
            for (int i = 0; i < _countsLen; ++i)
            {
                new (&_counts[i]) std::atomic<uint64_t> (0);
            }
        }
 
        BasicStats (const BasicStats& other) = delete;
 
        BasicStats& operator= (const BasicStats& other) = delete;
 
        BasicStats (BasicStats&& other) = delete;
 
        BasicStats& operator= (BasicStats&& other) = delete;
 
        ~BasicStats () noexcept
        {
            for (int i = 0; i < _countsLen; ++i)
            {
                _counts[i].~atomic<uint64_t> ();
            }
        }
 
        const std::string& name () const noexcept
        {
            return _name;
        }
 
        TimePoint start () const noexcept
        {
            return ClockPolicy::now ();
        }
 
        void stop (TimePoint startTime) noexcept
        {
            const uint64_t ns =
                static_cast<uint64_t> (std::chrono::duration_cast<Duration> (ClockPolicy::now () - startTime).count ());
 
            _sum.fetch_add (ns, std::memory_order_relaxed);
            _last.store (ns, std::memory_order_relaxed);
 
            auto prev = _min.load (std::memory_order_relaxed);
            while (ns < prev &&
                   !_min.compare_exchange_weak (prev, ns, std::memory_order_relaxed, std::memory_order_relaxed))
                ;
 
            prev = _max.load (std::memory_order_relaxed);
            while (ns > prev &&
                   !_max.compare_exchange_weak (prev, ns, std::memory_order_relaxed, std::memory_order_relaxed))
                ;
 
            _counts[countsIndex (ns)].fetch_add (1, std::memory_order_relaxed);
            _count.fetch_add (1, std::memory_order_release);
        }
 
        void reset () noexcept
        {
            _count.store (0, std::memory_order_relaxed);
            _last.store (0, std::memory_order_relaxed);
            _min.store (std::numeric_limits<uint64_t>::max (), std::memory_order_relaxed);
            _max.store (0, std::memory_order_relaxed);
            _sum.store (0, std::memory_order_relaxed);
            for (int i = 0; i < _countsLen; ++i)
            {
                _counts[i].store (0, std::memory_order_relaxed);
            }
        }
 
        uint64_t count () const noexcept
        {
            return _count.load (std::memory_order_acquire);
        }
 
        Duration last () const noexcept
        {
            if (_count.load (std::memory_order_acquire) == 0)
            {
                return Duration (0);
            }
            return Duration (_last.load (std::memory_order_relaxed));
        }
 
        Duration min () const noexcept
        {
            if (_count.load (std::memory_order_acquire) == 0)
            {
                return Duration (0);
            }
            return Duration (_min.load (std::memory_order_relaxed));
        }
 
        Duration max () const noexcept
        {
            if (_count.load (std::memory_order_acquire) == 0)
            {
                return Duration (0);
            }
            return Duration (_max.load (std::memory_order_relaxed));
        }
 
        std::chrono::duration<double, std::nano> mean () const noexcept
        {
            const auto count = _count.load (std::memory_order_acquire);
            if (count == 0)
            {
                return std::chrono::duration<double, std::nano> (0.0);
            }
            return std::chrono::duration<double, std::nano> (
                static_cast<double> (_sum.load (std::memory_order_relaxed)) / static_cast<double> (count));
        }
 
        double throughput () const noexcept
        {
            const auto count = _count.load (std::memory_order_acquire);
            const auto sum = _sum.load (std::memory_order_acquire);
 
            if (count == 0 || sum == 0)
            {
                return 0.0;
            }
 
            return (static_cast<double> (count) * 1e9) / static_cast<double> (sum);
        }
 
        Duration percentile (double p) const noexcept
        {
            const uint64_t total = _count.load (std::memory_order_acquire);
            if (total == 0)
            {
                return Duration (0);
            }
 
            const uint64_t target = static_cast<uint64_t> (p / 100.0 * static_cast<double> (total));
            uint64_t cumulative = 0;
 
            for (int i = 0; i < _countsLen; ++i)
            {
                cumulative += _counts[i].load (std::memory_order_relaxed);
 
                if (cumulative > target)
                {
                    return Duration (static_cast<typename Duration::rep> (bucketUpperBound (i)));
                }
            }
 
            return Duration (static_cast<typename Duration::rep> (_maxTrackableValue));
        }
 
#ifdef JOIN_HAS_NUMA
        int mbind (int numa) const noexcept
        {
            return _countsMem.mbind (numa);
        }
#endif
 
        int mlock () const noexcept
        {
            return _countsMem.mlock ();
        }
 
    private:
        static int hdrBucketIndex (uint64_t v) noexcept
        {
            const int pow2ceiling = 64 - __builtin_clzll (v | static_cast<uint64_t> (_subBucketCount - 1));
            return std::max (0, pow2ceiling - (_subBucketHalfCountMagnitude + 1));
        }
 
        static int countsIndex (uint64_t ns) noexcept
        {
            if (ns == 0)
            {
                ns = 1;  // LCOV_EXCL_LINE
            }
 
            const int bi = hdrBucketIndex (ns);
            const int si = static_cast<int> (ns >> bi);
            const int idx = (bi + 1) * _subBucketHalfCount + si - _subBucketHalfCount;
 
            if (JOIN_UNLIKELY (idx >= _buckets))
            {
                return _overflowIdx;  // LCOV_EXCL_LINE
            }
 
            return idx;
        }
 
        static uint64_t bucketUpperBound (int idx) noexcept
        {
            if (idx >= _overflowIdx)
            {
                return _maxTrackableValue;  // LCOV_EXCL_LINE
            }
 
            const int bi = std::max (0, idx / _subBucketHalfCount - 1);
            const int si = idx - bi * _subBucketHalfCount;
 
            return static_cast<uint64_t> (si + 1) << bi;
        }
 
        static constexpr int _subBucketHalfCountMagnitude = 7;
 
        static constexpr int _subBucketCount = 1 << (_subBucketHalfCountMagnitude + 1);
 
        static constexpr int _subBucketHalfCount = _subBucketCount >> 1;
 
        static constexpr int _bucketCount = 30;
 
        static constexpr int _buckets = (_bucketCount + 1) * _subBucketHalfCount;
 
        static constexpr int _countsLen = _buckets + 1;
 
        static constexpr int _overflowIdx = _buckets;
 
        static constexpr uint64_t _maxTrackableValue = static_cast<uint64_t> (_subBucketCount) << (_bucketCount - 1);
 
        static constexpr uint64_t _countsSize = static_cast<uint64_t> (_countsLen) * sizeof (std::atomic<uint64_t>);
 
        LocalMem _countsMem;
 
        std::atomic<uint64_t>* const _counts;
 
        alignas (64) std::atomic_uint64_t _count{0};
 
        alignas (64) std::atomic_uint64_t _last{0};
 
        alignas (64) std::atomic_uint64_t _min{std::numeric_limits<uint64_t>::max ()};
 
        alignas (64) std::atomic_uint64_t _max{0};
 
        alignas (64) std::atomic_uint64_t _sum{0};
 
        const std::string _name;
 
        ClockPolicy _clock;
    };
 
    namespace details
    {
        constexpr int colMetric = 24;
 
        constexpr int colCount = 14;
 
        constexpr int colThroughput = 20;
 
        constexpr int colLatency = 16;
 
        constexpr int colTotal = colMetric + colCount + colThroughput + colLatency * 6;
 
        inline int latencyScaleIndex () noexcept
        {
            static int idx = std::ios_base::xalloc ();
            return idx;
        }
 
        inline int throughputScaleIndex () noexcept
        {
            static int idx = std::ios_base::xalloc ();
            return idx;
        }
    }
 
    inline std::ostream& statsHeader (std::ostream& out)
    {
        out << std::left << std::setw (details::colMetric) << "Metric" << std::right << std::setw (details::colCount)
            << "Count" << std::setw (details::colThroughput) << "Throughput" << std::setw (details::colLatency) << "Min"
            << std::setw (details::colLatency) << "Mean" << std::setw (details::colLatency) << "Max"
            << std::setw (details::colLatency) << "P50" << std::setw (details::colLatency) << "P90"
            << std::setw (details::colLatency) << "P99" << "\n"
            << std::string (details::colTotal, '-');
 
        return out;
    }
 
    inline std::ostream& nsec (std::ostream& out)
    {
        out.iword (details::latencyScaleIndex ()) = 1;
        return out;
    }
 
    inline std::ostream& usec (std::ostream& out)
    {
        out.iword (details::latencyScaleIndex ()) = 1'000;
        return out;
    }
 
    inline std::ostream& msec (std::ostream& out)
    {
        out.iword (details::latencyScaleIndex ()) = 1'000'000;
        return out;
    }
 
    inline std::ostream& sec (std::ostream& out)
    {
        out.iword (details::latencyScaleIndex ()) = 1'000'000'000;
        return out;
    }
 
    inline std::ostream& ops (std::ostream& out)
    {
        out.iword (details::throughputScaleIndex ()) = 1;
        return out;
    }
 
    inline std::ostream& kops (std::ostream& out)
    {
        out.iword (details::throughputScaleIndex ()) = 1'000;
        return out;
    }
 
    inline std::ostream& mops (std::ostream& out)
    {
        out.iword (details::throughputScaleIndex ()) = 1'000'000;
        return out;
    }
 
    inline std::ostream& gops (std::ostream& out)
    {
        out.iword (details::throughputScaleIndex ()) = 1'000'000'000;
        return out;
    }
 
    template <class ClockPolicy>
    inline std::ostream& operator<< (std::ostream& out, const BasicStats<ClockPolicy>& statistics)
    {
        // latency scale.
        const long lscale = [&] {
            const long s = out.iword (details::latencyScaleIndex ());
            return s == 0 ? 1L : s;
        }();
 
        const double dlscale = static_cast<double> (lscale);
        const char* lunit = "ns";
        if (lscale == 1'000'000'000)
        {
            lunit = "s";
        }
        else if (lscale == 1'000'000)
        {
            lunit = "ms";
        }
        else if (lscale == 1'000)
        {
            lunit = "us";
        }
 
        // throughput scale.
        const long tscale = [&] {
            const long s = out.iword (details::throughputScaleIndex ());
            return s == 0 ? 1L : s;
        }();
 
        const double dtscale = static_cast<double> (tscale);
        const char* tunit = "ops/s";
        if (tscale == 1'000'000'000)
        {
            tunit = "Gops/s";
        }
        else if (tscale == 1'000'000)
        {
            tunit = "Mops/s";
        }
        else if (tscale == 1'000)
        {
            tunit = "Kops/s";
        }
 
        // format statistics.
        const auto count = statistics.count ();
        const auto min = statistics.min ();
        const auto mean = statistics.mean ();
        const auto max = statistics.max ();
        const auto thr = statistics.throughput ();
        const auto p50 = statistics.percentile (50.0);
        const auto p90 = statistics.percentile (90.0);
        const auto p99 = statistics.percentile (99.0);
 
        auto printLatCol = [&] (double v) {
            std::ostringstream ss;
            ss << std::fixed << std::setprecision (out.precision ()) << v << " (" << lunit << ")";
            out << std::setw (details::colLatency) << ss.str ();
        };
        std::ostringstream oss;
        oss << std::fixed << std::setprecision (out.precision ()) << thr / dtscale << " (" << tunit << ")";
        out << std::left << std::setw (details::colMetric) << statistics.name () << std::right
            << std::setw (details::colCount) << count << std::setw (details::colThroughput) << oss.str ();
        printLatCol (static_cast<double> (min.count ()) / dlscale);
        printLatCol (mean.count () / dlscale);
        printLatCol (static_cast<double> (max.count ()) / dlscale);
        printLatCol (static_cast<double> (p50.count ()) / dlscale);
        printLatCol (static_cast<double> (p90.count ()) / dlscale);
        printLatCol (static_cast<double> (p99.count ()) / dlscale);
 
        return out;
    }
 
    template <typename Statistics>
    class ScopedStats
    {
    public:
        using TimePoint = typename Statistics::TimePoint;
 
        explicit ScopedStats (Statistics& stats) noexcept
        : _statistics (stats)
        , _start (_statistics.start ())
        {
        }
 
        ScopedStats (const ScopedStats& other) = delete;
 
        ScopedStats& operator= (const ScopedStats& other) = delete;
 
        ScopedStats (ScopedStats&& other) = delete;
 
        ScopedStats& operator= (ScopedStats&& other) = delete;
 
        ~ScopedStats () noexcept
        {
            _statistics.stop (_start);
        }
 
    private:
        Statistics& _statistics;
 
        TimePoint _start;
    };
}
 
#endif